diff --git a/server.py b/server.py index 1c1dc43..1dda4ee 100644 --- a/server.py +++ b/server.py @@ -18,7 +18,7 @@ from bank_importer import ( manual_records, master_data, matching, multipart, period_close, personal_transit, positions, reminders, settings, subjects, ) -from bank_importer.db import connect, migrate, utc_now +from bank_importer.db import connect, migrate, transaction, utc_now ROOT = Path(__file__).resolve().parent @@ -2615,22 +2615,22 @@ class AppHandler(SimpleHTTPRequestHandler): ) return try: - result = matching.reconcile_rows(connection, writable, actor=user) + with transaction(connection): + result = matching.reconcile_rows(connection, writable, actor=user) + ledger_events.reconcile_bank_events(connection, actor=user) + result["late_arrivals"] = late + auth.audit( + connection, "transfer_reconcile", actor=user, + target=f"rows:{len(writable)}", + detail=( + f"created:{result['created_events']};" + f"updated:{result['updated_events']};late:{late}" + ), + ip=self._client_ip, + ) except Exception as exc: self._send_json(500, {"status": "error", "message": f"重跑匹配失败：{exc}"}) return - try: - ledger_events.reconcile_bank_events(connection, actor=user) - except Exception as exc: - self._send_json(500, {"status": "error", "message": f"同步往来事件失败：{exc}"}) - return - result["late_arrivals"] = late - auth.audit( - connection, "transfer_reconcile", actor=user, - target=f"rows:{len(writable)}", - detail=f"created:{result['created_events']};updated:{result['updated_events']};late:{late}", - ip=self._client_ip, - ) self._send_json(200, {"status": "ok", "matching": result, "late_arrivals": late}) finally: connection.close() @@ -2660,19 +2660,21 @@ class AppHandler(SimpleHTTPRequestHandler): return try: period_close.assert_event_writable(connection, event_id) - payload = matching.apply_manual_decision( - connection, - event_id, - action, - reason=str(data.get("reason") or ""), - expected_revision=expected_revision, - request_key=str(data.get("request_key") or "") or None, - actor=user, - source_row_ids=[int(item) for item in source_row_ids] - if source_row_ids - else None, - participant=data.get("participant"), - ) + with transaction(connection): + payload = matching.apply_manual_decision( + connection, + event_id, + action, + reason=str(data.get("reason") or ""), + expected_revision=expected_revision, + request_key=str(data.get("request_key") or "") or None, + actor=user, + source_row_ids=[int(item) for item in source_row_ids] + if source_row_ids + else None, + participant=data.get("participant"), + ) + ledger_events.reconcile_bank_events(connection, actor=user) except period_close.PeriodLockedError as exc: self._send_json(409, {"status": "error", "message": str(exc), "year_month": exc.year_month}) return @@ -2682,8 +2684,6 @@ class AppHandler(SimpleHTTPRequestHandler): except matching.MatchInputError as exc: self._send_json(400, {"status": "error", "message": str(exc)}) return - try: - ledger_events.reconcile_bank_events(connection, actor=user) except Exception as exc: self._send_json(500, {"status": "error", "message": f"同步往来事件失败：{exc}"}) return @@ -2976,19 +2976,21 @@ class AppHandler(SimpleHTTPRequestHandler): reason = str(data.get("reason") or "").strip() or "公司端确认单边流水" try: period_close.assert_event_writable(connection, event_id) - payload = matching.apply_manual_decision( - connection, - event_id, - "assign_participant", - reason=reason, - expected_revision=expected_revision, - request_key=request_key, - actor=user, - participant={ - "role": role, - "company_id": counterparty_company_id, - }, - ) + with transaction(connection): + payload = matching.apply_manual_decision( + connection, + event_id, + "assign_participant", + reason=reason, + expected_revision=expected_revision, + request_key=request_key, + actor=user, + participant={ + "role": role, + "company_id": counterparty_company_id, + }, + ) + ledger_events.reconcile_bank_events(connection, actor=user) except period_close.PeriodLockedError as exc: self._send_json(409, {"status": "error", "message": str(exc), "year_month": exc.year_month}) return @@ -2998,8 +3000,6 @@ class AppHandler(SimpleHTTPRequestHandler): except matching.MatchInputError as exc: self._send_json(400, {"status": "error", "message": str(exc)}) return - try: - ledger_events.reconcile_bank_events(connection, actor=user) except Exception as exc: self._send_json(500, {"status": "error", "message": f"同步往来事件失败：{exc}"}) return @@ -3417,51 +3417,56 @@ class AppHandler(SimpleHTTPRequestHandler): period_close.assert_ledger_writable(connection, event_id) if action in ("adjust", "reverse") and data.get("effective_at"): period_close.assert_date_writable(connection, str(data.get("effective_at"))) - if action == "reverse": - event_id, _revision_id = ledger_events.create_reversal( - connection, event_id, - source_kind="adjustment", - source_revision_token=None, - effective_at=data.get("effective_at") or None, - reason=reason, actor=user, idempotency_key=request_key, - ) - outcome: dict[str, object] = { - "action": "reverse", "ledger_event_id": event_id, - } - elif action == "adjust": - try: - effective_at = str(data.get("effective_at") or "") - amount = str(data.get("amount") or "") - currency = str(data.get("currency") or "") - payer = int(data["payer_company_id"]) - payee = int(data["payee_company_id"]) - perspective = int(data["perspective_company_id"]) - subject_code = str(data.get("subject_code") or "") - except (KeyError, TypeError, ValueError): - self._send_json( - 400, {"status": "error", "message": "adjust 参数不完整或无效。"} - ) - return - event_id, _revision_id = ledger_events.create_adjustment( - connection, event_id, - effective_at=effective_at, amount=amount, currency=currency, - payer_company_id=payer, payee_company_id=payee, - perspective_company_id=perspective, subject_code=subject_code, - reason=reason, actor=user, idempotency_key=request_key, - ) - outcome = {"action": "adjust", "ledger_event_id": event_id} - elif action == "reopen": - event_id, _revision_id = ledger_events.reopen_subject( - connection, event_id, reason=reason, actor=user, - idempotency_key=request_key, - ) - outcome = {"action": "reopen", "ledger_event_id": event_id} - else: + if action not in ("reverse", "adjust", "reopen"): self._send_json( 400, {"status": "error", "message": "action 必须是 reverse、adjust 或 reopen。"}, ) return + with transaction(connection): + if action == "reverse": + event_id, _revision_id = ledger_events.create_reversal( + connection, event_id, + source_kind="adjustment", + source_revision_token=None, + effective_at=data.get("effective_at") or None, + reason=reason, actor=user, idempotency_key=request_key, + ) + outcome = { + "action": "reverse", "ledger_event_id": event_id, + } + elif action == "adjust": + try: + effective_at = str(data.get("effective_at") or "") + amount = str(data.get("amount") or "") + currency = str(data.get("currency") or "") + payer = int(data["payer_company_id"]) + payee = int(data["payee_company_id"]) + perspective = int(data["perspective_company_id"]) + subject_code = str(data.get("subject_code") or "") + except (KeyError, TypeError, ValueError): + self._send_json( + 400, {"status": "error", "message": "adjust 参数不完整或无效。"} + ) + return + event_id, _revision_id = ledger_events.create_adjustment( + connection, event_id, + effective_at=effective_at, amount=amount, currency=currency, + payer_company_id=payer, payee_company_id=payee, + perspective_company_id=perspective, subject_code=subject_code, + reason=reason, actor=user, idempotency_key=request_key, + ) + outcome = {"action": "adjust", "ledger_event_id": event_id} + else: + event_id, _revision_id = ledger_events.reopen_subject( + connection, event_id, reason=reason, actor=user, + idempotency_key=request_key, + ) + outcome = {"action": "reopen", "ledger_event_id": event_id} + auth.audit( + connection, f"ledger_{action}", actor=user, + target=f"ledger_event:{event_id}", detail=reason, ip=self._client_ip, + ) except period_close.PeriodLockedError as exc: self._send_json(409, {"status": "error", "message": str(exc), "year_month": exc.year_month}) return @@ -3471,10 +3476,6 @@ class AppHandler(SimpleHTTPRequestHandler): except ledger_events.LedgerInputError as exc: self._send_json(400, {"status": "error", "message": str(exc)}) return - auth.audit( - connection, f"ledger_{action}", actor=user, - target=f"ledger_event:{event_id}", detail=reason, ip=self._client_ip, - ) self._send_json(200, {"status": "ok", **outcome}) finally: connection.close() diff --git a/src/bank_importer/auth.py b/src/bank_importer/auth.py index 3af1bc9..8db11d9 100644 --- a/src/bank_importer/auth.py +++ b/src/bank_importer/auth.py @@ -18,7 +18,7 @@ import secrets import sqlite3 import string -from .db import utc_now +from .db import transaction, utc_now MIN_PASSWORD_LENGTH = 8 @@ -308,7 +308,7 @@ def audit( ip: str | None = None, ) -> None: """Append an audit log entry. Never pass passwords in ``detail``.""" - with connection: + with transaction(connection): connection.execute( """ INSERT INTO audit_log ( diff --git a/src/bank_importer/calculation.py b/src/bank_importer/calculation.py index b2d7da9..6fccf84 100644 --- a/src/bank_importer/calculation.py +++ b/src/bank_importer/calculation.py @@ -7,8 +7,8 @@ from decimal import Decimal, InvalidOperation import json import sqlite3 -from .db import utc_now -from . import master_data, matching +from .db import transaction, utc_now +from . import auth, master_data, matching SETTING_START_DATE = "calculation_start_date" @@ -81,7 +81,7 @@ def set_calculation_start_date( raise LockedError("已有结账月份，起算日已锁定。") before = get_calculation_start_date(connection) now = utc_now() - with connection: + with transaction(connection): connection.execute( """ INSERT INTO system_settings (key, value, updated_at, updated_by) @@ -185,7 +185,7 @@ def create_opening_balance( raise ConflictError("该对公司已有确认期初，请使用修订。") revision = _next_revision(connection, low_id, high_id) now = utc_now() - with connection: + with transaction(connection): cursor = connection.execute( """ INSERT INTO opening_balance_revisions ( @@ -235,7 +235,7 @@ def confirm_opening_balance( reason = str(reason or "").strip() if len(reason) < 2: raise ValueError("确认期初必须填写原因。") - with connection: + with transaction(connection): connection.execute( """ UPDATE opening_balance_revisions SET status = 'confirmed', reason = ? @@ -278,7 +278,7 @@ def revise_opening_balance( high_id = int(row["company_id_high"]) revision = _next_revision(connection, low_id, high_id) now = utc_now() - with connection: + with transaction(connection): connection.execute( "UPDATE opening_balance_revisions SET status = 'superseded' WHERE id = ?", (revision_id,), @@ -333,7 +333,7 @@ def void_opening_balance( reason = str(reason or "").strip() if len(reason) < 2: raise ValueError("作废期初必须填写原因。") - with connection: + with transaction(connection): connection.execute( "UPDATE opening_balance_revisions SET status = 'void', reason = ? WHERE id = ?", (reason, revision_id), @@ -551,7 +551,7 @@ def recalculate_coverage_gaps(connection: sqlite3.Connection) -> int: "SELECT * FROM bank_accounts WHERE status = 'active'" ).fetchall() rebuilt = 0 - with connection: + with transaction(connection): for account in accounts: connection.execute( """ @@ -668,7 +668,7 @@ def submit_no_business_attestation( if account["company_id"] != company_id: raise ValueError("只能为本公司账户提交说明。") now = utc_now() - with connection: + with transaction(connection): cursor = connection.execute( """ INSERT INTO no_business_attestations ( @@ -688,6 +688,13 @@ def submit_no_business_attestation( ), ) attestation_id = int(cursor.lastrowid) + auth.audit( + connection, + "attestation_submit", + actor=actor, + target=f"attestation:{attestation_id}", + detail=f"account:{bank_account_id};gap:{gap_start}..{gap_end}", + ) return attestation_payload(connection, attestation_id) @@ -712,7 +719,7 @@ def review_no_business_attestation( raise ValueError("审核必须填写理由。") status = "approved" if decision == "approve" else "rejected" now = utc_now() - with connection: + with transaction(connection): connection.execute( """ UPDATE no_business_attestations @@ -748,6 +755,13 @@ def review_no_business_attestation( """, (row["bank_account_id"], row["gap_end"], row["gap_start"]), ) + auth.audit( + connection, + f"attestation_{decision}", + actor=actor, + target=f"attestation:{attestation_id}", + detail=review_reason, + ) return attestation_payload(connection, attestation_id) diff --git a/src/bank_importer/db.py b/src/bank_importer/db.py index 60c5e4c..3059b28 100644 --- a/src/bank_importer/db.py +++ b/src/bank_importer/db.py @@ -10,10 +10,12 @@ version order; each records itself in ``schema_migrations`` so re-running from __future__ import annotations import argparse +from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path import sqlite3 +from typing import Iterator DEFAULT_DB_PATH = Path("data/app.db") @@ -23,6 +25,31 @@ def utc_now() -> str: return datetime.now(timezone.utc).isoformat() +@contextmanager +def transaction(connection: sqlite3.Connection) -> Iterator[sqlite3.Connection]: + """Own a write transaction only when the caller has not already started one. + + Nested helpers join the outer boundary so business rows and their audit + trail commit or roll back together. Standalone callers still commit before + return, so ``connection.close()`` cannot silently drop the work (HEL-270). + ``sqlite3.Connection`` as a context manager always commits on exit even + when it did not begin the transaction; do not use it for nestable writes. + """ + began = False + if not connection.in_transaction: + connection.execute("BEGIN IMMEDIATE") + began = True + try: + yield connection + except Exception: + if began: + connection.rollback() + raise + else: + if began: + connection.commit() + + @dataclass(frozen=True) class Migration: version: int diff --git a/src/bank_importer/importing.py b/src/bank_importer/importing.py index 291fe9d..4a099ce 100644 --- a/src/bank_importer/importing.py +++ b/src/bank_importer/importing.py @@ -650,6 +650,14 @@ def review_sheets( actor=actor, ) ledger_events.reconcile_bank_events(connection, actor=actor) + if updated: + auth.audit( + connection, + f"sheet_{decision}", + actor=actor, + target=f"batch:{batch_id}", + detail=f"sheets:{','.join(updated)}" + (f";reason:{reason}" if reason else ""), + ) if began: connection.commit() except Exception: @@ -657,14 +665,6 @@ def review_sheets( connection.rollback() raise - if updated: - auth.audit( - connection, - f"sheet_{decision}", - actor=actor, - target=f"batch:{batch_id}", - detail=f"sheets:{','.join(updated)}" + (f";reason:{reason}" if reason else ""), - ) payload: dict[str, object] = {"updated": updated, "already": already} if matching_result is not None: payload["matching"] = matching_result diff --git a/src/bank_importer/ledger_events.py b/src/bank_importer/ledger_events.py index 694facc..8e4d7bb 100644 --- a/src/bank_importer/ledger_events.py +++ b/src/bank_importer/ledger_events.py @@ -18,7 +18,7 @@ from decimal import Decimal, InvalidOperation import json import sqlite3 -from .db import utc_now +from .db import transaction, utc_now from .subjects import MIRROR, SUBJECTS, mirror_subject @@ -428,26 +428,30 @@ def reopen_subject( raise LedgerInputError( "该事件没有银行来源，无法重新进入科目审核；请改用调整或冲销。" ) - if not _has_reversal(connection, ledger_event_id): - create_reversal( - connection, ledger_event_id, - source_kind=current["source_kind"], - source_revision_token=current["source_revision_token"], - reason="科目复核：原确认事件冲销", - actor=actor, - idempotency_key=(idempotency_key + ":rev" if idempotency_key else None), - rule_version=current["rule_version"], - ) ev = connection.execute( "SELECT * FROM eligible_intercompany_events WHERE event_id = ?", (bank_claim["bank_event_id"],), ).fetchone() if ev is None: raise LedgerInputError("银行事件已不再纳入往来，无法重新入账。") - event_id, revision_id = _create_bank_event( - connection, ev, actor, reason="科目复核后重新入账，待确认科目", - replacing_claim=bank_claim, - ) + # One nestable transaction: reversal, replacement event, source re-claim + # and suggestions commit together. create_event used to commit on its own, + # leaving the bank-source UPDATE uncommitted for connection.close(). + with transaction(connection): + if not _has_reversal(connection, ledger_event_id): + create_reversal( + connection, ledger_event_id, + source_kind=current["source_kind"], + source_revision_token=current["source_revision_token"], + reason="科目复核：原确认事件冲销", + actor=actor, + idempotency_key=(idempotency_key + ":rev" if idempotency_key else None), + rule_version=current["rule_version"], + ) + event_id, revision_id = _create_bank_event( + connection, ev, actor, reason="科目复核后重新入账，待确认科目", + replacing_claim=bank_claim, + ) return event_id, revision_id @@ -553,51 +557,52 @@ def _create_bank_event( reason: str, replacing_claim: sqlite3.Row | None = None, ) -> tuple[int, int]: - event_id, revision_id = create_event( - connection, - state="pending_subject", - effective_at=event["effective_at"], - amount=event["amount"], - currency=event["currency"], - payer_company_id=event["payer_company_id"], - payee_company_id=event["payee_company_id"], - perspective_company_id=None, - subject_code=None, - source_kind="bank", - source_revision_token=event["decision_id"], - posting_kind="normal", - rule_version=SUBJECT_RULE_VERSION, - evidence_json=json.dumps( - { - "bank_event_id": event["event_id"], - "decision_id": event["decision_id"], - "pairing": event["pairing"], - "evidence_count": event["evidence_count"], - }, - ensure_ascii=False, - ), - actor=actor, - reason=reason, - ) - if replacing_claim is not None: - connection.execute( - """ - UPDATE ledger_event_bank_sources SET ledger_event_id = ? - WHERE bank_event_id = ? - """, - (event_id, event["event_id"]), + with transaction(connection): + event_id, revision_id = create_event( + connection, + state="pending_subject", + effective_at=event["effective_at"], + amount=event["amount"], + currency=event["currency"], + payer_company_id=event["payer_company_id"], + payee_company_id=event["payee_company_id"], + perspective_company_id=None, + subject_code=None, + source_kind="bank", + source_revision_token=event["decision_id"], + posting_kind="normal", + rule_version=SUBJECT_RULE_VERSION, + evidence_json=json.dumps( + { + "bank_event_id": event["event_id"], + "decision_id": event["decision_id"], + "pairing": event["pairing"], + "evidence_count": event["evidence_count"], + }, + ensure_ascii=False, + ), + actor=actor, + reason=reason, ) - else: - connection.execute( - """ - INSERT INTO ledger_event_bank_sources (bank_event_id, ledger_event_id) - VALUES (?, ?) - """, - (event["event_id"], event_id), - ) - from .subjects import store_suggestions + if replacing_claim is not None: + connection.execute( + """ + UPDATE ledger_event_bank_sources SET ledger_event_id = ? + WHERE bank_event_id = ? + """, + (event_id, event["event_id"]), + ) + else: + connection.execute( + """ + INSERT INTO ledger_event_bank_sources (bank_event_id, ledger_event_id) + VALUES (?, ?) + """, + (event["event_id"], event_id), + ) + from .subjects import store_suggestions - store_suggestions(connection, event_id) + store_suggestions(connection, event_id) return event_id, revision_id diff --git a/src/bank_importer/master_data.py b/src/bank_importer/master_data.py index 60cddbd..30bb8fd 100644 --- a/src/bank_importer/master_data.py +++ b/src/bank_importer/master_data.py @@ -17,7 +17,7 @@ import json import re import sqlite3 -from .db import utc_now +from .db import transaction, utc_now ACCOUNT_TYPES = ("基本户", "一般户", "专用户") @@ -143,7 +143,7 @@ def create_company( raise ValueError("公司名称不能为空。") now = utc_now() try: - with connection: + with transaction(connection): cursor = connection.execute( """ INSERT INTO companies ( @@ -153,16 +153,15 @@ def create_company( (name, (credit_code or "").strip() or None, (cashier_name or "").strip() or None, now, now), ) + company_id = int(cursor.lastrowid) + record_change( + connection, "company", company_id, "create", + None, {"name": name, "credit_code": credit_code or None, + "cashier_name": cashier_name or None, "status": "active"}, + None, actor, + ) except sqlite3.IntegrityError as exc: raise ConflictError("公司名称已存在。") from exc - company_id = int(cursor.lastrowid) - with connection: - record_change( - connection, "company", company_id, "create", - None, {"name": name, "credit_code": credit_code or None, - "cashier_name": cashier_name or None, "status": "active"}, - None, actor, - ) return company_id @@ -210,7 +209,7 @@ def submit_bank_account( if existing is None: try: - with connection: + with transaction(connection): cursor = connection.execute( """ INSERT INTO bank_accounts ( @@ -222,23 +221,22 @@ def submit_bank_account( (company_id, number, holder, bank, kind, requested_from, actor["id"] if actor else None, now, now), ) + account_id = int(cursor.lastrowid) + record_change( + connection, "bank_account", account_id, "submit", None, + {"company_id": company_id, "account_number": number, + "bank_name": bank, "account_type": kind, "status": "pending", + "effective_from": requested_from}, + None, actor, + ) except sqlite3.IntegrityError as exc: # Lost a concurrent-insert race on the UNIQUE constraint. raise ConflictError("该银行账号已登记，请等待现有申请处理。") from exc - account_id = int(cursor.lastrowid) - with connection: - record_change( - connection, "bank_account", account_id, "submit", None, - {"company_id": company_id, "account_number": number, - "bank_name": bank, "account_type": kind, "status": "pending", - "effective_from": requested_from}, - None, actor, - ) return get_account(connection, account_id) if existing["status"] == "returned" and existing["company_id"] == company_id: before = _snapshot(existing) - with connection: + with transaction(connection): connection.execute( """ UPDATE bank_accounts @@ -287,7 +285,7 @@ def review_bank_account( if account["status"] != "pending": raise ConflictError("只有待复核的账户可以审核通过。") start = validate_date(effective_from, "启用日期") or account["effective_from"] or today - with connection: + with transaction(connection): connection.execute( """ UPDATE bank_accounts @@ -306,7 +304,7 @@ def review_bank_account( raise ConflictError("只有待复核的账户可以退回。") if reason is None: raise ValueError("退回必须填写原因。") - with connection: + with transaction(connection): connection.execute( """ UPDATE bank_accounts @@ -326,7 +324,7 @@ def review_bank_account( if reason is None: raise ValueError("停用必须填写原因。") end = validate_date(effective_to, "停用日期") or today - with connection: + with transaction(connection): connection.execute( """ UPDATE bank_accounts @@ -448,7 +446,7 @@ def add_alias( if start and end and end < start: raise ValueError("别名失效日期不能早于生效日期。") try: - with connection: + with transaction(connection): cursor = connection.execute( """ INSERT INTO account_aliases ( @@ -459,17 +457,16 @@ def add_alias( (account_id, alias_kind, value, rank, start, end, actor["id"] if actor else None, utc_now()), ) + alias_id = int(cursor.lastrowid) + record_change( + connection, "account_alias", alias_id, "create", None, + {"bank_account_id": account_id, "alias_kind": alias_kind, + "alias_value": value, "priority": rank, + "effective_from": start, "effective_to": end}, + None, actor, + ) except sqlite3.IntegrityError as exc: raise ConflictError("该账户下相同别名已存在。") from exc - alias_id = int(cursor.lastrowid) - with connection: - record_change( - connection, "account_alias", alias_id, "create", None, - {"bank_account_id": account_id, "alias_kind": alias_kind, - "alias_value": value, "priority": rank, - "effective_from": start, "effective_to": end}, - None, actor, - ) return alias_id diff --git a/src/bank_importer/matching.py b/src/bank_importer/matching.py index 1d098b9..45ecc9a 100644 --- a/src/bank_importer/matching.py +++ b/src/bank_importer/matching.py @@ -33,7 +33,7 @@ import re import sqlite3 from .auth import audit -from .db import utc_now +from .db import transaction, utc_now from .master_data import ( is_identifiable, normalize_account_number, @@ -1457,36 +1457,36 @@ def rebuild_current_projection(connection: sqlite3.Connection) -> int: no current pointer and no claims. Returns the number of current decisions rebuilt. Intended as a recovery/consistency entry point. """ - connection.execute("DELETE FROM transfer_observation_claims") - connection.execute("DELETE FROM current_transfer_decisions") - events = connection.execute( - """ - SELECT e.id AS event_id, - (SELECT d2.id FROM transfer_match_decisions d2 - WHERE d2.event_id = e.id - ORDER BY d2.revision DESC LIMIT 1) AS latest_id - FROM canonical_transfer_events e - WHERE e.lifecycle = 'active' - """ - ).fetchall() rebuilt = 0 - for event in events: - if event["latest_id"] is None: - continue - latest = connection.execute( - "SELECT mode FROM transfer_match_decisions WHERE id = ?", - (event["latest_id"],), - ).fetchone() - if latest is None or latest["mode"] == MODE_REVERSAL: - continue - observations = connection.execute( + with transaction(connection): + connection.execute("DELETE FROM transfer_observation_claims") + connection.execute("DELETE FROM current_transfer_decisions") + events = connection.execute( """ - SELECT source_row_id FROM transfer_decision_observations - WHERE decision_id = ? ORDER BY id + SELECT e.id AS event_id, + (SELECT d2.id FROM transfer_match_decisions d2 + WHERE d2.event_id = e.id + ORDER BY d2.revision DESC LIMIT 1) AS latest_id + FROM canonical_transfer_events e + WHERE e.lifecycle = 'active' """, - (event["latest_id"],), ).fetchall() - with connection: + for event in events: + if event["latest_id"] is None: + continue + latest = connection.execute( + "SELECT mode FROM transfer_match_decisions WHERE id = ?", + (event["latest_id"],), + ).fetchone() + if latest is None or latest["mode"] == MODE_REVERSAL: + continue + observations = connection.execute( + """ + SELECT source_row_id FROM transfer_decision_observations + WHERE decision_id = ? ORDER BY id + """, + (event["latest_id"],), + ).fetchall() connection.execute( """ INSERT OR REPLACE INTO current_transfer_decisions (event_id, decision_id) @@ -1502,7 +1502,7 @@ def rebuild_current_projection(connection: sqlite3.Connection) -> int: """, (observation["source_row_id"], event["event_id"], event["latest_id"]), ) - rebuilt += 1 + rebuilt += 1 return rebuilt diff --git a/src/bank_importer/period_close.py b/src/bank_importer/period_close.py index f3f140f..ccbc402 100644 --- a/src/bank_importer/period_close.py +++ b/src/bank_importer/period_close.py @@ -16,7 +16,7 @@ import json import re import sqlite3 -from .db import utc_now +from .db import transaction, utc_now from . import calculation, dashboard, settings as settings_mod @@ -237,10 +237,9 @@ def record_late_arrivals( return 0 now = utc_now() inserted = 0 - # One transaction so the rows and their audit trail commit or roll back - # together; without it the caller's connection.close() silently rolled - # the late-arrival records back while the API still reported them. - with connection: + # Nestable transaction: rows and their audit trail commit or roll back + # together. ``with connection:`` would commit an outer caller early. + with transaction(connection): for row_id, year_month in items: existing = connection.execute( "SELECT id FROM period_late_arrivals WHERE source_row_id = ?", @@ -530,7 +529,7 @@ def ensure_pending_tasks( first = earliest_month(connection) or last created: list[str] = [] month = first - with connection: + with transaction(connection): while month <= last: existing = _current_run(connection, month) if existing is None: @@ -598,7 +597,7 @@ def execute_close( snapshot = build_snapshot(connection, year_month) digest = _hash_payload(snapshot) now = utc_now() - with connection: + with transaction(connection): if current is not None and current["status"] == "pending": version = int(current["version"]) report_no = _report_no(year_month, version) @@ -665,7 +664,7 @@ def mark_close_failed( reason = str(reason or "").strip() or "结账失败" current = _current_run(connection, year_month) now = utc_now() - with connection: + with transaction(connection): if current is None: version = 1 connection.execute( @@ -727,7 +726,7 @@ def request_reopen( if pending is not None: raise PeriodConflictError("该账期已有待审批的重开申请。") now = utc_now() - with connection: + with transaction(connection): cursor = connection.execute( """ INSERT INTO period_reopen_requests ( @@ -786,7 +785,7 @@ def decide_reopen( close_run = connection.execute( "SELECT * FROM period_close_runs WHERE id = ?", (row["period_close_id"],) ).fetchone() - with connection: + with transaction(connection): if approve: window_end = ( datetime.now(timezone.utc) + timedelta(days=int(row["window_days"])) @@ -873,7 +872,7 @@ def expire_reopen_windows( for row in rows: year_month = row["year_month"] now = utc_now() - with connection: + with transaction(connection): connection.execute( """ UPDATE period_close_runs diff --git a/src/bank_importer/personal_transit.py b/src/bank_importer/personal_transit.py index 8559b7f..6f1b248 100644 --- a/src/bank_importer/personal_transit.py +++ b/src/bank_importer/personal_transit.py @@ -12,7 +12,7 @@ from __future__ import annotations import sqlite3 -from .db import utc_now +from .db import transaction, utc_now from .master_data import ( ConflictError, mask_account_number, @@ -67,7 +67,7 @@ def submit_mapping( ).fetchone() if existing is None: try: - with connection: + with transaction(connection): cursor = connection.execute( """ INSERT INTO personal_transit_mappings ( @@ -81,22 +81,21 @@ def submit_mapping( start, actor["id"] if actor else None, now, now, ), ) + mapping_id = int(cursor.lastrowid) + record_change( + connection, "personal_transit_mapping", mapping_id, "submit", None, + {"account_number": number, "account_name": holder, + "represented_company_id": represented_company_id, + "allowed_direction": direction, "status": "pending", + "effective_from": start}, + None, actor, + ) except sqlite3.IntegrityError as exc: raise ConflictError("该个人过账账号已登记，请等待现有申请处理。") from exc - mapping_id = int(cursor.lastrowid) - with connection: - record_change( - connection, "personal_transit_mapping", mapping_id, "submit", None, - {"account_number": number, "account_name": holder, - "represented_company_id": represented_company_id, - "allowed_direction": direction, "status": "pending", - "effective_from": start}, - None, actor, - ) return get_mapping(connection, mapping_id) if existing["status"] == "returned": - with connection: + with transaction(connection): connection.execute( """ UPDATE personal_transit_mappings @@ -143,7 +142,7 @@ def review_mapping( if mapping["status"] != "pending": raise ConflictError("只有待复核的映射可以审核通过。") start = validate_date(effective_from, "生效日期") or mapping["effective_from"] or today - with connection: + with transaction(connection): connection.execute( """ UPDATE personal_transit_mappings @@ -162,7 +161,7 @@ def review_mapping( raise ConflictError("只有待复核的映射可以退回。") if reason is None: raise ValueError("退回必须填写原因。") - with connection: + with transaction(connection): connection.execute( """ UPDATE personal_transit_mappings @@ -182,7 +181,7 @@ def review_mapping( if reason is None: raise ValueError("停用必须填写原因。") end = validate_date(effective_to, "停用日期") or today - with connection: + with transaction(connection): connection.execute( """ UPDATE personal_transit_mappings diff --git a/src/bank_importer/settings.py b/src/bank_importer/settings.py index 8700601..162bcb1 100644 --- a/src/bank_importer/settings.py +++ b/src/bank_importer/settings.py @@ -12,7 +12,7 @@ import re import sqlite3 from datetime import datetime -from .db import utc_now +from .db import utc_now, transaction # Defaults are applied when a key is absent; the value type is always string. @@ -93,7 +93,7 @@ def update_settings( if not cleaned: raise ValueError("没有需要保存的设置项。") current = get_settings(connection) - with connection: + with transaction(connection): for key, new_value in cleaned.items(): old_value = current.get(key) if old_value == new_value: diff --git a/tests/test_calculation.py b/tests/test_calculation.py index 68cd81e..f552084 100644 --- a/tests/test_calculation.py +++ b/tests/test_calculation.py @@ -349,6 +349,66 @@ class CoverageGapTests(CalculationBase): ).fetchone() self.assertEqual("closed_attested", closed["status"]) + def test_attestation_and_audit_survive_connection_close(self) -> None: + """HEL-282: attestation writes used to skip the change log; review also + nested-committed coverage recalculation before the overlap close.""" + from bank_importer.db import connect as db_connect + + self.add_confirmed_row( + self.company_a, + account_id=self.account_a["id"], + own_account="6222000000000001", + at="2026-06-21T10:00:00", + ) + calculation.recalculate_coverage_gaps(self.connection) + gap = self.connection.execute( + "SELECT * FROM coverage_gaps WHERE status = 'open'" + ).fetchone() + cashier_id = auth.create_user( + self.connection, "cashier-close", "CashierA123", "company", self.company_a + ) + cashier = self.connection.execute( + "SELECT * FROM users WHERE id = ?", (cashier_id,) + ).fetchone() + att = calculation.submit_no_business_attestation( + self.connection, + company_id=self.company_a, + bank_account_id=self.account_a["id"], + gap_start=gap["gap_start"], + gap_end=gap["gap_end"], + reason="当日账户无资金往来", + evidence=None, + actor=cashier, + ) + calculation.review_no_business_attestation( + self.connection, att["id"], "approve", "审核通过", self.admin + ) + att_id = att["id"] + self.connection.close() + fresh = db_connect(self.db_path) + try: + row = fresh.execute( + "SELECT status FROM no_business_attestations WHERE id = ?", (att_id,) + ).fetchone() + actions = [ + item["action"] + for item in fresh.execute( + """ + SELECT action FROM audit_log + WHERE action LIKE 'attestation_%' + ORDER BY id + """ + ).fetchall() + ] + closed = fresh.execute( + "SELECT status FROM coverage_gaps WHERE id = ?", (gap["id"],) + ).fetchone() + finally: + fresh.close() + self.assertEqual("approved", row["status"]) + self.assertEqual(["attestation_submit", "attestation_approve"], actions) + self.assertEqual("closed_attested", closed["status"]) + class BalanceBasisTests(CalculationBase): def setUp(self) -> None: diff --git a/tests/test_ledger_events.py b/tests/test_ledger_events.py index 4e272cf..4159e3f 100644 --- a/tests/test_ledger_events.py +++ b/tests/test_ledger_events.py @@ -257,6 +257,56 @@ class ProjectionTests(LedgerBase): self.assertEqual("confirmed", revision["state"]) self.assertEqual("receivable", revision["subject_code"]) + def test_reopen_subject_survives_connection_close(self) -> None: + """HEL-282: create_event used to commit the replacement event while + the bank-source re-claim stayed uncommitted; close() dropped the claim.""" + from bank_importer.db import connect as db_connect + + self.pair(self.company_a, self.company_b, "100.00") + ledger_events.reconcile_bank_events(self.connection, actor=self.admin) + original_id = self.ledger_events()[0]["id"] + subjects.confirm_subject( + self.connection, original_id, + perspective_company_id=self.company_a, subject_code="receivable", + reason="确认应收", expected_revision=1, request_key="k1", + actor=self.admin, + ) + new_id, _ = ledger_events.reopen_subject( + self.connection, original_id, + reason="科目复核更正为其他应收", actor=self.admin, + ) + self.connection.close() + fresh = db_connect(self.db_path) + try: + claim = fresh.execute( + "SELECT ledger_event_id FROM ledger_event_bank_sources" + ).fetchone() + new_state = fresh.execute( + """ + SELECT r.state FROM current_ledger_event_revisions c + JOIN ledger_event_revisions r ON r.id = c.revision_id + WHERE c.ledger_event_id = ? + """, + (new_id,), + ).fetchone() + suggestions = fresh.execute( + "SELECT COUNT(*) AS n FROM ledger_subject_suggestions WHERE ledger_event_id = ?", + (new_id,), + ).fetchone()["n"] + reversal = fresh.execute( + """ + SELECT COUNT(*) AS n FROM ledger_event_revisions + WHERE posting_kind = 'reversal' AND reverses_ledger_event_id = ? + """, + (original_id,), + ).fetchone()["n"] + finally: + fresh.close() + self.assertEqual(new_id, claim["ledger_event_id"]) + self.assertEqual("pending_subject", new_state["state"]) + self.assertGreaterEqual(suggestions, 1) + self.assertEqual(1, reversal) + class SubjectSuggestionTests(LedgerBase): def test_mirror_mapping_is_symmetric(self) -> None: diff --git a/tests/test_master_data.py b/tests/test_master_data.py index bf41a0d..78e48d1 100644 --- a/tests/test_master_data.py +++ b/tests/test_master_data.py @@ -210,6 +210,40 @@ class MasterDataUnitTests(unittest.TestCase): ) +class MasterDataCommitTests(unittest.TestCase): + """File-database checks that business rows and audit share one commit.""" + + def setUp(self) -> None: + self.temp_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.temp_dir.cleanup) + self.db_path = Path(self.temp_dir.name) / "app.db" + self.connection = connect(self.db_path) + self.addCleanup(self.connection.close) + migrate(self.connection) + + def test_create_company_and_audit_survive_connection_close(self) -> None: + company_id = master_data.create_company( + self.connection, "丁公司", None, None, actor=None + ) + self.connection.close() + fresh = connect(self.db_path) + try: + company = fresh.execute( + "SELECT name FROM companies WHERE id = ?", (company_id,) + ).fetchone() + change = fresh.execute( + """ + SELECT action, entity_id FROM master_data_changes + WHERE entity_type = 'company' + """ + ).fetchone() + finally: + fresh.close() + self.assertEqual("丁公司", company["name"]) + self.assertEqual("create", change["action"]) + self.assertEqual(company_id, change["entity_id"]) + + class MasterDataApiTests(unittest.TestCase): """Live-server workflow tests for account registration and review.""" diff --git a/tests/test_matching.py b/tests/test_matching.py index 01a5b29..964c9b5 100644 --- a/tests/test_matching.py +++ b/tests/test_matching.py @@ -943,6 +943,38 @@ class ProjectionRebuildTests(MatchingBase): self.assertEqual(sorted(before), sorted(after)) self.assertEqual(sorted(claims_before), sorted(claims_after)) + def test_rebuild_clears_stale_projection_and_survives_close(self) -> None: + """HEL-282: DELETEs used to stay uncommitted when nothing was restored.""" + from bank_importer.db import connect as db_connect + + row_a = self.add_row( + self.company_a, own_account="6222000000000001", + cp_account="6222000000000002", expense="100.00", + ) + row_b = self.add_row( + self.company_b, own_account="6222000000000002", + cp_account="6222000000000001", income="100.00", + ) + matching.reconcile_rows(self.connection, [row_a, row_b]) + with self.connection: + self.connection.execute( + "UPDATE canonical_transfer_events SET lifecycle = 'superseded'" + ) + matching.rebuild_current_projection(self.connection) + self.connection.close() + fresh = db_connect(self.db_path) + try: + remaining = fresh.execute( + "SELECT COUNT(*) AS n FROM current_transfer_decisions" + ).fetchone()["n"] + claims = fresh.execute( + "SELECT COUNT(*) AS n FROM transfer_observation_claims" + ).fetchone()["n"] + finally: + fresh.close() + self.assertEqual(0, remaining) + self.assertEqual(0, claims) + class ConcurrentReconcileTests(MatchingBase): def test_concurrent_reconcile_creates_one_event(self) -> None: diff --git a/tests/test_period_close.py b/tests/test_period_close.py index 5e9c492..93c429f 100644 --- a/tests/test_period_close.py +++ b/tests/test_period_close.py @@ -189,6 +189,42 @@ class PeriodCloseTests(LedgerBase): self.assertEqual(1, versions[0]["version"]) self.assertEqual(2, versions[-1]["version"]) + def test_close_and_reopen_request_survive_connection_close(self) -> None: + """HEL-282: monthly close / reopen request must persist with audit.""" + from bank_importer.db import connect as db_connect + + self._cover_month() + closed = self._close() + req = period_close.request_reopen( + self.connection, self.MONTH, self.admin, + reason="补录金牛煤业七月运输费并核对金额", + ) + report_no = closed["report_no"] + request_id = req["id"] + self.connection.close() + fresh = db_connect(self.db_path) + try: + run = fresh.execute( + "SELECT status, report_no FROM period_close_runs WHERE year_month = ?", + (self.MONTH,), + ).fetchone() + reopen = fresh.execute( + "SELECT status FROM period_reopen_requests WHERE id = ?", + (request_id,), + ).fetchone() + actions = { + row["action"] + for row in fresh.execute( + "SELECT action FROM period_audit_events" + ).fetchall() + } + finally: + fresh.close() + self.assertEqual("closed", run["status"]) + self.assertEqual(report_no, run["report_no"]) + self.assertEqual("pending", reopen["status"]) + self.assertTrue({"close_execute", "reopen_request"} <= actions) + def test_wal_on_file_database(self) -> None: mode = self.connection.execute("PRAGMA journal_mode").fetchone()[0] self.assertEqual("wal", str(mode).lower())