B-44: 返修——多币种对方汇总、镜像科目筛选、冲销独立生效日、创建来源撤销边界与并发幂等

2026-08-19 13:28:28 +08:00
parent d59576739f
commit f16e8aad34
7 changed files with 528 additions and 91 deletions
@@ -10,7 +10,7 @@ is never edited. Idempotency keys and UNIQUE claims prevent double counting.

 from __future__ import annotations

-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 from decimal import Decimal, InvalidOperation
 import json
 import sqlite3
@@ -60,6 +60,11 @@ def _validate_date(value: object, field: str) -> str:
    return text


+def _business_today() -> str:
+    """Shanghai business date (the default reversal effective date)."""
+    return datetime.now(timezone(timedelta(hours=8))).date().isoformat()
+
+
 # ---------------------------------------------------------------------------
 # Submit
 # ---------------------------------------------------------------------------
@@ -144,40 +149,56 @@ def submit(
            raise ManualInputError("supersedes_record_id 无效。")
        supersedes_id = int(supersedes_record_id)

-    existing = connection.execute(
-        "SELECT id FROM manual_records WHERE company_id = ? AND request_key = ?",
-        (int(company_id), request_key),
-    ).fetchone()
-    if existing is not None:
-        return _record_payload(connection, existing["id"], idempotent_replay=True)
-
    now = utc_now()
    began = False
    if not connection.in_transaction:
        connection.execute("BEGIN IMMEDIATE")
        began = True
    try:
-        cursor = connection.execute(
-            """
-            INSERT INTO manual_records (
-                company_id, counterparty_company_id, occurred_at, direction,
-                amount, amount_scale, currency, funding_source, bank_account_id,
-                personal_transit_mapping_id, related_source_row_id,
-                requested_subject, summary, reason, evidence_json, request_key,
-                supersedes_record_id, submitted_by, created_at
-            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-            """,
-            (
-                int(company_id), int(counterparty_company_id), occurred_at, direction,
-                amount, _scale_of(amount), currency, funding_source, bank_account_id,
-                mapping_id, related_row["id"] if related_row is not None else None,
-                requested_subject, str(summary or "") or None,
-                str(reason or "") or None,
-                json.dumps(evidence, ensure_ascii=False) if evidence else None,
-                request_key, supersedes_id,
-                actor["id"], now,
-            ),
-        )
+        # Re-check inside the write transaction: concurrent identical submits
+        # serialize here, so a replay is found before any INSERT.
+        existing = connection.execute(
+            "SELECT id FROM manual_records WHERE company_id = ? AND request_key = ?",
+            (int(company_id), request_key),
+        ).fetchone()
+        if existing is not None:
+            if began:
+                connection.commit()
+            return _record_payload(connection, existing["id"], idempotent_replay=True)
+        try:
+            cursor = connection.execute(
+                """
+                INSERT INTO manual_records (
+                    company_id, counterparty_company_id, occurred_at, direction,
+                    amount, amount_scale, currency, funding_source, bank_account_id,
+                    personal_transit_mapping_id, related_source_row_id,
+                    requested_subject, summary, reason, evidence_json, request_key,
+                    supersedes_record_id, submitted_by, created_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    int(company_id), int(counterparty_company_id), occurred_at, direction,
+                    amount, _scale_of(amount), currency, funding_source, bank_account_id,
+                    mapping_id, related_row["id"] if related_row is not None else None,
+                    requested_subject, str(summary or "") or None,
+                    str(reason or "") or None,
+                    json.dumps(evidence, ensure_ascii=False) if evidence else None,
+                    request_key, supersedes_id,
+                    actor["id"], now,
+                ),
+            )
+        except sqlite3.IntegrityError:
+            # A concurrent identical submit won the race and committed first;
+            # surface the existing record idempotently instead of a UNIQUE 500.
+            if began:
+                connection.rollback()
+            existing = connection.execute(
+                "SELECT id FROM manual_records WHERE company_id = ? AND request_key = ?",
+                (int(company_id), request_key),
+            ).fetchone()
+            if existing is not None:
+                return _record_payload(connection, existing["id"], idempotent_replay=True)
+            raise
        record_id = int(cursor.lastrowid)
        _append_decision(
            connection, record_id, state="pending", action="submit",
@@ -279,14 +300,16 @@ def decide(
    actor: sqlite3.Row,
    subject_code: object = None,
    target_ledger_event_id: object = None,
+    effective_at: object = None,
 ) -> dict[str, object]:
    """Apply an administrator decision to a manual record.

    ``approve_new`` creates a confirmed ledger event; ``approve_link`` joins an
    existing ledger event without adding a second economic impact; ``return``
    and ``exception`` never produce a balance; ``reverse`` creates an opposite
-    reversal event (or detaches a linked claim). Replays return the earlier
-    outcome via ``idempotency_key``.
+    reversal event (or detaches a linked claim) with an independent business
+    effective date — explicit ``effective_at`` or the approval business day.
+    Replays return the earlier outcome via ``idempotency_key``.
    """
    reason = (reason or "").strip()
    if not reason:
@@ -350,7 +373,10 @@ def decide(
        else:  # reverse
            if current["state"] != "approved":
                raise ManualConflictError("只有已批准记录可以冲销。")
-            outcome = _reverse(connection, record, current, actor, request_key, reason)
+            outcome = _reverse(
+                connection, record, current, actor, request_key, reason,
+                effective_at=effective_at,
+            )

        _store_audit(connection, record, current, action, outcome, reason, actor)
    except Exception:
@@ -455,6 +481,7 @@ def _reverse(
    actor: sqlite3.Row,
    request_key: str | None,
    reason: str,
+    effective_at: object = None,
 ) -> dict[str, object]:
    claim = manual_source_claim(connection, record["id"])
    if claim is None:
@@ -464,25 +491,30 @@ def _reverse(
    if revision is None:
        raise ManualConflictError("关联的往来事件没有当前修订。")

-    if revision["source_kind"] == "manual" and not _event_has_other_sources(
-        connection, event_id, record["id"]
-    ):
-        # The manual record is the only economic source: reverse it with a new
-        # opposite event at an independent effective date.
+    if _is_manual_creation_source(revision, record["id"]):
+        # This record created the event (approve_new): it added the economic
+        # impact, so reversing it must always produce an equal-amount reversal
+        # event — even when other manual evidence was later linked onto the
+        # same event. The original impact must not survive in balances.
+        if effective_at is not None and str(effective_at).strip():
+            effective_at = _validate_date(effective_at, "冲销生效日")
+        else:
+            effective_at = _business_today()
        create_reversal(
            connection,
            event_id,
            source_kind="manual",
            source_revision_token=str(record["id"]),
+            effective_at=effective_at,
            reason=reason,
            actor=actor,
            idempotency_key=request_key,
            rule_version="manual-record-v1",
        )
    else:
-        # The manual was linked evidence on a bank event: it never added a
-        # second impact, so reversing detaches the claim without a reversal
-        # event; the bank impact stays.
+        # The manual was linked evidence (approve_link) on an event it never
+        # created: it added no second impact, so reversing detaches the claim
+        # and the underlying economic impact stays.
        connection.execute(
            "DELETE FROM ledger_event_manual_sources WHERE manual_record_id = ?",
            (record["id"],),
@@ -495,17 +527,20 @@ def _reverse(
    return {"decision_id": decision_id, "ledger_event_id": None}


-def _event_has_other_sources(connection, event_id: int, manual_record_id: int) -> bool:
-    bank = connection.execute(
-        "SELECT 1 FROM ledger_event_bank_sources WHERE ledger_event_id = ? LIMIT 1",
-        (event_id,),
-    ).fetchone()
-    other_manual = connection.execute(
-        "SELECT 1 FROM ledger_event_manual_sources "
-        "WHERE ledger_event_id = ? AND manual_record_id != ? LIMIT 1",
-        (event_id, manual_record_id),
-    ).fetchone()
-    return bank is not None or other_manual is not None
+def _is_manual_creation_source(revision: sqlite3.Row, manual_record_id: int) -> bool:
+    """True when ``manual_record_id`` created the event via ``approve_new``.
+
+    The creation source is recorded in the event's immutable revision
+    ``evidence_json``; a linked evidence record is never the creation source
+    and carries no second economic impact.
+    """
+    if revision["source_kind"] != "manual":
+        return False
+    try:
+        evidence = json.loads(revision["evidence_json"] or "{}")
+    except (TypeError, ValueError):
+        return False
+    return evidence.get("manual_record_id") == manual_record_id


 def _store_audit(
@@ -133,10 +133,11 @@ def _event_filters(
    if subject:
        if subject not in SUBJECTS:
            raise PositionInputError("科目筛选无效。")
-        conditions.append(
-            "(p.subject_code = ? OR (p.subject_code = ? AND p.subject_code = ?))"
-        )
-        params.extend([subject, MIRROR[subject], subject])
+        # The stored subject lives on one company's perspective; the viewer on
+        # the other side sees its mirror. Match both so mirror events are never
+        # dropped from the filter.
+        conditions.append("(p.subject_code = ? OR p.subject_code = ?)")
+        params.extend([subject, MIRROR[subject]])
    if posting_kind:
        conditions.append("p.posting_kind = ?")
        params.append(posting_kind)
@@ -209,36 +209,42 @@ def confirm_subject(
        raise SubjectInputError("必须填写科目确认依据。")
    if subject_code not in SUBJECTS:
        raise SubjectInputError("科目必须是应收/应付/其他应收/其他应付之一。")
-    subject_code = subject_code
-
-    if request_key:
-        existing = connection.execute(
-            """
-            SELECT * FROM ledger_event_revisions
-            WHERE ledger_event_id = ? AND idempotency_key = ?
-            ORDER BY id LIMIT 1
-            """,
-            (ledger_event_id, request_key),
-        ).fetchone()
-        if existing is not None:
-            return _revision_payload(connection, existing)
-
-    current = current_revision(connection, ledger_event_id)
-    if current is None:
-        raise SubjectConflictError("该事件不存在或没有当前修订。")
-    if current["state"] != "pending_subject":
-        raise SubjectConflictError("只有待确认科目的事件可以确认科目。")
-    if expected_revision is not None and int(expected_revision) != current["revision"]:
-        raise SubjectConflictError("事件已发生变更，请刷新后重试。")
-    participants = {current["payer_company_id"], current["payee_company_id"]}
-    if perspective_company_id not in participants:
-        raise SubjectInputError("视角公司必须是事件参与方。")

    began = False
    if not connection.in_transaction:
        connection.execute("BEGIN IMMEDIATE")
        began = True
    try:
+        if request_key:
+            existing = connection.execute(
+                """
+                SELECT * FROM ledger_event_revisions
+                WHERE ledger_event_id = ? AND idempotency_key = ?
+                ORDER BY id LIMIT 1
+                """,
+                (ledger_event_id, request_key),
+            ).fetchone()
+            if existing is not None:
+                if began:
+                    connection.commit()
+                return _revision_payload(connection, existing)
+
+        current = current_revision(connection, ledger_event_id)
+        if current is None:
+            raise SubjectConflictError("该事件不存在或没有当前修订。")
+        if current["state"] != "pending_subject":
+            raise SubjectConflictError("只有待确认科目的事件可以确认科目。")
+        # ``expected_revision`` may be the revision row id (what the API/UI
+        # sends as ``ledger_revision_id``) or the per-event sequence number;
+        # both identify the exact revision the client saw.
+        if expected_revision is not None and int(expected_revision) not in (
+            current["id"], current["revision"],
+        ):
+            raise SubjectConflictError("事件已发生变更，请刷新后重试。")
+        participants = {current["payer_company_id"], current["payee_company_id"]}
+        if perspective_company_id not in participants:
+            raise SubjectInputError("视角公司必须是事件参与方。")
+
        revision_id = append_revision(
            connection,
            ledger_event_id,