Spaces:
Runtime error
Runtime error
File size: 4,869 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 | """FastAPI application for the template-agnostic v2 backend."""
from __future__ import annotations
import json
import logging
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from backend.config import settings
from backend.core import pii_scrubber, template_discoverer
from backend.core.rag_store import TIER_MASTER, TIER_REFERENCE, get_rag_store
from backend.startup import run_startup_ingest
from backend.utils.runtime_paths import ensure_data_drive_runtime_dirs
ensure_data_drive_runtime_dirs()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
_FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
_STARTUP_SUMMARY: dict = {}
_PII_SCAN_PATHS = {"/api/report/generate", "/api/report/preview"}
@asynccontextmanager
async def lifespan(app: FastAPI):
global _STARTUP_SUMMARY
_STARTUP_SUMMARY = run_startup_ingest()
from backend.core import document_library
recovered = document_library.recover_all_tenants_stale_processing()
if recovered:
logger.info("Startup recovered %d stale processing document(s)", recovered)
yield
app = FastAPI(title="RICS Template-Agnostic Backend (v2)", version="2.0.0", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
@app.middleware("http")
async def pii_request_middleware(request: Request, call_next):
"""Log PII in inbound notes for awareness (notes are not scrubbed; RAG docs are)."""
if request.method == "POST" and request.url.path in _PII_SCAN_PATHS:
body = await request.body()
async def _receive():
return {"type": "http.request", "body": body, "more_body": False}
request._receive = _receive # re-inject consumed body
try:
raw_notes = json.loads(body or b"{}").get("raw_notes", "")
audit = pii_scrubber.detect_pii(raw_notes)
flagged = {k: v for k, v in audit.items() if k in ("EMAIL", "POSTCODE")}
if flagged:
logger.warning(
"Inbound notes contain PII (notes kept verbatim; scrub RAG uploads only): %s",
flagged,
)
except Exception: # noqa: BLE001
pass
response = await call_next(request)
return response
# Routers
from backend.api.routes import admin, auth, compat, photos, report, schema, upload # noqa: E402
app.include_router(auth.router)
app.include_router(compat.router)
app.include_router(schema.router)
app.include_router(report.router)
app.include_router(photos.router)
app.include_router(upload.router)
if settings.master_template_upload_enabled:
app.include_router(admin.router)
@app.get("/health", tags=["health"])
def health() -> dict:
tenant = settings.default_tenant_id
loaded = template_discoverer.load_schema(tenant) is not None
store = get_rag_store()
master_faiss_count = store.count(tenant, TIER_MASTER)
reference_faiss_count = store.count(tenant, TIER_REFERENCE)
# Reference-only model: readiness depends on the (canonical) schema being
# installed, not on a shared master being seeded. When the operator master is
# explicitly enabled, its chunks must be present for an "ok" status.
master_ready = (not settings.master_template_auto_ingest) or master_faiss_count > 0
return {
"status": "ok" if loaded and master_ready else "degraded",
"master_loaded": loaded,
"master_faiss_count": master_faiss_count,
"reference_faiss_count": reference_faiss_count,
"reference_ready": reference_faiss_count > 0,
"startup": _STARTUP_SUMMARY,
"admin_override_enabled": settings.master_template_upload_enabled,
"ui_path": "/",
}
if _FRONTEND_DIR.is_dir():
app.mount("/static", StaticFiles(directory=str(_FRONTEND_DIR)), name="static")
@app.get("/", include_in_schema=False)
def root_ui() -> FileResponse:
index = _FRONTEND_DIR / "index.html"
if index.is_file():
return FileResponse(index)
return FileResponse(_FRONTEND_DIR / "v2.html")
@app.get("/v2.html", include_in_schema=False)
def v2_ui() -> FileResponse:
return FileResponse(_FRONTEND_DIR / "v2.html")
@app.get("/index.html", include_in_schema=False)
def legacy_index() -> FileResponse:
return FileResponse(_FRONTEND_DIR / "index.html")
@app.get("/favicon.ico", include_in_schema=False)
def favicon() -> FileResponse:
path = _FRONTEND_DIR / "favicon.ico"
if path.is_file():
return FileResponse(path)
return FileResponse(_FRONTEND_DIR / "v2.html")
|