import hashlib import uuid from dataclasses import asdict from datetime import date, timedelta from pathlib import Path from typing import Any from sqlalchemy import func from sqlalchemy.orm import Session from app.audit import service as audit from app.categorization import mcc from app.categorization import service as categorization from app.categorization.models import MccCode from app.config import get_settings from app.db.base import utcnow from app.db.session import session_scope from app.errors import AppError, bad_request, conflict, not_found from app.identity.models import User from app.imports.adapters import detect_adapter, magic_kind from app.imports.adapters.base import SourceRow, StatementDraft, record_fingerprint from app.imports.blob import get_blob_store, new_object_key from app.imports.crypto import decrypt_file, encrypt_file from app.imports.models import ( DedupCandidate, HoldResolution, ParseRun, SourceRecord, SourceRecordLink, StatementDocument, ) from app.ledger.models import Account, Operation from app.ledger.service import ( create_expense, create_income, create_transfer, ensure_period_open_for_date, ) HOLD_MANUAL_ACTIONS = ("confirm_as_posted", "discard_as_cancelled") def _document(db: Session, *, space_id: uuid.UUID, document_id: uuid.UUID) -> StatementDocument: document = ( db.query(StatementDocument).filter_by(id=document_id, space_id=space_id).one_or_none() ) if document is None: raise not_found("IMPORT_NOT_FOUND", "Statement import not found") return document def create_document( db: Session, *, space_id: uuid.UUID, user: User, filename: str, raw: bytes, account_id: uuid.UUID | None, ) -> tuple[StatementDocument, bool]: settings = get_settings() if not raw or len(raw) > settings.max_upload_bytes: raise AppError( "File size is outside the allowed range", code="IMPORT_SIZE_INVALID", status_code=413 ) kind = magic_kind(raw) if kind is None: raise AppError( "Only XLSX and text-layer PDF statements are supported", code="IMPORT_TYPE_INVALID", status_code=415, ) suffix = Path(filename).suffix.casefold() if suffix not in ({".xlsx"} if kind == "xlsx" else {".pdf"}): raise AppError( "File extension does not match its content", code="IMPORT_TYPE_MISMATCH", status_code=422, ) if account_id is not None: account = db.query(Account.id).filter_by(id=account_id, space_id=space_id).one_or_none() if account is None: raise not_found("LEDGER_ACCOUNT_NOT_FOUND", "Account not found") digest = hashlib.sha256(raw).hexdigest() existing = ( db.query(StatementDocument) .filter_by(space_id=space_id, content_sha256=digest) .one_or_none() ) if existing is not None: return existing, False encrypted = encrypt_file(raw) object_key = new_object_key() get_blob_store().put(object_key, encrypted.ciphertext) try: document = StatementDocument( space_id=space_id, account_id=account_id, original_filename=Path(filename).name[:300], size_bytes=len(raw), content_sha256=digest, object_key=object_key, dek_wrapped=encrypted.dek_wrapped, dek_wrap_nonce=encrypted.wrap_nonce, content_nonce=encrypted.content_nonce, status="queued", uploaded_by_user_id=user.id, ) db.add(document) db.flush() except Exception: get_blob_store().delete(object_key) raise audit.record( db, event_type="import.uploaded", space_id=space_id, user_id=user.id, detail={"document_id": str(document.id), "size_bytes": len(raw)}, ) return document, True def _draft_json(draft: StatementDraft) -> dict[str, Any]: value = asdict(draft) for key in ("period_start", "period_end"): item = value[key] value[key] = item.isoformat() if item else None for row in value["rows"]: for key in ("occurred_on", "posted_on"): item = row[key] row[key] = item.isoformat() if item else None return value def _source_from_row( *, space_id: uuid.UUID, document_id: uuid.UUID, run_id: uuid.UUID, row: SourceRow ) -> SourceRecord: critical_warnings = {"occurred_date_missing", "amount_unparseable", "bank_status_invalid"} warnings = list(row.warnings) if row.direction not in {"in", "out"}: warnings.append("direction_invalid") if row.amount_kopecks == 0: warnings.append("amount_zero") bank_status = getattr(row, "bank_status", "posted") if bank_status not in {"posted", "hold"}: warnings.append("bank_status_invalid") bank_status = "posted" return SourceRecord( space_id=space_id, document_id=document_id, parse_run_id=run_id, external_id=row.external_id, occurred_on=row.occurred_on, posted_on=row.posted_on, amount_kopecks=abs(row.amount_kopecks), direction=row.direction, bank_status=bank_status, description=row.description, mcc=row.mcc, bank_category=row.bank_category, location=row.location, row_ref=row.row_ref, field_warnings={"codes": sorted(set(warnings))} if warnings else None, critical=( bool(set(warnings) & critical_warnings) or row.amount_kopecks == 0 or row.direction not in {"in", "out"} ), record_fp=record_fingerprint(row.occurred_on, abs(row.amount_kopecks), row.description), review_decision="savings_transfer" if row.savings_transfer else "create", ) def _category_id_for_source( db: Session, *, space_id: uuid.UUID, source: SourceRecord ) -> uuid.UUID | None: if source.mcc is None: return None category = categorization.category_for_mcc(db, space_id=space_id, code=source.mcc) return category.id if category is not None else None def _reconciliation_summary(draft: Any) -> dict[str, Any]: """Reconcile parsed rows against declared bank controls with stable diagnostics. Posted rows are compared with the declared inflow/outflow controls, HOLD rows with the declared pending total, and where both balances exist the balance equation must hold. Every check keeps an explicit tri-state field: True (matched), False (mismatch, review-blocking), None (not declared). """ rows = list(draft.rows) posted = [row for row in rows if getattr(row, "bank_status", "posted") == "posted"] holds = [row for row in rows if getattr(row, "bank_status", "posted") == "hold"] posted_inflow = sum(row.amount_kopecks for row in posted if row.direction == "in") posted_outflow = sum(row.amount_kopecks for row in posted if row.direction == "out") hold_total = sum(abs(row.amount_kopecks) for row in holds) declared_inflow = getattr(draft, "declared_inflow_kopecks", None) declared_outflow = getattr(draft, "declared_outflow_kopecks", None) declared_hold = getattr(draft, "declared_hold_kopecks", None) full = draft.opening_balance_kopecks is not None and draft.closing_balance_kopecks is not None checks: list[bool] = [] balance_matched: bool | None = None computed_closing: int | None = None if full: computed_closing = int(draft.opening_balance_kopecks) + posted_inflow - posted_outflow balance_matched = computed_closing == int(draft.closing_balance_kopecks) checks.append(balance_matched) posted_totals_matched: bool | None = None if declared_inflow is not None or declared_outflow is not None: posted_totals_matched = (declared_inflow is None or declared_inflow == posted_inflow) and ( declared_outflow is None or declared_outflow == posted_outflow ) checks.append(posted_totals_matched) hold_totals_matched: bool | None = None if declared_hold is not None: hold_totals_matched = declared_hold == hold_total checks.append(hold_totals_matched) return { "currency": draft.currency, "opening_balance_kopecks": draft.opening_balance_kopecks, "closing_balance_kopecks": draft.closing_balance_kopecks, "full": full, "declared_inflow_kopecks": declared_inflow, "declared_outflow_kopecks": declared_outflow, "declared_hold_kopecks": declared_hold, "parsed_posted_inflow_kopecks": posted_inflow, "parsed_posted_outflow_kopecks": posted_outflow, "parsed_hold_total_kopecks": hold_total, "posted_totals_matched": posted_totals_matched, "hold_totals_matched": hold_totals_matched, "balance_matched": balance_matched, "computed_closing_kopecks": computed_closing, "matched": all(checks) if checks else None, "confirmed_without_full_reconciliation": False, } def _build_candidates(db: Session, *, document: StatementDocument, run: ParseRun) -> bool: pending = False for source in db.query(SourceRecord).filter_by(parse_run_id=run.id).all(): if source.bank_status == "hold": continue duplicate = None if source.external_id: match_query = ( db.query(SourceRecord) .join(ParseRun, ParseRun.id == SourceRecord.parse_run_id) .join(StatementDocument, StatementDocument.id == SourceRecord.document_id) .join(SourceRecordLink, SourceRecordLink.source_record_id == SourceRecord.id) .filter( SourceRecord.space_id == document.space_id, SourceRecord.document_id != document.id, SourceRecord.id != source.id, SourceRecord.external_id == source.external_id, SourceRecord.bank_status == "posted", ParseRun.adapter_id == run.adapter_id, StatementDocument.status == "confirmed", StatementDocument.active_parse_run_id == SourceRecord.parse_run_id, ) ) if document.account_id is not None: match_query = match_query.filter( StatementDocument.account_id == document.account_id ) matches = match_query.all() duplicate = matches[0] if len(matches) == 1 else None if duplicate is None: match_query = ( db.query(SourceRecord) .join(ParseRun, ParseRun.id == SourceRecord.parse_run_id) .join(StatementDocument, StatementDocument.id == SourceRecord.document_id) .join(SourceRecordLink, SourceRecordLink.source_record_id == SourceRecord.id) .filter( SourceRecord.space_id == document.space_id, SourceRecord.document_id != document.id, SourceRecord.id != source.id, SourceRecord.record_fp == source.record_fp, SourceRecord.direction == source.direction, SourceRecord.bank_status == "posted", ParseRun.adapter_id == run.adapter_id, StatementDocument.status == "confirmed", StatementDocument.active_parse_run_id == SourceRecord.parse_run_id, ) ) if document.account_id is not None: match_query = match_query.filter( StatementDocument.account_id == document.account_id ) matches = match_query.all() duplicate = matches[0] if len(matches) == 1 else None if duplicate is not None: link = db.query(SourceRecordLink).filter_by(source_record_id=duplicate.id).one() db.add( DedupCandidate( space_id=document.space_id, source_record_id=source.id, candidate_kind="duplicate", existing_source_record_id=duplicate.id, existing_operation_id=link.operation_id, confidence="strong" if source.external_id else "possible", ) ) pending = True continue if source.occurred_on is None: continue opposite = "in" if source.direction == "out" else "out" transfer_query = ( db.query(SourceRecord) .join(SourceRecordLink, SourceRecordLink.source_record_id == SourceRecord.id) .join(StatementDocument, StatementDocument.id == SourceRecord.document_id) .filter( SourceRecord.space_id == document.space_id, SourceRecord.document_id != document.id, SourceRecord.id != source.id, SourceRecord.amount_kopecks == source.amount_kopecks, SourceRecord.direction == opposite, SourceRecord.bank_status == "posted", SourceRecord.occurred_on >= source.occurred_on - timedelta(days=1), SourceRecord.occurred_on <= source.occurred_on + timedelta(days=1), StatementDocument.status == "confirmed", StatementDocument.active_parse_run_id == SourceRecord.parse_run_id, ) ) if document.account_id is not None: transfer_query = transfer_query.filter( StatementDocument.account_id != document.account_id ) transfer_matches = transfer_query.all() if len(transfer_matches) == 1: counterpart = transfer_matches[0] link = db.query(SourceRecordLink).filter_by(source_record_id=counterpart.id).one() db.add( DedupCandidate( space_id=document.space_id, source_record_id=source.id, candidate_kind="transfer", existing_source_record_id=counterpart.id, existing_operation_id=link.operation_id, confidence="possible", ) ) pending = True return pending def _match_hold_resolutions( db: Session, *, document: StatementDocument, run: ParseRun ) -> dict[uuid.UUID, uuid.UUID]: """Conservatively match unresolved confirmed-doc HOLDs to new posted records. Candidate pairs are searched only within the same space, mapped account, and statement contract (adapter). A stable external ID wins; otherwise a single unambiguous date/amount/direction/normalized-description fingerprint match is required. Any overlap ambiguity leaves every involved HOLD unresolved (review-required). Resolutions are written only when the new statement is confirmed. """ if run.adapter_id is None or document.account_id is None: return {} holds = ( db.query(SourceRecord) .join( StatementDocument, (StatementDocument.space_id == SourceRecord.space_id) & (StatementDocument.id == SourceRecord.document_id), ) .outerjoin(HoldResolution, HoldResolution.hold_source_record_id == SourceRecord.id) .filter( SourceRecord.space_id == document.space_id, SourceRecord.bank_status == "hold", SourceRecord.parse_run_id != run.id, StatementDocument.status == "confirmed", StatementDocument.active_parse_run_id == SourceRecord.parse_run_id, StatementDocument.adapter_id == run.adapter_id, StatementDocument.account_id == document.account_id, HoldResolution.id.is_(None), ) .all() ) if not holds: return {} posted = ( db.query(SourceRecord) .filter_by(parse_run_id=run.id, bank_status="posted") .filter(SourceRecord.review_decision != "discard") .all() ) by_external: dict[str, list[SourceRecord]] = {} by_fingerprint: dict[tuple[str, str], list[SourceRecord]] = {} for record in posted: if record.external_id: by_external.setdefault(record.external_id, []).append(record) by_fingerprint.setdefault((record.record_fp, record.direction), []).append(record) candidates: dict[uuid.UUID, set[uuid.UUID]] = {} for hold in holds: matched: set[uuid.UUID] = set() if hold.external_id: matched.update(item.id for item in by_external.get(hold.external_id, [])) matched.update(item.id for item in by_fingerprint.get((hold.record_fp, hold.direction), [])) if len(matched) == 1: candidates[hold.id] = matched claimed: dict[uuid.UUID, int] = {} for posted_ids in candidates.values(): for posted_id in posted_ids: claimed[posted_id] = claimed.get(posted_id, 0) + 1 return { hold_id: next(iter(posted_ids)) for hold_id, posted_ids in candidates.items() if all(claimed[posted_id] == 1 for posted_id in posted_ids) } def _apply_hold_matches( db: Session, *, document: StatementDocument, run: ParseRun, user: User, matches: dict[uuid.UUID, uuid.UUID], ) -> None: for hold_id, posted_id in matches.items(): db.add( HoldResolution( space_id=document.space_id, hold_source_record_id=hold_id, resolution="matched_posted", resolving_source_record_id=posted_id, resolved_at=utcnow(), ) ) if matches: db.flush() audit.record( db, event_type="import.holds_matched_posted", space_id=document.space_id, user_id=user.id, detail={ "document_id": str(document.id), "holds": sorted(str(hold_id) for hold_id in matches), }, ) def process_statement_document(document_id: uuid.UUID, space_id: uuid.UUID) -> str: with session_scope(space_id) as db: document = _document(db, space_id=space_id, document_id=document_id) if document.status in {"confirmed", "deleted"}: return document.status document.status = "processing" run = ParseRun(space_id=space_id, document_id=document.id, status="running") db.add(run) db.flush() try: if not all((document.dek_wrapped, document.dek_wrap_nonce, document.content_nonce)): raise AppError( "Document key material is missing", code="IMPORT_DECRYPT_FAILED", status_code=409, ) assert document.dek_wrapped is not None assert document.dek_wrap_nonce is not None assert document.content_nonce is not None raw = decrypt_file( ciphertext=get_blob_store().get(document.object_key), content_nonce=document.content_nonce, dek_wrapped=document.dek_wrapped, wrap_nonce=document.dek_wrap_nonce, ) kind = magic_kind(raw) if kind is None: raise AppError( "Unsupported document content", code="IMPORT_TYPE_INVALID", status_code=415 ) adapter = detect_adapter(raw, kind) draft = adapter.parse(raw) run.adapter_id = adapter.adapter_id run.parser_version = draft.parser_version run.draft = _draft_json(draft) run.status = "succeeded" document.adapter_id = adapter.adapter_id document.account_fingerprint = draft.account_fingerprint document.active_parse_run_id = run.id reconciliation = _reconciliation_summary(draft) document.reconciliation = reconciliation if document.account_id is None and draft.account_fingerprint: matches = ( db.query(Account) .filter_by(space_id=space_id, fingerprint=draft.account_fingerprint) .all() ) if len(matches) == 1: document.account_id = matches[0].id for row in draft.rows: db.add( _source_from_row( space_id=space_id, document_id=document.id, run_id=run.id, row=row ) ) db.flush() has_candidates = _build_candidates(db, document=document, run=run) critical = ( db.query(SourceRecord.id).filter_by(parse_run_id=run.id, critical=True).first() ) document.needs_account_mapping = document.account_id is None needs_manual_reconciliation = not reconciliation["full"] document.status = ( "review_required" if critical or document.needs_account_mapping or needs_manual_reconciliation or has_candidates or reconciliation.get("matched") is False else "ready" ) except AppError as exc: run.status = "failed" run.error_code = exc.code document.status = "failed" document.error_code = exc.code except Exception: run.status = "failed" run.error_code = "IMPORT_PARSE_FAILED" document.status = "failed" document.error_code = "IMPORT_PARSE_FAILED" return document.status def list_documents(db: Session, *, space_id: uuid.UUID) -> list[StatementDocument]: return ( db.query(StatementDocument) .filter_by(space_id=space_id) .order_by(StatementDocument.created_at.desc()) .all() ) def unresolved_holds( db: Session, *, space_id: uuid.UUID, document: StatementDocument ) -> list[SourceRecord]: """Unresolved HOLD rows of the document's active parse run.""" if not document.active_parse_run_id: return [] return ( db.query(SourceRecord) .outerjoin(HoldResolution, HoldResolution.hold_source_record_id == SourceRecord.id) .filter( SourceRecord.space_id == space_id, SourceRecord.parse_run_id == document.active_parse_run_id, SourceRecord.bank_status == "hold", HoldResolution.id.is_(None), ) .all() ) def unresolved_hold_counts( db: Session, *, space_id: uuid.UUID, document_ids: list[uuid.UUID], month: tuple[int, int] | None = None, ) -> dict[uuid.UUID, int]: """Count unresolved HOLD rows for active parse runs in one query.""" if not document_ids: return {} query = ( db.query(StatementDocument.id, func.count(SourceRecord.id)) .join( SourceRecord, (SourceRecord.document_id == StatementDocument.id) & (SourceRecord.parse_run_id == StatementDocument.active_parse_run_id), ) .outerjoin(HoldResolution, HoldResolution.hold_source_record_id == SourceRecord.id) .filter( StatementDocument.space_id == space_id, StatementDocument.id.in_(document_ids), SourceRecord.bank_status == "hold", HoldResolution.id.is_(None), ) ) if month is not None: year, month_number = month start = date(year, month_number, 1) end = date(year + 1, 1, 1) if month_number == 12 else date(year, month_number + 1, 1) query = query.filter(SourceRecord.occurred_on >= start, SourceRecord.occurred_on < end) return {document_id: count for document_id, count in query.group_by(StatementDocument.id)} def parse_month_filter(value: str | None) -> tuple[int, int] | None: if value is None: return None try: year_text, month_text = value.split("-", 1) year, month = int(year_text), int(month_text) except (ValueError, AttributeError) as exc: raise bad_request("IMPORT_MONTH_INVALID", "Month filter must use YYYY-MM") from exc if not 1 <= month <= 12 or year < 2000: raise bad_request("IMPORT_MONTH_INVALID", "Month filter must use YYYY-MM") return year, month def document_detail(db: Session, *, space_id: uuid.UUID, document_id: uuid.UUID) -> dict: document = _document(db, space_id=space_id, document_id=document_id) records = [] if document.active_parse_run_id: records = ( db.query(SourceRecord) .filter_by(parse_run_id=document.active_parse_run_id) .order_by(SourceRecord.created_at) .all() ) record_ids = [record.id for record in records] codes = {record.mcc for record in records if record.mcc is not None} mcc_names: dict[int, str] = { code: title for code, title in ( db.query(MccCode.code, MccCode.title).filter(MccCode.code.in_(codes)).all() if codes else [] ) } resolutions: dict[uuid.UUID, dict[str, Any]] = {} candidates: dict[uuid.UUID, list[dict[str, Any]]] = {} if record_ids: for item in ( db.query(HoldResolution) .filter( HoldResolution.space_id == space_id, HoldResolution.hold_source_record_id.in_(record_ids), ) .all() ): resolutions[item.hold_source_record_id] = { "resolution": item.resolution, "resolving_source_record_id": item.resolving_source_record_id, "operation_id": item.operation_id, "resolved_by_user_id": item.resolved_by_user_id, "reason": item.reason, "resolved_at": item.resolved_at, } for candidate in ( db.query(DedupCandidate) .filter( DedupCandidate.space_id == space_id, DedupCandidate.source_record_id.in_(record_ids), ) .order_by(DedupCandidate.created_at) .all() ): candidates.setdefault(candidate.source_record_id, []).append( { "id": candidate.id, "kind": candidate.candidate_kind, "confidence": candidate.confidence, "resolution": candidate.resolution, "existing_operation_id": candidate.existing_operation_id, } ) return { "id": document.id, "status": document.status, "adapter_id": document.adapter_id, "account_id": document.account_id, "needs_account_mapping": document.needs_account_mapping, "error_code": document.error_code, "reconciliation": document.reconciliation, "records": [ { "id": item.id, "external_id": item.external_id, "occurred_on": item.occurred_on, "posted_on": item.posted_on, "amount_kopecks": item.amount_kopecks, "direction": item.direction, "bank_status": item.bank_status, "description": item.description, "mcc": item.mcc, "mcc_name": mcc_names.get(item.mcc) or (mcc.russian_title(item.mcc) if item.mcc is not None else None), "critical": item.critical, "warnings": item.field_warnings, "review_decision": item.review_decision, "duplicate_operation_id": item.duplicate_operation_id, "hold_resolution": resolutions.get(item.id), "candidates": candidates.get(item.id, []), } for item in records ], } def decide_source_record( db: Session, *, space_id: uuid.UUID, document_id: uuid.UUID, source_record_id: uuid.UUID, decision: str, duplicate_operation_id: uuid.UUID | None, user: User, ) -> None: document = _document(db, space_id=space_id, document_id=document_id) source = ( db.query(SourceRecord) .filter_by( id=source_record_id, space_id=space_id, parse_run_id=document.active_parse_run_id, ) .one_or_none() ) if source is None: raise not_found("IMPORT_SOURCE_RECORD_NOT_FOUND", "Import row not found") if decision not in {"create", "discard", "duplicate", "transfer", "savings_transfer"}: raise conflict("IMPORT_DECISION_INVALID", "Unsupported row decision") if source.bank_status == "hold" and decision != "discard": raise conflict("IMPORT_DECISION_INVALID", "HOLD rows are resolved via hold resolution") if decision == "savings_transfer" and ( source.bank_status != "posted" or source.direction != "out" ): raise conflict( "IMPORT_DECISION_INVALID", "Only posted outflows can be transferred to savings", ) candidates = db.query(DedupCandidate).filter_by(source_record_id=source.id).all() matching = [item for item in candidates if item.candidate_kind == decision] if decision == "duplicate" and duplicate_operation_id is None and len(matching) == 1: duplicate_operation_id = matching[0].existing_operation_id if decision in {"duplicate", "transfer"} and len(matching) != 1: raise conflict("IMPORT_CANDIDATE_AMBIGUOUS", "A unique matching candidate is required") if decision == "duplicate": operation = ( db.query(Operation) .filter_by(id=duplicate_operation_id, space_id=space_id) .one_or_none() ) if operation is None: raise not_found("LEDGER_OPERATION_NOT_FOUND", "Duplicate operation not found") source.review_decision = decision source.duplicate_operation_id = duplicate_operation_id for candidate in candidates: if candidate not in matching: candidate.resolution = "is_new" elif decision == "duplicate": candidate.resolution = "is_duplicate" else: candidate.resolution = "accepted_transfer" candidate.decided_by_user_id = user.id if source.critical and decision in {"discard", "duplicate"}: source.critical = False unresolved = ( db.query(SourceRecord.id) .filter_by(parse_run_id=document.active_parse_run_id, critical=True) .first() ) pending_candidate = ( db.query(DedupCandidate.id) .join(SourceRecord, SourceRecord.id == DedupCandidate.source_record_id) .filter( SourceRecord.parse_run_id == document.active_parse_run_id, DedupCandidate.resolution == "pending", ) .first() ) reconciliation = document.reconciliation or {} reconciled = reconciliation.get("full") and reconciliation.get("matched") is not False reconciled = reconciled or reconciliation.get("confirmed_without_full_reconciliation") if not unresolved and not pending_candidate and document.account_id and reconciled: document.status = "ready" def set_review_confirmation( db: Session, *, space_id: uuid.UUID, document_id: uuid.UUID, account_id: uuid.UUID, confirmed_without_full_reconciliation: bool, reason: str | None = None, ) -> StatementDocument: document = _document(db, space_id=space_id, document_id=document_id) account = db.query(Account).filter_by(id=account_id, space_id=space_id).one_or_none() if account is None: raise not_found("LEDGER_ACCOUNT_NOT_FOUND", "Account not found") if account.system_key is not None: raise conflict( "IMPORT_SYSTEM_ACCOUNT_INVALID", "System accounts cannot be used as statement accounts", ) document.account_id = account.id document.needs_account_mapping = False reconciliation = dict(document.reconciliation or {}) reconciliation["confirmed_without_full_reconciliation"] = confirmed_without_full_reconciliation reconciliation["reason"] = reason.strip() or None if reason is not None else None document.reconciliation = reconciliation critical = ( db.query(SourceRecord.id) .filter_by(parse_run_id=document.active_parse_run_id, critical=True) .first() ) pending_candidate = ( db.query(DedupCandidate.id) .join(SourceRecord, SourceRecord.id == DedupCandidate.source_record_id) .filter( SourceRecord.parse_run_id == document.active_parse_run_id, DedupCandidate.resolution == "pending", ) .first() ) if not critical and not pending_candidate and reconciliation.get("matched") is not False: document.status = "ready" return document def resolve_hold( db: Session, *, space_id: uuid.UUID, document_id: uuid.UUID, source_record_id: uuid.UUID, user: User, action: str, reason: str | None = None, account_id: uuid.UUID | None = None, ) -> HoldResolution: """Manually settle one HOLD: book it as a normal operation or discard it.""" document = _document(db, space_id=space_id, document_id=document_id) if document.status != "confirmed": raise conflict( "IMPORT_HOLD_DOCUMENT_NOT_CONFIRMED", "Holds can only be resolved after the import is confirmed", ) if action not in HOLD_MANUAL_ACTIONS: raise bad_request("IMPORT_HOLD_ACTION_INVALID", "Unknown hold resolution action") reason = reason.strip() or None if reason is not None else None source = ( db.query(SourceRecord) .filter_by( id=source_record_id, space_id=space_id, parse_run_id=document.active_parse_run_id, ) .one_or_none() ) if source is None: raise not_found("IMPORT_HOLD_NOT_FOUND", "Import row not found") if source.bank_status != "hold": raise conflict("IMPORT_HOLD_INVALID", "Only hold rows can be resolved manually") existing = db.query(HoldResolution.id).filter_by(hold_source_record_id=source.id).one_or_none() if existing is not None: raise conflict("IMPORT_HOLD_ALREADY_RESOLVED", "Hold is already resolved") if source.occurred_on is None: raise conflict("IMPORT_HOLD_DATE_MISSING", "Hold has no usable operation date") if source.direction not in {"in", "out"}: raise conflict("IMPORT_HOLD_DIRECTION_INVALID", "Hold direction is invalid") resolution = HoldResolution( space_id=space_id, hold_source_record_id=source.id, resolution="confirmed_as_posted" if action == "confirm_as_posted" else "discarded_as_cancelled", resolved_by_user_id=user.id, reason=reason, resolved_at=utcnow(), ) if action == "confirm_as_posted": mapped_account_id = account_id or document.account_id if mapped_account_id is None: raise conflict( "IMPORT_HOLD_ACCOUNT_REQUIRED", "A mapped account is required to confirm a hold as posted", ) account = db.query(Account).filter_by(id=mapped_account_id, space_id=space_id).one_or_none() if account is None: raise not_found("LEDGER_ACCOUNT_NOT_FOUND", "Account not found") if source.direction == "in": operation = create_income( db, space_id=space_id, user=user, account_id=account.id, amount_kopecks=source.amount_kopecks, occurred_on=source.occurred_on, posted_on=source.posted_on, description=source.description, request_key=f"import:{source.id}", ) else: operation = create_expense( db, space_id=space_id, user=user, account_id=account.id, amount_kopecks=source.amount_kopecks, category_id=_category_id_for_source(db, space_id=space_id, source=source), occurred_on=source.occurred_on, posted_on=source.posted_on, description=source.description, request_key=f"import:{source.id}", ) db.add( SourceRecordLink( space_id=space_id, source_record_id=source.id, operation_id=operation.id, kind="hold_confirmed", ) ) resolution.operation_id = operation.id db.add(resolution) db.flush() audit.record( db, event_type="import.hold_confirmed_as_posted" if action == "confirm_as_posted" else "import.hold_discarded_as_cancelled", space_id=space_id, user_id=user.id, detail={ "document_id": str(document.id), "source_record_id": str(source.id), "reason": reason, }, ) return resolution def confirm_document( db: Session, *, space_id: uuid.UUID, document_id: uuid.UUID, user: User ) -> StatementDocument: document = _document(db, space_id=space_id, document_id=document_id) db.refresh(document, with_for_update=True) if document.status == "confirmed": return document if document.status not in {"ready", "review_required"} or document.account_id is None: raise conflict("IMPORT_NOT_READY", "Import review is not complete") pending = ( db.query(DedupCandidate.id) .join(SourceRecord, SourceRecord.id == DedupCandidate.source_record_id) .filter( SourceRecord.parse_run_id == document.active_parse_run_id, DedupCandidate.resolution == "pending", ) .first() ) if pending: raise conflict("IMPORT_DECISIONS_REQUIRED", "Duplicate or transfer decisions remain") records = db.query(SourceRecord).filter_by(parse_run_id=document.active_parse_run_id).all() if any(item.critical for item in records): raise conflict("IMPORT_FIELDS_UNRESOLVED", "Critical source fields remain unresolved") reconciliation = document.reconciliation or {} if not reconciliation.get("full") and not reconciliation.get( "confirmed_without_full_reconciliation" ): raise conflict("IMPORT_RECONCILIATION_REQUIRED", "Reconciliation must be confirmed") if reconciliation.get("matched") is False: raise conflict("IMPORT_RECONCILIATION_MISMATCH", "Statement arithmetic does not reconcile") if document.status != "ready": raise conflict("IMPORT_NOT_READY", "Import review is not complete") run = db.get(ParseRun, document.active_parse_run_id) if run is not None: _apply_hold_matches( db, document=document, run=run, user=user, matches=_match_hold_resolutions(db, document=document, run=run), ) for source in records: if source.bank_status == "hold": continue if source.review_decision == "discard": continue if source.review_decision == "transfer": candidate = ( db.query(DedupCandidate) .filter_by( source_record_id=source.id, candidate_kind="transfer", resolution="accepted_transfer", ) .one() ) assert candidate.existing_source_record_id is not None counterpart = db.get(SourceRecord, candidate.existing_source_record_id) old_link = ( db.query(SourceRecordLink) .filter_by(source_record_id=candidate.existing_source_record_id) .one() ) old_operation = db.get(Operation, old_link.operation_id) if ( counterpart is None or old_operation is None or old_operation.account_id is None or old_operation.type not in {"income", "expense"} ): raise conflict( "IMPORT_TRANSFER_TARGET_MISSING", "Transfer counterpart is unavailable", ) other_links = ( db.query(SourceRecordLink.id) .filter( SourceRecordLink.operation_id == old_operation.id, SourceRecordLink.id != old_link.id, ) .first() ) if other_links: raise conflict( "IMPORT_TRANSFER_TARGET_AMBIGUOUS", "Counterpart has multiple links", ) ensure_period_open_for_date(db, space_id=space_id, day=old_operation.occurred_on) from app.budgeting.models import ExpectedIncomeMatch has_income_match = ( db.query(ExpectedIncomeMatch.id).filter_by(operation_id=old_operation.id).first() ) has_refund = ( db.query(Operation.id) .filter_by(parent_operation_id=old_operation.id, status="confirmed") .first() ) if has_income_match or has_refund: raise conflict( "IMPORT_TRANSFER_TARGET_HAS_DEPENDENCIES", "Unlink matches and refunds before converting this operation to a transfer", ) candidate.replacement_snapshot = { "type": old_operation.type, "account_id": str(old_operation.account_id), "amount_kopecks": old_operation.amount_kopecks, "occurred_on": old_operation.occurred_on.isoformat(), "posted_on": old_operation.posted_on.isoformat() if old_operation.posted_on else None, "category_id": str(old_operation.category_id) if old_operation.category_id else None, "income_source_id": str(old_operation.income_source_id) if old_operation.income_source_id else None, "payee_id": str(old_operation.payee_id) if old_operation.payee_id else None, "description": old_operation.description, "request_key": old_operation.request_key, "link_kind": old_link.kind, } source_account_id = ( document.account_id if source.direction == "out" else old_operation.account_id ) target_account_id = ( old_operation.account_id if source.direction == "out" else document.account_id ) db.query(DedupCandidate).filter_by(existing_operation_id=old_operation.id).update( {DedupCandidate.existing_operation_id: None} ) db.delete(old_link) db.delete(old_operation) db.flush() assert source.occurred_on is not None operation = create_transfer( db, space_id=space_id, user=user, source_account_id=source_account_id, target_account_id=target_account_id, amount_kopecks=source.amount_kopecks, occurred_on=source.occurred_on, posted_on=source.posted_on, description=source.description, request_key=f"import-transfer:{source.id}", ) db.add_all( [ SourceRecordLink( space_id=space_id, source_record_id=source.id, operation_id=operation.id, kind="transfer", ), SourceRecordLink( space_id=space_id, source_record_id=counterpart.id, operation_id=operation.id, kind="transfer", ), ] ) continue if source.review_decision == "savings_transfer": from app.ledger.savings import ensure_savings_resources savings_account, _savings_category = ensure_savings_resources(db, space_id=space_id) if document.account_id == savings_account.id: raise conflict( "IMPORT_SAVINGS_SOURCE_INVALID", "The mapped statement account cannot be the savings account itself", ) assert source.occurred_on is not None operation = create_transfer( db, space_id=space_id, user=user, source_account_id=document.account_id, target_account_id=savings_account.id, amount_kopecks=source.amount_kopecks, occurred_on=source.occurred_on, posted_on=source.posted_on, description=source.description, request_key=f"import-savings:{source.id}", ) db.add( SourceRecordLink( space_id=space_id, source_record_id=source.id, operation_id=operation.id, kind="created", ) ) continue if source.review_decision == "duplicate" and source.duplicate_operation_id: duplicate_operation = ( db.query(Operation) .filter_by(id=source.duplicate_operation_id, space_id=space_id) .one_or_none() ) if duplicate_operation is None: raise conflict( "IMPORT_DUPLICATE_TARGET_MISSING", "Duplicate target no longer exists", ) db.add( SourceRecordLink( space_id=space_id, source_record_id=source.id, operation_id=duplicate_operation.id, kind="duplicate", ) ) continue existing_links = ( db.query(SourceRecordLink) .join(SourceRecord, SourceRecord.id == SourceRecordLink.source_record_id) .filter( SourceRecord.space_id == space_id, SourceRecord.external_id == source.external_id, SourceRecord.external_id.is_not(None), ) .all() if source.external_id else [] ) linked_operation = ( db.get(Operation, existing_links[0].operation_id) if len(existing_links) == 1 else None ) kind = "duplicate" if linked_operation is None: assert source.occurred_on is not None if source.direction == "in": linked_operation = create_income( db, space_id=space_id, user=user, account_id=document.account_id, amount_kopecks=source.amount_kopecks, occurred_on=source.occurred_on, posted_on=source.posted_on, description=source.description, request_key=f"import:{source.id}", ) else: linked_operation = create_expense( db, space_id=space_id, user=user, account_id=document.account_id, amount_kopecks=source.amount_kopecks, category_id=_category_id_for_source(db, space_id=space_id, source=source), occurred_on=source.occurred_on, posted_on=source.posted_on, description=source.description, request_key=f"import:{source.id}", ) kind = "created" db.add( SourceRecordLink( space_id=space_id, source_record_id=source.id, operation_id=linked_operation.id, kind=kind, ) ) document.status = "confirmed" audit.record( db, event_type="import.confirmed", space_id=space_id, user_id=user.id, detail={"document_id": str(document.id), "records": len(records)}, ) return document def prepare_reparse( db: Session, *, space_id: uuid.UUID, document_id: uuid.UUID, user: User ) -> StatementDocument: document = _document(db, space_id=space_id, document_id=document_id) if document.status == "confirmed": raise conflict( "IMPORT_REPARSE_CONFIRMED", "Revert the confirmed import before reparsing its source", ) if document.status == "deleted": raise conflict("IMPORT_SOURCE_DELETED", "Deleted source files cannot be reparsed") if document.status in {"queued", "processing"}: return document document.status = "queued" document.error_code = None audit.record( db, event_type="import.reparse_queued", space_id=space_id, user_id=user.id, detail={"document_id": str(document.id)}, ) return document def revert_document( db: Session, *, space_id: uuid.UUID, document_id: uuid.UUID, user: User ) -> StatementDocument: from app.budgeting.models import BudgetPeriod document = _document(db, space_id=space_id, document_id=document_id) if document.status != "confirmed": return document if document.active_parse_run_id: db.query(HoldResolution).filter( HoldResolution.space_id == space_id, HoldResolution.resolution == "matched_posted", HoldResolution.resolving_source_record_id.in_( db.query(SourceRecord.id).filter_by(parse_run_id=document.active_parse_run_id) ), ).delete(synchronize_session=False) links = ( db.query(SourceRecordLink) .join(SourceRecord, SourceRecord.id == SourceRecordLink.source_record_id) .filter( SourceRecord.parse_run_id == document.active_parse_run_id, SourceRecordLink.kind != "hold_confirmed", ) .all() ) transfer_candidates = ( db.query(DedupCandidate) .join(SourceRecord, SourceRecord.id == DedupCandidate.source_record_id) .filter( SourceRecord.parse_run_id == document.active_parse_run_id, DedupCandidate.candidate_kind == "transfer", DedupCandidate.resolution == "accepted_transfer", ) .all() ) created = [db.get(Operation, link.operation_id) for link in links if link.kind == "created"] transfer_operations = [ db.get(Operation, link.operation_id) for link in links if link.kind == "transfer" ] affected = { (operation.occurred_on.year, operation.occurred_on.month) for operation in created + transfer_operations if operation is not None } for candidate in transfer_candidates: candidate_snapshot = candidate.replacement_snapshot or {} if candidate_snapshot.get("occurred_on"): original_day = date.fromisoformat(candidate_snapshot["occurred_on"]) affected.add((original_day.year, original_day.month)) closed = ( db.query(BudgetPeriod) .filter( BudgetPeriod.space_id == space_id, BudgetPeriod.status == "closed", ) .all() ) impacted = [item for item in closed if (item.year, item.month) in affected] if impacted: raise conflict( "IMPORT_REVERT_REQUIRES_REOPEN", "Affected budget periods must be reopened before reverting", details={"period_ids": [str(item.id) for item in impacted]}, ) created_ids = [ operation.id for operation in created + transfer_operations if operation is not None ] dependent_refund = ( db.query(Operation.id) .filter(Operation.parent_operation_id.in_(created_ids), Operation.status == "confirmed") .first() if created_ids else None ) if dependent_refund: raise conflict( "IMPORT_REVERT_HAS_DEPENDENCIES", "Void dependent refunds before reverting this import", ) handled_link_ids: set[uuid.UUID] = set() for candidate in transfer_candidates: snapshot = candidate.replacement_snapshot if not snapshot or candidate.existing_source_record_id is None: raise conflict( "IMPORT_REVERT_SNAPSHOT_MISSING", "Transfer replacement metadata is unavailable", ) current_link = ( db.query(SourceRecordLink) .filter_by(source_record_id=candidate.source_record_id, kind="transfer") .one() ) transfer_operation = db.get(Operation, current_link.operation_id) if transfer_operation is None: raise conflict("IMPORT_REVERT_OPERATION_MISSING", "Transfer operation is unavailable") transfer_links = ( db.query(SourceRecordLink) .filter_by(operation_id=transfer_operation.id, kind="transfer") .all() ) handled_link_ids.update(link.id for link in transfer_links) for transfer_link in transfer_links: db.delete(transfer_link) db.delete(transfer_operation) db.flush() account_id = uuid.UUID(snapshot["account_id"]) occurred_on = date.fromisoformat(snapshot["occurred_on"]) posted_on = date.fromisoformat(snapshot["posted_on"]) if snapshot["posted_on"] else None if snapshot["type"] == "income": restored = create_income( db, space_id=space_id, user=user, account_id=account_id, amount_kopecks=int(snapshot["amount_kopecks"]), occurred_on=occurred_on, posted_on=posted_on, income_source_id=( uuid.UUID(snapshot["income_source_id"]) if snapshot["income_source_id"] else None ), description=snapshot["description"], request_key=snapshot["request_key"], ) else: restored = create_expense( db, space_id=space_id, user=user, account_id=account_id, amount_kopecks=int(snapshot["amount_kopecks"]), occurred_on=occurred_on, posted_on=posted_on, category_id=( uuid.UUID(snapshot["category_id"]) if snapshot["category_id"] else None ), payee_id=uuid.UUID(snapshot["payee_id"]) if snapshot["payee_id"] else None, description=snapshot["description"], request_key=snapshot["request_key"], ) db.add( SourceRecordLink( space_id=space_id, source_record_id=candidate.existing_source_record_id, operation_id=restored.id, kind=snapshot["link_kind"], ) ) for link in links: if link.id in handled_link_ids: continue operation = db.get(Operation, link.operation_id) db.delete(link) if link.kind == "created" and operation is not None: db.delete(operation) document.status = "ready" audit.record( db, event_type="import.reverted", space_id=space_id, user_id=user.id, detail={"document_id": str(document.id)}, ) return document def delete_source_file( db: Session, *, space_id: uuid.UUID, document_id: uuid.UUID, user: User ) -> StatementDocument: document = _document(db, space_id=space_id, document_id=document_id) document.dek_wrapped = None document.dek_wrap_nonce = None document.content_nonce = None document.status = "deleted" audit.record( db, event_type="import.source_deleted", space_id=space_id, user_id=user.id, detail={"document_id": str(document.id)}, ) return document