From 51c301d9df8bcccb181af0ca7459a12db8ddaaf4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=85=BE=E8=AE=AFWorkBuddy?= Date: Sun, 16 Aug 2026 01:49:03 +0800 Subject: [PATCH] =?UTF-8?q?B-41:=20=E5=8A=A8=E6=80=81=E4=B8=BB=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E2=80=94=E2=80=94=E5=85=AC=E5=8F=B8=E3=80=81=E9=93=B6?= =?UTF-8?q?=E8=A1=8C=E8=B4=A6=E6=88=B7=E3=80=81=E5=88=AB=E5=90=8D=E4=B8=8E?= =?UTF-8?q?=E5=AE=A1=E6=A0=B8=E8=BD=A8=E8=BF=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 公司、公司账号、银行账户、账户类型、户名/账号别名和生效区间 全部来自数据库，新增公司无需改代码即可检索与使用。 - 公司端提交账户登记，总账端审核为启用/退回/停用；未启用账户 不参与所有权识别、上传或覆盖计算。 - 账号归一化、脱敏显示与数据库 UNIQUE 约束，并发提交不产生重复 有效账户；别名匹配有优先级和有效期。 - 所有主数据变更记录前值、后值、操作人、时间和原因。 - 决策记录见 docs/decisions/004-master-data.md。 --- docs/decisions/004-master-data.md | 47 +++ src/bank_importer/master_data.py | 570 ++++++++++++++++++++++++++++ tests/test_master_data.py | 593 ++++++++++++++++++++++++++++++ 3 files changed, 1210 insertions(+) create mode 100644 docs/decisions/004-master-data.md create mode 100644 src/bank_importer/master_data.py create mode 100644 tests/test_master_data.py diff --git a/docs/decisions/004-master-data.md b/docs/decisions/004-master-data.md new file mode 100644 index 0000000..eab87e2 --- /dev/null +++ b/docs/decisions/004-master-data.md @@ -0,0 +1,47 @@ +# 004 动态主数据技术决策 + +对应 Issue：B-41（`docs/issues/004-p1-dynamic-master-data.md`）。 + +## 账户状态机与生效区间 + +- `bank_accounts.status`：`pending → active → disabled`，审核退回为 + `returned`；退回后同公司重新提交复用同一行并回到 `pending`，不产生 + 第二条记录。停用这个动作只对 `active` 开放，退回只对 `pending` 开放， + 越界操作返回 409。 +- 生效区间 `effective_from` / `effective_to`（含当天）：公司提交时填写 + 期望启用日，管理员审核通过时确认；停用时写入 `effective_to`（默认当天）。 +- 两个判定函数分开： + - `is_usable`（上传选择、当期操作）：仅 `active` 且在生效区间内。 + - `is_identifiable`（历史流水所有权识别）：`active` 或 `disabled` 且在 + 生效区间内——停用只终止未来参与，不改写历史归属。 + +## 账号规范化与唯一性 + +- 账号入库前统一规范化：去空格/短横线、全角数字转半角，只存 6–32 位 + 数字的规范形式；唯一性由数据库 `UNIQUE(account_number)` 保证，并发 + 提交必然只有一个成功（有并发测试）。 +- 公司端任何响应只含脱敏形式 `****尾四位`；完整账号只在总账管理端 + （授权审计视图）返回。审计日志 detail 也只写脱敏账号。 + +## 别名匹配优先级 + +- `account_aliases` 支持户名（`name`）与账号（`account`）两类别名， + 各自带生效区间和 `priority` 列。 +- 匹配优先级确定：精确账号（0）< 账号别名（1000+priority）< + 户名别名（2000+priority）；同层内按 `priority` 再按账户 id 排序。 + 别名只在自身生效区间且账户 `is_identifiable` 时参与匹配。 + +## 审计轨迹 + +- 所有主数据变更写入 `master_data_changes`：实体、动作、前值 JSON、 + 后值 JSON、原因、操作人、时间。公司创建、账户提交/退回重提/ + 审核/停用、别名创建全部覆盖，管理员可按实体查询。 + +## 前端边界 + +- 公司下拉（往来查询、审核筛选、流水筛选、提醒、期初）和公司表格全部 + 由 `/api/admin/companies` 渲染，新增公司无需改代码即可被检索。 +- 公司端账户目录、上传账户选择、手工记录资金来源只来自 + `/api/company/accounts`，且只有 `usable` 账户进入上传选择。 +- 手工记录仍是浏览器 localStorage 演示数据（属 Issue 007/012 范围）， + 账户登记已完全切换到服务端。 diff --git a/src/bank_importer/master_data.py b/src/bank_importer/master_data.py new file mode 100644 index 0000000..71dee00 --- /dev/null +++ b/src/bank_importer/master_data.py @@ -0,0 +1,570 @@ +"""Dynamic master data: companies, bank accounts, aliases and audit trail. + +Companies, cashier logins and bank accounts live in the database instead of +being hard-coded in the UI. Bank account numbers are stored normalized +(digits only) under a database UNIQUE constraint, so two concurrent +submissions can never produce two usable accounts for the same number. +Company-side registrations are requests: only administrator-approved +(``active``) accounts inside their effective interval may identify ownership, +accept uploads or take part in coverage calculation. Every change is recorded +in ``master_data_changes`` with before/after values, actor, time and reason. +""" + +from __future__ import annotations + +from datetime import date, datetime, timezone +import json +import re +import sqlite3 + +from .db import utc_now + + +ACCOUNT_TYPES = ("基本户", "一般户", "专用户") +ACCOUNT_STATUSES = ("pending", "active", "returned", "disabled") +ALIAS_KINDS = ("name", "account") + +# Deterministic alias match priority: an exact account-number hit always +# beats an account alias, which always beats a name alias. The per-alias +# ``priority`` column only orders matches within the same tier. +PRIORITY_EXACT_ACCOUNT = 0 +PRIORITY_ACCOUNT_ALIAS = 1000 +PRIORITY_NAME_ALIAS = 2000 + +_ACCOUNT_NUMBER_PATTERN = re.compile(r"^[0-9]{6,32}$") +_DATE_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}$") +_FULL_WIDTH_DIGITS = str.maketrans("０１２３４５６７８９", "0123456789") + + +class ConflictError(ValueError): + """A uniqueness or ownership conflict (mapped to HTTP 409).""" + + +def utc_today() -> str: + return datetime.now(timezone.utc).date().isoformat() + + +def normalize_account_number(raw: object) -> str: + """Normalize an account number to digits only; raise ValueError if invalid. + + Spaces, dashes and full-width digits entered by cashiers or emitted by + bank exports collapse to one canonical form, which is what the UNIQUE + constraint and all matching operate on. + """ + text = str(raw or "").translate(_FULL_WIDTH_DIGITS) + text = re.sub(r"[\s\-‐-‒–—]+", "", text) + if not _ACCOUNT_NUMBER_PATTERN.fullmatch(text): + raise ValueError("银行账号须为 6-32 位数字（可含空格或短横线分隔）。") + return text + + +def mask_account_number(number: str) -> str: + """Masked display form; the full number stays server-side.""" + if len(number) <= 4: + return f"****{number}" + return f"****{number[-4:]}" + + +def validate_date(value: object, field: str, *, required: bool = False) -> str | None: + text = str(value or "").strip() + if not text: + if required: + raise ValueError(f"{field}不能为空。") + return None + if not _DATE_PATTERN.fullmatch(text): + raise ValueError(f"{field}须为 YYYY-MM-DD 格式。") + try: + date.fromisoformat(text) + except ValueError: + raise ValueError(f"{field}不是有效日期。") from None + return text + + +def validate_account_type(value: object) -> str: + text = str(value or "").strip() or "一般户" + if text not in ACCOUNT_TYPES: + raise ValueError(f"账户类型必须是：{'、'.join(ACCOUNT_TYPES)}。") + return text + + +def _snapshot(row: sqlite3.Row | None) -> dict[str, object] | None: + if row is None: + return None + return {key: row[key] for key in row.keys()} + + +def record_change( + connection: sqlite3.Connection, + entity_type: str, + entity_id: int, + action: str, + before: dict[str, object] | None, + after: dict[str, object] | None, + reason: str | None, + actor: sqlite3.Row | None, +) -> None: + """Append a before/after audit entry for a master data change.""" + connection.execute( + """ + INSERT INTO master_data_changes ( + entity_type, entity_id, action, before_json, after_json, + reason, actor_user_id, actor_username, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + entity_type, + entity_id, + action, + json.dumps(before, ensure_ascii=False) if before is not None else None, + json.dumps(after, ensure_ascii=False) if after is not None else None, + reason, + actor["id"] if actor is not None else None, + actor["username"] if actor is not None else None, + utc_now(), + ), + ) + + +# --------------------------------------------------------------------------- +# Companies +# --------------------------------------------------------------------------- + + +def create_company( + connection: sqlite3.Connection, + name: str, + credit_code: str | None, + cashier_name: str | None, + actor: sqlite3.Row | None, +) -> int: + """Create a company master record; returns the new id.""" + name = name.strip() + if not name: + raise ValueError("公司名称不能为空。") + now = utc_now() + try: + with connection: + cursor = connection.execute( + """ + INSERT INTO companies ( + name, credit_code, cashier_name, status, created_at, updated_at + ) VALUES (?, ?, ?, 'active', ?, ?) + """, + (name, (credit_code or "").strip() or None, + (cashier_name or "").strip() or None, now, now), + ) + except sqlite3.IntegrityError as exc: + raise ConflictError("公司名称已存在。") from exc + company_id = int(cursor.lastrowid) + with connection: + record_change( + connection, "company", company_id, "create", + None, {"name": name, "credit_code": credit_code or None, + "cashier_name": cashier_name or None, "status": "active"}, + None, actor, + ) + return company_id + + +# --------------------------------------------------------------------------- +# Bank accounts +# --------------------------------------------------------------------------- + + +def get_account(connection: sqlite3.Connection, account_id: int) -> sqlite3.Row | None: + return connection.execute( + "SELECT * FROM bank_accounts WHERE id = ?", (account_id,) + ).fetchone() + + +def submit_bank_account( + connection: sqlite3.Connection, + *, + company_id: int, + bank_name: str, + account_type: object, + account_number: object, + account_name: object = None, + start_date: object = None, + actor: sqlite3.Row | None, +) -> sqlite3.Row: + """Register an account for review; returns the resulting account row. + + A normalized number already present is never duplicated: resubmitting a + returned request from the same company reopens that same row as pending; + any other existing row is a conflict. The UNIQUE constraint on + ``account_number`` is the final guard for concurrent submissions. + """ + number = normalize_account_number(account_number) + bank = str(bank_name or "").strip() + if not bank: + raise ValueError("开户银行不能为空。") + kind = validate_account_type(account_type) + holder = str(account_name or "").strip() or None + requested_from = validate_date(start_date, "启用日期") + now = utc_now() + + existing = connection.execute( + "SELECT * FROM bank_accounts WHERE account_number = ?", (number,) + ).fetchone() + + if existing is None: + try: + with connection: + cursor = connection.execute( + """ + INSERT INTO bank_accounts ( + company_id, account_number, account_name, bank_name, + account_type, status, effective_from, submitted_by, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, 'pending', ?, ?, ?, ?) + """, + (company_id, number, holder, bank, kind, + requested_from, actor["id"] if actor else None, now, now), + ) + except sqlite3.IntegrityError as exc: + # Lost a concurrent-insert race on the UNIQUE constraint. + raise ConflictError("该银行账号已登记，请等待现有申请处理。") from exc + account_id = int(cursor.lastrowid) + with connection: + record_change( + connection, "bank_account", account_id, "submit", None, + {"company_id": company_id, "account_number": number, + "bank_name": bank, "account_type": kind, "status": "pending", + "effective_from": requested_from}, + None, actor, + ) + return get_account(connection, account_id) + + if existing["status"] == "returned" and existing["company_id"] == company_id: + before = _snapshot(existing) + with connection: + connection.execute( + """ + UPDATE bank_accounts + SET account_name = ?, bank_name = ?, account_type = ?, + status = 'pending', effective_from = ?, + submitted_by = ?, reviewed_by = NULL, reviewed_at = NULL, + review_reason = NULL, updated_at = ? + WHERE id = ? AND status = 'returned' + """, + (holder, bank, kind, requested_from, + actor["id"] if actor else None, now, existing["id"]), + ) + record_change( + connection, "bank_account", existing["id"], "resubmit", + before, {"account_number": number, "bank_name": bank, + "account_type": kind, "status": "pending", + "effective_from": requested_from}, + "退回后重新提交", actor, + ) + return get_account(connection, existing["id"]) + + if existing["company_id"] == company_id: + raise ConflictError("该银行账号已登记，请等待现有申请处理。") + raise ConflictError("该银行账号已被其他公司登记，请联系总账管理员核对。") + + +def review_bank_account( + connection: sqlite3.Connection, + account_id: int, + decision: str, + reason: str | None, + actor: sqlite3.Row, + *, + effective_from: object = None, + effective_to: object = None, +) -> sqlite3.Row: + """Approve, return or disable a registration; returns the updated row.""" + account = get_account(connection, account_id) + if account is None: + raise LookupError("账户不存在。") + reason = (reason or "").strip() or None + today = utc_today() + before = _snapshot(account) + + if decision == "approve": + if account["status"] != "pending": + raise ConflictError("只有待复核的账户可以审核通过。") + start = validate_date(effective_from, "启用日期") or account["effective_from"] or today + with connection: + connection.execute( + """ + UPDATE bank_accounts + SET status = 'active', effective_from = ?, effective_to = NULL, + reviewed_by = ?, reviewed_at = ?, review_reason = ?, updated_at = ? + WHERE id = ? + """, + (start, actor["id"], utc_now(), reason, utc_now(), account_id), + ) + record_change( + connection, "bank_account", account_id, "approve", before, + {"status": "active", "effective_from": start}, reason, actor, + ) + elif decision == "return": + if account["status"] != "pending": + raise ConflictError("只有待复核的账户可以退回。") + if reason is None: + raise ValueError("退回必须填写原因。") + with connection: + connection.execute( + """ + UPDATE bank_accounts + SET status = 'returned', reviewed_by = ?, reviewed_at = ?, + review_reason = ?, updated_at = ? + WHERE id = ? + """, + (actor["id"], utc_now(), reason, utc_now(), account_id), + ) + record_change( + connection, "bank_account", account_id, "return", before, + {"status": "returned"}, reason, actor, + ) + elif decision == "disable": + if account["status"] != "active": + raise ConflictError("只有已启用的账户可以停用。") + if reason is None: + raise ValueError("停用必须填写原因。") + end = validate_date(effective_to, "停用日期") or today + with connection: + connection.execute( + """ + UPDATE bank_accounts + SET status = 'disabled', effective_to = ?, + reviewed_by = ?, reviewed_at = ?, review_reason = ?, updated_at = ? + WHERE id = ? + """, + (end, actor["id"], utc_now(), reason, utc_now(), account_id), + ) + record_change( + connection, "bank_account", account_id, "disable", before, + {"status": "disabled", "effective_to": end}, reason, actor, + ) + else: + raise ValueError("审核决定必须是 approve、return 或 disable。") + return get_account(connection, account_id) + + +def in_effective_window(account: sqlite3.Row, on_date: str) -> bool: + """True when ``on_date`` falls inside the account's effective interval. + + ``effective_to`` is the last participating day (inclusive). + """ + if account["effective_from"] and on_date < account["effective_from"]: + return False + if account["effective_to"] and on_date > account["effective_to"]: + return False + return True + + +def is_usable(account: sqlite3.Row, on_date: str) -> bool: + """True when the account may accept uploads / be selected on a date.""" + return account["status"] == "active" and in_effective_window(account, on_date) + + +def is_identifiable(account: sqlite3.Row, on_date: str) -> bool: + """True when the account may resolve ownership for a transaction date. + + A disabled account keeps identifying historical rows inside its effective + window — disabling stops future participation, it never rewrites history. + """ + return account["status"] in ("active", "disabled") and in_effective_window( + account, on_date + ) + + +def list_accounts( + connection: sqlite3.Connection, + company_id: int | None = None, + status: str | None = None, +) -> list[sqlite3.Row]: + conditions: list[str] = [] + params: list[object] = [] + if company_id is not None: + conditions.append("a.company_id = ?") + params.append(company_id) + if status is not None: + if status not in ACCOUNT_STATUSES: + raise ValueError("无效的账户状态。") + conditions.append("a.status = ?") + params.append(status) + where = f"WHERE {' AND '.join(conditions)}" if conditions else "" + return connection.execute( + f""" + SELECT a.*, c.name AS company_name + FROM bank_accounts a + JOIN companies c ON c.id = a.company_id + {where} + ORDER BY a.id + """, + params, + ).fetchall() + + +def usable_accounts( + connection: sqlite3.Connection, company_id: int, on_date: str | None = None +) -> list[sqlite3.Row]: + """Accounts allowed to upload / identify ownership for the company today.""" + day = on_date or utc_today() + return [ + account + for account in list_accounts(connection, company_id=company_id) + if is_usable(account, day) + ] + + +# --------------------------------------------------------------------------- +# Aliases +# --------------------------------------------------------------------------- + + +def add_alias( + connection: sqlite3.Connection, + account_id: int, + alias_kind: str, + alias_value: object, + priority: object = None, + effective_from: object = None, + effective_to: object = None, + actor: sqlite3.Row | None = None, +) -> int: + """Attach a name/account alias with an effective interval; returns the id.""" + if get_account(connection, account_id) is None: + raise LookupError("账户不存在。") + if alias_kind not in ALIAS_KINDS: + raise ValueError("别名类型必须是 name 或 account。") + if alias_kind == "account": + value = normalize_account_number(alias_value) + else: + value = re.sub(r"\s+", "", str(alias_value or "")) + if not value: + raise ValueError("户名别名不能为空。") + try: + rank = int(priority) if priority not in (None, "") else 100 + except (TypeError, ValueError): + raise ValueError("优先级必须是整数。") from None + start = validate_date(effective_from, "别名生效日期") + end = validate_date(effective_to, "别名失效日期") + if start and end and end < start: + raise ValueError("别名失效日期不能早于生效日期。") + try: + with connection: + cursor = connection.execute( + """ + INSERT INTO account_aliases ( + bank_account_id, alias_kind, alias_value, priority, + effective_from, effective_to, created_by, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + (account_id, alias_kind, value, rank, start, end, + actor["id"] if actor else None, utc_now()), + ) + except sqlite3.IntegrityError as exc: + raise ConflictError("该账户下相同别名已存在。") from exc + alias_id = int(cursor.lastrowid) + with connection: + record_change( + connection, "account_alias", alias_id, "create", None, + {"bank_account_id": account_id, "alias_kind": alias_kind, + "alias_value": value, "priority": rank, + "effective_from": start, "effective_to": end}, + None, actor, + ) + return alias_id + + +def list_aliases(connection: sqlite3.Connection, account_id: int) -> list[sqlite3.Row]: + return connection.execute( + "SELECT * FROM account_aliases WHERE bank_account_id = ? ORDER BY id", + (account_id,), + ).fetchall() + + +def match_account( + connection: sqlite3.Connection, + *, + account_number: object = None, + name: object = None, + on_date: str | None = None, +) -> list[dict[str, object]]: + """Resolve an observed counterparty to operating, in-window accounts. + + Deterministic priority: exact account number < account alias < name + alias; within one tier the alias ``priority`` column orders the hits. + Pending and returned accounts never match; disabled accounts still match + transaction dates inside their effective window (see ``is_identifiable``). + """ + day = on_date or utc_today() + hits: list[tuple[int, sqlite3.Row, str]] = [] + + number = str(account_number or "").strip() + if number: + try: + normalized = normalize_account_number(number) + except ValueError: + normalized = None + if normalized is not None: + exact = connection.execute( + "SELECT * FROM bank_accounts WHERE account_number = ?", + (normalized,), + ).fetchone() + if exact is not None and is_identifiable(exact, day): + hits.append((PRIORITY_EXACT_ACCOUNT, exact, "account_exact")) + alias_rows = connection.execute( + """ + SELECT a.*, al.priority AS alias_priority, + al.effective_from AS alias_from, al.effective_to AS alias_to + FROM account_aliases al + JOIN bank_accounts a ON a.id = al.bank_account_id + WHERE al.alias_kind = 'account' AND al.alias_value = ? + """, + (normalized,), + ).fetchall() + for alias in alias_rows: + if _alias_in_window(alias, day) and is_identifiable(alias, day): + hits.append( + (PRIORITY_ACCOUNT_ALIAS + alias["alias_priority"], alias, "account_alias") + ) + + normalized_name = re.sub(r"\s+", "", str(name or "")) + if normalized_name: + alias_rows = connection.execute( + """ + SELECT a.*, al.priority AS alias_priority, + al.effective_from AS alias_from, al.effective_to AS alias_to + FROM account_aliases al + JOIN bank_accounts a ON a.id = al.bank_account_id + WHERE al.alias_kind = 'name' AND al.alias_value = ? + """, + (normalized_name,), + ).fetchall() + for alias in alias_rows: + if _alias_in_window(alias, day) and is_identifiable(alias, day): + hits.append( + (PRIORITY_NAME_ALIAS + alias["alias_priority"], alias, "name_alias") + ) + + hits.sort(key=lambda item: (item[0], item[1]["id"])) + seen: set[int] = set() + results: list[dict[str, object]] = [] + for rank, account, via in hits: + if account["id"] in seen: + continue + seen.add(account["id"]) + results.append( + { + "bank_account_id": account["id"], + "company_id": account["company_id"], + "via": via, + "priority": rank, + } + ) + return results + + +def _alias_in_window(alias: sqlite3.Row, on_date: str) -> bool: + if alias["alias_from"] and on_date < alias["alias_from"]: + return False + if alias["alias_to"] and on_date > alias["alias_to"]: + return False + return True diff --git a/tests/test_master_data.py b/tests/test_master_data.py new file mode 100644 index 0000000..1ea5b55 --- /dev/null +++ b/tests/test_master_data.py @@ -0,0 +1,593 @@ +"""Tests for dynamic master data (companies, bank accounts, aliases). + +Unit tests exercise normalization, masking, effective-window boundaries and +alias-match priority against an in-memory database; the HTTP integration +class drives a live server to cover the submission -> review -> disable +workflow, uniqueness under concurrency, masking boundaries and RBAC. +""" + +from __future__ import annotations + +from pathlib import Path +import tempfile +import threading +import unittest + +from bank_importer import auth, master_data +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" + + +class MasterDataUnitTests(unittest.TestCase): + def setUp(self) -> None: + self.connection = connect(":memory:") + migrate(self.connection) + now = master_data.utc_now() + cursor = self.connection.execute( + "INSERT INTO companies (name, created_at, updated_at) VALUES ('甲公司', ?, ?)", + (now, now), + ) + self.company_id = int(cursor.lastrowid) + self.connection.commit() + + def tearDown(self) -> None: + self.connection.close() + + def _active_account(self, number: str = "1234567890123") -> object: + return master_data.submit_bank_account( + self.connection, + company_id=self.company_id, + bank_name="中信银行", + account_type="基本户", + account_number=number, + start_date="2026-01-01", + actor=None, + ) + + def test_normalize_account_number(self) -> None: + self.assertEqual("1234567890", master_data.normalize_account_number("1234 5678-90")) + self.assertEqual("1234567890", master_data.normalize_account_number("１２３４５６７８９０")) + for invalid in ("", "123", "abc123456", "12345678901234567890123456789012345"): + with self.assertRaises(ValueError): + master_data.normalize_account_number(invalid) + + def test_mask_account_number(self) -> None: + self.assertEqual("****9012", master_data.mask_account_number("123456789012")) + self.assertEqual("****123", master_data.mask_account_number("123")) + + def test_is_usable_window_boundaries(self) -> None: + account = self._active_account() + account = master_data.review_bank_account( + self.connection, account["id"], "approve", None, self._admin(), + effective_from="2026-02-01", + ) + self.assertFalse(master_data.is_usable(account, "2026-01-31")) + self.assertTrue(master_data.is_usable(account, "2026-02-01")) + # A scheduled effective_to bounds the usable window (inclusive). + with self.connection: + self.connection.execute( + "UPDATE bank_accounts SET effective_to = '2026-06-30' WHERE id = ?", + (account["id"],), + ) + account = master_data.get_account(self.connection, account["id"]) + self.assertTrue(master_data.is_usable(account, "2026-06-30")) + self.assertFalse(master_data.is_usable(account, "2026-07-01")) + # Disabled accounts never accept new uploads, even inside the window. + account = master_data.review_bank_account( + self.connection, account["id"], "disable", "账户销户", self._admin(), + effective_to="2026-06-30", + ) + self.assertFalse(master_data.is_usable(account, "2026-06-30")) + # …but they still identify historical rows inside the window. + self.assertTrue(master_data.is_identifiable(account, "2026-06-30")) + self.assertFalse(master_data.is_identifiable(account, "2026-07-01")) + + def _admin(self): + user = self.connection.execute( + "SELECT * FROM users WHERE username = 'admin-u'" + ).fetchone() + if user is None: + auth.create_user(self.connection, "admin-u", "AdminPass123", "admin") + user = self.connection.execute( + "SELECT * FROM users WHERE username = 'admin-u'" + ).fetchone() + return user + + def test_pending_and_returned_accounts_never_usable(self) -> None: + account = master_data.submit_bank_account( + self.connection, + company_id=self.company_id, + bank_name="中信银行", + account_type="一般户", + account_number="9988776655", + start_date="2026-01-01", + actor=None, + ) + self.assertFalse(master_data.is_usable(account, "2026-06-01")) + account = master_data.review_bank_account( + self.connection, account["id"], "return", "资料不全", self._admin() + ) + self.assertEqual("returned", account["status"]) + self.assertFalse(master_data.is_usable(account, "2026-06-01")) + + def test_alias_match_priority_and_window(self) -> None: + exact_account = self._active_account("1111222233334") + exact_account = master_data.review_bank_account( + self.connection, exact_account["id"], "approve", None, self._admin(), + effective_from="2026-01-01", + ) + alias_account = master_data.submit_bank_account( + self.connection, + company_id=self.company_id, + bank_name="建设银行", + account_type="一般户", + account_number="5555666677778", + start_date="2026-01-01", + actor=None, + ) + alias_account = master_data.review_bank_account( + self.connection, alias_account["id"], "approve", None, self._admin(), + effective_from="2026-01-01", + ) + master_data.add_alias( + self.connection, alias_account["id"], "account", "9999000011112", + priority=5, actor=None, + ) + master_data.add_alias( + self.connection, alias_account["id"], "name", "甲公司郑州分部", + effective_from="2026-03-01", effective_to="2026-03-31", actor=None, + ) + + # Exact account number beats the account alias tier. + master_data.add_alias( + self.connection, alias_account["id"], "account", "1111222233334", + priority=1, actor=None, + ) + hits = master_data.match_account( + self.connection, account_number="1111-2222 3333 4", on_date="2026-06-01" + ) + self.assertEqual(exact_account["id"], hits[0]["bank_account_id"]) + self.assertEqual("account_exact", hits[0]["via"]) + self.assertEqual(alias_account["id"], hits[1]["bank_account_id"]) + self.assertEqual("account_alias", hits[1]["via"]) + + # Name alias only matches inside its effective window. + hits = master_data.match_account( + self.connection, name="甲公司郑州分部 ", on_date="2026-03-15" + ) + self.assertEqual(alias_account["id"], hits[0]["bank_account_id"]) + self.assertEqual("name_alias", hits[0]["via"]) + self.assertEqual( + [], + master_data.match_account( + self.connection, name="甲公司郑州分部", on_date="2026-04-01" + ), + ) + + def test_match_never_returns_pending_or_disabled(self) -> None: + pending = master_data.submit_bank_account( + self.connection, + company_id=self.company_id, + bank_name="郑州银行", + account_type="一般户", + account_number="7777888899990", + actor=None, + ) + self.assertEqual( + [], + master_data.match_account(self.connection, account_number="7777888899990"), + ) + account = master_data.review_bank_account( + self.connection, pending["id"], "approve", None, self._admin(), + effective_from="2026-01-01", + ) + self.assertTrue( + master_data.match_account(self.connection, account_number="7777888899990") + ) + master_data.review_bank_account( + self.connection, account["id"], "disable", "销户", self._admin(), + effective_to="2026-06-30", + ) + # Disabled accounts still resolve dates inside their effective window… + self.assertTrue( + master_data.match_account( + self.connection, account_number="7777888899990", on_date="2026-06-30" + ) + ) + # …and never resolve dates after it. + self.assertEqual( + [], + master_data.match_account( + self.connection, account_number="7777888899990", on_date="2026-07-01" + ), + ) + + +class MasterDataApiTests(unittest.TestCase): + """Live-server workflow tests for account registration and review.""" + + @classmethod + def setUpClass(cls) -> None: + import os + + cls.temp_dir = tempfile.TemporaryDirectory() + root = Path(cls.temp_dir.name) + cls.db_path = root / "app.db" + cls.storage = root / "files" + + cls._old_db_path = server.DB_PATH + cls._old_storage = server.STORAGE_DIR + server.DB_PATH = cls.db_path + server.STORAGE_DIR = cls.storage + + os.environ["APP_BOOTSTRAP_ADMIN_PASSWORD"] = BOOTSTRAP_PASSWORD + connection = connect(cls.db_path) + migrate(connection) + server.ensure_bootstrap_admin(connection) + connection.close() + + class QuietHandler(server.AppHandler): + def log_message(self, *args) -> None: + pass + + cls.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler) + cls.port = cls.httpd.server_address[1] + cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True) + cls.thread.start() + + cls.admin = Client("127.0.0.1", cls.port) + status, _, data = cls.admin.post_json( + "/api/login", + {"username": "group-admin", "password": BOOTSTRAP_PASSWORD, "portal": "admin"}, + ) + assert status == 200, data + status, _, data = cls.admin.post_json( + "/api/password/change", + {"old_password": BOOTSTRAP_PASSWORD, "new_password": ADMIN_PASSWORD}, + ) + assert status == 200, data + + # Company A created together with its cashier login in one call. + status, _, data = cls.admin.post_json( + "/api/admin/companies", + {"name": "甲公司", "credit_code": "91410100TEST", "cashier_name": "牛女士", + "username": "cashier-a"}, + ) + assert status == 200, data + payload = as_json(data) + cls.company_a = payload["company_id"] + assert payload["initial_password"] == "cashier-a" + + status, _, data = cls.admin.post_json("/api/admin/companies", {"name": "乙公司"}) + assert status == 200, data + cls.company_b = as_json(data)["company_id"] + status, _, data = cls.admin.post_json( + "/api/admin/users", {"username": "cashier-b", "company_id": cls.company_b} + ) + assert status == 200, data + + cls.cashier_a = cls._login_company_user("cashier-a", "cashier-a") + cls.cashier_b = cls._login_company_user("cashier-b", "cashier-b") + + @classmethod + def _login_company_user(cls, username: str, initial: str) -> Client: + client = Client("127.0.0.1", cls.port) + 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 + + @classmethod + def tearDownClass(cls) -> None: + import os + + cls.httpd.shutdown() + cls.httpd.server_close() + server.DB_PATH = cls._old_db_path + server.STORAGE_DIR = cls._old_storage + os.environ.pop("APP_BOOTSTRAP_ADMIN_PASSWORD", None) + cls.temp_dir.cleanup() + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + def submit_account(self, client: Client, number: str, bank: str = "郑州银行"): + return client.post_json( + "/api/company/accounts", + {"bank_name": bank, "account_type": "一般户", + "account_number": number, "start_date": "2026-08-01"}, + ) + + def review(self, account_id: int, decision: str, reason: str = ""): + return self.admin.post_json( + f"/api/admin/accounts/{account_id}/review", + {"decision": decision, "reason": reason}, + ) + + # ------------------------------------------------------------------ + # Company management + # ------------------------------------------------------------------ + + def test_new_company_appears_in_admin_list_with_user(self) -> None: + status, _, data = self.admin.get("/api/admin/companies") + self.assertEqual(200, status) + companies = {c["name"]: c for c in as_json(data)["companies"]} + self.assertIn("甲公司", companies) + self.assertEqual("91410100TEST", companies["甲公司"]["credit_code"]) + self.assertEqual("牛女士", companies["甲公司"]["cashier_name"]) + self.assertIn("cashier-a", companies["甲公司"]["usernames"]) + self.assertEqual(0, companies["乙公司"]["account_count"]) + + def test_company_user_cannot_change_own_company_binding(self) -> None: + # A company_id in the submission body is ignored: the account is + # always bound to the session company. + status, _, data = self.cashier_a.post_json( + "/api/company/accounts", + {"bank_name": "民生银行", "account_type": "一般户", + "account_number": "6001000100010001", "company_id": self.company_b}, + ) + self.assertEqual(200, status, data) + self.assertEqual(self.company_a, as_json(data)["account"]["company_id"]) + + # ------------------------------------------------------------------ + # Account registration -> review workflow + # ------------------------------------------------------------------ + + def test_full_registration_workflow(self) -> None: + status, _, data = self.submit_account(self.cashier_a, "6222 0210-0100 8888") + self.assertEqual(200, status, data) + account = as_json(data)["account"] + account_id = account["id"] + self.assertEqual("pending", account["status"]) + # Company view is masked only. + self.assertEqual("****8888", account["account_number_masked"]) + self.assertNotIn("account_number", account) + self.assertFalse(account.get("usable", True)) + + # Admin audit view shows the full normalized number. + status, _, data = self.admin.get("/api/admin/accounts") + self.assertEqual(200, status) + full = [a for a in as_json(data)["accounts"] if a["id"] == account_id][0] + self.assertEqual("62220210010088 88".replace(" ", ""), full["account_number"]) + self.assertEqual("甲公司", full["company_name"]) + + # Pending account cannot upload. + sample = Path("流水模板/中信银行账户流水.xlsx") + status, _, data = self.cashier_a.post_multipart( + "/api/parse", {"bank_account_id": str(account_id)}, + sample.name, sample.read_bytes(), + ) + self.assertEqual(409, status, data) + + # Approve -> active and usable. + status, _, data = self.review(account_id, "approve") + self.assertEqual(200, status, data) + self.assertEqual("active", as_json(data)["account"]["status"]) + self.assertEqual("2026-08-01", as_json(data)["account"]["effective_from"]) + + status, _, data = self.cashier_a.get("/api/company/accounts") + mine = [a for a in as_json(data)["accounts"] if a["id"] == account_id][0] + self.assertTrue(mine["usable"]) + + # Approved account accepts uploads. + status, _, data = self.cashier_a.post_multipart( + "/api/parse", {"bank_account_id": str(account_id)}, + sample.name, sample.read_bytes(), + ) + self.assertEqual(200, status, data) + + # Disable -> leaves the usable window. + status, _, data = self.review(account_id, "disable", "账户销户") + self.assertEqual(200, status, data) + disabled = as_json(data)["account"] + self.assertEqual("disabled", disabled["status"]) + self.assertIsNotNone(disabled["effective_to"]) + + status, _, data = self.cashier_a.get("/api/company/accounts") + mine = [a for a in as_json(data)["accounts"] if a["id"] == account_id][0] + self.assertFalse(mine["usable"]) + + def test_return_and_resubmit_reuses_same_row(self) -> None: + status, _, data = self.submit_account(self.cashier_a, "3100998877665") + self.assertEqual(200, status, data) + account_id = as_json(data)["account"]["id"] + + status, _, data = self.review(account_id, "return", "开户许可证模糊") + self.assertEqual(200, status, data) + self.assertEqual("returned", as_json(data)["account"]["status"]) + + # Return requires a reason. + status, _, data = self.submit_account(self.cashier_a, "3100998877666") + other_id = as_json(data)["account"]["id"] + status, _, _ = self.review(other_id, "return") + self.assertEqual(400, status) + + # Resubmission reopens the same row as pending. + status, _, data = self.submit_account(self.cashier_a, "3100 9988-7766 5") + self.assertEqual(200, status, data) + resubmitted = as_json(data)["account"] + self.assertEqual(account_id, resubmitted["id"]) + self.assertEqual("pending", resubmitted["status"]) + + # The whole trail is auditable with actor, time and reason. + status, _, data = self.admin.get( + f"/api/admin/master-changes?entity_type=bank_account&entity_id={account_id}" + ) + self.assertEqual(200, status) + changes = as_json(data)["changes"] + actions = [change["action"] for change in changes] + self.assertEqual(["resubmit", "return", "submit"], actions) + returned = [c for c in changes if c["action"] == "return"][0] + self.assertEqual("开户许可证模糊", returned["reason"]) + self.assertEqual("group-admin", returned["actor_username"]) + self.assertIn('"pending"', returned["before_json"]) + self.assertIn('"returned"', returned["after_json"]) + + def test_duplicate_normalized_number_conflicts(self) -> None: + status, _, _ = self.submit_account(self.cashier_a, "4501111222233") + self.assertEqual(200, status) + # Same company, same number with different separators. + status, _, data = self.submit_account(self.cashier_a, "4501-1112 2223-3") + self.assertEqual(409, status, data) + # Another company registering the same number conflicts too. + status, _, data = self.submit_account(self.cashier_b, "4501111222233") + self.assertEqual(409, status, data) + self.assertIn("其他公司", as_json(data)["message"]) + + def test_concurrent_submissions_create_only_one_account(self) -> None: + number = "8800123456789" + results: list[int] = [] + + def submit() -> None: + client = Client("127.0.0.1", self.port) + client.cookies.update(self.cashier_a.cookies) + status, _, _ = client.post_json( + "/api/company/accounts", + {"bank_name": "中信银行", "account_type": "基本户", + "account_number": number}, + ) + results.append(status) + + threads = [threading.Thread(target=submit) for _ in range(2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + self.assertEqual(sorted(results), [200, 409]) + + connection = connect(self.db_path) + try: + count = connection.execute( + "SELECT COUNT(*) AS n FROM bank_accounts WHERE account_number = ?", + (number,), + ).fetchone()["n"] + finally: + connection.close() + self.assertEqual(1, count) + + # ------------------------------------------------------------------ + # RBAC and masking + # ------------------------------------------------------------------ + + def test_company_user_forbidden_on_admin_account_endpoints(self) -> None: + for call in ( + lambda: self.cashier_a.get("/api/admin/accounts"), + lambda: self.cashier_a.post_json( + "/api/admin/accounts/1/review", {"decision": "approve"} + ), + lambda: self.cashier_a.get("/api/admin/accounts/1/aliases"), + lambda: self.cashier_a.post_json( + "/api/admin/accounts/1/aliases", + {"alias_kind": "name", "alias_value": "x"}, + ), + lambda: self.cashier_a.get("/api/admin/master-changes"), + ): + status, _, data = call() + self.assertEqual(403, status, data) + + def test_admin_cannot_use_company_account_endpoint(self) -> None: + status, _, _ = self.admin.get("/api/company/accounts") + self.assertEqual(403, status) + status, _, _ = self.submit_account(self.admin, "1000200030004") + self.assertEqual(403, status) + + def test_unauthenticated_master_data_calls_return_401(self) -> None: + anon = Client("127.0.0.1", self.port) + for call in ( + lambda: anon.get("/api/company/accounts"), + lambda: anon.get("/api/admin/accounts"), + lambda: anon.post_json("/api/company/accounts", {}), + lambda: anon.get("/api/admin/master-changes"), + ): + status, _, _ = call() + self.assertEqual(401, status) + + def test_company_account_list_scoped_and_masked(self) -> None: + status, _, data = self.cashier_b.get("/api/company/accounts") + self.assertEqual(200, status) + for account in as_json(data)["accounts"]: + self.assertEqual(self.company_b, account["company_id"]) + self.assertNotIn("account_number", account) + self.assertTrue(account["account_number_masked"].startswith("****")) + + def test_upload_with_other_companys_account_returns_404(self) -> None: + status, _, data = self.submit_account(self.cashier_a, "5550001112223") + account_id = as_json(data)["account"]["id"] + status, _, _ = self.review(account_id, "approve") + self.assertEqual(200, status) + sample = Path("流水模板/中信银行账户流水.xlsx") + status, _, data = self.cashier_b.post_multipart( + "/api/parse", {"bank_account_id": str(account_id)}, + sample.name, sample.read_bytes(), + ) + self.assertEqual(404, status, data) + + # ------------------------------------------------------------------ + # Aliases via API + # ------------------------------------------------------------------ + + def test_alias_crud_and_audit(self) -> None: + status, _, data = self.submit_account(self.cashier_a, "6601234509876") + account_id = as_json(data)["account"]["id"] + status, _, _ = self.review(account_id, "approve") + self.assertEqual(200, status) + + status, _, data = self.admin.post_json( + f"/api/admin/accounts/{account_id}/aliases", + {"alias_kind": "name", "alias_value": "甲公司工会", + "effective_from": "2026-01-01", "effective_to": "2026-12-31", + "priority": 10}, + ) + self.assertEqual(200, status, data) + alias_id = as_json(data)["alias_id"] + + # Duplicate alias on the same account conflicts. + status, _, _ = self.admin.post_json( + f"/api/admin/accounts/{account_id}/aliases", + {"alias_kind": "name", "alias_value": "甲公司工会"}, + ) + self.assertEqual(409, status) + + status, _, data = self.admin.get(f"/api/admin/accounts/{account_id}/aliases") + self.assertEqual(200, status) + aliases = as_json(data)["aliases"] + self.assertEqual(1, len(aliases)) + self.assertEqual("甲公司工会", aliases[0]["alias_value"]) + self.assertEqual(10, aliases[0]["priority"]) + + status, _, data = self.admin.get( + f"/api/admin/master-changes?entity_type=account_alias&entity_id={alias_id}" + ) + changes = as_json(data)["changes"] + self.assertEqual("create", changes[0]["action"]) + self.assertIn("2026-12-31", changes[0]["after_json"]) + + # Invalid alias input is rejected, unknown account is 404. + status, _, _ = self.admin.post_json( + f"/api/admin/accounts/{account_id}/aliases", + {"alias_kind": "account", "alias_value": "abc"}, + ) + self.assertEqual(400, status) + status, _, _ = self.admin.post_json( + "/api/admin/accounts/99999/aliases", + {"alias_kind": "name", "alias_value": "x"}, + ) + self.assertEqual(404, status) + + +if __name__ == "__main__": + unittest.main()