Compare commits

...
Author SHA1 Message Date
腾讯WorkBuddy 826cb45bb0 B-248: 收口登录错误态、焦点与桌面总宽 2026-08-20 18:03:37 +08:00
腾讯WorkBuddyandCursor 79bd32f46e B-229: 登录页改为桌面双区布局并收紧三档响应式
把居中手机卡改回横向双区,端口选择复用分段控件节奏,避免再叠加一套登录样式。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-20 16:40:22 +08:00
腾讯WorkBuddyandCursor 4add17f2b8 B-44: 返修——768px 超大期末金额与未决状态避让
在 376–900px 五列目录把方向 chip 与期末金额分行,并加 Chromium 盒模型回归,避免 15 位金额盖住未决状态。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-20 03:22:13 +08:00
腾讯WorkBuddyandCursor 95ad9c939a B-44: 返修——375px 超大金额降级与审核弹窗必展字段
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-20 01:54:32 +08:00
腾讯WorkBuddyandCursor 1fe073066e B-44: 返修——余额目录视觉、八列事件表、抽屉键盘与审核三项决定
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-20 00:37:27 +08:00
腾讯WorkBuddy f16e8aad34 B-44: 返修——多币种对方汇总、镜像科目筛选、冲销独立生效日、创建来源撤销边界与并发幂等 2026-08-19 13:28:28 +08:00
腾讯WorkBuddy d59576739f B-44: fix empty-state retry selector in balance directory 2026-08-19 12:00:59 +08:00
腾讯WorkBuddy 85293b79df B-44: intercompany ledger events, subject review and drill-down evidence
- migration 6: manual_records, ledger_event_revisions chain, current
  projections, source claims, subject suggestions, eligible_position_events
- ledger_events.py: bank-event reconciliation, reversal/adjustment/reopen,
  append-only revision chain and rebuildable current projection
- subjects.py: fixed subject mirror, draft suggestion dictionary, explicit
  administrator subject confirmation with expected_revision + idempotency
- manual_records.py: submit, approve new/link, return/exception/reverse,
  candidate hints, idempotent replay and concurrency-safe claims
- positions.py: Decimal aggregation, both-perspective conservation asserts,
  cutoff window, unresolved gross buckets, keyset pagination, evidence
  visibility (visible/masked/missing)
- server.py: admin + company intercompany APIs with tenant isolation (404 on
  cross-tenant reads, 403 on company writes) and auto reconcile wiring
- admin/company portals: balance directory, pair drill-down drawer, evidence
  drawer, subject/manual audit queue, company balance summary
- tests: ledger events, subjects, manual records, positions, HTTP API and
  migration persistence (233 total, all green)
