Spaces:
Runtime error
Runtime error
File size: 6,376 Bytes
aad7814 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """Startup ingest: report template (schema) + standard paragraphs (MASTER RAG).
The operator bundle in ``Master Standard report and paragraphs/`` contains:
- PDF report template -> section structure (schema.json)
- Word standard paras -> approved wording (MASTER FAISS)
No user upload is required before the first report.
"""
from __future__ import annotations
import logging
import shutil
from backend.config import settings
from backend.core import ingest, notes_parser, pii_scrubber, template_discoverer
from backend.core.rics_canonical_l3 import PARENT_SECTION_COUNT
from backend.core.rag_store import TIER_MASTER, TIER_REFERENCE, get_rag_store
from backend.core.vector_index_guard import ensure_reference_indices_clean
from backend.utils import tenant_store
from backend.utils.runtime_paths import ensure_data_drive_runtime_dirs
logger = logging.getLogger(__name__)
def _scrub_stale_faiss_write_artifacts() -> None:
"""Remove half-written FAISS/meta files left by interrupted persists."""
tenants_root = settings.data_dir_path / "tenants"
if not tenants_root.is_dir():
return
for pattern in ("*.write.faiss", "*.write.json", "index.faiss.bad", "meta.json.bad"):
for path in tenants_root.rglob(pattern):
try:
path.unlink(missing_ok=True)
except OSError as exc:
logger.debug("Could not remove stale artifact %s: %s", path, exc)
def _load_prebuilt() -> bool:
"""Copy prebuilt schema.json + MASTER FAISS into the tenant store, if set."""
schema_src = settings.master_template_prebuilt_schema
faiss_src = settings.master_template_prebuilt_faiss
if not (schema_src and faiss_src):
return False
schema_path = settings.resolve_path(schema_src)
faiss_path = settings.resolve_path(faiss_src)
if not (schema_path.is_file() and faiss_path.is_dir()):
logger.warning("Prebuilt artifacts configured but missing; falling back to discovery.")
return False
tenant = settings.default_tenant_id
shutil.copyfile(schema_path, tenant_store.schema_path(tenant))
dest = tenant_store.faiss_dir(tenant, TIER_MASTER)
for item in faiss_path.iterdir():
shutil.copy2(item, dest / item.name)
logger.info("Loaded prebuilt master artifacts for tenant=%s", tenant)
return True
def run_startup_ingest() -> dict:
"""Ingest the operator bundle. Returns a summary for the health endpoint."""
ensure_data_drive_runtime_dirs()
_scrub_stale_faiss_write_artifacts()
tenant = settings.default_tenant_id
summary: dict = {
"master_loaded": False,
"sections": 0,
"paragraph_chunks": 0,
"report_template": settings.report_template_filename,
"standard_paragraphs": settings.standard_paragraphs_filename,
"reference_documents": 0,
}
if not settings.master_template_auto_ingest:
logger.info(
"master_template_auto_ingest disabled; reference-only generation. "
"Purging any persisted MASTER tier for tenant=%s.",
tenant,
)
template_discoverer.ensure_canonical_schema(tenant)
# Reference-only model: make removal real and idempotent so stale operator
# boilerplate from a prior master-enabled run cannot linger in the index.
get_rag_store().clear_tier(tenant, TIER_MASTER)
index_guard = ensure_reference_indices_clean()
summary["reference_index_guard"] = index_guard
summary["master_loaded"] = True # schema (canonical) is present
summary["sections"] = PARENT_SECTION_COUNT
summary["section_anchor_vectors"] = notes_parser.initialize_section_anchors()
if settings.reference_auto_ingest_enabled and not index_guard.get("rebuilt"):
try:
ref = ingest.auto_ingest_reference_dir(tenant)
summary["reference_documents"] = ref["documents"]
except Exception as exc: # noqa: BLE001
logger.warning("Reference auto-ingest failed: %s", exc)
return summary
loaded = False
if _load_prebuilt():
loaded = True
else:
try:
res = ingest.ingest_operator_bundle(tenant)
summary["sections"] = res["sections"]
summary["paragraph_chunks"] = res["paragraph_chunks"]
loaded = True
logger.info(
"Operator bundle ingested: schema from %s (%d sections), "
"paragraphs from %s (%d chunks)",
res["report_template"], res["sections"],
res["standard_paragraphs"], res["paragraph_chunks"],
)
except pii_scrubber.PiiDetectedError as exc:
logger.critical(
"STANDARD PARAGRAPHS REJECTED — contains property-identifying PII: %s", exc
)
except FileNotFoundError as exc:
logger.error("Operator bundle file missing: %s", exc)
except Exception as exc: # noqa: BLE001
logger.exception("Operator bundle ingest failed: %s", exc)
template_discoverer.ensure_canonical_schema(tenant)
index_guard = ensure_reference_indices_clean()
summary["reference_index_guard"] = index_guard
summary["master_loaded"] = loaded or template_discoverer.load_schema(tenant) is not None
summary["sections"] = PARENT_SECTION_COUNT
if settings.reference_auto_ingest_enabled and not index_guard.get("rebuilt"):
try:
ref = ingest.auto_ingest_reference_dir(tenant)
summary["reference_documents"] = ref["documents"]
except Exception as exc: # noqa: BLE001
logger.warning("Reference auto-ingest failed: %s", exc)
elif index_guard.get("rebuilt"):
summary["reference_documents"] = sum(
row.get("auto_ingest_docs", 0) + row.get("library_reingested", 0)
for row in index_guard.get("rebuilt", [])
)
anchor_count = notes_parser.initialize_section_anchors()
summary["section_anchor_vectors"] = anchor_count
store = get_rag_store()
logger.info(
"Startup complete: loaded=%s sections=%d paragraph_chunks=%d reference_chunks=%d anchors=%d",
summary["master_loaded"], summary["sections"],
store.count(tenant, TIER_MASTER), store.count(tenant, TIER_REFERENCE),
anchor_count,
)
return summary
|