from datetime import datetime from pathlib import Path import re import threading import time from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile, status from fastapi.responses import FileResponse from app.api.deps import require_session from app.models.fiscal_documents import ( FiscalDocumentLineItemRead, FiscalDocumentOrderMatchLineRead, FiscalDocumentOrderMatchRead, FiscalDocumentInboxItemRead, FiscalDocumentInboxListResponse, FiscalDocumentInboxStatusRead, FiscalDocumentNotificationSummary, FiscalDocumentRead, FiscalDocumentsListResponse, FiscalDocumentSettingsRead, FiscalDocumentSettingsUpdatePayload, FiscalDocumentV2ReviewPayload, ) from app.services.fiscal_engine_v2_client import ( FiscalEngineV2Error, analyze_fiscal_document_v2, fiscal_engine_v2_enabled, get_fiscal_document_v2_analysis, record_fiscal_document_v2_application, review_fiscal_document_v2, ) from app.services.fiscal_document_inbox_service import sync_fiscal_document_inbox from app.services.google_workspace_client import is_google_workspace_configured from app.services.google_workspace_store import get_google_workspace_store from app.services.fiscal_document_service import ( apply_delivery_note_storno_to_matched_order, build_fiscal_document_order_match, ingest_fiscal_document, persist_fiscal_engine_v2_analysis, ) from app.services.tenant_store import ( FiscalDocumentInboxItemRecord, FiscalDocumentLineItemRecord, FiscalDocumentRecord, SessionIdentity, get_tenant_store, ) router = APIRouter() _SIMPLE_EMAIL_PATTERN = re.compile(r"^[^\s@]+@[^\s@]+\.[^\s@]+$") _GMAIL_READONLY_SCOPE = "https://www.googleapis.com/auth/gmail.readonly" _ORDER_RECONCILABLE_DOCUMENT_TYPES = {"delivery_note", "instant_invoice"} _DISCREPANCY_CACHE_TTL_SECONDS = 300.0 _DISCREPANCY_CACHE_LOCK = threading.Lock() _DISCREPANCY_COMPUTE_LOCK = threading.Lock() _DISCREPANCY_CACHE: dict[tuple[str, str], tuple[float, str, bool]] = {} _REFERENCED_DDT_PATTERN = re.compile( r"(?:d\.?\s*d\.?\s*t\.?|documento\s+di\s+trasporto)\s*(?:n\.?|num(?:ero)?\.?)?\s*([A-Z0-9./-]+)(?:\s+del\s+(\d{1,2}[./-]\d{1,2}[./-]\d{2,4}))?", re.IGNORECASE, ) def _require_tenant_admin(session: SessionIdentity) -> SessionIdentity: if not get_tenant_store().session_has_permission(session, "fiscal_documents"): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Questo account non puo accedere a Documenti fiscali.", ) return session def _serialize_fiscal_document_line_item(record: FiscalDocumentLineItemRecord) -> FiscalDocumentLineItemRead: return FiscalDocumentLineItemRead( line_index=record.line_index, product_code=record.product_code, iso_code=record.iso_code, description=record.description, category_code=record.category_code, unit_code=record.unit_code, pack_count=record.pack_count, quantity=record.quantity, gross_quantity=record.gross_quantity, tare_quantity=record.tare_quantity, net_quantity=record.net_quantity, unit_price=record.unit_price, line_total=record.line_total, vat_code=record.vat_code, raw_row_text=record.raw_row_text, ) def _fiscal_document_item_order_match_payload(item: FiscalDocumentLineItemRead | FiscalDocumentLineItemRecord) -> dict[str, object]: return { "line_index": item.line_index, "description": item.description, "raw_row_text": item.raw_row_text, "pack_count": item.pack_count, "quantity": item.quantity, "gross_quantity": item.gross_quantity, "tare_quantity": item.tare_quantity, "net_quantity": item.net_quantity, "unit_code": item.unit_code, "category_code": item.category_code, "product_code": item.product_code, "unit_price": item.unit_price, "line_total": item.line_total, "vat_code": item.vat_code, } def _serialize_order_match_line(line: dict[str, object]) -> FiscalDocumentOrderMatchLineRead: def numeric_value(key: str) -> float | None: value = line.get(key) if isinstance(value, (int, float)) and not isinstance(value, bool): return float(value) return None return FiscalDocumentOrderMatchLineRead( status=str(line.get("status") or "partial"), # type: ignore[arg-type] confidence=float(line.get("confidence") or 0.0), order_item_id=int(line["order_item_id"]) if isinstance(line.get("order_item_id"), int) else None, order_product_name=str(line.get("order_product_name") or "") or None, order_lot_code=str(line.get("order_lot_code") or "") or None, order_supplier_name=str(line.get("order_supplier_name") or "") or None, ordered_quantity=numeric_value("ordered_quantity"), delivered_quantity=numeric_value("delivered_quantity"), comparable_delivered_quantity=numeric_value("comparable_delivered_quantity"), missing_quantity=numeric_value("missing_quantity"), extra_quantity=numeric_value("extra_quantity"), document_line_index=int(line["document_line_index"]) if isinstance(line.get("document_line_index"), int) else None, document_description=str(line.get("document_description") or "") or None, document_raw_row_text=str(line.get("document_raw_row_text") or "") or None, document_unit_code=str(line.get("document_unit_code") or "") or None, document_quantity=numeric_value("document_quantity"), ) def _serialize_order_match(match: dict[str, object] | None) -> FiscalDocumentOrderMatchRead | None: if not isinstance(match, dict): return None return FiscalDocumentOrderMatchRead( status=str(match.get("status") or "no_candidate"), # type: ignore[arg-type] matched_batch_id=int(match["matched_batch_id"]) if isinstance(match.get("matched_batch_id"), int) else None, matched_batch_confirmed_at=str(match.get("matched_batch_confirmed_at") or "") or None, matched_batch_staff=str(match.get("matched_batch_staff") or "") or None, matched_supplier_name=str(match.get("matched_supplier_name") or "") or None, score=float(match.get("score") or 0.0), line_match_count=int(match.get("line_match_count") or 0), exact_line_count=int(match.get("exact_line_count") or 0), missing_line_count=int(match.get("missing_line_count") or 0), extra_line_count=int(match.get("extra_line_count") or 0), can_apply_storno=bool(match.get("can_apply_storno")), lines=[_serialize_order_match_line(line) for line in match.get("lines", []) if isinstance(line, dict)], ) def _normalize_document_number(value: str | None) -> str: return re.sub(r"[^a-z0-9]+", "", (value or "").casefold()) def _normalize_supplier_key(value: str | None) -> str: normalized = (value or "").casefold() normalized = re.sub(r"\b(?:srl|s\.r\.l\.|snc|s\.n\.c\.|sas|s\.a\.s\.|spa|s\.p\.a\.)\b", " ", normalized) return re.sub(r"[^a-z0-9]+", "", normalized) def _parse_reference_date(value: str | None) -> str | None: raw_value = (value or "").strip() if not raw_value: return None for pattern in ("%d/%m/%Y", "%d-%m-%Y", "%d.%m.%Y", "%d/%m/%y", "%d-%m-%y", "%d.%m.%y"): try: return datetime.strptime(raw_value, pattern).date().isoformat() except ValueError: continue return None def _extract_referenced_delivery_note(document: FiscalDocumentRecord) -> tuple[str | None, str | None]: if document.document_type != "instant_invoice": return None, None searchable = f"{document.preview_text}\n{document.extracted_text}" for match in _REFERENCED_DDT_PATTERN.finditer(searchable): number = _normalize_document_number(match.group(1)) if not number: continue return number, _parse_reference_date(match.group(2)) or document.document_date return None, None def _find_covering_delivery_note(document: FiscalDocumentRecord) -> FiscalDocumentRecord | None: referenced_number, referenced_date = _extract_referenced_delivery_note(document) if not referenced_number: return None supplier_key = _normalize_supplier_key(document.supplier_name) for candidate in get_tenant_store().list_fiscal_documents(document.tenant_id): if candidate.id == document.id or candidate.document_type != "delivery_note": continue if supplier_key and _normalize_supplier_key(candidate.supplier_name) != supplier_key: continue if _normalize_document_number(candidate.document_number) != referenced_number: continue if referenced_date and candidate.document_date and candidate.document_date != referenced_date: continue return candidate return None def _discrepancy_cache_signature(record: FiscalDocumentRecord) -> str: return "|".join( [ record.updated_at or "", record.status or "", record.review_status or "", record.matching_status or "", record.document_type or "", record.document_date or "", record.document_number or "", record.supplier_name or "", ] ) def _cached_document_has_open_order_discrepancy( session: SessionIdentity, record: FiscalDocumentRecord, *, covering_delivery_note: FiscalDocumentRecord | None = None, ) -> bool: key = (record.tenant_id, record.id) signature = _discrepancy_cache_signature(record) now = time.monotonic() stale_value: bool | None = None with _DISCREPANCY_CACHE_LOCK: cached = _DISCREPANCY_CACHE.get(key) if cached is not None and cached[1] == signature: stale_value = cached[2] if now - cached[0] <= _DISCREPANCY_CACHE_TTL_SECONDS: return cached[2] # Avoid stacking expensive fuzzy order matches from sidebar polling. if not _DISCREPANCY_COMPUTE_LOCK.acquire(blocking=False): return stale_value if stale_value is not None else False try: with _DISCREPANCY_CACHE_LOCK: cached = _DISCREPANCY_CACHE.get(key) if ( cached is not None and cached[1] == signature and time.monotonic() - cached[0] <= _DISCREPANCY_CACHE_TTL_SECONDS ): return cached[2] has_discrepancy = _document_has_open_order_discrepancy( session, record, covering_delivery_note=covering_delivery_note, ) with _DISCREPANCY_CACHE_LOCK: _DISCREPANCY_CACHE[key] = (time.monotonic(), signature, has_discrepancy) return has_discrepancy finally: _DISCREPANCY_COMPUTE_LOCK.release() def _document_has_open_order_discrepancy( session: SessionIdentity, record: FiscalDocumentRecord, *, covering_delivery_note: FiscalDocumentRecord | None = None, order_match: dict[str, object] | None = None, ) -> bool: if record.status != "ready" or record.review_status != "to_review": return False if record.document_type not in _ORDER_RECONCILABLE_DOCUMENT_TYPES: return False if covering_delivery_note is not None: return False match = order_match if match is None: if fiscal_engine_v2_enabled(session): try: v2_analysis = get_fiscal_document_v2_analysis(session, record.id) except FiscalEngineV2Error: return False match = v2_analysis.get("order_match") if isinstance(v2_analysis, dict) and isinstance(v2_analysis.get("order_match"), dict) else None else: document_items = get_tenant_store().list_fiscal_document_items(record.tenant_id, record.id) match = build_fiscal_document_order_match( session, supplier_name=record.supplier_name, document_date=record.document_date, document_type=record.document_type, document_items=[_fiscal_document_item_order_match_payload(item) for item in document_items], ) return _match_has_order_discrepancy(match) def _serialize_fiscal_document(session: SessionIdentity, record: FiscalDocumentRecord) -> FiscalDocumentRead: analysis_engine = "legacy" analysis_engine_version = None analysis_run_id = None analysis_decision = None analysis_confidence = None source_page_count = None validation_messages: list[dict[str, object]] = [] v2_analysis = None if fiscal_engine_v2_enabled(session): try: v2_analysis = get_fiscal_document_v2_analysis(session, record.id) except FiscalEngineV2Error as exc: validation_messages = [ { "code": "engine_unavailable", "severity": "error", "message": str(exc), "line_index": None, } ] if v2_analysis is not None: line_items = [ FiscalDocumentLineItemRead.model_validate(item) for item in v2_analysis.get("line_items") or [] if isinstance(item, dict) ] analysis_engine = str(v2_analysis.get("engine") or "fiscal-engine-v2") analysis_engine_version = str(v2_analysis.get("engine_version") or "") or None analysis_run_id = str(v2_analysis.get("analysis_run_id") or "") or None analysis_decision = str(v2_analysis.get("decision") or "") or None analysis_confidence = float(v2_analysis["confidence"]) if v2_analysis.get("confidence") is not None else None source_page_count = int(v2_analysis.get("source_page_count") or 0) validation_messages = [ message for message in v2_analysis.get("validation_messages") or [] if isinstance(message, dict) ] else: line_items = [ _serialize_fiscal_document_line_item(item) for item in get_tenant_store().list_fiscal_document_items(record.tenant_id, record.id) ] covering_delivery_note = _find_covering_delivery_note(record) order_match = None if covering_delivery_note is None: if v2_analysis is not None: order_match = v2_analysis.get("order_match") if isinstance(v2_analysis.get("order_match"), dict) else None elif not fiscal_engine_v2_enabled(session): order_match = build_fiscal_document_order_match( session, supplier_name=record.supplier_name, document_date=record.document_date, document_type=record.document_type, document_items=[_fiscal_document_item_order_match_payload(item) for item in line_items], ) return FiscalDocumentRead( id=record.id, original_name=record.original_name, display_name=record.display_name, mime_type=record.mime_type, kind=record.kind, file_size_bytes=record.file_size_bytes, document_type=record.document_type, # type: ignore[arg-type] document_number=record.document_number, document_date=record.document_date, supplier_name=record.supplier_name, total_amount=record.total_amount, currency=record.currency, summary_text=record.summary_text, extracted_text=record.extracted_text, preview_text=record.preview_text, drive_file_id=record.drive_file_id, drive_web_url=record.drive_web_url, drive_uploaded_at=record.drive_uploaded_at, covered_by_document_id=covering_delivery_note.id if covering_delivery_note else None, covered_by_document_name=covering_delivery_note.display_name if covering_delivery_note else None, line_items=line_items, order_match=_serialize_order_match(order_match), has_order_discrepancy=_document_has_open_order_discrepancy( session, record, covering_delivery_note=covering_delivery_note, order_match=order_match, ), status=record.status, # type: ignore[arg-type] matching_status=record.matching_status, # type: ignore[arg-type] review_status=record.review_status, # type: ignore[arg-type] error_detail=record.error_detail, analysis_engine=analysis_engine, analysis_engine_version=analysis_engine_version, analysis_run_id=analysis_run_id, analysis_decision=analysis_decision, # type: ignore[arg-type] analysis_confidence=analysis_confidence, source_page_count=source_page_count, validation_messages=validation_messages, created_at=record.created_at, updated_at=record.updated_at, ) def _serialize_fiscal_document_summary(session: SessionIdentity, record: FiscalDocumentRecord) -> FiscalDocumentRead: covering_delivery_note = _find_covering_delivery_note(record) return FiscalDocumentRead( id=record.id, original_name=record.original_name, display_name=record.display_name, mime_type=record.mime_type, kind=record.kind, file_size_bytes=record.file_size_bytes, document_type=record.document_type, # type: ignore[arg-type] document_number=record.document_number, document_date=record.document_date, supplier_name=record.supplier_name, total_amount=record.total_amount, currency=record.currency, summary_text=record.summary_text, extracted_text="", preview_text="", drive_file_id=record.drive_file_id, drive_web_url=record.drive_web_url, drive_uploaded_at=record.drive_uploaded_at, covered_by_document_id=covering_delivery_note.id if covering_delivery_note else None, covered_by_document_name=covering_delivery_note.display_name if covering_delivery_note else None, line_items=[], order_match=None, has_order_discrepancy=_cached_document_has_open_order_discrepancy( session, record, covering_delivery_note=covering_delivery_note, ), status=record.status, # type: ignore[arg-type] matching_status=record.matching_status, # type: ignore[arg-type] review_status=record.review_status, # type: ignore[arg-type] error_detail=record.error_detail, created_at=record.created_at, updated_at=record.updated_at, ) def _record_search_text(record: FiscalDocumentRecord) -> str: return " ".join( [ record.display_name, record.original_name, record.document_number or "", record.document_date or "", record.supplier_name or "", record.summary_text or "", ] ).casefold() def _filter_fiscal_records( records: list[FiscalDocumentRecord], *, fiscal_filter: str, query: str | None, year: int | None, month: int | None, supplier: str | None, ) -> list[FiscalDocumentRecord]: normalized_query = (query or "").strip().casefold() normalized_supplier = (supplier or "").strip().casefold() filtered: list[FiscalDocumentRecord] = [] for record in records: if fiscal_filter == "invoice" and record.document_type not in {"invoice", "instant_invoice"}: continue if fiscal_filter in {"instant_invoice", "delivery_note"} and record.document_type != fiscal_filter: continue if fiscal_filter == "to_review" and record.review_status != "to_review": continue if year is not None: record_year = str(record.document_date or record.created_at or "")[:4] if record_year != str(year): continue if month is not None: record_month = str(record.document_date or record.created_at or "")[5:7] if record_month != f"{month:02d}": continue if normalized_supplier and normalized_supplier not in (record.supplier_name or "").casefold(): continue if normalized_query and normalized_query not in _record_search_text(record): continue filtered.append(record) return filtered def _build_list_response( session: SessionIdentity, records: list[FiscalDocumentRecord], *, all_records: list[FiscalDocumentRecord] | None = None, limit: int = 50, offset: int = 0, ) -> FiscalDocumentsListResponse: source_records = all_records if all_records is not None else records page_records = records[offset : offset + limit] items = [_serialize_fiscal_document_summary(session, record) for record in page_records] return FiscalDocumentsListResponse( items=items, total_count=len(records), archive_total_count=len(source_records), invoice_count=sum(1 for item in source_records if item.document_type in {"invoice", "instant_invoice"}), instant_invoice_count=sum(1 for item in source_records if item.document_type == "instant_invoice"), delivery_note_count=sum(1 for item in source_records if item.document_type == "delivery_note"), to_review_count=sum(1 for item in source_records if item.review_status == "to_review"), limit=limit, offset=offset, has_more=offset + limit < len(records), ) def _match_has_order_discrepancy(match: dict[str, object] | None) -> bool: if not isinstance(match, dict) or match.get("status") != "matched": return False if int(match.get("missing_line_count") or 0) > 0 or int(match.get("extra_line_count") or 0) > 0: return True lines = match.get("lines") if isinstance(match.get("lines"), list) else [] return any(isinstance(line, dict) and line.get("status") not in {None, "exact"} for line in lines) def _build_notification_summary(session: SessionIdentity) -> FiscalDocumentNotificationSummary: store = get_tenant_store() discrepancy_count = 0 for record in store.list_fiscal_documents(session.tenant_id): if record.review_status != "to_review" or record.status != "ready": continue if record.document_type not in _ORDER_RECONCILABLE_DOCUMENT_TYPES: continue covering_delivery_note = _find_covering_delivery_note(record) if covering_delivery_note is not None: continue if _cached_document_has_open_order_discrepancy( session, record, covering_delivery_note=covering_delivery_note, ): discrepancy_count += 1 return FiscalDocumentNotificationSummary(discrepancy_count=discrepancy_count) def _serialize_inbox_item(record: FiscalDocumentInboxItemRecord) -> FiscalDocumentInboxItemRead: return FiscalDocumentInboxItemRead( id=record.id, message_id=record.message_id, attachment_id=record.attachment_id, subject=record.subject, sender=record.sender, received_at=record.received_at, attachment_name=record.attachment_name, mime_type=record.mime_type, sync_status=record.sync_status, # type: ignore[arg-type] document_id=record.document_id, error_detail=record.error_detail, created_at=record.created_at, updated_at=record.updated_at, ) def _build_inbox_list_response(records: list[FiscalDocumentInboxItemRecord]) -> FiscalDocumentInboxListResponse: items = [_serialize_inbox_item(record) for record in records] return FiscalDocumentInboxListResponse( items=items, total_count=len(items), imported_count=sum(1 for item in items if item.sync_status == "imported"), unsupported_count=sum(1 for item in items if item.sync_status == "unsupported"), error_count=sum(1 for item in items if item.sync_status == "error"), ) def _serialize_settings(session: SessionIdentity) -> FiscalDocumentSettingsRead: settings = get_tenant_store().get_fiscal_document_settings(session.tenant_id) return FiscalDocumentSettingsRead( inbound_email=settings.inbound_email, mailbox_configured=bool(settings.inbound_email), updated_at=settings.updated_at, ) def _normalize_inbound_email(value: str | None) -> str | None: normalized = (value or "").strip().lower() if not normalized: return None if not _SIMPLE_EMAIL_PATTERN.match(normalized): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Inserisci una casella email valida per ricevere bolle e fatture.", ) return normalized def _build_inbox_status(session: SessionIdentity) -> FiscalDocumentInboxStatusRead: settings = get_tenant_store().get_fiscal_document_settings(session.tenant_id) connection = get_google_workspace_store().get_connection(session.tenant_id, adopt_legacy_if_needed=True) google_configured = is_google_workspace_configured() granted_scopes = [item.strip() for item in (connection.scope.split() if connection and connection.scope else []) if item.strip()] gmail_scope_granted = _GMAIL_READONLY_SCOPE in granted_scopes inbound_email = settings.inbound_email google_account_email = connection.account_email if connection else None mailbox_matches_google_account = bool( inbound_email and google_account_email and inbound_email.strip().lower() == google_account_email.strip().lower() ) ready_for_sync = bool( inbound_email and google_configured and connection is not None and gmail_scope_granted and mailbox_matches_google_account ) if not inbound_email: next_step = "Configura prima la mailbox del locale che ricevera bolle e fatture." elif not google_configured: next_step = "Configura l'app Google Workspace nel backend per poter leggere la mailbox del locale." elif connection is None: next_step = "Collega l'account Google della mailbox documenti per abilitare la lettura delle email in arrivo." elif not gmail_scope_granted: next_step = "Ricollega Google e accetta anche il permesso Gmail in sola lettura per leggere le email in arrivo." elif not mailbox_matches_google_account: next_step = "La mailbox salvata non coincide con l'account Google collegato. Usa lo stesso indirizzo o collega l'account corretto." else: next_step = "Mailbox pronta. La lettura automatica delle email e' attiva; il pulsante Sincronizza serve solo a forzare un controllo immediato." return FiscalDocumentInboxStatusRead( mailbox_configured=bool(inbound_email), inbound_email=inbound_email, google_configured=google_configured, google_connected=connection is not None, google_account_email=google_account_email, gmail_scope_granted=gmail_scope_granted, mailbox_matches_google_account=mailbox_matches_google_account, ready_for_sync=ready_for_sync, next_step=next_step, ) @router.get("/settings", response_model=FiscalDocumentSettingsRead) def get_fiscal_document_settings(session: SessionIdentity = Depends(require_session)) -> FiscalDocumentSettingsRead: admin_session = _require_tenant_admin(session) return _serialize_settings(admin_session) @router.put("/settings", response_model=FiscalDocumentSettingsRead) def update_fiscal_document_settings( payload: FiscalDocumentSettingsUpdatePayload, session: SessionIdentity = Depends(require_session), ) -> FiscalDocumentSettingsRead: admin_session = _require_tenant_admin(session) inbound_email = _normalize_inbound_email(payload.inbound_email) get_tenant_store().upsert_fiscal_document_settings(admin_session.tenant_id, inbound_email=inbound_email) return _serialize_settings(admin_session) @router.get("/inbox-status", response_model=FiscalDocumentInboxStatusRead) def get_fiscal_document_inbox_status(session: SessionIdentity = Depends(require_session)) -> FiscalDocumentInboxStatusRead: admin_session = _require_tenant_admin(session) return _build_inbox_status(admin_session) @router.get("/inbox", response_model=FiscalDocumentInboxListResponse) def list_fiscal_document_inbox(session: SessionIdentity = Depends(require_session)) -> FiscalDocumentInboxListResponse: admin_session = _require_tenant_admin(session) records = get_tenant_store().list_fiscal_document_inbox_items(admin_session.tenant_id) return _build_inbox_list_response(records) @router.post("/inbox/sync", response_model=FiscalDocumentInboxListResponse) async def sync_fiscal_document_inbox_route(session: SessionIdentity = Depends(require_session)) -> FiscalDocumentInboxListResponse: admin_session = _require_tenant_admin(session) await sync_fiscal_document_inbox(admin_session) records = get_tenant_store().list_fiscal_document_inbox_items(admin_session.tenant_id) return _build_inbox_list_response(records) @router.get("/notifications", response_model=FiscalDocumentNotificationSummary) def get_fiscal_document_notifications(session: SessionIdentity = Depends(require_session)) -> FiscalDocumentNotificationSummary: admin_session = _require_tenant_admin(session) return _build_notification_summary(admin_session) @router.get("", response_model=FiscalDocumentsListResponse) def list_fiscal_documents( filter: str = Query(default="all", pattern="^(all|invoice|instant_invoice|delivery_note|to_review)$"), q: str | None = Query(default=None, max_length=180), year: int | None = Query(default=None, ge=2000, le=2100), month: int | None = Query(default=None, ge=1, le=12), supplier: str | None = Query(default=None, max_length=180), limit: int = Query(default=50, ge=1, le=100), offset: int = Query(default=0, ge=0), session: SessionIdentity = Depends(require_session), ) -> FiscalDocumentsListResponse: admin_session = _require_tenant_admin(session) records = get_tenant_store().list_fiscal_documents(admin_session.tenant_id) filtered_records = _filter_fiscal_records( records, fiscal_filter=filter, query=q, year=year, month=month, supplier=supplier, ) return _build_list_response(admin_session, filtered_records, all_records=records, limit=limit, offset=offset) @router.delete("/archive") def clear_fiscal_documents_archive(session: SessionIdentity = Depends(require_session)) -> dict[str, int]: admin_session = _require_tenant_admin(session) deleted_documents, deleted_inbox_items = get_tenant_store().clear_fiscal_documents_archive(admin_session.tenant_id) return { "deleted_documents": deleted_documents, "deleted_inbox_items": deleted_inbox_items, } @router.delete("/{document_id}") def delete_fiscal_document( document_id: str, session: SessionIdentity = Depends(require_session), ) -> dict[str, object]: admin_session = _require_tenant_admin(session) deleted = get_tenant_store().delete_fiscal_document(admin_session.tenant_id, document_id) if not deleted: raise HTTPException(status_code=404, detail="Documento fiscale non trovato.") return {"ok": True, "deleted_document_id": document_id} @router.post("/upload", response_model=FiscalDocumentsListResponse) async def upload_fiscal_documents( files: list[UploadFile] = File(...), session: SessionIdentity = Depends(require_session), ) -> FiscalDocumentsListResponse: admin_session = _require_tenant_admin(session) if not files: raise HTTPException(status_code=400, detail="Seleziona almeno un documento fiscale da caricare.") for upload in files: await ingest_fiscal_document(admin_session, upload) records = get_tenant_store().list_fiscal_documents(admin_session.tenant_id) return _build_list_response(admin_session, records, all_records=records) @router.get("/{document_id}", response_model=FiscalDocumentRead) def get_fiscal_document( document_id: str, session: SessionIdentity = Depends(require_session), ) -> FiscalDocumentRead: admin_session = _require_tenant_admin(session) record = get_tenant_store().get_fiscal_document(admin_session.tenant_id, document_id) if record is None: raise HTTPException(status_code=404, detail="Documento fiscale non trovato.") return _serialize_fiscal_document(admin_session, record) @router.post("/{document_id}/apply-storno") def apply_fiscal_document_storno( document_id: str, session: SessionIdentity = Depends(require_session), ) -> FiscalDocumentRead: admin_session = _require_tenant_admin(session) if not get_tenant_store().session_has_permission(admin_session, "ordini"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Questo account non puo modificare gli ordini.") record = get_tenant_store().get_fiscal_document(admin_session.tenant_id, document_id) if record is None: raise HTTPException(status_code=404, detail="Documento fiscale non trovato.") if fiscal_engine_v2_enabled(admin_session): try: analysis = get_fiscal_document_v2_analysis(admin_session, document_id) except FiscalEngineV2Error as exc: raise HTTPException(status_code=503, detail=str(exc)) from exc if analysis is None: raise HTTPException(status_code=409, detail="Analisi Fiscal Engine V2 non ancora disponibile.") order_match = analysis.get("order_match") if isinstance(analysis.get("order_match"), dict) else {} if not order_match.get("can_apply_storno"): raise HTTPException(status_code=400, detail="Il confronto v2 richiede una revisione prima di aggiornare l'ordine.") idempotency_key = f"{analysis.get('analysis_run_id')}:apply-order" started = record_fiscal_document_v2_application( admin_session, document_id, idempotency_key=idempotency_key, application_status="started", detail={"matched_batch_ids": order_match.get("matched_batch_ids") or []}, ) if not started.get("already_completed"): try: result = apply_delivery_note_storno_to_matched_order( admin_session, document_id=document_id, order_match=order_match, ) except Exception as exc: record_fiscal_document_v2_application( admin_session, document_id, idempotency_key=idempotency_key, application_status="failed", detail={"error": str(exc)}, ) raise record_fiscal_document_v2_application( admin_session, document_id, idempotency_key=idempotency_key, application_status="completed", detail=result, ) else: document_items = get_tenant_store().list_fiscal_document_items(admin_session.tenant_id, document_id) order_match = build_fiscal_document_order_match( admin_session, supplier_name=record.supplier_name, document_date=record.document_date, document_type=record.document_type, document_items=[_fiscal_document_item_order_match_payload(item) for item in document_items], ) apply_delivery_note_storno_to_matched_order( admin_session, document_id=document_id, order_match=order_match or {}, ) refreshed = get_tenant_store().get_fiscal_document(admin_session.tenant_id, document_id) if refreshed is None: raise HTTPException(status_code=404, detail="Documento fiscale non trovato dopo l'aggiornamento ordine.") return _serialize_fiscal_document(admin_session, refreshed) @router.post("/{document_id}/mark-reviewed", response_model=FiscalDocumentRead) def mark_fiscal_document_reviewed( document_id: str, session: SessionIdentity = Depends(require_session), ) -> FiscalDocumentRead: admin_session = _require_tenant_admin(session) record = get_tenant_store().get_fiscal_document(admin_session.tenant_id, document_id) if record is None: raise HTTPException(status_code=404, detail="Documento fiscale non trovato.") if fiscal_engine_v2_enabled(admin_session): try: analysis = review_fiscal_document_v2( admin_session, document_id, corrections=[], note="Confronto confermato manualmente dall'admin.", confirm_document=True, ) except FiscalEngineV2Error as exc: raise HTTPException(status_code=503, detail=str(exc)) from exc updated = persist_fiscal_engine_v2_analysis(admin_session, record, analysis) return _serialize_fiscal_document(admin_session, updated) updated = get_tenant_store().update_fiscal_document( admin_session.tenant_id, document_id, review_status="reviewed", ) return _serialize_fiscal_document(admin_session, updated) @router.post("/{document_id}/v2/reanalyze", response_model=FiscalDocumentRead) def reanalyze_fiscal_document_v2( document_id: str, session: SessionIdentity = Depends(require_session), ) -> FiscalDocumentRead: admin_session = _require_tenant_admin(session) if not fiscal_engine_v2_enabled(admin_session): raise HTTPException(status_code=404, detail="Fiscal Engine V2 non e abilitato per questo tenant.") record = get_tenant_store().get_fiscal_document(admin_session.tenant_id, document_id) if record is None: raise HTTPException(status_code=404, detail="Documento fiscale non trovato.") try: analysis = analyze_fiscal_document_v2(admin_session, record, force=True) except FiscalEngineV2Error as exc: raise HTTPException(status_code=503, detail=str(exc)) from exc updated = persist_fiscal_engine_v2_analysis(admin_session, record, analysis) return _serialize_fiscal_document(admin_session, updated) @router.post("/{document_id}/v2/review", response_model=FiscalDocumentRead) def review_fiscal_document_with_v2( document_id: str, payload: FiscalDocumentV2ReviewPayload, session: SessionIdentity = Depends(require_session), ) -> FiscalDocumentRead: admin_session = _require_tenant_admin(session) if not fiscal_engine_v2_enabled(admin_session): raise HTTPException(status_code=404, detail="Fiscal Engine V2 non e abilitato per questo tenant.") record = get_tenant_store().get_fiscal_document(admin_session.tenant_id, document_id) if record is None: raise HTTPException(status_code=404, detail="Documento fiscale non trovato.") try: analysis = review_fiscal_document_v2( admin_session, document_id, corrections=[correction.model_dump(exclude_unset=True) for correction in payload.corrections], note=payload.note, confirm_document=payload.confirm_document, ) except FiscalEngineV2Error as exc: raise HTTPException(status_code=503, detail=str(exc)) from exc updated = persist_fiscal_engine_v2_analysis(admin_session, record, analysis) return _serialize_fiscal_document(admin_session, updated) @router.get("/{document_id}/download") def download_fiscal_document( document_id: str, session: SessionIdentity = Depends(require_session), ): admin_session = _require_tenant_admin(session) record = get_tenant_store().get_fiscal_document(admin_session.tenant_id, document_id) if record is None: raise HTTPException(status_code=404, detail="Documento fiscale non trovato.") storage_path = Path(record.storage_path) if not storage_path.exists(): raise HTTPException(status_code=404, detail="Il file non e piu disponibile sul server.") return FileResponse(storage_path, media_type=record.mime_type, filename=record.display_name)