leefer/caiwuzongzhang
1
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity

HEL-282: 写库路径事务提交与审计留痕自查修复

Browse Source 
同类未提交即 close 回滚、业务与审计拆成两笔事务的路径一并收进可嵌套事务边界。

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
总工
2026-08-30 23:02:14 +08:00
co-authored by Cursor multica-agent
parent 9e0e4a103a
commit 9193a3fce0
16 changed files with 511 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server.py
+89 -88
View File
@@ -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()
src/bank_importer/auth.py
+2 -2
View File
@@ -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 (
src/bank_importer/calculation.py
+24 -10
View File
@@ -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)
src/bank_importer/db.py
+27
View File
@@ -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
src/bank_importer/importing.py
+8 -8
View File
@@ -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
src/bank_importer/ledger_events.py
+63 -58
View File
@@ -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
src/bank_importer/master_data.py
+31 -34
View File
@@ -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
src/bank_importer/matching.py
+28 -28
View File
@@ -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
src/bank_importer/period_close.py
+10 -11
View File
@@ -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
src/bank_importer/personal_transit.py
+15 -16
View File
@@ -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
src/bank_importer/settings.py
+2 -2
View File
@@ -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:
tests/test_calculation.py
+60
View File
@@ -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:
tests/test_ledger_events.py
+50
View File
@@ -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:
tests/test_master_data.py
+34
View File
@@ -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."""
tests/test_matching.py
+32
View File
@@ -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:
tests/test_period_close.py
+36
View File
@@ -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())