import logging from typing import Annotated from datetime import date from sqlalchemy.ext.asyncio import async_sessionmaker from sqlalchemy.future import select from fastapi import APIRouter, Request, Depends from fastapi.responses import HTMLResponse from app.core.config import templates from app.db.schema import get_async_sessionmaker from app.models.recon_job import ReconJob from app.db.schema import ( ReconJob as ReconJobSchema, ReconConfig as ReconConfigSchema ) from app.models.recon_config import config_from_schema router = APIRouter( prefix="", tags=["User_Interface"] ) @router.get("/", response_class=HTMLResponse) async def dashboard_view(request: Request): return templates.TemplateResponse( request=request, name="dashboard.html", context={"title": "Dashboard"}, ) @router.get("/results", response_class=HTMLResponse) async def results_view( request: Request, sessionmaker: Annotated[async_sessionmaker, Depends(get_async_sessionmaker)], cursor: int = 999999, recon_job_name: str|None = None, as_at_date: str|None = None ): PAGE_SIZE = 20 async with sessionmaker() as session: query = ( select(ReconJobSchema) .order_by(ReconJobSchema.id.desc()) .filter(ReconJobSchema.id < cursor) .limit(PAGE_SIZE) ) if recon_job_name: query = query.filter(ReconJobSchema.name == recon_job_name) if as_at_date: query = query.filter(ReconJobSchema.as_at_date == as_at_date) result = await session.execute(query) dbjobs = result.scalars().all() next_cursor = None prev_cursor = None results = [] for dbjob in dbjobs: job = ReconJob.model_validate(dbjob) if job.results: results.append(job.results) if len(dbjobs) > 0: prev_cursor = dbjobs[0].id + PAGE_SIZE next_cursor = dbjobs[-1].id - 1 return templates.TemplateResponse( request=request, name="results.html", context={ "title": "Results", "results": results, "next_cursor": next_cursor, "prev_cursor": prev_cursor, } ) @router.get("/config_helper", response_class=HTMLResponse) async def config_helper_view( request: Request, sessionmaker: Annotated[async_sessionmaker, Depends(get_async_sessionmaker)], reference: str|None = None, as_at_date: str = f"{date.today().isoformat()}" ): config = None if reference is not None: async with sessionmaker() as session: query = ( select(ReconConfigSchema) .filter(ReconConfigSchema.reference == reference) ) result = await session.execute(query) if result: config = config_from_schema(result.scalar_one_or_none()) return templates.TemplateResponse( request=request, name="config_helper.html", context={ "title": "Config Helper", "config": config, "as_at_date": as_at_date, } )