2026-08-19 11:58:25 +08:00
25 changed files with 9315 additions and 231 deletions
@@ -0,0 +1,87 @@
# 006: Intercompany positions, subject review and drill-down evidence (B-44)
Status: accepted (2026-08-19)
## Decision
The intercompany ledger adds one canonical layer above B-43's eligible events:
- **Append-only ledger events.** A canonical `ledger_event` carries an
append-only `ledger_event_revisions` chain. A B-43 eligible bank event first
becomes a `pending_subject` revision; an administrator confirms the subject
into a `confirmed` revision. Corrections are never in-place edits — a
reversal or adjustment is a *new* ledger event with its own effective date,
and the original event keeps its history so earlier cutoffs are not
rewritten.
- **Manual records as immutable submitted facts.** Companies submit
`manual_records`; only an administrator-approved record becomes a confirmed
ledger event (`approve_new`) or joins one (`approve_link`). Returned,
exception and pending records never affect a balance and never leak to the
counterparty. Approved facts change only through `reverse` (a new opposite
event) — the original is never edited.
- **One perspective, fixed mirror.** Subjects are stored from one participating
company's perspective (`receivable/payable/other_receivable/other_payable`);
the other side is the fixed mirror (应收<->应付, 其他应收<->其他应付), so the
two companies can never book conflicting subjects.
- **Subjects are confirmed, never auto-posted.** Bank summary/purpose text only
feeds a deterministic *suggestion* dictionary (`subject-suggest-draft-v1`,
not group-approved). Without an approved trade dictionary every bank event
stays in subject review until an administrator confirms.
- **Decimal-only aggregation.** All money is stored as TEXT decimal strings and
aggregated with Python `Decimal`. SQLite `SUM`, JavaScript `Number` and
Python `float` never touch financial math. Different currencies are
aggregated and displayed separately; nothing is converted to a group total.
- **Conservation is asserted.** For every company pair and currency the two
perspectives must mirror exactly (`C_A == -C_B`); a violation raises a
calculation exception instead of rendering an unbalanced number.
- **Unresolved is absolute gross.** Unresolved amounts are summed by absolute
value per currency (never netted), broken down by reason
(`subject_review`, `unmatched_single`, `manual_pending`), with count and
gross amount exposed on every balance response.
- **B-45 boundary.** Until the B-45 opening balance exists, every response
returns `opening.status=unavailable`, `opening.amount=null` and
`result.kind=period_net_change`; the UI labels this "期间净变动", never
"期末余额".
## Background
B-43 produces `eligible_intercompany_events` as the only bank-event entry
point. Before B-44 there was no canonical financial event, no statutory
subject, no manual-record approval, and no server-side balance API. The
revision-chain design is inherited from `transfer_match_decisions` in
migration 5 and from the append-only audit posture of the rest of the system.
## Consequences
- **Positive:** balances are deterministic, conservable, auditable and
drillable from a group directory down to bank source rows; manual records
cannot double count; corrections never mutate evidence; company portals are
tenant-scoped on the server.
- **Negative:** pending bank events and returned/exception manual records are
intentionally invisible to counterparties, which can surprise cashiers who
expect symmetric disclosure; subject confirmation is manual until a
group-approved dictionary exists.
- **Operational:** migration 6 is forward-only for production once
approvals/revisions exist; pre-production it can be rolled back with
`--rollback-to 5`. Read aggregation runs against the rebuildable current
projection (no day snapshots yet); if B-45 monthly close needs them, immutable
monthly snapshots can be added behind the same API contract.
## Files
- `src/bank_importer/ledger_events.py` — event lifecycle, revision chain,
bank reconciliation, reversal/adjustment/reopen, projection rebuild.
- `src/bank_importer/subjects.py` — subject constants/mirror, suggestion
dictionary, `confirm_subject`.
- `src/bank_importer/manual_records.py` — submission, approval
(new/link), return/exception/reverse, candidate hints, idempotency.
- `src/bank_importer/positions.py` — Decimal aggregation, directories, pairs,
events, evidence visibility, unresolved buckets, keyset pagination.
- `server.py``/api/admin/intercompany/*` and `/api/company/intercompany/*`
plus `/api/admin/subject-reviews` and `/api/admin/manual-records`.
- `db.py` migration 6 — `manual_records`, `manual_record_decisions`,
`ledger_events`, `ledger_event_revisions`, current-pointer projections,
source claim tables, `ledger_subject_suggestions`, `eligible_position_events`.
- Tests: `test_ledger_events.py`, `test_manual_records.py`,
`test_positions.py`, `test_positions_api.py`, extended
`test_persistence.py`.
+829 -1
View File
@@ -10,7 +10,10 @@ from http.cookies import SimpleCookie
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse
from bank_importer import auth, importing, master_data, matching, multipart, personal_transit
from bank_importer import (
auth, importing, ledger_events, manual_records, master_data, matching,
multipart, personal_transit, positions, subjects,
)
from bank_importer.db import connect, migrate, utc_now
@@ -101,6 +104,68 @@ class AppHandler(SimpleHTTPRequestHandler):
self._handle_company_transfer_event_detail(int(company_event_match.group(1)))
return
# B-44 intercompany positions (admin)
if path == "/api/admin/intercompany/balances":
self._handle_admin_intercompany_balances(query)
return
admin_pair = re.fullmatch(
r"/api/admin/intercompany/pairs/(\d+)/(\d+)", path
)
if admin_pair:
self._handle_admin_intercompany_pair(
int(admin_pair.group(1)), int(admin_pair.group(2)), query
)
return
if path == "/api/admin/intercompany/events":
self._handle_admin_intercompany_events(query)
return
if path == "/api/admin/subject-reviews":
self._handle_admin_subject_reviews(query)
return
if path == "/api/admin/manual-records":
self._handle_admin_manual_records(query)
return
admin_event_match = re.fullmatch(r"/api/admin/intercompany/events/(\d+)", path)
if admin_event_match:
self._handle_admin_intercompany_event_detail(int(admin_event_match.group(1)))
return
admin_evidence = re.fullmatch(
r"/api/admin/intercompany/events/(\d+)/evidence", path
)
if admin_evidence:
self._handle_admin_intercompany_evidence(int(admin_evidence.group(1)))
return
# B-44 intercompany positions (company, own-company scope)
if path == "/api/company/intercompany/balances":
self._handle_company_intercompany_balances(query)
return
company_pair = re.fullmatch(r"/api/company/intercompany/pairs/(\d+)", path)
if company_pair:
self._handle_company_intercompany_pair(int(company_pair.group(1)), query)
return
if path == "/api/company/intercompany/events":
self._handle_company_intercompany_events(query)
return
if path == "/api/company/manual-records":
self._handle_company_manual_records(query)
return
if path == "/api/company/companies":
self._handle_company_companies()
return
company_event_match = re.fullmatch(r"/api/company/intercompany/events/(\d+)", path)
if company_event_match:
self._handle_company_intercompany_event_detail(
int(company_event_match.group(1))
)
return
company_evidence = re.fullmatch(
r"/api/company/intercompany/events/(\d+)/evidence", path
)
if company_evidence:
self._handle_company_intercompany_evidence(int(company_evidence.group(1)))
return
if path == "/admin.html" and not self._guard_page("admin"):
return
if path == "/company.html" and not self._guard_page("company"):
@@ -167,6 +232,29 @@ class AppHandler(SimpleHTTPRequestHandler):
if mapping_review:
self._handle_admin_review_personal_mapping(int(mapping_review.group(1)))
return
# B-44 intercompany positions (admin writes)
subject_decision = re.fullmatch(
r"/api/admin/intercompany/events/(\d+)/subject-decisions", path
)
if subject_decision:
self._handle_admin_subject_decision(int(subject_decision.group(1)))
return
adjustment = re.fullmatch(
r"/api/admin/intercompany/events/(\d+)/adjustments", path
)
if adjustment:
self._handle_admin_intercompany_adjustment(int(adjustment.group(1)))
return
manual_decision = re.fullmatch(r"/api/admin/manual-records/(\d+)/decisions", path)
if manual_decision:
self._handle_admin_manual_record_decision(int(manual_decision.group(1)))
return
# B-44 intercompany positions (company writes)
if path == "/api/company/manual-records":
self._handle_company_manual_records_submit()
return
self._send_json(404, {"status": "error", "message": "接口不存在。"})
# ------------------------------------------------------------------
@@ -1611,6 +1699,11 @@ class AppHandler(SimpleHTTPRequestHandler):
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
auth.audit(
connection, "transfer_reconcile", actor=user,
target=f"rows:{len(row_ids)}",
@@ -1664,6 +1757,11 @@ 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
self._send_json(200, {"status": "ok", "decision": payload})
finally:
connection.close()
@@ -1974,6 +2072,736 @@ class AppHandler(SimpleHTTPRequestHandler):
return rows
# ------------------------------------------------------------------
# B-44 intercompany positions (admin)
# ------------------------------------------------------------------
def _parse_window(self, query: dict[str, list[str]]) -> tuple[str, str]:
raw_from = (query.get("from") or [None])[0]
raw_cutoff = (query.get("cutoff") or [None])[0]
if raw_cutoff is None:
raw_cutoff = positions.today_shanghai()
try:
return positions.validate_window(raw_from, raw_cutoff)
except positions.PositionInputError as exc:
self._send_json(400, {"status": "error", "message": str(exc)})
raise
def _parse_limit(self, query: dict[str, list[str]], default: int = 50) -> int:
raw = (query.get("limit") or [str(default)])[0]
try:
return max(1, min(int(raw), 200))
except ValueError:
return default
def _parse_int(self, query: dict[str, list[str]], key: str, label: str) -> int | None:
raw = (query.get(key) or [None])[0]
if not raw:
return None
try:
return int(raw)
except ValueError:
self._send_json(400, {"status": "error", "message": f"{label}参数无效。"})
raise
def _handle_admin_intercompany_balances(self, query: dict[str, list[str]]) -> None:
connection = connect(DB_PATH)
try:
user = self._require_admin(connection)
if user is None:
return
try:
from_, cutoff = self._parse_window(query)
company_id = self._parse_int(query, "company_id", "company_id")
currency = (query.get("currency") or [None])[0]
limit = self._parse_limit(query)
cursor = (query.get("cursor") or [None])[0]
except Exception:
return
try:
payload = positions.company_balances(
connection, from_=from_, cutoff=cutoff, currency=currency,
company_id=company_id, limit=limit, cursor=cursor,
)
except positions.PositionInputError as exc:
self._send_json(400, {"status": "error", "message": str(exc)})
return
self._send_json(200, {"status": "ok", **payload})
finally:
connection.close()
def _handle_admin_intercompany_pair(
self, company_a: int, company_b: int, query: dict[str, list[str]]
) -> None:
connection = connect(DB_PATH)
try:
user = self._require_admin(connection)
if user is None:
return
if not self._companies_exist(connection, (company_a, company_b)):
self._send_json(404, {"status": "error", "message": "公司不存在。"})
return
try:
from_, cutoff = self._parse_window(query)
currency = (query.get("currency") or [None])[0]
except Exception:
return
try:
payload = positions.pair_detail(
connection, company_a, company_b,
from_=from_, cutoff=cutoff, currency=currency,
)
except positions.PositionError as exc:
self._send_json(422, {"status": "error", "message": str(exc)})
return
except positions.PositionInputError as exc:
self._send_json(400, {"status": "error", "message": str(exc)})
return
self._send_json(200, {"status": "ok", **payload})
finally:
connection.close()
def _companies_exist(self, connection, ids: tuple[int, ...]) -> bool:
for company_id in ids:
row = connection.execute(
"SELECT id FROM companies WHERE id = ?", (company_id,)
).fetchone()
if row is None:
return False
return True
def _handle_admin_intercompany_events(self, query: dict[str, list[str]]) -> None:
connection = connect(DB_PATH)
try:
user = self._require_admin(connection)
if user is None:
return
try:
from_, cutoff = self._parse_window(query)
currency = (query.get("currency") or [None])[0]
company_id = self._parse_int(query, "company_id", "company_id")
company_a = self._parse_int(query, "company_a", "company_a")
company_b = self._parse_int(query, "company_b", "company_b")
subject = (query.get("subject") or [None])[0]
state = (query.get("state") or [None])[0]
posting_kind = (query.get("posting_kind") or [None])[0]
source_kind = (query.get("source_kind") or [None])[0]
limit = self._parse_limit(query)
cursor = (query.get("cursor") or [None])[0]
except Exception:
return
try:
payload = positions.list_events(
connection, from_=from_, cutoff=cutoff, currency=currency,
company_id=company_id, company_a=company_a, company_b=company_b,
subject=subject, state=state, posting_kind=posting_kind,
source_kind=source_kind, limit=limit, cursor=cursor,
)
except positions.PositionInputError as exc:
self._send_json(400, {"status": "error", "message": str(exc)})
return
self._send_json(200, {"status": "ok", **payload})
finally:
connection.close()
def _handle_admin_subject_reviews(self, query: dict[str, list[str]]) -> None:
connection = connect(DB_PATH)
try:
user = self._require_admin(connection)
if user is None:
return
try:
from_, cutoff = self._parse_window(query)
company_id = self._parse_int(query, "company_id", "company_id")
limit = self._parse_limit(query)
cursor = (query.get("cursor") or [None])[0]
except Exception:
return
try:
payload = positions.subject_review_queue(
connection, from_=from_, cutoff=cutoff,
company_id=company_id, limit=limit, cursor=cursor,
)
except positions.PositionInputError as exc:
self._send_json(400, {"status": "error", "message": str(exc)})
return
self._send_json(200, {"status": "ok", **payload})
finally:
connection.close()
def _handle_admin_manual_records(self, query: dict[str, list[str]]) -> None:
connection = connect(DB_PATH)
try:
user = self._require_admin(connection)
if user is None:
return
company_id = self._parse_int(query, "company_id", "company_id")
if company_id is None and (query.get("company_id") or [None])[0]:
return
state = (query.get("state") or [None])[0]
limit = self._parse_limit(query, 100)
try:
rows = manual_records.list_records(
connection, company_id=company_id, state=state, limit=limit
)
except manual_records.ManualInputError as exc:
self._send_json(400, {"status": "error", "message": str(exc)})
return
items = []
for row in rows:
payload = manual_records._row_payload(connection, row)
items.append(payload)
self._send_json(200, {"status": "ok", "records": items})
finally:
connection.close()
def _handle_admin_intercompany_event_detail(self, event_id: int) -> None:
connection = connect(DB_PATH)
try:
user = self._require_admin(connection)
if user is None:
return
payload = positions.event_detail(connection, event_id)
if payload is None:
self._send_json(404, {"status": "error", "message": "事件不存在。"})
return
self._send_json(200, {"status": "ok", **payload})
finally:
connection.close()
def _handle_admin_intercompany_evidence(self, event_id: int) -> None:
connection = connect(DB_PATH)
try:
user = self._require_admin(connection)
if user is None:
return
payload = positions.event_evidence(connection, event_id)
if payload is None:
self._send_json(404, {"status": "error", "message": "事件不存在。"})
return
self._send_json(200, {"status": "ok", **payload})
finally:
connection.close()
def _handle_admin_subject_decision(self, event_id: int) -> None:
connection = connect(DB_PATH)
try:
user = self._require_admin(connection)
if user is None:
return
data = self._read_json_body()
if data is None:
return
try:
perspective_company_id = int(data["perspective_company_id"])
expected_revision = (
int(data["expected_revision"])
if data.get("expected_revision") is not None
else None
)
except (KeyError, TypeError, ValueError):
self._send_json(
400, {"status": "error", "message": "perspective_company_id 或 expected_revision 无效。"}
)
return
try:
action = str(data.get("action") or "confirm")
if action in ("return", "exception"):
payload = subjects.park_subject(
connection,
event_id,
disposition=action,
reason=str(data.get("reason") or ""),
expected_revision=expected_revision,
request_key=str(data.get("request_key") or "") or None,
actor=user,
)
else:
payload = subjects.confirm_subject(
connection,
event_id,
perspective_company_id=perspective_company_id,
subject_code=str(data.get("subject_code") or ""),
reason=str(data.get("reason") or ""),
expected_revision=expected_revision,
request_key=str(data.get("request_key") or "") or None,
actor=user,
)
except subjects.SubjectConflictError as exc:
self._send_json(409, {"status": "error", "message": str(exc)})
return
except subjects.SubjectInputError as exc:
self._send_json(400, {"status": "error", "message": str(exc)})
return
auth.audit(
connection, "subject_confirm", actor=user,
target=f"ledger_event:{event_id}",
detail=f"revision:{payload['revision']};subject:{payload['subject_code']}",
ip=self._client_ip,
)
self._send_json(200, {"status": "ok", "revision": payload})
finally:
connection.close()
def _handle_admin_intercompany_adjustment(self, event_id: int) -> None:
connection = connect(DB_PATH)
try:
user = self._require_admin(connection)
if user is None:
return
data = self._read_json_body()
if data is None:
return
action = str(data.get("action") or "")
reason = str(data.get("reason") or "")
request_key = str(data.get("request_key") or "") or None
if not reason:
self._send_json(400, {"status": "error", "message": "必须填写操作原因。"})
return
try:
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:
self._send_json(
400,
{"status": "error", "message": "action 必须是 reverse、adjust 或 reopen。"},
)
return
except ledger_events.LedgerConflictError as exc:
self._send_json(409, {"status": "error", "message": str(exc)})
return
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()
def _handle_admin_manual_record_decision(self, record_id: int) -> None:
connection = connect(DB_PATH)
try:
user = self._require_admin(connection)
if user is None:
return
data = self._read_json_body()
if data is None:
return
try:
expected_decision_id = (
int(data["expected_decision_id"])
if data.get("expected_decision_id") is not None
else None
)
except (TypeError, ValueError):
self._send_json(400, {"status": "error", "message": "expected_decision_id 无效。"})
return
try:
payload = manual_records.decide(
connection,
record_id,
str(data.get("action") or ""),
reason=str(data.get("reason") or ""),
expected_decision_id=expected_decision_id,
request_key=str(data.get("request_key") or "") or None,
actor=user,
subject_code=data.get("subject_code"),
target_ledger_event_id=data.get("target_ledger_event_id"),
effective_at=data.get("effective_at"),
)
except manual_records.ManualConflictError as exc:
self._send_json(409, {"status": "error", "message": str(exc)})
return
except manual_records.ManualInputError as exc:
self._send_json(400, {"status": "error", "message": str(exc)})
return
self._send_json(200, {"status": "ok", "decision": payload})
finally:
connection.close()
# ------------------------------------------------------------------
# B-44 intercompany positions (company, own-company scope only)
# ------------------------------------------------------------------
def _company_intercompany_scope(self, connection):
"""Reject company users with 403; return the session company id."""
user = self._require_user(connection)
if user is None:
return None, None
if user["role"] != "company":
self._send_json(403, {"status": "error", "message": "该操作仅限公司用户。"})
return None, None
return user, user["company_id"]
def _handle_company_intercompany_balances(self, query: dict[str, list[str]]) -> None:
connection = connect(DB_PATH)
try:
user, company_id = self._company_intercompany_scope(connection)
if company_id is None:
return
try:
from_, cutoff = self._parse_window(query)
currency = (query.get("currency") or [None])[0]
limit = self._parse_limit(query)
cursor = (query.get("cursor") or [None])[0]
except Exception:
return
try:
payload = positions.company_balances(
connection, from_=from_, cutoff=cutoff, currency=currency,
company_id=company_id, limit=limit, cursor=cursor,
)
counterparties = self._company_counterparty_summary(
connection, company_id, from_, cutoff, currency
)
except positions.PositionInputError as exc:
self._send_json(400, {"status": "error", "message": str(exc)})
return
self._send_json(200, {"status": "ok", **payload, "counterparties": counterparties})
finally:
connection.close()
def _company_counterparty_summary(
self, connection, company_id: int, from_: str, cutoff: str, currency: str | None
) -> list[dict[str, object]]:
events = positions.load_events(
connection, from_=from_, cutoff=cutoff, currency=currency,
company_id=company_id,
)
# Buckets are keyed by ``(counterparty, currency)``: amounts never mix
# across currencies, so one counterparty with CNY and USD yields two rows.
buckets: dict[tuple[int, str], dict[str, object]] = {}
for event in events:
counterparty = (
event["payee_company_id"]
if int(event["payer_company_id"]) == int(company_id)
else event["payer_company_id"]
)
name = (
event["payee_company_name"]
if int(event["payer_company_id"]) == int(company_id)
else event["payer_company_name"]
)
key = (int(counterparty), event["currency"])
bucket = buckets.setdefault(
key,
{
"counterparty_company_id": counterparty,
"counterparty_company_name": name,
"currency": event["currency"],
"signed": 0,
"event_count": 0,
},
)
bucket["signed"] += positions.signed_amount(event, company_id)
bucket["event_count"] += 1
# Counterparties with pending-subject exposure also show up so the
# company portal never hides unconfirmed balances.
pending = connection.execute(
"""
SELECT p.payer_company_id, p.payee_company_id, p.amount, p.currency,
cpayer.name AS payer_company_name, cpayee.name AS payee_company_name
FROM current_ledger_event_revisions cur
JOIN ledger_event_revisions p ON p.id = cur.revision_id
JOIN companies cpayer ON cpayer.id = p.payer_company_id
JOIN companies cpayee ON cpayee.id = p.payee_company_id
WHERE p.state = 'pending_subject'
AND p.effective_at >= ? AND p.effective_at <= ?
AND (p.payer_company_id = ? OR p.payee_company_id = ?)
""",
(from_, cutoff + "T23:59:59", company_id, company_id),
).fetchall()
for row in pending:
if currency and row["currency"] != currency:
continue
counterparty = (
row["payee_company_id"]
if int(row["payer_company_id"]) == int(company_id)
else row["payer_company_id"]
)
name = (
row["payee_company_name"]
if int(row["payer_company_id"]) == int(company_id)
else row["payer_company_name"]
)
buckets.setdefault(
(int(counterparty), row["currency"]),
{
"counterparty_company_id": counterparty,
"counterparty_company_name": name,
"currency": row["currency"],
"signed": 0,
"event_count": 0,
},
)
items = []
for (counterparty, cur), bucket in sorted(buckets.items()):
signed = bucket["signed"]
unresolved = positions.unresolved_for_company(
connection, company_id, cutoff, currency=cur,
counterparty_filter=(company_id, counterparty),
)
direction = "receivable" if signed > 0 else ("payable" if signed < 0 else None)
items.append(
{
"counterparty_company_id": counterparty,
"counterparty_company_name": bucket["counterparty_company_name"],
"currency": cur,
"result": {
"kind": "period_net_change",
"signed_amount": str(signed),
"direction": direction,
"label": "期间净变动",
},
"unresolved": unresolved,
"event_count": bucket["event_count"],
}
)
return items
def _handle_company_intercompany_pair(
self, counterparty_id: int, query: dict[str, list[str]]
) -> None:
connection = connect(DB_PATH)
try:
user, company_id = self._company_intercompany_scope(connection)
if company_id is None:
return
if not self._companies_exist(connection, (counterparty_id,)):
self._send_json(404, {"status": "error", "message": "公司不存在。"})
return
try:
from_, cutoff = self._parse_window(query)
currency = (query.get("currency") or [None])[0]
except Exception:
return
try:
payload = positions.pair_detail(
connection, company_id, counterparty_id,
from_=from_, cutoff=cutoff, currency=currency,
)
except positions.PositionError as exc:
self._send_json(422, {"status": "error", "message": str(exc)})
return
except positions.PositionInputError as exc:
self._send_json(400, {"status": "error", "message": str(exc)})
return
self._send_json(200, {"status": "ok", **payload})
finally:
connection.close()
def _handle_company_intercompany_events(self, query: dict[str, list[str]]) -> None:
connection = connect(DB_PATH)
try:
user, company_id = self._company_intercompany_scope(connection)
if company_id is None:
return
try:
from_, cutoff = self._parse_window(query)
currency = (query.get("currency") or [None])[0]
subject = (query.get("subject") or [None])[0]
state = (query.get("state") or [None])[0]
posting_kind = (query.get("posting_kind") or [None])[0]
source_kind = (query.get("source_kind") or [None])[0]
limit = self._parse_limit(query)
cursor = (query.get("cursor") or [None])[0]
except Exception:
return
try:
payload = positions.list_events(
connection, from_=from_, cutoff=cutoff, currency=currency,
subject=subject, state=state, posting_kind=posting_kind,
source_kind=source_kind, viewer_company_id=company_id,
limit=limit, cursor=cursor,
)
except positions.PositionInputError as exc:
self._send_json(400, {"status": "error", "message": str(exc)})
return
self._send_json(200, {"status": "ok", **payload})
finally:
connection.close()
def _handle_company_intercompany_event_detail(self, event_id: int) -> None:
connection = connect(DB_PATH)
try:
user, company_id = self._company_intercompany_scope(connection)
if company_id is None:
return
payload = positions.event_detail(connection, event_id)
if payload is None:
self._send_json(404, {"status": "error", "message": "事件不存在。"})
return
event = payload["event"]
if company_id not in (event["payer_company_id"], event["payee_company_id"]):
self._send_json(404, {"status": "error", "message": "事件不存在。"})
return
event.update(
positions.event_payload(
connection,
connection.execute(
positions._DETAIL_SELECT + " WHERE p.ledger_event_id = ?",
(event_id,),
).fetchone(),
viewer_company_id=company_id,
)
)
self._send_json(200, {"status": "ok", **payload})
finally:
connection.close()
def _handle_company_intercompany_evidence(self, event_id: int) -> None:
connection = connect(DB_PATH)
try:
user, company_id = self._company_intercompany_scope(connection)
if company_id is None:
return
detail = positions.event_detail(connection, event_id)
if detail is None:
self._send_json(404, {"status": "error", "message": "事件不存在。"})
return
event = detail["event"]
if company_id not in (event["payer_company_id"], event["payee_company_id"]):
self._send_json(404, {"status": "error", "message": "事件不存在。"})
return
payload = positions.event_evidence(
connection, event_id, viewer_company_id=company_id
)
if payload is None:
self._send_json(404, {"status": "error", "message": "事件不存在。"})
return
self._send_json(200, {"status": "ok", **payload})
finally:
connection.close()
def _handle_company_companies(self) -> None:
connection = connect(DB_PATH)
try:
user, company_id = self._company_intercompany_scope(connection)
if company_id is None:
return
rows = connection.execute(
"SELECT id, name FROM companies ORDER BY id"
).fetchall()
self._send_json(
200,
{"status": "ok", "companies": [dict(row) for row in rows]},
)
finally:
connection.close()
def _handle_company_manual_records(self, query: dict[str, list[str]]) -> None:
connection = connect(DB_PATH)
try:
user, company_id = self._company_intercompany_scope(connection)
if company_id is None:
return
state = (query.get("state") or [None])[0]
limit = self._parse_limit(query, 100)
try:
rows = manual_records.list_records(
connection, company_id=company_id, state=state, limit=limit
)
except manual_records.ManualInputError as exc:
self._send_json(400, {"status": "error", "message": str(exc)})
return
# A company only ever sees its own submissions, never declarations
# where it is merely the counterparty.
own = [
manual_records._row_payload(connection, row)
for row in rows
if row["company_id"] == company_id
]
self._send_json(200, {"status": "ok", "records": own})
finally:
connection.close()
def _handle_company_manual_records_submit(self) -> None:
connection = connect(DB_PATH)
try:
user, company_id = self._company_intercompany_scope(connection)
if company_id is None:
return
data = self._read_json_body()
if data is None:
return
try:
payload = manual_records.submit(
connection,
company_id=company_id,
counterparty_company_id=int(data["counterparty_company_id"]),
occurred_at=str(data.get("occurred_at") or ""),
direction=str(data.get("direction") or ""),
amount=str(data.get("amount") or ""),
currency=str(data.get("currency") or ""),
funding_source=str(data.get("funding_source") or ""),
requested_subject=str(data.get("requested_subject") or ""),
request_key=str(data.get("request_key") or ""),
actor=user,
bank_account_id=data.get("bank_account_id"),
personal_transit_mapping_id=data.get("personal_transit_mapping_id"),
related_source_row_id=data.get("related_source_row_id"),
summary=data.get("summary"),
reason=data.get("reason"),
evidence=data.get("evidence"),
)
except manual_records.ManualConflictError as exc:
self._send_json(409, {"status": "error", "message": str(exc)})
return
except manual_records.ManualInputError as exc:
self._send_json(400, {"status": "error", "message": str(exc)})
return
auth.audit(
connection, "manual_submit", actor=user,
target=f"manual_record:{payload['id']}",
detail=f"amount:{payload['amount']};currency:{payload['currency']}",
ip=self._client_ip,
)
self._send_json(200, {"status": "ok", "record": payload})
finally:
connection.close()
# ------------------------------------------------------------------
# Request/response plumbing
# ------------------------------------------------------------------
+229
View File
@@ -569,6 +569,235 @@ MIGRATIONS: tuple[Migration, ...] = (
ALTER TABLE import_batches DROP COLUMN upload_bank_account_id;
""",
),
Migration(
version=6,
name="0006_intercompany_ledger_events",
# B-44 canonical intercompany ledger layer. Manual records and the
# ledger event revision chain are append-only facts; current pointers
# (current revision per ledger event / manual decision, source claims)
# are rebuildable projections. Bank events enter only through
# ``eligible_intercompany_events``; nothing here rewrites bank rows.
up="""
CREATE TABLE manual_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
company_id INTEGER NOT NULL REFERENCES companies (id),
counterparty_company_id INTEGER NOT NULL REFERENCES companies (id),
occurred_at TEXT NOT NULL,
direction TEXT NOT NULL CHECK (direction IN ('outgoing', 'incoming')),
amount TEXT NOT NULL,
amount_scale INTEGER NOT NULL,
currency TEXT NOT NULL,
funding_source TEXT NOT NULL CHECK (funding_source IN (
'approved_bank_account', 'personal_transit', 'other'
)),
bank_account_id INTEGER REFERENCES bank_accounts (id),
personal_transit_mapping_id INTEGER REFERENCES personal_transit_mappings (id),
related_source_row_id INTEGER REFERENCES source_rows (id),
requested_subject TEXT NOT NULL CHECK (requested_subject IN (
'receivable', 'payable', 'other_receivable', 'other_payable'
)),
summary TEXT,
reason TEXT,
evidence_json TEXT,
request_key TEXT NOT NULL,
supersedes_record_id INTEGER REFERENCES manual_records (id),
submitted_by INTEGER REFERENCES users (id),
created_at TEXT NOT NULL,
UNIQUE (company_id, request_key),
CHECK (counterparty_company_id != company_id)
);
CREATE TABLE manual_record_decisions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
record_id INTEGER NOT NULL REFERENCES manual_records (id),
revision INTEGER NOT NULL,
state TEXT NOT NULL CHECK (state IN (
'pending', 'approved', 'returned', 'exception', 'reversed'
)),
action TEXT NOT NULL,
reason TEXT,
actor_user_id INTEGER REFERENCES users (id),
actor_username TEXT,
idempotency_key TEXT,
supersedes_decision_id INTEGER REFERENCES manual_record_decisions (id),
created_at TEXT NOT NULL,
UNIQUE (record_id, revision)
);
CREATE TABLE current_manual_record_decisions (
record_id INTEGER PRIMARY KEY REFERENCES manual_records (id),
decision_id INTEGER NOT NULL UNIQUE REFERENCES manual_record_decisions (id)
);
CREATE TABLE ledger_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
lifecycle TEXT NOT NULL DEFAULT 'active'
CHECK (lifecycle IN ('active', 'superseded')),
created_at TEXT NOT NULL
);
CREATE TABLE ledger_event_revisions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ledger_event_id INTEGER NOT NULL REFERENCES ledger_events (id),
revision INTEGER NOT NULL,
state TEXT NOT NULL CHECK (state IN ('pending_subject', 'confirmed')),
effective_at TEXT NOT NULL,
amount TEXT NOT NULL,
amount_scale INTEGER NOT NULL,
currency TEXT NOT NULL,
payer_company_id INTEGER NOT NULL REFERENCES companies (id),
payee_company_id INTEGER NOT NULL REFERENCES companies (id),
perspective_company_id INTEGER REFERENCES companies (id),
subject_code TEXT CHECK (subject_code IN (
'receivable', 'payable', 'other_receivable', 'other_payable'
)),
source_kind TEXT NOT NULL CHECK (source_kind IN ('bank', 'manual', 'adjustment')),
source_revision_token TEXT,
posting_kind TEXT NOT NULL CHECK (posting_kind IN (
'normal', 'reversal', 'adjustment'
)),
reverses_ledger_event_id INTEGER REFERENCES ledger_events (id),
adjusts_ledger_event_id INTEGER REFERENCES ledger_events (id),
rule_version TEXT,
evidence_json TEXT,
idempotency_key TEXT,
actor_user_id INTEGER REFERENCES users (id),
actor_username TEXT,
reason TEXT,
supersedes_revision_id INTEGER REFERENCES ledger_event_revisions (id),
created_at TEXT NOT NULL,
UNIQUE (ledger_event_id, revision),
CHECK (payer_company_id != payee_company_id),
CHECK (state = 'confirmed' OR subject_code IS NULL),
CHECK (state = 'confirmed' OR perspective_company_id IS NULL),
CHECK (
state != 'confirmed'
OR (perspective_company_id IS NOT NULL AND subject_code IS NOT NULL)
)
);
CREATE TABLE current_ledger_event_revisions (
ledger_event_id INTEGER PRIMARY KEY REFERENCES ledger_events (id),
revision_id INTEGER NOT NULL UNIQUE REFERENCES ledger_event_revisions (id)
);
CREATE TABLE ledger_event_bank_sources (
bank_event_id INTEGER PRIMARY KEY REFERENCES canonical_transfer_events (id),
ledger_event_id INTEGER NOT NULL REFERENCES ledger_events (id),
UNIQUE (ledger_event_id, bank_event_id)
);
CREATE TABLE ledger_event_manual_sources (
manual_record_id INTEGER PRIMARY KEY REFERENCES manual_records (id),
ledger_event_id INTEGER NOT NULL REFERENCES ledger_events (id),
UNIQUE (ledger_event_id, manual_record_id)
);
CREATE TABLE ledger_subject_suggestions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ledger_event_id INTEGER NOT NULL REFERENCES ledger_events (id),
source_revision_id INTEGER NOT NULL REFERENCES ledger_event_revisions (id),
suggested_perspective_company_id INTEGER NOT NULL REFERENCES companies (id),
suggested_subject_code TEXT NOT NULL CHECK (suggested_subject_code IN (
'receivable', 'payable', 'other_receivable', 'other_payable'
)),
rule_version TEXT,
evidence_json TEXT,
created_at TEXT NOT NULL
);
CREATE INDEX idx_ledger_revisions_event ON ledger_event_revisions (ledger_event_id, revision);
CREATE INDEX idx_ledger_revisions_effective ON ledger_event_revisions (effective_at);
CREATE INDEX idx_ledger_revisions_pair_currency
ON ledger_event_revisions (payer_company_id, payee_company_id, currency);
CREATE INDEX idx_ledger_revisions_state
ON ledger_event_revisions (state, effective_at);
CREATE INDEX idx_manual_records_company ON manual_records (company_id, occurred_at);
CREATE INDEX idx_manual_decisions_record ON manual_record_decisions (record_id, revision);
CREATE INDEX idx_manual_decisions_state ON manual_record_decisions (state);
CREATE INDEX idx_subject_suggestions_event ON ledger_subject_suggestions (ledger_event_id);
CREATE VIEW eligible_position_events AS
SELECT le.id AS ledger_event_id,
cur.revision_id AS ledger_revision_id,
r.effective_at AS effective_at,
r.amount AS amount, r.amount_scale AS amount_scale,
r.currency AS currency,
r.payer_company_id AS payer_company_id,
r.payee_company_id AS payee_company_id,
r.perspective_company_id AS perspective_company_id,
r.subject_code AS subject_code,
r.source_kind AS source_kind, r.posting_kind AS posting_kind,
r.reverses_ledger_event_id AS reverses_ledger_event_id,
r.adjusts_ledger_event_id AS adjusts_ledger_event_id,
COALESCE(
(SELECT bs.bank_event_id FROM ledger_event_bank_sources bs
WHERE bs.ledger_event_id = le.id LIMIT 1),
(SELECT ms.manual_record_id FROM ledger_event_manual_sources ms
WHERE ms.ledger_event_id = le.id LIMIT 1)
) AS source_id,
((SELECT COUNT(*) FROM ledger_event_bank_sources bs
WHERE bs.ledger_event_id = le.id)
+ (SELECT COUNT(*) FROM ledger_event_manual_sources ms
WHERE ms.ledger_event_id = le.id)) AS evidence_count
FROM ledger_events le
JOIN current_ledger_event_revisions cur ON cur.ledger_event_id = le.id
JOIN ledger_event_revisions r ON r.id = cur.revision_id
WHERE le.lifecycle = 'active' AND r.state = 'confirmed';
CREATE TRIGGER ledger_events_no_delete BEFORE DELETE ON ledger_events
BEGIN SELECT RAISE (ABORT, 'ledger_events rows are immutable'); END;
CREATE TRIGGER ledger_events_no_update BEFORE UPDATE ON ledger_events
BEGIN
SELECT RAISE (ABORT, 'ledger_events only allow lifecycle changes')
WHERE OLD.lifecycle = NEW.lifecycle
OR OLD.id IS NOT NEW.id
OR OLD.created_at IS NOT NEW.created_at;
END;
CREATE TRIGGER ledger_event_revisions_no_update BEFORE UPDATE ON ledger_event_revisions
BEGIN SELECT RAISE (ABORT, 'ledger_event_revisions rows are immutable'); END;
CREATE TRIGGER ledger_event_revisions_no_delete BEFORE DELETE ON ledger_event_revisions
BEGIN SELECT RAISE (ABORT, 'ledger_event_revisions rows are immutable'); END;
CREATE TRIGGER ledger_subject_suggestions_no_update BEFORE UPDATE ON ledger_subject_suggestions
BEGIN SELECT RAISE (ABORT, 'ledger_subject_suggestions rows are immutable'); END;
CREATE TRIGGER ledger_subject_suggestions_no_delete BEFORE DELETE ON ledger_subject_suggestions
BEGIN SELECT RAISE (ABORT, 'ledger_subject_suggestions rows are immutable'); END;
CREATE TRIGGER manual_records_no_update BEFORE UPDATE ON manual_records
BEGIN SELECT RAISE (ABORT, 'manual_records rows are immutable'); END;
CREATE TRIGGER manual_records_no_delete BEFORE DELETE ON manual_records
BEGIN SELECT RAISE (ABORT, 'manual_records rows are immutable'); END;
CREATE TRIGGER manual_record_decisions_no_update BEFORE UPDATE ON manual_record_decisions
BEGIN SELECT RAISE (ABORT, 'manual_record_decisions rows are immutable'); END;
CREATE TRIGGER manual_record_decisions_no_delete BEFORE DELETE ON manual_record_decisions
BEGIN SELECT RAISE (ABORT, 'manual_record_decisions rows are immutable'); END;
""",
down="""
DROP VIEW IF EXISTS eligible_position_events;
DROP TRIGGER IF EXISTS manual_record_decisions_no_delete;
DROP TRIGGER IF EXISTS manual_record_decisions_no_update;
DROP TRIGGER IF EXISTS manual_records_no_delete;
DROP TRIGGER IF EXISTS manual_records_no_update;
DROP TRIGGER IF EXISTS ledger_subject_suggestions_no_delete;
DROP TRIGGER IF EXISTS ledger_subject_suggestions_no_update;
DROP TRIGGER IF EXISTS ledger_event_revisions_no_delete;
DROP TRIGGER IF EXISTS ledger_event_revisions_no_update;
DROP TRIGGER IF EXISTS ledger_events_no_update;
DROP TRIGGER IF EXISTS ledger_events_no_delete;
DROP TABLE IF EXISTS ledger_subject_suggestions;
DROP TABLE IF EXISTS ledger_event_manual_sources;
DROP TABLE IF EXISTS ledger_event_bank_sources;
DROP TABLE IF EXISTS current_ledger_event_revisions;
DROP TABLE IF EXISTS ledger_event_revisions;
DROP TABLE IF EXISTS ledger_events;
DROP TABLE IF EXISTS current_manual_record_decisions;
DROP TABLE IF EXISTS manual_record_decisions;
DROP TABLE IF EXISTS manual_records;
""",
),
)
+2
View File
@@ -25,6 +25,7 @@ import sqlite3
import tempfile
from . import auth
from . import ledger_events
from . import matching
from .db import utc_now
from .models import SheetResult, StatementBatch
@@ -648,6 +649,7 @@ def review_sheets(
[item["id"] for item in confirmed_rows],
actor=actor,
)
ledger_events.reconcile_bank_events(connection, actor=actor)
if began:
connection.commit()
except Exception:
+684
View File
@@ -0,0 +1,684 @@
"""Canonical intercompany ledger events and the revision chain (B-44).
Bank source rows and approved manual records are immutable evidence. This
module turns them into one canonical ledger event each through ``reconcile_*``
functions and an append-only revision chain. Corrections are never in-place
edits: a reversal or adjustment is a new ledger event with its own effective
date, and the original event keeps its history so no earlier cutoff is
rewritten. Current revisions and source claims are rebuildable projections.
Only B-43 ``eligible_intercompany_events`` feeds bank facts here; same-company
transfers, external transactions, unresolved rows and unlocked single
observations never reach the confirmed balance.
"""
from __future__ import annotations
from decimal import Decimal, InvalidOperation
import json
import sqlite3
from .db import utc_now
from .subjects import MIRROR, SUBJECTS, mirror_subject
class LedgerConflictError(ValueError):
"""A revision/claim/idempotency conflict (mapped to HTTP 409)."""
class LedgerInputError(ValueError):
"""Invalid input for a ledger operation (mapped to HTTP 400/422)."""
SUBJECT_RULE_VERSION = "subject-suggest-draft-v1"
def amount_scale(amount: object) -> int:
"""Decimal places of a decimal-string amount, never negative."""
try:
exponent = Decimal(str(amount)).as_tuple().exponent
except InvalidOperation:
return 0
return max(0, -int(exponent))
def parse_amount(amount: object) -> Decimal:
"""Parse a positive, valid decimal-string amount."""
try:
value = Decimal(str(amount))
except InvalidOperation:
raise LedgerInputError("金额不是有效的十进制数。") from None
if not value.is_finite() or value <= 0:
raise LedgerInputError("金额必须大于零。")
return value
def _company_exists(connection: sqlite3.Connection, company_id: int, label: str) -> None:
row = connection.execute(
"SELECT id FROM companies WHERE id = ?", (company_id,)
).fetchone()
if row is None:
raise LedgerInputError(f"{label}指向的公司不存在。")
# ---------------------------------------------------------------------------
# Revision helpers
# ---------------------------------------------------------------------------
def _ensure_transaction(connection: sqlite3.Connection) -> bool:
"""Begin an immediate transaction unless one is already open.
Write helpers may run standalone (they own the transaction) or nested
inside a caller's transaction (e.g. the sheet-confirm flow); nested calls
never start their own commit.
"""
began = False
if not connection.in_transaction:
connection.execute("BEGIN IMMEDIATE")
began = True
return began
def current_revision(connection: sqlite3.Connection, ledger_event_id: int) -> sqlite3.Row | None:
return connection.execute(
"""
SELECT r.* FROM current_ledger_event_revisions c
JOIN ledger_event_revisions r ON r.id = c.revision_id
WHERE c.ledger_event_id = ?
""",
(ledger_event_id,),
).fetchone()
def _event_lifecycle(connection: sqlite3.Connection, ledger_event_id: int) -> str | None:
row = connection.execute(
"SELECT lifecycle FROM ledger_events WHERE id = ?", (ledger_event_id,)
).fetchone()
return row["lifecycle"] if row is not None else None
def _next_revision_number(connection: sqlite3.Connection, ledger_event_id: int) -> int:
row = connection.execute(
"SELECT COALESCE(MAX(revision), 0) AS m FROM ledger_event_revisions WHERE ledger_event_id = ?",
(ledger_event_id,),
).fetchone()
return int(row["m"]) + 1
def create_event(
connection: sqlite3.Connection,
*,
state: str,
effective_at: str,
amount: str,
currency: str,
payer_company_id: int,
payee_company_id: int,
perspective_company_id: int | None,
subject_code: str | None,
source_kind: str,
source_revision_token: str | None,
posting_kind: str,
reverses_ledger_event_id: int | None = None,
adjusts_ledger_event_id: int | None = None,
rule_version: str | None = None,
evidence_json: str | None = None,
idempotency_key: str | None = None,
actor: sqlite3.Row | None = None,
reason: str | None = None,
supersedes_revision_id: int | None = None,
) -> tuple[int, int]:
"""Insert a new ledger event with one revision. Returns ``(event_id, revision_id)``."""
if state == "confirmed":
if perspective_company_id is None or subject_code is None:
raise LedgerInputError("已确认事件必须提供视角公司与科目。")
if subject_code not in SUBJECTS:
raise LedgerInputError("科目必须是应收/应付/其他应收/其他应付之一。")
if int(payer_company_id) == int(payee_company_id):
raise LedgerInputError("付款公司与收款公司不能相同。")
if perspective_company_id is not None and perspective_company_id not in (
int(payer_company_id), int(payee_company_id),
):
raise LedgerInputError("视角公司必须是事件参与方。")
began = _ensure_transaction(connection)
try:
now = utc_now()
cursor = connection.execute(
"INSERT INTO ledger_events (lifecycle, created_at) VALUES ('active', ?)",
(now,),
)
event_id = int(cursor.lastrowid)
revision_id = append_revision(
connection,
event_id,
state=state,
effective_at=effective_at,
amount=amount,
currency=currency,
payer_company_id=payer_company_id,
payee_company_id=payee_company_id,
perspective_company_id=perspective_company_id,
subject_code=subject_code,
source_kind=source_kind,
source_revision_token=source_revision_token,
posting_kind=posting_kind,
reverses_ledger_event_id=reverses_ledger_event_id,
adjusts_ledger_event_id=adjusts_ledger_event_id,
rule_version=rule_version,
evidence_json=evidence_json,
idempotency_key=idempotency_key,
actor=actor,
reason=reason,
supersedes_revision_id=supersedes_revision_id,
)
except Exception:
if began:
connection.rollback()
raise
else:
if began:
connection.commit()
return event_id, revision_id
def append_revision(
connection: sqlite3.Connection,
ledger_event_id: int,
*,
state: str,
effective_at: str,
amount: str,
currency: str,
payer_company_id: int,
payee_company_id: int,
perspective_company_id: int | None,
subject_code: str | None,
source_kind: str,
source_revision_token: str | None,
posting_kind: str,
reverses_ledger_event_id: int | None = None,
adjusts_ledger_event_id: int | None = None,
rule_version: str | None = None,
evidence_json: str | None = None,
idempotency_key: str | None = None,
actor: sqlite3.Row | None = None,
reason: str | None = None,
supersedes_revision_id: int | None = None,
) -> int:
"""Append one immutable revision and repoint the current projection."""
if _event_lifecycle(connection, ledger_event_id) != "active":
raise LedgerConflictError("该事件已停用,不能追加修订。")
if state == "confirmed":
if perspective_company_id is None or subject_code is None:
raise LedgerInputError("已确认事件必须提供视角公司与科目。")
if subject_code not in SUBJECTS:
raise LedgerInputError("科目必须是应收/应付/其他应收/其他应付之一。")
if perspective_company_id not in (int(payer_company_id), int(payee_company_id)):
raise LedgerInputError("视角公司必须是事件参与方。")
elif state != "pending_subject":
raise LedgerInputError("事件状态必须是 pending_subject 或 confirmed。")
revision = _next_revision_number(connection, ledger_event_id)
now = utc_now()
cursor = connection.execute(
"""
INSERT INTO ledger_event_revisions (
ledger_event_id, revision, state, effective_at, amount,
amount_scale, currency, payer_company_id, payee_company_id,
perspective_company_id, subject_code, source_kind,
source_revision_token, posting_kind, reverses_ledger_event_id,
adjusts_ledger_event_id, rule_version, evidence_json,
idempotency_key, actor_user_id, actor_username, reason,
supersedes_revision_id, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
ledger_event_id, revision, state, effective_at, amount,
amount_scale(amount), currency, payer_company_id, payee_company_id,
perspective_company_id, subject_code, source_kind,
source_revision_token, posting_kind, reverses_ledger_event_id,
adjusts_ledger_event_id, rule_version, evidence_json,
idempotency_key,
actor["id"] if actor is not None else None,
actor["username"] if actor is not None else None,
reason, supersedes_revision_id, now,
),
)
connection.execute(
"""
INSERT OR REPLACE INTO current_ledger_event_revisions (ledger_event_id, revision_id)
VALUES (?, ?)
""",
(ledger_event_id, cursor.lastrowid),
)
return int(cursor.lastrowid)
# ---------------------------------------------------------------------------
# Source claims
# ---------------------------------------------------------------------------
def bank_source_claim(connection: sqlite3.Connection, bank_event_id: int) -> sqlite3.Row | None:
return connection.execute(
"SELECT * FROM ledger_event_bank_sources WHERE bank_event_id = ?",
(bank_event_id,),
).fetchone()
def manual_source_claim(connection: sqlite3.Connection, manual_record_id: int) -> sqlite3.Row | None:
return connection.execute(
"SELECT * FROM ledger_event_manual_sources WHERE manual_record_id = ?",
(manual_record_id,),
).fetchone()
def _facts_of(connection: sqlite3.Connection, ledger_event_id: int) -> dict[str, object]:
revision = current_revision(connection, ledger_event_id)
if revision is None:
return {}
return {
"amount": revision["amount"],
"currency": revision["currency"],
"effective_at": revision["effective_at"],
"payer_company_id": revision["payer_company_id"],
"payee_company_id": revision["payee_company_id"],
}
def _eligible_facts(event: sqlite3.Row) -> dict[str, object]:
return {
"amount": event["amount"],
"currency": event["currency"],
"effective_at": event["effective_at"],
"payer_company_id": event["payer_company_id"],
"payee_company_id": event["payee_company_id"],
}
def _has_reversal(connection: sqlite3.Connection, original_event_id: int) -> bool:
row = connection.execute(
"""
SELECT 1 FROM ledger_event_revisions r
JOIN current_ledger_event_revisions c ON c.revision_id = r.id
WHERE r.reverses_ledger_event_id = ? AND r.posting_kind = 'reversal'
LIMIT 1
""",
(original_event_id,),
).fetchone()
return row is not None
def create_reversal(
connection: sqlite3.Connection,
original_event_id: int,
*,
source_kind: str,
source_revision_token: str | None = None,
effective_at: str | None = None,
reason: str,
actor: sqlite3.Row | None,
idempotency_key: str | None = None,
rule_version: str | None = None,
) -> tuple[int, int]:
"""Create an equal-amount, opposite-direction reversal as a new ledger event.
The subject mirrors the original (应收<->应付, 其他应收<->其他应付). ``effective_at``
defaults to the original event's effective date so an earlier cutoff keeps
the original impact and later cutoffs see the net zero. The original event
is never modified or deleted.
"""
original = current_revision(connection, original_event_id)
if original is None:
raise LedgerConflictError("原事件不存在或没有当前修订。")
if original["state"] != "confirmed":
raise LedgerInputError("只有已确认事件才能生成冲销。")
perspective = mirror_perspective(original)
if effective_at is None:
effective_at = original["effective_at"]
return create_event(
connection,
state="confirmed",
effective_at=effective_at,
amount=original["amount"],
currency=original["currency"],
payer_company_id=original["payee_company_id"],
payee_company_id=original["payer_company_id"],
perspective_company_id=perspective,
subject_code=mirror_subject(original["subject_code"]),
source_kind=source_kind,
source_revision_token=source_revision_token,
posting_kind="reversal",
reverses_ledger_event_id=original_event_id,
rule_version=rule_version or original["rule_version"],
idempotency_key=idempotency_key,
actor=actor,
reason=reason,
)
def mirror_perspective(revision: sqlite3.Row) -> int:
"""The counterparty company from ``revision``'s perspective."""
perspective = int(revision["perspective_company_id"])
if perspective == int(revision["payer_company_id"]):
return int(revision["payee_company_id"])
return int(revision["payer_company_id"])
def create_adjustment(
connection: sqlite3.Connection,
ledger_event_id: int,
*,
effective_at: str,
amount: str,
currency: str,
payer_company_id: int,
payee_company_id: int,
perspective_company_id: int,
subject_code: str,
reason: str,
actor: sqlite3.Row,
idempotency_key: str | None = None,
rule_version: str | None = None,
) -> tuple[int, int]:
"""Create an audit adjustment event; the original event stays unchanged."""
return create_event(
connection,
state="confirmed",
effective_at=effective_at,
amount=str(parse_amount(amount)),
currency=currency,
payer_company_id=payer_company_id,
payee_company_id=payee_company_id,
perspective_company_id=perspective_company_id,
subject_code=subject_code,
source_kind="adjustment",
source_revision_token=None,
posting_kind="adjustment",
adjusts_ledger_event_id=ledger_event_id,
rule_version=rule_version or SUBJECT_RULE_VERSION,
idempotency_key=idempotency_key,
actor=actor,
reason=reason,
)
def reopen_subject(
connection: sqlite3.Connection,
ledger_event_id: int,
*,
reason: str,
actor: sqlite3.Row,
idempotency_key: str | None = None,
) -> tuple[int, int]:
"""Reverse a confirmed event and re-open it for subject re-review.
Creates an equal-amount reversal plus a fresh ``pending_subject`` event
that re-claims the original bank source, so the administrator can confirm
a corrected subject. The original event and its reversal keep history.
"""
current = current_revision(connection, ledger_event_id)
if current is None or current["state"] != "confirmed":
raise LedgerConflictError("只有已确认事件可以重新进入科目审核。")
bank_claim = connection.execute(
"SELECT * FROM ledger_event_bank_sources WHERE ledger_event_id = ?",
(ledger_event_id,),
).fetchone()
if bank_claim is None:
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,
)
return event_id, revision_id
# ---------------------------------------------------------------------------
# Bank event reconciliation
# ---------------------------------------------------------------------------
def reconcile_bank_events(
connection: sqlite3.Connection, actor: sqlite3.Row | None = None
) -> dict[str, object]:
"""Reconcile the current eligible intercompany events into ledger events.
Idempotent: first sight creates a ``pending_subject`` event; a changed B-43
decision updates a still-pending event's revision, or (for a confirmed
event) creates a reversal plus a fresh pending event. A source that left
the eligible set with a confirmed impact gets one reversal. Runs inside the
caller's transaction when one is open, otherwise in its own transaction.
"""
began = _ensure_transaction(connection)
try:
eligible = {
row["event_id"]: row
for row in connection.execute(
"SELECT * FROM eligible_intercompany_events"
).fetchall()
}
claims = {
row["bank_event_id"]: row
for row in connection.execute(
"SELECT * FROM ledger_event_bank_sources"
).fetchall()
}
stats = {
"created": 0, "updated_pending": 0, "reversal": 0,
"reopened": 0, "unchanged": 0, "sources": len(eligible),
}
for bank_event_id, event in sorted(eligible.items()):
claim = claims.get(bank_event_id)
if claim is None:
_create_bank_event(
connection, event, actor, reason="B-43 事件首次入账,待确认科目"
)
stats["created"] += 1
continue
current = current_revision(connection, claim["ledger_event_id"])
if current is None or _facts_of(connection, claim["ledger_event_id"]) != _eligible_facts(event):
if current is not None and current["state"] == "confirmed":
if not _has_reversal(connection, claim["ledger_event_id"]):
create_reversal(
connection, claim["ledger_event_id"],
source_kind="bank",
source_revision_token=event["decision_id"],
reason="B-43 事件事实变更,原确认事件冲销",
actor=actor,
)
stats["reversal"] += 1
_create_bank_event(
connection, event, actor,
reason="B-43 事件事实变更后重新入账,待确认科目",
replacing_claim=claim,
)
stats["reopened"] += 1
elif current is None or current["state"] == "pending_subject":
_append_bank_pending_revision(
connection, claim["ledger_event_id"], event, actor
)
stats["updated_pending"] += 1
else:
stats["unchanged"] += 1
else:
stats["unchanged"] += 1
for bank_event_id, claim in sorted(claims.items()):
if bank_event_id in eligible:
continue
current = current_revision(connection, claim["ledger_event_id"])
if current is not None and current["state"] == "confirmed":
if not _has_reversal(connection, claim["ledger_event_id"]):
create_reversal(
connection, claim["ledger_event_id"],
source_kind="bank",
source_revision_token=None,
reason="B-43 事件不再纳入往来,原确认事件冲销",
actor=actor,
)
stats["reversal"] += 1
except Exception:
if began:
connection.rollback()
raise
else:
if began:
connection.commit()
return stats
def _create_bank_event(
connection: sqlite3.Connection,
event: sqlite3.Row,
actor: sqlite3.Row | None,
*,
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"]),
)
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)
return event_id, revision_id
def _append_bank_pending_revision(
connection: sqlite3.Connection,
ledger_event_id: int,
event: sqlite3.Row,
actor: sqlite3.Row | None,
) -> int:
current = current_revision(connection, ledger_event_id)
revision_id = append_revision(
connection,
ledger_event_id,
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="B-43 事件事实更新,追加待审修订",
supersedes_revision_id=current["id"] if current is not None else None,
)
from .subjects import store_suggestions
store_suggestions(connection, ledger_event_id)
return revision_id
# ---------------------------------------------------------------------------
# Projection rebuild
# ---------------------------------------------------------------------------
def rebuild_current_ledger_projection(connection: sqlite3.Connection) -> int:
"""Rebuild current ledger revisions from the append-only log."""
began = _ensure_transaction(connection)
try:
connection.execute("DELETE FROM current_ledger_event_revisions")
rows = connection.execute(
"""
SELECT e.id AS ledger_event_id,
(SELECT r2.id FROM ledger_event_revisions r2
WHERE r2.ledger_event_id = e.id
ORDER BY r2.revision DESC LIMIT 1) AS latest_id
FROM ledger_events e
WHERE e.lifecycle = 'active'
""",
).fetchall()
rebuilt = 0
for row in rows:
if row["latest_id"] is None:
continue
connection.execute(
"""
INSERT OR REPLACE INTO current_ledger_event_revisions (ledger_event_id, revision_id)
VALUES (?, ?)
""",
(row["ledger_event_id"], row["latest_id"]),
)
rebuilt += 1
except Exception:
if began:
connection.rollback()
raise
else:
if began:
connection.commit()
return rebuilt
+818
View File
@@ -0,0 +1,818 @@
"""Manual evidence records, administrator approval and audit-safe reversal.
Manual records are immutable submitted facts. Only an approved record becomes
a canonical ledger event (``approve_new``) or joins one (``approve_link``);
returned/exception/pending records never affect a balance and never leak to
the counterparty. Approved facts change only through a ``reverse`` decision
that creates an opposite new event (or detaches a linked claim) — the original
is never edited. Idempotency keys and UNIQUE claims prevent double counting.
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation
import json
import sqlite3
from .db import utc_now
from .ledger_events import (
LedgerConflictError,
LedgerInputError,
create_event,
create_reversal,
current_revision,
manual_source_claim,
)
from .subjects import SUBJECTS
MANUAL_STATES = ("pending", "approved", "returned", "exception", "reversed")
FUNDING_SOURCES = ("approved_bank_account", "personal_transit", "other")
DATE_KEYS = ("occurred_at",)
class ManualConflictError(ValueError):
"""A claim/idempotency/revision conflict (mapped to HTTP 409)."""
class ManualInputError(ValueError):
"""Invalid manual record input (mapped to HTTP 400/422)."""
def _parse_amount(amount: object) -> Decimal:
try:
value = Decimal(str(amount))
except InvalidOperation:
raise ManualInputError("金额不是有效的十进制数。") from None
if not value.is_finite() or value <= 0:
raise ManualInputError("金额必须大于零。")
return value
def _validate_date(value: object, field: str) -> str:
text = str(value or "").strip()
if len(text) < 10:
raise ManualInputError(f"{field}必须是 YYYY-MM-DD 或完整时间。")
try:
datetime.fromisoformat(text[:10])
except ValueError:
raise ManualInputError(f"{field}必须是 YYYY-MM-DD 或完整时间。") from None
return text
def _business_today() -> str:
"""Shanghai business date (the default reversal effective date)."""
return datetime.now(timezone(timedelta(hours=8))).date().isoformat()
# ---------------------------------------------------------------------------
# Submit
# ---------------------------------------------------------------------------
def submit(
connection: sqlite3.Connection,
*,
company_id: int,
counterparty_company_id: int,
occurred_at: str,
direction: str,
amount: str,
currency: str,
funding_source: str,
requested_subject: str,
request_key: str,
actor: sqlite3.Row,
bank_account_id: object = None,
personal_transit_mapping_id: object = None,
related_source_row_id: object = None,
summary: object = None,
reason: object = None,
evidence: object = None,
supersedes_record_id: object = None,
) -> dict[str, object]:
"""Submit one manual record for review. Idempotent on ``(company_id, request_key)``."""
request_key = str(request_key or "").strip()
if not request_key:
raise ManualInputError("必须提供提交幂等键 request_key。")
if int(company_id) == int(counterparty_company_id):
raise ManualInputError("对方公司不能与本公司相同。")
if direction not in ("outgoing", "incoming"):
raise ManualInputError("方向必须是 outgoing 或 incoming。")
if funding_source not in FUNDING_SOURCES:
raise ManualInputError(f"资金来源必须是:{''.join(FUNDING_SOURCES)}")
if requested_subject not in SUBJECTS:
raise ManualInputError("科目必须是应收/应付/其他应收/其他应付之一。")
_validate_date(occurred_at, "业务日期")
currency = str(currency or "").strip()
if not currency:
raise ManualInputError("币种不能为空。")
amount = str(_parse_amount(amount))
for label, raw in (
("company_id", company_id), ("counterparty_company_id", counterparty_company_id),
):
row = connection.execute("SELECT id FROM companies WHERE id = ?", (int(raw),)).fetchone()
if row is None:
raise ManualInputError(f"{label} 指向的公司不存在。")
bank_account_id = _resolve_account_ref(
connection, bank_account_id, company_id, "银行账户"
)
mapping_id = _resolve_account_ref(
connection, personal_transit_mapping_id, company_id, "个人过账映射"
)
related_row = None
if related_source_row_id not in (None, ""):
related_row = connection.execute(
"SELECT r.id, b.company_id FROM source_rows r "
"JOIN sheet_batches s ON s.id = r.sheet_batch_id "
"JOIN import_batches b ON b.id = s.import_batch_id "
"WHERE r.id = ?",
(int(related_source_row_id),),
).fetchone()
if related_row is None:
raise ManualInputError("关联银行源行不存在。")
if funding_source == "approved_bank_account" and bank_account_id is None:
raise ManualInputError("资金来源为已批准账户时必须指定银行账户。")
if funding_source == "personal_transit" and mapping_id is None:
raise ManualInputError("资金来源为个人过账时必须指定个人过账映射。")
supersedes_id = None
if supersedes_record_id not in (None, ""):
parent = connection.execute(
"SELECT id, company_id FROM manual_records WHERE id = ?",
(int(supersedes_record_id),),
).fetchone()
if parent is None or parent["company_id"] != int(company_id):
raise ManualInputError("supersedes_record_id 无效。")
supersedes_id = int(supersedes_record_id)
now = utc_now()
began = False
if not connection.in_transaction:
connection.execute("BEGIN IMMEDIATE")
began = True
try:
# 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",
reason=str(reason or "") or None, actor=actor,
)
except Exception:
if began:
connection.rollback()
raise
else:
if began:
connection.commit()
return _record_payload(connection, record_id)
def _resolve_account_ref(connection, raw, company_id: int, label: str) -> int | None:
if raw in (None, ""):
return None
row = connection.execute(
"SELECT id, company_id FROM bank_accounts WHERE id = ?", (int(raw),)
).fetchone()
if row is None:
raise ManualInputError(f"{label}不存在。")
if row["company_id"] != int(company_id):
raise ManualInputError(f"{label}必须属于提交公司。")
return int(raw)
def _scale_of(amount: str) -> int:
exponent = Decimal(amount).as_tuple().exponent
return max(0, -int(exponent))
# ---------------------------------------------------------------------------
# Decisions
# ---------------------------------------------------------------------------
def _append_decision(
connection: sqlite3.Connection,
record_id: int,
*,
state: str,
action: str,
reason: str | None,
actor: sqlite3.Row,
idempotency_key: str | None = None,
supersedes_decision_id: int | None = None,
) -> int:
row = connection.execute(
"SELECT COALESCE(MAX(revision), 0) AS m FROM manual_record_decisions WHERE record_id = ?",
(record_id,),
).fetchone()
revision = int(row["m"]) + 1
now = utc_now()
cursor = connection.execute(
"""
INSERT INTO manual_record_decisions (
record_id, revision, state, action, reason, actor_user_id,
actor_username, idempotency_key, supersedes_decision_id, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
record_id, revision, state, action, reason,
actor["id"], actor["username"], idempotency_key,
supersedes_decision_id, now,
),
)
decision_id = int(cursor.lastrowid)
connection.execute(
"""
INSERT OR REPLACE INTO current_manual_record_decisions (record_id, decision_id)
VALUES (?, ?)
""",
(record_id, decision_id),
)
return decision_id
def _current_decision(connection: sqlite3.Connection, record_id: int) -> sqlite3.Row | None:
return connection.execute(
"""
SELECT d.* FROM current_manual_record_decisions c
JOIN manual_record_decisions d ON d.id = c.decision_id
WHERE c.record_id = ?
""",
(record_id,),
).fetchone()
def decide(
connection: sqlite3.Connection,
record_id: int,
action: str,
*,
reason: str,
expected_decision_id: int | None,
request_key: str | None,
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) 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:
raise ManualInputError("必须填写审核原因。")
if action not in ("approve_new", "approve_link", "return", "exception", "reverse"):
raise ManualInputError("未知的审核决定类型。")
began = False
if not connection.in_transaction:
connection.execute("BEGIN IMMEDIATE")
began = True
try:
record = connection.execute(
"SELECT * FROM manual_records WHERE id = ?", (record_id,)
).fetchone()
if record is None:
raise ManualConflictError("手工记录不存在。")
if request_key:
existing = connection.execute(
"SELECT * FROM manual_record_decisions WHERE record_id = ? AND idempotency_key = ?",
(record_id, request_key),
).fetchone()
if existing is not None:
if began:
connection.commit()
return _decision_payload(connection, record_id, existing["id"])
current = _current_decision(connection, record_id)
if current is None:
raise ManualConflictError("该记录没有当前状态。")
if expected_decision_id is not None and int(expected_decision_id) != current["id"]:
raise ManualConflictError("记录已发生变更,请刷新后重试。")
if action == "approve_new":
outcome = _approve_new(
connection, record, current, actor, subject_code, reason, request_key
)
elif action == "approve_link":
outcome = _approve_link(
connection, record, current, actor, target_ledger_event_id, reason,
request_key,
)
elif action == "return":
if current["state"] != "pending":
raise ManualConflictError("只有待复核的记录可以退回。")
decision_id = _append_decision(
connection, record_id, state="returned", action=action,
reason=reason, actor=actor, idempotency_key=request_key,
supersedes_decision_id=current["id"],
)
outcome = {"decision_id": decision_id, "ledger_event_id": None}
elif action == "exception":
if current["state"] != "pending":
raise ManualConflictError("只有待复核的记录可以转为异常。")
decision_id = _append_decision(
connection, record_id, state="exception", action=action,
reason=reason, actor=actor, idempotency_key=request_key,
supersedes_decision_id=current["id"],
)
outcome = {"decision_id": decision_id, "ledger_event_id": None}
else: # reverse
if current["state"] != "approved":
raise ManualConflictError("只有已批准记录可以冲销。")
outcome = _reverse(
connection, record, current, actor, request_key, reason,
effective_at=effective_at,
)
_store_audit(connection, record, current, action, outcome, reason, actor)
except Exception:
if began:
connection.rollback()
raise
else:
if began:
connection.commit()
return _decision_payload(connection, record_id, outcome["decision_id"])
def _approve_new(
connection: sqlite3.Connection,
record: sqlite3.Row,
current: sqlite3.Row,
actor: sqlite3.Row,
subject_code: object,
reason: str,
idempotency_key: str | None,
) -> dict[str, object]:
subject = str(subject_code or record["requested_subject"] or "")
if subject not in SUBJECTS:
raise ManualInputError("科目必须是应收/应付/其他应收/其他应付之一。")
if record["direction"] == "outgoing":
payer, payee = record["company_id"], record["counterparty_company_id"]
else:
payer, payee = record["counterparty_company_id"], record["company_id"]
event_id, _revision_id = create_event(
connection,
state="confirmed",
effective_at=record["occurred_at"],
amount=record["amount"],
currency=record["currency"],
payer_company_id=payer,
payee_company_id=payee,
perspective_company_id=record["company_id"],
subject_code=subject,
source_kind="manual",
source_revision_token=None,
posting_kind="normal",
rule_version="manual-record-v1",
evidence_json=json.dumps({"manual_record_id": record["id"]}, ensure_ascii=False),
actor=actor,
reason=reason,
)
decision_id = _append_decision(
connection, record["id"], state="approved", action="approve_new",
reason=reason, actor=actor,
idempotency_key=idempotency_key,
supersedes_decision_id=current["id"],
)
connection.execute(
"""
INSERT INTO ledger_event_manual_sources (manual_record_id, ledger_event_id)
VALUES (?, ?)
""",
(record["id"], event_id),
)
return {"decision_id": decision_id, "ledger_event_id": event_id}
def _approve_link(
connection: sqlite3.Connection,
record: sqlite3.Row,
current: sqlite3.Row,
actor: sqlite3.Row,
target_ledger_event_id: object,
reason: str,
idempotency_key: str | None,
) -> dict[str, object]:
if target_ledger_event_id in (None, ""):
raise ManualInputError("approve_link 必须指定目标往来事件。")
target = connection.execute(
"SELECT id, lifecycle FROM ledger_events WHERE id = ?",
(int(target_ledger_event_id),),
).fetchone()
if target is None or target["lifecycle"] != "active":
raise ManualConflictError("目标往来事件不存在。")
if manual_source_claim(connection, record["id"]) is not None:
raise ManualConflictError("该手工记录已关联往来事件。")
decision_id = _append_decision(
connection, record["id"], state="approved", action="approve_link",
reason=reason, actor=actor,
idempotency_key=idempotency_key,
supersedes_decision_id=current["id"],
)
connection.execute(
"""
INSERT INTO ledger_event_manual_sources (manual_record_id, ledger_event_id)
VALUES (?, ?)
""",
(record["id"], int(target_ledger_event_id)),
)
return {"decision_id": decision_id, "ledger_event_id": int(target_ledger_event_id)}
def _reverse(
connection: sqlite3.Connection,
record: sqlite3.Row,
current: sqlite3.Row,
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:
raise ManualConflictError("该记录尚未关联往来事件,无法冲销。")
event_id = claim["ledger_event_id"]
revision = current_revision(connection, event_id)
if revision is None:
raise ManualConflictError("关联的往来事件没有当前修订。")
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 (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"],),
)
decision_id = _append_decision(
connection, record["id"], state="reversed", action="reverse",
reason=reason, actor=actor, idempotency_key=request_key,
supersedes_decision_id=current["id"],
)
return {"decision_id": decision_id, "ledger_event_id": 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(
connection, record, current, action, outcome, reason, actor
) -> None:
from .auth import audit
audit(
connection,
f"manual_{action}",
actor=actor,
target=f"manual_record:{record['id']}",
detail=(
f"decision:{outcome['decision_id']};"
f"ledger_event:{outcome.get('ledger_event_id')};reason:{reason}"
),
)
# ---------------------------------------------------------------------------
# Candidates and queries
# ---------------------------------------------------------------------------
def find_candidates(connection: sqlite3.Connection, record_id: int) -> list[dict[str, object]]:
"""Deterministic hints shown before approval; never auto-merged."""
record = connection.execute(
"SELECT * FROM manual_records WHERE id = ?", (record_id,)
).fetchone()
if record is None:
return []
wanted_direction = "incoming" if record["direction"] == "outgoing" else "outgoing"
date_prefix = str(record["occurred_at"])[:10]
candidates: list[dict[str, object]] = []
bank_rows = connection.execute(
"""
SELECT e.event_id, e.amount, e.currency, e.effective_at, e.pairing,
e.payer_company_id, e.payee_company_id, e.decision_id
FROM eligible_intercompany_events e
WHERE (e.payer_company_id = ? AND e.payee_company_id = ?)
OR (e.payer_company_id = ? AND e.payee_company_id = ?)
ORDER BY e.event_id
""",
(
record["company_id"], record["counterparty_company_id"],
record["counterparty_company_id"], record["company_id"],
),
).fetchall()
for row in bank_rows:
if row["amount"] != record["amount"] or row["currency"] != record["currency"]:
continue
event_direction = (
"outgoing" if row["payer_company_id"] == record["company_id"] else "incoming"
)
if event_direction != wanted_direction:
continue
candidates.append(
{
"kind": "bank_event",
"ledger_event_id": _ledger_event_of_bank(connection, row["event_id"]),
"bank_event_id": row["event_id"],
"decision_id": row["decision_id"],
"amount": row["amount"],
"currency": row["currency"],
"effective_at": row["effective_at"],
"pairing": row["pairing"],
"hint": "已存在匹配的银行规范事件,建议关联",
}
)
manual_rows = connection.execute(
"""
SELECT m.id, m.company_id, m.counterparty_company_id, m.direction,
m.amount, m.currency, m.occurred_at, d.state
FROM manual_records m
JOIN current_manual_record_decisions c ON c.record_id = m.id
JOIN manual_record_decisions d ON d.id = c.decision_id
WHERE m.id != ? AND m.amount = ? AND m.currency = ?
AND (
(m.company_id = ? AND m.counterparty_company_id = ?)
OR (m.company_id = ? AND m.counterparty_company_id = ?)
)
ORDER BY m.id
""",
(
record["id"], record["amount"], record["currency"],
record["company_id"], record["counterparty_company_id"],
record["counterparty_company_id"], record["company_id"],
),
).fetchall()
for row in manual_rows:
if row["direction"] != wanted_direction:
continue
if row["state"] not in ("approved", "pending"):
continue
candidates.append(
{
"kind": "manual_record",
"ledger_event_id": None,
"manual_record_id": row["id"],
"amount": row["amount"],
"currency": row["currency"],
"occurred_at": row["occurred_at"],
"state": row["state"],
"hint": "存在方向相反的同额手工记录,建议核对后关联",
}
)
return candidates
def _ledger_event_of_bank(connection, bank_event_id: int) -> int | None:
claim = connection.execute(
"SELECT ledger_event_id FROM ledger_event_bank_sources WHERE bank_event_id = ?",
(bank_event_id,),
).fetchone()
return claim["ledger_event_id"] if claim is not None else None
def list_records(
connection: sqlite3.Connection,
*,
company_id: int | None = None,
state: str | None = None,
limit: int = 100,
) -> list[sqlite3.Row]:
conditions: list[str] = []
params: list[object] = []
if company_id is not None:
conditions.append("(m.company_id = ? OR m.counterparty_company_id = ?)")
params.extend([company_id, company_id])
if state is not None:
if state not in MANUAL_STATES:
raise ManualInputError("无效的记录状态。")
conditions.append("d.state = ?")
params.append(state)
where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
return connection.execute(
f"""
SELECT m.*, d.id AS decision_id, d.state AS state, d.revision AS decision_revision,
d.action AS action, d.reason AS decision_reason,
d.actor_username AS decision_actor, d.created_at AS decision_at,
c.name AS company_name, cc.name AS counterparty_company_name,
u.username AS submitted_by_username
FROM manual_records m
JOIN current_manual_record_decisions c ON c.record_id = m.id
JOIN manual_record_decisions d ON d.id = c.decision_id
LEFT JOIN companies c ON c.id = m.company_id
LEFT JOIN companies cc ON cc.id = m.counterparty_company_id
LEFT JOIN users u ON u.id = m.submitted_by
{where}
ORDER BY m.id DESC
LIMIT ?
""",
(*params, max(1, int(limit))),
).fetchall()
def rebuild_current_manual_projection(connection: sqlite3.Connection) -> int:
began = False
if not connection.in_transaction:
connection.execute("BEGIN IMMEDIATE")
began = True
try:
connection.execute("DELETE FROM current_manual_record_decisions")
rows = connection.execute(
"""
SELECT m.id AS record_id,
(SELECT d2.id FROM manual_record_decisions d2
WHERE d2.record_id = m.id
ORDER BY d2.revision DESC LIMIT 1) AS latest_id
FROM manual_records m
"""
).fetchall()
rebuilt = 0
for row in rows:
if row["latest_id"] is None:
continue
connection.execute(
"""
INSERT OR REPLACE INTO current_manual_record_decisions (record_id, decision_id)
VALUES (?, ?)
""",
(row["record_id"], row["latest_id"]),
)
rebuilt += 1
except Exception:
if began:
connection.rollback()
raise
else:
if began:
connection.commit()
return rebuilt
# ---------------------------------------------------------------------------
# Payloads
# ---------------------------------------------------------------------------
def _record_payload(connection: sqlite3.Connection, record_id: int, *, idempotent_replay: bool = False) -> dict[str, object]:
rows = list_records(connection, limit=1000)
row = next((item for item in rows if item["id"] == record_id), None)
if row is None:
raise ManualInputError("手工记录不存在。")
payload = _row_payload(connection, row)
if idempotent_replay:
payload["idempotent_replay"] = True
return payload
def _row_payload(connection: sqlite3.Connection, row: sqlite3.Row) -> dict[str, object]:
evidence = json.loads(row["evidence_json"] or "{}") if row["evidence_json"] else {}
return {
"id": row["id"],
"company_id": row["company_id"],
"company_name": row["company_name"],
"counterparty_company_id": row["counterparty_company_id"],
"counterparty_company_name": row["counterparty_company_name"],
"occurred_at": row["occurred_at"],
"direction": row["direction"],
"amount": row["amount"],
"currency": row["currency"],
"funding_source": row["funding_source"],
"bank_account_id": row["bank_account_id"],
"personal_transit_mapping_id": row["personal_transit_mapping_id"],
"related_source_row_id": row["related_source_row_id"],
"requested_subject": row["requested_subject"],
"summary": row["summary"],
"reason": row["reason"],
"request_key": row["request_key"],
"supersedes_record_id": row["supersedes_record_id"],
"submitted_by": row["submitted_by"],
"submitted_by_username": row["submitted_by_username"] if "submitted_by_username" in row.keys() else None,
"attachment_name": evidence.get("attachment_name"),
"created_at": row["created_at"],
"state": row["state"],
"decision_id": row["decision_id"],
"decision_revision": row["decision_revision"],
"decision_action": row["action"],
"decision_reason": row["decision_reason"],
"decision_actor": row["decision_actor"],
"decision_at": row["decision_at"],
"candidates": find_candidates(connection, row["id"]),
}
def _decision_payload(connection: sqlite3.Connection, record_id: int, decision_id: int) -> dict[str, object]:
row = connection.execute(
"""
SELECT d.* FROM manual_record_decisions d
WHERE d.id = ?
""",
(decision_id,),
).fetchone()
record = connection.execute(
"SELECT * FROM manual_records WHERE id = ?", (record_id,)
).fetchone()
claim = manual_source_claim(connection, record_id)
return {
"record_id": record_id,
"decision_id": decision_id,
"revision": row["revision"],
"state": row["state"],
"action": row["action"],
"reason": row["reason"],
"actor_username": row["actor_username"],
"created_at": row["created_at"],
"ledger_event_id": claim["ledger_event_id"] if claim is not None else None,
"requested_subject": record["requested_subject"],
"amount": record["amount"],
"currency": record["currency"],
"occurred_at": record["occurred_at"],
}
File diff suppressed because it is too large Load Diff
+398
View File
@@ -0,0 +1,398 @@
"""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",
"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"],
}
+86
View File
@@ -0,0 +1,86 @@
"use strict";
const path = require("path");
const assert = require("assert");
const ui = require(path.join(__dirname, "..", "web", "app.js"));
assert.strictEqual(ui.fmtAbsMoney(-1280), "1,280.00");
assert.strictEqual(ui.fmtAbsMoney(1280), "1,280.00");
assert.strictEqual(ui.fmtAbsMoney("60.5"), "60.50");
assert.strictEqual(ui.eventIsNegative({ posting_kind: "reversal" }), true);
assert.strictEqual(ui.eventIsNegative({ posting_kind: "normal", is_repayment: true }), true);
assert.strictEqual(ui.eventIsNegative({ posting_kind: "normal", is_repayment: false }), false);
assert.strictEqual(ui.cycleTab(0, 4, false), 1);
assert.strictEqual(ui.cycleTab(3, 4, false), 0);
assert.strictEqual(ui.cycleTab(0, 4, true), 3);
assert.strictEqual(ui.cycleTab(2, 5, true), 1);
assert.strictEqual(ui.cycleTab(0, 0, false), 0);
assert.strictEqual(ui.drawerEscAction(1), "close");
assert.strictEqual(ui.drawerEscAction(0), "close");
assert.strictEqual(ui.drawerEscAction(2), "back");
assert.strictEqual(ui.drawerEscAction(3), "back");
const abs = ui.amountWithCurrency(1280, "CNY");
assert.ok(abs.includes("CNY"));
assert.ok(abs.includes("1,280.00"));
assert.ok(!abs.includes("+"));
assert.ok(!abs.includes(""));
const repay = ui.amountWithCurrency(3200000, "CNY", { signed: true, negative: true });
assert.ok(repay.includes(""));
assert.ok(!repay.includes("+"));
assert.ok(repay.includes("3,200,000.00"));
assert.strictEqual(ui.resultDirection(10).label, "应收");
assert.strictEqual(ui.resultDirection(-10).label, "应付");
assert.strictEqual(ui.resultDirection(0).label, "持平");
assert.strictEqual(ui.isCompactAmount(1280), false);
assert.strictEqual(ui.isCompactAmount(999999999), false);
assert.strictEqual(ui.isCompactAmount(1000000000), true);
assert.strictEqual(ui.isCompactAmount("123456789012345"), true);
assert.ok(ui.amountWithCurrency("123456789012345", "CNY").includes("is-compact"));
assert.ok(!ui.amountWithCurrency(1280, "CNY").includes("is-compact"));
assert.strictEqual(ui.cashDirectionLabel("outgoing"), "转出");
assert.strictEqual(ui.cashDirectionLabel("incoming"), "转入");
assert.strictEqual(ui.relatedFlowLabel(null), "无关联流水");
assert.strictEqual(ui.relatedFlowLabel(""), "无关联流水");
assert.strictEqual(ui.relatedFlowLabel("42"), "42");
const subjectFields = ui.auditEvidenceFields("subject-review", {
direction: "outgoing",
effectiveAt: "2026-07-18T09:00:00",
amount: "600",
currency: "CNY",
summary: "资金调拨",
});
assert.deepStrictEqual(subjectFields.map((item) => item[0]), ["方向", "日期", "金额", "摘要"]);
assert.strictEqual(subjectFields[0][1], "转出");
const manualFields = ui.auditEvidenceFields("manual-review", {
counterpartyName: "乙公司",
direction: "incoming",
relatedSourceRowId: "",
submittedBy: "出纳甲",
attachmentName: "",
amount: "80",
currency: "CNY",
summary: "补记",
});
assert.deepStrictEqual(
manualFields.map((item) => item[0]),
["对方公司名", "方向", "关联流水号", "提交人", "附件", "摘要", "金额"],
);
assert.strictEqual(manualFields[0][1], "乙公司");
assert.strictEqual(manualFields[1][1], "转入");
assert.strictEqual(manualFields[2][1], "无关联流水");
assert.ok(ui.accountCell({ visibility: "visible", label: "中信 5316" }).includes("中信 5316"));
assert.ok(ui.accountCell({ visibility: "masked" }).includes("按对方授权不可见"));
assert.ok(ui.accountCell({ visibility: "missing" }).includes("源行缺失"));
console.log("b44_ui_check ok");
+138
View File
@@ -0,0 +1,138 @@
<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>B-44 余额目录布局回归</title>
<link rel="stylesheet" href="../../web/styles.css" />
</head>
<body data-portal="admin">
<div class="app-shell">
<aside class="sidebar" id="sidebar" aria-label="总账管理导航">
<div class="brand"><span class="brand-mark"></span><span class="brand-copy"><strong>金牛集团</strong><small>总账管理端</small></span></div>
<nav class="nav-list">
<button class="nav-item is-active" type="button"><span>往来查询</span></button>
</nav>
</aside>
<div class="workspace">
<main id="main-content">
<section class="app-view is-active">
<header class="page-heading"><div><h1>往来查询</h1><p>公司间往来余额目录</p></div></header>
<section class="panel company-ledger-panel" aria-label="公司余额目录">
<div class="panel-heading"><div><h2>公司余额目录</h2><p>每行余额都附带截止日、期初状态、本期借贷、结果与未决金额</p></div></div>
<div class="ledger-head is-balances"><span>公司</span><span class="ledger-hide-md">借方合计</span><span class="ledger-hide-md">贷方合计</span><span>期末结果</span><span>未决</span><span>截止日</span><span></span></div>
<div class="company-ledgers">
<details class="company-ledger is-balances">
<summary>
<span class="company-name" title="甲公司"><i></i><b>甲公司</b><small class="currency-tag">CNY</small></span>
<strong class="amount debit ledger-hide-md"><span class="amount-with-currency"><span class="currency-code">CNY</span>1,280.00</span></strong>
<strong class="amount credit ledger-hide-md"><span class="amount-with-currency"><span class="currency-code">CNY</span>320.00</span></strong>
<span class="ledger-result"><em class="status success">应收</em><b><span class="amount-with-currency"><span class="currency-code">CNY</span>960.00</span></b></span>
<span class="ledger-unresolved is-empty"><span class="status neutral">未决 0.00</span></span>
<span class="ledger-cutoff">截止 2026.07.31</span>
<svg></svg>
</summary>
</details>
<details class="company-ledger is-balances" id="stressRow">
<summary>
<span class="company-name" title="东南沿海综合贸易与供应链管理股份有限公司"><i></i><b>东南沿海综合贸易与供应链管理股份有限公司</b><small class="currency-tag">CNY</small></span>
<strong class="amount debit ledger-hide-md"><span class="amount-with-currency is-compact"><span class="currency-code">CNY</span>123,456,789,012,345.00</span></strong>
<strong class="amount credit ledger-hide-md"><span class="amount-with-currency is-compact"><span class="currency-code">CNY</span>100,000,000,000,000.00</span></strong>
<span class="ledger-result"><em class="status success">应收</em><b><span class="amount-with-currency is-compact"><span class="currency-code">CNY</span>23,456,789,012,345.00</span></b></span>
<span class="ledger-unresolved is-active"><span class="status warning">未决 2,150.00 · 3 笔</span><small style="display:block;color:var(--color-ink-muted);font-size:10px">手工记录待审 800.00 · 1 笔;待确认科目 1,350.00 · 2 笔</small></span>
<span class="ledger-cutoff">截止 2026.07.31</span>
<svg></svg>
</summary>
</details>
</div>
</section>
</section>
</main>
</div>
</div>
<aside class="drawer is-open" id="evidenceDrawer">
<header class="drawer-header">
<div>
<div class="drawer-breadcrumb"><button type="button">往来查询</button> / 甲公司 ↔ 乙公司</div>
<h2>甲公司 ↔ 乙公司</h2>
</div>
<button type="button" class="icon-button" aria-label="关闭">×</button>
</header>
<div class="drawer-body">
<div class="pair-balance-line is-six" id="drawerSix">
<div><span>期初余额</span><strong class="amount-neutral">期初不可用</strong></div>
<div><span>本期借方</span><strong><span class="amount-with-currency is-compact"><span class="currency-code">CNY</span>123,456,789,012,345.00</span></strong></div>
<div><span>本期贷方</span><strong><span class="amount-with-currency"><span class="currency-code">CNY</span>320.00</span></strong></div>
<div class="pair-final"><span>期末结果</span><strong><em class="status success">应收</em> <span class="amount-with-currency is-compact"><span class="currency-code">CNY</span>23,456,789,012,345.00</span></strong></div>
<div class="pair-unresolved is-empty"><span>未决金额</span><strong><span class="amount-with-currency"><span class="currency-code">CNY</span>0.00</span></strong></div>
<div><span>截止日</span><strong class="amount-neutral">2026.07.31</strong></div>
</div>
</div>
</aside>
<pre id="b175-metrics" hidden></pre>
<script>
function box(el) {
const r = el.getBoundingClientRect();
return { left: r.left, right: r.right, top: r.top, bottom: r.bottom, width: r.width, height: r.height };
}
function overlap(a, b) {
const dx = Math.min(a.right, b.right) - Math.max(a.left, b.left);
const dy = Math.min(a.bottom, b.bottom) - Math.max(a.top, b.top);
if (dx <= 0 || dy <= 0) return 0;
return Math.round(Math.min(dx, dy) === dy ? dx : dx);
}
function clipped(el) {
return el.scrollWidth > el.clientWidth + 1;
}
function measure() {
const row = document.querySelector("#stressRow");
const summary = row.querySelector("summary");
const amount = row.querySelector(".ledger-result .amount-with-currency");
const direction = row.querySelector(".ledger-result .status");
const unresolved = row.querySelector(".ledger-unresolved .status");
const cutoff = row.querySelector(".ledger-cutoff");
const chevron = row.querySelector("summary > svg");
const head = document.querySelector(".ledger-head.is-balances");
const drawerAmount = document.querySelector("#drawerSix .pair-final .amount-with-currency");
const amountBox = box(amount);
const unresolvedBox = box(unresolved);
const cutoffBox = box(cutoff);
const chevronBox = box(chevron);
const directionBox = box(direction);
const metrics = {
viewport: window.innerWidth,
clientWidth: document.documentElement.clientWidth,
fiveColVisible: getComputedStyle(head).display !== "none",
amountText: amount.textContent.replace(/\s+/g, " ").trim(),
unresolvedText: unresolved.textContent.replace(/\s+/g, " ").trim(),
cutoffText: cutoff.textContent.replace(/\s+/g, " ").trim(),
amountUnresolvedOverlap: overlap(amountBox, unresolvedBox),
amountCutoffOverlap: overlap(amountBox, cutoffBox),
unresolvedCutoffOverlap: overlap(unresolvedBox, cutoffBox),
unresolvedChevronOverlap: overlap(unresolvedBox, chevronBox),
cutoffChevronOverlap: overlap(cutoffBox, chevronBox),
directionUnresolvedOverlap: overlap(directionBox, unresolvedBox),
amountClipped: clipped(amount),
unresolvedClipped: clipped(unresolved),
cutoffClipped: clipped(cutoff),
amountVisible: amountBox.width > 4 && amountBox.height > 4,
unresolvedVisible: unresolvedBox.width > 4 && unresolvedBox.height > 4,
cutoffVisible: cutoffBox.width > 4 && cutoffBox.height > 4,
stackedResult: getComputedStyle(row.querySelector(".ledger-result")).flexDirection === "column",
drawerAmountClipped: clipped(drawerAmount),
summaryWidth: Math.round(box(summary).width),
};
const node = document.getElementById("b175-metrics");
node.hidden = false;
node.textContent = JSON.stringify(metrics);
document.title = "B175 " + node.textContent;
}
window.measureB175 = measure;
if (document.readyState === "complete") {
measure();
} else {
window.addEventListener("load", measure);
}
</script>
</body>
</html>
+175
View File
@@ -0,0 +1,175 @@
"""Shared fixtures for B-44 ledger / manual / position tests."""
from __future__ import annotations
from pathlib import Path
import sqlite3
import tempfile
import unittest
from bank_importer import auth, matching, master_data
from bank_importer.db import connect, migrate, utc_now
class LedgerBase(unittest.TestCase):
def setUp(self) -> None:
self.temp_dir = tempfile.TemporaryDirectory()
self.addCleanup(self.temp_dir.cleanup)
root = Path(self.temp_dir.name)
self.db_path = root / "app.db"
self.connection = connect(self.db_path)
self.addCleanup(self.connection.close)
migrate(self.connection)
self.admin = self._admin()
self.company_a = self._company("甲公司")
self.company_b = self._company("乙公司")
self.company_c = self._company("丙公司")
self.account_a = self._approved_account(self.company_a, "6222000000000001")
self.account_b = self._approved_account(self.company_b, "6222000000000002")
def _admin(self):
auth.create_user(self.connection, "admin-u", "AdminPass123", "admin")
return self.connection.execute(
"SELECT * FROM users WHERE username = 'admin-u'"
).fetchone()
def _company(self, name: str) -> int:
with self.connection:
cursor = self.connection.execute(
"INSERT INTO companies (name, created_at, updated_at) VALUES (?, ?, ?)",
(name, utc_now(), utc_now()),
)
return int(cursor.lastrowid)
def _approved_account(self, company_id: int, number: str, start: str = "2026-01-01"):
account = master_data.submit_bank_account(
self.connection, company_id=company_id, bank_name="中信银行",
account_type="基本户", account_number=number, start_date=start,
actor=None,
)
return master_data.review_bank_account(
self.connection, account["id"], "approve", None, self.admin,
effective_from=start,
)
def add_row(
self,
company_id: int,
*,
own_account: str,
cp_account: str | None = None,
income: str = "0",
expense: str = "0",
at: str = "2026-01-05T10:00:00",
currency: str = "CNY",
reference: str | None = None,
summary: str | None = None,
purpose: str | None = None,
) -> int:
with self.connection:
cursor = self.connection.execute(
"""
INSERT INTO source_files (sha256, original_filename, size_bytes, storage_path, created_at)
VALUES (?, '测试.xlsx', 1, 'data/files/测试.xlsx', ?)
""",
(utc_now(), utc_now()),
)
source_file_id = int(cursor.lastrowid)
cursor = self.connection.execute(
"""
INSERT INTO import_batches (source_file_id, status, company_id, created_at, updated_at)
VALUES (?, 'parsing', ?, ?, ?)
""",
(source_file_id, company_id, utc_now(), utc_now()),
)
batch_id = int(cursor.lastrowid)
cursor = self.connection.execute(
"""
INSERT INTO sheet_batches (
import_batch_id, sheet_name, bank_name, template_id, template_version,
header_row, own_account, own_name, period_start, period_end,
transaction_count, warnings, created_at
) VALUES (?, '流水', '测试银行', 'test-v1', 1, 1, NULL, NULL, NULL, NULL, 1, '[]', ?)
""",
(batch_id, utc_now()),
)
sheet_batch_id = int(cursor.lastrowid)
self.connection.execute(
"""
INSERT INTO sheet_reviews (
import_batch_id, sheet_name, outcome, sheet_batch_id,
review_status, created_at
) VALUES (?, '流水', 'parsed', ?, 'confirmed', ?)
""",
(batch_id, sheet_batch_id, utc_now()),
)
cursor = self.connection.execute(
"""
INSERT INTO source_rows (
sheet_batch_id, source_row, transaction_at, income, expense, balance,
own_account, own_name, counterparty_account, counterparty_name,
counterparty_bank, summary, purpose, reference, currency, created_at
) VALUES (?, 1, ?, ?, ?, NULL, ?, '测试', ?, '对方', NULL, ?, ?, ?, ?, ?)
""",
(
sheet_batch_id, at, income, expense, own_account,
cp_account, summary, purpose, reference, currency, utc_now(),
),
)
return int(cursor.lastrowid)
def pair(
self,
payer: int,
payee: int,
amount: str,
at: str = "2026-01-05T10:00:00",
*,
currency: str = "CNY",
summary: str = "借款",
purpose: str = "往来款",
) -> tuple[int, int]:
"""Create a mirrored A/B pair and reconcile into an eligible event."""
payer_account = self._account_of(payer)
payee_account = self._account_of(payee)
row_payer = self.add_row(
payer, own_account=payer_account, cp_account=payee_account,
expense=amount, at=at, currency=currency, summary=summary, purpose=purpose,
)
row_payee = self.add_row(
payee, own_account=payee_account, cp_account=payer_account,
income=amount, at=at.replace("T10:", "T11:"), currency=currency,
summary=summary, purpose=purpose,
)
matching.reconcile_rows(self.connection, [row_payer, row_payee])
return row_payer, row_payee
def _account_of(self, company_id: int) -> str:
if company_id == self.company_a:
return self.account_a["account_number"]
if company_id == self.company_b:
return self.account_b["account_number"]
return "6222000000000005"
def eligible(self) -> list[sqlite3.Row]:
return matching.eligible_intercompany_events(self.connection)
def ledger_events(self) -> list[sqlite3.Row]:
return self.connection.execute(
"SELECT * FROM ledger_events ORDER BY id"
).fetchall()
def current(self, ledger_event_id: int) -> sqlite3.Row | None:
return self.connection.execute(
"""
SELECT r.* FROM current_ledger_event_revisions c
JOIN ledger_event_revisions r ON r.id = c.revision_id
WHERE c.ledger_event_id = ?
""",
(ledger_event_id,),
).fetchone()
def position_events(self) -> list[sqlite3.Row]:
return self.connection.execute(
"SELECT * FROM eligible_position_events ORDER BY ledger_event_id"
).fetchall()
+109
View File
@@ -0,0 +1,109 @@
"""Static + Node checks for the B-44 visual rework.
Covers unique ids, nav copy, drawer/directory breakpoints, eight-column
event table, and the exported keyboard/amount helpers.
"""
from __future__ import annotations
from pathlib import Path
import re
import subprocess
import unittest
ROOT = Path(__file__).resolve().parents[1]
WEB = ROOT / "web"
class B44FrontendContractTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.admin = (WEB / "admin.html").read_text(encoding="utf-8")
cls.company = (WEB / "company.html").read_text(encoding="utf-8")
cls.css = (WEB / "styles.css").read_text(encoding="utf-8")
cls.js = (WEB / "app.js").read_text(encoding="utf-8")
def test_admin_ids_are_unique(self) -> None:
self.assertEqual(1, self.admin.count('id="companyLedgers"'))
self.assertEqual(1, self.admin.count('id="balanceLedgers"'))
self.assertIn('data-view="pair"', self.admin)
self.assertRegex(self.admin, r'data-view="pair"[^>]*>[\s\S]*?<span>往来查询</span>')
self.assertIn("<h1>往来查询</h1>", self.admin)
self.assertIn("转为异常后,该记录暂不纳入余额计算", self.admin)
self.assertIn('id="auditExceptionNote"', self.admin)
def test_company_balance_groups_and_nav(self) -> None:
self.assertIn('id="companyBalanceGroups"', self.company)
self.assertNotIn('id="companyBalanceLine"', self.company)
self.assertRegex(self.company, r'data-view="balances"[^>]*>[\s\S]*?<span>往来余额</span>')
self.assertIn("<h1>往来余额</h1>", self.company)
def test_css_directory_and_drawer_breakpoints(self) -> None:
self.assertIn("@media (max-width: 375px)", self.css)
self.assertIn("@media (max-width: 1179px)", self.css)
self.assertIn("@media (min-width: 376px) and (max-width: 900px)", self.css)
self.assertIn(
"minmax(80px, 0.85fr) minmax(152px, 1.2fr) minmax(96px, 0.9fr) minmax(84px, 0.5fr) 24px",
self.css,
)
self.assertIn("@media (max-width: 767px)", self.css)
self.assertIn(".drawer {", self.css)
self.assertIn("width: 640px", self.css)
self.assertIn(".drawer { width: 480px; }", self.css)
self.assertIn(".drawer { width: 100%; }", self.css)
self.assertIn(".drawer .pair-balance-line.is-six", self.css)
self.assertIn("repeat(3, 1fr)", self.css)
self.assertIn(".drawer .event-table { min-width: 860px; }", self.css)
self.assertIn(".company-row.is-balance-counterparty", self.css)
self.assertIn(".ledger-hide-md", self.css)
self.assertIn(".company-name b { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; min-width: 0; }", self.css)
self.assertIn(".amount-with-currency.is-compact { font-size: 11px; }", self.css)
self.assertIn("minmax(88px, 1fr) minmax(0, max-content) 24px", self.css)
self.assertIn("grid-row: 1 / span 2", self.css)
self.assertIn("flex-direction: column; align-items: flex-end; gap: 2px;", self.css)
def test_js_selectors_and_event_table(self) -> None:
self.assertIn('loadAdminBalances($("#balanceLedgers"))', self.js)
self.assertNotIn('loadAdminBalances($("#companyLedgers"))', self.js)
self.assertIn("交易日期", self.js)
self.assertIn("本方账户", self.js)
self.assertIn("对方账户", self.js)
self.assertIn("摘要", self.js)
self.assertIn("匹配状态", self.js)
self.assertIn("function eventTableHead()", self.js)
self.assertIn("colspan=\"8\"", self.js)
self.assertIn("function cycleTab(", self.js)
self.assertIn("function drawerEscAction(", self.js)
self.assertIn("function eventIsNegative(", self.js)
self.assertIn("fmtAbsMoney", self.js)
self.assertIn("is-balance-counterparty", self.js)
self.assertIn("function isCompactAmount(", self.js)
self.assertIn("function auditEvidenceFields(", self.js)
self.assertIn("无关联流水", self.js)
self.assertIn('row.dataset.direction = "outgoing"', self.js)
self.assertIn("row.dataset.counterpartyName", self.js)
self.assertIn("row.dataset.relatedSourceRowId", self.js)
heads = re.search(
r"function eventTableHead\(\) \{\s*return `([^`]+)`",
self.js,
)
self.assertIsNotNone(heads)
markup = heads.group(1)
self.assertNotIn("来源", markup)
for label in ("交易日期", "方向", "科目", "本方账户", "对方账户", "摘要", "匹配状态", "金额"):
self.assertIn(label, markup)
self.assertEqual(8, len(re.findall(r"<th\b", markup)))
def test_node_keyboard_and_amount_helpers(self) -> None:
result = subprocess.run(
["node", str(ROOT / "tests" / "b44_ui_check.js")],
capture_output=True,
text=True,
cwd=str(ROOT),
)
self.assertEqual(0, result.returncode, result.stdout + result.stderr)
self.assertIn("b44_ui_check ok", result.stdout)
if __name__ == "__main__":
unittest.main()
+530
View File
@@ -0,0 +1,530 @@
"""Chromium layout regression for the B-44 admin balance directory.
Measures 15-digit period-end amounts against the unresolved chip using
getBoundingClientRect — not static text presence.
"""
from __future__ import annotations
import base64
import hashlib
import json
import os
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
import socket
import subprocess
import tempfile
import threading
import time
from urllib.parse import urlparse
from urllib.request import urlopen
import unittest
import sys
ROOT = Path(__file__).resolve().parents[1]
if str(Path(__file__).resolve().parent) not in sys.path:
sys.path.insert(0, str(Path(__file__).resolve().parent))
FIXTURE = "/tests/fixtures/b44-balance-directory.html"
CHROME_CANDIDATES = [
Path(os.environ.get("PROGRAMFILES", r"C:\Program Files"))
/ "Google/Chrome/Application/chrome.exe",
Path(os.environ.get("PROGRAMFILES(X86)", r"C:\Program Files (x86)"))
/ "Microsoft/Edge/Application/msedge.exe",
]
VIEWPORTS = (720, 768, 800, 900, 375, 1024, 1440)
def chrome_bin() -> Path:
for path in CHROME_CANDIDATES:
if path.exists():
return path
raise FileNotFoundError("Chrome/Edge not found for layout regression")
class QuietHandler(SimpleHTTPRequestHandler):
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, directory=str(ROOT), **kwargs)
def log_message(self, *_args) -> None:
pass
def handle(self) -> None:
try:
super().handle()
except (ConnectionResetError, BrokenPipeError, TimeoutError):
pass
class _Cdp:
def __init__(self, ws_url: str) -> None:
parsed = urlparse(ws_url)
self._sock = socket.create_connection((parsed.hostname, parsed.port or 80), timeout=10)
self._sock.settimeout(10)
key = base64.b64encode(os.urandom(16)).decode()
path = parsed.path + (f"?{parsed.query}" if parsed.query else "")
self._sock.sendall(
(
f"GET {path} HTTP/1.1\r\n"
f"Host: {parsed.netloc}\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
f"Sec-WebSocket-Key: {key}\r\n"
"Sec-WebSocket-Version: 13\r\n\r\n"
).encode()
)
header = b""
while b"\r\n\r\n" not in header:
chunk = self._sock.recv(4096)
if not chunk:
raise ConnectionError("CDP websocket handshake failed")
header += chunk
expected = base64.b64encode(
hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()).digest()
).decode()
if expected not in header.decode("latin1"):
raise ConnectionError("CDP websocket accept mismatch")
leftover = header.split(b"\r\n\r\n", 1)[1]
self._buf = leftover
self._next_id = 1
def call(self, method: str, params: dict | None = None, timeout: float = 15) -> dict:
msg_id = self._next_id
self._next_id += 1
self._send({"id": msg_id, "method": method, "params": params or {}})
deadline = time.time() + timeout
while time.time() < deadline:
payload = self._recv()
if payload.get("id") == msg_id:
if "error" in payload:
raise RuntimeError(f"{method}: {payload['error']}")
return payload.get("result") or {}
raise TimeoutError(method)
def wait_event(self, name: str, timeout: float = 15) -> dict:
deadline = time.time() + timeout
while time.time() < deadline:
payload = self._recv()
if payload.get("method") == name:
return payload.get("params") or {}
raise TimeoutError(name)
def close(self) -> None:
try:
self._sock.close()
except OSError:
pass
def _send(self, obj: dict) -> None:
data = json.dumps(obj, separators=(",", ":")).encode()
mask = os.urandom(4)
header = bytearray([0x81])
length = len(data)
if length < 126:
header.append(0x80 | length)
elif length < 65536:
header.append(0x80 | 126)
header.extend(length.to_bytes(2, "big"))
else:
header.append(0x80 | 127)
header.extend(length.to_bytes(8, "big"))
header.extend(mask)
masked = bytes(b ^ mask[i % 4] for i, b in enumerate(data))
self._sock.sendall(header + masked)
def _recv(self) -> dict:
while True:
opcode, payload = self._read_frame()
if opcode == 0x9:
self._send_pong(payload)
continue
if opcode == 0xA:
continue
if opcode == 0x8:
raise ConnectionError("CDP websocket closed")
return json.loads(payload.decode())
def _send_pong(self, payload: bytes) -> None:
mask = os.urandom(4)
header = bytearray([0x8A, 0x80 | len(payload)])
header.extend(mask)
masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
self._sock.sendall(header + masked)
def _read_frame(self) -> tuple[int, bytes]:
header = self._read_exact(2)
opcode = header[0] & 0x0F
length = header[1] & 0x7F
masked = bool(header[1] & 0x80)
if length == 126:
length = int.from_bytes(self._read_exact(2), "big")
elif length == 127:
length = int.from_bytes(self._read_exact(8), "big")
mask = self._read_exact(4) if masked else b""
payload = self._read_exact(length)
if masked:
payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
return opcode, payload
def _read_exact(self, size: int) -> bytes:
while len(self._buf) < size:
chunk = self._sock.recv(4096)
if not chunk:
raise ConnectionError("CDP websocket closed")
self._buf += chunk
data, self._buf = self._buf[:size], self._buf[size:]
return data
class B44LayoutRegressionTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.chrome = chrome_bin()
cls.httpd = ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler)
cls.http_thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True)
cls.http_thread.start()
cls.port = cls.httpd.server_address[1]
cls.url = f"http://127.0.0.1:{cls.port}{FIXTURE}"
cls.tmp = tempfile.TemporaryDirectory(prefix="b44-layout-", ignore_cleanup_errors=True)
user_dir = cls.tmp.name
cls.proc = subprocess.Popen(
[
str(cls.chrome),
"--headless=new",
"--disable-gpu",
"--no-first-run",
"--disable-extensions",
"--remote-debugging-port=0",
f"--user-data-dir={user_dir}",
"about:blank",
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
port_file = Path(user_dir) / "DevToolsActivePort"
deadline = time.time() + 15
listing = None
while time.time() < deadline:
if port_file.exists() and port_file.stat().st_size:
text = port_file.read_text(encoding="utf-8").strip().splitlines()
if text:
cls.debug_port = int(text[0])
try:
listing = json.loads(
urlopen(
f"http://127.0.0.1:{cls.debug_port}/json/list",
timeout=5,
).read()
)
except Exception:
listing = None
if listing and any(item.get("type") == "page" for item in listing):
break
time.sleep(0.05)
else:
raise RuntimeError("Chrome DevTools port not ready")
page = next(item for item in listing if item.get("type") == "page")
cls.cdp = _Cdp(page["webSocketDebuggerUrl"])
cls.cdp.call("Runtime.enable")
cls.cdp.call("Page.enable")
@classmethod
def tearDownClass(cls) -> None:
if getattr(cls, "cdp", None):
cls.cdp.close()
if getattr(cls, "proc", None):
cls.proc.terminate()
try:
cls.proc.wait(timeout=5)
except subprocess.TimeoutExpired:
cls.proc.kill()
cls.proc.wait(timeout=5)
time.sleep(0.2)
if getattr(cls, "httpd", None):
cls.httpd.shutdown()
cls.httpd.server_close()
if getattr(cls, "tmp", None):
cls.tmp.cleanup()
def _metrics(self, width: int) -> dict:
self.cdp.call("Emulation.setDeviceMetricsOverride", {
"width": width,
"height": 1100,
"deviceScaleFactor": 1,
"mobile": False,
})
self.cdp.call("Page.navigate", {"url": f"{self.url}?w={width}"})
deadline = time.time() + 10
last = None
while time.time() < deadline:
result = self.cdp.call(
"Runtime.evaluate",
{
"expression": (
"typeof window.measureB175 === 'function' "
"? (window.measureB175(), document.getElementById('b175-metrics').textContent) "
": null"
),
"returnByValue": True,
},
)
last = result.get("result", {}).get("value")
if last:
return json.loads(last)
time.sleep(0.1)
raise TimeoutError(f"layout metrics not ready: {last}")
def test_fifteen_digit_amount_clear_of_unresolved_chip(self) -> None:
for width in VIEWPORTS:
with self.subTest(width=width):
metrics = self._metrics(width)
self.assertEqual(width, metrics["viewport"], metrics)
self.assertIn("23,456,789,012,345.00", metrics["amountText"])
self.assertIn("未决", metrics["unresolvedText"])
self.assertIn("2026.07.31", metrics["cutoffText"])
self.assertTrue(metrics["amountVisible"])
self.assertTrue(metrics["unresolvedVisible"])
self.assertTrue(metrics["cutoffVisible"])
self.assertFalse(metrics["amountClipped"], metrics)
self.assertFalse(metrics["unresolvedClipped"], metrics)
self.assertFalse(metrics["cutoffClipped"], metrics)
self.assertEqual(0, metrics["amountUnresolvedOverlap"], metrics)
self.assertEqual(0, metrics["amountCutoffOverlap"], metrics)
self.assertEqual(0, metrics["unresolvedCutoffOverlap"], metrics)
self.assertEqual(0, metrics["directionUnresolvedOverlap"], metrics)
self.assertEqual(0, metrics["cutoffChevronOverlap"], metrics)
self.assertFalse(metrics["drawerAmountClipped"], metrics)
if 376 <= width <= 1179:
self.assertTrue(metrics["fiveColVisible"], metrics)
if 376 <= width <= 900:
self.assertTrue(metrics["stackedResult"], metrics)
if width <= 375:
self.assertFalse(metrics["fiveColVisible"], metrics)
LIVE_MEASURE = r"""
(() => {
const pair = document.querySelector('[data-view="pair"]');
if (pair) pair.click();
const rows = [...document.querySelectorAll("#balanceLedgers .company-ledger.is-balances")];
if (!rows.length) return null;
const row = rows.find((item) => (item.textContent || "").includes("123,456,789,012,345")) || rows[0];
const amount = row.querySelector(".ledger-result .amount-with-currency");
const direction = row.querySelector(".ledger-result .status");
const unresolved = row.querySelector(".ledger-unresolved .status") || row.querySelector(".ledger-unresolved");
const cutoff = row.querySelector(".ledger-cutoff");
const chevron = row.querySelector("summary > svg");
const box = (el) => {
const r = el.getBoundingClientRect();
return { left: r.left, right: r.right, top: r.top, bottom: r.bottom };
};
const overlap = (a, b) => {
const dx = Math.min(a.right, b.right) - Math.max(a.left, b.left);
const dy = Math.min(a.bottom, b.bottom) - Math.max(a.top, b.top);
if (dx <= 0 || dy <= 0) return 0;
return Math.round(dx);
};
const clipped = (el) => el.scrollWidth > el.clientWidth + 1;
const amountBox = box(amount);
const unresolvedBox = box(unresolved);
return {
viewport: window.innerWidth,
amountText: (amount.textContent || "").replace(/\s+/g, " ").trim(),
unresolvedText: (unresolved.textContent || "").replace(/\s+/g, " ").trim(),
cutoffText: (cutoff.textContent || "").replace(/\s+/g, " ").trim(),
amountUnresolvedOverlap: overlap(amountBox, unresolvedBox),
directionUnresolvedOverlap: overlap(box(direction), unresolvedBox),
amountCutoffOverlap: overlap(amountBox, box(cutoff)),
unresolvedCutoffOverlap: overlap(unresolvedBox, box(cutoff)),
cutoffChevronOverlap: overlap(box(cutoff), box(chevron)),
amountClipped: clipped(amount),
unresolvedClipped: clipped(unresolved),
cutoffClipped: clipped(cutoff),
fiveColVisible: getComputedStyle(document.querySelector(".ledger-head.is-balances")).display !== "none",
stackedResult: getComputedStyle(row.querySelector(".ledger-result")).flexDirection === "column",
rowCount: rows.length,
};
})()
"""
COMPANY_MEASURE = r"""
(() => {
const balances = document.querySelector('[data-view="balances"]');
if (balances) balances.click();
const row = document.querySelector(".company-row.is-balance-counterparty");
if (!row) return null;
const amount = row.querySelector(".amount-with-currency") || row.querySelector(".company-row-figure strong");
const direction = row.querySelector(".status");
const cutoff = row.querySelector(".company-row-figure small");
const box = (el) => {
const r = el.getBoundingClientRect();
return { left: r.left, right: r.right, top: r.top, bottom: r.bottom, width: r.width, height: r.height };
};
const overlap = (a, b) => {
const dx = Math.min(a.right, b.right) - Math.max(a.left, b.left);
const dy = Math.min(a.bottom, b.bottom) - Math.max(a.top, b.top);
if (dx <= 0 || dy <= 0) return 0;
return Math.round(dx);
};
return {
viewport: window.innerWidth,
amountText: (amount?.textContent || "").replace(/\s+/g, " ").trim(),
directionText: (direction?.textContent || "").replace(/\s+/g, " ").trim(),
cutoffText: (cutoff?.textContent || "").replace(/\s+/g, " ").trim(),
amountDirectionOverlap: amount && direction ? overlap(box(amount), box(direction)) : 0,
visible: Boolean(amount && amount.getBoundingClientRect().width > 4),
};
})()
"""
class LiveAdminDirectoryLayoutTests(unittest.TestCase):
"""Chromium against a real login/API page with B-168-scale 15-digit amounts."""
@classmethod
def setUpClass(cls) -> None:
from test_positions_api import IntercompanyApiTests
cls.api = IntercompanyApiTests("test_admin_balances_directory")
cls.api.setUp()
confirmed = cls.api._fresh_pair("123456789012345.00")
cls.api._confirm(confirmed["ledger_event_id"])
cls.api._fresh_pair("2150.00")
cls.tmp = tempfile.TemporaryDirectory(prefix="b44-live-", ignore_cleanup_errors=True)
cls.proc = subprocess.Popen(
[
str(chrome_bin()),
"--headless=new",
"--disable-gpu",
"--no-first-run",
"--disable-extensions",
"--remote-debugging-port=0",
f"--user-data-dir={cls.tmp.name}",
"about:blank",
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
port_file = Path(cls.tmp.name) / "DevToolsActivePort"
deadline = time.time() + 15
listing = None
while time.time() < deadline:
if port_file.exists() and port_file.stat().st_size:
text = port_file.read_text(encoding="utf-8").strip().splitlines()
if text:
cls.debug_port = int(text[0])
try:
listing = json.loads(
urlopen(
f"http://127.0.0.1:{cls.debug_port}/json/list",
timeout=5,
).read()
)
except Exception:
listing = None
if listing and any(item.get("type") == "page" for item in listing):
break
time.sleep(0.05)
else:
raise RuntimeError("Chrome DevTools port not ready")
page = next(item for item in listing if item.get("type") == "page")
cls.cdp = _Cdp(page["webSocketDebuggerUrl"])
cls.cdp.call("Runtime.enable")
cls.cdp.call("Page.enable")
cls.cdp.call("Network.enable")
token = cls.api.admin.cookies["cw_session"]
cls.cdp.call(
"Network.setCookie",
{
"name": "cw_session",
"value": token,
"url": f"http://127.0.0.1:{cls.api.port}/",
},
)
@classmethod
def tearDownClass(cls) -> None:
if getattr(cls, "cdp", None):
cls.cdp.close()
if getattr(cls, "proc", None):
cls.proc.terminate()
try:
cls.proc.wait(timeout=5)
except subprocess.TimeoutExpired:
cls.proc.kill()
cls.proc.wait(timeout=5)
time.sleep(0.2)
if getattr(cls, "api", None):
cls.api.tearDown()
cls.api.doCleanups()
if getattr(cls, "tmp", None):
cls.tmp.cleanup()
def _eval(self, expression: str):
deadline = time.time() + 12
last = None
while time.time() < deadline:
result = self.cdp.call(
"Runtime.evaluate",
{"expression": expression, "returnByValue": True},
)
last = result.get("result", {}).get("value")
if last:
return last
time.sleep(0.15)
raise TimeoutError(f"live page metrics not ready: {last}")
def _open(self, path: str, width: int) -> None:
self.cdp.call("Emulation.setDeviceMetricsOverride", {
"width": width,
"height": 1100,
"deviceScaleFactor": 1,
"mobile": False,
})
self.cdp.call("Page.navigate", {"url": f"http://127.0.0.1:{self.api.port}/{path}"})
time.sleep(0.4)
def test_live_admin_directory_viewports(self) -> None:
for width in (720, 768, 800, 900, 375, 1024, 1440):
with self.subTest(width=width):
self._open("admin.html", width)
metrics = self._eval(LIVE_MEASURE)
self.assertEqual(width, metrics["viewport"], metrics)
self.assertGreaterEqual(metrics["rowCount"], 1, metrics)
self.assertIn("123,456,789,012,345", metrics["amountText"], metrics)
self.assertIn("未决", metrics["unresolvedText"], metrics)
self.assertIn("2026.07.31", metrics["cutoffText"], metrics)
self.assertFalse(metrics["amountClipped"], metrics)
self.assertFalse(metrics["unresolvedClipped"], metrics)
self.assertEqual(0, metrics["amountUnresolvedOverlap"], metrics)
self.assertEqual(0, metrics["directionUnresolvedOverlap"], metrics)
self.assertEqual(0, metrics["amountCutoffOverlap"], metrics)
if 376 <= width <= 900:
self.assertTrue(metrics["stackedResult"], metrics)
if width <= 375:
self.assertFalse(metrics["fiveColVisible"], metrics)
def test_live_company_balance_rows(self) -> None:
token = self.api.cashier_a.cookies["cw_session"]
self.cdp.call(
"Network.setCookie",
{
"name": "cw_session",
"value": token,
"url": f"http://127.0.0.1:{self.api.port}/",
},
)
for width in (375, 1440):
with self.subTest(width=width):
self._open("company.html", width)
metrics = self._eval(COMPANY_MEASURE)
self.assertTrue(metrics["visible"], metrics)
self.assertTrue(metrics["amountText"], metrics)
self.assertTrue(metrics["directionText"], metrics)
self.assertIn("截止", metrics["cutoffText"], metrics)
self.assertEqual(0, metrics["amountDirectionOverlap"], metrics)
if __name__ == "__main__":
unittest.main()
+306
View File
@@ -0,0 +1,306 @@
"""B-44 ledger event layer tests: bank reconciliation, revision chain,
reversal/adjustment, projection rebuild and immutability boundaries."""
from __future__ import annotations
from decimal import Decimal
import sqlite3
import unittest
from bank_importer import ledger_events, matching, subjects
from ledger_helpers import LedgerBase
class BankReconcileTests(LedgerBase):
def test_paired_eligible_event_creates_pending_subject_only(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
self.assertEqual(1, len(self.eligible()))
stats = ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
self.assertEqual(1, stats["created"])
events = self.ledger_events()
self.assertEqual(1, len(events))
revision = self.current(events[0]["id"])
self.assertEqual("pending_subject", revision["state"])
self.assertEqual(Decimal("100.00"), Decimal(revision["amount"]))
self.assertEqual("CNY", revision["currency"])
# Nothing confirmed yet -> no position event.
self.assertEqual([], self.position_events())
# The event is unresolved as subject_review.
from bank_importer import positions
unresolved = positions.unresolved_for_company(self.connection, self.company_a, "2026-12-31")
self.assertEqual("100.00", unresolved["by_reason"]["subject_review"]["gross_amount"])
def test_unlocked_single_never_enters_ledger(self) -> None:
self.add_row(
self.company_a, own_account="6222000000000001",
cp_account="6222000000000002", expense="100.00",
)
matching.reconcile_rows(self.connection, [1])
self.assertEqual([], self.eligible())
stats = ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
self.assertEqual(0, stats["created"])
self.assertEqual([], self.ledger_events())
def test_same_company_transfer_never_enters_ledger(self) -> None:
self.add_row(
self.company_a, own_account="6222000000000001",
cp_account="6222000000000001", expense="100.00",
)
self.add_row(
self.company_a, own_account="6222000000000001",
cp_account="6222000000000001", income="100.00",
)
matching.reconcile_rows(self.connection, [1, 2])
eligible = self.eligible()
self.assertEqual(0, len([e for e in eligible if e["payer_company_id"] == e["payee_company_id"]]))
stats = ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
self.assertEqual(0, stats["created"])
def test_reconcile_is_idempotent(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
events = self.ledger_events()
self.assertEqual(1, len(events))
before_revisions = self.connection.execute(
"SELECT COUNT(*) AS n FROM ledger_event_revisions"
).fetchone()["n"]
stats = ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
self.assertEqual(1, stats["unchanged"])
after_revisions = self.connection.execute(
"SELECT COUNT(*) AS n FROM ledger_event_revisions"
).fetchone()["n"]
self.assertEqual(before_revisions, after_revisions)
def test_reconcile_picks_up_newly_linked_rows_as_new_pending_event(self) -> None:
# A single observation locked as intercompany -> pending ledger event.
row_a = self.add_row(
self.company_a, own_account="6222000000000001",
cp_account="6222000000000002", expense="100.00",
)
matching.reconcile_rows(self.connection, [row_a])
single = self.connection.execute(
"SELECT * FROM current_transfer_decisions"
).fetchone()
event_id = single["event_id"]
decision = self.connection.execute(
"SELECT * FROM transfer_match_decisions WHERE id = ?",
(single["decision_id"],),
).fetchone()
matching.apply_manual_decision(
self.connection, event_id, "assign_participant",
reason="函证确认对方",
expected_revision=decision["revision"], request_key="assign-1",
actor=self.admin,
participant={"role": "payee", "company_id": self.company_b},
)
self.assertEqual(1, len(self.eligible()))
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
self.assertEqual(1, len(self.ledger_events()))
# The administrator links two fresh rows with a different amount onto a
# new canonical event; reconcile creates a second pending event.
row_a2 = self.add_row(
self.company_a, own_account="6222000000000001",
cp_account="6222000000000002", expense="150.00",
at="2026-02-01T10:00:00",
)
row_b2 = self.add_row(
self.company_b, own_account="6222000000000002",
cp_account="6222000000000001", income="150.00",
at="2026-02-01T11:00:00",
)
current_row = self.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()
matching.apply_manual_decision(
self.connection, event_id, "link_rows",
reason="补录双边流水",
expected_revision=current_row["revision"], request_key="link-1",
actor=self.admin, source_row_ids=[row_a2, row_b2],
)
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
amounts = sorted(
row["amount"] for row in self.connection.execute(
"SELECT amount FROM ledger_event_revisions WHERE state = 'pending_subject'"
).fetchall()
)
self.assertEqual(["100.00", "150.00"], amounts)
def test_confirm_subject_enters_position_view(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
subjects.confirm_subject(
self.connection, event_id,
perspective_company_id=self.company_a, subject_code="other_receivable",
reason="借款建议其他应收", expected_revision=1, request_key="k1",
actor=self.admin,
)
positions = self.position_events()
self.assertEqual(1, len(positions))
self.assertEqual("other_receivable", positions[0]["subject_code"])
self.assertEqual(self.company_a, positions[0]["perspective_company_id"])
def test_confirm_stale_revision_conflicts(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
subjects.confirm_subject(
self.connection, event_id,
perspective_company_id=self.company_a, subject_code="receivable",
reason="确认应收", expected_revision=1, request_key="k1",
actor=self.admin,
)
with self.assertRaises(subjects.SubjectConflictError):
subjects.confirm_subject(
self.connection, event_id,
perspective_company_id=self.company_a, subject_code="receivable",
reason="重复确认", expected_revision=1, request_key="k2",
actor=self.admin,
)
def test_reversal_mirrors_direction_and_subject(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
subjects.confirm_subject(
self.connection, event_id,
perspective_company_id=self.company_a, subject_code="other_receivable",
reason="借款", expected_revision=1, request_key="k1", actor=self.admin,
)
reversal_id, _ = ledger_events.create_reversal(
self.connection, event_id, source_kind="adjustment",
reason="冲销错误确认", actor=self.admin,
)
reversal = self.current(reversal_id)
self.assertEqual("reversal", reversal["posting_kind"])
self.assertEqual(self.company_b, reversal["payer_company_id"])
self.assertEqual(self.company_a, reversal["payee_company_id"])
self.assertEqual("other_payable", reversal["subject_code"])
self.assertEqual(self.company_b, reversal["perspective_company_id"])
# Original event stays confirmed and still in the position view.
self.assertEqual("confirmed", self.current(event_id)["state"])
positions = self.position_events()
self.assertEqual(2, len(positions))
signed = sum(
Decimal(position["amount"]) * (
1 if position["payer_company_id"] == self.company_a else -1
)
for position in positions
)
self.assertEqual(Decimal("0"), signed)
def test_evidence_rows_stay_immutable_under_reversal(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
subjects.confirm_subject(
self.connection, event_id,
perspective_company_id=self.company_a, subject_code="receivable",
reason="确认", expected_revision=1, request_key="k1", actor=self.admin,
)
before = self.connection.execute(
"SELECT income, expense FROM source_rows ORDER BY id"
).fetchall()
ledger_events.create_reversal(
self.connection, event_id, source_kind="adjustment",
reason="冲销", actor=self.admin,
)
after = self.connection.execute(
"SELECT income, expense FROM source_rows ORDER BY id"
).fetchall()
self.assertEqual(before, after)
def test_reversal_idempotent_guard(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
subjects.confirm_subject(
self.connection, event_id,
perspective_company_id=self.company_a, subject_code="receivable",
reason="确认", expected_revision=1, request_key="k1", actor=self.admin,
)
# Simulate the B-43 event leaving the eligible set.
with self.connection:
self.connection.execute(
"UPDATE canonical_transfer_events SET lifecycle = 'superseded' WHERE id = ?",
(self.eligible()[0]["event_id"],),
)
stats = ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
self.assertEqual(1, stats["reversal"])
reversals = self.connection.execute(
"SELECT COUNT(*) AS n FROM ledger_event_revisions WHERE posting_kind = 'reversal'"
).fetchone()["n"]
self.assertEqual(1, reversals)
# Re-run does not create a second reversal.
stats = ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
self.assertEqual(0, stats["reversal"])
class ProjectionTests(LedgerBase):
def test_rebuild_current_projection(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
subjects.confirm_subject(
self.connection, event_id,
perspective_company_id=self.company_a, subject_code="receivable",
reason="确认", expected_revision=1, request_key="k1", actor=self.admin,
)
self.connection.execute("DELETE FROM current_ledger_event_revisions")
rebuilt = ledger_events.rebuild_current_ledger_projection(self.connection)
self.assertGreaterEqual(rebuilt, 1)
revision = self.current(event_id)
self.assertEqual("confirmed", revision["state"])
self.assertEqual("receivable", revision["subject_code"])
class SubjectSuggestionTests(LedgerBase):
def test_mirror_mapping_is_symmetric(self) -> None:
for subject, mirror in (
("receivable", "payable"), ("payable", "receivable"),
("other_receivable", "other_payable"), ("other_payable", "other_receivable"),
):
self.assertEqual(mirror, subjects.mirror_subject(subject))
self.assertEqual(subject, subjects.mirror_subject(mirror))
def test_loan_keyword_suggests_payer_other_receivable(self) -> None:
self.pair(self.company_a, self.company_b, "100.00", summary="借款", purpose="往来款")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
suggestions = subjects.compute_suggestions(self.connection, event_id)
self.assertEqual(1, len(suggestions))
self.assertEqual(self.company_a, suggestions[0]["suggested_perspective_company_id"])
self.assertEqual("other_receivable", suggestions[0]["suggested_subject_code"])
self.assertFalse(suggestions[0]["evidence"]["approved"])
def test_repay_keyword_suggests_payee(self) -> None:
self.pair(self.company_a, self.company_b, "100.00", summary="归还往来款", purpose="还款")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
suggestions = subjects.compute_suggestions(self.connection, event_id)
self.assertEqual(1, len(suggestions))
self.assertEqual(self.company_b, suggestions[0]["suggested_perspective_company_id"])
def test_trade_keyword_never_suggests(self) -> None:
# Trade vocabulary needs group approval; no dictionary hit -> review.
self.pair(self.company_a, self.company_b, "100.00", summary="货款", purpose="采购货款")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
self.assertEqual([], subjects.compute_suggestions(self.connection, event_id))
def test_name_alone_never_confirms(self) -> None:
# A name match on the counterparty cannot decide the subject.
self.pair(self.company_a, self.company_b, "100.00", summary="某客户往来", purpose=None)
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
current = self.current(event_id)
self.assertEqual("pending_subject", current["state"])
self.assertEqual([], subjects.compute_suggestions(self.connection, event_id))
if __name__ == "__main__":
unittest.main()
+84
View File
@@ -0,0 +1,84 @@
"""Static contracts for the B-229 login desktop layout."""
from __future__ import annotations
from pathlib import Path
import re
import unittest
ROOT = Path(__file__).resolve().parents[1]
WEB = ROOT / "web"
class LoginFrontendContractTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.html = (WEB / "index.html").read_text(encoding="utf-8")
cls.css = (WEB / "styles.css").read_text(encoding="utf-8")
cls.js = (WEB / "app.js").read_text(encoding="utf-8")
def test_dual_zone_markup_and_auth_controls(self) -> None:
self.assertIn('class="entry-context"', self.html)
self.assertIn('class="entry-form-wrap"', self.html)
self.assertIn('id="loginForm"', self.html)
self.assertIn('id="togglePassword"', self.html)
self.assertIn('id="changePassword"', self.html)
self.assertIn('id="loginError"', self.html)
self.assertIn('name="username"', self.html)
self.assertIn('name="password"', self.html)
self.assertIn('name="new_password"', self.html)
self.assertIn('name="confirm_password"', self.html)
self.assertIn('value="admin"', self.html)
self.assertIn('value="company"', self.html)
self.assertIn("总账管理端", self.html)
self.assertIn("公司业务端", self.html)
self.assertIn("服务状态", self.html)
self.assertIn("当前账期", self.html)
self.assertNotIn("entry-card", self.html)
self.assertNotIn('value="group-admin"', self.html)
self.assertNotIn('value="demo123456"', self.html)
def test_login_css_lives_in_one_section(self) -> None:
self.assertIn("minmax(460px, 520px)", self.css)
self.assertIn("max-width: 1180px", self.css)
self.assertIn("margin: 0 auto", self.css)
self.assertIn("width: min(440px, 100%)", self.css)
self.assertIn("@media (max-width: 1023px)", self.css)
self.assertIn("@media (max-width: 720px)", self.css)
self.assertIn("minmax(400px, 440px)", self.css)
self.assertNotIn(".entry-card", self.css)
self.assertNotIn("0 0 60px rgba(55, 235, 137, 0.06)", self.css)
login_block = self.css.split("/* Login */", 1)[1]
after_login = login_block.split("/* ---- B-44", 1)[0] if "/* ---- B-44" in login_block else login_block
self.assertEqual(1, after_login.count(".entry-shell { width: 100%;"))
self.assertEqual(3, len(re.findall(r"\.entry-shell \{", after_login)))
later = self.css.split("/* ---- B-44 intercompany balances ---- */", 1)[-1]
self.assertNotIn(".entry-shell", later)
self.assertNotIn(".entry-form {", later)
self.assertNotIn(".role-switch {", later)
def test_login_error_and_keyboard_focus_follow_tokens(self) -> None:
self.assertIn(".entry-form #loginError { color: var(--color-danger); }", self.css)
self.assertIn(
".entry-form .field input:focus-visible { outline: 2px solid var(--color-primary-strong); outline-offset: 3px; }",
self.css,
)
def test_role_switch_reuses_segmented_rhythm(self) -> None:
self.assertIn(".role-switch { display: flex; gap: 4px; padding: 4px;", self.css)
self.assertIn("min-height: 36px", self.css)
self.assertIn('role="radiogroup"', self.html)
self.assertIn("class=\"sr-only\"", self.html)
def test_auth_javascript_untouched(self) -> None:
self.assertIn("function initEntry()", self.js)
self.assertIn('fetch("/api/login"', self.js)
self.assertIn('fetch("/api/password/change"', self.js)
self.assertIn("must_change_password", self.js)
self.assertIn("#togglePassword", self.js)
self.assertIn("进入总账管理端", self.js)
self.assertIn("进入公司业务端", self.js)
if __name__ == "__main__":
unittest.main()
+228
View File
@@ -0,0 +1,228 @@
"""Chromium layout regression for the B-229 login page.
Measures dual-zone geometry, form width, overflow and overlap at the
accepted viewports. Screenshots are written next to other visual evidence.
"""
from __future__ import annotations
import base64
import json
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from http.server import ThreadingHTTPServer
from pathlib import Path
from urllib.request import urlopen
ROOT = Path(__file__).resolve().parents[1]
if str(Path(__file__).resolve().parent) not in sys.path:
sys.path.insert(0, str(Path(__file__).resolve().parent))
from test_b44_layout import QuietHandler, _Cdp, chrome_bin # noqa: E402
SCREEN_DIR = ROOT / "screenshots"
LOGIN_URL = "/web/index.html"
MEASURE = r"""
(() => {
const box = (el) => {
if (!el) return null;
const r = el.getBoundingClientRect();
return {
left: r.left, right: r.right, top: r.top, bottom: r.bottom,
width: r.width, height: r.height
};
};
const overlapArea = (a, b) => {
if (!a || !b) return 0;
const dx = Math.min(a.right, b.right) - Math.max(a.left, b.left);
const dy = Math.min(a.bottom, b.bottom) - Math.max(a.top, b.top);
if (dx <= 0 || dy <= 0) return 0;
return Math.round(dx * dy);
};
const shell = document.querySelector(".entry-shell");
const context = document.querySelector(".entry-context");
const wrap = document.querySelector(".entry-form-wrap");
const form = document.querySelector(".entry-form");
const role = document.querySelector(".role-switch");
const spans = [...document.querySelectorAll(".role-switch span")];
const brand = document.querySelector("#product-name");
const status = document.querySelector(".entry-status");
const statement = document.querySelector(".entry-statement h1");
const sh = box(shell);
const ctx = box(context);
const wr = box(wrap);
const fm = box(form);
const sp = spans.map(box);
const clipped = (el) => Boolean(el && el.scrollWidth > el.clientWidth + 1);
return {
viewport: window.innerWidth,
viewportH: window.innerHeight,
grid: getComputedStyle(shell).gridTemplateColumns,
shellBox: sh,
shellCentered: Boolean(sh && Math.abs((sh.left + sh.right) / 2 - window.innerWidth / 2) <= 1),
contextBox: ctx,
wrapBox: wr,
formBox: fm,
formWidth: fm ? Math.round(fm.width) : 0,
roleHeight: role ? Math.round(role.getBoundingClientRect().height) : 0,
roleHorizontal: sp.length === 2 && sp[0].right <= sp[1].left + 2,
dualZone: Boolean(ctx && wr && ctx.right <= wr.left + 2 && Math.abs(ctx.top - wr.top) < 48),
stacked: Boolean(ctx && wr && ctx.bottom <= wr.top + 8),
overlap: overlapArea(ctx, fm),
pageOverflowX: document.documentElement.scrollWidth - window.innerWidth,
formBottom: fm ? fm.bottom : 0,
formFitsViewport: Boolean(fm && fm.top >= -1 && fm.bottom <= window.innerHeight + 2),
brandClipped: clipped(brand),
statementClipped: clipped(statement),
statusText: (status?.textContent || "").replace(/\s+/g, " ").trim(),
periodText: (document.querySelector(".entry-facts")?.textContent || "").replace(/\s+/g, " ").trim(),
hasToggle: Boolean(document.querySelector("#togglePassword")),
hasError: Boolean(document.querySelector("#loginError")),
changeHidden: document.querySelector("#changePassword")?.hidden === true,
};
})()
"""
VIEWPORTS = (
(1440, 900, "desktop", "b229-login-1440.png"),
(1366, 768, "desktop", "b229-login-1366.png"),
(1024, 768, "desktop", "b229-login-1024.png"),
(768, 1024, "compact", "b229-login-768.png"),
(375, 812, "mobile", "b229-login-375.png"),
)
class LoginLayoutTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.chrome = chrome_bin()
cls.httpd = ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler)
cls.http_thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True)
cls.http_thread.start()
cls.port = cls.httpd.server_address[1]
cls.tmp = tempfile.TemporaryDirectory(prefix="b229-login-", ignore_cleanup_errors=True)
cls.proc = subprocess.Popen(
[
str(cls.chrome),
"--headless=new",
"--disable-gpu",
"--no-first-run",
"--disable-extensions",
"--hide-scrollbars",
"--remote-debugging-port=0",
f"--user-data-dir={cls.tmp.name}",
"about:blank",
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
port_file = Path(cls.tmp.name) / "DevToolsActivePort"
deadline = time.time() + 15
listing = None
while time.time() < deadline:
if port_file.exists() and port_file.stat().st_size:
text = port_file.read_text(encoding="utf-8").strip().splitlines()
if text:
cls.debug_port = int(text[0])
try:
listing = json.loads(
urlopen(f"http://127.0.0.1:{cls.debug_port}/json/list", timeout=5).read()
)
except Exception:
listing = None
if listing and any(item.get("type") == "page" for item in listing):
break
time.sleep(0.05)
else:
raise RuntimeError("Chrome DevTools port not ready")
page = next(item for item in listing if item.get("type") == "page")
cls.cdp = _Cdp(page["webSocketDebuggerUrl"])
cls.cdp.call("Runtime.enable")
cls.cdp.call("Page.enable")
SCREEN_DIR.mkdir(exist_ok=True)
@classmethod
def tearDownClass(cls) -> None:
if getattr(cls, "cdp", None):
cls.cdp.close()
if getattr(cls, "proc", None):
cls.proc.terminate()
try:
cls.proc.wait(timeout=5)
except Exception:
cls.proc.kill()
cls.proc.wait(timeout=5)
time.sleep(0.2)
if getattr(cls, "httpd", None):
cls.httpd.shutdown()
cls.httpd.server_close()
if getattr(cls, "tmp", None):
cls.tmp.cleanup()
def _open(self, width: int, height: int) -> dict:
self.cdp.call(
"Emulation.setDeviceMetricsOverride",
{
"width": width,
"height": height,
"deviceScaleFactor": 1,
"mobile": width <= 720,
},
)
self.cdp.call("Page.navigate", {"url": f"http://127.0.0.1:{self.port}{LOGIN_URL}"})
time.sleep(0.4)
deadline = time.time() + 8
last = None
while time.time() < deadline:
result = self.cdp.call("Runtime.evaluate", {"expression": MEASURE, "returnByValue": True})
last = result.get("result", {}).get("value")
if last and last.get("formWidth"):
return last
time.sleep(0.1)
raise TimeoutError(f"login metrics not ready: {last}")
def _shot(self, name: str) -> None:
raw = self.cdp.call("Page.captureScreenshot", {"format": "png", "captureBeyondViewport": False})
(SCREEN_DIR / name).write_bytes(base64.b64decode(raw["data"]))
def test_login_viewports(self) -> None:
for width, height, mode, shot in VIEWPORTS:
with self.subTest(width=width, height=height, mode=mode):
metrics = self._open(width, height)
self._shot(shot)
self.assertEqual(width, metrics["viewport"], metrics)
self.assertEqual(0, metrics["overlap"], metrics)
self.assertLessEqual(metrics["pageOverflowX"], 1, metrics)
self.assertFalse(metrics["brandClipped"], metrics)
self.assertFalse(metrics["statementClipped"], metrics)
self.assertTrue(metrics["roleHorizontal"], metrics)
self.assertLessEqual(metrics["roleHeight"], 52, metrics)
self.assertIn("运行中", metrics["statusText"], metrics)
self.assertIn("2026 年 7 月", metrics["periodText"], metrics)
self.assertTrue(metrics["hasToggle"], metrics)
self.assertTrue(metrics["hasError"], metrics)
self.assertTrue(metrics["changeHidden"], metrics)
if mode == "desktop":
self.assertTrue(metrics["dualZone"], metrics)
self.assertLessEqual(metrics["shellBox"]["width"], 1180, metrics)
self.assertTrue(metrics["shellCentered"], metrics)
self.assertFalse(metrics["stacked"], metrics)
self.assertGreaterEqual(metrics["formWidth"], 420, metrics)
self.assertLessEqual(metrics["formWidth"], 460, metrics)
self.assertTrue(metrics["formFitsViewport"], metrics)
elif mode == "compact":
self.assertTrue(metrics["dualZone"], metrics)
self.assertFalse(metrics["stacked"], metrics)
self.assertGreaterEqual(metrics["formWidth"], 360, metrics)
self.assertLessEqual(metrics["formWidth"], 440, metrics)
else:
self.assertTrue(metrics["stacked"], metrics)
self.assertFalse(metrics["dualZone"], metrics)
if __name__ == "__main__":
unittest.main()
+607
View File
@@ -0,0 +1,607 @@
"""B-44 manual record tests: submit idempotency, approve new/link, return,
exception, reverse, deduplication, idempotent replay and concurrency."""
from __future__ import annotations
from decimal import Decimal
import threading
import unittest
from bank_importer import ledger_events, manual_records, matching, subjects
from ledger_helpers import LedgerBase
class SubmitTests(LedgerBase):
def submit(self, **overrides):
params = dict(
company_id=self.company_a,
counterparty_company_id=self.company_b,
occurred_at="2026-01-10T09:00:00",
direction="incoming",
amount="50.00",
currency="CNY",
funding_source="other",
requested_subject="other_receivable",
request_key="mr-key",
actor=self.admin,
summary="归还往来款",
)
params.update(overrides)
return manual_records.submit(self.connection, **params)
def test_submit_creates_pending_decision(self) -> None:
payload = self.submit()
self.assertEqual("pending", payload["state"])
self.assertEqual(Decimal("50.00"), Decimal(payload["amount"]))
self.assertEqual(1, payload["decision_revision"])
self.assertEqual([], payload["candidates"])
def test_submit_idempotent_on_request_key(self) -> None:
first = self.submit()
second = self.submit()
self.assertEqual(first["id"], second["id"])
self.assertTrue(second["idempotent_replay"])
decisions = self.connection.execute(
"SELECT COUNT(*) AS n FROM manual_record_decisions"
).fetchone()["n"]
self.assertEqual(1, decisions)
def test_submit_validations(self) -> None:
cases = (
({"counterparty_company_id": self.company_a}, "不能与本公司相同"),
({"amount": "0"}, "必须大于零"),
({"amount": "abc"}, "十进制"),
({"funding_source": "credit_card"}, "资金来源"),
({"requested_subject": "equity"}, "科目"),
({"request_key": ""}, "request_key"),
({"counterparty_company_id": 9999}, "公司不存在"),
)
for overrides, expected in cases:
with self.subTest(overrides=overrides):
with self.assertRaises(manual_records.ManualInputError) as ctx:
self.submit(**overrides)
self.assertIn(expected, str(ctx.exception))
def test_bank_account_must_belong_to_submitting_company(self) -> None:
with self.assertRaises(manual_records.ManualInputError):
self.submit(
funding_source="approved_bank_account",
bank_account_id=self.account_b["id"],
)
def test_related_source_row_never_modified(self) -> None:
row_id = self.add_row(
self.company_a, own_account="6222000000000001",
cp_account="6222000000000002", expense="100.00",
)
payload = self.submit(related_source_row_id=row_id)
self.assertEqual(row_id, payload["related_source_row_id"])
with self.assertRaises(Exception):
self.connection.execute(
"UPDATE source_rows SET expense = '0' WHERE id = ?", (row_id,)
)
self.connection.rollback()
class DecisionTests(LedgerBase):
def setUp(self) -> None:
super().setUp()
self.record = manual_records.submit(
self.connection,
company_id=self.company_a, counterparty_company_id=self.company_b,
occurred_at="2026-01-10T09:00:00", direction="incoming",
amount="50.00", currency="CNY", funding_source="other",
requested_subject="other_receivable", request_key="mr-1",
actor=self.admin, summary="归还往来款",
)
def decide(self, action, **overrides):
params = dict(
record_id=self.record["id"],
action=action,
reason="管理员审核",
expected_decision_id=self.record["decision_id"],
request_key="dec-key",
actor=self.admin,
)
params.update(overrides)
return manual_records.decide(self.connection, **params)
def test_approve_new_creates_confirmed_event_with_requested_subject(self) -> None:
outcome = self.decide("approve_new")
self.assertEqual("approved", outcome["state"])
event_id = outcome["ledger_event_id"]
revision = ledger_events.current_revision(self.connection, event_id)
self.assertEqual("confirmed", revision["state"])
# incoming 50 from A's perspective: B is the payer, A the payee.
self.assertEqual(self.company_b, revision["payer_company_id"])
self.assertEqual(self.company_a, revision["payee_company_id"])
self.assertEqual(self.company_a, revision["perspective_company_id"])
self.assertEqual("other_receivable", revision["subject_code"])
# Exactly one position impact.
positions = self.connection.execute(
"SELECT * FROM eligible_position_events"
).fetchall()
self.assertEqual(1, len(positions))
def test_approve_new_replay_is_idempotent(self) -> None:
first = self.decide("approve_new")
second = self.decide("approve_new")
self.assertEqual(first["decision_id"], second["decision_id"])
events = self.connection.execute(
"SELECT COUNT(*) AS n FROM ledger_events"
).fetchone()["n"]
self.assertEqual(1, events)
def test_return_and_exception_never_produce_balance(self) -> None:
for action in ("return", "exception"):
with self.subTest(action=action):
fresh = manual_records.submit(
self.connection,
company_id=self.company_a, counterparty_company_id=self.company_b,
occurred_at="2026-01-11T09:00:00", direction="incoming",
amount="10.00", currency="CNY", funding_source="other",
requested_subject="receivable", request_key=f"mr-{action}",
actor=self.admin,
)
outcome = manual_records.decide(
self.connection, fresh["id"], action,
reason="材料不足" if action == "return" else "转入异常",
expected_decision_id=fresh["decision_id"],
request_key=f"dec-{action}", actor=self.admin,
)
expected_state = "returned" if action == "return" else "exception"
self.assertEqual(expected_state, outcome["state"])
self.assertEqual(0, self.connection.execute(
"SELECT COUNT(*) AS n FROM eligible_position_events"
).fetchone()["n"])
def test_reverse_of_approved_record_creates_reversal_event(self) -> None:
approved = self.decide("approve_new")
reversed_outcome = manual_records.decide(
self.connection, self.record["id"], "reverse",
reason="误录,冲销", expected_decision_id=approved["decision_id"],
request_key="dec-rev", actor=self.admin,
)
self.assertEqual("reversed", reversed_outcome["state"])
rows = self.connection.execute(
"SELECT * FROM eligible_position_events ORDER BY ledger_event_id"
).fetchall()
self.assertEqual(2, len(rows))
original, reversal = rows
self.assertEqual("normal", original["posting_kind"])
self.assertEqual("reversal", reversal["posting_kind"])
self.assertEqual(original["payer_company_id"], reversal["payee_company_id"])
self.assertEqual(Decimal(original["amount"]), Decimal(reversal["amount"]))
# Net position is zero.
signed = Decimal(original["amount"]) * (
1 if original["payer_company_id"] == self.company_a else -1
) + Decimal(reversal["amount"]) * (
1 if reversal["payer_company_id"] == self.company_a else -1
)
self.assertEqual(Decimal("0"), signed)
def test_approve_link_to_bank_event_adds_evidence_not_impact(self) -> None:
# Build an eligible bank event and confirm its subject.
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
bank_ledger_id = self.ledger_events()[0]["id"]
subjects.confirm_subject(
self.connection, bank_ledger_id,
perspective_company_id=self.company_a, subject_code="other_receivable",
reason="借款", expected_revision=1, request_key="subj-1",
actor=self.admin,
)
before = len(self.position_events())
outcome = manual_records.decide(
self.connection, self.record["id"], "approve_link",
reason="与银行事件同源", expected_decision_id=self.record["decision_id"],
request_key="dec-link", actor=self.admin,
target_ledger_event_id=bank_ledger_id,
)
self.assertEqual("approved", outcome["state"])
self.assertEqual(bank_ledger_id, outcome["ledger_event_id"])
after = len(self.position_events())
self.assertEqual(before, after)
sources = self.connection.execute(
"SELECT * FROM ledger_event_manual_sources WHERE manual_record_id = ?",
(self.record["id"],),
).fetchall()
self.assertEqual(1, len(sources))
self.assertEqual(bank_ledger_id, sources[0]["ledger_event_id"])
def test_approve_link_requires_target(self) -> None:
with self.assertRaises(manual_records.ManualInputError):
self.decide("approve_link")
def test_reverse_of_linked_record_detaches_claim_only(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
bank_ledger_id = self.ledger_events()[0]["id"]
subjects.confirm_subject(
self.connection, bank_ledger_id,
perspective_company_id=self.company_a, subject_code="other_receivable",
reason="借款", expected_revision=1, request_key="subj-1",
actor=self.admin,
)
approved = manual_records.decide(
self.connection, self.record["id"], "approve_link",
reason="同源", expected_decision_id=self.record["decision_id"],
request_key="dec-link", actor=self.admin,
target_ledger_event_id=bank_ledger_id,
)
reversed_outcome = manual_records.decide(
self.connection, self.record["id"], "reverse",
reason="撤销关联", expected_decision_id=approved["decision_id"],
request_key="dec-unlink", actor=self.admin,
)
self.assertEqual("reversed", reversed_outcome["state"])
# No reversal event is created for a linked claim.
reversals = self.connection.execute(
"SELECT COUNT(*) AS n FROM ledger_event_revisions WHERE posting_kind = 'reversal'"
).fetchone()["n"]
self.assertEqual(0, reversals)
# The bank impact is untouched.
self.assertEqual(1, len(self.position_events()))
def test_stale_expected_decision_conflicts(self) -> None:
approved = self.decide("approve_new")
with self.assertRaises(manual_records.ManualConflictError):
manual_records.decide(
self.connection, self.record["id"], "reverse",
reason="冲销", expected_decision_id=self.record["decision_id"],
request_key="dec-stale", actor=self.admin,
)
def test_returned_record_cannot_be_reversed(self) -> None:
returned = self.decide("return", reason="材料不足")
with self.assertRaises(manual_records.ManualConflictError):
manual_records.decide(
self.connection, self.record["id"], "reverse",
reason="冲销", expected_decision_id=returned["decision_id"],
request_key="dec-x", actor=self.admin,
)
def test_reverse_uses_explicit_effective_date_across_cutoff(self) -> None:
# Reversing with an independent effective date must keep the original
# impact for cutoffs before it and net it to zero only on/after it —
# never rewrite the historical period retroactively.
from bank_importer import positions
approved = self.decide("approve_new")
event_id = approved["ledger_event_id"]
self.assertEqual("2026-01-10", self.current(event_id)["effective_at"][:10])
manual_records.decide(
self.connection, self.record["id"], "reverse",
reason="误录冲销", expected_decision_id=approved["decision_id"],
request_key="dec-rev-date", actor=self.admin,
effective_at="2026-06-15",
)
reversal = self.connection.execute(
"SELECT * FROM ledger_event_revisions WHERE posting_kind = 'reversal'"
).fetchone()
self.assertEqual("2026-06-15", reversal["effective_at"][:10])
before = positions.company_balances(
self.connection, from_="2026-01-01", cutoff="2026-05-31"
)
after = positions.company_balances(
self.connection, from_="2026-01-01", cutoff="2026-07-31"
)
for item in before["items"]:
if item["company_id"] == self.company_a:
self.assertEqual("-50.00", item["result"]["signed_amount"])
for item in after["items"]:
if item["company_id"] == self.company_a:
self.assertEqual("0.00", item["result"]["signed_amount"])
def test_reverse_defaults_to_approval_business_day(self) -> None:
from datetime import datetime, timedelta, timezone
from bank_importer import positions
approved = self.decide("approve_new")
manual_records.decide(
self.connection, self.record["id"], "reverse",
reason="误录冲销", expected_decision_id=approved["decision_id"],
request_key="dec-rev-default", actor=self.admin,
)
today = datetime.now(timezone(timedelta(hours=8))).date().isoformat()
reversal = self.connection.execute(
"SELECT * FROM ledger_event_revisions WHERE posting_kind = 'reversal'"
).fetchone()
self.assertEqual(today, reversal["effective_at"][:10])
# The day before the approval business day keeps the original impact.
before = positions.company_balances(
self.connection, from_="2026-01-01",
cutoff=(datetime.fromisoformat(today) - timedelta(days=1)).date().isoformat(),
)
for item in before["items"]:
if item["company_id"] == self.company_a:
self.assertEqual("-50.00", item["result"]["signed_amount"])
def test_reverse_of_creation_source_with_later_linked_evidence(self) -> None:
# approve_new created the event; a second manual record later links
# onto it. Reversing the creation source must still create an
# equal-amount reversal — the original economic impact must not survive
# just because other evidence was attached later.
approved = self.decide("approve_new")
event_id = approved["ledger_event_id"]
linked = manual_records.submit(
self.connection, company_id=self.company_a,
counterparty_company_id=self.company_b,
occurred_at="2026-01-12T09:00:00", direction="incoming",
amount="50.00", currency="CNY", funding_source="other",
requested_subject="receivable", request_key="mr-link-ev",
actor=self.admin,
)
manual_records.decide(
self.connection, linked["id"], "approve_link",
reason="同源补充证据", expected_decision_id=linked["decision_id"],
request_key="dec-link-ev", actor=self.admin,
target_ledger_event_id=event_id,
)
self.assertEqual(1, len(self.position_events()))
reversed_outcome = manual_records.decide(
self.connection, self.record["id"], "reverse",
reason="误录冲销", expected_decision_id=approved["decision_id"],
request_key="dec-rev-create", actor=self.admin,
)
self.assertEqual("reversed", reversed_outcome["state"])
rows = self.connection.execute(
"SELECT * FROM eligible_position_events ORDER BY ledger_event_id"
).fetchall()
self.assertEqual(2, len(rows))
original, reversal = rows
self.assertEqual("normal", original["posting_kind"])
self.assertEqual("reversal", reversal["posting_kind"])
self.assertEqual(Decimal(original["amount"]), Decimal(reversal["amount"]))
# The later linked evidence stays attached to the event.
links = self.connection.execute(
"SELECT * FROM ledger_event_manual_sources ORDER BY manual_record_id"
).fetchall()
self.assertEqual(2, len(links))
def test_reverse_of_linked_record_after_creation_reversal_detaches_only(self) -> None:
# M1 creates the event, M2 links; after M1's reversal created the
# offset, reversing the linked M2 must only detach, never add a second
# reversal event.
approved = self.decide("approve_new")
event_id = approved["ledger_event_id"]
linked = manual_records.submit(
self.connection, company_id=self.company_a,
counterparty_company_id=self.company_b,
occurred_at="2026-01-12T09:00:00", direction="incoming",
amount="50.00", currency="CNY", funding_source="other",
requested_subject="receivable", request_key="mr-link-ev2",
actor=self.admin,
)
linked_approved = manual_records.decide(
self.connection, linked["id"], "approve_link",
reason="同源补充证据", expected_decision_id=linked["decision_id"],
request_key="dec-link-ev2", actor=self.admin,
target_ledger_event_id=event_id,
)
manual_records.decide(
self.connection, self.record["id"], "reverse",
reason="误录冲销", expected_decision_id=approved["decision_id"],
request_key="dec-rev-create2", actor=self.admin,
)
manual_records.decide(
self.connection, linked["id"], "reverse",
reason="撤销关联证据", expected_decision_id=linked_approved["decision_id"],
request_key="dec-rev-link2", actor=self.admin,
)
reversals = self.connection.execute(
"SELECT COUNT(*) AS n FROM ledger_event_revisions WHERE posting_kind = 'reversal'"
).fetchone()["n"]
self.assertEqual(1, reversals)
links = self.connection.execute(
"SELECT * FROM ledger_event_manual_sources"
).fetchall()
self.assertEqual(1, len(links))
class DedupAndConcurrencyTests(LedgerBase):
def test_duplicate_manual_record_pairing_candidate(self) -> None:
manual_records.submit(
self.connection, company_id=self.company_a,
counterparty_company_id=self.company_b,
occurred_at="2026-01-10T09:00:00", direction="outgoing",
amount="40.00", currency="CNY", funding_source="other",
requested_subject="other_receivable", request_key="mr-a",
actor=self.admin,
)
other = manual_records.submit(
self.connection, company_id=self.company_b,
counterparty_company_id=self.company_a,
occurred_at="2026-01-10T10:00:00", direction="incoming",
amount="40.00", currency="CNY", funding_source="other",
requested_subject="other_payable", request_key="mr-b",
actor=self.admin,
)
candidates = manual_records.find_candidates(self.connection, other["id"])
self.assertEqual(1, len(candidates))
self.assertEqual("manual_record", candidates[0]["kind"])
def test_duplicate_bank_event_candidate_hints_link(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
record = manual_records.submit(
self.connection, company_id=self.company_a,
counterparty_company_id=self.company_b,
occurred_at="2026-01-05T12:00:00", direction="incoming",
amount="100.00", currency="CNY", funding_source="other",
requested_subject="other_receivable", request_key="mr-c",
actor=self.admin,
)
candidates = manual_records.find_candidates(self.connection, record["id"])
self.assertTrue(any(c["kind"] == "bank_event" for c in candidates))
def test_concurrent_approve_new_counts_once(self) -> None:
record = manual_records.submit(
self.connection, company_id=self.company_a,
counterparty_company_id=self.company_b,
occurred_at="2026-01-10T09:00:00", direction="incoming",
amount="50.00", currency="CNY", funding_source="other",
requested_subject="receivable", request_key="mr-conc",
actor=self.admin,
)
errors: list[Exception] = []
def worker(thread_key: str) -> None:
import sqlite3
try:
connection = sqlite3.connect(self.db_path, timeout=30)
connection.row_factory = sqlite3.Row
try:
manual_records.decide(
connection, record["id"], "approve_new",
reason="并发审批", expected_decision_id=record["decision_id"],
request_key=f"dec-{thread_key}", actor=self.admin,
)
finally:
connection.close()
except Exception as exc: # pragma: no cover
errors.append(exc)
threads = [
threading.Thread(target=worker, args=(f"t{i}",)) for i in range(4)
]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
# Two connections can still double-commit, so the UNIQUE manual source
# claim must make the second write fail or be a no-op; either way only
# one ledger event may exist.
events = self.connection.execute(
"SELECT COUNT(*) AS n FROM ledger_events"
).fetchone()["n"]
self.assertEqual(1, events)
def test_concurrent_submit_same_request_key_is_idempotent(self) -> None:
import sqlite3
results: list[dict] = []
errors: list[Exception] = []
def worker(thread_key: str) -> None:
try:
connection = sqlite3.connect(self.db_path, timeout=30)
connection.row_factory = sqlite3.Row
try:
payload = manual_records.submit(
connection, company_id=self.company_a,
counterparty_company_id=self.company_b,
occurred_at="2026-01-10T09:00:00", direction="incoming",
amount="50.00", currency="CNY", funding_source="other",
requested_subject="receivable",
request_key="mr-conc-submit", actor=self.admin,
)
results.append(payload)
finally:
connection.close()
except Exception as exc: # pragma: no cover
errors.append(exc)
threads = [
threading.Thread(target=worker, args=(f"t{i}",)) for i in range(4)
]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
self.assertEqual([], errors)
rows = self.connection.execute(
"SELECT COUNT(*) AS n FROM manual_records"
).fetchone()["n"]
self.assertEqual(1, rows)
self.assertEqual(1, len({payload["id"] for payload in results}))
decisions = self.connection.execute(
"SELECT COUNT(*) AS n FROM manual_record_decisions"
).fetchone()["n"]
self.assertEqual(1, decisions)
def test_concurrent_confirm_subject_same_key_creates_single_revision(self) -> None:
import sqlite3
self.pair(self.company_a, self.company_b, "100.00")
from bank_importer import ledger_events
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
outcomes: list[dict] = []
errors: list[Exception] = []
def worker(thread_key: str) -> None:
try:
connection = sqlite3.connect(self.db_path, timeout=30)
connection.row_factory = sqlite3.Row
try:
payload = subjects.confirm_subject(
connection, event_id,
perspective_company_id=self.company_a,
subject_code="other_receivable", reason="并发确认",
expected_revision=1, request_key="conc-subj-key",
actor=self.admin,
)
outcomes.append(payload)
finally:
connection.close()
except Exception as exc: # pragma: no cover
errors.append(exc)
threads = [
threading.Thread(target=worker, args=(f"t{i}",)) for i in range(4)
]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
self.assertEqual([], errors)
confirmed = self.connection.execute(
"SELECT COUNT(*) AS n FROM ledger_event_revisions WHERE state = 'confirmed'"
).fetchone()["n"]
self.assertEqual(1, confirmed)
self.assertEqual(1, len({payload["revision_id"] for payload in outcomes}))
def test_repeated_confirm_same_key_appends_no_noise_revision(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
from bank_importer import ledger_events
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
first = subjects.confirm_subject(
self.connection, event_id,
perspective_company_id=self.company_a, subject_code="other_receivable",
reason="确认", expected_revision=1, request_key="noise-key",
actor=self.admin,
)
second = subjects.confirm_subject(
self.connection, event_id,
perspective_company_id=self.company_a, subject_code="other_receivable",
reason="重复确认", expected_revision=1, request_key="noise-key",
actor=self.admin,
)
self.assertEqual(first["revision_id"], second["revision_id"])
revisions = self.connection.execute(
"SELECT COUNT(*) AS n FROM ledger_event_revisions WHERE ledger_event_id = ?",
(event_id,),
).fetchone()["n"]
# One pending_subject + one confirmed; the replay appended nothing.
self.assertEqual(2, revisions)
confirmed = self.connection.execute(
"SELECT COUNT(*) AS n FROM ledger_event_revisions "
"WHERE ledger_event_id = ? AND state = 'confirmed'",
(event_id,),
).fetchone()["n"]
self.assertEqual(1, confirmed)
if __name__ == "__main__":
unittest.main()
+105 -5
View File
@@ -41,7 +41,7 @@ class PersistenceTestCase(unittest.TestCase):
class MigrationTests(PersistenceTestCase):
def test_migrate_creates_schema_and_is_idempotent(self) -> None:
first = applied_versions(self.connection)
self.assertEqual([1, 2, 3, 4, 5], first)
self.assertEqual([1, 2, 3, 4, 5, 6], first)
self.assertEqual([], migrate(self.connection))
self.assertEqual(first, applied_versions(self.connection))
tables = {
@@ -73,19 +73,28 @@ class MigrationTests(PersistenceTestCase):
"transfer_match_candidates",
"current_transfer_decisions",
"transfer_observation_claims",
"manual_records",
"manual_record_decisions",
"current_manual_record_decisions",
"ledger_events",
"ledger_event_revisions",
"current_ledger_event_revisions",
"ledger_event_bank_sources",
"ledger_event_manual_sources",
"ledger_subject_suggestions",
"schema_migrations",
):
self.assertIn(table, tables)
def test_rollback_removes_schema_and_forward_rebuilds_it(self) -> None:
self.assertEqual([5, 4, 3, 2, 1], rollback(self.connection, 0))
self.assertEqual([6, 5, 4, 3, 2, 1], rollback(self.connection, 0))
self.assertEqual([], applied_versions(self.connection))
remaining = self.connection.execute(
"SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'source_rows'"
).fetchone()
self.assertIsNone(remaining)
self.assertEqual([1, 2, 3, 4, 5], migrate(self.connection))
self.assertEqual([1, 2, 3, 4, 5], applied_versions(self.connection))
self.assertEqual([1, 2, 3, 4, 5, 6], migrate(self.connection))
self.assertEqual([1, 2, 3, 4, 5, 6], applied_versions(self.connection))
def test_rollback_to_4_keeps_bank_evidence_and_drops_event_layer(self) -> None:
self.import_sample()
@@ -93,7 +102,7 @@ class MigrationTests(PersistenceTestCase):
"SELECT COUNT(*) AS n FROM source_rows"
).fetchone()["n"]
self.assertGreater(row_count, 0)
self.assertEqual([5], rollback(self.connection, 4))
self.assertEqual([6, 5], rollback(self.connection, 4))
# The pre-migration evidence and schema are untouched.
self.assertEqual(
row_count,
@@ -109,6 +118,10 @@ class MigrationTests(PersistenceTestCase):
"SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'transfer_match_decisions'"
).fetchone()
self.assertIsNone(remaining)
ledger_remaining = self.connection.execute(
"SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'ledger_event_revisions'"
).fetchone()
self.assertIsNone(ledger_remaining)
def test_event_layer_views_exist_after_migration(self) -> None:
view = self.connection.execute(
@@ -116,6 +129,93 @@ class MigrationTests(PersistenceTestCase):
).fetchone()
self.assertIsNotNone(view)
def test_ledger_layer_views_exist_after_migration(self) -> None:
view = self.connection.execute(
"SELECT name FROM sqlite_master WHERE type = 'view' AND name = 'eligible_position_events'"
).fetchone()
self.assertIsNotNone(view)
table = self.connection.execute(
"SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'ledger_event_revisions'"
).fetchone()
self.assertIsNotNone(table)
def test_ledger_revision_log_is_immutable(self) -> None:
with self.connection:
self.connection.execute(
"INSERT INTO companies (name, created_at, updated_at) VALUES ('甲公司', ?, ?)",
(utc_now(), utc_now()),
)
self.connection.execute(
"INSERT INTO companies (name, created_at, updated_at) VALUES ('乙公司', ?, ?)",
(utc_now(), utc_now()),
)
company_a, company_b = [
row["id"]
for row in self.connection.execute("SELECT id FROM companies ORDER BY id")
]
self.connection.execute(
"INSERT INTO ledger_events (lifecycle, created_at) VALUES ('active', ?)",
(utc_now(),),
)
event_id = self.connection.execute(
"SELECT id FROM ledger_events LIMIT 1"
).fetchone()["id"]
self.connection.execute(
"""
INSERT INTO ledger_event_revisions (
ledger_event_id, revision, state, effective_at, amount,
amount_scale, currency, payer_company_id, payee_company_id,
source_kind, posting_kind, created_at
) VALUES (?, 1, 'pending_subject', '2026-01-01T00:00:00', '1.00', 2,
'CNY', ?, ?, 'bank', 'normal', ?)
""",
(event_id, company_a, company_b, utc_now()),
)
for statement in (
"UPDATE ledger_event_revisions SET amount = '9.99'",
"DELETE FROM ledger_event_revisions",
"UPDATE ledger_events SET created_at = '2020-01-01T00:00:00'",
"DELETE FROM ledger_events",
):
with self.subTest(statement=statement):
with self.assertRaises(sqlite3.IntegrityError):
self.connection.execute(statement)
self.connection.rollback()
def test_manual_record_facts_are_immutable(self) -> None:
with self.connection:
self.connection.execute(
"INSERT INTO companies (name, created_at, updated_at) VALUES ('甲公司', ?, ?)",
(utc_now(), utc_now()),
)
self.connection.execute(
"INSERT INTO companies (name, created_at, updated_at) VALUES ('乙公司', ?, ?)",
(utc_now(), utc_now()),
)
company_a, company_b = [
row["id"]
for row in self.connection.execute("SELECT id FROM companies ORDER BY id")
]
self.connection.execute(
"""
INSERT INTO manual_records (
company_id, counterparty_company_id, occurred_at, direction,
amount, amount_scale, currency, funding_source, requested_subject,
request_key, created_at
) VALUES (?, ?, '2026-01-01T00:00:00', 'incoming', '1.00', 2,
'CNY', 'other', 'receivable', 'k1', ?)
""",
(company_a, company_b, utc_now()),
)
for statement in (
"UPDATE manual_records SET amount = '9.99'",
"DELETE FROM manual_records",
):
with self.subTest(statement=statement):
with self.assertRaises(sqlite3.IntegrityError):
self.connection.execute(statement)
self.connection.rollback()
def test_decision_log_immutability_triggers(self) -> None:
self.import_sample()
source_row_id = self.connection.execute(
+430
View File
@@ -0,0 +1,430 @@
"""B-44 position aggregation tests: conservation, repayments, reversals across
cutoffs, currency isolation, Decimal precision, unresolved gross, normal
balance exceptions and keyset pagination."""
from __future__ import annotations
from decimal import Decimal, getcontext
import unittest
from bank_importer import ledger_events, manual_records, positions, subjects
from ledger_helpers import LedgerBase
def confirm(connection, event_id: int, perspective: int, subject: str, key: str, actor):
return subjects.confirm_subject(
connection, event_id,
perspective_company_id=perspective, subject_code=subject,
reason="审核确认", expected_revision=1, request_key=key, actor=actor,
)
class ConservationTests(LedgerBase):
def setUp(self) -> None:
super().setUp()
getcontext().prec = 50
def _confirm_all(self, subject: str = "other_receivable"):
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
for event in self.ledger_events():
revision = self.current(event["id"])
confirm(
self.connection, event["id"],
revision["payer_company_id"], subject, f"subj-{event['id']}",
self.admin,
)
def test_both_perspectives_mirror_exactly(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
self.pair(self.company_a, self.company_b, "30.00", at="2026-01-10T10:00:00")
self.pair(self.company_a, self.company_b, "70.00", at="2026-02-05T10:00:00")
self._confirm_all()
pair = positions.pair_detail(
self.connection, self.company_a, self.company_b,
from_="2026-01-01", cutoff="2026-07-31",
)
item = pair["items"][0]
self.assertTrue(item["conservation"]["opposite"])
self.assertTrue(item["conservation"]["abs_equal"])
self.assertEqual(Decimal("200.00"), Decimal(item["a"]["result"]["signed_amount"]))
self.assertEqual(Decimal("-200.00"), Decimal(item["b"]["result"]["signed_amount"]))
def test_repayment_reduces_the_original_balance(self) -> None:
# A lends B 100 (A pays) then B repays 40 (B pays).
self.pair(self.company_a, self.company_b, "100.00")
self.pair(self.company_b, self.company_a, "40.00", at="2026-03-01T10:00:00",
summary="还款", purpose="归还借款")
self._confirm_all()
balances = positions.company_balances(
self.connection, from_="2026-01-01", cutoff="2026-07-31"
)
for item in balances["items"]:
if item["company_id"] == self.company_a:
self.assertEqual(Decimal("60.00"), Decimal(item["result"]["signed_amount"]))
self.assertEqual(Decimal("100.00"), Decimal(item["period"]["debit"]))
self.assertEqual(Decimal("40.00"), Decimal(item["period"]["credit"]))
else:
self.assertEqual(Decimal("-60.00"), Decimal(item["result"]["signed_amount"]))
self.assertEqual(Decimal("40.00"), Decimal(item["period"]["debit"]))
self.assertEqual(Decimal("100.00"), Decimal(item["period"]["credit"]))
def test_over_repayment_flags_normal_balance_exception(self) -> None:
# A lends B 100, then B repays 150 while A books the repayment against
# its receivable: A's receivable and B's payable both go negative.
self.pair(self.company_a, self.company_b, "100.00")
self.pair(self.company_b, self.company_a, "150.00", at="2026-03-01T10:00:00",
summary="还款", purpose="归还借款")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
for event in self.ledger_events():
revision = self.current(event["id"])
# Book every event from A's perspective on the receivable side so
# the over-repayment produces abnormal balances.
confirm(
self.connection, event["id"], self.company_a,
"other_receivable", f"k-{event['id']}", self.admin,
)
pair = positions.pair_detail(
self.connection, self.company_a, self.company_b,
from_="2026-01-01", cutoff="2026-07-31",
)
item = pair["items"][0]
self.assertTrue(item["normal_balance_exception"])
self.assertIn("other_receivable", item["normal_balance_exception"])
self.assertIn("other_payable", item["normal_balance_exception"])
class CutoffAndReversalTests(LedgerBase):
def _loan(self) -> int:
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
confirm(self.connection, event_id, self.company_a, "other_receivable", "k1", self.admin)
return event_id
def test_cutoff_before_reversal_keeps_original_impact(self) -> None:
event_id = self._loan()
# Reversal dated after the original event.
ledger_events.create_reversal(
self.connection, event_id, source_kind="adjustment",
effective_at="2026-06-15T00:00:00", reason="冲销", actor=self.admin,
)
before = positions.company_balances(
self.connection, from_="2026-01-01", cutoff="2026-05-31"
)
after = positions.company_balances(
self.connection, from_="2026-01-01", cutoff="2026-07-31"
)
for item in before["items"]:
if item["company_id"] == self.company_a:
self.assertEqual(Decimal("100.00"), Decimal(item["result"]["signed_amount"]))
for item in after["items"]:
if item["company_id"] == self.company_a:
self.assertEqual(Decimal("0"), Decimal(item["result"]["signed_amount"]))
def test_from_gt_cutoff_is_rejected(self) -> None:
with self.assertRaises(positions.PositionInputError):
positions.validate_window("2026-08-01", "2026-07-31")
def test_events_before_from_are_excluded_from_period(self) -> None:
self._loan()
balances = positions.company_balances(
self.connection, from_="2026-06-01", cutoff="2026-07-31"
)
self.assertEqual([], balances["items"])
def test_reversal_uses_approved_effective_date(self) -> None:
event_id = self._loan()
ledger_events.create_reversal(
self.connection, event_id, source_kind="adjustment",
effective_at="2026-08-01T00:00:00", reason="未来冲销", actor=self.admin,
)
balances = positions.company_balances(
self.connection, from_="2026-01-01", cutoff="2026-07-31"
)
for item in balances["items"]:
if item["company_id"] == self.company_a:
# The future reversal does not rewrite the earlier cutoff.
self.assertEqual(Decimal("100.00"), Decimal(item["result"]["signed_amount"]))
class CurrencyAndPrecisionTests(LedgerBase):
def test_currencies_never_mix(self) -> None:
self.pair(self.company_a, self.company_b, "100.00", currency="CNY")
self.pair(self.company_a, self.company_b, "50.00", at="2026-01-20T10:00:00",
currency="USD")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
events = self.ledger_events()
for event in events:
revision = self.current(event["id"])
confirm(self.connection, event["id"], revision["payer_company_id"],
"other_receivable", f"k-{event['id']}", self.admin)
balances = positions.company_balances(
self.connection, from_="2026-01-01", cutoff="2026-07-31"
)
by_company = {}
for item in balances["items"]:
by_company.setdefault(item["company_id"], {})[item["currency"]] = item["result"]
self.assertEqual(Decimal("100.00"), Decimal(by_company[self.company_a]["CNY"]["signed_amount"]))
self.assertEqual(Decimal("50.00"), Decimal(by_company[self.company_a]["USD"]["signed_amount"]))
def test_decimal_precision_is_preserved(self) -> None:
self.pair(self.company_a, self.company_b, "0.01")
self.pair(self.company_a, self.company_b, "12345.6789", at="2026-01-11T10:00:00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
for event in self.ledger_events():
revision = self.current(event["id"])
confirm(self.connection, event["id"], revision["payer_company_id"],
"other_receivable", f"k-{event['id']}", self.admin)
balances = positions.company_balances(
self.connection, from_="2026-01-01", cutoff="2026-07-31"
)
for item in balances["items"]:
if item["company_id"] == self.company_a:
self.assertEqual(
Decimal("12345.6889"), Decimal(item["result"]["signed_amount"])
)
# The stored string keeps every decimal place.
self.assertIn("12345.6889", item["result"]["signed_amount"])
def test_manual_record_preserves_amount_scale(self) -> None:
record = manual_records.submit(
self.connection, company_id=self.company_a,
counterparty_company_id=self.company_b,
occurred_at="2026-02-01T09:00:00", direction="incoming",
amount="12.3400", currency="CNY", funding_source="other",
requested_subject="receivable", request_key="mr-prec",
actor=self.admin,
)
manual_records.decide(
self.connection, record["id"], "approve_new",
reason="确认", expected_decision_id=record["decision_id"],
request_key="dec-prec", actor=self.admin,
)
stored = self.connection.execute(
"SELECT amount, amount_scale FROM eligible_position_events"
).fetchone()
self.assertEqual("12.3400", stored["amount"])
self.assertEqual(4, stored["amount_scale"])
class UnresolvedTests(LedgerBase):
def test_unresolved_gross_never_nets(self) -> None:
# Two pending_subject events of opposite economic signs must sum their
# absolute values, never cancel.
self.pair(self.company_a, self.company_b, "100.00")
self.pair(self.company_a, self.company_b, "40.00", at="2026-01-20T10:00:00",
summary="还款", purpose="归还借款")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
unresolved = positions.unresolved_for_company(
self.connection, self.company_a, "2026-07-31"
)
self.assertEqual(2, unresolved["count"])
self.assertEqual(
Decimal("140.00"), Decimal(unresolved["gross_amount"])
)
self.assertEqual(
Decimal("140.00"),
Decimal(unresolved["by_reason"]["subject_review"]["gross_amount"]),
)
def test_unmatched_single_reason_bucket(self) -> None:
self.add_row(
self.company_a, own_account="6222000000000001",
cp_account="6222000000000002", expense="88.00",
)
from bank_importer import matching
matching.reconcile_rows(self.connection, [1])
unresolved = positions.unresolved_for_company(
self.connection, self.company_a, "2026-07-31"
)
self.assertEqual(1, unresolved["count"])
self.assertEqual(
Decimal("88.00"),
Decimal(unresolved["by_reason"]["unmatched_single"]["gross_amount"]),
)
def test_pending_manual_only_counts_for_submitting_company(self) -> None:
manual_records.submit(
self.connection, company_id=self.company_a,
counterparty_company_id=self.company_b,
occurred_at="2026-01-10T09:00:00", direction="incoming",
amount="50.00", currency="CNY", funding_source="other",
requested_subject="receivable", request_key="mr-a",
actor=self.admin,
)
a_unresolved = positions.unresolved_for_company(
self.connection, self.company_a, "2026-07-31"
)
b_unresolved = positions.unresolved_for_company(
self.connection, self.company_b, "2026-07-31"
)
self.assertEqual(1, a_unresolved["count"])
self.assertEqual(
Decimal("50.00"), Decimal(a_unresolved["by_reason"]["manual_pending"]["gross_amount"])
)
# The counterparty never sees the unapproved declaration.
self.assertEqual(0, b_unresolved["count"])
class PaginationTests(LedgerBase):
def test_subject_filter_matches_stored_and_mirror_subjects(self) -> None:
# A books "receivable" from its own perspective; B sees the mirror
# "payable". Filtering by B's mirror subject must still find the event.
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
confirm(self.connection, event_id, self.company_a, "receivable", "k-mirror",
self.admin)
stored = positions.list_events(
self.connection, from_="2026-01-01", cutoff="2026-07-31",
subject="receivable",
)
self.assertTrue(any(
item["ledger_event_id"] == event_id for item in stored["items"]
))
# Company B filters by its own perspective: the stored "receivable" is
# the mirror of "payable", so it must be included.
mirror = positions.list_events(
self.connection, from_="2026-01-01", cutoff="2026-07-31",
subject="payable", viewer_company_id=self.company_b,
)
self.assertTrue(any(
item["ledger_event_id"] == event_id for item in mirror["items"]
), mirror)
for item in mirror["items"]:
if item["ledger_event_id"] == event_id:
self.assertEqual("payable", item["own_subject"])
def test_subject_filter_without_match_returns_empty(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = self.ledger_events()[0]["id"]
confirm(self.connection, event_id, self.company_a, "receivable", "k-none",
self.admin)
other = positions.list_events(
self.connection, from_="2026-01-01", cutoff="2026-07-31",
subject="other_receivable",
)
self.assertEqual([], other["items"])
def test_pagination_is_stable_and_complete(self) -> None:
for index in range(7):
at = f"2026-01-{(index % 28) + 1:02d}T10:00:00"
self.pair(self.company_a, self.company_b, f"{index + 1}.00", at=at)
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
for event in self.ledger_events():
revision = self.current(event["id"])
confirm(self.connection, event["id"], revision["payer_company_id"],
"other_receivable", f"k-{event['id']}", self.admin)
seen: list[int] = []
cursor = None
pages = 0
while True:
page = positions.list_events(
self.connection, from_="2026-01-01", cutoff="2026-07-31",
limit=3, cursor=cursor,
)
pages += 1
seen.extend(item["ledger_event_id"] for item in page["items"])
cursor = page["next_cursor"]
if not page["has_more"]:
break
self.assertEqual(7, len(seen))
self.assertEqual(len(set(seen)), len(seen))
self.assertGreater(pages, 2)
def test_directory_pagination_complete(self) -> None:
for index in range(6):
at = f"2026-01-{(index % 28) + 1:02d}T10:00:00"
self.pair(self.company_a, self.company_b, f"{index + 1}.00", at=at)
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
for event in self.ledger_events():
revision = self.current(event["id"])
confirm(self.connection, event["id"], revision["payer_company_id"],
"other_receivable", f"k-{event['id']}", self.admin)
seen: set[tuple[int, str]] = set()
cursor = None
while True:
page = positions.company_balances(
self.connection, from_="2026-01-01", cutoff="2026-07-31",
limit=1, cursor=cursor,
)
seen.update((item["company_id"], item["currency"]) for item in page["items"])
cursor = page["next_cursor"]
if not page["has_more"]:
break
self.assertEqual(
{(self.company_a, "CNY"), (self.company_b, "CNY")}, seen
)
def test_event_payload_account_chip_and_repayment_flag(self) -> None:
self.pair(
self.company_a, self.company_b, "40.00",
summary="归还往来款", purpose="还款",
)
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = int(self.ledger_events()[0]["id"])
detail = positions.event_detail(self.connection, event_id)
event = detail["event"]
self.assertEqual("visible", event["payer_account"]["visibility"])
self.assertEqual("测试 0001", event["payer_account"]["label"])
self.assertEqual("visible", event["payee_account"]["visibility"])
self.assertEqual("测试 0002", event["payee_account"]["label"])
self.assertEqual("归还往来款", event["summary"])
self.assertTrue(event["is_repayment"])
company_view = positions.event_payload(
self.connection,
self.connection.execute(
positions._DETAIL_SELECT + " WHERE p.ledger_event_id = ?",
(event_id,),
).fetchone(),
viewer_company_id=self.company_a,
)
self.assertEqual("visible", company_view["payer_account"]["visibility"])
self.assertEqual("masked", company_view["payee_account"]["visibility"])
self.assertEqual("按对方授权不可见", company_view["payee_account"]["label"])
def test_park_subject_exception_hides_from_review_queue(self) -> None:
self.pair(self.company_a, self.company_b, "100.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = int(self.ledger_events()[0]["id"])
current = self.current(event_id)
subjects.park_subject(
self.connection, event_id,
disposition="exception", reason="转异常核查",
expected_revision=current["id"], request_key="park-exc-1",
actor=self.admin,
)
queue = positions.subject_review_queue(
self.connection, from_="2026-01-01", cutoff="2026-07-31",
)
self.assertFalse(
any(item["ledger_event_id"] == event_id for item in queue["items"])
)
revision = self.current(event_id)
self.assertEqual("pending_subject", revision["state"])
def test_park_subject_return_stays_in_review_queue(self) -> None:
self.pair(self.company_a, self.company_b, "80.00")
ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
event_id = int(self.ledger_events()[0]["id"])
current = self.current(event_id)
subjects.park_subject(
self.connection, event_id,
disposition="return", reason="退回补充摘要",
expected_revision=current["id"], request_key="park-ret-1",
actor=self.admin,
)
queue = positions.subject_review_queue(
self.connection, from_="2026-01-01", cutoff="2026-07-31",
)
self.assertTrue(
any(item["ledger_event_id"] == event_id for item in queue["items"])
)
if __name__ == "__main__":
unittest.main()
+594
View File
@@ -0,0 +1,594 @@
"""HTTP integration tests for the B-44 intercompany position APIs.
Covers the admin balances/pair/events/evidence/subject-review/manual-record
endpoints and the company-side scoped reads with masked evidence, plus
cross-tenant 404s and company-forbidden admin writes. Every test method spins
up its own server with a fresh database so subject confirmations and matched
fixtures never leak across tests.
"""
from __future__ import annotations
import io
import json
import os
from pathlib import Path
import tempfile
import threading
import unittest
from openpyxl import Workbook
from bank_importer.db import connect, migrate
import server
from test_server_auth import Client, as_json
BOOTSTRAP_PASSWORD = "BootAdmin123"
ADMIN_PASSWORD = "AdminPass123"
CASHIER_PASSWORD = "Cashier123"
CCB_HEADER = [
"客户账号", "账户名称", "交易时间", "借方发生额(支取)", "贷方发生额(收入)",
"余额", "币种", "对方户名", "对方账号", "对方开户机构", "摘要", "备注",
]
ACCOUNT_A = "6222000000000001"
ACCOUNT_B = "6222000000000002"
def workbook_bytes(rows) -> bytes:
workbook = Workbook()
sheet = workbook.active
sheet.title = "正常流水"
sheet.append(CCB_HEADER)
for row in rows:
sheet.append(row)
buffer = io.BytesIO()
workbook.save(buffer)
return buffer.getvalue()
def outgoing(own: str, cp: str, amount: str, at: str = "2026-01-05 10:00:00",
currency: str = "RMB"):
return [own, "测试公司", at, amount, "", "50000.00", currency, "对方", cp, "某银行", "借款", ""]
def incoming(own: str, cp: str, amount: str, at: str = "2026-01-05 11:00:00",
currency: str = "RMB"):
return [own, "测试公司", at, "", amount, "50000.00", currency, "对方", cp, "某银行", "借款", ""]
class IntercompanyApiTests(unittest.TestCase):
def setUp(self) -> None:
self.temp_dir = tempfile.TemporaryDirectory()
self.addCleanup(self.temp_dir.cleanup)
root = Path(self.temp_dir.name)
self.db_path = root / "app.db"
self.storage = root / "files"
self.initial_passwords: dict[str, str] = {}
self._old_db_path = server.DB_PATH
self._old_storage = server.STORAGE_DIR
server.DB_PATH = self.db_path
server.STORAGE_DIR = self.storage
os.environ["APP_BOOTSTRAP_ADMIN_PASSWORD"] = BOOTSTRAP_PASSWORD
connection = connect(self.db_path)
migrate(connection)
assert server.ensure_bootstrap_admin(connection) is None
connection.close()
class QuietHandler(server.AppHandler):
def log_message(self, *args) -> None:
pass
self.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler)
self.port = self.httpd.server_address[1]
self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
self.thread.start()
self.admin = Client("127.0.0.1", self.port)
status, _, data = self.admin.post_json(
"/api/login",
{"username": "group-admin", "password": BOOTSTRAP_PASSWORD, "portal": "admin"},
)
assert status == 200, data
status, _, data = self.admin.post_json(
"/api/password/change",
{"old_password": BOOTSTRAP_PASSWORD, "new_password": ADMIN_PASSWORD},
)
assert status == 200, data
self.company_a = self._create_company("甲公司", "cashier-a")
self.company_b = self._create_company("乙公司", "cashier-b")
self.company_c = self._create_company("丙公司", "cashier-c")
self.cashier_a = self._login_company("cashier-a", self.company_a)
self.cashier_b = self._login_company("cashier-b", self.company_b)
self.cashier_c = self._login_company("cashier-c", self.company_c)
self._approve_account(self.company_a, ACCOUNT_A)
self._approve_account(self.company_b, ACCOUNT_B)
def tearDown(self) -> None:
self.httpd.shutdown()
self.httpd.server_close()
server.DB_PATH = self._old_db_path
server.STORAGE_DIR = self._old_storage
os.environ.pop("APP_BOOTSTRAP_ADMIN_PASSWORD", None)
def _create_company(self, name: str, username: str) -> int:
status, _, data = self.admin.post_json(
"/api/admin/companies", {"name": name, "username": username}
)
assert status == 200, data
self.initial_passwords.setdefault(username, as_json(data)["initial_password"])
return as_json(data)["company_id"]
def _login_company(self, username: str, company_id: int) -> Client:
client = Client("127.0.0.1", self.port)
initial = self.initial_passwords[username]
status, _, data = client.post_json(
"/api/login", {"username": username, "password": initial, "portal": "company"}
)
assert status == 200, data
status, _, data = client.post_json(
"/api/password/change",
{"old_password": initial, "new_password": CASHIER_PASSWORD},
)
assert status == 200, data
return client
def _approve_account(self, company_id: int, number: str) -> None:
client = {self.company_a: self.cashier_a, self.company_b: self.cashier_b}[company_id]
status, _, data = client.post_json(
"/api/company/accounts",
{"bank_name": "中信银行", "account_type": "基本户",
"account_number": number, "start_date": "2026-01-01"},
)
assert status == 200, data
account_id = as_json(data)["account"]["id"]
status, _, data = self.admin.post_json(
f"/api/admin/accounts/{account_id}/review",
{"decision": "approve", "reason": "测试启用", "effective_from": "2026-01-01"},
)
assert status == 200, data
def _upload_and_confirm(self, client, company_id: int, rows) -> int:
content = workbook_bytes(rows)
status, _, data = self.admin.post_multipart(
"/api/parse", {"company_id": str(company_id)}, "账单.xlsx", content
)
assert status == 200, data
batch_id = as_json(data)["batch_id"]
status, _, data = client.get(f"/api/batches/{batch_id}/sheets")
names = [s["sheet_name"] for s in as_json(data)["sheets"] if s["outcome"] == "parsed"]
status, _, data = client.post_json(
f"/api/batches/{batch_id}/confirm", {"sheets": names}
)
assert status == 200, data
return batch_id
def _fresh_pair(self, amount: str = "100.00", at: str = "2026-01-05 10:00:00",
currency: str = "RMB") -> dict:
"""Upload+confirm a new A<->B pair; returns the pending review item."""
self._upload_and_confirm(
self.cashier_a, self.company_a,
[outgoing(ACCOUNT_A, ACCOUNT_B, amount, at, currency)],
)
self._upload_and_confirm(
self.cashier_b, self.company_b,
[incoming(ACCOUNT_B, ACCOUNT_A, amount, at.replace("10:", "11:"), currency)],
)
status, _, data = self.admin.get(
"/api/admin/subject-reviews?from=2026-01-01&cutoff=2026-12-31"
)
self.assertEqual(200, status, data)
items = as_json(data)["items"]
pending = [item for item in items if item["amount"] == amount]
self.assertEqual(1, len(pending), data)
return pending[0]
def _confirm(self, ledger_event_id: int) -> dict:
status, _, data = self.admin.get(
f"/api/admin/intercompany/events/{ledger_event_id}"
)
self.assertEqual(200, status, data)
detail = as_json(data)
status, _, data = self.admin.post_json(
f"/api/admin/intercompany/events/{ledger_event_id}/subject-decisions",
{
"perspective_company_id": detail["event"]["payer_company_id"],
"subject_code": "other_receivable",
"reason": "借款确认其他应收",
"expected_revision": detail["event"]["ledger_revision_id"],
"request_key": f"subj-{ledger_event_id}",
},
)
self.assertEqual(200, status, data)
return as_json(data)["revision"]
def balances(self, client=None, path="/api/admin/intercompany/balances") -> dict:
client = client or self.admin
status, _, data = client.get(path + "?from=2026-01-01&cutoff=2026-12-31")
self.assertEqual(200, status, data)
return as_json(data)
def _by_company(self, payload: dict, company_id: int) -> dict:
return next(item for item in payload["items"] if item["company_id"] == company_id)
# ------------------------------------------------------------------
# Admin balances / pair / events
# ------------------------------------------------------------------
def test_admin_balances_directory(self) -> None:
self._fresh_pair("100.00")
payload = self.balances()
self.assertEqual("2026-12-31", payload["window"]["cutoff"])
companies = {item["company_id"] for item in payload["items"]}
self.assertIn(self.company_a, companies)
self.assertIn(self.company_b, companies)
for item in payload["items"]:
self.assertEqual("unavailable", item["opening"]["status"])
self.assertIsNone(item["opening"]["amount"])
self.assertEqual("period_net_change", item["result"]["kind"])
self.assertIn("gross_amount", item["unresolved"])
self.assertIn("count", item["unresolved"])
self.assertIn("by_reason", item["unresolved"])
self.assertEqual("RMB", item["currency"])
def test_admin_balances_pending_subject_shows_unresolved(self) -> None:
self._fresh_pair("100.00")
payload = self.balances()
item = self._by_company(payload, self.company_a)
self.assertEqual(1, item["unresolved"]["count"])
self.assertEqual(
"100.00", item["unresolved"]["by_reason"]["subject_review"]["gross_amount"]
)
self.assertEqual("0", item["period"]["debit"])
self.assertEqual("0", item["period"]["credit"])
def test_admin_pair_detail_conserves(self) -> None:
self._fresh_pair("100.00")
status, _, data = self.admin.get(
f"/api/admin/intercompany/pairs/{self.company_a}/{self.company_b}"
"?from=2026-01-01&cutoff=2026-12-31"
)
self.assertEqual(200, status, data)
payload = as_json(data)
item = payload["items"][0]
self.assertTrue(item["conservation"]["opposite"])
self.assertTrue(item["conservation"]["abs_equal"])
self.assertEqual("unavailable", item["opening"]["status"])
self.assertIn("subjects", item)
self.assertEqual(1, item["unresolved"]["count"])
def test_admin_subject_decision_flows_to_balances(self) -> None:
pending = self._fresh_pair("100.00")
revision = self._confirm(pending["ledger_event_id"])
self.assertEqual("confirmed", revision["state"])
self.assertEqual("other_receivable", revision["subject_code"])
payload = self.balances()
item = self._by_company(payload, self.company_a)
self.assertEqual(0, item["unresolved"]["count"])
self.assertEqual("100.00", item["period"]["debit"])
self.assertEqual("100.00", item["result"]["signed_amount"])
def test_admin_subject_decision_stale_revision_conflicts(self) -> None:
pending = self._fresh_pair("100.00")
status, _, data = self.admin.post_json(
f"/api/admin/intercompany/events/{pending['ledger_event_id']}/subject-decisions",
{
"perspective_company_id": pending["payer_company_id"],
"subject_code": "other_receivable", "reason": "确认",
"expected_revision": 999, "request_key": "stale-key",
},
)
self.assertEqual(409, status, data)
def test_admin_events_list_and_evidence(self) -> None:
pending = self._fresh_pair("100.00")
revision = self._confirm(pending["ledger_event_id"])
status, _, data = self.admin.get(
"/api/admin/intercompany/events?from=2026-01-01&cutoff=2026-12-31"
)
self.assertEqual(200, status, data)
items = as_json(data)["items"]
self.assertTrue(items)
event = next(
item for item in items
if item["ledger_event_id"] == revision["ledger_event_id"]
)
self.assertEqual("confirmed", event["state"])
self.assertEqual("other_receivable", event["subject_code"])
self.assertIn("payer_company_name", event)
self.assertIn("amount", event)
status, _, data = self.admin.get(
f"/api/admin/intercompany/events/{event['ledger_event_id']}"
)
self.assertEqual(200, status, data)
self.assertEqual("confirmed", as_json(data)["state"])
status, _, data = self.admin.get(
f"/api/admin/intercompany/events/{event['ledger_event_id']}/evidence"
)
self.assertEqual(200, status, data)
blocks = as_json(data)["blocks"]
self.assertTrue(blocks)
for block in blocks:
self.assertEqual("visible", block["visibility"])
def test_admin_events_list_includes_pending_subject(self) -> None:
pending = self._fresh_pair("100.00")
status, _, data = self.admin.get(
"/api/admin/intercompany/events?from=2026-01-01&cutoff=2026-12-31"
)
self.assertEqual(200, status, data)
items = as_json(data)["items"]
item = next(
item for item in items
if item["ledger_event_id"] == pending["ledger_event_id"]
)
self.assertEqual("pending_subject", item["state"])
self.assertIsNone(item["subject_code"])
def test_admin_manual_record_decision_via_api(self) -> None:
status, _, data = self.cashier_a.post_json(
"/api/company/manual-records",
{
"counterparty_company_id": self.company_b,
"occurred_at": "2026-02-01T09:00:00",
"direction": "incoming", "amount": "20.00", "currency": "CNY",
"funding_source": "other", "requested_subject": "other_receivable",
"request_key": "mr-api-1", "summary": "还款",
},
)
self.assertEqual(200, status, data)
record = as_json(data)["record"]
self.assertEqual("pending", record["state"])
status, _, data = self.admin.get("/api/admin/manual-records")
self.assertEqual(200, status, data)
records = as_json(data)["records"]
self.assertTrue(any(item["id"] == record["id"] for item in records))
status, _, data = self.admin.post_json(
f"/api/admin/manual-records/{record['id']}/decisions",
{"action": "approve_new", "reason": "银行流水中无此事实",
"expected_decision_id": record["decision_id"], "request_key": "dec-api-1"},
)
self.assertEqual(200, status, data)
self.assertEqual("approved", as_json(data)["decision"]["state"])
status, _, data = self.cashier_a.get("/api/company/manual-records")
self.assertEqual(200, status, data)
own = [item for item in as_json(data)["records"] if item["id"] == record["id"]]
self.assertEqual(1, len(own))
self.assertEqual("approved", own[0]["state"])
status, _, data = self.cashier_b.get("/api/company/manual-records")
self.assertEqual(200, status, data)
self.assertEqual([], [
item for item in as_json(data)["records"] if item["id"] == record["id"]
])
def test_admin_adjustment_reverse_via_api(self) -> None:
pending = self._fresh_pair("100.00")
revision = self._confirm(pending["ledger_event_id"])
status, _, data = self.admin.post_json(
f"/api/admin/intercompany/events/{revision['ledger_event_id']}/adjustments",
{"action": "reverse", "reason": "科目误判,冲销"},
)
self.assertEqual(200, status, data)
self.assertEqual("reverse", as_json(data)["action"])
# ------------------------------------------------------------------
# Company scope and masking
# ------------------------------------------------------------------
def test_company_balances_scoped_to_own_company(self) -> None:
self._fresh_pair("100.00")
status, _, data = self.cashier_a.get(
"/api/company/intercompany/balances?from=2026-01-01&cutoff=2026-12-31"
)
self.assertEqual(200, status, data)
payload = as_json(data)
for item in payload["items"]:
self.assertEqual(self.company_a, item["company_id"])
self.assertIn("counterparties", payload)
self.assertTrue(payload["counterparties"])
def test_company_pair_and_events_are_own_scope(self) -> None:
self._fresh_pair("100.00")
status, _, data = self.cashier_a.get(
f"/api/company/intercompany/pairs/{self.company_b}"
"?from=2026-01-01&cutoff=2026-12-31"
)
self.assertEqual(200, status, data)
self.assertEqual(self.company_a, as_json(data)["companies"]["a"]["company_id"])
status, _, data = self.cashier_a.get(
"/api/company/intercompany/events?from=2026-01-01&cutoff=2026-12-31"
)
self.assertEqual(200, status, data)
items = as_json(data)["items"]
self.assertTrue(items)
for item in items:
self.assertIn("direction", item)
self.assertIn("own_subject_label", item)
self.assertIn(item["direction"], ("incoming", "outgoing"))
def test_company_evidence_masks_counterparty_side(self) -> None:
pending = self._fresh_pair("100.00")
self._confirm(pending["ledger_event_id"])
status, _, data = self.cashier_a.get(
f"/api/company/intercompany/events/{pending['ledger_event_id']}/evidence"
)
self.assertEqual(200, status, data)
blocks = as_json(data)["blocks"]
visibilities = {block["visibility"] for block in blocks}
self.assertTrue(visibilities & {"visible", "masked"})
for block in blocks:
if block["visibility"] == "masked":
self.assertEqual("按对方授权不可见", block["fields"].get("note"))
self.assertNotIn(ACCOUNT_B, json.dumps(block["fields"], ensure_ascii=False))
def test_company_cannot_read_or_write_admin_intercompany(self) -> None:
self._fresh_pair("100.00")
status, _, data = self.cashier_a.get("/api/admin/intercompany/balances")
self.assertEqual(403, status, data)
status, _, data = self.cashier_a.get("/api/admin/manual-records")
self.assertEqual(403, status, data)
status, _, data = self.cashier_a.post_json(
"/api/admin/manual-records/1/decisions", {"action": "approve_new", "reason": "x"}
)
self.assertEqual(403, status, data)
status, _, data = self.cashier_a.post_json(
"/api/admin/intercompany/events/1/adjustments",
{"action": "reverse", "reason": "越权"},
)
self.assertEqual(403, status, data)
def test_cross_company_reads_are_404(self) -> None:
pending = self._fresh_pair("100.00")
status, _, data = self.cashier_c.get(
f"/api/company/intercompany/events/{pending['ledger_event_id']}"
)
self.assertEqual(404, status, data)
status, _, data = self.cashier_c.get(
f"/api/company/intercompany/events/{pending['ledger_event_id']}/evidence"
)
self.assertEqual(404, status, data)
def test_company_counterparty_summary_splits_by_currency(self) -> None:
# A->B 100 CNY and A->B 50 USD must render two rows, never a mixed
# "CNY 150.00" bucket.
cny = self._fresh_pair("100.00", currency="CNY")
usd = self._fresh_pair("50.00", at="2026-01-20 10:00:00", currency="USD")
for pending in (cny, usd):
self._confirm(pending["ledger_event_id"])
status, _, data = self.cashier_a.get(
"/api/company/intercompany/balances?from=2026-01-01&cutoff=2026-12-31"
)
self.assertEqual(200, status, data)
counterparties = as_json(data)["counterparties"]
rows = [row for row in counterparties if row["counterparty_company_id"] == self.company_b]
self.assertEqual(2, len(rows), counterparties)
by_currency = {row["currency"]: row for row in rows}
self.assertEqual({"CNY", "USD"}, set(by_currency))
self.assertEqual("100.00", by_currency["CNY"]["result"]["signed_amount"])
self.assertEqual("50.00", by_currency["USD"]["result"]["signed_amount"])
for row in rows:
self.assertNotEqual("150.00", row["result"]["signed_amount"])
self.assertEqual(1, row["event_count"])
def test_company_events_subject_filter_matches_mirror(self) -> None:
# A confirms subject "receivable" (stored from A's perspective); B must
# still find the event when filtering by the mirror subject "payable".
pending = self._fresh_pair("100.00")
status, _, data = self.admin.post_json(
f"/api/admin/intercompany/events/{pending['ledger_event_id']}/subject-decisions",
{
"perspective_company_id": self.company_a,
"subject_code": "receivable", "reason": "确认应收",
"expected_revision": 1, "request_key": "mirror-subj",
},
)
self.assertEqual(200, status, data)
status, _, data = self.cashier_b.get(
"/api/company/intercompany/events?from=2026-01-01&cutoff=2026-12-31&subject=payable"
)
self.assertEqual(200, status, data)
items = as_json(data)["items"]
self.assertTrue(any(
item["ledger_event_id"] == pending["ledger_event_id"] for item in items
), data)
def test_manual_reverse_via_api_uses_explicit_effective_date(self) -> None:
status, _, data = self.cashier_a.post_json(
"/api/company/manual-records",
{
"counterparty_company_id": self.company_b,
"occurred_at": "2026-02-01T09:00:00",
"direction": "incoming", "amount": "20.00", "currency": "CNY",
"funding_source": "other", "requested_subject": "other_receivable",
"request_key": "mr-rev-api", "summary": "还款",
},
)
self.assertEqual(200, status, data)
record = as_json(data)["record"]
status, _, data = self.admin.post_json(
f"/api/admin/manual-records/{record['id']}/decisions",
{"action": "approve_new", "reason": "确认入账",
"expected_decision_id": record["decision_id"], "request_key": "dec-rev-api"},
)
self.assertEqual(200, status, data)
event_id = as_json(data)["decision"]["ledger_event_id"]
status, _, data = self.admin.get(
f"/api/admin/intercompany/events/{event_id}"
)
self.assertEqual(200, status, data)
revision_id = as_json(data)["event"]["ledger_revision_id"]
status, _, data = self.admin.post_json(
f"/api/admin/manual-records/{record['id']}/decisions",
{"action": "reverse", "reason": "误录冲销", "effective_at": "2026-06-15",
"expected_decision_id": None, "request_key": "dec-rev-api-2"},
)
self.assertEqual(200, status, data)
status, _, data = self.admin.get(
"/api/admin/intercompany/events?from=2026-01-01&cutoff=2026-12-31"
)
self.assertEqual(200, status, data)
reversals = [item for item in as_json(data)["items"] if item["posting_kind"] == "reversal"]
self.assertEqual(1, len(reversals))
self.assertEqual("2026-06-15", reversals[0]["effective_at"][:10])
self.assertEqual(event_id, reversals[0]["reverses_ledger_event_id"])
def test_admin_events_include_account_chips(self) -> None:
pending = self._fresh_pair("100.00")
revision = self._confirm(pending["ledger_event_id"])
status, _, data = self.admin.get(
"/api/admin/intercompany/events?from=2026-01-01&cutoff=2026-12-31"
)
self.assertEqual(200, status, data)
event = next(
item for item in as_json(data)["items"]
if item["ledger_event_id"] == revision["ledger_event_id"]
)
self.assertIn("payer_account", event)
self.assertIn("payee_account", event)
self.assertEqual("visible", event["payer_account"]["visibility"])
self.assertTrue(event["payer_account"]["label"])
self.assertIn("summary", event)
self.assertIn("is_repayment", event)
def test_subject_exception_drops_from_review_queue(self) -> None:
pending = self._fresh_pair("77.00")
status, _, data = self.admin.post_json(
f"/api/admin/intercompany/events/{pending['ledger_event_id']}/subject-decisions",
{
"perspective_company_id": pending["payer_company_id"],
"action": "exception",
"reason": "转异常待核查",
"expected_revision": pending["revision_id"],
"request_key": "park-exc-api",
},
)
self.assertEqual(200, status, data)
self.assertEqual("pending_subject", as_json(data)["revision"]["state"])
status, _, data = self.admin.get(
"/api/admin/subject-reviews?from=2026-01-01&cutoff=2026-12-31"
)
self.assertEqual(200, status, data)
ids = [item["ledger_event_id"] for item in as_json(data)["items"]]
self.assertNotIn(pending["ledger_event_id"], ids)
if __name__ == "__main__":
unittest.main()
+23 -12
View File
@@ -147,21 +147,21 @@
</section>
<section class="app-view" data-page="pair">
<header class="page-heading"><div><h1>往来查询</h1><p>公司对查询双方口径、科目和逐笔凭证</p></div></header>
<header class="page-heading"><div><h1>往来查询</h1><p>公司间往来余额目录、公司对明细与逐层追溯</p></div>
<span class="status neutral" id="balanceOpeningNote">期初不可用 · 仅显示期间净变动</span>
</header>
<section class="query-band">
<form class="pair-query-form" data-pair-form>
<label class="field"><span>本方公司</span><select name="from"><option>A公司</option><option>B公司</option><option>C公司</option><option>D公司</option><option>E公司</option><option>F公司</option></select></label>
<button class="swap-button" type="button" data-swap aria-label="交换公司" title="交换公司"><svg><use href="icons.svg#arrow-left-right"/></svg></button>
<label class="field"><span>对方公司</span><select name="to"><option>B公司</option><option>C公司</option><option>D公司</option><option>E公司</option><option>F公司</option><option>A公司</option></select></label>
<label class="field"><span>统计截止</span><input name="end" type="date" value="2026-07-31" /></label>
<form class="pair-query-form" data-balance-form>
<label class="field"><span>统计起始</span><input name="from" type="date" value="2026-01-01" /></label>
<label class="field"><span>统计截止</span><input name="cutoff" type="date" value="2026-07-31" /></label>
<label class="field"><span>币种</span><select name="currency"><option value="">全部币种</option><option value="CNY">CNY</option><option value="RMB">RMB</option></select></label>
<button class="button primary" type="submit"><svg><use href="icons.svg#search"/></svg>查询</button>
</form>
</section>
<section class="pair-report" id="pairReport">
<header class="pair-report-heading"><div><h2><span data-pair-from>A公司</span><span data-pair-to>B公司</span></h2><p id="pairPeriod">统计口径 2026.01.01—2026.07.31</p></div><span class="status warning" id="pairReviewStatus">含 1 笔待审核</span></header>
<div class="pair-balance-line"><div><span>期初余额</span><strong id="pairOpening">0.00</strong></div><div><span>本期借方</span><strong id="pairDebit">2,416.00</strong></div><div><span>本期贷方</span><strong id="pairCredit">1,136.00</strong></div><div class="pair-final"><span>期末结果</span><strong id="pairFinal">应收 1,280.00<small>万元</small></strong></div></div>
<div class="subject-strip"><button class="is-active" data-subject-filter="all" aria-pressed="true"><span>全部往来</span><strong data-subject-total="all">5 笔</strong></button><button data-subject-filter="应收" aria-pressed="false"><span>应收</span><strong data-subject-total="应收">1,280.00</strong></button><button data-subject-filter="其他应收" aria-pressed="false"><span>其他应收</span><strong data-subject-total="其他应收">640.00</strong></button><button data-subject-filter="应付" aria-pressed="false"><span>应付</span><strong data-subject-total="应付">728.00</strong></button><button data-subject-filter="其他应付" aria-pressed="false"><span>其他应付</span><strong data-subject-total="其他应付">408.00</strong></button></div>
<div class="table-scroll"><table class="data-table" id="pairTransactions"><thead><tr><th>交易日期</th><th>方向</th><th>科目</th><th>本方账户</th><th>对方账户</th><th>摘要</th><th>匹配</th><th class="number">金额(万元)</th></tr></thead><tbody><tr data-subject="应收"><td>2026.07.18</td><td>转出</td><td>应收</td><td>中信 · 5316</td><td>工行 · 9481</td><td>往来款</td><td><span class="status success">双边匹配</span></td><td class="number">1,000.00</td></tr><tr data-subject="应收"><td>2026.07.06</td><td>转入</td><td>应收</td><td>中信 · 5316</td><td>工行 · 9481</td><td>归还往来款</td><td><span class="status success">双边匹配</span></td><td class="number">320.00</td></tr><tr data-subject="其他应收"><td>2026.06.27</td><td>转出</td><td>其他应收</td><td>建行 · 0845</td><td>工行 · 9481</td><td>资金调拨</td><td><span class="status warning">单边待核</span></td><td class="number">600.00</td></tr></tbody></table></div>
<section class="panel company-ledger-panel" aria-label="公司余额目录">
<div class="panel-heading"><div><h2>公司余额目录</h2><p>每行余额都附带截止日、期初状态、本期借贷、结果与未决金额</p></div></div>
<div class="ledger-head is-balances"><span>公司</span><span class="ledger-hide-md">借方合计</span><span class="ledger-hide-md">贷方合计</span><span>期末结果</span><span>未决</span><span>截止日</span><span></span></div>
<div id="balanceLedgers" class="company-ledgers" aria-live="polite"></div>
</section>
</section>
@@ -219,7 +219,7 @@
<dialog id="auditDialog" class="dialog">
<form id="auditForm">
<header><div><h2 id="auditDialogTitle">审核事项</h2><p id="auditDialogMeta">原始证据与处理决定将一并留痕</p></div><button type="button" class="icon-button" data-close-audit aria-label="关闭" title="关闭"><svg><use href="icons.svg#x"/></svg></button></header>
<div class="dialog-body"><div class="evidence-block" id="auditEvidence"></div><label class="field"><span>处理决定</span><select name="decision" required><option value="">请选择</option><option>确认并纳入计算</option><option>退回公司补充材料</option><option>转为异常待后续处理</option></select></label><label class="field"><span>处理依据</span><textarea name="reason" rows="3" required placeholder="填写核验账号、摘要、回单或说明"></textarea></label></div>
<div class="dialog-body"><div class="evidence-block" id="auditEvidence"></div><label class="field"><span>处理决定</span><select name="decision" required><option value="">请选择</option><option>确认并纳入计算</option><option>退回公司补充材料</option><option>转为异常</option></select></label><p class="decision-danger" id="auditExceptionNote" hidden>转为异常后,该记录暂不纳入余额计算,转入异常队列等待人工核查;此操作将写入审计日志。</p><label class="field"><span>处理依据</span><textarea name="reason" rows="3" required placeholder="填写核验账号、摘要、回单或说明"></textarea></label></div>
<footer><button type="button" class="button secondary" data-close-audit>取消</button><button class="button primary" type="submit">提交审核结果</button></footer>
</form>
</dialog>
@@ -230,6 +230,17 @@
<footer><button type="button" class="button secondary" data-close-closing>取消</button><button class="button primary" type="submit">确认执行结账</button></footer>
</form>
</dialog>
<div class="drawer-scrim" id="drawerScrim" data-close-drawer aria-hidden="true"></div>
<aside class="drawer" id="evidenceDrawer" aria-label="往来明细与源行证据抽屉" aria-hidden="true">
<header class="drawer-header">
<div>
<div class="drawer-breadcrumb" id="drawerBreadcrumb"></div>
<h2 id="drawerTitle">往来明细</h2>
</div>
<button type="button" class="icon-button" data-close-drawer aria-label="关闭抽屉" title="关闭"><svg><use href="icons.svg#x"/></svg></button>
</header>
<div class="drawer-body" id="drawerBody" tabindex="-1"></div>
</aside>
<div class="toast-region" id="toastRegion" aria-live="polite"></div>
<script src="app.js"></script>
</body>
+1231 -142
View File
File diff suppressed because it is too large Load Diff
+33 -2
View File
@@ -26,7 +26,8 @@
<button class="nav-item" data-view="upload"><svg><use href="icons.svg#upload"/></svg><span>流水导入</span></button>
<button class="nav-item" data-view="manual"><svg><use href="icons.svg#plus"/></svg><span>手工记录</span></button>
<button class="nav-item" data-view="flows"><svg><use href="icons.svg#file-spreadsheet"/></svg><span>流水管理</span></button>
<button class="nav-item" data-view="reconcile"><svg><use href="icons.svg#arrow-left-right"/></svg><span>往来确认</span><b>2</b></button>
<button class="nav-item" data-view="balances"><svg><use href="icons.svg#arrow-left-right"/></svg><span>往来余额</span></button>
<button class="nav-item" data-view="reconcile"><svg><use href="icons.svg#circle-check"/></svg><span>往来确认</span><b>2</b></button>
<button class="nav-item" data-view="accounts"><svg><use href="icons.svg#landmark"/></svg><span>银行账户</span></button>
<button class="nav-item" data-view="notifications"><svg><use href="icons.svg#bell"/></svg><span>通知</span><b>2</b></button>
</nav>
@@ -140,7 +141,7 @@
<div class="form-body">
<div class="form-grid"><label class="field"><span>交易日期</span><input name="transactionDate" type="date" value="2026-08-06" required /></label><label class="field"><span>收付方向</span><select name="direction" required><option>付款</option><option>收款</option></select></label></div>
<div class="form-grid"><label class="field"><span>金额(元)</span><input name="amount" type="number" min="0.01" step="0.01" required /></label><label class="field"><span>资金来源</span><select name="sourceAccount" required><option value="">请选择</option><option>个人过账</option></select></label></div>
<div class="form-grid"><label class="field"><span>对方类型</span><select name="counterpartyType" required><option>集团内部公司</option><option>个人过账方</option><option>外部单位</option></select></label><label class="field"><span>对方名称</span><input name="counterparty" maxlength="100" placeholder="公司全称或个人姓名" required /></label></div>
<div class="form-grid"><label class="field"><span>对方类型</span><select name="counterpartyType" required><option>集团内部公司</option><option>个人过账方</option><option>外部单位</option></select></label><label class="field"><span>对方公司</span><select name="counterparty" required><option value="">请选择集团内部公司</option></select></label></div>
<div class="form-grid"><label class="field"><span>对方账号</span><input name="counterpartyAccount" maxlength="64" placeholder="可选" /></label><label class="field"><span>往来科目</span><select name="subject" required><option>应收</option><option>应付</option><option>其他应收</option><option>其他应付</option></select></label></div>
<label class="field"><span>业务摘要</span><input name="summary" maxlength="120" placeholder="例如:个人代付后转回" required /></label>
<label class="field"><span>补充说明</span><textarea name="remark" rows="3" maxlength="500" placeholder="说明形成原因和核对依据" required></textarea></label>
@@ -162,6 +163,25 @@
<section class="panel"><div class="table-summary"><span>当前结果 <strong id="flowCount">3</strong></span><span>金额单位:元</span></div><div class="table-scroll"><table class="data-table" id="flowTable"><thead><tr><th>日期</th><th>银行账户</th><th>方向</th><th>对方户名 / 账号</th><th>摘要</th><th>银行流水号</th><th>归集状态</th><th class="number">金额</th></tr></thead><tbody><tr data-bank="中信银行"><td>2026.07.18</td><td>中信银行 · 5316</td><td>转出</td><td><strong>B公司</strong><small>尾号 9481</small></td><td>往来款</td><td>CIT260718018</td><td><span class="status success">双边匹配</span></td><td class="number">10,000,000.00</td></tr><tr data-bank="建设银行"><td>2026.07.12</td><td>建设银行 · 0845</td><td>转出</td><td><strong>A公司</strong><small>尾号 5316</small></td><td>同名账户调拨</td><td>CCB260712031</td><td><span class="status neutral">同公司调拨</span></td><td class="number">2,000,000.00</td></tr><tr data-bank="中信银行"><td>2026.07.06</td><td>中信银行 · 5316</td><td>转入</td><td><strong>B公司</strong><small>尾号 9481</small></td><td>归还往来款</td><td>CIT260706041</td><td><span class="status success">双边匹配</span></td><td class="number">3,200,000.00</td></tr></tbody></table></div></section>
</section>
<section class="app-view" data-page="balances">
<header class="page-heading"><div><h1>往来余额</h1><p>本公司往来余额、对方公司明细与允许查看的源证据</p></div>
<span class="status neutral" id="companyOpeningNote">期初不可用 · 仅显示期间净变动</span>
</header>
<div class="company-alert balance-alert" id="companyUnresolvedAlert" hidden>
<svg><use href="icons.svg#circle-alert"/></svg>
<div><strong id="companyUnresolvedText"></strong><p>未确认事项不影响已确认余额,处理前请先核对证据。</p></div>
<button class="button secondary" data-view-link="reconcile">去确认</button>
</div>
<section class="pair-report" id="companyBalanceSummary">
<header class="pair-report-heading"><div><h2>本公司往来汇总</h2><p id="companyBalancePeriod">统计口径</p></div></header>
<div id="companyBalanceGroups" class="balance-currency-groups"></div>
</section>
<section class="panel company-stack-panel">
<div class="panel-heading"><div><h2>对方公司明细</h2><p>点击任意对方公司查看往来事件与允许的源证据</p></div></div>
<div class="company-list" id="companyCounterparties" aria-live="polite"></div>
</section>
</section>
<section class="app-view" data-page="reconcile">
<header class="page-heading"><div><h1>往来确认</h1><p>系统计算为主,只处理无法确定的匹配与科目</p></div></header>
<section class="reconcile-summary"><div><span>自动确认</span><strong>18 笔</strong><small>无需人工处理</small></div><div><span>单边待匹配</span><strong id="matchPendingCount">1 笔</strong><small>需选择对方证据</small></div><div><span>科目待确认</span><strong id="subjectPendingCount">1 笔</strong><small>需选择会计科目</small></div></section>
@@ -195,6 +215,17 @@
<footer><button type="button" class="button secondary" data-close-account>取消</button><button class="button primary" type="submit">提交登记</button></footer>
</form>
</dialog>
<div class="drawer-scrim" id="drawerScrim" data-close-drawer aria-hidden="true"></div>
<aside class="drawer" id="evidenceDrawer" aria-label="往来明细与源行证据抽屉" aria-hidden="true">
<header class="drawer-header">
<div>
<div class="drawer-breadcrumb" id="drawerBreadcrumb"></div>
<h2 id="drawerTitle">往来明细</h2>
</div>
<button type="button" class="icon-button" data-close-drawer aria-label="关闭抽屉" title="关闭"><svg><use href="icons.svg#x"/></svg></button>
</header>
<div class="drawer-body" id="drawerBody" tabindex="-1"></div>
</aside>
<div class="toast-region" id="toastRegion" aria-live="polite"></div>
<script src="app.js"></script>
</body>
+17 -10
View File
@@ -11,19 +11,31 @@
<!--
THESIS: 登录页只完成身份确认,并明确区分总账管理端与公司业务端。
OWN-WORLD: 深黑身份场景、石墨玻璃表单和荧光绿当前状态,延续双端工作台的材料语言。
STORY: 居中集团登录卡:先确认集团身份系统名称,再选择工作端口登录。
FIRST VIEWPORT: 一张悬浮玻璃登录卡居于氛围光中央,集团名称置顶,端口选择、账号表单与系统事实依次排列
STORY: 桌面双区入口:左侧确认集团身份系统名称与账期状态,右侧选择工作端口登录。
FIRST VIEWPORT: 横向双区;左侧入口信息,右侧 420-460px 登录表单。端口选择为紧凑分段控件
FORM: 用户参考图锁定的深色玻璃财务工作台,Operate 模式;seed key fe8a50aa。
FINISH: unreviewed and undocumented is unfinished; this build ends with the finish review, the verdict, and DESIGN.md
-->
<main class="entry-shell">
<section class="entry-card" aria-labelledby="login-title">
<section class="entry-context" aria-labelledby="product-name">
<div class="entry-brand"><span class="brand-mark"></span><span><strong id="product-name">河南金牛实业集团</strong><small>集团资金往来管理系统</small></span></div>
<p class="entry-status"><span>服务状态</span><b>运行中</b></p>
<div class="entry-statement">
<h1>一笔往来,追溯到双方银行凭证。</h1>
<p>选择与账号一致的工作端口后进入对应工作台。总账管理端查看集团全貌,公司业务端只处理本公司账务。</p>
</div>
<dl class="entry-facts">
<div><dt>全局起算日</dt><dd>2026.01.01</dd></div>
<div><dt>当前账期</dt><dd>2026 年 7 月</dd></div>
<div><dt>覆盖银行</dt><dd>已对接 6 家</dd></div>
</dl>
</section>
<section class="entry-form-wrap" aria-labelledby="login-title">
<form class="entry-form" id="loginForm">
<header><h2 id="login-title">登录</h2><p>请选择与账号一致的工作端口</p></header>
<div class="role-switch" role="radiogroup" aria-label="工作端口">
<label><input type="radio" name="role" value="admin" checked /><span><svg><use href="icons.svg#shield-check"/></svg><b>总账管理端</b><small>集团管理员</small></span></label>
<label><input type="radio" name="role" value="company" /><span><svg><use href="icons.svg#building"/></svg><b>公司业务端</b><small>公司出纳</small></span></label>
<label><input type="radio" name="role" value="admin" checked /><span><svg><use href="icons.svg#shield-check"/></svg><b>总账管理端</b><small class="sr-only">集团管理员</small></span></label>
<label><input type="radio" name="role" value="company" /><span><svg><use href="icons.svg#building"/></svg><b>公司业务端</b><small class="sr-only">公司出纳</small></span></label>
</div>
<label class="field"><span>账号</span><input name="username" autocomplete="username" required /></label>
<label class="field"><span>密码</span><span class="password-field"><input name="password" type="password" autocomplete="current-password" required /><button type="button" class="inside-icon" id="togglePassword" aria-label="显示密码" title="显示密码"><svg><use href="icons.svg#eye"/></svg></button></span></label>
@@ -36,11 +48,6 @@
<label class="check-field"><input type="checkbox" checked />记住本次登录</label>
<button class="button primary wide" type="submit"><span id="loginAction">进入总账管理端</span><svg><use href="icons.svg#chevron-right"/></svg></button>
</form>
<dl class="entry-facts">
<div><dt>全局起算日</dt><dd>2026.01.01</dd></div>
<div><dt>当前账期</dt><dd>2026 年 7 月</dd></div>
<div><dt>覆盖银行</dt><dd>已对接 6 家</dd></div>
</dl>
</section>
</main>
<script src="app.js"></script>
+258 -59
View File
@@ -26,6 +26,8 @@
--color-danger-wash: rgba(255, 98, 109, 0.11);
--color-info: #66a8ff;
--color-info-wash: rgba(102, 168, 255, 0.11);
--row-h-evidence: 55px;
--z-drawer: 150;
--radius-sm: 10px;
--radius-md: 16px;
--radius-lg: 22px;
@@ -218,7 +220,7 @@ main { width: min(1560px, 100%); margin: 0 auto; padding: 12px 32px 58px; }
.company-row.is-warning .company-row-mark { background: var(--color-warning-wash); color: var(--color-warning); }
.company-row.is-danger .company-row-mark { background: var(--color-danger-wash); color: var(--color-danger); }
.company-row-body { display: flex; flex-direction: column; min-width: 0; }
.company-row-body strong { font-size: 14px; }
.company-row-body strong { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; font-size: 14px; }
.company-row-body small { color: var(--color-ink-muted); font-size: 11px; }
.company-row-figure { display: flex; flex-direction: column; align-items: flex-end; text-align: right; min-width: 92px; }
.company-row-figure strong { font: 600 14px var(--font-data); color: var(--color-ink); }
@@ -259,8 +261,9 @@ main { width: min(1560px, 100%); margin: 0 auto; padding: 12px 32px 58px; }
.company-ledger summary:hover { background: rgba(255, 255, 255, 0.035); }
.company-ledger summary > svg { transition: transform var(--duration-standard) var(--ease-out); }
.company-ledger[open] summary > svg { transform: rotate(180deg); }
.company-name { display: grid; grid-template-columns: 38px minmax(0, 1fr); align-items: center; column-gap: 10px; }
.company-name { display: grid; grid-template-columns: 38px minmax(0, 1fr); align-items: center; column-gap: 10px; min-width: 0; }
.company-name i { width: 36px; height: 36px; grid-row: 1 / 3; display: grid; place-items: center; border-radius: 12px; background: var(--color-primary-wash); color: var(--color-primary); font-style: normal; font-weight: 800; }
.company-name b { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; min-width: 0; }
.company-name small { color: var(--color-ink-muted); }
.amount { font: 600 13px var(--font-data); }
.amount.debit { color: var(--color-positive); }
@@ -432,32 +435,61 @@ main { width: min(1560px, 100%); margin: 0 auto; padding: 12px 32px 58px; }
/* Login */
.entry-page { overflow-x: hidden; }
.entry-shell { min-height: 100vh; display: grid; grid-template-columns: minmax(0, 1.15fr) minmax(430px, 0.85fr); }
.entry-context { position: relative; min-height: 100vh; display: flex; flex-direction: column; justify-content: space-between; padding: clamp(32px, 5vw, 76px); border-right: 1px solid var(--color-line); background: #080808; }
.entry-context::after { content: ""; position: absolute; inset: 12% 8% 12% auto; width: 1px; background: rgba(55, 235, 137, 0.32); box-shadow: 0 0 38px rgba(55, 235, 137, 0.45); }
.entry-brand { display: flex; align-items: center; gap: 12px; }
.entry-brand > span:last-child { display: flex; flex-direction: column; }
.entry-shell { width: 100%; max-width: 1180px; min-height: 100vh; display: grid; grid-template-columns: minmax(0, 1fr) minmax(460px, 520px); margin: 0 auto; padding: 0; }
.entry-context { position: relative; min-width: 0; min-height: 100vh; display: flex; flex-direction: column; justify-content: space-between; padding: clamp(28px, 4vw, 56px) clamp(28px, 4vw, 64px); border-right: 1px solid var(--color-line); background: var(--color-bg-soft); }
.entry-brand { display: flex; align-items: center; gap: 12px; min-width: 0; }
.entry-brand > span:last-child { display: flex; flex-direction: column; min-width: 0; }
.entry-brand strong { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
.entry-brand small { color: var(--color-ink-muted); }
.entry-statement { max-width: 680px; margin: 80px 0; }
.entry-statement h1 { max-width: 640px; font-size: clamp(38px, 5vw, 70px); line-height: 1.12; letter-spacing: 0; }
.entry-statement p { margin-top: 20px; color: var(--color-ink-muted); }
.entry-facts { display: grid; grid-template-columns: repeat(3, 1fr); gap: 12px; max-width: 670px; }
.entry-facts div { padding: 15px; border: 1px solid var(--color-line); border-radius: var(--radius-md); background: var(--color-surface-muted); }
.entry-facts dt { color: var(--color-ink-muted); }
.entry-facts dd { margin-top: 4px; color: var(--color-primary); }
.entry-form-wrap { min-height: 100vh; display: grid; place-items: center; padding: 40px; background: var(--color-bg); }
.entry-form { width: min(420px, 100%); display: grid; gap: 18px; padding: 28px; border: 1px solid var(--color-line); border-radius: var(--radius-xl); background: var(--color-surface); box-shadow: var(--shadow-panel); backdrop-filter: blur(24px); }
.entry-form header h2 { font-size: 25px; }
.entry-status { display: flex; align-items: center; gap: 8px; margin: 14px 0 0; color: var(--color-ink-soft); font-size: 12px; }
.entry-status span { color: var(--color-ink-muted); }
.entry-status b { color: var(--color-ink); font-weight: 700; }
.entry-status b::before { content: ""; display: inline-block; width: 7px; height: 7px; margin-right: 6px; border-radius: 50%; background: var(--color-primary); vertical-align: 1px; }
.entry-statement { max-width: 640px; margin: 48px 0 auto; }
.entry-statement h1 { max-width: 620px; font-size: clamp(30px, 3.4vw, 48px); line-height: 1.18; letter-spacing: 0; }
.entry-statement p { margin-top: 12px; color: var(--color-ink-muted); max-width: 520px; }
.entry-facts { display: grid; grid-template-columns: repeat(3, minmax(0, 1fr)); gap: 10px; max-width: 640px; }
.entry-facts div { min-width: 0; padding: 12px; border: 1px solid var(--color-line); border-radius: var(--radius-sm); background: var(--color-surface-muted); }
.entry-facts dt { color: var(--color-ink-muted); font-size: 11px; }
.entry-facts dd { margin-top: 4px; color: var(--color-primary); overflow-wrap: anywhere; }
.entry-form-wrap { min-width: 0; min-height: 100vh; display: grid; place-items: center; padding: 28px 24px; background: var(--color-bg); }
.entry-form { width: min(440px, 100%); display: grid; gap: 11px; padding: 22px 22px 20px; border: 1px solid var(--color-line); border-radius: var(--radius-lg); background: var(--color-surface); box-shadow: var(--shadow-low); backdrop-filter: blur(18px); }
.entry-form header h2 { font-size: 20px; }
.entry-form header p { margin-top: 4px; font-size: 12px; }
.entry-form header p, .entry-note { color: var(--color-ink-muted); }
.role-switch { display: grid; grid-template-columns: 1fr 1fr; gap: 9px; }
.role-switch label { cursor: pointer; }
.role-switch input { position: absolute; opacity: 0; }
.role-switch span { min-height: 83px; display: grid; grid-template-columns: 30px 1fr; align-content: center; gap: 1px 9px; padding: 12px; border: 1px solid var(--color-line); border-radius: var(--radius-md); background: rgba(0, 0, 0, 0.14); transition: border-color var(--duration-fast), background var(--duration-fast), transform var(--duration-fast); }
.role-switch svg { grid-row: 1 / 3; align-self: center; color: var(--color-ink-muted); }
.role-switch small { color: var(--color-ink-muted); }
.role-switch input:checked + span { border-color: rgba(55, 235, 137, 0.4); background: var(--color-primary-wash); transform: translateY(-1px); }
.entry-form #loginError { color: var(--color-danger); }
.entry-form .field { gap: 4px; }
.entry-form .field input:focus-visible { outline: 2px solid var(--color-primary-strong); outline-offset: 3px; }
.role-switch { display: flex; gap: 4px; padding: 4px; border: 1px solid var(--color-line); border-radius: 13px; background: rgba(0, 0, 0, 0.18); }
.role-switch label { position: relative; flex: 1; min-width: 0; cursor: pointer; }
.role-switch input { position: absolute; opacity: 0; width: 1px; height: 1px; }
.role-switch span { min-height: 36px; display: flex; align-items: center; justify-content: center; gap: 6px; padding: 0 8px; border: 0; border-radius: 9px; background: transparent; color: var(--color-ink-muted); transition: color var(--duration-fast), background var(--duration-fast); }
.role-switch svg { width: 15px; height: 15px; flex: 0 0 auto; }
.role-switch b { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; font-size: 12px; }
.role-switch input:checked + span { background: var(--color-primary-wash); color: var(--color-ink); }
.role-switch input:checked + span svg { color: var(--color-primary); }
.entry-note { font-size: 10px; }
.role-switch input:focus-visible + span { outline: 2px solid var(--color-primary-strong); outline-offset: 2px; }
.entry-note { font-size: 12px; }
@media (max-width: 1023px) {
.entry-shell { grid-template-columns: minmax(0, 1fr) minmax(400px, 440px); }
.entry-context { padding: 24px 20px; }
.entry-statement { margin: 28px 0 auto; }
.entry-statement h1 { font-size: clamp(24px, 3.6vw, 34px); }
.entry-facts { grid-template-columns: 1fr; max-width: none; }
.entry-form-wrap { padding: 20px 16px; }
.entry-form { width: min(400px, 100%); padding: 18px; gap: 10px; }
}
@media (max-width: 720px) {
.entry-shell { grid-template-columns: 1fr; }
.entry-context { min-height: auto; padding: 22px 16px 16px; border-right: 0; border-bottom: 1px solid var(--color-line); }
.entry-statement { margin: 18px 0 14px; }
.entry-statement h1 { font-size: 24px; }
.entry-facts { grid-template-columns: 1fr; }
.entry-form-wrap { min-height: auto; padding: 16px 16px 32px; align-content: start; }
.entry-form { width: 100%; padding: 16px; }
}
@media (max-width: 1180px) {
.app-shell { grid-template-columns: 88px minmax(0, 1fr); }
@@ -485,8 +517,8 @@ main { width: min(1560px, 100%); margin: 0 auto; padding: 12px 32px 58px; }
.pair-query-form .field:nth-of-type(3) { grid-column: 1 / 3; }
.pair-query-form .button { grid-column: 3; }
.ledger-head { display: none; }
.company-ledger summary { grid-template-columns: minmax(180px, 1.4fr) repeat(2, minmax(100px, 0.7fr)) 28px; }
.company-ledger summary > span:nth-child(4), .company-ledger summary > span:nth-child(5) { display: none; }
.company-ledger:not(.is-balances) summary { grid-template-columns: minmax(180px, 1.4fr) repeat(2, minmax(100px, 0.7fr)) 28px; }
.company-ledger:not(.is-balances) summary > span:nth-child(4), .company-ledger:not(.is-balances) summary > span:nth-child(5) { display: none; }
.ledger-breakdown { grid-template-columns: 1fr; }
.ledger-breakdown section + section { border-top: 1px solid var(--color-line); border-left: 0; }
.pair-balance-line { grid-template-columns: repeat(3, 1fr); }
@@ -494,9 +526,6 @@ main { width: min(1560px, 100%); margin: 0 auto; padding: 12px 32px 58px; }
.subject-strip { grid-template-columns: repeat(3, 1fr); }
.work-progress { grid-template-columns: 1fr 24px 1fr; row-gap: 16px; }
.work-progress i:nth-of-type(2) { display: none; }
.entry-shell { grid-template-columns: 1fr; }
.entry-context { min-height: 58vh; border-right: 0; border-bottom: 1px solid var(--color-line); }
.entry-form-wrap { min-height: auto; }
}
@media (max-width: 720px) {
@@ -532,8 +561,8 @@ main { width: min(1560px, 100%); margin: 0 auto; padding: 12px 32px 58px; }
.task-list button > b { display: none; }
.quick-pair-form { grid-template-columns: 1fr 38px 1fr; padding-inline: 12px; }
.quick-result { margin-inline: 12px; }
.company-ledger summary { grid-template-columns: minmax(145px, 1fr) 100px 24px; padding-inline: 12px; }
.company-ledger summary > strong:nth-of-type(2), .company-ledger summary > span:nth-child(4), .company-ledger summary > span:nth-child(5) { display: none; }
.company-ledger:not(.is-balances) summary { grid-template-columns: minmax(145px, 1fr) 100px 24px; padding-inline: 12px; }
.company-ledger:not(.is-balances) summary > strong:nth-of-type(2), .company-ledger:not(.is-balances) summary > span:nth-child(4), .company-ledger:not(.is-balances) summary > span:nth-child(5) { display: none; }
.company-name { grid-template-columns: 30px minmax(0, 1fr); }
.company-name i { width: 28px; height: 28px; }
.filter-bar { align-items: stretch; flex-direction: column; }
@@ -570,12 +599,6 @@ main { width: min(1560px, 100%); margin: 0 auto; padding: 12px 32px 58px; }
.form-grid { grid-template-columns: 1fr; }
.toast-region { right: 12px; bottom: 12px; left: 12px; }
.toast { min-width: 0; max-width: none; }
.entry-context { min-height: auto; padding: 30px 22px; }
.entry-statement { margin: 54px 0 42px; }
.entry-statement h1 { font-size: 39px; }
.entry-facts { grid-template-columns: 1fr; }
.entry-form-wrap { padding: 32px 16px 48px; }
.entry-form { padding: 22px; }
}
@media (max-width: 460px) {
@@ -583,7 +606,6 @@ main { width: min(1560px, 100%); margin: 0 auto; padding: 12px 32px 58px; }
.metric-copy { min-height: 52px; }
.metric-label { font-size: 11px; }
.metric-value { font-size: 23px; }
.role-switch { grid-template-columns: 1fr; }
}
@media (max-width: 1180px) {
@@ -592,8 +614,8 @@ main { width: min(1560px, 100%); margin: 0 auto; padding: 12px 32px 58px; }
}
@media (max-width: 900px) {
.company-row { grid-template-columns: 34px minmax(0, 1fr) auto; }
.company-row-figure { display: none; }
.company-row:not(.is-balance-counterparty) { grid-template-columns: 34px minmax(0, 1fr) auto; }
.company-row:not(.is-balance-counterparty) .company-row-figure { display: none; }
}
@media (max-width: 720px) {
@@ -715,24 +737,201 @@ body::before { content: ""; position: fixed; inset: 0; z-index: 0; pointer-event
.table-scroll { padding: 2px 14px 8px; }
.table-summary { margin: 0 14px; }
/* ===== 登录页重做:居中集团登录卡 ===== */
.entry-shell { display: grid; grid-template-columns: 1fr; place-items: center; min-height: 100vh; padding: 48px 20px; }
.entry-card { position: relative; width: min(470px, 100%); overflow: hidden; border: 1px solid rgba(255, 255, 255, 0.13); border-radius: var(--radius-xl); background: linear-gradient(165deg, rgba(36, 36, 36, 0.9) 0%, rgba(16, 16, 16, 0.92) 60%, rgba(10, 10, 10, 0.94) 100%); box-shadow: var(--shadow-panel), 0 0 60px rgba(55, 235, 137, 0.06); backdrop-filter: blur(28px) saturate(125%); }
.entry-card::before { content: ""; position: absolute; inset: 0; background: radial-gradient(150% 110% at 50% 118%, rgba(255, 255, 255, 0.07), transparent 55%), linear-gradient(165deg, rgba(255, 255, 255, 0.12) 0%, rgba(255, 255, 255, 0.03) 42%, transparent 70%); pointer-events: none; }
.entry-card::after { content: ""; position: absolute; top: 0; left: 18%; right: 18%; height: 2px; background: linear-gradient(90deg, transparent, rgba(55, 235, 137, 0.65), transparent); }
.entry-brand { flex-direction: column; justify-content: center; gap: 15px; min-height: 0; padding: 36px 28px 24px; border-bottom: 1px solid var(--color-line); text-align: center; }
.entry-brand .brand-mark { width: 54px; height: 54px; border-radius: 17px; font-size: 26px; }
.entry-brand > span:last-child { align-items: center; }
.entry-brand strong { font-size: 21px; letter-spacing: 0.03em; }
.entry-brand small { margin-top: 4px; font-size: 12px; letter-spacing: 0.08em; }
.entry-form { width: 100%; gap: 16px; padding: 24px 28px 10px; border: 0; background: transparent; box-shadow: none; backdrop-filter: none; }
.entry-form header { text-align: center; }
.entry-form header h2 { font-size: 22px; }
.entry-form header p { margin-top: 4px; color: var(--color-ink-muted); font-size: 12px; }
.entry-facts { max-width: none; gap: 10px; padding: 16px 28px 28px; }
.entry-facts div { padding: 12px; text-align: center; }
.entry-facts dd { margin-top: 2px; font-size: 12px; }
@media (prefers-reduced-motion: reduce) {
*, *::before, *::after { scroll-behavior: auto !important; animation-duration: 0.01ms !important; animation-iteration-count: 1 !important; transition-duration: 0.01ms !important; }
}
/* ---- B-44 intercompany balances ---- */
.status.info { border-color: rgba(102, 168, 255, 0.18); background: var(--color-info-wash); color: var(--color-info); }
.amount-neutral { color: var(--color-ink-muted); }
.currency-code { font-size: 10px; color: var(--color-ink-muted); margin-right: 4px; }
.amount-with-currency { white-space: nowrap; font-variant-numeric: tabular-nums; }
.amount-with-currency.is-compact { font-size: 11px; }
.account-cell { white-space: nowrap; }
.summary-cell { max-width: 220px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
.candidate-list { margin: 8px 0 0; padding: 0; list-style: none; color: var(--color-ink-soft); font-size: 12px; }
.candidate-list li { padding: 4px 0; border-bottom: 1px dashed var(--color-line); }
.candidate-list li:last-child { border-bottom: 0; }
.reason-full { max-height: calc(1.6em * 6); overflow: auto; white-space: pre-wrap; }
.decision-danger { margin-top: 8px; color: var(--color-danger); font-size: 12px; }
.balance-currency-groups { display: grid; gap: 12px; }
.balance-currency-groups .pair-balance-line { border: 1px solid var(--color-line); border-radius: var(--radius-md); overflow: hidden; }
/* Balance directory: 公司 | 借方 | 贷方 | 期末结果 | 未决 | 截止日 | ▸ */
.ledger-head.is-balances, .company-ledger.is-balances summary {
grid-template-columns: minmax(180px, 1.3fr) minmax(0, 0.7fr) minmax(0, 0.7fr) minmax(0, 0.9fr) minmax(0, 0.7fr) minmax(96px, 0.55fr) 24px;
}
.company-ledger.is-balances summary { min-height: 72px; }
.company-ledger.is-balances summary .company-name { min-width: 96px; }
.company-ledger.is-balances summary .ledger-cutoff { color: var(--color-ink-muted); font-size: 11px; white-space: nowrap; }
.company-ledger.is-balances .ledger-result { display: flex; align-items: center; gap: 7px; min-width: 0; }
.company-ledger.is-balances .ledger-result b { font: 600 14px var(--font-data); font-variant-numeric: tabular-nums; white-space: nowrap; min-width: 0; }
.company-ledger.is-balances .ledger-unresolved { font-size: 11px; white-space: nowrap; }
.company-ledger.is-balances .ledger-unresolved.is-empty { color: var(--color-ink-muted); }
.company-ledger.is-balances .ledger-unresolved.is-active { color: var(--color-warning); }
.company-ledger.is-balances .currency-tag { color: var(--color-ink-muted); font-size: 10px; }
.company-ledger.is-balances .subject-row small { white-space: nowrap; }
@media (max-width: 1179px) {
.ledger-head.is-balances, .company-ledger.is-balances summary {
grid-template-columns: minmax(120px, 1.2fr) minmax(136px, 0.9fr) minmax(0, 0.7fr) minmax(96px, 0.55fr) 24px;
}
.ledger-head.is-balances .ledger-hide-md,
.company-ledger.is-balances summary .ledger-hide-md { display: none; }
.company-ledger.is-balances summary .amount-with-currency .currency-code { display: none; }
}
@media (min-width: 376px) and (max-width: 900px) {
.ledger-head.is-balances { display: grid; }
.ledger-head.is-balances, .company-ledger.is-balances summary {
grid-template-columns: minmax(80px, 0.85fr) minmax(152px, 1.2fr) minmax(96px, 0.9fr) minmax(84px, 0.5fr) 24px;
}
.company-ledger.is-balances summary .ledger-result {
flex-direction: column;
align-items: flex-start;
justify-content: center;
gap: 2px;
}
.company-ledger.is-balances .ledger-result b { font-size: 11px; }
.company-ledger.is-balances .ledger-unresolved {
min-width: 0;
white-space: normal;
}
.company-ledger.is-balances .ledger-unresolved .status {
width: auto;
max-width: 100%;
height: auto;
white-space: normal;
justify-content: flex-start;
text-align: left;
}
}
@media (max-width: 375px) {
.ledger-head.is-balances { display: none; }
.company-ledger.is-balances summary {
grid-template-columns: minmax(88px, 1fr) minmax(0, max-content) 24px;
grid-template-rows: auto auto;
align-items: center;
column-gap: 10px;
row-gap: 4px;
}
.company-ledger.is-balances summary .company-name { grid-column: 1; grid-row: 1; min-width: 88px; }
.company-ledger.is-balances summary .ledger-result {
grid-column: 2; grid-row: 1; justify-self: end;
flex-direction: column; align-items: flex-end; gap: 2px;
}
.company-ledger.is-balances summary .ledger-result .currency-code { display: none; }
.company-ledger.is-balances summary .ledger-unresolved {
grid-column: 1; grid-row: 2; min-width: 0; white-space: normal;
}
.company-ledger.is-balances summary .ledger-cutoff { grid-column: 2; grid-row: 2; justify-self: end; }
.company-ledger.is-balances summary > svg { grid-column: 3; grid-row: 1 / span 2; align-self: center; }
.company-ledger.is-balances summary .ledger-hide-md { display: none; }
}
/* Pair report six-cell: page may be six-across; drawer is always two rows of three */
.pair-balance-line.is-six { grid-template-columns: repeat(6, 1fr); }
.pair-balance-line.is-six .pair-final { grid-column: auto; border-left: 1px solid var(--color-line); }
.pair-balance-line .pair-open-note { color: var(--color-ink-muted); font-size: 10px; margin-top: 3px; }
.pair-balance-line .pair-unresolved b { color: var(--color-warning); }
.pair-balance-line .pair-unresolved.is-empty b { color: var(--color-ink); }
.pair-balance-line .pair-cutoff { color: var(--color-ink-muted); font-size: 10px; margin-top: 3px; }
@media (max-width: 1180px) {
.pair-balance-line.is-six { grid-template-columns: repeat(3, 1fr); }
.pair-balance-line.is-six div:nth-child(4) { border-top: 1px solid var(--color-line); border-left: 0; }
.pair-balance-line.is-six div:nth-child(5), .pair-balance-line.is-six div:nth-child(6) { border-top: 1px solid var(--color-line); }
}
.amount-with-currency { font-variant-numeric: tabular-nums; white-space: nowrap; }
.amount-with-currency.is-compact { font-size: 11px; }
.amount-with-currency .currency-code { margin-right: 4px; color: var(--color-ink-muted); font-size: 10px; font-weight: 500; }
/* Evidence drawer: <768 full, 7681179 480px, >=1180 640px */
.drawer-scrim { position: fixed; inset: 0; z-index: calc(var(--z-drawer) - 1); background: rgba(5, 5, 5, 0.6); opacity: 0; pointer-events: none; transition: opacity var(--duration-standard) var(--ease-out); }
.drawer-scrim.is-open { opacity: 1; pointer-events: auto; }
.drawer { position: fixed; top: 0; right: 0; bottom: 0; z-index: var(--z-drawer); width: 640px; max-width: 100%; display: flex; flex-direction: column; background: rgba(17, 17, 17, 0.96); border-left: 1px solid var(--color-line-strong); box-shadow: var(--shadow-panel); transform: translateX(24px); opacity: 0; visibility: hidden; transition: transform var(--duration-standard) var(--ease-out), opacity var(--duration-standard) var(--ease-out), visibility 0s linear var(--duration-standard); }
.drawer.is-open { transform: translateX(0); opacity: 1; visibility: visible; transition-delay: 0s; }
@media (max-width: 1179px) { .drawer { width: 480px; } }
@media (max-width: 767px) { .drawer { width: 100%; } }
.drawer-header { min-height: 74px; display: flex; align-items: flex-start; justify-content: space-between; gap: 12px; padding: 16px 18px; border-bottom: 1px solid var(--color-line); }
.drawer-header .drawer-breadcrumb { display: flex; flex-wrap: wrap; align-items: center; gap: 5px; color: var(--color-ink-muted); font-size: 11px; }
.drawer-header .drawer-breadcrumb button { border: 0; padding: 0; background: none; color: var(--color-info); cursor: pointer; font-size: 11px; }
.drawer-header .drawer-breadcrumb button:hover { text-decoration: underline; }
.drawer-header h2 { margin-top: 6px; font-size: 17px; }
.drawer-body { flex: 1; overflow: auto; padding: 16px 18px 24px; contain: layout paint; }
.drawer .pair-balance-line { border: 1px solid var(--color-line); border-radius: var(--radius-md); overflow: hidden; margin-bottom: 16px; }
.drawer .pair-balance-line div { min-height: 64px; min-width: 0; overflow: hidden; }
.drawer .pair-balance-line strong { overflow-wrap: anywhere; }
.drawer .pair-balance-line .amount-with-currency.is-compact { white-space: normal; overflow-wrap: anywhere; }
.drawer .pair-balance-line .pair-cutoff,
.drawer .pair-balance-line strong.amount-neutral { white-space: nowrap; }
.drawer .pair-balance-line.is-six {
grid-template-columns: repeat(3, 1fr);
}
.drawer .pair-balance-line.is-six .pair-final { grid-column: auto; }
.drawer .pair-balance-line.is-six div:nth-child(4) { border-top: 1px solid var(--color-line); border-left: 0; }
.drawer .pair-balance-line.is-six div:nth-child(5),
.drawer .pair-balance-line.is-six div:nth-child(6) { border-top: 1px solid var(--color-line); }
.drawer .event-table { min-width: 860px; }
.drawer .event-table td { height: var(--row-h-evidence); }
.drawer .event-row { cursor: pointer; }
.drawer .event-row:focus-visible { outline: 2px solid var(--color-primary-strong); outline-offset: -2px; }
.drawer .evidence-stack { display: flex; flex-direction: column; gap: 10px; }
.drawer .evidence-midline { display: flex; align-items: center; gap: 10px; color: var(--color-ink-muted); font-size: 11px; }
.drawer .evidence-midline::before, .drawer .evidence-midline::after { content: ""; flex: 1; height: 1px; background: var(--color-line); }
.evidence-block.is-missing { border-color: rgba(102, 168, 255, 0.28); background: var(--color-info-wash); }
.evidence-block .evidence-grid { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 7px 14px; margin-top: 10px; }
.evidence-block .evidence-grid div { display: flex; flex-direction: column; min-width: 0; }
.evidence-block .evidence-grid dt { color: var(--color-ink-muted); font-size: 10px; }
.evidence-block .evidence-grid dd { color: var(--color-ink-soft); font-size: 12px; word-break: break-all; }
.evidence-block .evidence-masked { color: var(--color-ink-muted); font-style: normal; }
@media (max-width: 720px) { .evidence-block .evidence-grid { grid-template-columns: 1fr; } }
.company-row.is-balance-counterparty { grid-template-columns: 34px minmax(0, 1fr) minmax(0, auto); }
.company-row.is-balance-counterparty .company-row-figure {
display: flex; flex-direction: column; align-items: flex-end; gap: 4px; min-width: 0;
}
.company-row.is-balance-counterparty .company-row-figure strong { display: flex; flex-wrap: wrap; align-items: center; justify-content: flex-end; gap: 6px; }
@media (max-width: 900px) {
.company-row.is-balance-counterparty {
grid-template-columns: 34px minmax(0, 1fr);
row-gap: 8px;
}
.company-row.is-balance-counterparty .company-row-figure {
grid-column: 2;
align-items: flex-start;
text-align: left;
flex-wrap: wrap;
flex-direction: row;
column-gap: 10px;
row-gap: 4px;
}
}
/* Loading skeletons */
.skeleton-row { height: var(--row-h-evidence); display: flex; align-items: center; gap: 12px; padding: 0 14px; border-bottom: 1px solid var(--color-line); }
.skeleton-row span { height: 12px; border-radius: 6px; background: rgba(255, 255, 255, 0.06); }
.skeleton-row span:nth-child(1) { width: 22%; } .skeleton-row span:nth-child(2) { width: 14%; }
.skeleton-row span:nth-child(3) { width: 14%; } .skeleton-row span:nth-child(4) { width: 18%; }
.skeleton-row span:nth-child(5) { width: 12%; }
.skeleton-block { animation: skeleton-breathe 1200ms ease-in-out infinite; }
/* Empty / error / no-permission states */
.state-panel { display: flex; flex-direction: column; align-items: center; gap: 10px; justify-content: center; min-height: 240px; padding: 28px 18px; text-align: center; color: var(--color-ink-muted); }
.state-panel .state-icon { width: 40px; height: 40px; color: var(--color-ink-muted); opacity: 0.7; }
.state-panel strong { color: var(--color-ink-soft); font-size: 14px; }
.state-panel p { max-width: 420px; font-size: 12px; }
.state-panel .button { margin-top: 6px; }
.state-panel.is-error { border: 1px solid rgba(255, 98, 109, 0.3); border-radius: var(--radius-lg); background: var(--color-danger-wash); }
.state-panel.is-error strong { color: var(--color-danger); }
.state-panel.is-denied { border: 1px solid var(--color-line); border-radius: var(--radius-lg); background: var(--color-surface); }
/* Company portal alert variant + direction chips */
.balance-alert { border-color: rgba(255, 188, 82, 0.24); background: var(--color-warning-wash); }
.balance-alert > svg { color: var(--color-warning); }
.balance-alert .button { color: var(--color-warning); }
.company-row .result-direction { font-weight: 600; font-variant-numeric: tabular-nums; }
.subject-strip button { white-space: nowrap; }
/* Direction chips: text-first, arrows as affordance, never color-only */
.direction-chip { display: inline-flex; align-items: center; gap: 4px; min-height: 23px; padding: 2px 8px; border-radius: 8px; border: 1px solid var(--color-line); background: rgba(255, 255, 255, 0.045); color: var(--color-ink-soft); font-size: 10px; white-space: nowrap; }
@keyframes skeleton-breathe { 0%, 100% { opacity: 0.55; } 50% { opacity: 1; } }
@media (prefers-reduced-motion: reduce) { .skeleton-block { animation: none; } }