app/services/knowledge_base.py

"""Local standards knowledge base ingestion.

Indexes PDFs/DOCX from configured folders into the vector store under a reserved
tenant id. This allows the agentic pipeline to cite RICS standards/exemplar
material without requiring uploads through the UI.

Files are NOT committed to git; the ingestion simply reads from disk.
"""

from __future__ import annotations

import hashlib
import logging
from pathlib import Path

from langchain_core.documents import Document

from app.config import settings
from app.ingest.hierarchy import build_hierarchical_documents
from app.ingest.pipeline import load_raw_documents, stamp_chunks_for_index
from app.retrieval.chunk_role import classify_chunk_role
from app.vectorstore.factory import get_vectorstore

logger = logging.getLogger(__name__)


def _stable_kb_doc_id(path: Path) -> str:
    # Stable across restarts; changes if file path changes.
    h = hashlib.sha1(str(path).encode("utf-8"), usedforsecurity=False).hexdigest()[:20]
    return f"kb-{h}"


def iter_knowledge_base_roots() -> list[Path]:
    """Resolved directory (or single-file parent) roots listed in ``knowledge_base_dirs``."""
    raw = (settings.knowledge_base_dirs or "").strip()
    if not raw:
        return []
    parts = [p.strip() for p in raw.split(",") if p.strip()]
    root = Path(__file__).resolve().parents[2]  # repo root
    out: list[Path] = []
    for part in parts:
        p = Path(part)
        if not p.is_absolute():
            p = root / p
        if not p.exists():
            continue
        out.append(p if p.is_dir() else p.parent)
    # de-dup while preserving order
    seen: set[str] = set()
    uniq: list[Path] = []
    for p in out:
        k = str(p.resolve()).lower()
        if k in seen:
            continue
        seen.add(k)
        uniq.append(p)
    return uniq


def iter_kb_files() -> list[Path]:
    """List all candidate KB documents from configured dirs."""
    raw = (settings.knowledge_base_dirs or "").strip()
    if not raw:
        return []
    parts = [p.strip() for p in raw.split(",") if p.strip()]
    root = Path(__file__).resolve().parents[2]  # repo root
    out: list[Path] = []
    for part in parts:
        p = Path(part)
        if not p.is_absolute():
            p = root / p
        if not p.exists():
            continue
        if p.is_file():
            out.append(p)
            continue
        for ext in (".pdf", ".docx"):
            out.extend(sorted(p.rglob(f"*{ext}")))
    # de-dup
    seen: set[str] = set()
    uniq: list[Path] = []
    for p in out:
        k = str(p.resolve()).lower()
        if k in seen:
            continue
        seen.add(k)
        uniq.append(p)
    return uniq


def upsert_knowledge_base() -> dict[str, int]:
    """(Re)index all KB files into the vector store.

    Returns counts for observability; safe to call at startup.
    """
    if not settings.knowledge_base_enabled:
        return {"files": 0, "chunks": 0, "skipped": 0}

    files = iter_kb_files()
    if not files:
        logger.info("KB ingest: no knowledge base files found")
        return {"files": 0, "chunks": 0, "skipped": 0}

    vs = get_vectorstore()
    tenant_id = settings.knowledge_base_tenant_id
    chunks_total = 0
    skipped = 0

    for fp in files:
        suf = fp.suffix.lower()
        if suf not in (".pdf", ".docx"):
            skipped += 1
            continue
        doc_id = _stable_kb_doc_id(fp)
        try:
            # Best-effort upsert: delete prior chunks for this kb doc_id.
            vs.delete_document(doc_id)

            raw_docs = load_raw_documents(fp)
            # Ensure a reasonable source label even without DB Document rows.
            label = fp.name
            for d in raw_docs:
                d.metadata = {**(d.metadata or {}), "source": label}

            hier = build_hierarchical_documents(
                raw_docs,
                doc_id=doc_id,
                filename=label,
                chunk_size=settings.chunk_size,
                chunk_overlap=settings.chunk_overlap,
            )
            stamped = stamp_chunks_for_index(hier, doc_id=doc_id, tenant_id=tenant_id)
            # Mark provenance as knowledge base for downstream agents and
            # classify each chunk so retrieval can bias toward boilerplate at
            # assembly tier.
            for d in stamped:
                d.metadata["kb"] = True
                d.metadata["kb_path"] = str(fp)
                d.metadata["chunk_role"] = classify_chunk_role(d.page_content or "")
            vs.add_documents(stamped)
            chunks_total += len(stamped)
        except Exception as exc:  # noqa: BLE001
            logger.warning("KB ingest failed for %s: %s", fp, exc)
            skipped += 1

    if chunks_total:
        import asyncio

        from app.retrieval.semantic_cache import invalidate_semantic_cache_for_tenant

        try:
            asyncio.get_running_loop()
            pending = True
        except RuntimeError:
            pending = False
        if pending:
            asyncio.create_task(invalidate_semantic_cache_for_tenant(tenant_id))
        else:
            asyncio.run(invalidate_semantic_cache_for_tenant(tenant_id))

    logger.info("KB ingest complete: files=%d chunks=%d skipped=%d", len(files), chunks_total, skipped)
    return {"files": len(files), "chunks": chunks_total, "skipped": skipped}