app/services/style_corpus.py

"""Per-tenant Style Corpus — the user's own past reports, reused for voice/style only.

What this module does
=====================
Each tenant has a private library of documents. Two distinct purposes:

* ``report_source`` — factual input for a specific report job. These are the
  user's current RICS survey upload(s) and any reference docs they attached
  to *this* report. RAG retrieval pulls facts from here.

* ``style_corpus`` — the user's PAST completed reports, PII-scrubbed and
  indexed so the AI can imitate their voice on new jobs. RAG retrieval
  for **factual** evidence explicitly skips these (see
  ``app.retrieval.retriever._STYLE_CORPUS_EXCLUDE``). This module provides
  the corollary: helpers that read FROM the style corpus only, so the LLM
  can be fed past paragraphs as style examples without those past
  paragraphs polluting the new report's facts.

What this module deliberately does NOT do
=========================================
* Train OpenAI. The contract is "RAG + style learning", not fine-tuning.
* Mix tenants. Every helper is tenant-scoped at the retrieval layer.
* Bypass PII redaction. Style-corpus uploads always run through the
  sanitiser in ``app.services.document_sanitiser`` regardless of the
  per-tenant policy switch — see the ``force=True`` plumbing on the
  ingest path.
"""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING

from app.config import settings
from app.models.schemas import SearchResult, WritingStyleProfile
from app.retrieval.retriever import retrieve_async, retrieve_unified

if TYPE_CHECKING:
    pass

logger = logging.getLogger(__name__)


_STYLE_PURPOSE_SET: frozenset[str] = frozenset({"style_corpus"})

# Diverse queries used to seed style-corpus retrieval. We sample a broad
# spread of RICS-style content so the style profile / paragraph mirroring
# captures the user's voice across condition descriptions, defects, and
# recommendations — not just whichever section happens to be top of mind.
DEFAULT_STYLE_QUERIES: tuple[str, ...] = (
    "building condition survey inspection findings",
    "roof structure tiles condition recommendation",
    "main walls external condition damp penetration",
    "services electrical gas plumbing drainage condition",
    "property description construction type age",
    "chimney stack flashing pointing condition",
    "floors structure condition movement settlement",
    "windows doors frames condition repair",
    "recommendations next steps further investigation",
    "executive summary overall condition",
)


async def has_style_corpus(tenant_id: str) -> bool:
    """Cheap probe: does this tenant have any style-corpus chunks indexed yet?

    We do a tiny retrieval with the most generic survey query and a single
    purpose filter — enough to confirm at least one chunk exists. Empty
    style corpora are the default state for new tenants, so this check is
    on the hot path; it MUST be cheap.
    """
    try:
        hits = await retrieve_async(
            query="building survey",
            tenant_id=tenant_id,
            k=1,
            purpose_in=_STYLE_PURPOSE_SET,
        )
    except Exception as exc:  # noqa: BLE001
        logger.debug("style-corpus probe failed tenant=%s: %s", tenant_id, exc)
        return False
    return bool(hits)


async def retrieve_style_paragraphs(
    *,
    tenant_id: str,
    query: str,
    k: int = 6,
) -> list[SearchResult]:
    """Pull verbatim paragraphs ONLY from the tenant's style corpus.

    Used to inject few-shot style examples into generation prompts. The
    paragraphs returned are exact quotations from past reports — the LLM
    is expected to *mirror their voice*, not copy their content.
    """
    if not tenant_id:
        return []
    try:
        return await retrieve_async(
            query=query,
            tenant_id=tenant_id,
            k=k,
            purpose_in=_STYLE_PURPOSE_SET,
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning(
            "Style-corpus retrieval failed tenant=%s: %s", tenant_id, exc
        )
        return []


async def sample_style_corpus_texts(
    *,
    tenant_id: str,
    queries: tuple[str, ...] | None = None,
    per_query_k: int = 3,
    max_texts: int = 24,
) -> list[str]:
    """Sample a diverse slice of the tenant's past reports for style analysis.

    Combines results from several broad RICS queries to avoid biasing the
    style profile toward a single section. Deduplicates by text so a
    repeating paragraph (boilerplate) does not skew the analyser.
    """
    qs = queries if queries is not None else DEFAULT_STYLE_QUERIES
    out: list[str] = []
    seen: set[str] = set()
    for q in qs:
        if len(out) >= max_texts:
            break
        hits = await retrieve_style_paragraphs(
            tenant_id=tenant_id, query=q, k=per_query_k
        )
        for r in hits:
            text = (r.text or "").strip()
            if not text or text in seen:
                continue
            seen.add(text)
            out.append(text)
            if len(out) >= max_texts:
                break
    return out


async def build_style_corpus_profile(
    tenant_id: str,
) -> WritingStyleProfile | None:
    """Build a ``WritingStyleProfile`` from the tenant's style corpus only.

    Returns ``None`` when no style-corpus chunks exist; the caller should
    fall back to whichever profile source it already used (the broader
    tenant corpus, the KB reference profile, or the mock).
    """
    if not (settings.openai_api_key or "").strip():
        # No key → style analysis returns the mock profile; better to let
        # the caller fall back to its existing logic than to cache a mock
        # under the style-corpus key.
        return None
    texts = await sample_style_corpus_texts(tenant_id=tenant_id)
    if not texts:
        return None
    from app.generator.style_analyzer import build_reference_style_profile

    # ``build_reference_style_profile`` is the thorough variant — it asks the
    # LLM for verbatim example paragraphs, which is exactly what we want
    # for personal-style mirroring (the lightweight ``analyze_writing_style``
    # may skip example extraction).
    return await _call_in_thread(
        build_reference_style_profile,
        sample_texts=texts,
        openai_api_key=settings.openai_api_key,
        chat_model=settings.chat_model,
    )


async def _call_in_thread(fn, **kwargs):  # type: ignore[no-untyped-def]
    """Run a blocking helper in the executor so we never block the loop."""
    from app.async_executor import run_sync_in_executor

    return await run_sync_in_executor(fn, **kwargs)


def style_corpus_purpose_set() -> frozenset[str]:
    """Public accessor for the purpose-filter set (for callers that need the literal)."""
    return _STYLE_PURPOSE_SET


def _unused_imports_keep() -> None:
    """Keep retrieve_unified import live for forward-compat callers."""
    _ = retrieve_unified