app/services/document_sanitiser.py

"""LLM + regex sanitisation before tenant uploads are indexed for RAG / style learning.

On-disk uploads are unchanged; only text written to the vector index is sanitised.
"""

from __future__ import annotations

import logging
import re
from typing import TYPE_CHECKING

from app.config import settings

if TYPE_CHECKING:
    from langchain_core.documents import Document

logger = logging.getLogger(__name__)


class DocumentSanitisationError(RuntimeError):
    """Raised when sanitisation is required but cannot complete safely."""


RAG_UPLOAD_SANITISATION_SYSTEM_PROMPT = """\
You are a document sanitisation assistant for a professional property report-writing platform.

The purpose of this process is to prepare users' previously written reports for storage in a private user-specific RAG system. These reports will be used solely to learn the user's preferred writing style, report structure, technical reasoning, and professional wording.

Your task is to remove all personal, confidential, identifying, financial, legal, and transaction-specific information while preserving the technical content, report format, headings, narrative style, and professional observations.

IMPORTANT INSTRUCTIONS

- Preserve the report's structure, headings, section order, and formatting wherever possible.
- Preserve the user's writing style, sentence structure, tone, and terminology.
- Preserve all technical observations, building pathology descriptions, assumptions, caveats, and recommendations.
- Preserve generic market commentary and valuation reasoning.
- Preserve standard boilerplate text and report templates.
- Remove sensitive information entirely rather than replacing it with placeholders.
- Do not summarise, rewrite, shorten, or paraphrase unless necessary to remove identifying information.
- Do not invent any new information.
- Ensure that the resulting document contains no information that could identify a specific individual, property, company, or transaction.

REMOVE THE FOLLOWING INFORMATION

PERSONAL INFORMATION
- Client names
- Occupier names
- Tenant names
- Vendor names
- Purchaser names
- Solicitor names
- Estate agent names
- Surveyor names
- Witness names
- Signatures
- Initials where they identify a person
- Dates of birth

CONTACT INFORMATION
- Email addresses
- Telephone numbers
- Mobile numbers
- Fax numbers
- Website URLs where they identify a specific business or individual

ADDRESS INFORMATION
- Full property addresses
- Flat numbers
- House numbers
- Building names
- Street names
- Towns and cities where they identify the subject property
- Postcodes
- Correspondence addresses

PROPERTY IDENTIFIERS
- Land Registry title numbers
- UPRNs
- EPC certificate numbers
- Planning application numbers
- Building regulation references
- Local authority references
- Council tax account numbers

PROFESSIONAL IDENTIFIERS
- RICS membership numbers
- Registration numbers
- Company registration numbers
- VAT numbers
- Insurance policy numbers
- Certificate numbers
- Job numbers
- File references
- Invoice numbers

FINANCIAL INFORMATION
- Valuation figures
- Purchase prices
- Sale prices
- Premiums
- Mortgage balances
- Ground rent amounts
- Service charge amounts
- Insurance claim amounts
- Bank account details
- Sort codes
- IBAN numbers

LEGAL INFORMATION
- Tribunal case numbers
- Court claim numbers
- Lease numbers
- Policy claim references
- Solicitor reference numbers

DATES
- Inspection dates
- Report dates
- Exchange dates
- Completion dates
- Any specific dates linked to an identifiable matter or transaction

DIGITAL IDENTIFIERS
- IP addresses
- GPS coordinates
- Metadata references
- QR code content

SPECIAL CATEGORY OR CONFIDENTIAL INFORMATION
- Medical or health information
- Disability information
- Family circumstances
- Complaints
- Litigation details
- Insurance disputes
- Employment information
- Any other confidential notes

OUTPUT REQUIREMENTS

- Return only the sanitised report text.
- Preserve as much useful stylistic and technical content as possible.
- Remove all identifying and confidential information.
- Ensure the resulting text is suitable for storage in a private user-specific RAG system for personalised AI report generation.\
"""

_POSTCODE_RE = re.compile(
    r"\b([A-Z]{1,2}\d[A-Z\d]?\s*\d[A-Z]{2})\b", re.IGNORECASE
)
_ADDRESS_LINE_RE = re.compile(
    r"\b(\d{1,4}\s+[A-Za-z][A-Za-z'\-]*(?:\s+[A-Za-z][A-Za-z'\-]*){0,6}\s+"
    r"(Road|Rd|Street|St|Avenue|Ave|Lane|Ln|Drive|Dr|Crescent|Close|Place|Way|Gardens|Gdns|Court|Ct|Terrace|Terr))\b",
    re.IGNORECASE,
)
_MONEY_RE = re.compile(
    r"(£\s*\d[\d,]*(?:\.\d+)?|\b\d[\d,]*(?:\.\d+)?\s*(?:gbp|pounds)\b)",
    re.IGNORECASE,
)
_DATE_RE = re.compile(
    r"\b(?:\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|\d{1,2}\s+"
    r"(?:jan|feb|mar|apr|may|jun|jul|aug|sep|sept|oct|nov|dec)[a-z]*\s+\d{2,4})\b",
    re.IGNORECASE,
)
_LONG_NUMBER_RE = re.compile(r"\b\d{5,}\b")
_EMAIL_RE = re.compile(r"\b[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}\b")
_PHONE_RE = re.compile(
    r"\b(?:\+44\s?|0)(?:\d[\s-]?){9,12}\b|\b\d{3,4}[\s-]\d{3,4}[\s-]\d{3,4}\b"
)
_URL_RE = re.compile(r"https?://[^\s<>\"']+", re.IGNORECASE)


