From 5ed34c2b8f4901e0c2b406969ddcd89cdf9d6df1 Mon Sep 17 00:00:00 2001 From: caoxiaozhu Date: Tue, 14 Jul 2026 10:09:09 +0800 Subject: [PATCH] feat(expenses): backfill historical claims into expense cases --- .../CONCEPT.md | 10 + .../TODO.md | 6 +- .../backfill_legacy_expense_claim_cases.py | 532 ++++++++++++++++++ .../src/app/db/maintenance_database_target.py | 237 ++++++++ .../services/expense_case_legacy_backfill.py | 391 +++++++++++++ server/src/app/services/expense_cases.py | 3 +- ...backfill_legacy_expense_claim_cases_cli.py | 183 ++++++ .../test_expense_case_legacy_backfill.py | 385 +++++++++++++ .../tests/test_maintenance_database_target.py | 225 ++++++++ web/src/utils/expenseCaseTimeline.js | 26 +- web/tests/expense-case-timeline.test.mjs | 58 ++ 11 files changed, 2053 insertions(+), 3 deletions(-) create mode 100644 server/scripts/backfill_legacy_expense_claim_cases.py create mode 100644 server/src/app/db/maintenance_database_target.py create mode 100644 server/src/app/services/expense_case_legacy_backfill.py create mode 100644 server/tests/test_backfill_legacy_expense_claim_cases_cli.py create mode 100644 server/tests/test_expense_case_legacy_backfill.py create mode 100644 server/tests/test_maintenance_database_target.py diff --git a/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md b/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md index 29f2406..f093f4a 100644 --- a/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md +++ b/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md @@ -221,6 +221,7 @@ #### 最小事件词典 - `expense_case_created` +- `historical_claim_imported`：迁移前旧单的当前快照；只证明已纳入统一费用事件，不重建或伪造迁移前审批历史。 - `application_generated` / `application_submitted` / `application_approved` - `receipt_received` / `receipt_verified` / `ocr_corrected` - `field_suggested` / `field_accepted` / `field_edited` / `field_rejected` @@ -243,6 +244,9 @@ #### 兼容策略 - 迁移初期可用 shadow 事件校验现有 `ExpenseClaim` 的映射；某个状态一旦正式纳入事件模型，其业务写入与 Outbox 事件必须进入同一事务。 +- 迁移前已有 `ExpenseClaim` 通过显式维护命令写入一条 `historical_claim_imported` 快照事件：事件发生时间使用真实回填时间，原创建、发生、提交和更新时间只进入内部 payload；`history_reconstructed=false`，不得按当前状态反推并伪造历史提交、审批或付款动作。 +- 历史回填默认 dry-run，必须显式提供租户、创建时间边界和数据库目标；apply 还要求精确目标确认、迁移 head、advisory lock 和分批事务。稳定幂等键、源快照指纹、已有 Link 跳过及孤立 Event 冲突拒绝共同保证可审计重跑。 +- 历史快照事件使用 `delivery_status=suppressed`，供时间线和分析读取，但不进入实时 Outbox 投递；用户态 API 仍只返回事件白名单，不暴露回填批次、源指纹、幂等键和投递状态。 - 旧 `ReimbursementRequest` 进入只读兼容和迁移状态，停止新增第二套业务编排。 - 现有 `risk_flags_json` 保持读取兼容，新审批、付款、归档和关系事件写入结构化表。 - API 新字段优先追加，不在同一阶段破坏现有前端契约。 @@ -370,6 +374,7 @@ customer_contribution_margin ### 后端 - Expense Case 状态机、关系绑定、幂等和非法状态跃迁单元测试。 +- 历史 ExpenseClaim 回填的 dry-run 零写、租户/时间边界、快照真实性、批次回滚、冲突拒绝、重复执行和数据库目标防误连测试。 - Business Event、AI Decision、Feedback、Outcome、Memory、Automation Policy、Savings Ledger service 单元测试。 - 服务端会话、租户隔离、角色和动作权限的正向/越权测试。 - 连接器事件幂等、重试、回执、失败恢复和重复支付防护测试。 @@ -395,6 +400,7 @@ customer_contribution_margin ### 集成 - 一句话申请 → 票据归集 → 自动报销 → 预审 → 审批 → 付款事件 → 入账 → 归档端到端。 +- 持久开发库只读流式克隆 → Alembic 升级 → 历史回填 dry-run/apply/重复 apply → 服务端登录 → 旧单时间线查询 → 持久库不变验证。 - AI 建议 → 用户修改 → 退回/通过 → 记忆候选 → 下次建议变化闭环。 - 风险命中 → 人工确认/误报 → few-shot → 新版本回放 → Canary/回滚闭环。 - 节省机会 → 负责人执行 → 实际结果 → 财务确认 → ROI 看板闭环。 @@ -443,6 +449,7 @@ docker exec -w /app -e SERVER_VENV_DIR=/tmp/x-financial-server-venv \ - 范围过大：费用闭环、AI 学习、价值分析和商业化不能同时全量实现，需要按 P0/P1/P2 阶段交付。 - 领域模型迁移：旧 `ReimbursementRequest`、`ExpenseClaim` 和 JSON 状态并存，必须旁路记录、小步迁移和双读校验。 +- 历史证据边界：旧单快照只表达回填时可确认的当前状态，迁移前逐节点办理过程仍以原单据和既有审计为准；后续分析不得把快照事件误当成历史审批事实。 - 认证和租户：当前客户端身份头不适合自动化和 SaaS，多租户、记忆和高风险动作开发前必须修复。 - 会话运维：不透明会话已经替代客户端身份头，但仍需补充定时清理、活跃会话查看/全部退出、密钥轮换策略、登录限流和企业 SSO；当前 `tenant_id` 仍是最小契约，不代表跨租户查询守卫已经完成。 - 费用事件读取边界：用户态精简 DTO 和首批 HTTP 权限测试已完成；剩余风险是同一 URL 若已有外部客户端依赖旧内部字段会产生契约变更，且未来新增敏感 payload 字段必须继续显式进入白名单，不能恢复任意字典透传。 @@ -504,3 +511,6 @@ docker exec -w /app -e SERVER_VENV_DIR=/tmp/x-financial-server-venv \ - 2026-07-14（迁移所有权加固）：新增统一 `schema_ownership.py`，七个运行时初始化入口只创建 legacy 表；标准启动在 Alembic upgrade 前执行只读漂移预检，revision 与 migration-owned 表集合不一致时 fail-fast，且不会自动 stamp 或修改数据库。 - 2026-07-14（真实迁移验证）：在主应用容器连接的一次性 tmpfs PostgreSQL 17 中完成空库升级、重复升级、关键约束/索引、真实外键级联、降级到 base、legacy 哨兵保留、无版本自有表漂移拒绝和再次升级，`test_alembic_migrations.py` 4 项通过，最终 revision 为 `20260713_0002`；临时容器已自动清理，持久化开发数据库复查仍未迁移。 - 2026-07-14（剩余边界）：当前两条 revision 只覆盖 Expense Case、Business Event 和 Auth Session，完整 legacy schema baseline 及停止其余运行时 DDL 仍未完成；本轮受影响服务回归 46 项通过，既有员工目录历史部门归一化用例仍单独失败，未混入本次迁移安全范围。 +- 2026-07-14（历史旧单接入）：新增独立 `ExpenseCaseLegacyBackfillService` 和 `backfill_legacy_expense_claim_cases.py`。命令只读取显式 `DATABASE_URL`，默认 dry-run；apply 强制目标核验、精确确认、迁移 head、advisory lock 和批次事务。每张旧单只创建一条 `historical_claim_imported` 系统快照，保留源时间和指纹、明确不重建历史，并以 `suppressed` 阻止实时投递。 +- 2026-07-14（克隆库联调）：将持久开发库以只读 `pg_dump` 流式恢复到一次性 tmpfs PostgreSQL 17，原 40 张表、4 张费用单、105 名员工、248 条预算和 62 个 Agent 资产完整保留。升级到 `20260713_0002` 后首次 dry-run 识别 4 张旧单，apply 创建 4 组 Case/Link/Event，重复 apply 创建 0 条；隔离后端完成登录、`/auth/me`、旧单时间线 200 和登出，内部回填字段未出现在 API。临时数据库和隔离进程已清理，持久库复查仍为原数据签名且没有 migration-owned 表。 +- 2026-07-14（安全复核与质量验证）：交叉审查后补齐超长租户幂等键稳定哈希、Link/Case 租户一致性、孤立 Case 冲突、URL 路由参数覆盖防护和部分批次失败进度摘要；前端将快照语义明确为“纳入时状态/节点”。容器内 65 项后端定向测试、11 项前端测试、Ruff 和 Vite 生产构建通过。全库代码体积门禁仍被本次未修改的 `RiskRuleGenerationService` 807 行存量问题阻断，未混入当前功能提交。 diff --git a/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md b/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md index a0fa87d..4d0b95a 100644 --- a/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md +++ b/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md @@ -77,6 +77,8 @@ 证据：`reimbursements.py`、`test_reimbursement_endpoints.py`；对抗用例修复前返回 200，修复后返回 400，且目标申请和费用事件保持不变。 - [x] [CONCEPT: 兼容策略] 建立迁移桥接：服务启动先执行 Alembic，旧 metadata bootstrap 排除 migration-owned 表。 证据：`server_start.sh`、`schema_ownership.py`、`migration_preflight.py`、`20260713_0001_expense_case_business_events.py`、`20260713_0002_auth_sessions.py`；容器 Shell/静态检查及一次性 PostgreSQL 完整 upgrade/downgrade/re-upgrade 通过。 +- [x] [CONCEPT: 兼容策略] 为迁移前已有 `ExpenseClaim` 提供显式、幂等且不伪造办理历史的费用事件快照回填。 + 证据：`expense_case_legacy_backfill.py`、`maintenance_database_target.py`、`backfill_legacy_expense_claim_cases.py`；默认 dry-run，apply 强制租户/截止时间、精确目标、迁移 head、advisory lock 和批次事务，事件使用真实回填时间、`history_reconstructed=false` 与 `delivery_status=suppressed`。一次性克隆库首次创建 4 组 Case/Link/Event，重复 apply 创建 0 条。 - [ ] [CONCEPT: 兼容策略] 正式切换前以 shadow 事件校验现有 `ExpenseClaim` 映射；切换后禁止关键事件可丢弃写入。 - [ ] [CONCEPT: 兼容策略] 制定旧 `ReimbursementRequest` 只读兼容、迁移和停止新增编排的计划。 - [ ] [CONCEPT: 兼容策略] 把新增审批、付款、归档和关系事件移出 `risk_flags_json`，保留旧数据读取兼容。 @@ -166,9 +168,11 @@ - [ ] [CONCEPT: 测试方案] 跑通风险反馈 → few-shot → golden case → Canary → 回滚闭环。 - [ ] [CONCEPT: 测试方案] 跑通节省机会 → 执行 → 实现 → 财务确认 → ROI 看板闭环。 - [x] [CONCEPT: 测试方案] 为已有 Expense Case 事件时间线补充视图模型、404 降级、详情页接入及相关响应式回归，并完成前端生产构建。 - 证据：容器内 `node --test` 定向执行 99 项通过；`npm --prefix web run build` 通过。浏览器可打开本地应用，但当前未迁移数据库无法签发有效认证凭证，真实详情页联调留待迁移后完成。 + 证据：容器内 `node --test` 定向执行 99 项通过；`npm --prefix web run build` 通过。一次性克隆迁移库上的隔离后端已完成真实登录、身份读取和旧单时间线 200 联调；持久开发库仍未迁移，因此日常本地页面仍保持兼容提示。 - [x] [CONCEPT: 测试方案] 为 AI 申请草稿事件补充事务失败回滚、同快照幂等、同 run 多版本留痕和 Steward 重放回归。 证据：本轮受影响后端定向回归 36 项、Expense Case 前端兼容测试 9 项和 Python `ruff --select F,I` 在容器内通过。 +- [x] [CONCEPT: 测试方案] 在现有开发数据的只读一次性克隆上验证迁移、历史回填和真实认证时间线链路。 + 证据：源库与克隆初始签名均为 40 张表、4 张费用单、105 名员工、248 条预算、62 个 Agent 资产；克隆升级后 dry-run 为 eligible=4，apply created=4，重复 apply created=0，四条事件均为 system/suppressed；登录、`/auth/me`、旧单时间线和登出均返回 200，持久库复查不变且仍无 migration-owned 表。 - [ ] [CONCEPT: 测试方案] 补充其余前端组件、键盘操作、移动真实接口和完整浏览器关键流程验证。 - [ ] [CONCEPT: 测试方案] 所有后端、集成和迁移测试在当前主应用容器内执行，单条命令最大超时 60s。 - [ ] [CONCEPT: 指标与验收] 记录测试、lint、typecheck、构建、端到端和未覆盖风险证据。 diff --git a/server/scripts/backfill_legacy_expense_claim_cases.py b/server/scripts/backfill_legacy_expense_claim_cases.py new file mode 100644 index 0000000..7c61b33 --- /dev/null +++ b/server/scripts/backfill_legacy_expense_claim_cases.py @@ -0,0 +1,532 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import os +import sys +import uuid +from dataclasses import asdict +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from sqlalchemy import create_engine, text +from sqlalchemy.engine import Connection +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.orm import Session +from sqlalchemy.pool import NullPool + +SERVER_DIR = Path(__file__).resolve().parents[1] +SRC_DIR = SERVER_DIR / "src" +if str(SRC_DIR) not in sys.path: + sys.path.insert(0, str(SRC_DIR)) + +from app.db.maintenance_database_target import ( # noqa: E402 + MaintenanceDatabaseTarget, + MaintenanceDatabaseTargetError, + validate_maintenance_database_target, +) +from app.db.migration_preflight import ( # noqa: E402 + MigrationPreflightError, + validate_migration_state, +) +from app.services.expense_case_legacy_backfill import ( # noqa: E402 + DEFAULT_BATCH_SIZE, + MAX_BATCH_SIZE, + ExpenseCaseLegacyBackfillService, + LegacyBackfillCursor, +) + +REQUIRED_ALEMBIC_REVISION = "20260713_0002" +EXIT_CONFIGURATION = 2 +EXIT_SAFETY = 3 +EXIT_LOCKED = 4 +EXIT_CONFLICT = 5 +EXIT_RUNTIME = 6 + + +class BackfillCommandError(RuntimeError): + def __init__( + self, + message: str, + *, + exit_code: int, + code: str, + details: dict[str, Any] | None = None, + ) -> None: + super().__init__(message) + self.exit_code = exit_code + self.code = code + self.details = details + + +def parse_created_before(value: str) -> datetime: + normalized = str(value or "").strip() + if normalized.endswith("Z"): + normalized = f"{normalized[:-1]}+00:00" + try: + parsed = datetime.fromisoformat(normalized) + except ValueError as exc: + raise argparse.ArgumentTypeError( + "--created-before 必须是带时区的 ISO 8601 时间，例如 2026-07-14T00:00:00Z" + ) from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise argparse.ArgumentTypeError("--created-before 必须显式包含时区") + return parsed.astimezone(UTC) + + +def positive_int(value: str) -> int: + try: + parsed = int(value) + except ValueError as exc: + raise argparse.ArgumentTypeError("必须是正整数") from exc + if parsed < 1: + raise argparse.ArgumentTypeError("必须是正整数") + return parsed + + +def non_empty_text(value: str) -> str: + normalized = str(value or "").strip() + if not normalized: + raise argparse.ArgumentTypeError("不能为空") + return normalized + + +def batch_size(value: str) -> int: + parsed = positive_int(value) + if parsed > MAX_BATCH_SIZE: + raise argparse.ArgumentTypeError(f"不能超过 {MAX_BATCH_SIZE}") + return parsed + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="把迁移前 ExpenseClaim 诚实地接入统一费用事件；默认只预览。", + ) + mode = parser.add_mutually_exclusive_group() + mode.add_argument("--dry-run", action="store_true", help="只读预览（默认）。") + mode.add_argument("--apply", action="store_true", help="按批写入历史快照事件。") + parser.add_argument( + "--tenant-id", + required=True, + type=non_empty_text, + help="旧单明确归属的租户 ID。", + ) + parser.add_argument( + "--created-before", + required=True, + type=parse_created_before, + help="仅处理该时刻之前创建的单据，必须包含时区。", + ) + parser.add_argument("--batch-size", type=batch_size, default=DEFAULT_BATCH_SIZE) + parser.add_argument( + "--max-claims", + type=positive_int, + help="最多扫描的历史单据数，用于 canary。", + ) + parser.add_argument("--sample-limit", type=positive_int, default=20) + parser.add_argument("--expected-host", required=True) + parser.add_argument("--expected-database", required=True) + parser.add_argument( + "--confirm-target", + help="apply 时必须精确等于解析后的 host:port/database。", + ) + parser.add_argument( + "--allow-non-disposable-target", + action="store_true", + help="允许在非 probe 数据库 apply；仍需精确确认目标。", + ) + return parser + + +def _cursor_payload(cursor: LegacyBackfillCursor | None) -> dict[str, str] | None: + if cursor is None: + return None + return { + "created_at": cursor.created_at.astimezone(UTC).isoformat().replace("+00:00", "Z"), + "claim_id": cursor.claim_id, + } + + +def _item_payload(item: Any) -> dict[str, str]: + payload = asdict(item) + payload["disposition"] = item.disposition.value + return payload + + +def _base_summary( + *, + mode: str, + target: MaintenanceDatabaseTarget, + tenant_id: str, + created_before: datetime, + revision: str, +) -> dict[str, Any]: + return { + "mode": mode, + "database": { + "target": target.exact_target, + "url": target.sanitized_url, + "disposable": target.is_disposable, + "revision": revision, + }, + "tenant_id": tenant_id, + "created_before": created_before.isoformat().replace("+00:00", "Z"), + "inspected": 0, + "eligible": 0, + "already_linked": 0, + "conflicts": 0, + "created": 0, + "batches": 0, + "limited": False, + "last_cursor": None, + "samples": [], + } + + +def _page_limit(*, configured: int, remaining: int | None) -> int: + return configured if remaining is None else min(configured, remaining) + + +def _execution_progress(summary: dict[str, Any], **extra: Any) -> dict[str, Any]: + progress = { + "run_id": summary.get("run_id"), + "partial_commit": bool(summary.get("batches")), + "committed_batches": int(summary.get("batches") or 0), + "committed_claims": int(summary.get("inspected") or 0), + "created": int(summary.get("created") or 0), + "already_linked": int(summary.get("already_linked") or 0), + "last_committed_cursor": summary.get("last_cursor"), + } + progress.update(extra) + return progress + + +def _preview( + session: Session, + *, + tenant_id: str, + created_before: datetime, + configured_batch_size: int, + max_claims: int | None, + sample_limit: int, + summary: dict[str, Any], +) -> None: + service = ExpenseCaseLegacyBackfillService( + session, + tenant_id=tenant_id, + cutoff=created_before, + ) + cursor: LegacyBackfillCursor | None = None + remaining = max_claims + + while remaining is None or remaining > 0: + page = service.preview( + batch_size=_page_limit(configured=configured_batch_size, remaining=remaining), + after=cursor, + ) + if not page.items: + break + summary["batches"] += 1 + summary["inspected"] += page.inspected + summary["eligible"] += page.eligible + summary["already_linked"] += page.linked + summary["conflicts"] += page.conflicts + available_samples = max(0, sample_limit - len(summary["samples"])) + summary["samples"].extend(_item_payload(item) for item in page.items[:available_samples]) + cursor = page.next_cursor + summary["last_cursor"] = _cursor_payload(cursor) + if remaining is not None: + remaining -= page.inspected + if not page.has_more: + break + + summary["limited"] = bool(remaining == 0 and page.has_more) if "page" in locals() else False + + +def _acquire_advisory_lock(connection: Connection, tenant_id: str) -> str: + lock_name = f"legacy-expense-case-backfill:{tenant_id}" + acquired = connection.scalar( + text("SELECT pg_try_advisory_lock(hashtextextended(:lock_name, 0))"), + {"lock_name": lock_name}, + ) + connection.commit() + if not acquired: + raise BackfillCommandError( + "同一租户已有历史费用事件回填任务正在运行。", + exit_code=EXIT_LOCKED, + code="advisory_lock_unavailable", + ) + return lock_name + + +def _release_advisory_lock(connection: Connection, lock_name: str) -> None: + if connection.in_transaction(): + connection.rollback() + connection.execute( + text("SELECT pg_advisory_unlock(hashtextextended(:lock_name, 0))"), + {"lock_name": lock_name}, + ) + connection.commit() + + +def _apply( + connection: Connection, + *, + tenant_id: str, + created_before: datetime, + configured_batch_size: int, + max_claims: int | None, + sample_limit: int, + summary: dict[str, Any], +) -> None: + lock_name = _acquire_advisory_lock(connection, tenant_id) + run_id = f"historical-import-{uuid.uuid4().hex}" + backfilled_at = datetime.now(UTC) + summary["run_id"] = run_id + summary["backfilled_at"] = backfilled_at.isoformat().replace("+00:00", "Z") + cursor: LegacyBackfillCursor | None = None + remaining = max_claims + + try: + with Session(bind=connection, autoflush=False, expire_on_commit=False) as session: + service = ExpenseCaseLegacyBackfillService( + session, + tenant_id=tenant_id, + cutoff=created_before, + ) + while remaining is None or remaining > 0: + session.execute(text("SET LOCAL lock_timeout = '5s'")) + result = service.apply_batch( + run_id=run_id, + batch_size=_page_limit( + configured=configured_batch_size, + remaining=remaining, + ), + after=cursor, + backfilled_at=backfilled_at, + ) + if not result.items: + session.rollback() + break + if result.conflicts: + session.rollback() + conflict_ids = [ + item.claim_id + for item in result.items + if item.disposition.value == "conflict" + ] + raise BackfillCommandError( + f"检测到 {result.conflicts} 个数据冲突，当前批次已回滚：" + f"{', '.join(conflict_ids)}", + exit_code=EXIT_CONFLICT, + code="legacy_claim_conflict", + details=_execution_progress( + summary, + current_batch_conflict_ids=conflict_ids, + ), + ) + + session.commit() + summary["batches"] += 1 + summary["inspected"] += result.inspected + summary["created"] += result.created + summary["already_linked"] += result.skipped_linked + available_samples = max(0, sample_limit - len(summary["samples"])) + summary["samples"].extend( + _item_payload(item) for item in result.items[:available_samples] + ) + cursor = result.next_cursor + summary["last_cursor"] = _cursor_payload(cursor) + if remaining is not None: + remaining -= result.inspected + if not result.has_more: + break + + summary["limited"] = bool(remaining == 0 and "result" in locals() and result.has_more) + finally: + _release_advisory_lock(connection, lock_name) + + +def _verify_connected_database( + connection: Connection, + *, + expected_database: str, +) -> tuple[str, str]: + database_name, database_user = connection.execute( + text("SELECT current_database(), current_user") + ).one() + if str(database_name) != expected_database: + raise BackfillCommandError( + "连接后的 current_database() 与 DATABASE_URL 不一致。", + exit_code=EXIT_SAFETY, + code="connected_database_mismatch", + ) + return str(database_name), str(database_user) + + +def run(args: argparse.Namespace) -> dict[str, Any]: + database_url = os.environ.get("DATABASE_URL", "") + if args.apply and str(args.confirm_target or "").strip() == "": + raise BackfillCommandError( + "--apply 必须提供 --confirm-target 精确确认数据库目标。", + exit_code=EXIT_SAFETY, + code="confirm_target_required", + ) + target = validate_maintenance_database_target( + database_url, + expected_host=args.expected_host, + expected_database=args.expected_database, + apply=args.apply, + allow_non_disposable=args.allow_non_disposable_target, + confirm_target=args.confirm_target, + ) + engine = create_engine( + database_url, + pool_pre_ping=True, + poolclass=NullPool, + ) + try: + with engine.connect() as connection: + connected_database, connected_user = _verify_connected_database( + connection, + expected_database=target.database, + ) + state = validate_migration_state(connection) + if state.revision != REQUIRED_ALEMBIC_REVISION: + raise BackfillCommandError( + "数据库必须先升级到费用事件迁移 head：" + f"actual={state.revision or 'unversioned/base'}; " + f"required={REQUIRED_ALEMBIC_REVISION}", + exit_code=EXIT_SAFETY, + code="migration_revision_mismatch", + ) + connection.rollback() + + summary = _base_summary( + mode="apply" if args.apply else "dry-run", + target=target, + tenant_id=args.tenant_id, + created_before=args.created_before, + revision=state.revision, + ) + summary["database"]["connected_database"] = connected_database + summary["database"]["connected_user"] = connected_user + + with Session(bind=connection, autoflush=False) as preview_session: + _preview( + preview_session, + tenant_id=args.tenant_id, + created_before=args.created_before, + configured_batch_size=args.batch_size, + max_claims=args.max_claims, + sample_limit=args.sample_limit, + summary=summary, + ) + preview_session.rollback() + + if not args.apply: + summary["would_create_cases"] = summary["eligible"] + summary["would_create_links"] = summary["eligible"] + summary["would_create_events"] = summary["eligible"] + return summary + if summary["conflicts"]: + raise BackfillCommandError( + f"预览发现 {summary['conflicts']} 个数据冲突；未执行任何写入。", + exit_code=EXIT_CONFLICT, + code="legacy_claim_conflict", + ) + + preview = { + "inspected": summary["inspected"], + "eligible": summary["eligible"], + "already_linked": summary["already_linked"], + "samples": summary["samples"], + } + summary.update( + inspected=0, + eligible=preview["eligible"], + already_linked=0, + conflicts=0, + created=0, + batches=0, + limited=False, + last_cursor=None, + samples=[], + preview=preview, + ) + try: + _apply( + connection, + tenant_id=args.tenant_id, + created_before=args.created_before, + configured_batch_size=args.batch_size, + max_claims=args.max_claims, + sample_limit=args.sample_limit, + summary=summary, + ) + except BackfillCommandError: + raise + except SQLAlchemyError as exc: + raise BackfillCommandError( + "数据库执行失败；已提交批次不会回滚，请依据进度摘要安全重跑。", + exit_code=EXIT_RUNTIME, + code="database_runtime_error", + details=_execution_progress(summary), + ) from exc + return summary + finally: + engine.dispose() + + +def _error_payload( + exc: Exception, + *, + code: str, + details: dict[str, Any] | None = None, +) -> dict[str, Any]: + payload: dict[str, Any] = {"status": "error", "code": code, "message": str(exc)} + if details is not None: + payload["details"] = details + return payload + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + payload = run(args) + except BackfillCommandError as exc: + print( + json.dumps( + _error_payload(exc, code=exc.code, details=exc.details), + ensure_ascii=False, + ), + file=sys.stderr, + ) + return exc.exit_code + except MaintenanceDatabaseTargetError as exc: + print( + json.dumps(_error_payload(exc, code=exc.code), ensure_ascii=False), + file=sys.stderr, + ) + return EXIT_SAFETY + except MigrationPreflightError as exc: + print( + json.dumps(_error_payload(exc, code="migration_preflight_failed"), ensure_ascii=False), + file=sys.stderr, + ) + return EXIT_SAFETY + except (SQLAlchemyError, OSError) as exc: + print( + json.dumps(_error_payload(exc, code="database_runtime_error"), ensure_ascii=False), + file=sys.stderr, + ) + return EXIT_RUNTIME + + print(json.dumps(payload, ensure_ascii=False, indent=2, default=str)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/server/src/app/db/maintenance_database_target.py b/server/src/app/db/maintenance_database_target.py new file mode 100644 index 0000000..2c9ebe5 --- /dev/null +++ b/server/src/app/db/maintenance_database_target.py @@ -0,0 +1,237 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Any + +from sqlalchemy.engine import URL, make_url +from sqlalchemy.exc import ArgumentError + +DEFAULT_POSTGRESQL_PORT = 5432 +DISPOSABLE_TARGET_MARKERS = ("migration-probe", "disposable-probe") +SENSITIVE_QUERY_KEY_PARTS = ( + "credential", + "passfile", + "password", + "secret", + "token", +) +FORBIDDEN_ROUTING_QUERY_KEYS = frozenset( + { + "database", + "dbname", + "host", + "hostaddr", + "options", + "port", + "service", + "servicefile", + "user", + "username", + } +) + + +class MaintenanceDatabaseTargetError(ValueError): + """维护命令数据库目标不满足安全约束。""" + + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + + +@dataclass(frozen=True, slots=True) +class MaintenanceDatabaseTarget: + host: str + port: int + database: str + username: str | None + sanitized_url: str + exact_target: str + is_disposable: bool + + +def _raise_target_error(code: str, message: str) -> None: + raise MaintenanceDatabaseTargetError(code, message) + + +def _normalize_host(value: str) -> str: + normalized = str(value or "").strip().lower() + if normalized.startswith("[") and normalized.endswith("]"): + normalized = normalized[1:-1] + return normalized.rstrip(".") + + +def _normalize_probe_component(value: str) -> str: + return re.sub(r"[^a-z0-9]+", "-", str(value or "").lower()).strip("-") + + +def _matches_disposable_marker(value: str) -> bool: + normalized = _normalize_probe_component(value) + return any( + normalized == marker + or normalized.startswith(f"{marker}-") + or normalized.startswith(f"x-financial-{marker}-") + for marker in DISPOSABLE_TARGET_MARKERS + ) + + +def is_disposable_maintenance_target(*, host: str, database: str) -> bool: + """只有主机名和数据库名都带显式 probe 标记时才视为一次性目标。""" + + return _matches_disposable_marker(host) and _matches_disposable_marker(database) + + +def _sanitize_query(url: URL) -> URL: + sanitized_query: dict[str, Any] = {} + for key, value in url.query.items(): + normalized_key = str(key).lower() + if any(part in normalized_key for part in SENSITIVE_QUERY_KEY_PARTS): + if isinstance(value, tuple): + sanitized_query[key] = tuple("***" for _ in value) + else: + sanitized_query[key] = "***" + continue + sanitized_query[key] = value + return url.set(query=sanitized_query) + + +def _format_target_host(host: str) -> str: + return f"[{host}]" if ":" in host else host + + +def parse_maintenance_database_target( + database_url: str, + *, + expected_host: str, + expected_database: str, +) -> MaintenanceDatabaseTarget: + """解析显式数据库 URL，并核对操作人声明的目标主机和数据库。""" + + raw_url = str(database_url or "").strip() + if not raw_url: + _raise_target_error( + "database_url_required", + "维护命令必须显式提供 DATABASE_URL，禁止回退到环境文件或默认配置。", + ) + + try: + parsed_url = make_url(raw_url) + except (ArgumentError, TypeError, ValueError) as exc: + raise MaintenanceDatabaseTargetError( + "invalid_database_url", + "DATABASE_URL 不是有效的 SQLAlchemy 数据库 URL。", + ) from exc + + if parsed_url.get_backend_name() != "postgresql": + _raise_target_error( + "postgresql_required", + "维护命令只允许连接 PostgreSQL 数据库。", + ) + + routing_query_keys = sorted( + str(key).lower() + for key in parsed_url.query + if str(key).lower() in FORBIDDEN_ROUTING_QUERY_KEYS + ) + if routing_query_keys: + _raise_target_error( + "database_routing_query_forbidden", + f"DATABASE_URL 查询参数不得覆盖连接目标或 schema：{', '.join(routing_query_keys)}", + ) + + host = _normalize_host(parsed_url.host or "") + if not host: + _raise_target_error( + "database_host_required", + "DATABASE_URL 必须包含显式 PostgreSQL 主机名。", + ) + + database = str(parsed_url.database or "").strip() + if not database: + _raise_target_error( + "database_name_required", + "DATABASE_URL 必须包含显式数据库名。", + ) + + normalized_expected_host = _normalize_host(expected_host) + if not normalized_expected_host: + _raise_target_error( + "expected_host_required", + "必须通过 expected_host 声明预期数据库主机。", + ) + if host != normalized_expected_host: + _raise_target_error( + "expected_host_mismatch", + f"DATABASE_URL 主机与预期不一致：actual={host}; expected={normalized_expected_host}", + ) + + normalized_expected_database = str(expected_database or "").strip() + if not normalized_expected_database: + _raise_target_error( + "expected_database_required", + "必须通过 expected_database 声明预期数据库名。", + ) + if database != normalized_expected_database: + _raise_target_error( + "expected_database_mismatch", + "DATABASE_URL 数据库名与预期不一致：" + f"actual={database}; expected={normalized_expected_database}", + ) + + port = int(parsed_url.port or DEFAULT_POSTGRESQL_PORT) + exact_target = f"{_format_target_host(host)}:{port}/{database}" + sanitized_url = _sanitize_query(parsed_url).render_as_string(hide_password=True) + + return MaintenanceDatabaseTarget( + host=host, + port=port, + database=database, + username=parsed_url.username, + sanitized_url=sanitized_url, + exact_target=exact_target, + is_disposable=is_disposable_maintenance_target(host=host, database=database), + ) + + +def validate_maintenance_database_target( + database_url: str, + *, + expected_host: str, + expected_database: str, + apply: bool = False, + allow_non_disposable: bool = False, + confirm_target: str | None = None, +) -> MaintenanceDatabaseTarget: + """校验维护目标；非一次性数据库 apply 必须显式放行并精确确认目标。""" + + target = parse_maintenance_database_target( + database_url, + expected_host=expected_host, + expected_database=expected_database, + ) + if not apply: + return target + + normalized_confirmation = str(confirm_target or "").strip() + if target.is_disposable: + if normalized_confirmation and normalized_confirmation != target.exact_target: + _raise_target_error( + "confirm_target_mismatch", + "confirm_target 与解析后的数据库目标不一致：" + f"actual={normalized_confirmation}; expected={target.exact_target}", + ) + return target + + if not allow_non_disposable: + _raise_target_error( + "non_disposable_apply_forbidden", + "非一次性数据库 apply 必须显式启用 allow_non_disposable。", + ) + if normalized_confirmation != target.exact_target: + _raise_target_error( + "confirm_target_mismatch", + "非一次性数据库 apply 必须通过 confirm_target 精确确认：" + f"expected={target.exact_target}", + ) + return target diff --git a/server/src/app/services/expense_case_legacy_backfill.py b/server/src/app/services/expense_case_legacy_backfill.py new file mode 100644 index 0000000..f5a39c3 --- /dev/null +++ b/server/src/app/services/expense_case_legacy_backfill.py @@ -0,0 +1,391 @@ +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from datetime import UTC, datetime +from decimal import Decimal +from enum import StrEnum + +from sqlalchemy import and_, or_, select +from sqlalchemy.orm import Session + +from app.models.expense_case import BusinessEvent, ExpenseCase, ExpenseCaseLink +from app.models.financial_record import ExpenseClaim +from app.services.expense_cases import ExpenseCaseService + +HISTORICAL_CLAIM_IMPORTED_EVENT = "historical_claim_imported" +HISTORICAL_IMPORT_VERSION = 1 +HISTORICAL_IMPORT_DELIVERY_STATUS = "suppressed" +DEFAULT_BATCH_SIZE = 100 +MAX_BATCH_SIZE = 1000 + + +class LegacyBackfillDisposition(StrEnum): + ELIGIBLE = "eligible" + LINKED = "linked" + CONFLICT = "conflict" + + +@dataclass(frozen=True, slots=True) +class LegacyBackfillCursor: + created_at: datetime + claim_id: str + + +@dataclass(frozen=True, slots=True) +class LegacyBackfillItem: + claim_id: str + claim_no: str + disposition: LegacyBackfillDisposition + source_fingerprint: str + reason: str = "" + + +@dataclass(frozen=True, slots=True) +class LegacyBackfillPreview: + tenant_id: str + cutoff: datetime + inspected: int + eligible: int + linked: int + conflicts: int + has_more: bool + next_cursor: LegacyBackfillCursor | None + items: tuple[LegacyBackfillItem, ...] + + +@dataclass(frozen=True, slots=True) +class LegacyBackfillBatchResult: + tenant_id: str + cutoff: datetime + run_id: str + inspected: int + created: int + skipped_linked: int + conflicts: int + has_more: bool + next_cursor: LegacyBackfillCursor | None + items: tuple[LegacyBackfillItem, ...] + + +class ExpenseCaseLegacyBackfillService: + """把旧费用单诚实地接入费用事件，不虚构迁移前的逐节点历史。""" + + def __init__(self, db: Session, *, tenant_id: str, cutoff: datetime) -> None: + self.db = db + self.tenant_id = self._require_text(tenant_id, field_name="tenant_id", max_length=64) + self.cutoff = self._require_aware_datetime(cutoff, field_name="cutoff") + self.expense_cases = ExpenseCaseService(db) + + def preview( + self, + *, + batch_size: int = DEFAULT_BATCH_SIZE, + after: LegacyBackfillCursor | None = None, + ) -> LegacyBackfillPreview: + """预览一批旧单；该方法只查询，不 flush、不 commit。""" + + with self.db.no_autoflush: + claims, has_more = self._load_claims( + batch_size=self._normalize_batch_size(batch_size), + after=after, + lock_rows=False, + ) + items = self._classify(claims) + return LegacyBackfillPreview( + tenant_id=self.tenant_id, + cutoff=self.cutoff, + inspected=len(items), + eligible=self._count(items, LegacyBackfillDisposition.ELIGIBLE), + linked=self._count(items, LegacyBackfillDisposition.LINKED), + conflicts=self._count(items, LegacyBackfillDisposition.CONFLICT), + has_more=has_more, + next_cursor=self._next_cursor(claims), + items=items, + ) + + def apply_batch( + self, + *, + run_id: str, + batch_size: int = DEFAULT_BATCH_SIZE, + after: LegacyBackfillCursor | None = None, + backfilled_at: datetime | None = None, + ) -> LegacyBackfillBatchResult: + """应用一批回填但不提交；调用方拥有完整的批次事务边界。""" + + normalized_run_id = self._require_text(run_id, field_name="run_id", max_length=64) + normalized_backfilled_at = self._require_aware_datetime( + backfilled_at or datetime.now(UTC), + field_name="backfilled_at", + ) + claims, has_more = self._load_claims( + batch_size=self._normalize_batch_size(batch_size), + after=after, + lock_rows=True, + ) + classified = self._classify(claims) + claims_by_id = {claim.id: claim for claim in claims} + result_items: list[LegacyBackfillItem] = [] + created = 0 + + for item in classified: + if item.disposition is not LegacyBackfillDisposition.ELIGIBLE: + result_items.append(item) + continue + + claim = claims_by_id[item.claim_id] + _expense_case, event = self.expense_cases.record_claim_event( + claim, + event_type=HISTORICAL_CLAIM_IMPORTED_EVENT, + actor_id="system", + tenant_id=self.tenant_id, + correlation_id=normalized_run_id, + idempotency_key=self.idempotency_key(claim.id), + previous_status="", + previous_approval_stage="", + extra_payload=self._event_payload( + claim, + run_id=normalized_run_id, + backfilled_at=normalized_backfilled_at, + source_fingerprint=item.source_fingerprint, + ), + delivery_status=HISTORICAL_IMPORT_DELIVERY_STATUS, + ) + # 事件发生时间表达真实回填动作；旧单业务时间只保留在 payload 中。 + event.occurred_at = normalized_backfilled_at + created += 1 + result_items.append(item) + + return LegacyBackfillBatchResult( + tenant_id=self.tenant_id, + cutoff=self.cutoff, + run_id=normalized_run_id, + inspected=len(result_items), + created=created, + skipped_linked=self._count(result_items, LegacyBackfillDisposition.LINKED), + conflicts=self._count(result_items, LegacyBackfillDisposition.CONFLICT), + has_more=has_more, + next_cursor=self._next_cursor(claims), + items=tuple(result_items), + ) + + def idempotency_key(self, claim_id: str) -> str: + normalized_claim_id = str(claim_id or "").strip() + if not normalized_claim_id: + raise ValueError("claim_id must not be empty") + key = ( + f"historical-import:v{HISTORICAL_IMPORT_VERSION}:{self.tenant_id}:{normalized_claim_id}" + ) + if len(key) <= 120: + return key + digest = hashlib.sha256(key.encode("utf-8")).hexdigest() + return f"historical-import:v{HISTORICAL_IMPORT_VERSION}:sha256:{digest}" + + @classmethod + def source_fingerprint(cls, claim: ExpenseClaim) -> str: + encoded = json.dumps( + cls._source_snapshot(claim), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return f"sha256:{hashlib.sha256(encoded).hexdigest()}" + + def _load_claims( + self, + *, + batch_size: int, + after: LegacyBackfillCursor | None, + lock_rows: bool, + ) -> tuple[list[ExpenseClaim], bool]: + stmt = select(ExpenseClaim).where(ExpenseClaim.created_at < self.cutoff) + if after is not None: + normalized_claim_id = self._require_text( + after.claim_id, + field_name="after.claim_id", + max_length=36, + ) + stmt = stmt.where( + or_( + ExpenseClaim.created_at > after.created_at, + and_( + ExpenseClaim.created_at == after.created_at, + ExpenseClaim.id > normalized_claim_id, + ), + ) + ) + stmt = stmt.order_by(ExpenseClaim.created_at.asc(), ExpenseClaim.id.asc()).limit( + batch_size + 1 + ) + if lock_rows: + stmt = stmt.with_for_update() + + claims = list(self.db.scalars(stmt).all()) + return claims[:batch_size], len(claims) > batch_size + + def _classify(self, claims: list[ExpenseClaim]) -> tuple[LegacyBackfillItem, ...]: + if not claims: + return () + + claim_ids = [claim.id for claim in claims] + links_by_claim_id = { + link.resource_id: (link, expense_case) + for link, expense_case in self.db.execute( + select(ExpenseCaseLink, ExpenseCase) + .outerjoin(ExpenseCase, ExpenseCase.id == ExpenseCaseLink.expense_case_id) + .where( + ExpenseCaseLink.resource_type == "expense_claim", + ExpenseCaseLink.resource_id.in_(claim_ids), + ) + ).all() + } + event_claim_ids = set( + self.db.scalars( + select(BusinessEvent.aggregate_id).where( + BusinessEvent.aggregate_type == "expense_claim", + BusinessEvent.aggregate_id.in_(claim_ids), + ) + ).all() + ) + expected_case_nos = {f"CASE-{str(claim.claim_no or claim.id).strip()}" for claim in claims} + existing_case_nos = set( + self.db.scalars( + select(ExpenseCase.case_no).where(ExpenseCase.case_no.in_(expected_case_nos)) + ).all() + ) + + items: list[LegacyBackfillItem] = [] + for claim in claims: + fingerprint = self.source_fingerprint(claim) + linked_record = links_by_claim_id.get(claim.id) + expected_case_no = f"CASE-{str(claim.claim_no or claim.id).strip()}" + if linked_record is not None: + link, expense_case = linked_record + if link.tenant_id == self.tenant_id: + if expense_case is None or expense_case.tenant_id != self.tenant_id: + disposition = LegacyBackfillDisposition.CONFLICT + reason = "expense claim link points to an invalid tenant expense case" + else: + disposition = LegacyBackfillDisposition.LINKED + reason = "expense claim already belongs to an expense case" + else: + disposition = LegacyBackfillDisposition.CONFLICT + reason = "expense claim is linked under another tenant" + elif claim.id in event_claim_ids: + disposition = LegacyBackfillDisposition.CONFLICT + reason = "business event exists without an expense case link" + elif expected_case_no in existing_case_nos: + disposition = LegacyBackfillDisposition.CONFLICT + reason = "expense case exists without an expense claim link" + else: + disposition = LegacyBackfillDisposition.ELIGIBLE + reason = "" + items.append( + LegacyBackfillItem( + claim_id=claim.id, + claim_no=str(claim.claim_no or ""), + disposition=disposition, + source_fingerprint=fingerprint, + reason=reason, + ) + ) + return tuple(items) + + @classmethod + def _source_snapshot(cls, claim: ExpenseClaim) -> dict[str, object]: + return { + "id": str(claim.id or ""), + "claim_no": str(claim.claim_no or ""), + "employee_id": str(claim.employee_id or ""), + "expense_type": str(claim.expense_type or ""), + "amount": cls._money_text(claim.amount), + "currency": str(claim.currency or "CNY"), + "status": str(claim.status or ""), + "approval_stage": str(claim.approval_stage or ""), + "source_times": cls._source_times(claim), + } + + @classmethod + def _event_payload( + cls, + claim: ExpenseClaim, + *, + run_id: str, + backfilled_at: datetime, + source_fingerprint: str, + ) -> dict[str, object]: + return { + "schema_version": HISTORICAL_IMPORT_VERSION, + "backfill_version": HISTORICAL_IMPORT_VERSION, + "source": "legacy_expense_claim", + "history_reconstructed": False, + "backfill_run_id": run_id, + "backfilled_at": cls._isoformat(backfilled_at), + "performed_by": "expense_case_legacy_backfill_cli", + "source_times": cls._source_times(claim), + "source_fingerprint": source_fingerprint, + "reason": "该单据已纳入统一费用事件；迁移前的逐节点办理明细未重建。", + } + + @classmethod + def _source_times(cls, claim: ExpenseClaim) -> dict[str, str | None]: + return { + "occurred_at": cls._optional_isoformat(claim.occurred_at), + "submitted_at": cls._optional_isoformat(claim.submitted_at), + "created_at": cls._optional_isoformat(claim.created_at), + "updated_at": cls._optional_isoformat(claim.updated_at), + } + + @staticmethod + def _money_text(value: Decimal | None) -> str: + return f"{Decimal(value or Decimal('0.00')).quantize(Decimal('0.01')):.2f}" + + @classmethod + def _optional_isoformat(cls, value: datetime | None) -> str | None: + return cls._isoformat(value) if value is not None else None + + @staticmethod + def _isoformat(value: datetime) -> str: + normalized = value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) + return normalized.isoformat().replace("+00:00", "Z") + + @staticmethod + def _require_aware_datetime(value: datetime, *, field_name: str) -> datetime: + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError(f"{field_name} must include a timezone") + return value.astimezone(UTC) + + @staticmethod + def _require_text(value: str, *, field_name: str, max_length: int) -> str: + normalized = str(value or "").strip() + if not normalized: + raise ValueError(f"{field_name} must not be empty") + if len(normalized) > max_length: + raise ValueError(f"{field_name} must be at most {max_length} characters") + return normalized + + @staticmethod + def _normalize_batch_size(value: int) -> int: + normalized = int(value) + if normalized < 1 or normalized > MAX_BATCH_SIZE: + raise ValueError(f"batch_size must be between 1 and {MAX_BATCH_SIZE}") + return normalized + + @staticmethod + def _count( + items: tuple[LegacyBackfillItem, ...] | list[LegacyBackfillItem], + disposition: LegacyBackfillDisposition, + ) -> int: + return sum(1 for item in items if item.disposition is disposition) + + @staticmethod + def _next_cursor(claims: list[ExpenseClaim]) -> LegacyBackfillCursor | None: + if not claims: + return None + last_claim = claims[-1] + return LegacyBackfillCursor( + created_at=last_claim.created_at, + claim_id=last_claim.id, + ) diff --git a/server/src/app/services/expense_cases.py b/server/src/app/services/expense_cases.py index 01606b6..a5b4a25 100644 --- a/server/src/app/services/expense_cases.py +++ b/server/src/app/services/expense_cases.py @@ -151,6 +151,7 @@ class ExpenseCaseService: expense_case: ExpenseCase | None = None, relation_type: str | None = None, update_case_state: bool = True, + delivery_status: str = "pending", ) -> tuple[ExpenseCase, BusinessEvent]: normalized_tenant = self.normalize_tenant_id(tenant_id) if expense_case is None: @@ -215,7 +216,7 @@ class ExpenseCaseService: actor_id=str(actor_id or "system").strip() or "system", actor_type="system" if str(actor_id or "").strip() == "system" else "user", payload_json=payload, - delivery_status="pending", + delivery_status=str(delivery_status or "pending").strip() or "pending", occurred_at=datetime.now(UTC), ) self.db.add(event) diff --git a/server/tests/test_backfill_legacy_expense_claim_cases_cli.py b/server/tests/test_backfill_legacy_expense_claim_cases_cli.py new file mode 100644 index 0000000..1ef66fa --- /dev/null +++ b/server/tests/test_backfill_legacy_expense_claim_cases_cli.py @@ -0,0 +1,183 @@ +from __future__ import annotations + +import argparse +import importlib.util +from datetime import UTC, datetime +from pathlib import Path + +import pytest + + +def load_script_module(): + script_path = ( + Path(__file__).resolve().parents[1] / "scripts" / "backfill_legacy_expense_claim_cases.py" + ) + spec = importlib.util.spec_from_file_location( + "backfill_legacy_expense_claim_cases_cli", + script_path, + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +MODULE = load_script_module() + + +def test_created_before_requires_timezone_and_normalizes_to_utc() -> None: + assert MODULE.parse_created_before("2026-07-14T08:00:00+08:00") == datetime( + 2026, + 7, + 14, + tzinfo=UTC, + ) + with pytest.raises(argparse.ArgumentTypeError, match="时区"): + MODULE.parse_created_before("2026-07-14T00:00:00") + + +@pytest.mark.parametrize("value", ["0", "-1", "not-a-number"]) +def test_positive_int_rejects_invalid_values(value: str) -> None: + with pytest.raises(argparse.ArgumentTypeError): + MODULE.positive_int(value) + + +def test_parser_defaults_to_dry_run_and_requires_explicit_scope() -> None: + args = MODULE.build_parser().parse_args( + [ + "--tenant-id", + "default", + "--created-before", + "2026-07-14T00:00:00Z", + "--expected-host", + "migration-probe", + "--expected-database", + "migration_probe", + ] + ) + + assert args.apply is False + assert args.dry_run is False + assert args.batch_size == 100 + assert args.tenant_id == "default" + + +def test_parser_rejects_apply_and_dry_run_together() -> None: + with pytest.raises(SystemExit): + MODULE.build_parser().parse_args( + [ + "--apply", + "--dry-run", + "--tenant-id", + "default", + "--created-before", + "2026-07-14T00:00:00Z", + "--expected-host", + "migration-probe", + "--expected-database", + "migration_probe", + ] + ) + + +def test_parser_rejects_blank_tenant() -> None: + with pytest.raises(SystemExit): + MODULE.build_parser().parse_args( + [ + "--tenant-id", + " ", + "--created-before", + "2026-07-14T00:00:00Z", + "--expected-host", + "migration-probe", + "--expected-database", + "migration_probe", + ] + ) + + +def test_apply_requires_exact_target_confirmation_before_connecting(monkeypatch) -> None: + monkeypatch.setenv( + "DATABASE_URL", + "postgresql://finance:secret@migration-probe/migration_probe", + ) + args = MODULE.build_parser().parse_args( + [ + "--apply", + "--tenant-id", + "default", + "--created-before", + "2026-07-14T00:00:00Z", + "--expected-host", + "migration-probe", + "--expected-database", + "migration_probe", + ] + ) + + with pytest.raises(MODULE.BackfillCommandError) as exc_info: + MODULE.run(args) + + assert exc_info.value.code == "confirm_target_required" + assert exc_info.value.exit_code == MODULE.EXIT_SAFETY + + +def test_missing_database_url_is_rejected_without_configuration_fallback(monkeypatch) -> None: + monkeypatch.delenv("DATABASE_URL", raising=False) + args = MODULE.build_parser().parse_args( + [ + "--tenant-id", + "default", + "--created-before", + "2026-07-14T00:00:00Z", + "--expected-host", + "migration-probe", + "--expected-database", + "migration_probe", + ] + ) + + with pytest.raises(MODULE.MaintenanceDatabaseTargetError) as exc_info: + MODULE.run(args) + + assert exc_info.value.code == "database_url_required" + + +def test_partial_failure_payload_discloses_committed_progress() -> None: + summary = { + "run_id": "historical-import-run-1", + "batches": 2, + "inspected": 4, + "created": 3, + "already_linked": 1, + "last_cursor": { + "created_at": "2026-07-10T00:00:00Z", + "claim_id": "claim-4", + }, + } + details = MODULE._execution_progress( + summary, + current_batch_conflict_ids=["claim-5"], + ) + error = MODULE.BackfillCommandError( + "current batch rolled back", + exit_code=MODULE.EXIT_CONFLICT, + code="legacy_claim_conflict", + details=details, + ) + + payload = MODULE._error_payload(error, code=error.code, details=error.details) + + assert payload["details"] == { + "run_id": "historical-import-run-1", + "partial_commit": True, + "committed_batches": 2, + "committed_claims": 4, + "created": 3, + "already_linked": 1, + "last_committed_cursor": { + "created_at": "2026-07-10T00:00:00Z", + "claim_id": "claim-4", + }, + "current_batch_conflict_ids": ["claim-5"], + } diff --git a/server/tests/test_expense_case_legacy_backfill.py b/server/tests/test_expense_case_legacy_backfill.py new file mode 100644 index 0000000..0e2623f --- /dev/null +++ b/server/tests/test_expense_case_legacy_backfill.py @@ -0,0 +1,385 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from decimal import Decimal + +import pytest +from sqlalchemy import create_engine, func, select +from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy.pool import StaticPool + +from app.db.base import Base +from app.models.expense_case import BusinessEvent, ExpenseCase, ExpenseCaseLink +from app.models.financial_record import ExpenseClaim +from app.services.expense_case_legacy_backfill import ( + HISTORICAL_CLAIM_IMPORTED_EVENT, + HISTORICAL_IMPORT_DELIVERY_STATUS, + ExpenseCaseLegacyBackfillService, + LegacyBackfillDisposition, +) + +CUTOFF = datetime(2026, 7, 14, tzinfo=UTC) + + +def build_session() -> Session: + engine = create_engine( + "sqlite+pysqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + Base.metadata.create_all(bind=engine) + return sessionmaker(bind=engine, autoflush=False, autocommit=False)() + + +def build_claim(claim_no: str, *, created_at: datetime | None = None) -> ExpenseClaim: + return ExpenseClaim( + claim_no=claim_no, + employee_name="张三", + department_name="市场部", + project_code="PRJ-BACKFILL", + expense_type="travel", + reason="历史客户拜访", + location="上海", + amount=Decimal("128.50"), + currency="CNY", + invoice_count=1, + occurred_at=datetime(2026, 7, 10, 9, 0, tzinfo=UTC), + submitted_at=datetime(2026, 7, 11, 10, 0, tzinfo=UTC), + status="submitted", + approval_stage="直属领导审批", + risk_flags_json=[], + created_at=created_at or datetime(2026, 7, 10, 8, 0, tzinfo=UTC), + updated_at=datetime(2026, 7, 11, 10, 0, tzinfo=UTC), + ) + + +def seed_link(db: Session, claim: ExpenseClaim, *, tenant_id: str = "default") -> None: + expense_case = ExpenseCase( + tenant_id=tenant_id, + case_no=f"CASE-{claim.claim_no}", + scene_code="travel", + title=claim.reason, + current_stage="reviewing", + status="active", + ) + db.add(expense_case) + db.flush() + db.add( + ExpenseCaseLink( + tenant_id=tenant_id, + expense_case_id=expense_case.id, + resource_type="expense_claim", + resource_id=claim.id, + relation_type="claim", + ) + ) + db.flush() + + +def seed_event_without_link(db: Session, claim: ExpenseClaim) -> None: + expense_case = ExpenseCase( + tenant_id="default", + case_no=f"CASE-ORPHAN-{claim.claim_no}", + scene_code="travel", + title=claim.reason, + current_stage="reviewing", + status="active", + ) + db.add(expense_case) + db.flush() + db.add( + BusinessEvent( + tenant_id="default", + expense_case_id=expense_case.id, + aggregate_type="expense_claim", + aggregate_id=claim.id, + event_type="claim_submitted", + event_version=1, + idempotency_key=f"orphan:{claim.id}", + correlation_id="orphan-event", + actor_id="system", + actor_type="system", + payload_json={}, + delivery_status="pending", + occurred_at=datetime(2026, 7, 11, tzinfo=UTC), + ) + ) + db.flush() + + +def table_counts(db: Session) -> tuple[int, int, int]: + return ( + db.scalar(select(func.count()).select_from(ExpenseCase)) or 0, + db.scalar(select(func.count()).select_from(ExpenseCaseLink)) or 0, + db.scalar(select(func.count()).select_from(BusinessEvent)) or 0, + ) + + +def test_requires_explicit_tenant_and_timezone_aware_cutoff() -> None: + with build_session() as db: + with pytest.raises(ValueError, match="tenant_id"): + ExpenseCaseLegacyBackfillService(db, tenant_id="", cutoff=CUTOFF) + with pytest.raises(ValueError, match="timezone"): + ExpenseCaseLegacyBackfillService( + db, + tenant_id="default", + cutoff=datetime(2026, 7, 14), + ) + + +def test_fingerprint_is_stable_and_idempotency_key_is_tenant_scoped() -> None: + with build_session() as db: + claim = build_claim("EXP-HISTORY-FINGERPRINT") + db.add(claim) + db.flush() + default_service = ExpenseCaseLegacyBackfillService( + db, + tenant_id="default", + cutoff=CUTOFF, + ) + tenant_service = ExpenseCaseLegacyBackfillService( + db, + tenant_id="tenant-b", + cutoff=CUTOFF, + ) + + first_fingerprint = default_service.source_fingerprint(claim) + assert default_service.source_fingerprint(claim) == first_fingerprint + assert default_service.idempotency_key(claim.id) == ( + f"historical-import:v1:default:{claim.id}" + ) + assert tenant_service.idempotency_key(claim.id) == ( + f"historical-import:v1:tenant-b:{claim.id}" + ) + + claim.status = "approved" + assert default_service.source_fingerprint(claim) != first_fingerprint + + +def test_long_tenant_idempotency_key_is_stably_hashed_to_database_limit() -> None: + with build_session() as db: + service = ExpenseCaseLegacyBackfillService( + db, + tenant_id="t" * 64, + cutoff=CUTOFF, + ) + + first = service.idempotency_key("c" * 36) + second = service.idempotency_key("c" * 36) + + assert first == second + assert first.startswith("historical-import:v1:sha256:") + assert len(first) <= 120 + + +def test_preview_is_read_only_and_classifies_linked_and_orphan_event_conflicts() -> None: + with build_session() as db: + eligible = build_claim("EXP-HISTORY-ELIGIBLE") + linked = build_claim("EXP-HISTORY-LINKED", created_at=CUTOFF - timedelta(days=3)) + conflict = build_claim("EXP-HISTORY-CONFLICT", created_at=CUTOFF - timedelta(days=2)) + future = build_claim("EXP-HISTORY-FUTURE", created_at=CUTOFF + timedelta(seconds=1)) + db.add_all([eligible, linked, conflict, future]) + db.flush() + seed_link(db, linked) + seed_event_without_link(db, conflict) + db.commit() + + counts_before = table_counts(db) + preview = ExpenseCaseLegacyBackfillService( + db, + tenant_id="default", + cutoff=CUTOFF, + ).preview(batch_size=10) + + assert preview.inspected == 3 + assert preview.eligible == 1 + assert preview.linked == 1 + assert preview.conflicts == 1 + assert not preview.has_more + assert table_counts(db) == counts_before + dispositions = {item.claim_no: item.disposition for item in preview.items} + assert dispositions == { + "EXP-HISTORY-ELIGIBLE": LegacyBackfillDisposition.ELIGIBLE, + "EXP-HISTORY-LINKED": LegacyBackfillDisposition.LINKED, + "EXP-HISTORY-CONFLICT": LegacyBackfillDisposition.CONFLICT, + } + + +def test_apply_batch_writes_honest_snapshot_without_committing_and_is_idempotent() -> None: + with build_session() as db: + claim = build_claim("EXP-HISTORY-APPLY") + db.add(claim) + db.commit() + backfilled_at = datetime(2026, 7, 14, 12, 30, tzinfo=UTC) + service = ExpenseCaseLegacyBackfillService( + db, + tenant_id="default", + cutoff=CUTOFF, + ) + + result = service.apply_batch( + run_id="legacy-backfill-run-1", + batch_size=10, + backfilled_at=backfilled_at, + ) + + assert result.created == 1 + assert result.skipped_linked == 0 + assert result.conflicts == 0 + event = db.scalar(select(BusinessEvent)) + assert event is not None + assert event.event_type == HISTORICAL_CLAIM_IMPORTED_EVENT + assert event.idempotency_key == (f"historical-import:v1:default:{claim.id}") + assert event.idempotency_key == service.idempotency_key(claim.id) + assert event.actor_id == "system" + assert event.actor_type == "system" + assert event.delivery_status == HISTORICAL_IMPORT_DELIVERY_STATUS + assert event.occurred_at == backfilled_at + assert event.payload_json["history_reconstructed"] is False + assert event.payload_json["backfill_version"] == 1 + assert event.payload_json["backfilled_at"] == "2026-07-14T12:30:00Z" + assert event.payload_json["performed_by"] == "expense_case_legacy_backfill_cli" + assert event.payload_json["source_times"] == { + "occurred_at": "2026-07-10T09:00:00Z", + "submitted_at": "2026-07-11T10:00:00Z", + "created_at": "2026-07-10T08:00:00Z", + "updated_at": "2026-07-11T10:00:00Z", + } + assert event.payload_json["source_fingerprint"].startswith("sha256:") + assert "未重建" in event.payload_json["reason"] + + db.rollback() + assert table_counts(db) == (0, 0, 0) + + first_apply = service.apply_batch( + run_id="legacy-backfill-run-2", + batch_size=10, + backfilled_at=backfilled_at, + ) + db.commit() + repeated_apply = service.apply_batch( + run_id="legacy-backfill-run-3", + batch_size=10, + backfilled_at=backfilled_at + timedelta(minutes=1), + ) + + assert first_apply.created == 1 + assert repeated_apply.created == 0 + assert repeated_apply.skipped_linked == 1 + assert table_counts(db) == (1, 1, 1) + + +def test_apply_batch_reports_event_without_link_as_conflict() -> None: + with build_session() as db: + claim = build_claim("EXP-HISTORY-ORPHAN") + db.add(claim) + db.flush() + seed_event_without_link(db, claim) + db.commit() + + result = ExpenseCaseLegacyBackfillService( + db, + tenant_id="default", + cutoff=CUTOFF, + ).apply_batch( + run_id="legacy-backfill-conflict", + batch_size=10, + ) + + assert result.created == 0 + assert result.conflicts == 1 + assert result.items[0].disposition is LegacyBackfillDisposition.CONFLICT + assert "without an expense case link" in result.items[0].reason + assert db.scalar(select(func.count()).select_from(ExpenseCaseLink)) == 0 + + +def test_preview_rejects_damaged_link_tenant_and_unlinked_case() -> None: + with build_session() as db: + damaged_link_claim = build_claim("EXP-HISTORY-DAMAGED-LINK") + unlinked_case_claim = build_claim( + "EXP-HISTORY-UNLINKED-CASE", + created_at=CUTOFF - timedelta(days=2), + ) + db.add_all([damaged_link_claim, unlinked_case_claim]) + db.flush() + + foreign_case = ExpenseCase( + tenant_id="other-tenant", + case_no=f"CASE-{damaged_link_claim.claim_no}", + scene_code="travel", + title=damaged_link_claim.reason, + current_stage="reviewing", + status="active", + ) + orphan_case = ExpenseCase( + tenant_id="default", + case_no=f"CASE-{unlinked_case_claim.claim_no}", + scene_code="travel", + title=unlinked_case_claim.reason, + current_stage="reviewing", + status="active", + ) + db.add_all([foreign_case, orphan_case]) + db.flush() + db.add( + ExpenseCaseLink( + tenant_id="default", + expense_case_id=foreign_case.id, + resource_type="expense_claim", + resource_id=damaged_link_claim.id, + relation_type="claim", + ) + ) + db.commit() + + preview = ExpenseCaseLegacyBackfillService( + db, + tenant_id="default", + cutoff=CUTOFF, + ).preview(batch_size=10) + + assert preview.conflicts == 2 + reasons = {item.claim_no: item.reason for item in preview.items} + assert "invalid tenant" in reasons["EXP-HISTORY-DAMAGED-LINK"] + assert "without an expense claim link" in reasons["EXP-HISTORY-UNLINKED-CASE"] + + +def test_preview_cursor_provides_stable_keyset_batches() -> None: + with build_session() as db: + first = build_claim("EXP-HISTORY-001", created_at=CUTOFF - timedelta(days=3)) + second = build_claim("EXP-HISTORY-002", created_at=CUTOFF - timedelta(days=2)) + db.add_all([first, second]) + db.commit() + service = ExpenseCaseLegacyBackfillService( + db, + tenant_id="default", + cutoff=CUTOFF, + ) + + page_one = service.preview(batch_size=1) + page_two = service.preview(batch_size=1, after=page_one.next_cursor) + + assert page_one.has_more + assert page_one.items[0].claim_no == "EXP-HISTORY-001" + assert not page_two.has_more + assert page_two.items[0].claim_no == "EXP-HISTORY-002" + + +def test_standard_event_delivery_status_remains_pending_by_default() -> None: + with build_session() as db: + claim = build_claim("EXP-NORMAL-EVENT") + db.add(claim) + db.commit() + + _expense_case, event = ExpenseCaseLegacyBackfillService( + db, + tenant_id="default", + cutoff=CUTOFF, + ).expense_cases.record_claim_event( + claim, + event_type="claim_submitted", + actor_id="owner@example.com", + tenant_id="default", + ) + + assert event.delivery_status == "pending" diff --git a/server/tests/test_maintenance_database_target.py b/server/tests/test_maintenance_database_target.py new file mode 100644 index 0000000..60afe49 --- /dev/null +++ b/server/tests/test_maintenance_database_target.py @@ -0,0 +1,225 @@ +from __future__ import annotations + +import pytest + +from app.db.maintenance_database_target import ( + MaintenanceDatabaseTargetError, + is_disposable_maintenance_target, + parse_maintenance_database_target, + validate_maintenance_database_target, +) + + +def assert_target_error(code: str, callback) -> None: + with pytest.raises(MaintenanceDatabaseTargetError) as exc_info: + callback() + assert exc_info.value.code == code + + +def test_parse_target_supports_hostname_and_hides_password() -> None: + target = parse_maintenance_database_target( + "postgresql+psycopg://finance:super-secret@db.internal:5544/x_financial" + "?application_name=backfill", + expected_host="DB.INTERNAL.", + expected_database="x_financial", + ) + + assert target.host == "db.internal" + assert target.port == 5544 + assert target.database == "x_financial" + assert target.username == "finance" + assert target.exact_target == "db.internal:5544/x_financial" + assert "super-secret" not in target.sanitized_url + assert "***" in target.sanitized_url + assert "application_name=backfill" in target.sanitized_url + assert target.is_disposable is False + + +def test_parse_target_supports_ipv4_and_defaults_postgresql_port() -> None: + target = parse_maintenance_database_target( + "postgresql://finance:password@127.0.0.1/x_financial", + expected_host="127.0.0.1", + expected_database="x_financial", + ) + + assert target.port == 5432 + assert target.exact_target == "127.0.0.1:5432/x_financial" + assert "password" not in target.sanitized_url + + +def test_parse_target_redacts_sensitive_query_values() -> None: + target = parse_maintenance_database_target( + "postgresql://finance:authority-secret@db.internal/x_financial" + "?sslpassword=query-secret&application_name=safe-name", + expected_host="db.internal", + expected_database="x_financial", + ) + + assert "authority-secret" not in target.sanitized_url + assert "query-secret" not in target.sanitized_url + assert "application_name=safe-name" in target.sanitized_url + + +@pytest.mark.parametrize( + "routing_query", + [ + "host=x-financial-local-postgres", + "port=6432", + "dbname=production", + "user=other-user", + "service=production-service", + "options=-csearch_path%3Dother_schema", + ], +) +def test_parse_target_rejects_query_parameters_that_can_override_route( + routing_query: str, +) -> None: + assert_target_error( + "database_routing_query_forbidden", + lambda: parse_maintenance_database_target( + f"postgresql://finance:password@migration-probe/migration_probe?{routing_query}", + expected_host="migration-probe", + expected_database="migration_probe", + ), + ) + + +@pytest.mark.parametrize( + ("database_url", "code"), + [ + ("", "database_url_required"), + ("not a database url", "invalid_database_url"), + ("sqlite:///tmp/test.db", "postgresql_required"), + ("postgresql:///x_financial", "database_host_required"), + ("postgresql://finance@db.internal", "database_name_required"), + ], +) +def test_parse_target_rejects_invalid_or_incomplete_urls(database_url: str, code: str) -> None: + assert_target_error( + code, + lambda: parse_maintenance_database_target( + database_url, + expected_host="db.internal", + expected_database="x_financial", + ), + ) + + +@pytest.mark.parametrize( + ("expected_host", "expected_database", "code"), + [ + ("", "x_financial", "expected_host_required"), + ("other-db", "x_financial", "expected_host_mismatch"), + ("db.internal", "", "expected_database_required"), + ("db.internal", "other_database", "expected_database_mismatch"), + ], +) +def test_parse_target_rejects_expected_target_mismatches( + expected_host: str, + expected_database: str, + code: str, +) -> None: + assert_target_error( + code, + lambda: parse_maintenance_database_target( + "postgresql://finance:password@db.internal/x_financial", + expected_host=expected_host, + expected_database=expected_database, + ), + ) + + +@pytest.mark.parametrize( + ("host", "database", "expected"), + [ + ("migration-probe", "migration_probe", True), + ("x-financial-migration-probe-123", "migration_probe_clone_123", True), + ("disposable-probe-7", "disposable_probe_7", True), + ("x-financial-local-postgres", "migration_probe", False), + ("migration-probe-7", "x_financial", False), + ], +) +def test_disposable_target_requires_markers_on_host_and_database( + host: str, + database: str, + expected: bool, +) -> None: + assert is_disposable_maintenance_target(host=host, database=database) is expected + + +def test_dry_run_allows_non_disposable_target_without_apply_override() -> None: + target = validate_maintenance_database_target( + "postgresql://finance:password@db.internal/x_financial", + expected_host="db.internal", + expected_database="x_financial", + apply=False, + ) + + assert target.is_disposable is False + + +def test_apply_allows_disposable_target_without_non_disposable_override() -> None: + target = validate_maintenance_database_target( + "postgresql://finance:password@x-financial-migration-probe-1/migration_probe_1", + expected_host="x-financial-migration-probe-1", + expected_database="migration_probe_1", + apply=True, + ) + + assert target.is_disposable is True + + +def test_apply_rejects_wrong_optional_confirmation_for_disposable_target() -> None: + assert_target_error( + "confirm_target_mismatch", + lambda: validate_maintenance_database_target( + "postgresql://finance:password@migration-probe/migration_probe", + expected_host="migration-probe", + expected_database="migration_probe", + apply=True, + confirm_target="other:5432/migration_probe", + ), + ) + + +def test_apply_rejects_non_disposable_target_without_explicit_override() -> None: + assert_target_error( + "non_disposable_apply_forbidden", + lambda: validate_maintenance_database_target( + "postgresql://finance:password@db.internal/x_financial", + expected_host="db.internal", + expected_database="x_financial", + apply=True, + confirm_target="db.internal:5432/x_financial", + ), + ) + + +@pytest.mark.parametrize("confirm_target", [None, "", "db.internal:5432/other_database"]) +def test_apply_rejects_non_disposable_target_without_exact_confirmation( + confirm_target: str | None, +) -> None: + assert_target_error( + "confirm_target_mismatch", + lambda: validate_maintenance_database_target( + "postgresql://finance:password@db.internal/x_financial", + expected_host="db.internal", + expected_database="x_financial", + apply=True, + allow_non_disposable=True, + confirm_target=confirm_target, + ), + ) + + +def test_apply_allows_non_disposable_target_with_override_and_exact_confirmation() -> None: + target = validate_maintenance_database_target( + "postgresql://finance:password@db.internal/x_financial", + expected_host="db.internal", + expected_database="x_financial", + apply=True, + allow_non_disposable=True, + confirm_target="db.internal:5432/x_financial", + ) + + assert target.exact_target == "db.internal:5432/x_financial" diff --git a/web/src/utils/expenseCaseTimeline.js b/web/src/utils/expenseCaseTimeline.js index 1fa31e5..8c6ee8d 100644 --- a/web/src/utils/expenseCaseTimeline.js +++ b/web/src/utils/expenseCaseTimeline.js @@ -26,6 +26,7 @@ const BUSINESS_STATUS_LABELS = { const EVENT_PRESENTATION = { claim_draft_created: { label: '报销草稿已创建', icon: 'mdi mdi-file-plus-outline', tone: 'info' }, claim_draft_updated: { label: '报销草稿已更新', icon: 'mdi mdi-file-edit-outline', tone: 'info' }, + historical_claim_imported: { label: '历史单据已纳入统一费用事件', icon: 'mdi mdi-history', tone: 'info' }, application_submitted: { label: '费用申请已提交', icon: 'mdi mdi-send-outline', tone: 'info' }, claim_submitted: { label: '报销单已提交', icon: 'mdi mdi-send-outline', tone: 'info' }, approval_stage_completed: { label: '审批节点已完成', icon: 'mdi mdi-account-check-outline', tone: 'success' }, @@ -83,7 +84,30 @@ function buildStateTransitionSummary(payload) { return transition || (nextStage ? `当前节点：${nextStage}` : '') } +function buildHistoricalClaimImportSummary(payload) { + const currentStatus = resolveStatusLabel(readField(payload, 'next_status', 'nextStatus')) + const currentStage = compactText(readField(payload, 'next_approval_stage', 'nextApprovalStage')) + const currentStateParts = [] + + if (currentStatus) { + currentStateParts.push(`纳入时状态：${currentStatus}`) + } + if (currentStage) { + currentStateParts.push(`纳入时节点：${currentStage}`) + } + + const currentStateSummary = currentStateParts.length + ? `${currentStateParts.join(' · ')}。` + : '' + + return `系统已根据原单据建立费用事件档案。${currentStateSummary}导入前办理记录以原单据为准，后续动作将在此持续记录。` +} + function buildEventSummary(eventType, payload) { + if (eventType === 'historical_claim_imported') { + return buildHistoricalClaimImportSummary(payload) + } + const reason = compactText(payload?.reason) if (reason) { return reason @@ -159,7 +183,7 @@ function normalizeEvent(event, index) { icon: presentation.icon, tone: presentation.tone, summary: buildEventSummary(eventType, payload), - actorLabel: resolveActorLabel(event), + actorLabel: eventType === 'historical_claim_imported' ? '系统' : resolveActorLabel(event), occurredAt: compactText(occurredAt), occurredAtLabel: formatOccurredAt(occurredAt), sortTimestamp: Number.isNaN(timestamp) ? null : timestamp, diff --git a/web/tests/expense-case-timeline.test.mjs b/web/tests/expense-case-timeline.test.mjs index 90befb7..bc957d6 100644 --- a/web/tests/expense-case-timeline.test.mjs +++ b/web/tests/expense-case-timeline.test.mjs @@ -100,6 +100,64 @@ test('expense case timeline gives returned events an actionable reason', () => { assert.equal(timeline.events[0].summary, '住宿票据缺少入住人信息') }) +test('expense case timeline explains a historical import without pretending it is a business action', () => { + const timeline = buildExpenseCaseTimelineViewModel({ + events: [ + { + id: 'historical-import-1', + event_type: 'historical_claim_imported', + actor_id: 'migration-worker', + actor_type: 'user', + occurred_at: '2026-07-14T03:00:00Z', + correlation_id: 'private-correlation', + delivery_status: 'pending', + payload_json: { + reason: '不应覆盖历史导入事件的专用说明', + next_status: 'submitted', + next_approval_stage: '直属领导审批', + migration_batch_id: 'private-batch', + source_table: 'expense_claims' + } + } + ] + }) + + const [event] = timeline.events + assert.equal(event.label, '历史单据已纳入统一费用事件') + assert.equal(event.icon, 'mdi mdi-history') + assert.equal(event.tone, 'info') + assert.equal(event.actorLabel, '系统') + assert.equal( + event.summary, + '系统已根据原单据建立费用事件档案。纳入时状态：已提交 · 纳入时节点：直属领导审批。导入前办理记录以原单据为准，后续动作将在此持续记录。' + ) + assert.doesNotMatch( + JSON.stringify(event), + /不应覆盖|private-correlation|pending|private-batch|expense_claims/ + ) +}) + +test('expense case timeline keeps the historical import summary honest when state is missing', () => { + const timeline = buildExpenseCaseTimelineViewModel({ + events: [ + { + id: 'historical-import-without-state', + event_type: 'historical_claim_imported', + actor_id: 'system', + occurred_at: '2026-07-14T03:00:00Z', + payload_json: {} + } + ] + }) + + const [event] = timeline.events + assert.equal( + event.summary, + '系统已根据原单据建立费用事件档案。导入前办理记录以原单据为准，后续动作将在此持续记录。' + ) + assert.doesNotMatch(event.summary, /纳入时状态|纳入时节点|undefined|null/) +}) + test('expense case timeline describes application archive against the paid reimbursement', () => { const timeline = buildExpenseCaseTimelineViewModel({ events: [