399 lines
15 KiB
Python
399 lines
15 KiB
Python
"""Statutory subject suggestions, mirror mapping and confirmation (B-44).
|
|
|
|
Subjects are stored from one participating company's perspective and the
|
|
other side is the fixed mirror (应收<->应付, 其他应收<->其他应付), so the two
|
|
companies can never record conflicting subjects. Bank summary/purpose text
|
|
only ever produces a *suggestion*; nothing here confirms a subject
|
|
automatically. Confirmation is an explicit administrator decision carrying
|
|
``expected_revision`` and an idempotency key.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
|
|
from . import matching
|
|
|
|
SUBJECTS = ("receivable", "payable", "other_receivable", "other_payable")
|
|
SUBJECT_RULE_VERSION = "subject-suggest-draft-v1"
|
|
|
|
MIRROR = {
|
|
"receivable": "payable",
|
|
"payable": "receivable",
|
|
"other_receivable": "other_payable",
|
|
"other_payable": "other_receivable",
|
|
}
|
|
|
|
SUBJECT_LABELS = {
|
|
"receivable": "应收",
|
|
"payable": "应付",
|
|
"other_receivable": "其他应收",
|
|
"other_payable": "其他应付",
|
|
}
|
|
|
|
_FULL_WIDTH = str.maketrans(
|
|
"ＡＢＣＤＥＦＧＨＩＪＫＬＭＮＯＰＱＲＳＴＵＶＷＸＹＺ"
|
|
"ａｂｃｄｅｆｇｈｉｊｋｌｍｎｏｐｑｒｓｔｕｖｗｘｙｚ０１２３４５６７８９",
|
|
"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
|
|
"abcdefghijklmnopqrstuvwxyz0123456789",
|
|
)
|
|
|
|
# Draft v1 dictionary. Exact-keyword matching only; every hit is a suggestion
|
|
# and never an automatic posting. Trade-type keywords are deliberately absent
|
|
# until the group supplies an approved dictionary (they always go to review).
|
|
_LOAN_LIKE = ("借款", "往来款", "资金往来", "临时借款", "资金调拨", "代垫", "垫付")
|
|
_REPAY_LIKE = ("还款", "归还借款", "归还往来款")
|
|
|
|
|
|
class SubjectConflictError(ValueError):
|
|
"""A stale revision or idempotency conflict (mapped to HTTP 409)."""
|
|
|
|
|
|
class SubjectInputError(ValueError):
|
|
"""Invalid input for a subject decision (mapped to HTTP 400/422)."""
|
|
|
|
|
|
def mirror_subject(subject_code: str) -> str:
|
|
if subject_code not in MIRROR:
|
|
raise SubjectInputError("科目必须是应收/应付/其他应收/其他应付之一。")
|
|
return MIRROR[subject_code]
|
|
|
|
|
|
def subject_label(subject_code: str) -> str:
|
|
return SUBJECT_LABELS.get(subject_code, subject_code)
|
|
|
|
|
|
def _normalize(text: object) -> str:
|
|
return re.sub(r"[\s\ufeff]+", "", str(text or "").translate(_FULL_WIDTH))
|
|
|
|
|
|
def _bank_evidence_texts(connection: sqlite3.Connection, ledger_event_id: int) -> dict[str, str]:
|
|
"""Purpose/summary text of the B-43 source rows behind a bank event."""
|
|
row = connection.execute(
|
|
"""
|
|
SELECT bs.bank_event_id FROM ledger_event_bank_sources bs
|
|
WHERE bs.ledger_event_id = ?
|
|
""",
|
|
(ledger_event_id,),
|
|
).fetchone()
|
|
if row is None:
|
|
return {"purpose": "", "summary": ""}
|
|
decision = matching._current_decision_for_event(connection, row["bank_event_id"])
|
|
if decision is None:
|
|
return {"purpose": "", "summary": ""}
|
|
observations = matching._decision_observations(connection, decision["id"])
|
|
texts: dict[str, list[str]] = {"purpose": [], "summary": []}
|
|
for observation in observations:
|
|
source = connection.execute(
|
|
"SELECT purpose, summary FROM source_rows WHERE id = ?",
|
|
(observation["source_row_id"],),
|
|
).fetchone()
|
|
if source is None:
|
|
continue
|
|
for key in ("purpose", "summary"):
|
|
value = str(source[key] or "").strip()
|
|
if value:
|
|
texts[key].append(value)
|
|
return {
|
|
"purpose": " ".join(texts["purpose"]),
|
|
"summary": " ".join(texts["summary"]),
|
|
}
|
|
|
|
|
|
def compute_suggestions(
|
|
connection: sqlite3.Connection, ledger_event_id: int
|
|
) -> list[dict[str, object]]:
|
|
""" Deterministic draft suggestions for a pending event, never confirmation.
|
|
|
|
Purpose rules take precedence over summary rules. When both loan-like and
|
|
repay-like keywords match, both candidates are returned as a conflict for
|
|
the reviewer; no priority breaks the tie.
|
|
"""
|
|
from .ledger_events import current_revision
|
|
|
|
revision = current_revision(connection, ledger_event_id)
|
|
if revision is None or revision["state"] != "pending_subject":
|
|
return []
|
|
texts = _bank_evidence_texts(connection, ledger_event_id)
|
|
purpose = _normalize(texts["purpose"])
|
|
summary = _normalize(texts["summary"])
|
|
search = purpose or summary
|
|
payer = revision["payer_company_id"]
|
|
payee = revision["payee_company_id"]
|
|
|
|
loan_hit = next((word for word in _LOAN_LIKE if word in search), None)
|
|
repay_hit = next((word for word in _REPAY_LIKE if word in search), None)
|
|
|
|
suggestions: list[dict[str, object]] = []
|
|
if loan_hit:
|
|
suggestions.append(
|
|
{
|
|
"suggested_perspective_company_id": payer,
|
|
"suggested_subject_code": "other_receivable",
|
|
"reason": f"匹配建议词典「{loan_hit}」，建议付款方其他应收",
|
|
"rule_version": SUBJECT_RULE_VERSION,
|
|
"evidence": {
|
|
"keyword": loan_hit,
|
|
"matched_text": search,
|
|
"approved": False,
|
|
},
|
|
}
|
|
)
|
|
if repay_hit:
|
|
suggestions.append(
|
|
{
|
|
"suggested_perspective_company_id": payee,
|
|
"suggested_subject_code": "other_receivable",
|
|
"reason": f"匹配建议词典「{repay_hit}」，建议收款方其他应收",
|
|
"rule_version": SUBJECT_RULE_VERSION,
|
|
"evidence": {
|
|
"keyword": repay_hit,
|
|
"matched_text": search,
|
|
"approved": False,
|
|
},
|
|
}
|
|
)
|
|
return suggestions
|
|
|
|
|
|
def store_suggestions(connection: sqlite3.Connection, ledger_event_id: int) -> int:
|
|
"""Compute and append suggestions for a pending event. Returns count stored."""
|
|
from .ledger_events import current_revision
|
|
|
|
revision = current_revision(connection, ledger_event_id)
|
|
if revision is None or revision["state"] != "pending_subject":
|
|
return 0
|
|
stored = 0
|
|
for suggestion in compute_suggestions(connection, ledger_event_id):
|
|
from .db import utc_now
|
|
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO ledger_subject_suggestions (
|
|
ledger_event_id, source_revision_id,
|
|
suggested_perspective_company_id, suggested_subject_code,
|
|
rule_version, evidence_json, created_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
ledger_event_id, revision["id"],
|
|
suggestion["suggested_perspective_company_id"],
|
|
suggestion["suggested_subject_code"],
|
|
suggestion["rule_version"],
|
|
json.dumps(suggestion.get("evidence", {}), ensure_ascii=False),
|
|
utc_now(),
|
|
),
|
|
)
|
|
stored += 1
|
|
return stored
|
|
|
|
|
|
def confirm_subject(
|
|
connection: sqlite3.Connection,
|
|
ledger_event_id: int,
|
|
*,
|
|
perspective_company_id: int,
|
|
subject_code: str,
|
|
reason: str,
|
|
expected_revision: int | None,
|
|
request_key: str | None,
|
|
actor: sqlite3.Row,
|
|
) -> dict[str, object]:
|
|
"""Confirm a subject, turning a pending event into a confirmed revision."""
|
|
from .ledger_events import append_revision, current_revision
|
|
|
|
reason = (reason or "").strip()
|
|
if not reason:
|
|
raise SubjectInputError("必须填写科目确认依据。")
|
|
if subject_code not in SUBJECTS:
|
|
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,
|
|
state="confirmed",
|
|
effective_at=current["effective_at"],
|
|
amount=current["amount"],
|
|
currency=current["currency"],
|
|
payer_company_id=current["payer_company_id"],
|
|
payee_company_id=current["payee_company_id"],
|
|
perspective_company_id=perspective_company_id,
|
|
subject_code=subject_code,
|
|
source_kind=current["source_kind"],
|
|
source_revision_token=current["source_revision_token"],
|
|
posting_kind=current["posting_kind"],
|
|
reverses_ledger_event_id=current["reverses_ledger_event_id"],
|
|
adjusts_ledger_event_id=current["adjusts_ledger_event_id"],
|
|
rule_version=current["rule_version"] or SUBJECT_RULE_VERSION,
|
|
evidence_json=current["evidence_json"],
|
|
idempotency_key=request_key,
|
|
actor=actor,
|
|
reason=reason,
|
|
supersedes_revision_id=current["id"],
|
|
)
|
|
row = connection.execute(
|
|
"SELECT * FROM ledger_event_revisions WHERE id = ?", (revision_id,)
|
|
).fetchone()
|
|
except Exception:
|
|
if began:
|
|
connection.rollback()
|
|
raise
|
|
else:
|
|
if began:
|
|
connection.commit()
|
|
return _revision_payload(connection, row)
|
|
|
|
|
|
def park_subject(
|
|
connection: sqlite3.Connection,
|
|
ledger_event_id: int,
|
|
*,
|
|
disposition: str,
|
|
reason: str,
|
|
expected_revision: int | None,
|
|
request_key: str | None,
|
|
actor: sqlite3.Row,
|
|
) -> dict[str, object]:
|
|
"""Record 退回/转异常 without confirming a statutory subject.
|
|
|
|
The event stays ``pending_subject`` so it never enters confirmed balances.
|
|
``exception`` is hidden from the active review queue; ``return`` remains
|
|
visible so the company can supplement materials.
|
|
"""
|
|
from .ledger_events import append_revision, current_revision
|
|
|
|
if disposition not in ("return", "exception"):
|
|
raise SubjectInputError("科目处理只能是退回或转异常。")
|
|
reason = (reason or "").strip()
|
|
if not reason:
|
|
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("只有待确认科目的事件可以退回或转异常。")
|
|
if expected_revision is not None and int(expected_revision) not in (
|
|
current["id"], current["revision"],
|
|
):
|
|
raise SubjectConflictError("事件已发生变更，请刷新后重试。")
|
|
evidence = json.loads(current["evidence_json"] or "{}") if current["evidence_json"] else {}
|
|
evidence["admin_disposition"] = disposition
|
|
revision_id = append_revision(
|
|
connection,
|
|
ledger_event_id,
|
|
state="pending_subject",
|
|
effective_at=current["effective_at"],
|
|
amount=current["amount"],
|
|
currency=current["currency"],
|
|
payer_company_id=current["payer_company_id"],
|
|
payee_company_id=current["payee_company_id"],
|
|
perspective_company_id=None,
|
|
subject_code=None,
|
|
source_kind=current["source_kind"],
|
|
source_revision_token=current["source_revision_token"],
|
|
posting_kind=current["posting_kind"],
|
|
reverses_ledger_event_id=current["reverses_ledger_event_id"],
|
|
adjusts_ledger_event_id=current["adjusts_ledger_event_id"],
|
|
rule_version=current["rule_version"] or SUBJECT_RULE_VERSION,
|
|
evidence_json=json.dumps(evidence, ensure_ascii=False),
|
|
idempotency_key=request_key,
|
|
actor=actor,
|
|
reason=reason,
|
|
supersedes_revision_id=current["id"],
|
|
)
|
|
row = connection.execute(
|
|
"SELECT * FROM ledger_event_revisions WHERE id = ?", (revision_id,)
|
|
).fetchone()
|
|
except Exception:
|
|
if began:
|
|
connection.rollback()
|
|
raise
|
|
else:
|
|
if began:
|
|
connection.commit()
|
|
return _revision_payload(connection, row)
|
|
|
|
|
|
def _revision_payload(connection: sqlite3.Connection, revision: sqlite3.Row) -> dict[str, object]:
|
|
company = connection.execute(
|
|
"SELECT name FROM companies WHERE id = ?", (revision["perspective_company_id"],)
|
|
).fetchone()
|
|
return {
|
|
"ledger_event_id": revision["ledger_event_id"],
|
|
"revision_id": revision["id"],
|
|
"revision": revision["revision"],
|
|
"state": revision["state"],
|
|
"effective_at": revision["effective_at"],
|
|
"amount": revision["amount"],
|
|
"currency": revision["currency"],
|
|
"payer_company_id": revision["payer_company_id"],
|
|
"payee_company_id": revision["payee_company_id"],
|
|
"perspective_company_id": revision["perspective_company_id"],
|
|
"perspective_company_name": company["name"] if company else None,
|
|
"subject_code": revision["subject_code"],
|
|
"subject_label": subject_label(revision["subject_code"])
|
|
if revision["subject_code"]
|
|
else None,
|
|
"posting_kind": revision["posting_kind"],
|
|
"source_kind": revision["source_kind"],
|
|
"reason": revision["reason"],
|
|
"created_at": revision["created_at"],
|
|
}
|