"""Generation and section-retrieval endpoints.""" from __future__ import annotations import json import logging import uuid from datetime import UTC, datetime from pathlib import Path from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession from app.api import rate_limit from app.config import settings from app.db.database import get_db, multi_section_parallel_enabled from app.jobs.models import GenerationJob, JobType from app.jobs.queue import dispatch_or_enqueue from app.db.models import Document, IngestStatus, Report, ReportSection, ReportSectionPhoto, ReportStatus from app.templates.registry import ALL_VALID_SECTION_CODES from app.models.schemas import ( SECTION_TEXT_MAX_BYTES, ExtractNotesResponse, RouteNotesRequest, RouteNotesResponse, GenerateRequest, GenerateResponse, Provenance, SectionPayload, SectionsResponse, SectionTextUpdateBody, TestIngestRequest, WritingStyleProfile, ) from app.services.survey_level_classifier import ( build_or_load_corpus_profiles, classify_file_path, classify_text, ) logger = logging.getLogger(__name__) router = APIRouter() def _meta_interference_level(meta: dict) -> str | None: raw = meta.get("interference_level") if isinstance(raw, str): t = raw.strip().lower() if t == "minimal": t = "minimum" if t in ("minimum", "medium", "maximum"): return t leg = meta.get("ai_involvement_tier") if isinstance(leg, str): t = leg.strip().lower() if t == "minimal": t = "minimum" if t in ("minimum", "medium", "maximum"): return t return None def _meta_int(meta: dict, key: str) -> int | None: v = meta.get(key) if isinstance(v, int): return v if isinstance(v, str) and v.strip().isdigit(): return int(v.strip()) return None # Maximum bytes accepted by the /extract-notes endpoint. _NOTES_MAX_BYTES: int = 10 * 1024 * 1024 # 10 MB @router.post("/reports/{report_id}/generate", response_model=GenerateResponse) async def generate_section( report_id: str, body: GenerateRequest, request: Request, db: AsyncSession = Depends(get_db), _rl: None = Depends(rate_limit.check_generate), ) -> GenerateResponse: """Trigger (or re-trigger) the retrieval + adapt pipeline for one section. Args: report_id: UUID of the report to update. body: Template ID, fact bullets, and optional force-regenerate flag. request: Provides ``state.tenant_id``. db: Injected database session. Returns: :class:`~app.models.schemas.GenerateResponse` with ``status="generating"``. Raises: HTTPException: 404 if the report does not exist or belongs to another tenant. """ tenant_id: str = request.state.tenant_id report = await db.get(Report, report_id) if report is None or report.tenant_id != tenant_id: raise HTTPException(status_code=404, detail="Report not found") from app.services.generation import _resolve_generation_sections, validate_section_codes section_codes = _resolve_generation_sections(body.template_id, body.template_ids) try: validate_section_codes(section_codes) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc # Atomic compare-and-swap: only flip status to "generating" if it is NOT # already "generating". This closes the race where two concurrent POSTs # could both pass a read-then-write 409 check. cas_result = await db.execute( update(Report) .where( Report.id == report_id, Report.tenant_id == tenant_id, Report.status != ReportStatus.generating, ) .values( status=ReportStatus.generating, generation_started_at=datetime.now(UTC), generation_section_total=len(section_codes), ) ) if cas_result.rowcount == 0: raise HTTPException( status_code=409, detail="A generation task is already running for this report. Wait for it to complete or poll /status.", ) await db.commit() from app.services.generation import mark_report_generation_failed, run_generation logger.debug( "POST /generate: ai_percent=%r ai_level=%r interference_level=%r template=%s mode=%s", body.ai_percent, body.ai_level, body.interference_level, body.template_id, body.mode, ) il_arg = str(body.interference_level.value) if body.interference_level is not None else None from app.workflows.client import try_start_report_generation_workflow from app.workflows.models import ReportWorkflowRequest wf_request = ReportWorkflowRequest( report_id=report_id, tenant_id=tenant_id, template_id=body.template_id, bullets=body.bullets, mode=body.mode, ai_level=body.ai_level or 3, ai_percent=body.ai_percent, retrieval_level=body.retrieval_level, force_regenerate=body.force_regenerate, strict_uploaded_only=bool(getattr(body, "strict_uploaded_only", False)), reference_document_ids=body.reference_document_ids, draft_paragraph=body.draft_paragraph, interference_level=il_arg, template_ids=body.template_ids, bullets_by_section=body.bullets_by_section, parallel_sections=bool( multi_section_parallel_enabled() and body.template_ids and len(section_codes) > 1 ), ) workflow_id = await try_start_report_generation_workflow(wf_request) if workflow_id: return GenerateResponse( report_id=report_id, status="generating", message="Generation queued (Temporal workflow).", workflow_id=workflow_id, ) gen_job = GenerationJob( job_type=JobType.generate, report_id=report_id, tenant_id=tenant_id, payload={ "template_id": body.template_id, "bullets": body.bullets, "mode": body.mode, "ai_level": body.ai_level or 3, "ai_percent": body.ai_percent, "retrieval_level": body.retrieval_level, "force_regenerate": body.force_regenerate, "strict_uploaded_only": bool(getattr(body, "strict_uploaded_only", False)), "reference_document_ids": body.reference_document_ids, "draft_paragraph": body.draft_paragraph, "interference_level": il_arg, "template_ids": body.template_ids, "bullets_by_section": body.bullets_by_section, }, ) async def _run_generation_task() -> None: try: await run_generation( report_id=report_id, tenant_id=tenant_id, template_id=body.template_id, bullets=body.bullets, mode=body.mode, ai_level=body.ai_level or 3, ai_percent=body.ai_percent, retrieval_level=body.retrieval_level, force_regenerate=body.force_regenerate, strict_uploaded_only=bool(getattr(body, "strict_uploaded_only", False)), reference_document_ids=body.reference_document_ids, draft_paragraph=body.draft_paragraph, interference_level=il_arg, template_ids=body.template_ids, bullets_by_section=body.bullets_by_section, ) except Exception as exc: # noqa: BLE001 logger.exception("Background generation task failed report=%s", report_id) await mark_report_generation_failed(report_id, tenant_id, str(exc)) queue_mode = await dispatch_or_enqueue(job=gen_job, inline_factory=_run_generation_task) msg = ( "Generation queued (Redis worker)." if queue_mode == "redis" else "Generation queued." ) return GenerateResponse( report_id=report_id, status="generating", message=msg, ) @router.post("/reports/{report_id}/generation/cancel") async def cancel_report_generation( report_id: str, request: Request, db: AsyncSession = Depends(get_db), ) -> dict[str, str]: """Release a report stuck in ``generating`` (client timeout or user cancel).""" tenant_id: str = request.state.tenant_id report = await db.get(Report, report_id) if report is None or report.tenant_id != tenant_id: raise HTTPException(status_code=404, detail="Report not found") if report.status != ReportStatus.generating: return {"report_id": report_id, "status": report.status.value} from sqlalchemy import select from app.db.models import ReportSection from app.services.generation import mark_report_generation_failed sec_result = await db.execute( select(ReportSection).where(ReportSection.report_id == report_id) ) sections = list(sec_result.scalars().all()) with_text = sum(1 for s in sections if (s.text or "").strip()) if with_text > 0: report.status = ReportStatus.partial report.generation_started_at = None report.generation_section_total = None report.error_message = ( f"Generation stopped by client ({with_text} section(s) already saved). " "Re-generate remaining sections from the Results page." )[:2000] await db.commit() return {"report_id": report_id, "status": ReportStatus.partial.value} await mark_report_generation_failed( report_id, tenant_id, "Generation cancelled or timed out before any section completed.", ) return {"report_id": report_id, "status": ReportStatus.failed.value} @router.post("/reports", response_model=GenerateResponse, status_code=201) async def create_report( document_id: str, request: Request, survey_level: int | None = Query( default=None, ge=1, le=3, description=( "RICS Home Survey product tier for this report (1 / 2 / 3). " "Overrides the upload's stored tier when set. " "When omitted, the primary document's ``survey_level`` is used." ), ), confirm_tier_mismatch: bool = Query( default=False, description=( "Set true to proceed even when the primary document is confidently classified " "as a different RICS survey level than ``survey_level``." ), ), db: AsyncSession = Depends(get_db), ) -> GenerateResponse: """Create a new report record linked to an ingested document. Args: document_id: UUID of the already-ingested document. survey_level: Optional tier override for retrieval filtering (library documents must match). request: Provides ``state.tenant_id``. db: Injected database session. Returns: :class:`~app.models.schemas.GenerateResponse` with a new ``report_id``. Raises: HTTPException: 404 if document not found or belongs to another tenant. HTTPException: 409 if the document has not finished ingestion. """ tenant_id: str = request.state.tenant_id doc = await db.get(Document, document_id) if doc is None or doc.tenant_id != tenant_id: raise HTTPException(status_code=404, detail="Document not found") if doc.status != IngestStatus.complete: raise HTTPException( status_code=409, detail=f"Document ingestion status is '{doc.status}'; must be 'complete'.", ) # Guardrail: do not silently generate a Level 2 report from a Level 3 survey (or vice versa). # The UI allows choosing a tier for the report; we enforce that the chosen tier is compatible # with the primary document's content unless the user explicitly confirms the mismatch. if survey_level is not None and not confirm_tier_mismatch: try: fp = Path(doc.file_path) if fp.is_file(): profiles = build_or_load_corpus_profiles(force_refresh=False) c = classify_file_path(fp, filename=doc.filename or fp.name, document_id=doc.id, profiles=profiles) predicted = int(getattr(c, "predicted_survey_level", 0) or 0) conf = float(getattr(c, "confidence", 0.0) or 0.0) if predicted in (1, 2, 3) and predicted != int(survey_level): # Four confidence bands — thresholds depend on whether the # mismatch is between *adjacent* tiers (L1↔L2 or L2↔L3) or # non-adjacent (L1↔L3): # # conf >= 0.45 → hard block (confident mismatch) # 0.30 <= conf < 0.45 → soft warning (clear lean but uncertain) # ADJACENT: 0.15 <= conf < 0.30 → soft warning (adjacent tiers are # easy to confuse; even a slight classifier lean is meaningful) # otherwise → silent pass (truly ambiguous) # # "Adjacent" means the predicted and requested levels differ by # exactly 1 (L1↔L2 or L2↔L3). Non-adjacent (L1↔L3) is a bigger # structural difference and the corpus is more decisive. adjacent = abs(predicted - int(survey_level)) == 1 soft_warn_threshold = 0.15 if adjacent else 0.30 if conf >= 0.45: raise HTTPException( status_code=422, detail={ "error": "Tier mismatch", "message": ( f"Primary document looks like RICS Level {predicted} " f"(confidence {conf:.0%}), but you requested Level {int(survey_level)}." ), "primary_document_id": doc.id, "requested_survey_level": int(survey_level), "predicted_survey_level": predicted, "confidence": round(conf, 3), "rationale": getattr(c, "rationale", ""), "hint": "Retry with confirm_tier_mismatch=true if you really want to proceed.", }, ) if conf >= soft_warn_threshold: adj_note = ( " Level 2 and Level 3 reports share many section headings " "so the classifier lean may be subtle — please verify the document type." ) if adjacent else "" raise HTTPException( status_code=422, detail={ "error": "Tier not verified", "message": ( f"The document analysis has a weak lean towards Level {predicted} " f"(confidence {conf:.0%}), but you selected Level {int(survey_level)}.{adj_note} " "If you know the correct tier, proceed or go back and correct it." ), "primary_document_id": doc.id, "requested_survey_level": int(survey_level), "predicted_survey_level": predicted, "confidence": round(conf, 3), "rationale": getattr(c, "rationale", ""), "hint": ( "Confidence is low — the document may be genuinely ambiguous. " "If you are sure of the tier, click 'Proceed' or fix it and re-upload." ), }, ) # Below threshold: classifier is essentially guessing. # Log for visibility but do not block. logger.info( "tier guardrail skipped (low confidence): doc=%s predicted=%s requested=%s conf=%.2f adjacent=%s", doc.id, predicted, survey_level, conf, adjacent, ) except HTTPException: raise except Exception as exc: # noqa: BLE001 # Guardrail could not run (missing file, loader error, empty corpus) — allow creation # but log a WARNING so operators can see this in production logs. logger.warning( "tier mismatch guardrail skipped for doc=%s (survey_level=%s): %s — " "report will proceed without tier validation.", doc.id, survey_level, exc, ) effective_level = survey_level if survey_level is not None else doc.survey_level report_id = str(uuid.uuid4()) report = Report( id=report_id, tenant_id=tenant_id, document_id=document_id, status=ReportStatus.pending, survey_level=effective_level, ) db.add(report) await db.commit() return GenerateResponse( report_id=report_id, status="pending", message="Report created.", ) @router.get("/reports/{report_id}/sections", response_model=SectionsResponse) async def get_sections( report_id: str, request: Request, db: AsyncSession = Depends(get_db), ) -> SectionsResponse: """Return all generated sections for a report. Args: report_id: UUID of the report. request: Provides ``state.tenant_id``. db: Injected database session. Returns: :class:`~app.models.schemas.SectionsResponse` mapping ``section_code → SectionPayload``. """ tenant_id: str = request.state.tenant_id report = await db.get(Report, report_id) if report is None or report.tenant_id != tenant_id: raise HTTPException(status_code=404, detail="Report not found") result = await db.execute( select(ReportSection).where(ReportSection.report_id == report_id) ) sections_orm = result.scalars().all() photos_res = await db.execute( select(ReportSectionPhoto).where( ReportSectionPhoto.tenant_id == tenant_id, ReportSectionPhoto.report_id == report_id, ) ) photos_rows = photos_res.scalars().all() photos_by_code: dict[str, list[dict]] = {} for p in photos_rows: photos_by_code.setdefault(p.section_code, []).append( { "photo_id": p.id, "original_filename": p.original_filename, "content_type": p.content_type, "created_at": p.created_at.isoformat(), "url": f"/reports/{report_id}/sections/{p.section_code}/photos/{p.id}", "selected_for_ai": bool(getattr(p, "selected_for_ai", False)), } ) sections: dict[str, SectionPayload] = {} for s in sections_orm: raw = json.loads(s.provenance or "[]") # Support both old flat list format and new {sources, meta} format if isinstance(raw, dict): sources = raw.get("sources", []) meta = raw.get("meta", {}) else: sources = raw meta = {} style_profile: WritingStyleProfile | None = None raw_sp = meta.get("style_profile") if isinstance(raw_sp, dict): style_profile = WritingStyleProfile(**raw_sp) prov_list: list[Provenance] = [] for p in sources: if not isinstance(p, dict) or "doc_id" not in p or "chunk_id" not in p or "score" not in p: continue try: prov_list.append(Provenance.model_validate(p)) except Exception: # noqa: BLE001 logger.debug("Skipping malformed provenance row for section=%s", s.section_code) sections[s.section_code] = SectionPayload( text=s.text, confidence=s.confidence, provenance=prov_list, cached=s.cached, mode=meta.get("mode", "generate"), style_profile=style_profile, ai_level=meta.get("ai_level"), ai_percent=meta.get("ai_percent"), ai_transparency=meta.get("ai_transparency") if isinstance(meta.get("ai_transparency"), dict) else None, photos=photos_by_code.get(s.section_code, []), pipeline=meta.get("pipeline"), fallback_used=meta.get("fallback_used"), inspector=meta.get("inspector") if isinstance(meta.get("inspector"), dict) else None, interference_level=_meta_interference_level(meta), word_count=_meta_int(meta, "word_count"), generated_at=meta.get("generated_at") if isinstance(meta.get("generated_at"), str) else None, citation_audit=meta.get("citation_audit") if isinstance(meta.get("citation_audit"), dict) else None, ) return SectionsResponse(report_id=report_id, sections=sections) @router.patch( "/reports/{report_id}/sections/{section_code}", response_model=SectionPayload, ) async def update_section_text( report_id: str, section_code: str, body: SectionTextUpdateBody, request: Request, db: AsyncSession = Depends(get_db), ) -> SectionPayload: """Persist an inline-edited section body for the current tenant. Behaviour: * 404 — ``report_id`` not found, or owned by another tenant. * 422 — ``section_code`` is not a known RICS section code. * If the report exists but there is no ``ReportSection`` row yet, one is **created** (upsert) so surveyors can type manual content before any AI run — same payload shape as an update. * 413 — ``body.text`` exceeds 200 KB (UTF-8 byte length). * 200 — ``ReportSection.text`` is updated, ``meta.user_edited`` is set to ``True`` inside the persisted ``provenance`` JSON, and the freshly loaded :class:`SectionPayload` is returned. All existing meta (provenance sources, mode, confidence, cached, ai_level, ai_percent, ai_transparency, style_profile, pipeline, fallback_used, inspector) is preserved. * Empty-string bodies are accepted (the user may want to clear the section). Tenant isolation is enforced via the standard ``request.state.tenant_id`` middleware: a user from tenant A who tries to PATCH a section that lives under a tenant B report receives 404 (we deliberately do not leak the section's existence with a 403). """ tenant_id: str = request.state.tenant_id # Size guard — count UTF-8 bytes so multibyte content is bounded too. # Done before the DB lookup so oversized payloads fail fast. text_bytes = body.text.encode("utf-8") if len(text_bytes) > SECTION_TEXT_MAX_BYTES: raise HTTPException( status_code=413, detail=( f"Section body too large: {len(text_bytes)} bytes " f"(max {SECTION_TEXT_MAX_BYTES} bytes / 200 KB UTF-8)." ), ) report = await db.get(Report, report_id) if report is None or report.tenant_id != tenant_id: raise HTTPException( status_code=404, detail=f"Report '{report_id}' not found.", ) res = await db.execute( select(ReportSection).where( ReportSection.report_id == report_id, ReportSection.section_code == section_code, ) ) section = res.scalar_one_or_none() if section is None: code = section_code.strip() if not code or code not in ALL_VALID_SECTION_CODES: raise HTTPException( status_code=422, detail=( f"section_code {code!r} is not a valid RICS section code " f"(known codes are listed in the template registry)." ), ) section = ReportSection( id=str(uuid.uuid4()), report_id=report_id, section_code=code, text=body.text, confidence=0.0, cached=False, provenance=json.dumps( { "sources": [], "meta": { "mode": "generate", "user_edited": True, "manual_placeholder": True, }, } ), ) db.add(section) else: # Preserve the {sources, meta} provenance shape; tolerate the legacy # bare-list shape some older sections may still carry. raw = json.loads(section.provenance or "[]") if isinstance(raw, dict): sources = raw.get("sources", []) or [] meta = dict(raw.get("meta", {}) or {}) elif isinstance(raw, list): sources = raw meta = {} else: sources = [] meta = {} section.text = body.text meta["user_edited"] = True meta["word_count"] = len(body.text.split()) section.provenance = json.dumps({"sources": sources, "meta": meta}) await db.commit() await db.refresh(section) raw = json.loads(section.provenance or "[]") if isinstance(raw, dict): sources = raw.get("sources", []) or [] meta = dict(raw.get("meta", {}) or {}) elif isinstance(raw, list): sources = raw meta = {} else: sources = [] meta = {} style_profile: WritingStyleProfile | None = None raw_sp = meta.get("style_profile") if isinstance(raw_sp, dict): try: style_profile = WritingStyleProfile(**raw_sp) except Exception: # noqa: BLE001 style_profile = None prov_list: list[Provenance] = [] for p in sources: if ( not isinstance(p, dict) or "doc_id" not in p or "chunk_id" not in p or "score" not in p ): continue try: prov_list.append(Provenance.model_validate(p)) except Exception: # noqa: BLE001 logger.debug("Skipping malformed provenance row for section=%s", section_code) photos_res = await db.execute( select(ReportSectionPhoto).where( ReportSectionPhoto.tenant_id == tenant_id, ReportSectionPhoto.report_id == report_id, ReportSectionPhoto.section_code == section_code, ) ) photos: list[dict] = [] for p in photos_res.scalars().all(): photos.append( { "photo_id": p.id, "original_filename": p.original_filename, "content_type": p.content_type, "created_at": p.created_at.isoformat(), "url": f"/reports/{report_id}/sections/{section_code}/photos/{p.id}", } ) return SectionPayload( text=section.text, confidence=section.confidence, provenance=prov_list, cached=section.cached, mode=meta.get("mode", "generate"), style_profile=style_profile, ai_level=meta.get("ai_level"), ai_percent=meta.get("ai_percent"), ai_transparency=( meta.get("ai_transparency") if isinstance(meta.get("ai_transparency"), dict) else None ), photos=photos, pipeline=meta.get("pipeline"), fallback_used=meta.get("fallback_used"), inspector=( meta.get("inspector") if isinstance(meta.get("inspector"), dict) else None ), interference_level=_meta_interference_level(meta), word_count=_meta_int(meta, "word_count"), generated_at=meta.get("generated_at") if isinstance(meta.get("generated_at"), str) else None, ) # ── Dev-only endpoint ───────────────────────────────────────────────────────── @router.post("/test/ingest", include_in_schema=settings.dev_mode) async def test_ingest( body: TestIngestRequest, db: AsyncSession = Depends(get_db), ) -> dict[str, str]: """Ingest a sample file directly from disk (dev/test only). Args: body: File path and tenant ID for the test ingestion. db: Injected database session. Returns: Dict with ``document_id`` and ``status``. """ if not settings.dev_mode: raise HTTPException(status_code=403, detail="Only available in dev mode") from pathlib import Path file_path = Path(body.file_path) if not file_path.exists(): raise HTTPException(status_code=404, detail=f"File not found: {file_path}") doc_id = str(uuid.uuid4()) doc = Document( id=doc_id, tenant_id=body.tenant_id, filename=file_path.name, file_path=str(file_path), status=IngestStatus.pending, ) db.add(doc) await db.commit() from app.ingest.pipeline import ingest_document await ingest_document(doc_id=doc_id, file_path=file_path) return {"document_id": doc_id, "status": "ingested"} # ── Notes extraction endpoint ───────────────────────────────────────────────── @router.post("/extract-notes", response_model=ExtractNotesResponse) async def extract_notes( file: UploadFile = File(...), ) -> ExtractNotesResponse: """Extract raw text lines from an uploaded messy notes document. Accepts .docx or .pdf and returns a flat list of non-empty text lines suitable for pasting directly into the bullets textarea on the frontend. No database write is performed — this is a pure parse-and-return endpoint. """ suffix = Path(file.filename or "").suffix.lower() if suffix not in {".docx", ".pdf", ".txt"}: raise HTTPException( status_code=422, detail="Unsupported file type. Upload a .docx, .pdf, or .txt notes file.", ) data = await file.read(_NOTES_MAX_BYTES + 1) # read one extra byte to detect overflow if len(data) > _NOTES_MAX_BYTES: raise HTTPException(status_code=413, detail="Notes file too large (max 10 MB).") lines: list[str] = [] try: if suffix == ".docx": import io from docx import Document as DocxDocument doc = DocxDocument(io.BytesIO(data)) for para in doc.paragraphs: text = para.text.strip() if text: lines.append(text) # Also pull from tables for table in doc.tables: for row in table.rows: for cell in row.cells: text = cell.text.strip() if text and text not in lines: lines.append(text) elif suffix == ".pdf": import io try: import pypdf reader = pypdf.PdfReader(io.BytesIO(data)) for page in reader.pages: page_text = page.extract_text() or "" for line in page_text.splitlines(): line = line.strip() if line: lines.append(line) except ImportError: raise HTTPException( status_code=501, detail="PDF extraction requires pypdf. Install it: pip install pypdf", ) from None else: # .txt text = data.decode("utf-8", errors="replace") lines = [ln.strip() for ln in text.splitlines() if ln.strip()] except HTTPException: raise except Exception as exc: raise HTTPException(status_code=500, detail=f"Failed to parse file: {exc}") from exc # Classify the *content* of the notes so the UI can warn when a user # uploads (e.g.) Level 3 inspection detail onto a Level 1 report. The # original-document tier guardrail at /reports POST only inspects the # primary survey PDF, never the field-notes file, so this is the only # place we can surface the mismatch before routing dumps everything # into Section A. We need a reasonable amount of text — short notes # (e.g. five line headings) carry no usable phrase signal; in that # case predicted_survey_level=0 keeps the classifier silent. predicted_level: int = 0 confidence: float = 0.0 rationale: str = "" joined = "\n".join(lines) if len(joined) >= 200: try: classification = classify_text( joined, filename=file.filename or "notes", ) # Only surface a tier prediction when the classifier has enough # confidence to be useful. Messy field-notes often lack formal # RICS language so a low-confidence guess (e.g. 0.2) would be # misleading — treat anything below 0.30 as unknown (level 0). if float(classification.confidence) >= 0.30: predicted_level = int(classification.predicted_survey_level) confidence = float(classification.confidence) rationale = classification.rationale else: predicted_level = 0 confidence = float(classification.confidence) rationale = ( f"{classification.rationale} " f"(confidence {classification.confidence:.0%} below threshold — " "treating as tier-unknown to avoid false warnings.)" ) except Exception as exc: # Classifier issues must never block the parse — they only # demote the response to "unknown tier". logger.warning("extract-notes tier classification failed: %s", exc) from app.generator.note_bullets import explode_note_lines lines = explode_note_lines(lines) return ExtractNotesResponse( filename=file.filename or "unknown", lines=lines, line_count=len(lines), predicted_survey_level=predicted_level, confidence=confidence, rationale=rationale, ) # ── Notes routing endpoint ──────────────────────────────────────────────────── @router.post("/route-notes", response_model=RouteNotesResponse) async def route_notes_endpoint(body: RouteNotesRequest) -> RouteNotesResponse: """Split a flat list of messy inspector notes into per-section bullets. Companion to ``/extract-notes``: the user uploads one big block of field notes that spans every RICS section (roof + services + grounds + …), this endpoint maps each line to the most relevant section code for the chosen product tier (L1/L2/L3) and returns a ``bullets_by_section`` map ready to feed into ``/reports/{report_id}/generate``. The deterministic keyword router runs first (offline, ~ms). When ``use_llm=true`` and ``OPENAI_API_KEY`` is configured, an LLM refinement pass adjudicates ambiguous lines using the section context — the deterministic match is preserved unless the LLM produces a higher- confidence routing. Lines that don't match any section confidently are returned in ``unrouted_lines`` so the surveyor can decide what to do with them (assign manually, or set ``duplicate_to_unmatched=true`` to bucket them into ``__unrouted__``). """ from app.generator.notes_router import route_notes result = await route_notes( body.lines, survey_level=body.survey_level, use_llm=body.use_llm, duplicate_to_unmatched=body.duplicate_to_unmatched, ) return RouteNotesResponse( survey_level=int(body.survey_level), bullets_by_section=result.bullets_by_section, unrouted_lines=result.unrouted_lines, routing_details=result.routing_details, used_llm=result.used_llm, )