app/services/runtime_rag_index.py

"""Re-index user-edited section text into the vector store for immediate hierarchical RAG."""

from __future__ import annotations

import logging

from langchain_core.documents import Document

from app.config import settings
from app.ingest.hierarchy import build_hierarchical_documents
from app.ingest.pipeline import stamp_chunks_for_index
from app.vectorstore.factory import get_vectorstore

logger = logging.getLogger(__name__)


def runtime_section_vector_doc_id(report_id: str, section_code: str) -> str:
    """Stable virtual ``doc_id`` for a report section overlay (document + section + paragraph tiers)."""
    return f"runtime-{report_id}-{section_code}"


def upsert_runtime_section_index(
    *,
    tenant_id: str,
    virtual_doc_id: str,
    section_title: str,
    body_text: str,
) -> int:
    """Remove prior chunks for ``virtual_doc_id`` and insert fresh hierarchical rows.

    Returns:
        Number of chunks written (0 if ``body_text`` is empty after stripping).
    """
    vs = get_vectorstore()
    vs.delete_document(virtual_doc_id)
    text = (body_text or "").strip()
    if not text:
        logger.info("runtime RAG: cleared index for doc=%s", virtual_doc_id)
        return 0

    from app.services.document_sanitiser import sanitise_text_for_rag_sync

    text = sanitise_text_for_rag_sync(text, tenant_id=tenant_id)
    if not text:
        logger.info("runtime RAG: empty after sanitisation doc=%s", virtual_doc_id)
        return 0

    label = (section_title or "").strip() or virtual_doc_id
    raw = [Document(page_content=text, metadata={"source": label})]
    hier = build_hierarchical_documents(
        raw,
        doc_id=virtual_doc_id,
        filename=label,
        chunk_size=settings.chunk_size,
        chunk_overlap=settings.chunk_overlap,
    )
    stamped = stamp_chunks_for_index(hier, doc_id=virtual_doc_id, tenant_id=tenant_id)
    vs.add_documents(stamped)
    logger.info(
        "runtime RAG: indexed %d chunks for doc=%s tenant=%s",
        len(stamped),
        virtual_doc_id,
        tenant_id,
    )
    return len(stamped)


async def upsert_runtime_section_index_async(
    *,
    tenant_id: str,
    virtual_doc_id: str,
    section_title: str,
    body_text: str,
) -> int:
    """Non-blocking wrapper for :func:`upsert_runtime_section_index`."""
    from app.async_executor import run_sync_in_executor

    return await run_sync_in_executor(
        lambda: upsert_runtime_section_index(
            tenant_id=tenant_id,
            virtual_doc_id=virtual_doc_id,
            section_title=section_title,
            body_text=body_text,
        )
    )