app/api/generate.py

"""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,
    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),
        )
    )
    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 app.services.generation import mark_report_generation_failed

    await mark_report_generation_failed(
        report_id,
        tenant_id,
        "Generation cancelled or timed out.",
    )
    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}",
            }
        )

    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,
        )

    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)

    return ExtractNotesResponse(
        filename=file.filename or "unknown",
        lines=lines,
        line_count=len(lines),
        predicted_survey_level=predicted_level,
        confidence=confidence,
        rationale=rationale,
    )