diff --git a/server.py b/server.py index d69217e..c334678 100644 --- a/server.py +++ b/server.py @@ -115,6 +115,9 @@ class AppHandler(SimpleHTTPRequestHandler): if path == "/api/company/match-exceptions": self._handle_company_match_exceptions(query) return + if path == "/api/company/workspace": + self._handle_company_workspace() + return company_event_match = re.fullmatch(r"/api/company/transfer-events/(\d+)", path) if company_event_match: self._handle_company_transfer_event_detail(int(company_event_match.group(1))) @@ -277,6 +280,12 @@ class AppHandler(SimpleHTTPRequestHandler): if path == "/api/company/manual-records": self._handle_company_manual_records_submit() return + company_confirm = re.fullmatch( + r"/api/company/transfer-events/(\d+)/confirm", path + ) + if company_confirm: + self._handle_company_transfer_confirm(int(company_confirm.group(1))) + return self._send_json(404, {"status": "error", "message": "接口不存在。"}) # ------------------------------------------------------------------ @@ -2195,6 +2204,8 @@ class AppHandler(SimpleHTTPRequestHandler): ) return { "event_id": row["event_id"], + "decision_id": row["decision_id"], + "revision": row["revision"], "classification": row["classification"], "pairing": row["pairing"], "status": matching.exposed_status(row), @@ -2209,6 +2220,20 @@ class AppHandler(SimpleHTTPRequestHandler): "evidence_count": row["evidence_count"], } + def _handle_company_workspace(self) -> None: + connection = connect(DB_PATH) + try: + user = self._require_user(connection) + if user is None: + return + if user["role"] != "company": + self._send_json(403, {"status": "error", "message": "该操作仅限公司用户。"}) + return + payload = matching.company_workspace_payload(connection, user["company_id"]) + self._send_json(200, {"status": "ok", **payload}) + finally: + connection.close() + def _handle_company_match_exceptions(self, query: dict[str, list[str]]) -> None: connection = connect(DB_PATH) try: @@ -2218,13 +2243,7 @@ class AppHandler(SimpleHTTPRequestHandler): if user["role"] != "company": self._send_json(403, {"status": "error", "message": "该操作仅限公司用户。"}) return - params: list[object] = [user["company_id"], user["company_id"]] - rows = connection.execute( - self._event_base_sql(True) - + "AND d.classification IN ('unresolved', 'needs_review') " - "ORDER BY d.id DESC LIMIT 500", - params, - ).fetchall() + rows = matching.company_pending_unilaterals(connection, user["company_id"]) items = [ self._company_event_list_item(row, user["company_id"]) for row in rows ] @@ -2232,6 +2251,109 @@ class AppHandler(SimpleHTTPRequestHandler): finally: connection.close() + def _handle_company_transfer_confirm(self, event_id: int) -> None: + """Company confirms a pending unilateral using existing assign_participant rules.""" + connection = connect(DB_PATH) + try: + user = self._require_user(connection) + if user is None: + return + if user["role"] != "company": + self._send_json(403, {"status": "error", "message": "该操作仅限公司用户。"}) + return + company_id = int(user["company_id"]) + data = self._read_json_body() + if data is None: + return + request_key = str(data.get("request_key") or "").strip() or None + # Idempotent replay: same request_key returns current workspace without re-decrement. + if request_key: + existing = connection.execute( + """ + SELECT id, actor_username + FROM transfer_match_decisions + WHERE idempotency_key = ? AND event_id = ? + """, + (request_key, event_id), + ).fetchone() + if existing is not None: + if existing["actor_username"] != user["username"]: + self._send_json(404, {"status": "error", "message": "待确认单边流水不存在或已处理。"}) + return + workspace = matching.company_workspace_payload(connection, company_id) + payload = matching._decision_payload(connection, event_id, existing["id"]) + self._send_json( + 200, + {"status": "ok", "decision": payload, "workspace": workspace}, + ) + return + pending = matching.company_pending_unilaterals(connection, company_id) + target = next((row for row in pending if int(row["event_id"]) == event_id), None) + if target is None: + self._send_json(404, {"status": "error", "message": "待确认单边流水不存在或已处理。"}) + return + try: + expected_revision = ( + int(data["expected_revision"]) + if data.get("expected_revision") is not None + else None + ) + except (TypeError, ValueError): + self._send_json(400, {"status": "error", "message": "expected_revision 参数无效。"}) + return + try: + counterparty_company_id = int(data["counterparty_company_id"]) + except (KeyError, TypeError, ValueError): + self._send_json( + 400, {"status": "error", "message": "counterparty_company_id 必须是有效的公司 id。"} + ) + return + if counterparty_company_id == company_id: + self._send_json(400, {"status": "error", "message": "对方公司不能是本公司。"}) + return + # Own side is already on the decision; assign the opposite role. + own_is_payer = target["payer_company_id"] == company_id + role = "payee" if own_is_payer else "payer" + if target["payer_company_id"] is None and target["payee_company_id"] is None: + self._send_json(400, {"status": "error", "message": "单边流水缺少本方参与方,无法确认。"}) + return + if not own_is_payer and target["payee_company_id"] != company_id: + self._send_json(404, {"status": "error", "message": "待确认单边流水不存在或已处理。"}) + return + reason = str(data.get("reason") or "").strip() or "公司端确认单边流水" + try: + payload = matching.apply_manual_decision( + connection, + event_id, + "assign_participant", + reason=reason, + expected_revision=expected_revision, + request_key=request_key, + actor=user, + participant={ + "role": role, + "company_id": counterparty_company_id, + }, + ) + except matching.MatchConflictError as exc: + self._send_json(409, {"status": "error", "message": str(exc)}) + return + 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 + workspace = matching.company_workspace_payload(connection, company_id) + self._send_json( + 200, + {"status": "ok", "decision": payload, "workspace": workspace}, + ) + finally: + connection.close() + def _handle_company_transfer_event_detail(self, event_id: int) -> None: connection = connect(DB_PATH) try: diff --git a/src/bank_importer/matching.py b/src/bank_importer/matching.py index e0f58da..c1aa1f1 100644 --- a/src/bank_importer/matching.py +++ b/src/bank_importer/matching.py @@ -1563,6 +1563,87 @@ def exposed_status(decision: sqlite3.Row | dict) -> str: return "unresolved" +def company_pending_unilaterals( + connection: sqlite3.Connection, company_id: int +) -> list[sqlite3.Row]: + """Authoritative pending unilateral queue for a company workspace. + + Same set as company match-exceptions: active events whose current decision + is ``unresolved`` / ``needs_review`` and the company participates. + Parentheses on the company filter avoid OR/AND precedence bugs. + """ + return connection.execute( + """ + SELECT c.event_id, d.id AS decision_id, d.revision, d.classification, + d.pairing, d.amount, d.currency, d.effective_at, d.mode, + d.locked, d.rule_version, d.created_at, + payer.company_id AS payer_company_id, + payee.company_id AS payee_company_id, + payer.bank_account_id AS payer_account_id, + payee.bank_account_id AS payee_account_id, + cpayer.name AS payer_company_name, + cpayee.name AS payee_company_name, + (SELECT COUNT(*) FROM transfer_decision_observations o + WHERE o.decision_id = d.id) AS evidence_count + FROM current_transfer_decisions c + JOIN transfer_match_decisions d ON d.id = c.decision_id + JOIN canonical_transfer_events e ON e.id = c.event_id + LEFT JOIN transfer_decision_participants payer + ON payer.decision_id = d.id AND payer.role = 'payer' + LEFT JOIN transfer_decision_participants payee + ON payee.decision_id = d.id AND payee.role = 'payee' + LEFT JOIN companies cpayer ON cpayer.id = payer.company_id + LEFT JOIN companies cpayee ON cpayee.id = payee.company_id + WHERE e.lifecycle = 'active' + AND d.classification IN ('unresolved', 'needs_review') + AND (payer.company_id = ? OR payee.company_id = ?) + ORDER BY d.id DESC + """, + (company_id, company_id), + ).fetchall() + + +def company_workspace_payload( + connection: sqlite3.Connection, company_id: int +) -> dict[str, object]: + """Workspace counts + list share one authoritative unilateral queue.""" + rows = company_pending_unilaterals(connection, company_id) + events = [ + { + "event_id": row["event_id"], + "decision_id": row["decision_id"], + "revision": row["revision"], + "classification": row["classification"], + "pairing": row["pairing"], + "status": exposed_status(row), + "amount": row["amount"], + "currency": row["currency"], + "effective_at": row["effective_at"], + "mode": row["mode"], + "locked": bool(row["locked"]), + "own_company_id": company_id, + "counterparty_company_id": ( + row["payee_company_id"] + if row["payer_company_id"] == company_id + else row["payer_company_id"] + ), + "counterparty_company_name": ( + row["payee_company_name"] + if row["payer_company_id"] == company_id + else row["payer_company_name"] + ), + "evidence_count": row["evidence_count"], + } + for row in rows + ] + count = len(events) + return { + "pending_unilateral": count, + "pending_total": count, + "unilateral_events": events, + } + + # --------------------------------------------------------------------------- # API payload helpers # --------------------------------------------------------------------------- diff --git a/tests/test_company_workspace.py b/tests/test_company_workspace.py new file mode 100644 index 0000000..bf78586 --- /dev/null +++ b/tests/test_company_workspace.py @@ -0,0 +1,313 @@ +"""HEL-168: company workspace pending unilateral counts sync after confirm.""" + +from __future__ import annotations + +import io +import os +from pathlib import Path +import tempfile +import threading +import unittest + +from openpyxl import Workbook + +import server +from test_server_auth import Client, as_json + +BOOTSTRAP_PASSWORD = "BootAdmin123" +ADMIN_PASSWORD = "AdminPass123" +CASHIER_PASSWORD = "Cashier123" + +CCB_HEADER = [ + "客户账号", "账户名称", "交易时间", "借方发生额(支取)", "贷方发生额(收入)", + "余额", "币种", "对方户名", "对方账号", "对方开户机构", "摘要", "备注", +] + +ACCOUNT_A = "6222000000001001" +ACCOUNT_B = "6222000000001002" + + +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) -> list: + return [own, "测试公司", at, amount, "", "50000.00", "RMB", "对方", cp, "某银行", "货款", ""] + + +class CompanyWorkspaceTodoTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + 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 + from bank_importer.db import connect, migrate + + connection = connect(cls.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 + + 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 + + cls.initial_passwords: dict[str, str] = {} + cls.company_a = cls._create_company("甲公司", "ws-cashier-a") + cls.company_b = cls._create_company("乙公司", "ws-cashier-b") + cls.cashier_a = cls._login_company("ws-cashier-a") + cls.cashier_b = cls._login_company("ws-cashier-b") + cls._approve_account(cls.cashier_a, ACCOUNT_A) + cls._approve_account(cls.cashier_b, ACCOUNT_B) + + @classmethod + def tearDownClass(cls) -> None: + 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() + + @classmethod + def _create_company(cls, name: str, username: str) -> int: + status, _, data = cls.admin.post_json( + "/api/admin/companies", + {"name": name, "username": username}, + ) + assert status == 200, data + payload = as_json(data) + cls.initial_passwords[username] = payload["initial_password"] + return payload["company_id"] + + @classmethod + def _login_company(cls, username: str) -> Client: + client = Client("127.0.0.1", cls.port) + initial = cls.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 + + @classmethod + def _approve_account(cls, company_client: Client, number: str) -> int: + status, _, data = company_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 = cls.admin.post_json( + f"/api/admin/accounts/{account_id}/review", + {"decision": "approve", "reason": "测试启用", "effective_from": "2026-01-01"}, + ) + assert status == 200, data + return account_id + + def _seed_unilateral(self, amount: str, at: str, cp_account: str = "9999999999999999") -> dict: + status, _, data = self.admin.post_multipart( + "/api/parse", + {"company_id": str(self.company_a)}, + f"单边-{amount}.xlsx", + workbook_bytes([outgoing(ACCOUNT_A, cp_account, amount, at)]), + ) + self.assertEqual(200, status, data) + batch_id = as_json(data)["batch_id"] + status, _, data = self.cashier_a.get(f"/api/batches/{batch_id}/sheets") + self.assertEqual(200, status, data) + names = [s["sheet_name"] for s in as_json(data)["sheets"] if s["outcome"] == "parsed"] + status, _, data = self.cashier_a.post_json( + f"/api/batches/{batch_id}/confirm", {"sheets": names} + ) + self.assertEqual(200, status, data) + + status, _, data = self.cashier_a.get("/api/company/workspace") + self.assertEqual(200, status, data) + events = [ + e for e in as_json(data)["unilateral_events"] if e["amount"] == amount + ] + self.assertEqual(1, len(events), data) + return events[0] + + def _workspace(self, client: Client) -> dict: + status, _, data = client.get("/api/company/workspace") + self.assertEqual(200, status, data) + payload = as_json(data) + self.assertEqual("ok", payload["status"]) + return payload + + def test_seed_three_confirm_sync_and_isolation(self) -> None: + # Seed 3 pending unilaterals for company A. + e1 = self._seed_unilateral("11.00", "2026-03-01 10:00:00") + e2 = self._seed_unilateral("22.00", "2026-03-02 10:00:00") + e3 = self._seed_unilateral("33.00", "2026-03-03 10:00:00") + seeded_ids = {e1["event_id"], e2["event_id"], e3["event_id"]} + + ws = self._workspace(self.cashier_a) + own = [e for e in ws["unilateral_events"] if e["event_id"] in seeded_ids] + self.assertEqual(3, len(own)) + self.assertEqual(ws["pending_unilateral"], ws["pending_total"]) + self.assertGreaterEqual(ws["pending_unilateral"], 3) + # 两处数字与列表同一权威口径 + self.assertEqual(ws["pending_unilateral"], len(ws["unilateral_events"])) + + # 另一公司看不到甲公司这 3 笔 + ws_b = self._workspace(self.cashier_b) + self.assertEqual( + [], + [e for e in ws_b["unilateral_events"] if e["event_id"] in seeded_ids], + ) + + # match-exceptions 与 workspace 同集合 + status, _, data = self.cashier_a.get("/api/company/match-exceptions") + self.assertEqual(200, status, data) + exceptions = as_json(data)["exceptions"] + exception_ids = {e["event_id"] for e in exceptions} + self.assertTrue(seeded_ids.issubset(exception_ids)) + + remaining = [e1, e2, e3] + for expected in (2, 1, 0): + event = remaining.pop(0) + status, _, data = self.cashier_a.post_json( + f"/api/company/transfer-events/{event['event_id']}/confirm", + { + "expected_revision": event["revision"], + "request_key": f"hel168-{event['event_id']}-ok", + "counterparty_company_id": self.company_b, + "reason": "公司端确认单边流水", + }, + ) + self.assertEqual(200, status, data) + body = as_json(data) + self.assertIn("workspace", body) + left = [ + e + for e in body["workspace"]["unilateral_events"] + if e["event_id"] in seeded_ids + ] + self.assertEqual(expected, len(left)) + self.assertEqual(body["workspace"]["pending_unilateral"], len(body["workspace"]["unilateral_events"])) + + # 刷新等价:重新 GET 仍为同步后的数字 + ws = self._workspace(self.cashier_a) + left = [e for e in ws["unilateral_events"] if e["event_id"] in seeded_ids] + self.assertEqual(expected, len(left)) + + # 重复确认:事件已离开待确认集合且 request_key 不同 → 404,数字不得误减 + status, _, data = self.cashier_a.post_json( + f"/api/company/transfer-events/{e1['event_id']}/confirm", + { + "expected_revision": e1["revision"], + "request_key": f"hel168-{e1['event_id']}-again", + "counterparty_company_id": self.company_b, + "reason": "公司端确认单边流水", + }, + ) + self.assertEqual(404, status, data) + ws = self._workspace(self.cashier_a) + left = [e for e in ws["unilateral_events"] if e["event_id"] in seeded_ids] + self.assertEqual(0, len(left)) + + # 同一 request_key 重放:幂等成功,数字不误减 + e4 = self._seed_unilateral("44.00", "2026-03-04 10:00:00") + key = f"hel168-{e4['event_id']}-replay" + status, _, data = self.cashier_a.post_json( + f"/api/company/transfer-events/{e4['event_id']}/confirm", + { + "expected_revision": e4["revision"], + "request_key": key, + "counterparty_company_id": self.company_b, + "reason": "公司端确认单边流水", + }, + ) + self.assertEqual(200, status, data) + after_first = self._workspace(self.cashier_a)["pending_unilateral"] + status, _, data = self.cashier_a.post_json( + f"/api/company/transfer-events/{e4['event_id']}/confirm", + { + "expected_revision": e4["revision"], + "request_key": key, + "counterparty_company_id": self.company_b, + "reason": "公司端确认单边流水", + }, + ) + self.assertEqual(200, status, data) + self.assertEqual(after_first, as_json(data)["workspace"]["pending_unilateral"]) + self.assertEqual(after_first, self._workspace(self.cashier_a)["pending_unilateral"]) + + # 失败请求不得误减:用不存在的事件 + before = self._workspace(self.cashier_a)["pending_unilateral"] + status, _, data = self.cashier_a.post_json( + "/api/company/transfer-events/999999/confirm", + { + "expected_revision": 1, + "request_key": "hel168-fail", + "counterparty_company_id": self.company_b, + }, + ) + self.assertEqual(404, status, data) + self.assertEqual(before, self._workspace(self.cashier_a)["pending_unilateral"]) + + # 乙公司不能确认甲公司事件(即使猜到 id) + status, _, data = self.cashier_b.post_json( + f"/api/company/transfer-events/{e2['event_id']}/confirm", + { + "expected_revision": e2["revision"], + "request_key": "hel168-cross", + "counterparty_company_id": self.company_a, + }, + ) + self.assertEqual(404, status, data) + + def test_admin_forbidden_on_company_workspace(self) -> None: + status, _, data = self.admin.get("/api/company/workspace") + self.assertEqual(403, status, data) + + +if __name__ == "__main__": + unittest.main() diff --git a/web/app.js b/web/app.js index 2a0690d..0e04cf5 100644 --- a/web/app.js +++ b/web/app.js @@ -2548,100 +2548,245 @@ async function loadImportBatches() { if (foot) foot.textContent = batches.length ? `共 ${batches.length} 个批次` : "暂无批次"; } -function refreshWorkspacePending() { - const card = $("#workspaceTodos"); - if (!card) return; - const count = $$(".list-row", card).length; - const status = $("#workspacePendingStatus"); - if (status) status.textContent = `${count} 项待处理`; +function formatWorkspaceAmount(amount, currency) { + const n = Number(amount); + const text = Number.isFinite(n) + ? n.toLocaleString("zh-CN", { minimumFractionDigits: 2, maximumFractionDigits: 2 }) + : String(amount ?? "—"); + return `${currency || "CNY"} ${text}`; } -function initReconcile() { - const matchCards = $$("[data-match-card]"); - const subjectRows = $$("[data-subject-row]"); - if (!matchCards.length && !subjectRows.length) return; +function applyCompanyWorkspace(payload) { + const events = Array.isArray(payload?.unilateral_events) ? payload.unilateral_events : []; + const pending = Number(payload?.pending_unilateral ?? events.length) || 0; + const total = Number(payload?.pending_total ?? pending) || 0; + state.workspace = { pending_unilateral: pending, pending_total: total, unilateral_events: events }; + + const cta = $("#workspaceUnilateralCta"); + if (cta) { + cta.textContent = pending ? `去确认单边流水 (${pending})` : "单边流水已全部确认"; + cta.classList.toggle("btn-primary", pending > 0); + } + + const status = $("#workspacePendingStatus"); + if (status) { + status.textContent = total ? `${total} 项待处理` : "已完成"; + status.className = `pill ${total ? "pill-warn" : "pill-success"}`; + } + + const sub = $("#workspaceTodoSub"); + if (sub) sub.textContent = total ? `权威待确认单边流水 ${pending} 笔` : "本月单边流水待办已清空"; + + const list = $("#workspaceTodoList"); + if (list) { + list.replaceChildren(); + if (!events.length) { + const empty = document.createElement("div"); + empty.className = "empty"; + empty.style.padding = "18px 16px"; + empty.innerHTML = '