from __future__ import annotations import time from copy import deepcopy from datetime import date from typing import Any from pydantic import BaseModel, EmailStr, Field from sqlalchemy import select from app.db import LedgerStateRow, SessionLocal from app.ledger_defaults import EMPTY_LEDGER_PAYLOAD class LineItem(BaseModel): id: str description: str qty: float unitPrice: float vatRate: float = 21 class Suggestion(BaseModel): ledger: str vatCode: str category: str relation: str reasoning: str editedBy: str | None = None class Invoice(BaseModel): id: str file: str clientId: str supplier: str number: str invoiceDate: str dueDate: str vatAmount: float total: float currency: str paymentRef: str uploadedAt: str confidence: float status: str lines: list[LineItem] suggestion: Suggestion duplicateOf: str | None = None fileHash: str | None = None duplicateStatus: str = "None" syncError: str | None = None externalInvoiceId: str | None = None syncedAt: str | None = None mimeType: str | None = None fileSize: int | None = None uploadedBy: str | None = None storageKey: str | None = None # Review / approval audit (populated by firm workflow; OCR fills fields later) processedAt: str | None = None reviewedAt: str | None = None reviewedBy: str | None = None approvedAt: str | None = None approvedBy: str | None = None rejectedAt: str | None = None rejectedBy: str | None = None rejectionReason: str | None = None updatedAt: str | None = None paidAmount: float = 0 # Local OCR / extraction (optional — older documents may have these as null) ocrText: str | None = None ocrStatus: str | None = None # pending | completed | failed | unavailable ocrEngine: str | None = None ocrProcessedAt: str | None = None ocrError: str | None = None # Snapshot of OCR-detected fields (before accountant corrections). # Authoritative bookkeeping values remain supplier/number/total/etc. extractedFields: dict[str, Any] | None = None class BankAllocation(BaseModel): id: str invoiceId: str amount: float createdAt: str class BankTx(BaseModel): id: str date: str description: str clientId: str amount: float suggestedInvoiceId: str | None = None confidence: float | None = None status: str note: str | None = None # Extended bank fields (SnelStart import + matching audit) externalId: str | None = None source: str = "manual" currency: str = "EUR" transactionType: str = "payment" bankAccount: str | None = None iban: str | None = None paymentReference: str | None = None matchedRecordId: str | None = None matchedRecordType: str | None = None matchedBy: str | None = None matchedAt: str | None = None ignoreReason: str | None = None allocations: list[BankAllocation] = Field(default_factory=list) createdAt: str | None = None updatedAt: str | None = None class Client(BaseModel): id: str name: str kvk: str snelstart: bool = False snelstartStatus: str = "not_connected" lastUpload: str email: str = "" class QueryMessage(BaseModel): id: str senderRole: str senderName: str message: str attachmentRef: str | None = None createdAt: str class ClientQuery(BaseModel): id: str clientId: str invoiceId: str | None = None title: str | None = None question: str createdAt: str createdBy: str | None = None status: str priority: str = "Normal" answer: str | None = None answeredAt: str | None = None proof: str | None = None resolvedAt: str | None = None resolvedBy: str | None = None updatedAt: str | None = None messages: list[QueryMessage] = Field(default_factory=list) class ActivityEvent(BaseModel): id: str type: str text: str actor: str at: int clientId: str | None = None invoiceId: str | None = None transactionId: str | None = None queryId: str | None = None class Settings(BaseModel): threshold: float autoSync: bool globalRule: str clientRules: dict[str, str] class LedgerState(BaseModel): clients: list[Client] invoices: list[Invoice] transactions: list[BankTx] queries: list[ClientQuery] activity: list[ActivityEvent] settings: Settings class UploadDocumentRequest(BaseModel): clientId: str fileName: str actor: str = "User" mimeType: str | None = None fileSize: int | None = None storageKey: str | None = None uploadedBy: str | None = None fileHash: str | None = None forceDuplicate: bool = False duplicateOf: str | None = None duplicateStatus: str = "None" class FirmActorMixin(BaseModel): platform: str = "firm" email: str = "" actor: str = "" class UpdateInvoiceRequest(FirmActorMixin): patch: dict[str, Any] class UpdateSuggestionRequest(FirmActorMixin): patch: dict[str, Any] editedBy: str = "" class InvoiceStatusRequest(FirmActorMixin): status: str class ApproveInvoiceRequest(FirmActorMixin): pass class RejectInvoiceRequest(FirmActorMixin): reason: str = Field(min_length=1, max_length=2000) class SyncInvoiceRequest(FirmActorMixin): pass class BulkInvoiceRequest(FirmActorMixin): ids: list[str] class DuplicateStatusRequest(FirmActorMixin): status: str class CreateQueryRequest(FirmActorMixin): clientId: str question: str invoiceId: str | None = None title: str | None = None priority: str = "Normal" class QueryMessageRequest(BaseModel): platform: str email: str = "" actor: str = "" message: str attachmentRef: str | None = None class AnswerQueryRequest(BaseModel): platform: str = "client" email: str = "" answer: str proof: str | None = None class QueryActorRequest(FirmActorMixin): pass class CancelQueryRequest(FirmActorMixin): reason: str | None = None class TxStatusRequest(FirmActorMixin): status: str class SettingsPatchRequest(BaseModel): patch: dict[str, Any] class ClientRuleRequest(BaseModel): rule: str class CreateClientRequest(BaseModel): name: str = Field(min_length=1, max_length=255) company_name: str = "" email: EmailStr password: str = Field(min_length=8, max_length=128) kvk: str = "" class BankActorRequest(FirmActorMixin): pass class MatchBankTxRequest(FirmActorMixin): invoiceId: str amount: float | None = None class IgnoreBankTxRequest(FirmActorMixin): reason: str = Field(min_length=1, max_length=2000) class CreateBankTxRequest(FirmActorMixin): """Manual / test import of one bank row — not used to seed demo UI data.""" clientId: str date: str description: str amount: float currency: str = "EUR" paymentReference: str = "" iban: str = "" bankAccount: str = "" externalId: str = "" class LedgerStore: def __init__(self) -> None: from app.db import init_db init_db() self._seq = 1000 loaded = self._load_from_db() if loaded is not None: self.state = loaded self._seq = self._infer_seq(self.state) return self.state = LedgerState.model_validate(EMPTY_LEDGER_PAYLOAD) self._persist() def _infer_seq(self, state: LedgerState) -> int: max_n = 1000 for collection in ( state.clients, state.invoices, state.queries, state.activity, state.transactions, ): for item in collection: parts = str(getattr(item, "id", "")).rsplit("-", 1) if len(parts) == 2 and parts[1].isdigit(): max_n = max(max_n, int(parts[1])) return max_n def _load_from_db(self) -> LedgerState | None: try: with SessionLocal() as session: row = session.get(LedgerStateRow, 1) if not row: return None return LedgerState.model_validate(row.payload) except Exception: return None def _persist(self) -> None: payload = self.state.model_dump(mode="json") with SessionLocal() as session: row = session.get(LedgerStateRow, 1) if row: row.payload = payload else: session.add(LedgerStateRow(id=1, payload=payload)) session.commit() def snapshot(self) -> LedgerState: snap = deepcopy(self.state) self._enrich_snelstart_clients(snap.clients) return snap def snapshot_for(self, platform: str, client_id: str | None) -> LedgerState: """Firm sees the full ledger. Clients only see their own records.""" snap = deepcopy(self.state) self._enrich_snelstart_clients(snap.clients) if platform != "client": return snap if not client_id: empty = deepcopy(EMPTY_LEDGER_PAYLOAD) empty["settings"] = snap.settings.model_dump() return LedgerState.model_validate(empty) hidden_activity = {"edit", "match", "ai", "sync", "sync-failed", "processing"} return LedgerState( clients=[c for c in snap.clients if c.id == client_id], invoices=[i for i in snap.invoices if i.clientId == client_id], transactions=[t for t in snap.transactions if t.clientId == client_id], queries=[q for q in snap.queries if q.clientId == client_id], activity=[ e for e in snap.activity if e.clientId == client_id and e.type not in hidden_activity ], settings=snap.settings, ) def _enrich_snelstart_clients(self, clients: list[Client]) -> None: from app.services.snelstart.connection import sync_legacy_client_flags sync_legacy_client_flags(clients) def find_client_by_email(self, email: str) -> Client | None: email_n = email.strip().lower() if not email_n: return None return next( (c for c in self.state.clients if (c.email or "").strip().lower() == email_n), None, ) def create_client( self, name: str, *, kvk: str = "", email: str = "", link_user: bool = True, ) -> Client: """Create or reuse a client administratie. Email is the portal join key.""" today = date.today().isoformat() email_n = email.strip().lower() if email_n: existing = self.find_client_by_email(email_n) if existing: if name.strip(): existing.name = name.strip() if kvk.strip(): existing.kvk = kvk.strip() self._persist() if link_user: self._link_portal_user(email_n, existing.id) return existing try: from app.services import users as user_service portal_user = user_service.get_user("client", email_n) except Exception: portal_user = None if portal_user and portal_user.client_id: owned = self.get_client(portal_user.client_id) if owned: owned.email = email_n if name.strip(): owned.name = name.strip() if kvk.strip(): owned.kvk = kvk.strip() self._persist() return owned client = Client( id=self._next_id("c"), name=name.strip() or "Unnamed client", kvk=kvk.strip(), snelstart=False, snelstartStatus="not_connected", lastUpload=today, email=email_n, ) self.state.clients.append(client) self._persist() if email_n and link_user: self._link_portal_user(email_n, client.id) return client def remove_client(self, client_id: str) -> None: """Remove a ledger client that has no documents or queries (rollback helper).""" if any(i.clientId == client_id for i in self.state.invoices): return if any(q.clientId == client_id for q in self.state.queries): return self.state.clients = [c for c in self.state.clients if c.id != client_id] self._persist() def _link_portal_user(self, email: str, client_id: str) -> None: try: from app.services import users as user_service if user_service.get_user("client", email): user_service.set_client_id("client", email, client_id) except Exception: pass def get_client(self, client_id: str) -> Client | None: return next((c for c in self.state.clients if c.id == client_id), None) @staticmethod def _client_snelstart_status(client: Client) -> str: if client.snelstartStatus: return client.snelstartStatus return "connected" if client.snelstart else "not_connected" @staticmethod def _client_sort_key(client: Client) -> tuple[int, str]: parts = client.id.rsplit("-", 1) num = int(parts[-1]) if len(parts) > 1 and parts[-1].isdigit() else 0 return (-num, client.name.lower()) def list_clients( self, *, search: str | None = None, connected_only: bool = False, page: int = 1, page_size: int = 20, ) -> dict[str, Any]: from app.services.client_overview import NEEDS_REVIEW_STATUSES, OPEN_QUERY_STATUSES clients = list(self.state.clients) self._enrich_snelstart_clients(clients) doc_counts: dict[str, int] = {} open_docs: dict[str, int] = {} for invoice in self.state.invoices: client_id = invoice.clientId doc_counts[client_id] = doc_counts.get(client_id, 0) + 1 if invoice.status in NEEDS_REVIEW_STATUSES: open_docs[client_id] = open_docs.get(client_id, 0) + 1 open_queries: dict[str, int] = {} for query in self.state.queries: if query.status in OPEN_QUERY_STATUSES: open_queries[query.clientId] = open_queries.get(query.clientId, 0) + 1 last_activity: dict[str, int] = {} for event in self.state.activity: if not event.clientId: continue previous = last_activity.get(event.clientId, 0) if event.at > previous: last_activity[event.clientId] = event.at rows = clients query_text = (search or "").strip().lower() if query_text: rows = [ client for client in rows if query_text in client.name.lower() or query_text in (client.email or "").lower() or query_text in client.kvk.lower() or query_text in client.id.lower() ] if connected_only: rows = [client for client in rows if self._client_snelstart_status(client) == "connected"] rows.sort(key=self._client_sort_key) total = len(rows) offset = max(page - 1, 0) * page_size page_rows = rows[offset : offset + page_size] items = [ { **client.model_dump(), "documentCount": doc_counts.get(client.id, 0), "openDocuments": open_docs.get(client.id, 0), "openQueries": open_queries.get(client.id, 0), "lastActivityAt": last_activity.get(client.id), } for client in page_rows ] return {"items": items, "total": total} def _next_id(self, prefix: str) -> str: self._seq += 1 return f"{prefix}-{self._seq}" def _log( self, event_type: str, text: str, actor: str, client_id: str | None = None, invoice_id: str | None = None, transaction_id: str | None = None, query_id: str | None = None, ) -> ActivityEvent: event = ActivityEvent( id=self._next_id("act"), type=event_type, text=text, actor=actor, at=int(time.time() * 1000), clientId=client_id, invoiceId=invoice_id, transactionId=transaction_id, queryId=query_id, ) self.state.activity.insert(0, event) self._persist() return event def _find_invoice(self, invoice_id: str) -> Invoice: for invoice in self.state.invoices: if invoice.id == invoice_id: return invoice raise KeyError(invoice_id) def _replace_invoice(self, invoice_id: str, updated: Invoice, *, persist: bool = True) -> Invoice: invoice = self._find_invoice(invoice_id) index = self.state.invoices.index(invoice) self.state.invoices[index] = updated if persist: self._persist() return updated def find_file_duplicate(self, client_id: str, file_hash: str) -> Invoice | None: from app.services.duplicate_detection import find_exact_file_duplicate return find_exact_file_duplicate( self.state.invoices, client_id=client_id, file_hash=file_hash, ) def upload_document(self, payload: UploadDocumentRequest) -> Invoice: from app.services.duplicate_detection import ( DUPLICATE_NONE, DUPLICATE_POSSIBLE, find_exact_file_duplicate, ) from app.services.invoice_workflow import STATUS_PROCESSING, today_iso, utc_now_iso dup_status = payload.duplicateStatus or DUPLICATE_NONE dup_of = payload.duplicateOf if payload.fileHash and not payload.forceDuplicate: existing = find_exact_file_duplicate( self.state.invoices, client_id=payload.clientId, file_hash=payload.fileHash, ) if existing: from app.services.duplicate_detection import DuplicateUploadError raise DuplicateUploadError(existing) if payload.forceDuplicate and payload.fileHash: existing = find_exact_file_duplicate( self.state.invoices, client_id=payload.clientId, file_hash=payload.fileHash, ) if existing: dup_status = DUPLICATE_POSSIBLE dup_of = existing.id today = today_iso() now = utc_now_iso() invoice_id = self._next_id("INV") invoice = Invoice( id=invoice_id, file=payload.fileName, clientId=payload.clientId, supplier="", number="", invoiceDate="", dueDate="", vatAmount=0, total=0, currency="EUR", paymentRef="", uploadedAt=today, confidence=0, status=STATUS_PROCESSING, lines=[], suggestion=Suggestion( ledger="", vatCode="", category="", relation="", reasoning="Not processed yet. Local OCR has not run.", ), mimeType=payload.mimeType, fileSize=payload.fileSize, uploadedBy=payload.uploadedBy or payload.actor, storageKey=payload.storageKey, fileHash=payload.fileHash, duplicateStatus=dup_status, duplicateOf=dup_of, updatedAt=now, ocrStatus="pending", ) self.state.invoices.insert(0, invoice) client_name = next( (c.name for c in self.state.clients if c.id == payload.clientId), payload.actor or "Client", ) for client in self.state.clients: if client.id == payload.clientId: client.lastUpload = today break self._log( "upload", f"Document {payload.fileName} uploaded", payload.actor, payload.clientId, invoice_id, ) if dup_status != DUPLICATE_NONE: self._log( "duplicate", f"Possible duplicate of {dup_of or 'existing document'}", payload.actor, payload.clientId, invoice_id, ) self._persist() try: from app.services import document_processing as processing from app.services import notifications as notif notif.on_document_uploaded( client_id=payload.clientId, client_name=client_name, invoice_id=invoice_id, file_name=payload.fileName, ) notif.notify_client( client_id=payload.clientId, type=notif.NotificationType.DOCUMENT_PROCESSING, title="Document received", message=f"We received {payload.fileName} and will process it shortly.", reference_type="invoice", reference_id=invoice_id, ) processing.enqueue_document_processing(invoice_id) except Exception: pass return self._find_invoice(invoice_id) def log_ocr_event(self, invoice_id: str, text: str) -> None: """Record an OCR-related activity event without changing invoice fields.""" try: invoice = self._find_invoice(invoice_id) except KeyError: return self._log("processing", text, "Amanah", invoice.clientId, invoice_id) self._persist() def apply_ocr_result(self, invoice_id: str, result) -> Invoice: """ Apply local OCR output onto the invoice. Populates bookkeeping fields only when OCR detected a value (never invents). Stores a snapshot in extractedFields; accountant edits later update supplier/etc. """ from app.services.invoice_workflow import utc_now_iso invoice = self._find_invoice(invoice_id) now = utc_now_iso() extraction = getattr(result, "extraction", None) extracted_dict = extraction.to_dict() if extraction else None updates: dict[str, Any] = { "ocrText": (result.text or "")[:100_000] or None, "ocrStatus": result.status, "ocrEngine": result.ocr_engine, "ocrProcessedAt": now, "ocrError": result.error, "extractedFields": extracted_dict, "updatedAt": now, } if extraction: updates["confidence"] = float(extraction.overall_confidence or 0) def _str_val(field) -> str: if field is None or field.value is None: return "" return str(field.value).strip() def _float_val(field) -> float | None: if field is None or field.value is None: return None try: return float(field.value) except (TypeError, ValueError): return None supplier = _str_val(extraction.supplier) number = _str_val(extraction.invoice_number) invoice_date = _str_val(extraction.invoice_date) due_date = _str_val(extraction.due_date) currency = _str_val(extraction.currency) or invoice.currency or "EUR" payment_ref = _str_val(extraction.payment_ref) vat = _float_val(extraction.vat) total = _float_val(extraction.total) # Only fill empty authoritative fields from OCR (don't wipe firm edits on re-run) if supplier and not (invoice.supplier or "").strip(): updates["supplier"] = supplier if number and not (invoice.number or "").strip(): updates["number"] = number if invoice_date and not (invoice.invoiceDate or "").strip(): updates["invoiceDate"] = invoice_date if due_date and not (invoice.dueDate or "").strip(): updates["dueDate"] = due_date if payment_ref and not (invoice.paymentRef or "").strip(): updates["paymentRef"] = payment_ref if currency: updates["currency"] = currency if vat is not None and float(invoice.vatAmount or 0) == 0: updates["vatAmount"] = vat if total is not None and float(invoice.total or 0) == 0: updates["total"] = total conf = float(extraction.overall_confidence or 0) if conf >= 80: band = "High" elif conf >= 60: band = "Medium" else: band = "Low" engine = result.ocr_engine or "local-ocr" updates["suggestion"] = invoice.suggestion.model_copy( update={ "reasoning": ( f"Local OCR/extraction confidence: {band} ({conf:.0f}%). " f"Engine: {engine}. " "Verify fields against the original document before approval." ) } ) updated = invoice.model_copy(update=updates) meta_dup = self._check_metadata_duplicate(updated) if meta_dup and updated.duplicateStatus in {"None", "Not duplicate"}: from app.services.duplicate_detection import DUPLICATE_POSSIBLE updated = updated.model_copy( update={"duplicateStatus": DUPLICATE_POSSIBLE, "duplicateOf": meta_dup.id} ) self._replace_invoice(invoice_id, updated) return updated def mark_ocr_failed(self, invoice_id: str, *, error: str) -> Invoice: """OCR unavailable/failed — move to Needs review for manual firm entry.""" from app.services.invoice_workflow import ( STATUS_NEEDS_REVIEW, STATUS_NEW, STATUS_PROCESSING, assert_transition, display_label, utc_now_iso, WorkflowError, ) invoice = self._find_invoice(invoice_id) now = utc_now_iso() safe_error = (error or "Local OCR could not process this document.").strip()[:500] updates: dict[str, Any] = { "ocrStatus": "failed" if "unavailable" not in safe_error.lower() else "unavailable", "ocrError": safe_error, "ocrProcessedAt": now, "ocrEngine": invoice.ocrEngine or "tesseract", "confidence": 0, "processedAt": now, "updatedAt": now, "suggestion": invoice.suggestion.model_copy( update={ "reasoning": ( "Document processing is currently unavailable or failed. " "Enter invoice fields manually from the original file." ) } ), } if invoice.status in {STATUS_NEW, STATUS_PROCESSING}: try: assert_transition(invoice.status, STATUS_NEEDS_REVIEW) updates["status"] = STATUS_NEEDS_REVIEW except WorkflowError: pass if "unavailable" in safe_error.lower() or "not installed" in safe_error.lower(): updates["ocrStatus"] = "unavailable" updated = invoice.model_copy(update=updates) self._replace_invoice(invoice_id, updated, persist=False) self._log( "processing", f"OCR processing failed for {invoice.file}", "Amanah", invoice.clientId, invoice_id, ) try: from app.services import notifications as notif notif.on_document_needs_review( client_id=invoice.clientId, invoice_id=invoice_id, supplier=display_label(invoice.file, invoice.supplier), total=invoice.total, file_name=invoice.file, ) notif.notify_client( client_id=invoice.clientId, type=notif.NotificationType.DOCUMENT_READY, title="Document under review", message=( f"We couldn't reliably read {invoice.file}. " "Your accountant will review it manually." ), reference_type="invoice", reference_id=invoice_id, ) except Exception: pass self._persist() return updated def mark_ready_for_review(self, invoice_id: str, *, confidence: float | None = None) -> Invoice: """Move Processing/New → Needs review or Auto-ready after OCR extraction.""" from app.services.invoice_workflow import ( STATUS_AUTO_READY, STATUS_NEEDS_REVIEW, STATUS_NEW, STATUS_PROCESSING, assert_transition, display_label, utc_now_iso, WorkflowError, ) invoice = self._find_invoice(invoice_id) if invoice.status not in {STATUS_NEW, STATUS_PROCESSING}: return invoice conf = float(confidence if confidence is not None else invoice.confidence or 0) threshold = float(self.state.settings.threshold or 85) # Auto-ready only when local extraction confidence is strong AND key fields exist has_basics = bool((invoice.supplier or "").strip() and float(invoice.total or 0) > 0) target = ( STATUS_AUTO_READY if conf >= threshold and has_basics and invoice.ocrStatus == "completed" else STATUS_NEEDS_REVIEW ) try: assert_transition(invoice.status, target) except WorkflowError: try: assert_transition(invoice.status, STATUS_NEEDS_REVIEW) target = STATUS_NEEDS_REVIEW except WorkflowError: return invoice now = utc_now_iso() updated = invoice.model_copy( update={ "status": target, "processedAt": now, "confidence": conf, "updatedAt": now, } ) self._replace_invoice(invoice_id, updated, persist=False) self._log( "processing", f"Document {invoice.file} ready for firm review", "Amanah", invoice.clientId, invoice_id, ) try: from app.services import notifications as notif notif.on_document_needs_review( client_id=invoice.clientId, invoice_id=invoice_id, supplier=display_label(updated.file, updated.supplier), total=updated.total, file_name=updated.file, ) notif.notify_client( client_id=invoice.clientId, type=notif.NotificationType.DOCUMENT_READY, title="Document under review", message=f"Your accountant is reviewing {invoice.file}.", reference_type="invoice", reference_id=invoice_id, ) except Exception: pass self._persist() return updated def set_duplicate_status( self, invoice_id: str, *, status: str, actor: str, ) -> Invoice: from app.services.duplicate_detection import ( DUPLICATE_CLEARED, DUPLICATE_CONFIRMED, DUPLICATE_POSSIBLE, ) from app.services.invoice_workflow import utc_now_iso invoice = self._find_invoice(invoice_id) allowed = {DUPLICATE_CONFIRMED, DUPLICATE_CLEARED, DUPLICATE_POSSIBLE} if status not in allowed: raise ValueError(f"Invalid duplicate status: {status}") now = utc_now_iso() updated = invoice.model_copy(update={"duplicateStatus": status, "updatedAt": now}) self._replace_invoice(invoice_id, updated, persist=False) if status == DUPLICATE_CONFIRMED: self._log( "duplicate", f"{actor} confirmed duplicate for {invoice.file}", actor, invoice.clientId, invoice_id, ) try: from app.services import notifications as notif notif.on_document_duplicate_confirmed( client_id=invoice.clientId, invoice_id=invoice_id, file_name=invoice.file, ) except Exception: pass else: self._log( "duplicate-cleared", f"{actor} cleared duplicate flag for {invoice.file}", actor, invoice.clientId, invoice_id, ) self._persist() return updated def _check_metadata_duplicate(self, invoice: Invoice) -> Invoice | None: from app.services.duplicate_detection import find_metadata_duplicate return find_metadata_duplicate( self.state.invoices, client_id=invoice.clientId, supplier=invoice.supplier, number=invoice.number, invoice_date=invoice.invoiceDate, total=float(invoice.total or 0), exclude_id=invoice.id, ) def update_invoice(self, invoice_id: str, patch: dict[str, Any], *, actor: str = "Firm") -> Invoice: from app.services.invoice_workflow import EDITABLE_STATUSES, utc_now_iso, WorkflowError invoice = self._find_invoice(invoice_id) if invoice.status not in EDITABLE_STATUSES: raise WorkflowError(f"Cannot edit a document with status “{invoice.status}”.") blocked = { "id", "clientId", "status", "storageKey", "mimeType", "fileSize", "uploadedAt", "uploadedBy", "processedAt", "reviewedAt", "reviewedBy", "approvedAt", "approvedBy", "rejectedAt", "rejectedBy", "rejectionReason", "syncError", "ocrText", "ocrStatus", "ocrEngine", "ocrProcessedAt", "ocrError", "extractedFields", "fileHash", "confidence", } clean = {k: v for k, v in patch.items() if k not in blocked} if not clean: return invoice updated = invoice.model_copy(update={**clean, "updatedAt": utc_now_iso()}) meta_dup = self._check_metadata_duplicate(updated) if meta_dup and updated.duplicateStatus in {"None", "Not duplicate"}: from app.services.duplicate_detection import DUPLICATE_POSSIBLE updated = updated.model_copy( update={"duplicateStatus": DUPLICATE_POSSIBLE, "duplicateOf": meta_dup.id} ) self._replace_invoice(invoice_id, updated, persist=False) label = updated.number or updated.file self._log( "edit", f"Document fields updated on {label}", actor, updated.clientId, invoice_id, ) return updated def update_suggestion( self, invoice_id: str, patch: dict[str, Any], edited_by: str ) -> Invoice: from app.services.invoice_workflow import EDITABLE_STATUSES, utc_now_iso, WorkflowError invoice = self._find_invoice(invoice_id) if invoice.status not in EDITABLE_STATUSES: raise WorkflowError(f"Cannot edit a document with status “{invoice.status}”.") suggestion = invoice.suggestion.model_copy(update={**patch, "editedBy": edited_by}) updated = invoice.model_copy( update={"suggestion": suggestion, "updatedAt": utc_now_iso()} ) self._replace_invoice(invoice_id, updated, persist=False) label = updated.number or updated.file self._log( "edit", f"Booking suggestion adjusted on {label}", edited_by, updated.clientId, invoice_id, ) return updated def set_invoice_status(self, invoice_id: str, status: str, *, actor: str = "Firm") -> Invoice: from app.services.invoice_workflow import assert_transition, utc_now_iso invoice = self._find_invoice(invoice_id) assert_transition(invoice.status, status) updated = invoice.model_copy(update={"status": status, "updatedAt": utc_now_iso()}) self._replace_invoice(invoice_id, updated, persist=False) self._log( "edit", f"Status of {updated.file} set to {status}", actor, updated.clientId, invoice_id, ) return updated def approve_invoice(self, invoice_id: str, *, actor: str) -> Invoice: from app.services.invoice_workflow import ( APPROVABLE_STATUSES, STATUS_APPROVED, assert_transition, display_label, utc_now_iso, WorkflowError, ) invoice = self._find_invoice(invoice_id) if invoice.status not in APPROVABLE_STATUSES: raise WorkflowError( f"Only documents that need review can be approved (current status: “{invoice.status}”)." ) assert_transition(invoice.status, STATUS_APPROVED) supplier = (invoice.supplier or "").strip() if not supplier: raise WorkflowError("Please enter the supplier before approving this document.") now = utc_now_iso() updated = invoice.model_copy( update={ "status": STATUS_APPROVED, "reviewedAt": now, "reviewedBy": actor, "approvedAt": now, "approvedBy": actor, "rejectedAt": None, "rejectedBy": None, "rejectionReason": None, "updatedAt": now, } ) self._replace_invoice(invoice_id, updated, persist=False) self._log( "approve", f"{actor} approved {invoice.file} — awaiting SnelStart integration", actor, invoice.clientId, invoice_id, ) try: from app.services import notifications as notif notif.on_invoice_approved( client_id=invoice.clientId, invoice_id=invoice_id, supplier=display_label(invoice.file, supplier), ) except Exception: pass self._persist() return updated def reject_invoice(self, invoice_id: str, *, actor: str, reason: str) -> Invoice: from app.services.invoice_workflow import ( REJECTABLE_STATUSES, STATUS_REJECTED, assert_transition, display_label, utc_now_iso, WorkflowError, ) reason_n = (reason or "").strip() if not reason_n: raise WorkflowError("A rejection reason is required.") invoice = self._find_invoice(invoice_id) if invoice.status not in REJECTABLE_STATUSES: raise WorkflowError( f"Cannot reject a document with status “{invoice.status}”." ) assert_transition(invoice.status, STATUS_REJECTED) now = utc_now_iso() updated = invoice.model_copy( update={ "status": STATUS_REJECTED, "reviewedAt": now, "reviewedBy": actor, "rejectedAt": now, "rejectedBy": actor, "rejectionReason": reason_n, "updatedAt": now, } ) self._replace_invoice(invoice_id, updated, persist=False) self._log( "reject", f"{actor} rejected {invoice.file}: {reason_n[:80]}", actor, invoice.clientId, invoice_id, ) try: from app.services import notifications as notif notif.on_invoice_rejected( client_id=invoice.clientId, invoice_id=invoice_id, supplier=display_label(invoice.file, invoice.supplier), reason=reason_n, ) except Exception: pass self._persist() return updated def sync_invoice(self, invoice_id: str, *, actor: str = "Amanah") -> Invoice: from app.services.snelstart.invoices import push_invoice return push_invoice(invoice_id, actor=actor) def bulk_approve(self, ids: list[str], *, actor: str) -> list[Invoice]: from app.services.invoice_workflow import WorkflowError updated: list[Invoice] = [] errors: list[str] = [] for invoice_id in ids: try: updated.append(self.approve_invoice(invoice_id, actor=actor)) except (KeyError, WorkflowError) as exc: errors.append(f"{invoice_id}: {exc}") if not updated and errors: raise WorkflowError( errors[0] if len(errors) == 1 else f"{len(errors)} documents could not be approved." ) if updated: self._log("approve", f"{actor} approved {len(updated)} documents in bulk", actor) return updated def bulk_needs_review(self, ids: list[str], *, actor: str) -> list[Invoice]: from app.services.invoice_workflow import ( STATUS_NEEDS_REVIEW, assert_transition, display_label, utc_now_iso, WorkflowError, ) updated: list[Invoice] = [] now = utc_now_iso() for invoice_id in ids: try: invoice = self._find_invoice(invoice_id) assert_transition(invoice.status, STATUS_NEEDS_REVIEW) next_inv = invoice.model_copy( update={"status": STATUS_NEEDS_REVIEW, "updatedAt": now} ) self._replace_invoice(invoice_id, next_inv, persist=False) updated.append(next_inv) except (KeyError, WorkflowError): continue if updated: self._log( "edit", f"{actor} marked {len(updated)} documents as needs review", actor, ) try: from app.services import notifications as notif for inv in updated: notif.on_document_needs_review( client_id=inv.clientId, invoice_id=inv.id, supplier=display_label(inv.file, inv.supplier), total=inv.total, file_name=inv.file, ) except Exception: pass self._persist() return updated def _find_query(self, query_id: str) -> ClientQuery: for query in self.state.queries: if query.id == query_id: return self._migrate_query(query) raise KeyError(query_id) def _replace_query(self, query_id: str, updated: ClientQuery, *, persist: bool = True) -> ClientQuery: for index, query in enumerate(self.state.queries): if query.id == query_id: self.state.queries[index] = updated if persist: self._persist() return updated raise KeyError(query_id) def _migrate_query(self, query: ClientQuery) -> ClientQuery: from app.services.query_workflow import migrate_legacy_status status = migrate_legacy_status(query.status) messages = list(query.messages or []) if not messages and query.question.strip(): messages = [ QueryMessage( id=f"{query.id}-m1", senderRole="firm", senderName=query.createdBy or "Accountant", message=query.question, createdAt=query.createdAt, ) ] if query.answer: messages.append( QueryMessage( id=f"{query.id}-m2", senderRole="client", senderName="Client", message=query.answer, attachmentRef=query.proof, createdAt=query.answeredAt or query.createdAt, ) ) changed = status != query.status or messages != (query.messages or []) if not changed: return query updated = query.model_copy(update={"status": status, "messages": messages}) idx = next(i for i, q in enumerate(self.state.queries) if q.id == query.id) self.state.queries[idx] = updated return updated def _query_invoice_context(self, invoice_id: str | None) -> tuple[str | None, str | None]: if not invoice_id: return None, None try: inv = self._find_invoice(invoice_id) return inv.supplier or inv.file, inv.id except KeyError: return None, invoice_id def list_queries( self, *, client_id: str | None = None, status: str | None = None, priority: str | None = None, date_from: str | None = None, date_to: str | None = None, query: str | None = None, invoice_id: str | None = None, ) -> list[ClientQuery]: rows = [self._migrate_query(q) for q in self.state.queries] if client_id: rows = [q for q in rows if q.clientId == client_id] if status and status != "all": rows = [q for q in rows if q.status == status] if priority and priority != "all": rows = [q for q in rows if (q.priority or "Normal") == priority] if date_from: rows = [q for q in rows if (q.createdAt or "") >= date_from] if date_to: rows = [q for q in rows if (q.createdAt or "") <= date_to] if invoice_id: rows = [q for q in rows if q.invoiceId == invoice_id] if query: qn = query.strip().lower() if qn: filtered: list[ClientQuery] = [] for row in rows: inv_label, _ = self._query_invoice_context(row.invoiceId) client_name = next( (c.name for c in self.state.clients if c.id == row.clientId), "", ) hay = " ".join( filter( None, [ row.question, row.title or "", client_name, inv_label or "", *(m.message for m in row.messages), ], ) ).lower() if qn in hay or row.id.lower().startswith(qn): filtered.append(row) rows = filtered return rows def query_stats(self, *, client_id: str | None = None) -> dict[str, int]: rows = self.list_queries(client_id=client_id) stats = { "requested": 0, "delivered": 0, "answered": 0, "resolved": 0, "cancelled": 0, "total": len(rows), } for row in rows: key = row.status.lower().replace(" ", "") if key == "requested": stats["requested"] += 1 elif key == "delivered": stats["delivered"] += 1 elif key == "answered": stats["answered"] += 1 elif key == "resolved": stats["resolved"] += 1 elif key == "cancelled": stats["cancelled"] += 1 return stats def create_query(self, payload: CreateQueryRequest, *, actor: str) -> ClientQuery: from app.services.query_workflow import PRIORITY_NORMAL, STATUS_DELIVERED, VALID_PRIORITIES from app.services.invoice_workflow import today_iso, utc_now_iso if payload.invoiceId: inv = self._find_invoice(payload.invoiceId) if inv.clientId != payload.clientId: raise ValueError("Invoice does not belong to this client") priority = payload.priority if payload.priority in VALID_PRIORITIES else PRIORITY_NORMAL now = utc_now_iso() today = today_iso() query_id = self._next_id("Q") first_message = QueryMessage( id=self._next_id("QM"), senderRole="firm", senderName=actor, message=payload.question.strip(), createdAt=now, ) query = ClientQuery( id=query_id, clientId=payload.clientId, invoiceId=payload.invoiceId, title=(payload.title or "").strip() or None, question=payload.question.strip(), createdAt=today, createdBy=actor, status=STATUS_DELIVERED, priority=priority, updatedAt=now, messages=[first_message], ) self.state.queries.insert(0, query) client = self.get_client(payload.clientId) client_label = client.name if client else payload.clientId self._log( "query", f"Question sent to {client_label}", actor, payload.clientId, payload.invoiceId, query_id=query_id, ) supplier, _ = self._query_invoice_context(payload.invoiceId) try: from app.services import notifications as notif notif.on_accountant_question( client_id=payload.clientId, query_id=query_id, invoice_id=payload.invoiceId, supplier=supplier, question=payload.question, ) except Exception: pass self._persist() return query def add_query_message( self, query_id: str, *, sender_role: str, sender_name: str, message: str, attachment_ref: str | None = None, ) -> ClientQuery: from app.services.query_workflow import ( STATUS_ANSWERED, STATUS_DELIVERED, can_client_reply, ) from app.services.invoice_workflow import utc_now_iso query = self._find_query(query_id) if sender_role == "client" and not can_client_reply(query.status): raise ValueError("This question is no longer open for client replies") if query.status in {"Resolved", "Cancelled"}: raise ValueError("This question is closed") now = utc_now_iso() msg = QueryMessage( id=self._next_id("QM"), senderRole=sender_role, senderName=sender_name, message=message.strip(), attachmentRef=attachment_ref, createdAt=now, ) updates: dict[str, object] = { "messages": [*query.messages, msg], "updatedAt": now, } if sender_role == "client": updates["answer"] = message.strip() updates["proof"] = attachment_ref updates["answeredAt"] = now[:10] updates["status"] = STATUS_ANSWERED elif sender_role == "firm" and query.status == STATUS_ANSWERED: # Firm follow-up keeps Answered until resolved pass elif sender_role == "firm" and query.status == STATUS_DELIVERED: pass updated = query.model_copy(update=updates) self._replace_query(query_id, updated, persist=False) client = self.get_client(query.clientId) client_name = client.name if client else "Client" supplier, _ = self._query_invoice_context(query.invoiceId) if sender_role == "client": self._log( "answer", f"{client_name} answered a question", client_name, query.clientId, query.invoiceId, query_id=query_id, ) try: from app.services import notifications as notif notif.on_question_answered( client_id=query.clientId, client_name=client_name, query_id=query_id, invoice_id=query.invoiceId, supplier=supplier, ) except Exception: pass else: self._log( "query-reply", f"{sender_name} replied in a client query", sender_name, query.clientId, query.invoiceId, query_id=query_id, ) try: from app.services import notifications as notif notif.on_firm_query_reply( client_id=query.clientId, query_id=query_id, invoice_id=query.invoiceId, supplier=supplier, message=message, ) except Exception: pass self._persist() return updated def answer_query(self, query_id: str, payload: AnswerQueryRequest) -> ClientQuery: client = None for q in self.state.queries: if q.id == query_id: client = self.get_client(q.clientId) break client_name = client.name if client else "Client" return self.add_query_message( query_id, sender_role="client", sender_name=client_name, message=payload.answer, attachment_ref=payload.proof, ) def resolve_query(self, query_id: str, *, actor: str) -> ClientQuery: from app.services.query_workflow import STATUS_RESOLVED, can_firm_resolve from app.services.invoice_workflow import utc_now_iso query = self._find_query(query_id) if not can_firm_resolve(query.status): raise ValueError("Only open questions can be resolved") now = utc_now_iso() updated = query.model_copy( update={ "status": STATUS_RESOLVED, "resolvedAt": now[:10], "resolvedBy": actor, "updatedAt": now, } ) self._replace_query(query_id, updated, persist=False) client = self.get_client(query.clientId) client_label = client.name if client else query.clientId self._log( "query-resolve", f"Question resolved for {client_label}", actor, query.clientId, query.invoiceId, query_id=query_id, ) try: from app.services import notifications as notif notif.on_question_resolved( client_id=query.clientId, query_id=query_id, invoice_id=query.invoiceId, ) except Exception: pass self._persist() return updated def cancel_query(self, query_id: str, *, actor: str, reason: str | None = None) -> ClientQuery: from app.services.query_workflow import STATUS_CANCELLED, can_firm_cancel from app.services.invoice_workflow import utc_now_iso query = self._find_query(query_id) if not can_firm_cancel(query.status): raise ValueError("This question cannot be cancelled") now = utc_now_iso() updated = query.model_copy( update={ "status": STATUS_CANCELLED, "updatedAt": now, "title": query.title or (f"Cancelled: {reason}" if reason else query.title), } ) self._replace_query(query_id, updated, persist=False) self._log( "query-cancel", f"Question cancelled{(f': {reason}' if reason else '')}", actor, query.clientId, query.invoiceId, query_id=query_id, ) self._persist() return updated def approve_query(self, query_id: str, *, actor: str = "Accountant") -> ClientQuery: """Legacy alias — resolves the query.""" return self.resolve_query(query_id, actor=actor) def _find_transaction(self, tx_id: str) -> BankTx: for tx in self.state.transactions: if tx.id == tx_id: return tx raise KeyError(tx_id) def _replace_transaction(self, tx_id: str, updated: BankTx, *, persist: bool = True) -> BankTx: tx = self._find_transaction(tx_id) index = self.state.transactions.index(tx) self.state.transactions[index] = updated if persist: self._persist() return updated def list_bank_transactions( self, *, client_id: str | None = None, status: str | None = None, date_from: str | None = None, date_to: str | None = None, query: str | None = None, ) -> list[BankTx]: rows = list(self.state.transactions) if client_id: rows = [t for t in rows if t.clientId == client_id] if status and status != "all": rows = [t for t in rows if t.status == status] if date_from: rows = [t for t in rows if (t.date or "") >= date_from] if date_to: rows = [t for t in rows if (t.date or "") <= date_to] if query: q = query.strip().lower() if q: filtered: list[BankTx] = [] for t in rows: inv = None if t.suggestedInvoiceId: try: inv = self._find_invoice(t.suggestedInvoiceId) except KeyError: inv = None hay = " ".join( filter( None, [ t.description, t.paymentReference or "", t.iban or "", t.bankAccount or "", str(t.amount), inv.supplier if inv else "", inv.number if inv else "", ], ) ).lower() client_name = next( (c.name for c in self.state.clients if c.id == t.clientId), "", ).lower() if q in hay or q in client_name: filtered.append(t) rows = filtered rows.sort(key=lambda t: (t.date or "", t.id), reverse=True) return rows INBOX_QUEUE_STATUSES = frozenset( {"Processing", "New", "AI processed", "Needs review", "Auto-ready"} ) INVOICE_LIST_EXCLUDE = {"ocrText", "extractedFields", "lines", "suggestion"} def list_invoices( self, *, client_id: str | None = None, search: str | None = None, status: str | None = None, confidence: str | None = None, uploaded_at: str | None = None, ) -> list[Invoice]: rows = list(self.state.invoices) if client_id: rows = [i for i in rows if i.clientId == client_id] status_n = (status or "").strip() if status_n and status_n not in {"all", "*"}: if status_n == "queue": rows = [i for i in rows if i.status in self.INBOX_QUEUE_STATUSES] else: rows = [i for i in rows if i.status == status_n] uploaded_n = (uploaded_at or "").strip() if uploaded_n: rows = [i for i in rows if (i.uploadedAt or "") == uploaded_n] confidence_n = (confidence or "").strip().lower() if confidence_n == "high": rows = [i for i in rows if i.confidence >= 85] elif confidence_n == "medium": rows = [i for i in rows if 65 <= i.confidence < 85] elif confidence_n == "low": rows = [i for i in rows if i.confidence < 65] query_text = (search or "").strip().lower() if query_text: names = {c.id: c.name for c in self.state.clients} filtered: list[Invoice] = [] for invoice in rows: hay = " ".join( filter( None, [ invoice.file, invoice.supplier, invoice.number, invoice.id, names.get(invoice.clientId, ""), ], ) ).lower() if query_text in hay: filtered.append(invoice) rows = filtered rows.sort(key=lambda i: (i.uploadedAt or "", i.id), reverse=True) return rows def list_invoices_page( self, *, client_id: str | None = None, search: str | None = None, status: str | None = None, confidence: str | None = None, uploaded_at: str | None = None, page: int = 1, page_size: int = 20, ) -> dict[str, Any]: rows = self.list_invoices( client_id=client_id, search=search, status=status, confidence=confidence, uploaded_at=uploaded_at, ) total = len(rows) offset = max(page - 1, 0) * page_size page_rows = rows[offset : offset + page_size] names = {c.id: c.name for c in self.state.clients} if client_id: has_any = any(i.clientId == client_id for i in self.state.invoices) else: has_any = bool(self.state.invoices) items = [] for invoice in page_rows: data = invoice.model_dump(exclude=self.INVOICE_LIST_EXCLUDE) data["clientName"] = names.get(invoice.clientId, "") items.append(data) return {"items": items, "total": total, "has_any": has_any} def list_activity(self, *, client_id: str | None = None, limit: int = 25) -> list[ActivityEvent]: rows = list(self.state.activity) if client_id: rows = [e for e in rows if e.clientId == client_id] rows.sort(key=lambda e: e.at, reverse=True) if limit > 0: return rows[:limit] return rows def bank_transaction_stats(self, *, client_id: str | None = None) -> dict[str, int]: from app.services.bank_workflow import ( STATUS_IGNORED, STATUS_MATCHED, STATUS_NEEDS_REVIEW, STATUS_PARTIAL, STATUS_SUGGESTED, STATUS_UNMATCHED, ) rows = self.list_bank_transactions(client_id=client_id) counts = { "unmatched": 0, "suggested": 0, "needsReview": 0, "matched": 0, "ignored": 0, "partial": 0, "total": len(rows), } for tx in rows: if tx.status == STATUS_UNMATCHED: counts["unmatched"] += 1 elif tx.status == STATUS_SUGGESTED: counts["suggested"] += 1 elif tx.status == STATUS_NEEDS_REVIEW: counts["needsReview"] += 1 elif tx.status in {STATUS_MATCHED, STATUS_PARTIAL}: counts["matched"] += 1 if tx.status == STATUS_PARTIAL: counts["partial"] += 1 elif tx.status == STATUS_IGNORED: counts["ignored"] += 1 return counts def _run_matcher_on_tx(self, tx: BankTx) -> BankTx: from app.services.bank_matching import apply_suggestion, best_match candidate = best_match(tx, self.state.invoices) patch = apply_suggestion(tx, candidate) return tx.model_copy(update=patch) def import_bank_transaction(self, client_id: str, row) -> BankTx: """Import one bank row (from SnelStart or manual API). Runs matcher.""" from app.services.bank_workflow import STATUS_UNMATCHED, utc_now_iso from app.services.snelstart.models import BankImportRow now = utc_now_iso() if isinstance(row, BankImportRow): data = row else: data = row external = getattr(data, "external_id", None) or getattr(data, "externalId", "") or "" source = getattr(data, "source", None) or "manual" if external: existing = next( ( t for t in self.state.transactions if t.clientId == client_id and t.externalId == external ), None, ) if existing: return existing tx = BankTx( id=self._next_id("T"), date=getattr(data, "transaction_date", None) or getattr(data, "date", ""), description=getattr(data, "description", ""), clientId=client_id, amount=float(getattr(data, "amount", 0)), status=STATUS_UNMATCHED, currency=getattr(data, "currency", "EUR") or "EUR", transactionType=getattr(data, "transaction_type", "payment") or "payment", bankAccount=getattr(data, "bank_account", "") or getattr(data, "bankAccount", "") or None, iban=getattr(data, "iban", "") or None, paymentReference=getattr(data, "payment_reference", "") or getattr(data, "paymentReference", "") or None, externalId=external or None, source=source, createdAt=now, updatedAt=now, ) tx = self._run_matcher_on_tx(tx) self.state.transactions.insert(0, tx) self._log( "match", f"Bank transaction imported: {tx.description}", "Amanah", client_id, transaction_id=tx.id, ) self._persist() return tx def create_bank_transaction_manual(self, payload: CreateBankTxRequest) -> BankTx: from app.services.snelstart.models import BankImportRow if not any(c.id == payload.clientId for c in self.state.clients): raise KeyError("client") row = BankImportRow( external_id=payload.externalId or f"manual-{int(time.time() * 1000)}", transaction_date=payload.date, description=payload.description, amount=payload.amount, currency=payload.currency, bank_account=payload.bankAccount, iban=payload.iban, payment_reference=payload.paymentReference, source="manual", ) return self.import_bank_transaction(payload.clientId, row) def get_transaction_matches(self, tx_id: str) -> list[dict]: from app.services.bank_matching import find_matches, invoice_outstanding tx = self._find_transaction(tx_id) candidates = find_matches(tx, self.state.invoices, limit=10) out: list[dict] = [] for c in candidates: inv = self._find_invoice(c.invoice_id) out.append( { "invoiceId": c.invoice_id, "score": c.score, "confidenceLabel": c.confidence_label, "signals": c.signals, "allocatedAmount": c.allocated_amount, "outstanding": c.outstanding, "isPartial": c.is_partial, "supplier": inv.supplier or inv.file, "number": inv.number or "—", "total": inv.total, "invoiceDate": inv.invoiceDate, } ) return out def search_transaction_matches(self, tx_id: str, query: str) -> list[dict]: tx = self._find_transaction(tx_id) q = query.strip().lower() if not q: return [] hits: list[dict] = [] for inv in self.state.invoices: if inv.clientId != tx.clientId: continue hay = f"{inv.supplier} {inv.number} {inv.paymentRef} {inv.file}".lower() if q not in hay and q not in inv.id.lower(): continue from app.services.bank_matching import score_match scored = score_match(tx, inv) if not scored: continue hits.append( { "invoiceId": inv.id, "score": scored.score, "confidenceLabel": scored.confidence_label, "signals": scored.signals, "allocatedAmount": scored.allocated_amount, "outstanding": scored.outstanding, "isPartial": scored.is_partial, "supplier": inv.supplier or inv.file, "number": inv.number or "—", "total": inv.total, "invoiceDate": inv.invoiceDate, } ) hits.sort(key=lambda h: (-h["score"], h["invoiceId"])) return hits[:20] def approve_bank_match( self, tx_id: str, *, invoice_id: str, actor: str, amount: float | None = None, ) -> BankTx: from app.services.bank_matching import invoice_outstanding, score_match from app.services.bank_workflow import ( STATUS_MATCHED, STATUS_PARTIAL, STATUS_UNMATCHED, utc_now_iso, BankWorkflowError, ) tx = self._find_transaction(tx_id) invoice = self._find_invoice(invoice_id) if invoice.clientId != tx.clientId: raise BankWorkflowError("Invoice belongs to a different client.") tx_amount = abs(float(tx.amount or 0)) if tx_amount <= 0: raise BankWorkflowError("Transaction amount must be non-zero.") outstanding = invoice_outstanding(invoice) allocate = amount if amount is not None else min(tx_amount, outstanding if outstanding > 0 else tx_amount) if allocate <= 0: raise BankWorkflowError("Allocation amount must be greater than zero.") if outstanding > 0 and allocate - outstanding > 0.01: raise BankWorkflowError("Allocation exceeds the invoice outstanding balance.") now = utc_now_iso() allocation = BankAllocation( id=self._next_id("BA"), invoiceId=invoice_id, amount=allocate, createdAt=now, ) allocations = list(tx.allocations) + [allocation] is_partial = outstanding > 0 and allocate + 0.01 < outstanding status = STATUS_PARTIAL if is_partial else STATUS_MATCHED inv_index = self.state.invoices.index(invoice) self.state.invoices[inv_index] = invoice.model_copy( update={"paidAmount": float(invoice.paidAmount or 0) + allocate} ) updated = tx.model_copy( update={ "status": status, "suggestedInvoiceId": invoice_id, "matchedRecordId": invoice_id, "matchedRecordType": "invoice", "matchedBy": actor, "matchedAt": now, "confidence": (score_match(tx, invoice).score if score_match(tx, invoice) else tx.confidence), "allocations": allocations, "updatedAt": now, } ) self._replace_transaction(tx_id, updated, persist=False) self._log( "match", f"{actor} matched {tx.description} to {invoice.number or invoice.file} ({allocate:.2f})", actor, tx.clientId, invoice_id, tx_id, ) self._persist() return updated def reject_bank_match(self, tx_id: str, *, actor: str) -> BankTx: from app.services.bank_workflow import STATUS_UNMATCHED, utc_now_iso tx = self._find_transaction(tx_id) updated = tx.model_copy( update={ "status": STATUS_UNMATCHED, "suggestedInvoiceId": None, "confidence": None, "matchedRecordId": None, "matchedRecordType": None, "updatedAt": utc_now_iso(), } ) self._replace_transaction(tx_id, updated, persist=False) self._log( "match", f"{actor} rejected match suggestion for {tx.description}", actor, tx.clientId, transaction_id=tx_id, ) self._persist() return updated def ignore_bank_transaction(self, tx_id: str, *, actor: str, reason: str) -> BankTx: from app.services.bank_workflow import STATUS_IGNORED, utc_now_iso, BankWorkflowError reason_n = (reason or "").strip() if not reason_n: raise BankWorkflowError("An ignore reason is required.") tx = self._find_transaction(tx_id) updated = tx.model_copy( update={ "status": STATUS_IGNORED, "ignoreReason": reason_n, "note": reason_n, "updatedAt": utc_now_iso(), } ) self._replace_transaction(tx_id, updated, persist=False) self._log( "match", f"{actor} ignored bank transaction {tx.description}: {reason_n[:80]}", actor, tx.clientId, transaction_id=tx_id, ) self._persist() return updated def set_tx_status(self, tx_id: str, status: str, *, actor: str = "Firm") -> BankTx: """Legacy status toggle — prefer approve_bank_match / reject / ignore.""" tx = self._find_transaction(tx_id) updated = tx.model_copy(update={"status": status}) self._replace_transaction(tx_id, updated, persist=False) self._log( "match", f"Transaction {tx.description} marked {status.lower()}", actor, tx.clientId, tx.suggestedInvoiceId, tx_id, ) self._persist() return updated def update_settings(self, patch: dict[str, Any]) -> Settings: self.state.settings = self.state.settings.model_copy(update=patch) self._persist() return self.state.settings def set_client_rule(self, client_id: str, rule: str) -> Settings: rules = dict(self.state.settings.clientRules) rules[client_id] = rule self.state.settings = self.state.settings.model_copy(update={"clientRules": rules}) self._persist() return self.state.settings store = LedgerStore()