""" nfr.api.routes.review ~~~~~~~~~~~~~~~~~~~~~ Review queue routes for the NAS File Router operator dashboard. All mutating actions (accept / reject / reroute) require a POST body that includes ``operator_notes`` so that every action is traceable in the audit log. Every action emits an audit event via ``nfr.audit.logger.emit``. Routes ------ GET /review List pending review items (oldest first) GET /review/{id} Review item detail POST /review/{id}/accept Accept proposed routing → triggers Safe NAS Writer POST /review/{id}/reject Mark item as rejected POST /review/{id}/reroute Operator selects a different project and subfolder Security / safety ----------------- - All mutating routes require a POST body (not GET) to prevent CSRF from simple link clicks. - The ``operator_notes`` field is mandatory for audit traceability. - No file is ever deleted. A rejected item remains on the NAS review queue path — only its DB status changes. - ``accept`` calls the Safe NAS Writer, which enforces all safety invariants (boundary check, template validation, dry-run mode). Never hard-code secrets here. """ from __future__ import annotations import logging from datetime import datetime, timezone from pathlib import Path from typing import Optional from fastapi import APIRouter, Form, HTTPException, Request, status from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse from fastapi.templating import Jinja2Templates from sqlalchemy import desc, func, select from nfr.database import get_async_session from nfr.models.db import AuditEvent, IngestRecord, Project, ReviewQueueItem, RoutingDecision from nfr.settings import get_settings logger = logging.getLogger(__name__) _HERE = Path(__file__).parent.parent templates = Jinja2Templates(directory=str(_HERE / "templates")) router = APIRouter(tags=["review"]) def _utcnow() -> datetime: return datetime.now(tz=timezone.utc) def _emit_audit(session, event_type: str, review_item: ReviewQueueItem, actor: str, payload: dict) -> None: """ Emit an audit event synchronously (for use inside async routes that have a session). Writes to the ``audit_events`` table. File-based audit log is handled separately by ``nfr.audit.logger``. """ from ulid import ULID # noqa: PLC0415 event = AuditEvent( event_id=str(ULID()), event_type=event_type, severity="INFO", component="review_queue", ingest_record_id=review_item.ingest_record_id, actor=actor, payload=payload, created_at=_utcnow(), ) session.add(event) # ── GET /review ─────────────────────────────────────────────────────────────── @router.get("/review", response_class=HTMLResponse, summary="Review queue") async def review_list( request: Request, reason_filter: Optional[str] = None, page: int = 1, page_size: int = 30, ) -> HTMLResponse: """ List all pending review items sorted oldest-first. Supports optional ``?reason_filter=LOW_CONFIDENCE_MATCH`` URL parameter to narrow by review reason code. """ settings = get_settings() offset = (page - 1) * page_size async with get_async_session() as session: stmt = ( select(ReviewQueueItem) .where(ReviewQueueItem.operator_status == "pending") .order_by(ReviewQueueItem.created_at.asc()) ) if reason_filter: stmt = stmt.where(ReviewQueueItem.review_reason == reason_filter) count_q = await session.execute( select(func.count()).select_from(stmt.subquery()) ) total_count: int = count_q.scalar_one_or_none() or 0 items_q = await session.execute(stmt.offset(offset).limit(page_size)) items = items_q.scalars().all() # Load associated ingest records for display ingest_ids = [item.ingest_record_id for item in items] ingest_map: dict[int, IngestRecord] = {} if ingest_ids: ing_q = await session.execute( select(IngestRecord).where(IngestRecord.id.in_(ingest_ids)) ) for rec in ing_q.scalars().all(): ingest_map[rec.id] = rec # Load proposed projects for display in the list project_ids = [item.proposed_project_id for item in items if item.proposed_project_id] project_map: dict[int, Project] = {} if project_ids: proj_q = await session.execute( select(Project).where(Project.id.in_(project_ids)) ) for proj in proj_q.scalars().all(): project_map[proj.id] = proj # Available reason codes for filter dropdown reason_q = await session.execute( select(ReviewQueueItem.review_reason).distinct() ) available_reasons = [row[0] for row in reason_q.all()] total_pages = max(1, (total_count + page_size - 1) // page_size) context = { "request": request, "settings": settings, "items": items, "ingest_map": ingest_map, "project_map": project_map, "total_count": total_count, "page": page, "page_size": page_size, "total_pages": total_pages, "reason_filter": reason_filter or "", "available_reasons": available_reasons, } return templates.TemplateResponse(request, "review.html", context) # ── GET /review/{id}/preview ───────────────────────────────────────────────── @router.get("/review/{item_id}/preview", summary="Preview review item file") async def review_preview(request: Request, item_id: int): """ Serve the file associated with a review item for inline preview. Looks for the file at ``review_nas_path`` first. If the file is not on disk (e.g. older dry-run items), falls back to a redirect to the original ClickUp download URL. Returns a FileResponse with the appropriate content type so the browser can render PDFs and images inline. """ async with get_async_session() as session: item: ReviewQueueItem | None = await session.get(ReviewQueueItem, item_id) if item is None: raise HTTPException(status_code=404, detail="Review item not found") ingest_record: IngestRecord | None = await session.get( IngestRecord, item.ingest_record_id ) # Try the NAS review path first if item.review_nas_path: file_path = Path(item.review_nas_path) if file_path.is_file(): content_type = (ingest_record.content_type if ingest_record else None) or "application/octet-stream" filename = ingest_record.original_filename if ingest_record else file_path.name return FileResponse( path=str(file_path), media_type=content_type, filename=filename, content_disposition_type="inline", ) # Fallback: try temp staging directory settings = get_settings() if ingest_record and ingest_record.ulid: staging_dir = settings.TEMP_STAGING_DIR if staging_dir.exists(): for f in staging_dir.iterdir(): if f.name.startswith(ingest_record.ulid): content_type = (ingest_record.content_type if ingest_record else None) or "application/octet-stream" return FileResponse( path=str(f), media_type=content_type, filename=ingest_record.original_filename or f.name, content_disposition_type="inline", ) # Last resort: redirect to the original download URL (may be expired) if ingest_record and ingest_record.download_url: from fastapi.responses import RedirectResponse as _RR # noqa: PLC0415 return _RR(url=ingest_record.download_url) raise HTTPException(status_code=404, detail="File not found — not on NAS and no download URL available") # ── GET /review/{id} ───────────────────────────────────────────────────────── @router.get("/review/{item_id}", response_class=HTMLResponse, summary="Review item detail") async def review_detail(request: Request, item_id: int) -> HTMLResponse: """Detail page for a single review queue item.""" settings = get_settings() async with get_async_session() as session: item: ReviewQueueItem | None = await session.get(ReviewQueueItem, item_id) if item is None: raise HTTPException(status_code=404, detail="Review item not found") ingest_record: IngestRecord | None = await session.get( IngestRecord, item.ingest_record_id ) proposed_project: Project | None = None if item.proposed_project_id: proposed_project = await session.get(Project, item.proposed_project_id) # Load routing decision for this ingest record routing_decision: RoutingDecision | None = None if ingest_record: rd_q = await session.execute( select(RoutingDecision) .where(RoutingDecision.ingest_record_id == ingest_record.id) .order_by(RoutingDecision.decided_at.desc()) .limit(1) ) routing_decision = rd_q.scalar_one_or_none() # All active projects for the reroute selector projects_q = await session.execute( select(Project) .where(Project.state == "active") .order_by(Project.year.desc(), Project.project_number) ) all_projects = projects_q.scalars().all() # ── Fetch ClickUp task context (best-effort, non-blocking) ─────────── clickup_task: dict | None = None clickup_comments: list[dict] = [] if ingest_record and ingest_record.clickup_task_id: import httpx as _httpx # noqa: PLC0415 cu_headers = { "Authorization": settings.CLICKUP_API_TOKEN, "Content-Type": "application/json", } try: async with _httpx.AsyncClient( base_url="https://api.clickup.com/api/v2", headers=cu_headers, timeout=_httpx.Timeout(connect=5.0, read=10.0, write=5.0, pool=5.0), ) as cu_client: task_resp = await cu_client.get(f"/task/{ingest_record.clickup_task_id}") if task_resp.status_code == 200: clickup_task = task_resp.json() comments_resp = await cu_client.get(f"/task/{ingest_record.clickup_task_id}/comment") if comments_resp.status_code == 200: clickup_comments = (comments_resp.json().get("comments") or [])[:10] except Exception as cu_exc: logger.warning("review_detail: failed to fetch ClickUp task data: %s", cu_exc) context = { "request": request, "settings": settings, "item": item, "ingest_record": ingest_record, "proposed_project": proposed_project, "all_projects": all_projects, "clickup_task": clickup_task, "clickup_comments": clickup_comments, "routing_decision": routing_decision, } return templates.TemplateResponse(request, "review_detail.html", context) # ── POST /review/{id}/accept ───────────────────────────────────────────────── @router.post("/review/{item_id}/accept", summary="Accept proposed routing") async def review_accept( request: Request, item_id: int, operator_notes: str = Form(..., description="Mandatory operator notes / reason"), operator_id: str = Form(default="operator"), ) -> RedirectResponse: """ Accept the proposed routing for a review item. Triggers the Safe NAS Writer to write the file from the review queue staging path to the proposed destination on the NAS. Requires ``operator_notes`` (POST body field). Emits ``REVIEW_ACCEPTED`` audit event. Redirects to ``/review`` on success. """ settings = get_settings() async with get_async_session() as session: item: ReviewQueueItem | None = await session.get(ReviewQueueItem, item_id) if item is None: raise HTTPException(status_code=404, detail="Review item not found") if item.operator_status != "pending": raise HTTPException( status_code=409, detail=f"Item is already in status {item.operator_status!r}", ) ingest_record: IngestRecord | None = await session.get( IngestRecord, item.ingest_record_id ) if ingest_record is None: raise HTTPException(status_code=404, detail="Associated IngestRecord not found") proposed_project: Project | None = None if item.proposed_project_id: proposed_project = await session.get(Project, item.proposed_project_id) if proposed_project is None: raise HTTPException( status_code=422, detail="No proposed project set on this review item — use /reroute to assign one", ) if not item.proposed_destination: raise HTTPException( status_code=422, detail="No proposed destination subfolder set — use /reroute to assign one", ) # ── Trigger Safe NAS Writer ────────────────────────────────────────── from nfr.nas.safe_writer import SafeNASWriter # noqa: PLC0415 writer = SafeNASWriter() review_file_path = Path(item.review_nas_path) write_result = writer.write( session=session, ingest_record=ingest_record, project=proposed_project, destination_subfolder=item.proposed_destination, temp_file_path=review_file_path, ) # ── Update DB state ────────────────────────────────────────────────── now = _utcnow() item.operator_status = "accepted" item.operator_id = operator_id item.operator_action_at = now item.operator_notes = operator_notes if write_result.nas_path: item.final_nas_path = write_result.nas_path ingest_record.nas_final_path = write_result.nas_path ingest_record.status = "done" if write_result.success else "error" ingest_record.matched_project_id = proposed_project.id ingest_record.routed_subfolder = item.proposed_destination ingest_record.processed_at = now # ── Audit event ────────────────────────────────────────────────────── _emit_audit(session, "REVIEW_ACCEPTED", item, actor=operator_id, payload={ "operator_notes": operator_notes, "proposed_destination": item.proposed_destination, "project_id": proposed_project.id, "project_number": proposed_project.project_number, "write_success": write_result.success, "dry_run": write_result.dry_run, "nas_path": write_result.nas_path, }) logger.info( "Review accepted: item_id=%d ingest_record_id=%d nas_path=%r dry_run=%s", item_id, item.ingest_record_id, write_result.nas_path, write_result.dry_run, ) return RedirectResponse(url="/review", status_code=status.HTTP_303_SEE_OTHER) # ── POST /review/{id}/reject ───────────────────────────────────────────────── @router.post("/review/{item_id}/reject", summary="Reject review item") async def review_reject( request: Request, item_id: int, operator_notes: str = Form(..., description="Mandatory operator notes / reason for rejection"), operator_id: str = Form(default="operator"), ) -> RedirectResponse: """ Mark a review item as rejected. The file is **not** deleted — it remains in the review queue NAS path. Only the DB record status changes. Emits ``REVIEW_REJECTED`` audit event. """ async with get_async_session() as session: item: ReviewQueueItem | None = await session.get(ReviewQueueItem, item_id) if item is None: raise HTTPException(status_code=404, detail="Review item not found") if item.operator_status != "pending": raise HTTPException( status_code=409, detail=f"Item is already in status {item.operator_status!r}", ) now = _utcnow() item.operator_status = "rejected" item.operator_id = operator_id item.operator_action_at = now item.operator_notes = operator_notes # Also update the ingest record status ingest_record: IngestRecord | None = await session.get( IngestRecord, item.ingest_record_id ) if ingest_record is not None: ingest_record.status = "skipped" ingest_record.processed_at = now _emit_audit(session, "REVIEW_REJECTED", item, actor=operator_id, payload={ "operator_notes": operator_notes, }) logger.info( "Review rejected: item_id=%d ingest_record_id=%d", item_id, item.ingest_record_id, ) return RedirectResponse(url="/review", status_code=status.HTTP_303_SEE_OTHER) # ── POST /review/{id}/reroute ───────────────────────────────────────────────── @router.post("/review/{item_id}/reroute", summary="Reroute to different project/folder") async def review_reroute( request: Request, item_id: int, project_id: int = Form(..., description="Target project ID"), destination_subfolder: str = Form( ..., description="Relative subfolder within the project root" ), operator_notes: str = Form(..., description="Mandatory operator notes"), operator_id: str = Form(default="operator"), ) -> RedirectResponse: """ The operator selects a different project and subfolder for this review item, then triggers the Safe NAS Writer to write the file to the new destination. ``project_id`` must be an existing active project. ``destination_subfolder`` must be a template-approved subfolder path. Emits ``REVIEW_ACCEPTED`` audit event with ``rerouted=True``. """ async with get_async_session() as session: item: ReviewQueueItem | None = await session.get(ReviewQueueItem, item_id) if item is None: raise HTTPException(status_code=404, detail="Review item not found") if item.operator_status != "pending": raise HTTPException( status_code=409, detail=f"Item is already in status {item.operator_status!r}", ) target_project: Project | None = await session.get(Project, project_id) if target_project is None: raise HTTPException(status_code=404, detail="Target project not found") ingest_record: IngestRecord | None = await session.get( IngestRecord, item.ingest_record_id ) if ingest_record is None: raise HTTPException(status_code=404, detail="Associated IngestRecord not found") # ── Update proposed routing on the item ────────────────────────────── item.proposed_project_id = project_id item.proposed_destination = destination_subfolder.strip("/") # ── Trigger Safe NAS Writer ────────────────────────────────────────── from nfr.nas.safe_writer import SafeNASWriter # noqa: PLC0415 writer = SafeNASWriter() review_file_path = Path(item.review_nas_path) write_result = writer.write( session=session, ingest_record=ingest_record, project=target_project, destination_subfolder=item.proposed_destination, temp_file_path=review_file_path, ) # ── Update DB state ────────────────────────────────────────────────── now = _utcnow() item.operator_status = "rerouted" item.operator_id = operator_id item.operator_action_at = now item.operator_notes = operator_notes if write_result.nas_path: item.final_nas_path = write_result.nas_path ingest_record.nas_final_path = write_result.nas_path ingest_record.status = "done" if write_result.success else "error" ingest_record.matched_project_id = project_id ingest_record.routed_subfolder = item.proposed_destination ingest_record.processed_at = now _emit_audit(session, "REVIEW_ACCEPTED", item, actor=operator_id, payload={ "operator_notes": operator_notes, "rerouted": True, "original_proposed_project_id": item.proposed_project_id, "new_project_id": project_id, "destination_subfolder": item.proposed_destination, "write_success": write_result.success, "dry_run": write_result.dry_run, "nas_path": write_result.nas_path, }) logger.info( "Review rerouted: item_id=%d ingest_record_id=%d project_id=%d subfolder=%r", item_id, item.ingest_record_id, project_id, item.proposed_destination, ) return RedirectResponse(url="/review", status_code=status.HTTP_303_SEE_OTHER)