def should_sanitise_for_rag(tenant_id: str) -> bool:
    """Whether uploads for this tenant should be sanitised before vector indexing."""
    if not settings.enable_rag_upload_sanitisation:
        return False
    if (
        settings.rag_sanitisation_skip_kb_tenant
        and tenant_id == settings.knowledge_base_tenant_id
    ):
        return False
    return True


def regex_sanitise_text(text: str) -> str:
    """Deterministic redaction when the LLM path is unavailable or as a safety net."""
    t = (text or "").strip()
    if not t:
        return ""
    t = _EMAIL_RE.sub("", t)
    t = _URL_RE.sub("", t)
    t = _PHONE_RE.sub("", t)
    t = _POSTCODE_RE.sub("", t)
    t = _ADDRESS_LINE_RE.sub("", t)
    t = _MONEY_RE.sub("", t)
    t = _DATE_RE.sub("", t)
    t = _LONG_NUMBER_RE.sub("", t)
    t = re.sub(r"[ \t]{2,}", " ", t)
    t = re.sub(r"\n{3,}", "\n\n", t)
    return t.strip()


def _split_text_chunks(text: str, max_chars: int) -> list[str]:
    body = (text or "").strip()
    if not body:
        return []
    if len(body) <= max_chars:
        return [body]
    parts = re.split(r"(\n\s*\n)", body)
    chunks: list[str] = []
    current = ""
    for part in parts:
        if not part:
            continue
        candidate = current + part
        if len(candidate) <= max_chars:
            current = candidate
            continue
        if current.strip():
            chunks.append(current.strip())
        if len(part) <= max_chars:
            current = part
        else:
            for i in range(0, len(part), max_chars):
                segment = part[i : i + max_chars].strip()
                if segment:
                    chunks.append(segment)
            current = ""
    if current.strip():
        chunks.append(current.strip())
    return chunks or [body[:max_chars]]


def _llm_sanitise_chunk_sync(chunk: str, *, part: int, total: int) -> str:
    from app.llm.openai_chat import chat_completions_create

    import asyncio

    prefix = (
        f"Sanitise this excerpt from a RICS property survey report "
        f"(part {part} of {total}).\n\n"
    )
    user_content = prefix + chunk

    async def _run() -> str:
        return await chat_completions_create(
            messages=[
                {"role": "system", "content": RAG_UPLOAD_SANITISATION_SYSTEM_PROMPT},
                {"role": "user", "content": user_content},
            ],
            model=settings.chat_model,
            max_tokens=int(settings.rag_sanitisation_max_output_tokens),
            temperature=0.0,
            phase="rag_upload_sanitise",
        )

    try:
        return asyncio.run(_run())
    except RuntimeError:
        # Already inside a running loop (should not happen from ingest thread).
        import concurrent.futures

        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
            return pool.submit(asyncio.run, _run()).result()


def sanitise_text_for_rag_sync(text: str, *, tenant_id: str) -> str:
    """Sanitise plain text before it is embedded into the tenant vector index."""
    if not should_sanitise_for_rag(tenant_id):
        return (text or "").strip()

    raw = (text or "").strip()
    if not raw:
        return ""

    max_chars = int(settings.rag_sanitisation_chunk_chars)
    chunks = _split_text_chunks(raw, max_chars)
    has_key = bool((settings.openai_api_key or "").strip())
    out_parts: list[str] = []

    use_llm = bool(settings.rag_sanitisation_use_llm)
    for i, chunk in enumerate(chunks, start=1):
        cleaned = ""
        if has_key and use_llm:
            try:
                cleaned = (_llm_sanitise_chunk_sync(chunk, part=i, total=len(chunks)) or "").strip()
            except Exception as exc:
                logger.warning(
                    "RAG sanitisation LLM failed tenant=%s part=%d/%d: %s",
                    tenant_id,
                    i,
                    len(chunks),
                    exc,
                )
        if not cleaned:
            cleaned = regex_sanitise_text(chunk)
            if has_key:
                logger.info(
                    "RAG sanitisation using regex fallback tenant=%s part=%d/%d",
                    tenant_id,
                    i,
                    len(chunks),
                )
        if not cleaned and settings.rag_sanitisation_fail_closed:
            raise DocumentSanitisationError(
                f"Sanitisation produced empty text for tenant={tenant_id} part={i}/{len(chunks)}"
            )
        out_parts.append(cleaned)

    combined = "\n\n".join(p for p in out_parts if p).strip()
    if not combined and settings.rag_sanitisation_fail_closed:
        raise DocumentSanitisationError(
            f"Sanitisation produced empty document for tenant={tenant_id}"
        )
    return combined or regex_sanitise_text(raw)


def sanitise_langchain_documents_sync(
    documents: list[Document],
    *,
    tenant_id: str,
) -> list[Document]:
    """Return documents whose ``page_content`` has been sanitised for RAG indexing."""
    if not should_sanitise_for_rag(tenant_id):
        return documents

    from langchain_core.documents import Document as LCDocument

    out: list[LCDocument] = []
    for doc in documents:
        meta = dict(doc.metadata or {})
        cleaned = sanitise_text_for_rag_sync(doc.page_content or "", tenant_id=tenant_id)
        meta["rag_sanitised"] = True
        out.append(LCDocument(page_content=cleaned, metadata=meta))
    return out