1602 lines
58 KiB
Python
1602 lines
58 KiB
Python
"""Canonical transfer matching: deterministic bilateral pairing and event layer.
|
|
|
|
Immutable bank source rows are observations. This module turns one or two
|
|
confirmed observations into one canonical transfer event by appending
|
|
decisions to an append-only log and maintaining a rebuildable current
|
|
projection. It never modifies ``source_rows``.
|
|
|
|
Confirmed business rules (B-43/B-114):
|
|
|
|
- A single observation stays in the unresolved amount bucket even when both
|
|
participants are uniquely identified by approved accounts; only a bilateral
|
|
merge (``intercompany`` + ``paired``) or an administrator confirmation based
|
|
on evidence (``intercompany`` + locked manual single) enters the confirmed
|
|
B-44 balance.
|
|
- The economic date of a cross-day event is the payer's outgoing bank posting
|
|
time, never the first-import time.
|
|
- Automatic window v1: same non-empty reference + mirrored accounts <= 3
|
|
calendar days (M1); exact account mirror without reference <= 1 day (M2);
|
|
alias/personal-mapping mirror <= 1 day with reference or summary equality
|
|
(M3). Anything over-window or with 2+ same-tier candidates goes to review.
|
|
- Personal transit mappings bind a specific account, an effective interval, an
|
|
allowed direction and a represented company, and require administrator
|
|
approval; names are supporting evidence only.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from decimal import Decimal, InvalidOperation
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
|
|
from .auth import audit
|
|
from .db import utc_now
|
|
from .master_data import (
|
|
is_identifiable,
|
|
normalize_account_number,
|
|
)
|
|
from . import personal_transit
|
|
|
|
RULE_VERSION = "transfer-match-v1"
|
|
|
|
CLASSIFICATION_UNRESOLVED = "unresolved"
|
|
CLASSIFICATION_NEEDS_REVIEW = "needs_review"
|
|
CLASSIFICATION_INTERNAL_SINGLE = "internal_single"
|
|
CLASSIFICATION_INTERCOMPANY = "intercompany"
|
|
CLASSIFICATION_SAME_COMPANY = "same_company"
|
|
CLASSIFICATION_EXTERNAL = "external"
|
|
|
|
PAIRING_SINGLE = "single"
|
|
PAIRING_PAIRED = "paired"
|
|
PAIRING_NA = "not_applicable"
|
|
|
|
MODE_AUTO = "auto"
|
|
MODE_MANUAL = "manual"
|
|
MODE_REVERSAL = "reversal"
|
|
|
|
MAX_DATE_DIFF_DAYS = 3
|
|
AUTO_TIERS = ("M1", "M2", "M3")
|
|
REVIEW_TIERS = ("R1", "R2")
|
|
|
|
_FULL_WIDTH = str.maketrans(
|
|
"ＡＢＣＤＥＦＧＨＩＪＫＬＭＮＯＰＱＲＳＴＵＶＷＸＹＺ"
|
|
"ａｂｃｄｅｆｇｈｉｊｋｌｍｎｏｐｑｒｓｔｕｖｗｘｙｚ０１２３４５６７８９",
|
|
"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
|
|
"abcdefghijklmnopqrstuvwxyz0123456789",
|
|
)
|
|
# Bank noise words stripped from summaries before comparison (M3). Only exact
|
|
# equality after normalization decides; no fuzzy matching or AI guesses.
|
|
_SUMMARY_NOISE = (
|
|
"转账", "汇款", "网上银行", "手机银行", "跨行", "行内", "对公",
|
|
"跨行转账", "行内转账", "网银转账", "速汇", "实时", "普通", "加急",
|
|
)
|
|
|
|
|
|
class MatchConflictError(ValueError):
|
|
"""A decision/claim/lock conflict (mapped to HTTP 409)."""
|
|
|
|
|
|
class MatchInputError(ValueError):
|
|
"""Invalid input for matching (mapped to HTTP 400/422)."""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Small value objects
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Resolved:
|
|
"""One participant endpoint resolved from approved identifiers."""
|
|
|
|
role: str
|
|
company_id: int
|
|
bank_account_id: int | None
|
|
resolve_method: str
|
|
alias_id: int | None = None
|
|
mapping_id: int | None = None
|
|
evidence: dict = field(default_factory=dict)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RowResolution:
|
|
direction: str | None
|
|
own: Resolved | None
|
|
own_reason: str | None
|
|
counterparty: Resolved | None
|
|
counterparty_reason: str | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CandidateHit:
|
|
row_id: int
|
|
tier: str
|
|
date_diff_days: int
|
|
account_mirror: str # exact | alias | none
|
|
reference_match: str | None # same | conflict | None
|
|
summary_match: bool
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Parsing and normalization helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def parse_direction(income: object, expense: object) -> str | None:
|
|
"""Exactly one positive amount decides the direction; anything else is None."""
|
|
income = Decimal(str(income or "0"))
|
|
expense = Decimal(str(expense or "0"))
|
|
if expense > 0 and income == 0:
|
|
return "outgoing"
|
|
if income > 0 and expense == 0:
|
|
return "incoming"
|
|
return None
|
|
|
|
|
|
def amount_of_row(row: sqlite3.Row) -> Decimal | None:
|
|
income = Decimal(str(row["income"] or "0"))
|
|
expense = Decimal(str(row["expense"] or "0"))
|
|
if income > 0 and expense == 0:
|
|
return income
|
|
if expense > 0 and income == 0:
|
|
return expense
|
|
return None
|
|
|
|
|
|
def normalize_reference(value: object) -> str | None:
|
|
text = str(value or "").translate(_FULL_WIDTH)
|
|
text = re.sub(r"\s+", "", text)
|
|
return text or None
|
|
|
|
|
|
def normalize_summary(row: sqlite3.Row) -> str:
|
|
parts = []
|
|
for key in ("summary", "purpose"):
|
|
text = str(row[key] or "").translate(_FULL_WIDTH)
|
|
text = re.sub(r"[\s\ufeff]+", "", text)
|
|
for noise in _SUMMARY_NOISE:
|
|
text = text.replace(noise, "")
|
|
if text:
|
|
parts.append(text)
|
|
return "|".join(parts)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Transaction helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _ensure_transaction(connection: sqlite3.Connection):
|
|
"""Begin an immediate transaction unless one is already open.
|
|
|
|
``reconcile_rows`` / ``apply_manual_decision`` may run standalone (they own
|
|
the transaction) or nested inside the caller's ``with connection:`` block
|
|
(the worksheet-confirm flow); nested calls never start their own commit.
|
|
"""
|
|
began = False
|
|
if not connection.in_transaction:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
began = True
|
|
return began
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Participant resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _normalized_account(value: object) -> str | None:
|
|
text = str(value or "").strip()
|
|
if not text:
|
|
return None
|
|
try:
|
|
return normalize_account_number(text)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _resolve_own(
|
|
connection: sqlite3.Connection,
|
|
row: sqlite3.Row,
|
|
on_date: str,
|
|
upload_account_id: int | None,
|
|
) -> tuple[Resolved | None, str | None]:
|
|
"""Resolve the owner endpoint: source-row own account first, the approved
|
|
upload account as fallback. A company conflict between the two is a review
|
|
signal, never a guess.
|
|
"""
|
|
own_number = _normalized_account(row["own_account"])
|
|
|
|
own: Resolved | None = None
|
|
own_reason: str | None = None
|
|
if own_number is not None:
|
|
account = connection.execute(
|
|
"SELECT * FROM bank_accounts WHERE account_number = ?", (own_number,)
|
|
).fetchone()
|
|
if account is not None and is_identifiable(account, on_date):
|
|
own = Resolved(
|
|
role="", company_id=account["company_id"],
|
|
bank_account_id=account["id"],
|
|
resolve_method="own_exact",
|
|
evidence={"account_number": own_number, "via": "own_account"},
|
|
)
|
|
else:
|
|
own_reason = "本方账号未能匹配到生效期内已批准的账户"
|
|
|
|
if upload_account_id is not None:
|
|
upload = connection.execute(
|
|
"SELECT * FROM bank_accounts WHERE id = ?", (upload_account_id,)
|
|
).fetchone()
|
|
if upload is not None and is_identifiable(upload, on_date):
|
|
if own is not None and own.company_id != upload["company_id"]:
|
|
return None, "本方账号与上传账户归属公司冲突"
|
|
if own is None:
|
|
own = Resolved(
|
|
role="", company_id=upload["company_id"],
|
|
bank_account_id=upload["id"],
|
|
resolve_method="upload_account",
|
|
evidence={"account_number": upload["account_number"],
|
|
"via": "upload_bank_account"},
|
|
)
|
|
own_reason = None
|
|
|
|
return own, own_reason
|
|
|
|
|
|
def _resolve_counterparty(
|
|
connection: sqlite3.Connection,
|
|
row: sqlite3.Row,
|
|
on_date: str,
|
|
direction: str | None,
|
|
) -> tuple[Resolved | None, str | None]:
|
|
"""Resolve the counterparty strictly by account identifier tiers.
|
|
|
|
Tier order: exact ``bank_accounts`` number -> approved ``account`` alias ->
|
|
approved in-window personal transit mapping (with direction allowed). Names
|
|
never resolve a counterparty; they only supply conflict evidence.
|
|
"""
|
|
number = _normalized_account(row["counterparty_account"])
|
|
if number is None:
|
|
return None, "无对方账号"
|
|
|
|
exact = connection.execute(
|
|
"SELECT * FROM bank_accounts WHERE account_number = ?", (number,)
|
|
).fetchone()
|
|
if exact is not None and is_identifiable(exact, on_date):
|
|
return _finish_counterparty(connection, row, exact, "counterparty_exact", None, None)
|
|
|
|
alias_rows = connection.execute(
|
|
"""
|
|
SELECT a.*, al.id AS alias_row_id, al.effective_from AS alias_from,
|
|
al.effective_to AS alias_to
|
|
FROM account_aliases al
|
|
JOIN bank_accounts a ON a.id = al.bank_account_id
|
|
WHERE al.alias_kind = 'account' AND al.alias_value = ?
|
|
""",
|
|
(number,),
|
|
).fetchall()
|
|
in_window = [
|
|
alias for alias in alias_rows
|
|
if _alias_in_window(alias, on_date) and is_identifiable(alias, on_date)
|
|
]
|
|
if len(in_window) == 1:
|
|
alias = in_window[0]
|
|
return _finish_counterparty(
|
|
connection, row, alias, "account_alias", alias["alias_row_id"], None,
|
|
)
|
|
if len(in_window) > 1:
|
|
return None, "对方账号命中多个账号别名，落入人工审核"
|
|
|
|
mapping = connection.execute(
|
|
"SELECT * FROM personal_transit_mappings WHERE account_number = ?", (number,)
|
|
).fetchone()
|
|
if mapping is not None and personal_transit.is_mapping_active(mapping, on_date):
|
|
if direction is not None and not personal_transit.direction_allowed(mapping, direction):
|
|
return None, "个人过账映射方向与该笔交易不符"
|
|
return _finish_counterparty(
|
|
connection, row, mapping, "personal_mapping", None, mapping["id"],
|
|
)
|
|
|
|
return None, "对方账号未能匹配到已批准账户、账号别名或个人过账映射"
|
|
|
|
|
|
def _finish_counterparty(
|
|
connection: sqlite3.Connection,
|
|
row: sqlite3.Row,
|
|
target: sqlite3.Row,
|
|
method: str,
|
|
alias_id: int | None,
|
|
mapping_id: int | None,
|
|
) -> tuple[Resolved | None, str | None]:
|
|
"""Attach a resolved counterparty; a conflicting name alias downgrades it.
|
|
|
|
A name alias belonging to a different company than the account evidence is
|
|
a conflict that must go to review. A name alias matching the same company
|
|
is recorded as supporting evidence only.
|
|
"""
|
|
company_id = target["company_id"] if "company_id" in target.keys() else target["represented_company_id"]
|
|
bank_account_id = target["id"] if "company_id" in target.keys() else None
|
|
evidence: dict[str, object] = {
|
|
"account_number": _normalized_account(row["counterparty_account"]),
|
|
"via": method,
|
|
}
|
|
if mapping_id is not None:
|
|
evidence["mapping_id"] = mapping_id
|
|
if alias_id is not None:
|
|
evidence["alias_id"] = alias_id
|
|
|
|
name = str(row["counterparty_name"] or "").strip()
|
|
if name:
|
|
name_rows = connection.execute(
|
|
"""
|
|
SELECT a.company_id, al.bank_account_id
|
|
FROM account_aliases al
|
|
JOIN bank_accounts a ON a.id = al.bank_account_id
|
|
WHERE al.alias_kind = 'name' AND al.alias_value = ?
|
|
""",
|
|
(re.sub(r"\s+", "", name),),
|
|
).fetchall()
|
|
conflicts = [hit for hit in name_rows if hit["company_id"] != company_id]
|
|
if conflicts:
|
|
return None, "对方户名别名归属与账号证据冲突，落入人工审核"
|
|
if name_rows:
|
|
evidence["name_alias"] = name
|
|
|
|
resolved = Resolved(
|
|
role="", company_id=company_id, bank_account_id=bank_account_id,
|
|
resolve_method=method, alias_id=alias_id, mapping_id=mapping_id,
|
|
evidence=evidence,
|
|
)
|
|
return resolved, None
|
|
|
|
|
|
def _alias_in_window(alias: sqlite3.Row, on_date: str) -> bool:
|
|
if alias["alias_from"] and on_date < alias["alias_from"]:
|
|
return False
|
|
if alias["alias_to"] and on_date > alias["alias_to"]:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _load_rows(
|
|
connection: sqlite3.Connection, row_ids: list[int]
|
|
) -> list[sqlite3.Row]:
|
|
"""Load source rows joined with their batch ownership/upload account."""
|
|
placeholders = ",".join("?" for _ in row_ids)
|
|
return connection.execute(
|
|
f"""
|
|
SELECT r.*, s.sheet_name, rv.review_status,
|
|
b.company_id AS batch_company_id,
|
|
b.upload_bank_account_id AS upload_bank_account_id
|
|
FROM source_rows r
|
|
JOIN sheet_batches s ON s.id = r.sheet_batch_id
|
|
JOIN sheet_reviews rv ON rv.sheet_batch_id = s.id
|
|
JOIN import_batches b ON b.id = s.import_batch_id
|
|
WHERE r.id IN ({placeholders})
|
|
ORDER BY r.id
|
|
""",
|
|
row_ids,
|
|
).fetchall()
|
|
|
|
|
|
def resolve_row(
|
|
connection: sqlite3.Connection, row: sqlite3.Row
|
|
) -> RowResolution:
|
|
"""Resolve one source row's direction, own and counterparty endpoints."""
|
|
on_date = _date_only(row["transaction_at"])
|
|
direction = parse_direction(row["income"], row["expense"])
|
|
own, own_reason = _resolve_own(
|
|
connection, row, on_date,
|
|
row["upload_bank_account_id"] if "upload_bank_account_id" in row.keys() else None,
|
|
)
|
|
|
|
if (
|
|
own is not None
|
|
and row["batch_company_id"] is not None
|
|
and own.resolve_method == "upload_account"
|
|
and own.company_id != row["batch_company_id"]
|
|
):
|
|
own, own_reason = None, "上传账户与批次公司不一致"
|
|
|
|
counterparty, cp_reason = None, None
|
|
if direction is not None and own is not None:
|
|
counterparty, cp_reason = _resolve_counterparty(
|
|
connection, row, on_date, direction
|
|
)
|
|
elif direction is not None:
|
|
cp_reason = "本方未解析，无法解析对方"
|
|
|
|
return RowResolution(
|
|
direction, own, own_reason, counterparty, cp_reason,
|
|
)
|
|
|
|
|
|
def _date_only(value: object) -> str:
|
|
text = str(value or "")[:10]
|
|
return text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Candidate search
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _confirmed_rows(connection: sqlite3.Connection) -> list[sqlite3.Row]:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT r.*, s.sheet_name, rv.review_status,
|
|
b.company_id AS batch_company_id,
|
|
b.upload_bank_account_id AS upload_bank_account_id
|
|
FROM source_rows r
|
|
JOIN sheet_batches s ON s.id = r.sheet_batch_id
|
|
JOIN sheet_reviews rv ON rv.sheet_batch_id = s.id
|
|
JOIN import_batches b ON b.id = s.import_batch_id
|
|
WHERE rv.review_status = 'confirmed'
|
|
ORDER BY r.id
|
|
"""
|
|
).fetchall()
|
|
locked = connection.execute(
|
|
"""
|
|
SELECT c.source_row_id FROM transfer_observation_claims c
|
|
JOIN transfer_match_decisions d ON d.id = c.decision_id
|
|
WHERE d.locked = 1
|
|
"""
|
|
).fetchall()
|
|
locked_ids = {item["source_row_id"] for item in locked}
|
|
return [row for row in rows if row["id"] not in locked_ids]
|
|
|
|
|
|
def _find_candidates(
|
|
connection: sqlite3.Connection,
|
|
row: sqlite3.Row,
|
|
res: RowResolution,
|
|
pool: list[sqlite3.Row],
|
|
resolutions: dict[int, RowResolution],
|
|
) -> tuple[list[CandidateHit], str | None]:
|
|
"""Deterministic candidate search over confirmed rows.
|
|
|
|
Returns ``(hits, None)`` when a unique auto-tier candidate decides the
|
|
match, ``(hits, reason)`` with reason ``r1_review``/``r2_ambiguous``/None
|
|
when the outcome needs review. Only hits passing the hard gates are kept.
|
|
"""
|
|
row_amount = amount_of_row(row)
|
|
row_currency = str(row["currency"] or "").strip()
|
|
if res.own is None or res.counterparty is None:
|
|
return [], None
|
|
if row_amount is None or not row_currency:
|
|
return [], None
|
|
|
|
row_date = _parse_datetime(row["transaction_at"]).date()
|
|
|
|
# Rows already awaiting review may not be grabbed by another automatic
|
|
# match: pairing with one of several same-tier candidates would be a
|
|
# processing-order tie-break, which is never allowed.
|
|
blocked = {
|
|
item["source_row_id"]
|
|
for item in connection.execute(
|
|
"""
|
|
SELECT c.source_row_id FROM transfer_observation_claims c
|
|
JOIN transfer_match_decisions d ON d.id = c.decision_id
|
|
WHERE d.classification = ?
|
|
""",
|
|
(CLASSIFICATION_NEEDS_REVIEW,),
|
|
)
|
|
}
|
|
|
|
hits: list[CandidateHit] = []
|
|
for other in pool:
|
|
if other["id"] == row["id"] or other["id"] in blocked:
|
|
continue
|
|
other_res = resolutions[other["id"]]
|
|
if other_res.direction is None or other_res.own is None or other_res.counterparty is None:
|
|
continue
|
|
if other_res.direction == res.direction:
|
|
continue
|
|
other_amount = amount_of_row(other)
|
|
other_currency = str(other["currency"] or "").strip()
|
|
if other_amount != row_amount or not other_currency or other_currency != row_currency:
|
|
continue
|
|
other_date = _parse_datetime(other["transaction_at"]).date()
|
|
date_diff = abs((row_date - other_date).days)
|
|
# Over-window pairs stay as R1 review candidates: the hard gate only
|
|
# filters money/direction/company reciprocity, never the date window.
|
|
if not _reciprocal(row, other, res, other_res):
|
|
continue
|
|
tier = _classify_tier(row, other, res, other_res)
|
|
if tier is None:
|
|
continue
|
|
mirror = _mirror_level(row, other, res, other_res) or "none"
|
|
ref_match, summary_match = _tier_evidence(row, other)
|
|
hits.append(
|
|
CandidateHit(
|
|
row_id=other["id"], tier=tier, date_diff_days=date_diff,
|
|
account_mirror=mirror, reference_match=ref_match,
|
|
summary_match=summary_match,
|
|
)
|
|
)
|
|
|
|
if not hits:
|
|
return [], None
|
|
|
|
best_tier = min(hits, key=lambda hit: AUTO_TIERS.index(hit.tier) if hit.tier in AUTO_TIERS else len(AUTO_TIERS)).tier
|
|
if best_tier in AUTO_TIERS:
|
|
same_tier = [hit for hit in hits if hit.tier == best_tier]
|
|
if len(same_tier) == 1:
|
|
return same_tier, None
|
|
return hits, "r2_ambiguous"
|
|
return hits, "r1_review"
|
|
|
|
|
|
def _reciprocal(
|
|
row: sqlite3.Row,
|
|
other: sqlite3.Row,
|
|
res: RowResolution,
|
|
other_res: RowResolution,
|
|
) -> bool:
|
|
return (
|
|
res.own.company_id == other_res.counterparty.company_id
|
|
and res.counterparty.company_id == other_res.own.company_id
|
|
)
|
|
|
|
|
|
def _mirror_level(
|
|
row: sqlite3.Row,
|
|
other: sqlite3.Row,
|
|
res: RowResolution,
|
|
other_res: RowResolution,
|
|
) -> str | None:
|
|
row_own = _normalized_account(row["own_account"])
|
|
row_cp = _normalized_account(row["counterparty_account"])
|
|
other_own = _normalized_account(other["own_account"])
|
|
other_cp = _normalized_account(other["counterparty_account"])
|
|
|
|
exact_pairs = 0
|
|
if row_own and row_own == other_cp:
|
|
exact_pairs += 1
|
|
if row_cp and row_cp == other_own:
|
|
exact_pairs += 1
|
|
if exact_pairs == 2:
|
|
return "exact"
|
|
if exact_pairs == 1:
|
|
methods = {
|
|
res.counterparty.resolve_method,
|
|
other_res.counterparty.resolve_method,
|
|
}
|
|
if methods & {"account_alias", "personal_mapping"}:
|
|
return "alias"
|
|
return None
|
|
|
|
|
|
def _reference_same(row: sqlite3.Row, other: sqlite3.Row) -> bool:
|
|
"""Positive reference evidence: both sides carry one equal reference.
|
|
|
|
``None == None`` must never count as reference agreement; only "both
|
|
reference numbers exist and are equal" satisfies the reference layer of the
|
|
M1/M3 evidence. A single-sided reference is not equality either.
|
|
"""
|
|
ref_row = normalize_reference(row["reference"])
|
|
ref_other = normalize_reference(other["reference"])
|
|
return bool(ref_row and ref_other and ref_row == ref_other)
|
|
|
|
|
|
def _reference_conflict(row: sqlite3.Row, other: sqlite3.Row) -> bool:
|
|
ref_row = normalize_reference(row["reference"])
|
|
ref_other = normalize_reference(other["reference"])
|
|
return bool(ref_row and ref_other and ref_row != ref_other)
|
|
|
|
|
|
def _classify_tier(
|
|
row: sqlite3.Row,
|
|
other: sqlite3.Row,
|
|
res: RowResolution,
|
|
other_res: RowResolution,
|
|
) -> str | None:
|
|
mirror = _mirror_level(row, other, res, other_res)
|
|
if mirror is None:
|
|
return "R1"
|
|
row_date = _parse_datetime(row["transaction_at"]).date()
|
|
other_date = _parse_datetime(other["transaction_at"]).date()
|
|
date_diff = abs((row_date - other_date).days)
|
|
ref_same = _reference_same(row, other)
|
|
ref_conflict = _reference_conflict(row, other)
|
|
|
|
if mirror == "exact":
|
|
if ref_same and date_diff <= 3:
|
|
return "M1"
|
|
if date_diff <= 1 and not ref_conflict:
|
|
return "M2"
|
|
return "R1"
|
|
# alias / personal-mapping mirror: only equal references (both present) or a
|
|
# deterministically equal summary is positive evidence; an absent reference
|
|
# is not.
|
|
if date_diff <= 1 and not ref_conflict and (ref_same or _summary_equal(row, other)):
|
|
return "M3"
|
|
return "R1"
|
|
|
|
|
|
def _tier_evidence(
|
|
row: sqlite3.Row, other: sqlite3.Row
|
|
) -> tuple[str | None, bool]:
|
|
if _reference_same(row, other):
|
|
ref_match = "same"
|
|
elif _reference_conflict(row, other):
|
|
ref_match = "conflict"
|
|
else:
|
|
ref_match = None
|
|
return ref_match, _summary_equal(row, other)
|
|
|
|
|
|
def _summary_equal(row: sqlite3.Row, other: sqlite3.Row) -> bool:
|
|
left = normalize_summary(row)
|
|
right = normalize_summary(other)
|
|
return bool(left) and left == right
|
|
|
|
|
|
def _parse_datetime(value: object) -> datetime:
|
|
text = str(value)
|
|
try:
|
|
return datetime.fromisoformat(text)
|
|
except ValueError:
|
|
return datetime.fromisoformat(text.replace("Z", "+00:00"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Derivation and application
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DerivedDecision:
|
|
classification: str
|
|
pairing: str
|
|
amount: str
|
|
currency: str | None
|
|
effective_at: str
|
|
observations: tuple[tuple[int, str], ...] # (row_id, role)
|
|
participants: tuple[dict[str, object], ...]
|
|
reason: str | None
|
|
candidates: tuple[dict[str, object], ...] = ()
|
|
|
|
|
|
def _derive(
|
|
connection: sqlite3.Connection,
|
|
row: sqlite3.Row,
|
|
res: RowResolution,
|
|
pool: list[sqlite3.Row],
|
|
resolutions: dict[int, RowResolution],
|
|
) -> DerivedDecision:
|
|
amount = amount_of_row(row)
|
|
amount_text = str(amount) if amount is not None else None
|
|
currency = str(row["currency"] or "").strip() or None
|
|
row_time = _parse_datetime(row["transaction_at"]).isoformat()
|
|
|
|
if res.direction is None:
|
|
return DerivedDecision(
|
|
CLASSIFICATION_UNRESOLVED, PAIRING_NA, amount_text, currency,
|
|
row_time, ((row["id"], "outgoing"),), (), "金额方向不满足一出一入",
|
|
)
|
|
if res.own is None:
|
|
classification = (
|
|
CLASSIFICATION_NEEDS_REVIEW if _own_is_conflict(res)
|
|
else CLASSIFICATION_UNRESOLVED
|
|
)
|
|
role = "outgoing" if res.direction == "outgoing" else "incoming"
|
|
return DerivedDecision(
|
|
classification, PAIRING_NA, amount_text, currency, row_time,
|
|
((row["id"], role),), (), res.own_reason,
|
|
)
|
|
role = "outgoing" if res.direction == "outgoing" else "incoming"
|
|
counter_role = "payee" if role == "outgoing" else "payer"
|
|
|
|
if res.counterparty is None:
|
|
classification = (
|
|
CLASSIFICATION_NEEDS_REVIEW
|
|
if _cp_is_ambiguous(res)
|
|
else CLASSIFICATION_UNRESOLVED
|
|
)
|
|
return DerivedDecision(
|
|
classification, PAIRING_NA, amount_text, currency,
|
|
row_time, ((row["id"], role),),
|
|
(_participant(res.own, "payer" if role == "outgoing" else "payee"),),
|
|
res.counterparty_reason,
|
|
)
|
|
|
|
own_participant = _participant(res.own, "payer" if role == "outgoing" else "payee")
|
|
cp_participant = _participant(res.counterparty, counter_role)
|
|
if res.own.company_id == res.counterparty.company_id:
|
|
# Same-company transfers pair when a mirror candidate exists; cash
|
|
# trail is kept but they never enter the intercompany balance.
|
|
hits, decision = _find_candidates(connection, row, res, pool, resolutions)
|
|
if decision == "r2_ambiguous":
|
|
return DerivedDecision(
|
|
CLASSIFICATION_NEEDS_REVIEW, PAIRING_NA, amount_text, currency,
|
|
row_time, ((row["id"], role),), (own_participant, cp_participant),
|
|
"同层多候选，落入人工审核", tuple(_candidate_rows(hits)),
|
|
)
|
|
if decision == "r1_review":
|
|
return DerivedDecision(
|
|
CLASSIFICATION_NEEDS_REVIEW, PAIRING_NA, amount_text, currency,
|
|
row_time, ((row["id"], role),), (own_participant, cp_participant),
|
|
"端点可确认但账号镜像或参考号证据不足，落入人工审核",
|
|
tuple(_candidate_rows(hits)),
|
|
)
|
|
if hits:
|
|
other = next(item for item in pool if item["id"] == hits[0].row_id)
|
|
return _derive_paired(row, res, other, resolutions[other["id"]], hits[0])
|
|
return DerivedDecision(
|
|
CLASSIFICATION_SAME_COMPANY, PAIRING_SINGLE, amount_text, currency,
|
|
row_time, ((row["id"], role),),
|
|
(own_participant, cp_participant), "本方与对方同属一家公司",
|
|
)
|
|
|
|
hits, decision = _find_candidates(connection, row, res, pool, resolutions)
|
|
if decision == "r2_ambiguous":
|
|
candidates = _candidate_rows(hits)
|
|
return DerivedDecision(
|
|
CLASSIFICATION_NEEDS_REVIEW, PAIRING_NA, amount_text, currency,
|
|
row_time, ((row["id"], role),), (own_participant, cp_participant),
|
|
"同层多候选，落入人工审核", tuple(candidates),
|
|
)
|
|
if decision == "r1_review":
|
|
candidates = _candidate_rows(hits)
|
|
return DerivedDecision(
|
|
CLASSIFICATION_NEEDS_REVIEW, PAIRING_NA, amount_text, currency,
|
|
row_time, ((row["id"], role),), (own_participant, cp_participant),
|
|
"端点可确认但账号镜像或参考号证据不足，落入人工审核",
|
|
tuple(candidates),
|
|
)
|
|
if hits:
|
|
other = next(item for item in pool if item["id"] == hits[0].row_id)
|
|
other_res = resolutions[other["id"]]
|
|
return _derive_paired(row, res, other, other_res, hits[0])
|
|
return DerivedDecision(
|
|
CLASSIFICATION_INTERNAL_SINGLE, PAIRING_SINGLE, amount_text, currency,
|
|
row_time, ((row["id"], role),), (own_participant, cp_participant),
|
|
"单边观察，等待另一方到账或人工确认",
|
|
)
|
|
|
|
|
|
def _own_is_conflict(res: RowResolution) -> bool:
|
|
return res.own_reason in (
|
|
"本方账号与上传账户归属公司冲突",
|
|
"上传账户与批次公司不一致",
|
|
)
|
|
|
|
|
|
def _cp_is_ambiguous(res: RowResolution) -> bool:
|
|
return res.counterparty_reason in (
|
|
"对方账号命中多个账号别名，落入人工审核",
|
|
"对方户名别名归属与账号证据冲突，落入人工审核",
|
|
)
|
|
|
|
|
|
def _participant(resolved: Resolved, role: str) -> dict[str, object]:
|
|
return {
|
|
"role": role,
|
|
"company_id": resolved.company_id,
|
|
"bank_account_id": resolved.bank_account_id,
|
|
"resolve_method": resolved.resolve_method,
|
|
"alias_id": resolved.alias_id,
|
|
"mapping_id": resolved.mapping_id,
|
|
"evidence": json.dumps(resolved.evidence, ensure_ascii=False),
|
|
}
|
|
|
|
|
|
def _candidate_rows(hits: list[CandidateHit]) -> list[dict[str, object]]:
|
|
return [
|
|
{
|
|
"source_row_id": hit.row_id,
|
|
"rule_tier": hit.tier,
|
|
"date_diff_days": hit.date_diff_days,
|
|
"account_mirror": 1 if hit.account_mirror != "none" else 0,
|
|
"reference_match": hit.reference_match,
|
|
"summary_match": 1 if hit.summary_match else 0,
|
|
"accepted": None,
|
|
"reason": None,
|
|
}
|
|
for hit in hits
|
|
]
|
|
|
|
|
|
def _derive_paired(
|
|
row: sqlite3.Row,
|
|
res: RowResolution,
|
|
other: sqlite3.Row,
|
|
other_res: RowResolution,
|
|
hit: CandidateHit,
|
|
) -> DerivedDecision:
|
|
row_role = "outgoing" if res.direction == "outgoing" else "incoming"
|
|
other_role = "incoming" if row_role == "outgoing" else "outgoing"
|
|
if row_role == "outgoing":
|
|
payer = _participant(res.own, "payer")
|
|
payee = _participant(other_res.own, "payee")
|
|
effective_at = _parse_datetime(row["transaction_at"]).isoformat()
|
|
else:
|
|
payer = _participant(other_res.own, "payer")
|
|
payee = _participant(res.own, "payee")
|
|
effective_at = _parse_datetime(other["transaction_at"]).isoformat()
|
|
|
|
classification = (
|
|
CLASSIFICATION_SAME_COMPANY
|
|
if payer["company_id"] == payee["company_id"]
|
|
else CLASSIFICATION_INTERCOMPANY
|
|
)
|
|
observations = (
|
|
(row["id"], row_role),
|
|
(other["id"], other_role),
|
|
)
|
|
return DerivedDecision(
|
|
classification, PAIRING_PAIRED,
|
|
str(amount_of_row(row)), str(row["currency"] or "").strip() or None,
|
|
effective_at, observations, (payer, payee), "双边镜像归并",
|
|
(_candidate_row(hit, accepted=1),),
|
|
)
|
|
|
|
|
|
def _candidate_row(hit: CandidateHit, *, accepted: int | None) -> dict[str, object]:
|
|
return {
|
|
"source_row_id": hit.row_id,
|
|
"rule_tier": hit.tier,
|
|
"date_diff_days": hit.date_diff_days,
|
|
"account_mirror": 1 if hit.account_mirror != "none" else 0,
|
|
"reference_match": hit.reference_match,
|
|
"summary_match": 1 if hit.summary_match else 0,
|
|
"accepted": accepted,
|
|
"reason": None,
|
|
}
|
|
|
|
|
|
def _derive_for_row(
|
|
connection: sqlite3.Connection,
|
|
row: sqlite3.Row,
|
|
res: RowResolution,
|
|
pool: list[sqlite3.Row],
|
|
resolutions: dict[int, RowResolution],
|
|
) -> DerivedDecision:
|
|
return _derive(connection, row, res, pool, resolutions)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Current claim helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _current_claim(connection: sqlite3.Connection, row_id: int) -> sqlite3.Row | None:
|
|
return connection.execute(
|
|
"""
|
|
SELECT c.event_id, c.decision_id, d.locked, d.mode, d.classification,
|
|
d.pairing, d.revision, d.rule_version
|
|
FROM transfer_observation_claims c
|
|
JOIN transfer_match_decisions d ON d.id = c.decision_id
|
|
WHERE c.source_row_id = ?
|
|
""",
|
|
(row_id,),
|
|
).fetchone()
|
|
|
|
|
|
def _current_decision_for_event(
|
|
connection: sqlite3.Connection, event_id: int
|
|
) -> sqlite3.Row | None:
|
|
return connection.execute(
|
|
"""
|
|
SELECT d.* FROM current_transfer_decisions c
|
|
JOIN transfer_match_decisions d ON d.id = c.decision_id
|
|
WHERE c.event_id = ?
|
|
""",
|
|
(event_id,),
|
|
).fetchone()
|
|
|
|
|
|
def _decision_observations(
|
|
connection: sqlite3.Connection, decision_id: int
|
|
) -> list[sqlite3.Row]:
|
|
return connection.execute(
|
|
"SELECT * FROM transfer_decision_observations WHERE decision_id = ? ORDER BY id",
|
|
(decision_id,),
|
|
).fetchall()
|
|
|
|
|
|
def _decision_participants(
|
|
connection: sqlite3.Connection, decision_id: int
|
|
) -> list[sqlite3.Row]:
|
|
return connection.execute(
|
|
"SELECT * FROM transfer_decision_participants WHERE decision_id = ? ORDER BY role",
|
|
(decision_id,),
|
|
).fetchall()
|
|
|
|
|
|
def _same_decision(
|
|
connection: sqlite3.Connection,
|
|
decision_id: int,
|
|
derived: DerivedDecision,
|
|
) -> bool:
|
|
"""Idempotency check: derived outcome equals the current decision."""
|
|
current = connection.execute(
|
|
"SELECT * FROM transfer_match_decisions WHERE id = ?", (decision_id,)
|
|
).fetchone()
|
|
if current is None:
|
|
return False
|
|
if current["classification"] != derived.classification:
|
|
return False
|
|
if current["pairing"] != derived.pairing:
|
|
return False
|
|
if (current["amount"] or None) != derived.amount:
|
|
return False
|
|
if (current["currency"] or None) != derived.currency:
|
|
return False
|
|
if (current["effective_at"] or None) != derived.effective_at:
|
|
return False
|
|
|
|
observed = {
|
|
(row["source_row_id"], row["role"])
|
|
for row in _decision_observations(connection, decision_id)
|
|
}
|
|
if observed != set(derived.observations):
|
|
return False
|
|
current_participants = {
|
|
(
|
|
row["role"], row["company_id"], row["bank_account_id"],
|
|
row["resolve_method"], row["alias_id"], row["mapping_id"],
|
|
)
|
|
for row in _decision_participants(connection, decision_id)
|
|
}
|
|
derived_participants = {
|
|
(
|
|
str(p["role"]), p["company_id"], p["bank_account_id"],
|
|
str(p["resolve_method"]), p["alias_id"], p["mapping_id"],
|
|
)
|
|
for p in derived.participants
|
|
}
|
|
return current_participants == derived_participants
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Applying decisions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _apply(
|
|
connection: sqlite3.Connection,
|
|
derived: DerivedDecision,
|
|
*,
|
|
mode: str,
|
|
locked: int,
|
|
reason: str | None,
|
|
actor: sqlite3.Row | None,
|
|
rule_version: str,
|
|
supersede_decision_id: int | None = None,
|
|
idempotency_key: str | None = None,
|
|
) -> dict[str, object]:
|
|
"""Write one decision + projection update inside the open transaction."""
|
|
now = utc_now()
|
|
superseded_events: set[int] = set()
|
|
|
|
claim_events: dict[int, sqlite3.Row] = {}
|
|
for row_id, _role in derived.observations:
|
|
claim = _current_claim(connection, row_id)
|
|
if claim is not None:
|
|
claim_events[row_id] = claim
|
|
|
|
if claim_events:
|
|
event_id = min(item["event_id"] for item in claim_events.values())
|
|
else:
|
|
cursor = connection.execute(
|
|
"INSERT INTO canonical_transfer_events (lifecycle, created_at) VALUES ('active', ?)",
|
|
(now,),
|
|
)
|
|
event_id = int(cursor.lastrowid)
|
|
|
|
previous = _current_decision_for_event(connection, event_id)
|
|
if supersede_decision_id is None and previous is not None:
|
|
supersede_decision_id = previous["id"]
|
|
|
|
revision_row = connection.execute(
|
|
"SELECT COALESCE(MAX(revision), 0) AS m FROM transfer_match_decisions WHERE event_id = ?",
|
|
(event_id,),
|
|
).fetchone()
|
|
revision = int(revision_row["m"]) + 1
|
|
|
|
cursor = connection.execute(
|
|
"""
|
|
INSERT INTO transfer_match_decisions (
|
|
event_id, revision, classification, pairing, amount, currency,
|
|
effective_at, mode, rule_version, locked, reason, idempotency_key,
|
|
actor_user_id, actor_username, supersedes_decision_id, created_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
event_id, revision, derived.classification, derived.pairing,
|
|
derived.amount, derived.currency, derived.effective_at,
|
|
mode, rule_version, locked, reason, idempotency_key,
|
|
actor["id"] if actor is not None else None,
|
|
actor["username"] if actor is not None else None,
|
|
supersede_decision_id,
|
|
now,
|
|
),
|
|
)
|
|
decision_id = int(cursor.lastrowid)
|
|
|
|
for row_id, role in derived.observations:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO transfer_decision_observations (decision_id, source_row_id, role, created_at)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
(decision_id, row_id, role, now),
|
|
)
|
|
for participant in derived.participants:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO transfer_decision_participants (
|
|
decision_id, role, company_id, bank_account_id, resolve_method,
|
|
alias_id, mapping_id, evidence, created_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
decision_id, participant["role"], participant["company_id"],
|
|
participant["bank_account_id"], participant["resolve_method"],
|
|
participant["alias_id"], participant["mapping_id"],
|
|
participant["evidence"], now,
|
|
),
|
|
)
|
|
for candidate in derived.candidates:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO transfer_match_candidates (
|
|
decision_id, source_row_id, rule_tier, date_diff_days,
|
|
account_mirror, reference_match, summary_match, accepted,
|
|
reason, rule_version, created_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
decision_id, candidate["source_row_id"], candidate["rule_tier"],
|
|
candidate["date_diff_days"], candidate["account_mirror"],
|
|
candidate["reference_match"], candidate["summary_match"],
|
|
candidate["accepted"], candidate["reason"], rule_version, now,
|
|
),
|
|
)
|
|
|
|
# Projection: point every involved observation's claim at the new decision.
|
|
for row_id, _role in derived.observations:
|
|
connection.execute(
|
|
"""
|
|
INSERT OR REPLACE INTO transfer_observation_claims (source_row_id, event_id, decision_id)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
(row_id, event_id, decision_id),
|
|
)
|
|
for row_id in claim_events:
|
|
if row_id not in {rid for rid, _ in derived.observations}:
|
|
connection.execute(
|
|
"DELETE FROM transfer_observation_claims WHERE source_row_id = ?",
|
|
(row_id,),
|
|
)
|
|
|
|
# Other involved events lose their current pointer and become superseded.
|
|
for other_event in {item["event_id"] for item in claim_events.values()}:
|
|
if other_event != event_id:
|
|
connection.execute(
|
|
"DELETE FROM current_transfer_decisions WHERE event_id = ?",
|
|
(other_event,),
|
|
)
|
|
connection.execute(
|
|
"UPDATE canonical_transfer_events SET lifecycle = 'superseded' WHERE id = ?",
|
|
(other_event,),
|
|
)
|
|
superseded_events.add(other_event)
|
|
connection.execute(
|
|
"""
|
|
INSERT OR REPLACE INTO current_transfer_decisions (event_id, decision_id)
|
|
VALUES (?, ?)
|
|
""",
|
|
(event_id, decision_id),
|
|
)
|
|
|
|
return {
|
|
"event_id": event_id,
|
|
"decision_id": decision_id,
|
|
"revision": revision,
|
|
"superseded_events": sorted(superseded_events),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public reconcile API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def reconcile_rows(
|
|
connection: sqlite3.Connection,
|
|
source_row_ids: list[int],
|
|
*,
|
|
rule_version: str = RULE_VERSION,
|
|
actor: sqlite3.Row | None = None,
|
|
) -> dict[str, object]:
|
|
"""Reconcile confirmed source rows into canonical events, idempotently.
|
|
|
|
Rows claimed by a locked (manual) decision are never touched. Rows whose
|
|
derived outcome already equals their current decision produce zero writes.
|
|
Runs inside the caller's transaction when one is open, otherwise in its
|
|
own ``BEGIN IMMEDIATE`` transaction.
|
|
"""
|
|
began = _ensure_transaction(connection)
|
|
try:
|
|
ids = sorted({int(row_id) for row_id in source_row_ids})
|
|
if not ids:
|
|
return {
|
|
"created_events": 0, "updated_events": 0, "unchanged": 0,
|
|
"skipped_locked": 0, "rows": 0,
|
|
}
|
|
stats = {
|
|
"created_events": 0, "updated_events": 0, "unchanged": 0,
|
|
"skipped_locked": 0, "rows": 0,
|
|
}
|
|
pool = _confirmed_rows(connection)
|
|
pool_by_id = {row["id"]: row for row in pool}
|
|
resolutions = {
|
|
row["id"]: resolve_row(connection, row)
|
|
for row in pool
|
|
}
|
|
input_rows = _load_rows(connection, ids)
|
|
|
|
for row in input_rows:
|
|
stats["rows"] += 1
|
|
claim = _current_claim(connection, row["id"])
|
|
if claim is not None and claim["locked"]:
|
|
stats["skipped_locked"] += 1
|
|
continue
|
|
if row["id"] not in pool_by_id:
|
|
stats["skipped_unconfirmed"] += 1
|
|
continue
|
|
derived = _derive_for_row(connection, row, resolutions[row["id"]], pool, resolutions)
|
|
if claim is not None and _same_decision(connection, claim["decision_id"], derived):
|
|
stats["unchanged"] += 1
|
|
continue
|
|
outcome = _apply(
|
|
connection, derived, mode=MODE_AUTO, locked=0,
|
|
reason=derived.reason, actor=actor, rule_version=rule_version,
|
|
)
|
|
if outcome["revision"] == 1:
|
|
stats["created_events"] += 1
|
|
else:
|
|
stats["updated_events"] += 1
|
|
except Exception:
|
|
if began:
|
|
connection.rollback()
|
|
raise
|
|
else:
|
|
if began:
|
|
connection.commit()
|
|
return stats
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Manual decisions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def apply_manual_decision(
|
|
connection: sqlite3.Connection,
|
|
event_id: int,
|
|
action: str,
|
|
*,
|
|
reason: str,
|
|
expected_revision: int | None,
|
|
request_key: str | None,
|
|
actor: sqlite3.Row,
|
|
source_row_ids: list[int] | None = None,
|
|
participant: dict[str, object] | None = None,
|
|
) -> dict[str, object]:
|
|
"""Apply an administrator decision on an event (link_rows/assign_participant
|
|
/mark_external/reverse). Replaces the current decision with a locked manual
|
|
one; corrections must go through ``reverse``, never in-place editing.
|
|
"""
|
|
reason = (reason or "").strip()
|
|
if not reason:
|
|
raise MatchInputError("必须填写操作原因。")
|
|
began = _ensure_transaction(connection)
|
|
try:
|
|
# Idempotency: a replayed request key returns the earlier decision
|
|
# without re-validating the (now changed) revision.
|
|
if request_key:
|
|
existing = connection.execute(
|
|
"SELECT id FROM transfer_match_decisions WHERE idempotency_key = ? AND event_id = ?",
|
|
(request_key, event_id),
|
|
).fetchone()
|
|
if existing is not None:
|
|
if began:
|
|
connection.commit()
|
|
return _decision_payload(connection, event_id, existing["id"])
|
|
|
|
current = _current_decision_for_event(connection, event_id)
|
|
if current is None:
|
|
raise MatchConflictError("该事件当前没有有效决定。")
|
|
if expected_revision is not None and int(expected_revision) != current["revision"]:
|
|
raise MatchConflictError("事件已发生变更，请刷新后重试。")
|
|
|
|
if action == "link_rows":
|
|
derived = _manual_link(
|
|
connection, event_id, current, source_row_ids, reason, actor
|
|
)
|
|
elif action == "assign_participant":
|
|
derived = _manual_assign_participant(
|
|
connection, event_id, current, participant, reason, actor
|
|
)
|
|
elif action == "mark_external":
|
|
derived = _manual_external(connection, event_id, current, reason)
|
|
elif action == "reverse":
|
|
derived = None
|
|
else:
|
|
raise MatchInputError("未知的人工决定类型。")
|
|
|
|
if action == "reverse":
|
|
outcome = _apply_reversal(
|
|
connection, event_id, current, reason, actor, request_key
|
|
)
|
|
else:
|
|
outcome = _apply(
|
|
connection, derived, mode=MODE_MANUAL, locked=1,
|
|
reason=reason, actor=actor,
|
|
rule_version=current["rule_version"] or RULE_VERSION,
|
|
idempotency_key=request_key,
|
|
)
|
|
audit(
|
|
connection,
|
|
f"transfer_{action}",
|
|
actor=actor,
|
|
target=f"transfer_event:{event_id}",
|
|
detail=f"decision:{outcome['decision_id']};reason:{reason}",
|
|
)
|
|
except Exception:
|
|
if began:
|
|
connection.rollback()
|
|
raise
|
|
else:
|
|
if began:
|
|
connection.commit()
|
|
return _decision_payload(connection, event_id, outcome["decision_id"])
|
|
|
|
|
|
def _manual_link(
|
|
connection: sqlite3.Connection,
|
|
event_id: int,
|
|
current: sqlite3.Row,
|
|
source_row_ids: list[int] | None,
|
|
reason: str,
|
|
actor: sqlite3.Row,
|
|
) -> DerivedDecision:
|
|
if not source_row_ids or len(source_row_ids) != 2:
|
|
raise MatchInputError("人工关联必须且只能提供两条源行。")
|
|
rows = _load_rows(connection, [int(row_id) for row_id in source_row_ids])
|
|
if len(rows) != 2:
|
|
raise MatchInputError("存在无效的源行。")
|
|
for row in rows:
|
|
claim = _current_claim(connection, row["id"])
|
|
if claim is not None and claim["decision_id"] != current["id"] and claim["locked"]:
|
|
raise MatchConflictError(f"源行 {row['id']} 已被锁定的人工决定占用。")
|
|
resolutions = {row["id"]: resolve_row(connection, row) for row in rows}
|
|
directions = {row["id"]: resolutions[row["id"]].direction for row in rows}
|
|
if any(value is None for value in directions.values()):
|
|
raise MatchInputError("人工关联的源行金额方向必须一出一入。")
|
|
if directions[rows[0]["id"]] == directions[rows[1]["id"]]:
|
|
raise MatchInputError("人工关联的源行金额方向必须相反。")
|
|
outgoing_row = rows[0] if directions[rows[0]["id"]] == "outgoing" else rows[1]
|
|
incoming_row = rows[1] if outgoing_row is rows[0] else rows[0]
|
|
out_res = resolutions[outgoing_row["id"]]
|
|
in_res = resolutions[incoming_row["id"]]
|
|
if out_res.own is None or in_res.own is None:
|
|
raise MatchInputError("人工关联的源行未能唯一确认本方归属，不能定案。")
|
|
payer = _participant(out_res.own, "payer")
|
|
payee = _participant(in_res.own, "payee")
|
|
classification = (
|
|
CLASSIFICATION_SAME_COMPANY
|
|
if payer["company_id"] == payee["company_id"]
|
|
else CLASSIFICATION_INTERCOMPANY
|
|
)
|
|
amount = amount_of_row(outgoing_row)
|
|
return DerivedDecision(
|
|
classification, PAIRING_PAIRED, str(amount),
|
|
str(outgoing_row["currency"] or "").strip() or None,
|
|
_parse_datetime(outgoing_row["transaction_at"]).isoformat(),
|
|
(
|
|
(outgoing_row["id"], "outgoing"),
|
|
(incoming_row["id"], "incoming"),
|
|
),
|
|
(payer, payee), "管理员人工关联",
|
|
)
|
|
|
|
|
|
def _manual_assign_participant(
|
|
connection: sqlite3.Connection,
|
|
event_id: int,
|
|
current: sqlite3.Row,
|
|
participant: dict[str, object] | None,
|
|
reason: str,
|
|
actor: sqlite3.Row,
|
|
) -> DerivedDecision:
|
|
if not participant or not isinstance(participant, dict):
|
|
raise MatchInputError("assign_participant 必须提供 participant。")
|
|
role = str(participant.get("role") or "")
|
|
if role not in ("payer", "payee"):
|
|
raise MatchInputError("participant.role 必须是 payer 或 payee。")
|
|
try:
|
|
company_id = int(participant["company_id"])
|
|
except (KeyError, TypeError, ValueError):
|
|
raise MatchInputError("participant.company_id 必须是有效的公司 id。") from None
|
|
company = connection.execute(
|
|
"SELECT id FROM companies WHERE id = ?", (company_id,)
|
|
).fetchone()
|
|
if company is None:
|
|
raise MatchInputError("participant.company_id 指向的公司不存在。")
|
|
|
|
observations = _decision_observations(connection, current["id"])
|
|
if len(observations) != 1:
|
|
raise MatchInputError("assign_participant 只适用于单边事件。")
|
|
observation = observations[0]
|
|
row = _load_rows(connection, [observation["source_row_id"]])[0]
|
|
resolved = resolve_row(connection, row)
|
|
if resolved.own is None or resolved.direction is None:
|
|
raise MatchInputError("assign_participant 需要本方归属已确认。")
|
|
own_role = "payer" if resolved.direction == "outgoing" else "payee"
|
|
if role == own_role:
|
|
raise MatchInputError("assign_participant 只能指定对方参与方角色。")
|
|
own_participant = _participant(resolved.own, own_role)
|
|
assigned = {
|
|
"role": role,
|
|
"company_id": company_id,
|
|
"bank_account_id": (
|
|
int(participant["bank_account_id"])
|
|
if participant.get("bank_account_id")
|
|
else None
|
|
),
|
|
"resolve_method": "manual",
|
|
"alias_id": None,
|
|
"mapping_id": None,
|
|
"evidence": json.dumps(
|
|
{"via": "manual", "company_id": company_id}, ensure_ascii=False
|
|
),
|
|
}
|
|
participants = (own_participant, assigned)
|
|
classification = (
|
|
CLASSIFICATION_SAME_COMPANY
|
|
if own_participant["company_id"] == company_id
|
|
else CLASSIFICATION_INTERCOMPANY
|
|
)
|
|
amount = amount_of_row(row)
|
|
return DerivedDecision(
|
|
classification, PAIRING_SINGLE, str(amount),
|
|
str(row["currency"] or "").strip() or None,
|
|
_parse_datetime(row["transaction_at"]).isoformat(),
|
|
((row["id"], observation["role"]),),
|
|
participants, "管理员按证据确认参与方",
|
|
)
|
|
|
|
|
|
def _manual_external(
|
|
connection: sqlite3.Connection, event_id: int, current: sqlite3.Row, reason: str
|
|
) -> DerivedDecision:
|
|
observations = _decision_observations(connection, current["id"])
|
|
participants = _decision_participants(connection, current["id"])
|
|
if len(observations) != 1:
|
|
raise MatchInputError("mark_external 只适用于单边事件。")
|
|
observation = observations[0]
|
|
row = _load_rows(connection, [observation["source_row_id"]])[0]
|
|
amount = amount_of_row(row)
|
|
participant_rows = [
|
|
{
|
|
"role": p["role"], "company_id": p["company_id"],
|
|
"bank_account_id": p["bank_account_id"],
|
|
"resolve_method": p["resolve_method"], "alias_id": p["alias_id"],
|
|
"mapping_id": p["mapping_id"], "evidence": p["evidence"],
|
|
}
|
|
for p in participants
|
|
]
|
|
return DerivedDecision(
|
|
CLASSIFICATION_EXTERNAL, PAIRING_NA, str(amount),
|
|
str(row["currency"] or "").strip() or None,
|
|
_parse_datetime(row["transaction_at"]).isoformat(),
|
|
((row["id"], observation["role"]),),
|
|
tuple(participant_rows), "管理员确认外部交易",
|
|
)
|
|
|
|
|
|
def _apply_reversal(
|
|
connection: sqlite3.Connection,
|
|
event_id: int,
|
|
current: sqlite3.Row,
|
|
reason: str,
|
|
actor: sqlite3.Row | None,
|
|
idempotency_key: str | None,
|
|
) -> dict[str, object]:
|
|
now = utc_now()
|
|
revision_row = connection.execute(
|
|
"SELECT COALESCE(MAX(revision), 0) AS m FROM transfer_match_decisions WHERE event_id = ?",
|
|
(event_id,),
|
|
).fetchone()
|
|
revision = int(revision_row["m"]) + 1
|
|
cursor = connection.execute(
|
|
"""
|
|
INSERT INTO transfer_match_decisions (
|
|
event_id, revision, classification, pairing, amount, currency,
|
|
effective_at, mode, rule_version, locked, reason, idempotency_key,
|
|
actor_user_id, actor_username, supersedes_decision_id, created_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
event_id, revision, current["classification"], current["pairing"],
|
|
current["amount"], current["currency"], current["effective_at"],
|
|
MODE_REVERSAL, current["rule_version"], reason, idempotency_key,
|
|
actor["id"] if actor is not None else None,
|
|
actor["username"] if actor is not None else None,
|
|
current["id"], now,
|
|
),
|
|
)
|
|
decision_id = int(cursor.lastrowid)
|
|
connection.execute(
|
|
"DELETE FROM transfer_observation_claims WHERE event_id = ?", (event_id,)
|
|
)
|
|
connection.execute(
|
|
"DELETE FROM current_transfer_decisions WHERE event_id = ?", (event_id,)
|
|
)
|
|
return {"decision_id": decision_id, "event_id": event_id}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Projection rebuild and B-44 read contract
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def rebuild_current_projection(connection: sqlite3.Connection) -> int:
|
|
"""Rebuild current decisions and row claims from the append-only log.
|
|
|
|
For every active event, the current decision is its highest-revision
|
|
decision — unless that decision is a reversal, in which case the event has
|
|
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(
|
|
"""
|
|
SELECT source_row_id FROM transfer_decision_observations
|
|
WHERE decision_id = ? ORDER BY id
|
|
""",
|
|
(event["latest_id"],),
|
|
).fetchall()
|
|
with connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT OR REPLACE INTO current_transfer_decisions (event_id, decision_id)
|
|
VALUES (?, ?)
|
|
""",
|
|
(event["event_id"], event["latest_id"]),
|
|
)
|
|
for observation in observations:
|
|
connection.execute(
|
|
"""
|
|
INSERT OR REPLACE INTO transfer_observation_claims (source_row_id, event_id, decision_id)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
(observation["source_row_id"], event["event_id"], event["latest_id"]),
|
|
)
|
|
rebuilt += 1
|
|
return rebuilt
|
|
|
|
|
|
def eligible_intercompany_events(connection: sqlite3.Connection) -> list[sqlite3.Row]:
|
|
return connection.execute(
|
|
"SELECT * FROM eligible_intercompany_events ORDER BY event_id"
|
|
).fetchall()
|
|
|
|
|
|
def unresolved_amounts(
|
|
connection: sqlite3.Connection,
|
|
company_id: int,
|
|
cutoff: str | None = None,
|
|
) -> list[sqlite3.Row]:
|
|
"""Unresolved amounts a company is exposed to as of ``cutoff``.
|
|
|
|
Only current decisions in ``unresolved``/``needs_review``/``internal_single``
|
|
count as unresolved; paired intercompany, same-company and external events
|
|
are resolved classifications and never appear here.
|
|
"""
|
|
cutoff_where = "AND d.effective_at <= ?" if cutoff else ""
|
|
params: list[object] = []
|
|
if cutoff:
|
|
params.append(cutoff)
|
|
rows = connection.execute(
|
|
f"""
|
|
SELECT d.id AS decision_id, d.event_id, d.classification, d.amount,
|
|
d.currency, d.effective_at, o.role AS direction
|
|
FROM current_transfer_decisions c
|
|
JOIN transfer_match_decisions d ON d.id = c.decision_id
|
|
JOIN transfer_decision_participants p ON p.decision_id = d.id
|
|
JOIN transfer_decision_observations o ON o.decision_id = d.id
|
|
WHERE d.classification IN ('unresolved', 'needs_review', 'internal_single')
|
|
AND p.company_id = ?
|
|
{cutoff_where}
|
|
ORDER BY d.id
|
|
""",
|
|
(company_id, *params),
|
|
).fetchall()
|
|
return rows
|
|
|
|
|
|
def exposed_status(decision: sqlite3.Row | dict) -> str:
|
|
classification = decision["classification"]
|
|
pairing = decision["pairing"]
|
|
if classification == CLASSIFICATION_INTERCOMPANY and pairing == PAIRING_PAIRED:
|
|
return "matched"
|
|
if classification == CLASSIFICATION_INTERCOMPANY:
|
|
return "confirmed_single"
|
|
if classification == CLASSIFICATION_SAME_COMPANY:
|
|
return "same_company_transfer"
|
|
if classification == CLASSIFICATION_EXTERNAL:
|
|
return "external"
|
|
if classification == CLASSIFICATION_NEEDS_REVIEW:
|
|
return "needs_review"
|
|
if classification == CLASSIFICATION_INTERNAL_SINGLE:
|
|
return "internal_single"
|
|
return "unresolved"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# API payload helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _decision_payload(
|
|
connection: sqlite3.Connection, event_id: int, decision_id: int
|
|
) -> dict[str, object]:
|
|
decision = connection.execute(
|
|
"SELECT * FROM transfer_match_decisions WHERE id = ?", (decision_id,)
|
|
).fetchone()
|
|
observations = _decision_observations(connection, decision_id)
|
|
participants = _decision_participants(connection, decision_id)
|
|
return {
|
|
"event_id": event_id,
|
|
"decision_id": decision_id,
|
|
"revision": decision["revision"],
|
|
"classification": decision["classification"],
|
|
"pairing": decision["pairing"],
|
|
"status": exposed_status(decision),
|
|
"amount": decision["amount"],
|
|
"currency": decision["currency"],
|
|
"effective_at": decision["effective_at"],
|
|
"mode": decision["mode"],
|
|
"locked": bool(decision["locked"]),
|
|
"rule_version": decision["rule_version"],
|
|
"source_row_ids": [o["source_row_id"] for o in observations],
|
|
"participants": [
|
|
{
|
|
"role": p["role"], "company_id": p["company_id"],
|
|
"bank_account_id": p["bank_account_id"],
|
|
"resolve_method": p["resolve_method"],
|
|
}
|
|
for p in participants
|
|
],
|
|
}
|