"""Persist structured FDI extracts into SQL.""" from __future__ import annotations import json import re import uuid from datetime import datetime, timezone from typing import Any from app.database import get_connection from app.fdi.extract import StructuredExtract def _now() -> str: return datetime.now(timezone.utc).isoformat() def _uid() -> str: return str(uuid.uuid4()) def _norm_alias(name: str) -> str: return re.sub(r"\s+", " ", (name or "").strip().lower()) def upsert_entity(conn: Any, org_id: str, entity_type: str, name: str) -> str: """Find or create entity by alias; return entity_id.""" alias = _norm_alias(name) if not alias: raise ValueError("empty entity name") row = conn.execute( """ SELECT e.id FROM fdi_entities e JOIN fdi_entity_aliases a ON a.entity_id = e.id WHERE a.org_id = ? AND a.alias = ? LIMIT 1 """, (org_id, alias), ).fetchone() if row: return str(row["id"] if isinstance(row, dict) else row[0]) entity_id = _uid() now = _now() canonical = re.sub(r"\s+", " ", name.strip()) conn.execute( """ INSERT INTO fdi_entities(id, org_id, entity_type, canonical_name, attributes_json, created_at) VALUES (?, ?, ?, ?, ?, ?) """, (entity_id, org_id, entity_type, canonical, None, now), ) conn.execute( """ INSERT INTO fdi_entity_aliases(id, org_id, entity_id, alias, source, weight, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) """, (_uid(), org_id, entity_id, alias, "extract", 1.0, now), ) return entity_id def delete_document_data(conn: Any, document_id: str) -> None: conn.execute("DELETE FROM fdi_line_items WHERE document_id = ?", (document_id,)) conn.execute("DELETE FROM fdi_identifiers WHERE document_id = ?", (document_id,)) conn.execute("DELETE FROM fdi_assertions WHERE document_id = ?", (document_id,)) conn.execute("DELETE FROM fdi_entity_links WHERE document_id = ?", (document_id,)) def persist_structured( *, org_id: str, agent_id: str, kb_doc_id: str, filename: str, storage_uri: str | None, sha256: str | None, mime_type: str | None, extract: StructuredExtract, ) -> str: """Replace any prior FDI rows for this kb_doc and write a fresh structured index.""" now = _now() with get_connection() as conn: existing = conn.execute( "SELECT id FROM fdi_documents WHERE kb_doc_id = ? LIMIT 1", (kb_doc_id,), ).fetchone() doc_id = str(existing["id"] if isinstance(existing, dict) else existing[0]) if existing else _uid() if existing: delete_document_data(conn, doc_id) conn.execute( """ UPDATE fdi_documents SET filename=?, storage_uri=?, sha256=?, mime_type=?, doc_type=?, doc_type_confidence=?, status=?, page_count=?, ocr_used=?, parse_confidence=?, currency_primary=?, period_start=?, period_end=?, issuer_name=?, holder_name=?, source_type=?, error=?, indexed_at=? WHERE id=? """, ( filename, storage_uri, sha256, mime_type, extract.doc_type, extract.doc_type_confidence, "ready", extract.page_count, 1 if extract.ocr_used else 0, extract.parse_confidence, extract.currency_primary, extract.period_start, extract.period_end, extract.issuer_name, extract.holder_name, extract.source_type, "; ".join(extract.warnings)[:2000] if extract.warnings else None, now, doc_id, ), ) else: conn.execute( """ INSERT INTO fdi_documents( id, org_id, agent_id, kb_doc_id, filename, storage_uri, sha256, mime_type, doc_type, doc_type_confidence, status, page_count, ocr_used, parse_confidence, currency_primary, period_start, period_end, issuer_name, holder_name, source_type, error, created_at, indexed_at ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( doc_id, org_id, agent_id, kb_doc_id, filename, storage_uri, sha256, mime_type, extract.doc_type, extract.doc_type_confidence, "ready", extract.page_count, 1 if extract.ocr_used else 0, extract.parse_confidence, extract.currency_primary, extract.period_start, extract.period_end, extract.issuer_name, extract.holder_name, extract.source_type, "; ".join(extract.warnings)[:2000] if extract.warnings else None, now, now, ), ) # Parties / entities name_to_entity: dict[str, str] = {} for etype, name in extract.parties: key = _norm_alias(name) if not key: continue if key not in name_to_entity: name_to_entity[key] = upsert_entity(conn, org_id, etype, name) for line in extract.lines: cp_id = None if line.counterparty_name: cp_key = _norm_alias(line.counterparty_name) if cp_key not in name_to_entity: name_to_entity[cp_key] = upsert_entity(conn, org_id, "merchant", line.counterparty_name) cp_id = name_to_entity.get(cp_key) conn.execute( """ INSERT INTO fdi_line_items( id, org_id, document_id, page_no, row_idx, event_date, posting_date, description_raw, description_norm, amount, debit, credit, balance_after, direction, currency, counterparty_entity_id, account_entity_id, category, event_class, extras_json, confidence, created_at ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( _uid(), org_id, doc_id, line.page_no, line.row_idx, line.event_date, None, line.description_raw, line.description_norm, line.amount, line.debit, line.credit, line.balance_after, line.direction, line.currency, cp_id, None, line.category, line.event_class, json.dumps(line.extras) if line.extras else None, line.confidence, now, ), ) for assertion in extract.assertions: conn.execute( """ INSERT INTO fdi_assertions( id, org_id, document_id, kind, label, value_numeric, value_text, unit, currency, source, confidence, evidence_json, created_at ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( _uid(), org_id, doc_id, assertion.kind, assertion.label, assertion.value_numeric, assertion.value_text, assertion.unit, assertion.currency, assertion.source, assertion.confidence, None, now, ), ) for ident in extract.identifiers: conn.execute( """ INSERT INTO fdi_identifiers( id, org_id, document_id, line_item_id, id_type, value_raw, value_norm, confidence, created_at ) VALUES (?,?,?,?,?,?,?,?,?) """, ( _uid(), org_id, doc_id, None, ident.id_type, ident.value_raw, ident.value_norm, ident.confidence, now, ), ) # Entity alias seeds + counterparty relink (outside txn connection scope above) try: from app.fdi.entities import ensure_seed_aliases, relink_line_counterparties ensure_seed_aliases(org_id) relink_line_counterparties(org_id, doc_id) except Exception: pass return doc_id