from datetime import datetime from pathlib import Path import re 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, ) 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, ) 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" _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 _serialize_fiscal_document(session: SessionIdentity, record: FiscalDocumentRecord) -> FiscalDocumentRead: 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: 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), 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 _serialize_fiscal_document_summary(record: FiscalDocumentRecord) -> FiscalDocumentRead: 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, line_items=[], order_match=None, 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( 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(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 _find_covering_delivery_note(record) is not None: continue line_items = store.list_fiscal_document_items(record.tenant_id, record.id) 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], ) if _match_has_order_discrepancy(order_match): 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(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(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.") 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.") updated = get_tenant_store().update_fiscal_document( admin_session.tenant_id, document_id, review_status="reviewed", ) 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)