Spaces:
Runtime error
Runtime error
| """Upload-time redaction strategy parsing and session PII context assembly. | |
| Builds the ``db_context`` map forwarded to :func:`sanitise_text_for_rag_sync` | |
| for exact-string wipes in the deterministic redaction path. Values originate | |
| from optional multipart form fields and, when absent, from the tenant's most | |
| recent report section text (best-effort identity extraction). | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| from typing import TYPE_CHECKING | |
| from app.services.document_sanitiser import RedactionStrategy | |
| if TYPE_CHECKING: | |
| from sqlalchemy.ext.asyncio import AsyncSession | |
| logger = logging.getLogger(__name__) | |
| class InvalidRedactionStrategyError(ValueError): | |
| """Raised when an upload ``redaction_strategy`` value cannot be mapped.""" | |
| _STRATEGY_ALIASES: dict[str, RedactionStrategy] = { | |
| "ai_hybrid": RedactionStrategy.AI_HYBRID, | |
| "ai": RedactionStrategy.AI_HYBRID, | |
| "hybrid": RedactionStrategy.AI_HYBRID, | |
| "deterministic": RedactionStrategy.DETERMINISTIC_CODE, | |
| "deterministic_code": RedactionStrategy.DETERMINISTIC_CODE, | |
| "code": RedactionStrategy.DETERMINISTIC_CODE, | |
| } | |
| def parse_redaction_strategy(raw: str | None) -> RedactionStrategy: | |
| """Map an upload form ``redaction_strategy`` string to :class:`RedactionStrategy`. | |
| Args: | |
| raw: User-supplied strategy name. ``None`` or blank defaults to | |
| ``AI_HYBRID``. | |
| Returns: | |
| The resolved enum member. | |
| Raises: | |
| InvalidRedactionStrategyError: When ``raw`` is non-empty but not recognised. | |
| """ | |
| if raw is None or not str(raw).strip(): | |
| return RedactionStrategy.AI_HYBRID | |
| key = str(raw).strip().lower().replace("-", "_") | |
| resolved = _STRATEGY_ALIASES.get(key) | |
| if resolved is None: | |
| allowed = sorted({"AI_HYBRID", "DETERMINISTIC_CODE", *_STRATEGY_ALIASES}) | |
| raise InvalidRedactionStrategyError( | |
| f"Invalid redaction_strategy {raw!r}. " | |
| f"Allowed values: {', '.join(allowed)}." | |
| ) | |
| return resolved | |
| def merge_db_context(*parts: dict[str, str] | None) -> dict[str, str]: | |
| """Merge context dicts; later keys do not overwrite earlier non-empty values.""" | |
| merged: dict[str, str] = {} | |
| for part in parts: | |
| if not part: | |
| continue | |
| for key, value in part.items(): | |
| v = (value or "").strip() | |
| if v and key not in merged: | |
| merged[key] = v | |
| return merged | |
| def db_context_from_form_fields( | |
| *, | |
| client_name: str | None = None, | |
| property_address: str | None = None, | |
| surveyor_name: str | None = None, | |
| ) -> dict[str, str]: | |
| """Build a ``db_context`` fragment from optional multipart form fields.""" | |
| ctx: dict[str, str] = {} | |
| if client_name and client_name.strip(): | |
| ctx["client_name"] = client_name.strip() | |
| if property_address and property_address.strip(): | |
| ctx["property_address"] = property_address.strip() | |
| if surveyor_name and surveyor_name.strip(): | |
| ctx["surveyor_name"] = surveyor_name.strip() | |
| return ctx | |
| async def fetch_tenant_db_context( | |
| db: AsyncSession, | |
| tenant_id: str, | |
| ) -> dict[str, str]: | |
| """Best-effort PII context from the tenant's recent report section text. | |
| Uses :func:`app.services.generation._extract_property_identity` on the | |
| latest report sections so uploads can wipe names/addresses already present | |
| in the tenant's active reporting session without re-entering them manually. | |
| Args: | |
| db: Active async database session. | |
| tenant_id: Authenticated tenant identifier. | |
| Returns: | |
| A (possibly empty) ``{label: sensitive_value}`` map suitable for | |
| ``db_context`` in the deterministic redaction path. | |
| """ | |
| ctx: dict[str, str] = {} | |
| try: | |
| from sqlalchemy import select | |
| from app.db.models import Report, ReportSection, ReportStatus | |
| from app.services.generation import _extract_property_identity | |
| res = await db.execute( | |
| select(ReportSection.text) | |
| .join(Report, Report.id == ReportSection.report_id) | |
| .where(Report.tenant_id == tenant_id) | |
| .where( | |
| Report.status.in_( | |
| ( | |
| ReportStatus.pending, | |
| ReportStatus.generating, | |
| ReportStatus.complete, | |
| ReportStatus.partial, | |
| ) | |
| ) | |
| ) | |
| .order_by(Report.updated_at.desc()) | |
| .limit(24) | |
| ) | |
| lines = [row[0] for row in res.all() if row and row[0]] | |
| if not lines: | |
| return ctx | |
| identity = _extract_property_identity(lines) | |
| addr = (identity.get("address") or "").strip() | |
| pc = (identity.get("postcode") or "").strip() | |
| if addr: | |
| if pc and pc.upper() not in addr.upper(): | |
| addr = f"{addr}, {pc}" | |
| ctx["property_address"] = addr | |
| elif pc: | |
| ctx["property_address"] = pc | |
| if identity.get("surveyor_name"): | |
| ctx["surveyor_name"] = str(identity["surveyor_name"]).strip() | |
| if identity.get("firm"): | |
| ctx["company_name"] = str(identity["firm"]).strip() | |
| except Exception as exc: # noqa: BLE001 — context fetch must never block upload | |
| logger.debug("Tenant db_context backfill skipped tenant=%s: %s", tenant_id, exc) | |
| return ctx | |
| async def build_upload_db_context( | |
| db: AsyncSession, | |
| tenant_id: str, | |
| *, | |
| client_name: str | None = None, | |
| property_address: str | None = None, | |
| surveyor_name: str | None = None, | |
| ) -> dict[str, str]: | |
| """Merge upload form PII fields with tenant session backfill from reports.""" | |
| form_ctx = db_context_from_form_fields( | |
| client_name=client_name, | |
| property_address=property_address, | |
| surveyor_name=surveyor_name, | |
| ) | |
| tenant_ctx = await fetch_tenant_db_context(db, tenant_id) | |
| return merge_db_context(tenant_ctx, form_ctx) | |
| def serialize_db_context(ctx: dict[str, str] | None) -> str | None: | |
| """Persist ``db_context`` on a :class:`~app.db.models.Document` row.""" | |
| if not ctx: | |
| return None | |
| cleaned = {k: v.strip() for k, v in ctx.items() if k and (v or "").strip()} | |
| return json.dumps(cleaned, ensure_ascii=False) if cleaned else None | |
| def deserialize_db_context(raw: str | None) -> dict[str, str] | None: | |
| """Load persisted ``db_context`` JSON; returns ``None`` when absent/invalid.""" | |
| if not raw or not str(raw).strip(): | |
| return None | |
| try: | |
| data = json.loads(raw) | |
| except json.JSONDecodeError: | |
| logger.warning("Invalid redaction_context_json on document; ignoring.") | |
| return None | |
| if not isinstance(data, dict): | |
| return None | |
| return { | |
| str(k): str(v).strip() | |
| for k, v in data.items() | |
| if k and isinstance(v, (str, int, float)) and str(v).strip() | |
| } | |
| def load_document_redaction_strategy(raw: str | None) -> RedactionStrategy: | |
| """Resolve a stored document strategy string (never raises — defaults safely).""" | |
| try: | |
| return parse_redaction_strategy(raw) | |
| except InvalidRedactionStrategyError: | |
| logger.warning("Unknown stored redaction_strategy %r; defaulting to AI_HYBRID.", raw) | |
| return RedactionStrategy.AI_HYBRID | |