"""Style library endpoints — manage a tenant's past-report corpus. The style library is the private, per-tenant slice of uploads tagged ``document_purpose=style_corpus``. These docs are: * PII-scrubbed via the LLM + regex sanitiser BEFORE they touch the vector index (forced on regardless of the global sanitisation flag). * Used to build the tenant's :class:`WritingStyleProfile` and to inject verbatim example paragraphs into generation prompts so the model mirrors the surveyor's voice. * DELIBERATELY EXCLUDED from factual retrieval (see ``app.retrieval.retriever._STYLE_CORPUS_EXCLUDE``) so observations from a past property cannot leak into a new report. The endpoints here mirror the shape of ``app.api.upload`` (single + batch upload, list, delete) but stamp ``DocumentPurpose.style_corpus`` so the ingest pipeline routes the file correctly. """ from __future__ import annotations import logging import tempfile import uuid from pathlib import Path from typing import Annotated from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile from sqlalchemy import delete, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.api.rate_limit import check_read from app.config import settings from app.db.database import get_db from app.db.models import Document, DocumentPurpose, IngestStatus, Report from app.ingest.schedule import schedule_ingest from app.services.redaction_upload import ( InvalidRedactionStrategyError, build_upload_db_context, parse_redaction_strategy, serialize_db_context, ) from app.models.schemas import ( BatchUploadItem, DocumentDeleteResponse, StyleLibraryBatchUploadResponse, StyleLibraryDocumentItem, StyleLibraryListResponse, StyleLibraryUploadResponse, ) logger = logging.getLogger(__name__) router = APIRouter() _ALLOWED_SUFFIXES: frozenset[str] = frozenset({".docx", ".pdf"}) _ZIP_SUFFIX: str = ".zip" async def _read_upload_limited(upload: UploadFile, max_bytes: int) -> bytes: data = await upload.read(max_bytes + 1) if len(data) > max_bytes: raise HTTPException( status_code=413, detail=f"File too large. Maximum allowed size is {max_bytes // (1024 * 1024)} MB.", ) return data def _save_style_record( db: AsyncSession, tenant_id: str, filename: str, contents: bytes, suffix: str, *, redaction_strategy: str = "ai_hybrid", redaction_context_json: str | None = None, ) -> tuple[str, Path]: """Persist a style-library upload row + write the bytes to the tenant's upload dir.""" doc_id = str(uuid.uuid4()) dest_dir = settings.upload_dir / tenant_id dest_dir.mkdir(parents=True, exist_ok=True) dest_path = dest_dir / f"{doc_id}{suffix}" dest_path.write_bytes(contents) doc = Document( id=doc_id, tenant_id=tenant_id, filename=filename, file_path=str(dest_path), status=IngestStatus.pending, document_purpose=DocumentPurpose.style_corpus, redaction_strategy=redaction_strategy, redaction_context_json=redaction_context_json, ) db.add(doc) return doc_id, dest_path def _resolve_upload_redaction_strategy(raw: str | None) -> str: try: return parse_redaction_strategy(raw).value except InvalidRedactionStrategyError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc def _invalidate_style_cache(tenant_id: str) -> None: """Drop the cached :class:`WritingStyleProfile` so it rebuilds from the new corpus.""" from app.cache import style_cache as _sc _sc.invalidate(tenant_id) @router.post( "/style-library/upload", response_model=StyleLibraryUploadResponse, status_code=202, summary="Upload one past completed report to the tenant's style library", ) async def upload_style_document( request: Request, file: UploadFile = File(...), redaction_strategy: Annotated[str | None, Form()] = None, client_name: Annotated[str | None, Form()] = None, property_address: Annotated[str | None, Form()] = None, surveyor_name: Annotated[str | None, Form()] = None, db: AsyncSession = Depends(get_db), ) -> StyleLibraryUploadResponse: """Add a past report to this tenant's private style corpus. The file is PII-scrubbed during ingest, indexed under ``document_purpose=style_corpus``, and used to learn the surveyor's voice. It is **never** retrieved as factual evidence for new reports. """ tenant_id: str = request.state.tenant_id suffix = Path(file.filename or "").suffix.lower() if suffix not in _ALLOWED_SUFFIXES: raise HTTPException( status_code=422, detail=f"Unsupported file type '{suffix}'. Allowed: {sorted(_ALLOWED_SUFFIXES)}", ) strategy_value = _resolve_upload_redaction_strategy(redaction_strategy) db_context = await build_upload_db_context( db, tenant_id, client_name=client_name, property_address=property_address, surveyor_name=surveyor_name, ) contents = await _read_upload_limited(file, settings.max_single_upload_bytes) doc_id, dest_path = _save_style_record( db, tenant_id, file.filename or "unknown", contents, suffix, redaction_strategy=strategy_value, redaction_context_json=serialize_db_context(db_context), ) await db.flush() await db.commit() schedule_ingest(doc_id=doc_id, file_path=dest_path) _invalidate_style_cache(tenant_id) logger.info( "Style-library upload accepted doc=%s tenant=%s filename=%s", doc_id, tenant_id, file.filename, ) return StyleLibraryUploadResponse( document_id=doc_id, tenant_id=tenant_id, filename=file.filename or "unknown", status="pending", redaction_strategy=strategy_value, ) @router.post( "/style-library/upload/batch", response_model=StyleLibraryBatchUploadResponse, status_code=202, summary="Upload many past reports (and/or a ZIP) to the style library", ) async def upload_style_batch( request: Request, files: list[UploadFile] = File(...), redaction_strategy: Annotated[str | None, Form()] = None, client_name: Annotated[str | None, Form()] = None, property_address: Annotated[str | None, Form()] = None, surveyor_name: Annotated[str | None, Form()] = None, db: AsyncSession = Depends(get_db), ) -> StyleLibraryBatchUploadResponse: """Bulk-upload past reports. Mirrors POST /upload/batch but stamps style_corpus.""" tenant_id: str = request.state.tenant_id if not files: raise HTTPException(status_code=422, detail="No files uploaded") strategy_value = _resolve_upload_redaction_strategy(redaction_strategy) db_context = await build_upload_db_context( db, tenant_id, client_name=client_name, property_address=property_address, surveyor_name=surveyor_name, ) context_json = serialize_db_context(db_context) items: list[BatchUploadItem] = [] accepted = 0 rejected = 0 logical_count = 0 cap = settings.max_upload_batch_files to_ingest: list[tuple[str, Path]] = [] async def _accept_bytes(original_name: str, body: bytes, suffix: str) -> None: nonlocal accepted, rejected, logical_count if logical_count >= cap: rejected += 1 items.append( BatchUploadItem( filename=original_name, status="rejected", message=f"Batch limit reached ({cap}). Send another batch.", ) ) return doc_id, dest_path = _save_style_record( db, tenant_id, original_name, body, suffix, redaction_strategy=strategy_value, redaction_context_json=context_json, ) await db.flush() to_ingest.append((doc_id, dest_path)) accepted += 1 logical_count += 1 items.append( BatchUploadItem( document_id=doc_id, filename=original_name, status="accepted", message="Stored under style_corpus; ingestion queued.", ) ) for upload in files: name = upload.filename or "unknown" suffix = Path(name).suffix.lower() if suffix == _ZIP_SUFFIX: zmax = settings.max_archive_upload_bytes zbytes = await _read_upload_limited(upload, zmax) with tempfile.TemporaryDirectory() as tmp: zpath = Path(tmp) / "bundle.zip" zpath.write_bytes(zbytes) try: extracted = extract_reference_documents(zpath, Path(tmp) / "out") except ValueError as exc: rejected += 1 items.append( BatchUploadItem( filename=name, status="rejected", message=str(exc) ) ) continue if not extracted: rejected += 1 items.append( BatchUploadItem( filename=name, status="rejected", message="ZIP contained no .docx or .pdf files.", ) ) continue for inner_name, inner_path in extracted: body = inner_path.read_bytes() suf = inner_path.suffix.lower() await _accept_bytes(inner_name, body, suf) continue if suffix not in _ALLOWED_SUFFIXES: rejected += 1 items.append( BatchUploadItem( filename=name, status="rejected", message=f"Unsupported type {suffix!r}; allowed .docx, .pdf, .zip", ) ) continue body = await _read_upload_limited(upload, settings.max_single_upload_bytes) await _accept_bytes(name, body, suffix) await db.commit() for doc_id, dest_path in to_ingest: schedule_ingest(doc_id=doc_id, file_path=dest_path) if accepted > 0: _invalidate_style_cache(tenant_id) msg = ( f"Style library: accepted {accepted}, rejected {rejected}. " "Each file is PII-scrubbed and indexed for voice mirroring only." ) logger.info( "Style-library batch upload tenant=%s accepted=%d rejected=%d", tenant_id, accepted, rejected, ) return StyleLibraryBatchUploadResponse( tenant_id=tenant_id, accepted=accepted, rejected=rejected, items=items, message=msg, ) @router.get( "/style-library", response_model=StyleLibraryListResponse, summary="List past reports in the tenant's style library", ) async def list_style_library( request: Request, db: AsyncSession = Depends(get_db), _: None = Depends(check_read), ) -> StyleLibraryListResponse: """Return all style-corpus documents for the authenticated tenant.""" tenant_id: str = request.state.tenant_id rows = ( ( await db.execute( select(Document) .where( Document.tenant_id == tenant_id, Document.document_purpose == DocumentPurpose.style_corpus, ) .order_by(Document.created_at.desc()) ) ) .scalars() .all() ) items = [ StyleLibraryDocumentItem( document_id=d.id, filename=d.filename, status=d.status.value if hasattr(d.status, "value") else str(d.status), error_message=d.error_message, survey_level=d.survey_level, created_at=d.created_at, updated_at=d.updated_at, ) for d in rows ] return StyleLibraryListResponse( tenant_id=tenant_id, total=len(items), items=items, ) @router.delete( "/style-library/{document_id}", response_model=DocumentDeleteResponse, summary="Remove a past report from the tenant's style library", ) async def delete_style_document( document_id: str, request: Request, db: AsyncSession = Depends(get_db), _: None = Depends(check_read), ) -> DocumentDeleteResponse: """Delete a style-library document and its vector-index chunks. Refuses with 404 when the document does not exist or does not belong to the authenticated tenant, and refuses with 409 when the document is still linked to a report job (which shouldn't happen for style_corpus docs, but is defended against here). """ 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="Style document not found") if doc.document_purpose != DocumentPurpose.style_corpus: raise HTTPException( status_code=400, detail=( "This document is not part of your style library " "(use DELETE /documents/{id} for report-source uploads)." ), ) cnt = await db.execute( select(func.count()).select_from(Report).where(Report.document_id == document_id) ) if int(cnt.scalar_one() or 0) > 0: raise HTTPException( status_code=409, detail=( "This style-library document is still linked to a report. " "Detach or delete the report first." ), ) try: from app.vectorstore.factory import get_vectorstore get_vectorstore().delete_document(document_id) except Exception as exc: # noqa: BLE001 logger.warning("Vector store delete failed for style doc=%s: %s", document_id, exc) fp = Path(doc.file_path) try: if fp.is_file(): fp.unlink() except OSError as exc: logger.warning("Could not remove style file %s: %s", fp, exc) await db.execute(delete(Document).where(Document.id == document_id)) await db.commit() _invalidate_style_cache(tenant_id) return DocumentDeleteResponse( document_id=document_id, detail="Style-library document removed from disk, database, and search index.", )