From 763a9406835770c712113b77d72e095fdab6dd7c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=85=BE=E8=AE=AFWorkBuddy?= Date: Sun, 16 Aug 2026 01:48:53 +0800 Subject: [PATCH] =?UTF-8?q?B-39:=20=E6=95=B0=E6=8D=AE=E5=BA=93=E6=8C=81?= =?UTF-8?q?=E4=B9=85=E5=8C=96=E4=B8=8E=E4=B8=8D=E5=8F=AF=E5=8F=98=E5=AF=BC?= =?UTF-8?q?=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 版本化 SQLite 迁移（向前/回滚）与连接助手。 - 导入管线：上传文件按 SHA-256 内容哈希不可变保存，文件/行/批次 幂等；重复上传返回 duplicate 并复用已有批次，不产生第二份事实。 - 每条规范化源行反查文件、工作表、原始行号与模板版本。 - 解析失败保留 exception/failed 批次与诊断，不产生已确认交易。 - 决策记录见 docs/decisions/002-persistence.md。 --- docs/decisions/002-persistence.md | 44 ++++ src/bank_importer/db.py | 410 ++++++++++++++++++++++++++++++ src/bank_importer/importing.py | 252 ++++++++++++++++++ src/bank_importer/models.py | 1 + src/bank_importer/parser.py | 1 + src/bank_importer/templates.py | 1 + tests/test_persistence.py | 285 +++++++++++++++++++++ 7 files changed, 994 insertions(+) create mode 100644 docs/decisions/002-persistence.md create mode 100644 src/bank_importer/db.py create mode 100644 src/bank_importer/importing.py create mode 100644 tests/test_persistence.py diff --git a/docs/decisions/002-persistence.md b/docs/decisions/002-persistence.md new file mode 100644 index 0000000..dd378c5 --- /dev/null +++ b/docs/decisions/002-persistence.md @@ -0,0 +1,44 @@ +# 002 持久化层技术决策 + +对应 Issue：B-39（`docs/issues/002-p0-persistence-and-immutable-imports.md`）。 + +## 数据库：SQLite（Python 标准库 `sqlite3`） + +- 单文件、事务完整、零新增依赖，与当前标准库服务器和离线内网部署环境匹配。 +- 外键约束、CHECK 约束、唯一约束和触发器均可用，足以承载不可变证据模型。 +- 金额以 `TEXT` 保存 `Decimal` 的原始字符串，读取时还原为 `Decimal`， + 绝不经过二进制浮点。 +- 后续若并发写入成为瓶颈，可开启 WAL 或平迁 PostgreSQL；迁移版本表 + `schema_migrations` 不绑定具体引擎方言之外的特性。 + +## 迁移工具：仓库内置版本化迁移器（`src/bank_importer/db.py`） + +- 每条迁移包含 `up` / `down` 两段 SQL，按版本号顺序执行并记录在 + `schema_migrations` 表中；重复执行无副作用。 +- 不引入 Alembic 等外部工具：当前模型规模小，内置迁移器保持零依赖， + 且回滚路径明确（`python -m bank_importer.db --rollback-to <版本>`）。 +- 服务启动时自动执行 `migrate`，空数据库即可完整建表。 + +## 文件存储：内容寻址的本地文件系统（`data/files/`） + +- 上传文件先计算 SHA-256，再写入 `data/files/<哈希前两位>/<完整哈希>.<扩展名>`。 +- 发布采用临时文件 + 硬链接：目标要么完整出现、要么不存在，且永不覆盖。 +- 相同内容（即使文件名不同）只保存一份文件、一条 `source_files` 记录。 +- `data/` 已加入 `.gitignore`，真实流水不进入版本库。 + +## 幂等与状态机 + +- 文件级幂等：`source_files.sha256` 唯一约束。重复上传插入一条 + `status='duplicate'` 的批次审计记录，指向首个批次，不产生第二份事实。 +- 行级幂等：`source_rows` 上 `UNIQUE (sheet_batch_id, source_row)`。 +- 批次状态：`parsing → parsed | exception | failed`，另有 `duplicate`。 + 失败/异常批次保留 `import_exceptions` 诊断（阶段、消息、原始文件名）， + 不产生任何已确认源行。 +- 事务边界：文件落盘后，`source_files` + 批次 + 工作表批次 + 源行在单个 + SQLite 事务内提交；解析失败时批次状态与异常记录同样在事务内落库。 + +## 不可变性 + +- `source_files`、`sheet_batches`、`source_rows` 三张表由数据库触发器禁止 + `UPDATE` 和 `DELETE`，任何修改只能走后续阶段的冲销/审计调整流程。 +- 该约束由 `tests/test_persistence.py` 自动化验证。 diff --git a/src/bank_importer/db.py b/src/bank_importer/db.py new file mode 100644 index 0000000..eda6221 --- /dev/null +++ b/src/bank_importer/db.py @@ -0,0 +1,410 @@ +"""SQLite persistence: versioned migrations and connection helpers. + +The database, migration runner and file storage choices are recorded in +``docs/decisions/002-persistence.md``. Migrations are plain SQL applied in +version order; each records itself in ``schema_migrations`` so re-running +``migrate`` on an existing database is a no-op. Every migration ships a +``down`` script so ``rollback`` can walk backwards for recovery and tests. +""" + +from __future__ import annotations + +import argparse +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +import sqlite3 + + +DEFAULT_DB_PATH = Path("data/app.db") + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +@dataclass(frozen=True) +class Migration: + version: int + name: str + up: str + down: str + + +MIGRATIONS: tuple[Migration, ...] = ( + Migration( + version=1, + name="0001_core_persistence", + up=""" + CREATE TABLE companies ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL UNIQUE, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + + CREATE TABLE bank_accounts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + company_id INTEGER REFERENCES companies (id), + account_number TEXT NOT NULL UNIQUE, + account_name TEXT, + bank_name TEXT, + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'active', 'disabled')), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + + CREATE TABLE source_files ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + sha256 TEXT NOT NULL UNIQUE, + original_filename TEXT NOT NULL, + size_bytes INTEGER NOT NULL, + storage_path TEXT NOT NULL, + created_at TEXT NOT NULL + ); + + CREATE TABLE import_batches ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source_file_id INTEGER NOT NULL REFERENCES source_files (id), + status TEXT NOT NULL + CHECK (status IN ('parsing', 'parsed', 'exception', 'failed', 'duplicate')), + duplicate_of_id INTEGER REFERENCES import_batches (id), + diagnostics TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + + CREATE TABLE sheet_batches ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + import_batch_id INTEGER NOT NULL REFERENCES import_batches (id), + sheet_name TEXT NOT NULL, + bank_name TEXT NOT NULL, + template_id TEXT NOT NULL, + template_version INTEGER NOT NULL, + header_row INTEGER NOT NULL, + own_account TEXT, + own_name TEXT, + period_start TEXT, + period_end TEXT, + transaction_count INTEGER NOT NULL, + warnings TEXT NOT NULL DEFAULT '[]', + created_at TEXT NOT NULL, + UNIQUE (import_batch_id, sheet_name) + ); + + CREATE TABLE source_rows ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + sheet_batch_id INTEGER NOT NULL REFERENCES sheet_batches (id), + source_row INTEGER NOT NULL, + transaction_at TEXT NOT NULL, + income TEXT NOT NULL, + expense TEXT NOT NULL, + balance TEXT, + own_account TEXT, + own_name TEXT, + counterparty_account TEXT, + counterparty_name TEXT, + counterparty_bank TEXT, + summary TEXT, + purpose TEXT, + reference TEXT, + currency TEXT, + created_at TEXT NOT NULL, + UNIQUE (sheet_batch_id, source_row) + ); + + CREATE TABLE import_exceptions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + import_batch_id INTEGER NOT NULL REFERENCES import_batches (id), + stage TEXT NOT NULL, + message TEXT NOT NULL, + diagnostics TEXT, + created_at TEXT NOT NULL + ); + + CREATE TRIGGER source_files_no_update BEFORE UPDATE ON source_files + BEGIN SELECT RAISE (ABORT, 'source_files rows are immutable'); END; + CREATE TRIGGER source_files_no_delete BEFORE DELETE ON source_files + BEGIN SELECT RAISE (ABORT, 'source_files rows are immutable'); END; + CREATE TRIGGER sheet_batches_no_update BEFORE UPDATE ON sheet_batches + BEGIN SELECT RAISE (ABORT, 'sheet_batches rows are immutable'); END; + CREATE TRIGGER sheet_batches_no_delete BEFORE DELETE ON sheet_batches + BEGIN SELECT RAISE (ABORT, 'sheet_batches rows are immutable'); END; + CREATE TRIGGER source_rows_no_update BEFORE UPDATE ON source_rows + BEGIN SELECT RAISE (ABORT, 'source_rows rows are immutable'); END; + CREATE TRIGGER source_rows_no_delete BEFORE DELETE ON source_rows + BEGIN SELECT RAISE (ABORT, 'source_rows rows are immutable'); END; + """, + down=""" + DROP TRIGGER IF EXISTS source_rows_no_delete; + DROP TRIGGER IF EXISTS source_rows_no_update; + DROP TRIGGER IF EXISTS sheet_batches_no_delete; + DROP TRIGGER IF EXISTS sheet_batches_no_update; + DROP TRIGGER IF EXISTS source_files_no_delete; + DROP TRIGGER IF EXISTS source_files_no_update; + DROP TABLE IF EXISTS import_exceptions; + DROP TABLE IF EXISTS source_rows; + DROP TABLE IF EXISTS sheet_batches; + DROP TABLE IF EXISTS import_batches; + DROP TABLE IF EXISTS source_files; + DROP TABLE IF EXISTS bank_accounts; + DROP TABLE IF EXISTS companies; + """, + ), + Migration( + version=2, + name="0002_auth_and_sessions", + up=""" + CREATE TABLE users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + username TEXT NOT NULL UNIQUE, + password_hash TEXT NOT NULL, + role TEXT NOT NULL CHECK (role IN ('admin', 'company')), + company_id INTEGER REFERENCES companies (id), + status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'disabled')), + must_change_password INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + CHECK (role != 'company' OR company_id IS NOT NULL), + CHECK (role != 'admin' OR company_id IS NULL) + ); + + CREATE TABLE sessions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + token_hash TEXT NOT NULL UNIQUE, + user_id INTEGER NOT NULL REFERENCES users (id), + created_at TEXT NOT NULL, + expires_at TEXT NOT NULL, + revoked_at TEXT + ); + + CREATE TABLE login_attempts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + username TEXT NOT NULL, + ip TEXT NOT NULL, + success INTEGER NOT NULL, + created_at TEXT NOT NULL + ); + + CREATE TABLE audit_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + actor_user_id INTEGER REFERENCES users (id), + actor_username TEXT, + action TEXT NOT NULL, + target TEXT, + detail TEXT, + ip TEXT, + created_at TEXT NOT NULL + ); + + ALTER TABLE import_batches ADD COLUMN company_id INTEGER REFERENCES companies (id); + """, + down=""" + DROP TABLE IF EXISTS audit_log; + DROP TABLE IF EXISTS login_attempts; + DROP TABLE IF EXISTS sessions; + DROP TABLE IF EXISTS users; + ALTER TABLE import_batches DROP COLUMN company_id; + """, + ), + Migration( + version=3, + name="0003_dynamic_master_data", + # bank_accounts is rebuilt (SQLite cannot alter CHECK constraints): + # status gains 'returned', and the account gains type, effective + # interval and review fields. account_number stays UNIQUE and stores + # the normalized digits-only form (see master_data.normalize). + up=""" + CREATE TABLE bank_accounts_new ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + company_id INTEGER NOT NULL REFERENCES companies (id), + account_number TEXT NOT NULL UNIQUE, + account_name TEXT, + bank_name TEXT NOT NULL DEFAULT '', + account_type TEXT NOT NULL DEFAULT '一般户' + CHECK (account_type IN ('基本户', '一般户', '专用户')), + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'active', 'returned', 'disabled')), + effective_from TEXT, + effective_to TEXT, + submitted_by INTEGER REFERENCES users (id), + reviewed_by INTEGER REFERENCES users (id), + reviewed_at TEXT, + review_reason TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + INSERT INTO bank_accounts_new ( + id, company_id, account_number, account_name, bank_name, + status, created_at, updated_at + ) + SELECT id, company_id, account_number, account_name, + COALESCE(bank_name, ''), status, created_at, updated_at + FROM bank_accounts WHERE company_id IS NOT NULL; + DROP TABLE bank_accounts; + ALTER TABLE bank_accounts_new RENAME TO bank_accounts; + + CREATE TABLE account_aliases ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + bank_account_id INTEGER NOT NULL REFERENCES bank_accounts (id), + alias_kind TEXT NOT NULL CHECK (alias_kind IN ('name', 'account')), + alias_value TEXT NOT NULL, + priority INTEGER NOT NULL DEFAULT 100, + effective_from TEXT, + effective_to TEXT, + created_by INTEGER REFERENCES users (id), + created_at TEXT NOT NULL, + UNIQUE (bank_account_id, alias_kind, alias_value) + ); + + CREATE TABLE master_data_changes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + entity_type TEXT NOT NULL + CHECK (entity_type IN ('company', 'user', 'bank_account', 'account_alias')), + entity_id INTEGER NOT NULL, + action TEXT NOT NULL, + before_json TEXT, + after_json TEXT, + reason TEXT, + actor_user_id INTEGER REFERENCES users (id), + actor_username TEXT, + created_at TEXT NOT NULL + ); + + ALTER TABLE companies ADD COLUMN credit_code TEXT; + ALTER TABLE companies ADD COLUMN cashier_name TEXT; + ALTER TABLE companies ADD COLUMN status TEXT NOT NULL DEFAULT 'active' + CHECK (status IN ('active', 'preparing', 'disabled')); + """, + down=""" + ALTER TABLE companies DROP COLUMN status; + ALTER TABLE companies DROP COLUMN cashier_name; + ALTER TABLE companies DROP COLUMN credit_code; + DROP TABLE IF EXISTS master_data_changes; + DROP TABLE IF EXISTS account_aliases; + + CREATE TABLE bank_accounts_new ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + company_id INTEGER REFERENCES companies (id), + account_number TEXT NOT NULL UNIQUE, + account_name TEXT, + bank_name TEXT, + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'active', 'disabled')), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + INSERT INTO bank_accounts_new ( + id, company_id, account_number, account_name, bank_name, + status, created_at, updated_at + ) + SELECT id, company_id, account_number, account_name, bank_name, + CASE WHEN status = 'returned' THEN 'pending' ELSE status END, + created_at, updated_at + FROM bank_accounts; + DROP TABLE bank_accounts; + ALTER TABLE bank_accounts_new RENAME TO bank_accounts; + """, + ), +) + + +def connect(path: str | Path) -> sqlite3.Connection: + db_path = Path(path) + if str(db_path) != ":memory:": + db_path.parent.mkdir(parents=True, exist_ok=True) + connection = sqlite3.connect(str(db_path)) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA foreign_keys = ON") + return connection + + +def applied_versions(connection: sqlite3.Connection) -> list[int]: + exists = connection.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'schema_migrations'" + ).fetchone() + if not exists: + return [] + rows = connection.execute( + "SELECT version FROM schema_migrations ORDER BY version" + ).fetchall() + return [row["version"] for row in rows] + + +def migrate(connection: sqlite3.Connection) -> list[int]: + """Apply every pending migration; returns the versions applied now.""" + connection.execute( + """ + CREATE TABLE IF NOT EXISTS schema_migrations ( + version INTEGER PRIMARY KEY, + name TEXT NOT NULL, + applied_at TEXT NOT NULL + ) + """ + ) + applied = set(applied_versions(connection)) + newly_applied: list[int] = [] + for migration in MIGRATIONS: + if migration.version in applied: + continue + with connection: + connection.executescript(migration.up) + connection.execute( + "INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, ?)", + (migration.version, migration.name, utc_now()), + ) + newly_applied.append(migration.version) + return newly_applied + + +def rollback(connection: sqlite3.Connection, target_version: int = 0) -> list[int]: + """Reverse migrations above ``target_version``; returns reversed versions.""" + applied = applied_versions(connection) + reversed_versions: list[int] = [] + for migration in sorted(MIGRATIONS, key=lambda item: item.version, reverse=True): + if migration.version <= target_version or migration.version not in applied: + continue + with connection: + connection.executescript(migration.down) + connection.execute( + "DELETE FROM schema_migrations WHERE version = ?", + (migration.version,), + ) + reversed_versions.append(migration.version) + return reversed_versions + + +def main() -> int: + parser = argparse.ArgumentParser(description="Apply or roll back database migrations.") + parser.add_argument( + "db_path", + type=Path, + nargs="?", + default=DEFAULT_DB_PATH, + help="SQLite database path (default: data/app.db)", + ) + parser.add_argument( + "--rollback-to", + type=int, + default=None, + metavar="VERSION", + help="Reverse migrations above VERSION instead of migrating forward", + ) + args = parser.parse_args() + + connection = connect(args.db_path) + try: + if args.rollback_to is None: + applied = migrate(connection) + print(f"applied migrations: {applied or 'none (already up to date)'}") + else: + reversed_versions = rollback(connection, args.rollback_to) + print(f"reversed migrations: {reversed_versions or 'none'}") + finally: + connection.close() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/bank_importer/importing.py b/src/bank_importer/importing.py new file mode 100644 index 0000000..e98b3ca --- /dev/null +++ b/src/bank_importer/importing.py @@ -0,0 +1,252 @@ +"""Immutable statement import pipeline. + +Every uploaded file is hashed (SHA-256) and written once to content-addressed +storage before parsing. A repeated upload of identical bytes never creates a +second set of facts: it records a ``duplicate`` batch that points at the +original batch. Parse failures keep the batch and its diagnostics as an +``exception`` batch without producing any confirmed source rows. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import json +import os +from pathlib import Path +import sqlite3 + +from .db import utc_now +from .models import StatementBatch +from .parser import StatementParseError, parse_statement + + +@dataclass(frozen=True) +class ImportResult: + batch_id: int + status: str # parsed | duplicate | exception + sha256: str + source_file_id: int + batches: tuple[StatementBatch, ...] = () + message: str | None = None + + +def import_statement( + connection: sqlite3.Connection, + storage_dir: str | Path, + original_filename: str, + content: bytes, + company_id: int | None = None, +) -> ImportResult: + sha256 = hashlib.sha256(content).hexdigest() + existing_file = connection.execute( + "SELECT id FROM source_files WHERE sha256 = ?", (sha256,) + ).fetchone() + + if existing_file is not None: + return _record_duplicate(connection, existing_file["id"], sha256, company_id) + + stored_path = _store_immutable(Path(storage_dir), original_filename, content, sha256) + now = utc_now() + with connection: + cursor = connection.execute( + """ + INSERT INTO source_files (sha256, original_filename, size_bytes, storage_path, created_at) + VALUES (?, ?, ?, ?, ?) + """, + (sha256, original_filename, len(content), str(stored_path), now), + ) + source_file_id = cursor.lastrowid + cursor = connection.execute( + """ + INSERT INTO import_batches (source_file_id, status, company_id, created_at, updated_at) + VALUES (?, 'parsing', ?, ?, ?) + """, + (source_file_id, company_id, now, now), + ) + batch_id = cursor.lastrowid + + try: + batches = parse_statement(stored_path) + except StatementParseError as exc: + message = _clean_message(str(exc), stored_path, original_filename) + with connection: + _insert_exception(connection, batch_id, "parse", message, original_filename) + _set_batch_status(connection, batch_id, "exception") + return ImportResult(batch_id, "exception", sha256, source_file_id, message=message) + except Exception as exc: + message = f"文件解析失败，请检查文件是否完整。（{type(exc).__name__}）" + with connection: + _insert_exception(connection, batch_id, "internal", message, original_filename) + _set_batch_status(connection, batch_id, "failed") + raise + + with connection: + for batch in batches: + _insert_sheet_batch(connection, batch_id, batch) + _set_batch_status(connection, batch_id, "parsed") + return ImportResult(batch_id, "parsed", sha256, source_file_id, batches=batches) + + +def _record_duplicate( + connection: sqlite3.Connection, + source_file_id: int, + sha256: str, + company_id: int | None = None, +) -> ImportResult: + original = connection.execute( + """ + SELECT id FROM import_batches + WHERE source_file_id = ? AND status = 'parsed' + ORDER BY id LIMIT 1 + """, + (source_file_id,), + ).fetchone() + if original is None: + original = connection.execute( + """ + SELECT id FROM import_batches + WHERE source_file_id = ? AND status != 'duplicate' + ORDER BY id LIMIT 1 + """, + (source_file_id,), + ).fetchone() + now = utc_now() + with connection: + connection.execute( + """ + INSERT INTO import_batches + (source_file_id, status, duplicate_of_id, company_id, diagnostics, created_at, updated_at) + VALUES (?, 'duplicate', ?, ?, ?, ?, ?) + """, + ( + source_file_id, + original["id"], + company_id, + json.dumps({"note": "内容哈希相同，复用已有批次，不产生第二份事实。"}, ensure_ascii=False), + now, + now, + ), + ) + return ImportResult( + original["id"], + "duplicate", + sha256, + source_file_id, + message="相同内容的文件已导入，本次按重复上传处理。", + ) + + +def _store_immutable( + storage_dir: Path, original_filename: str, content: bytes, sha256: str +) -> Path: + suffix = Path(original_filename).suffix.lower() + target_dir = storage_dir / sha256[:2] + target_dir.mkdir(parents=True, exist_ok=True) + target = target_dir / f"{sha256}{suffix}" + # Publish via a temporary file + hard link: the content-addressed target + # either appears complete or not at all, and is never overwritten. + temp = target_dir / f".{sha256}.tmp" + temp.write_bytes(content) + try: + os.link(temp, target) + except FileExistsError: + # Content-addressed name means identical bytes; never overwrite. + pass + finally: + temp.unlink(missing_ok=True) + return target + + +def _insert_sheet_batch( + connection: sqlite3.Connection, batch_id: int, batch: StatementBatch +) -> None: + now = utc_now() + cursor = connection.execute( + """ + INSERT INTO sheet_batches ( + import_batch_id, sheet_name, bank_name, template_id, template_version, + header_row, own_account, own_name, period_start, period_end, + transaction_count, warnings, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + batch_id, + batch.sheet_name, + batch.bank_name, + batch.template_id, + batch.template_version, + batch.header_row, + batch.own_account, + batch.own_name, + batch.period_start.isoformat() if batch.period_start else None, + batch.period_end.isoformat() if batch.period_end else None, + len(batch.transactions), + json.dumps(list(batch.warnings), ensure_ascii=False), + now, + ), + ) + sheet_batch_id = cursor.lastrowid + for transaction in batch.transactions: + connection.execute( + """ + INSERT INTO source_rows ( + sheet_batch_id, source_row, transaction_at, income, expense, balance, + own_account, own_name, counterparty_account, counterparty_name, + counterparty_bank, summary, purpose, reference, currency, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + sheet_batch_id, + transaction.source_row, + transaction.transaction_at.isoformat(), + str(transaction.income), + str(transaction.expense), + str(transaction.balance) if transaction.balance is not None else None, + transaction.own_account, + transaction.own_name, + transaction.counterparty_account, + transaction.counterparty_name, + transaction.counterparty_bank, + transaction.summary, + transaction.purpose, + transaction.reference, + transaction.currency, + now, + ), + ) + + +def _insert_exception( + connection: sqlite3.Connection, + batch_id: int, + stage: str, + message: str, + original_filename: str, +) -> None: + connection.execute( + """ + INSERT INTO import_exceptions (import_batch_id, stage, message, diagnostics, created_at) + VALUES (?, ?, ?, ?, ?) + """, + ( + batch_id, + stage, + message, + json.dumps({"original_filename": original_filename}, ensure_ascii=False), + utc_now(), + ), + ) + + +def _set_batch_status(connection: sqlite3.Connection, batch_id: int, status: str) -> None: + connection.execute( + "UPDATE import_batches SET status = ?, updated_at = ? WHERE id = ?", + (status, utc_now(), batch_id), + ) + + +def _clean_message(message: str, stored_path: Path, original_filename: str) -> str: + return message.replace(str(stored_path), original_filename).replace( + stored_path.name, original_filename + ) diff --git a/src/bank_importer/models.py b/src/bank_importer/models.py index 576db2b..3f0ac50 100644 --- a/src/bank_importer/models.py +++ b/src/bank_importer/models.py @@ -39,3 +39,4 @@ class StatementBatch: period_end: date | None transactions: tuple[NormalizedTransaction, ...] warnings: tuple[str, ...] + template_version: int = 1 diff --git a/src/bank_importer/parser.py b/src/bank_importer/parser.py index afcff95..0e83b15 100644 --- a/src/bank_importer/parser.py +++ b/src/bank_importer/parser.py @@ -190,6 +190,7 @@ def _parse_sheet(source: Path, sheet: RawSheet) -> StatementBatch: else None, transactions=tuple(transactions), warnings=tuple(warnings), + template_version=template.version, ) diff --git a/src/bank_importer/templates.py b/src/bank_importer/templates.py index b448bf2..0df3b9e 100644 --- a/src/bank_importer/templates.py +++ b/src/bank_importer/templates.py @@ -10,6 +10,7 @@ class BankTemplate: bank_name: str columns: dict[str, tuple[str, ...]] required: tuple[str, ...] + version: int = 1 def normalize_header(value: object) -> str: diff --git a/tests/test_persistence.py b/tests/test_persistence.py new file mode 100644 index 0000000..227c195 --- /dev/null +++ b/tests/test_persistence.py @@ -0,0 +1,285 @@ +from __future__ import annotations + +from decimal import Decimal +import hashlib +import sqlite3 +import tempfile +from pathlib import Path +import unittest + +from openpyxl import Workbook + +from bank_importer.db import applied_versions, connect, migrate, rollback +from bank_importer.importing import import_statement + + +ROOT = Path(__file__).resolve().parents[1] +SAMPLES = ROOT / "流水模板" +SAMPLE_FILE = SAMPLES / "中国建设银行账户流水.xls" + + +class PersistenceTestCase(unittest.TestCase): + def setUp(self) -> None: + self.temp_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.temp_dir.cleanup) + root = Path(self.temp_dir.name) + self.db_path = root / "app.db" + self.storage = root / "files" + self.connection = connect(self.db_path) + self.addCleanup(self.connection.close) + migrate(self.connection) + + def import_sample(self) -> object: + return import_statement( + self.connection, + self.storage, + SAMPLE_FILE.name, + SAMPLE_FILE.read_bytes(), + ) + + +class MigrationTests(PersistenceTestCase): + def test_migrate_creates_schema_and_is_idempotent(self) -> None: + first = applied_versions(self.connection) + self.assertEqual([1, 2, 3], first) + self.assertEqual([], migrate(self.connection)) + self.assertEqual(first, applied_versions(self.connection)) + tables = { + row["name"] + for row in self.connection.execute( + "SELECT name FROM sqlite_master WHERE type = 'table'" + ) + } + for table in ( + "companies", + "bank_accounts", + "account_aliases", + "master_data_changes", + "source_files", + "import_batches", + "sheet_batches", + "source_rows", + "import_exceptions", + "users", + "sessions", + "login_attempts", + "audit_log", + "schema_migrations", + ): + self.assertIn(table, tables) + + def test_rollback_removes_schema_and_forward_rebuilds_it(self) -> None: + self.assertEqual([3, 2, 1], rollback(self.connection, 0)) + self.assertEqual([], applied_versions(self.connection)) + remaining = self.connection.execute( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'source_rows'" + ).fetchone() + self.assertIsNone(remaining) + self.assertEqual([1, 2, 3], migrate(self.connection)) + self.assertEqual([1, 2, 3], applied_versions(self.connection)) + + +class ImportPersistenceTests(PersistenceTestCase): + def test_import_persists_batch_sheets_and_rows(self) -> None: + result = self.import_sample() + self.assertEqual("parsed", result.status) + + batch = self.connection.execute( + "SELECT status FROM import_batches WHERE id = ?", (result.batch_id,) + ).fetchone() + self.assertEqual("parsed", batch["status"]) + + sheets = self.connection.execute( + "SELECT * FROM sheet_batches WHERE import_batch_id = ?", (result.batch_id,) + ).fetchall() + self.assertEqual(1, len(sheets)) + self.assertEqual("中国建设银行", sheets[0]["bank_name"]) + self.assertEqual("ccb-account-detail-v1", sheets[0]["template_id"]) + self.assertEqual(1, sheets[0]["template_version"]) + + rows = self.connection.execute( + "SELECT * FROM source_rows WHERE sheet_batch_id = ?", (sheets[0]["id"],) + ).fetchall() + self.assertEqual(sheets[0]["transaction_count"], len(rows)) + + def test_amounts_keep_decimal_precision(self) -> None: + result = self.import_sample() + rows = self.connection.execute( + """ + SELECT income, expense FROM source_rows + WHERE sheet_batch_id IN ( + SELECT id FROM sheet_batches WHERE import_batch_id = ? + ) + """, + (result.batch_id,), + ).fetchall() + self.assertTrue(rows) + for row in rows: + self.assertEqual(row["income"], str(Decimal(row["income"]))) + self.assertEqual(row["expense"], str(Decimal(row["expense"]))) + + def test_every_row_traces_back_to_file_sheet_and_template(self) -> None: + result = self.import_sample() + row = self.connection.execute( + """ + SELECT r.source_row, s.sheet_name, s.template_id, s.template_version, f.sha256 + FROM source_rows r + JOIN sheet_batches s ON s.id = r.sheet_batch_id + JOIN import_batches b ON b.id = s.import_batch_id + JOIN source_files f ON f.id = b.source_file_id + WHERE b.id = ? + LIMIT 1 + """, + (result.batch_id,), + ).fetchone() + self.assertIsNotNone(row) + self.assertEqual( + hashlib.sha256(SAMPLE_FILE.read_bytes()).hexdigest(), row["sha256"] + ) + self.assertEqual("ccb-account-detail-v1", row["template_id"]) + self.assertEqual(1, row["template_version"]) + self.assertGreater(row["source_row"], 0) + self.assertTrue(row["sheet_name"]) + + def test_source_file_is_stored_immutably_by_content_hash(self) -> None: + self.import_sample() + stored = self.connection.execute( + "SELECT storage_path FROM source_files" + ).fetchone() + stored_path = Path(stored["storage_path"]) + self.assertTrue(stored_path.is_file()) + self.assertEqual(SAMPLE_FILE.read_bytes(), stored_path.read_bytes()) + self.assertIn(hashlib.sha256(SAMPLE_FILE.read_bytes()).hexdigest(), stored_path.name) + + def test_data_survives_reconnect(self) -> None: + result = self.import_sample() + self.connection.close() + + reopened = connect(self.db_path) + self.addCleanup(reopened.close) + batch = reopened.execute( + "SELECT status FROM import_batches WHERE id = ?", (result.batch_id,) + ).fetchone() + self.assertEqual("parsed", batch["status"]) + files = reopened.execute("SELECT COUNT(*) AS n FROM source_files").fetchone() + self.assertEqual(1, files["n"]) + rows = reopened.execute("SELECT COUNT(*) AS n FROM source_rows").fetchone() + self.assertGreater(rows["n"], 0) + + +class IdempotencyTests(PersistenceTestCase): + def test_duplicate_upload_creates_no_second_facts(self) -> None: + first = self.import_sample() + second = self.import_sample() + + self.assertEqual("parsed", first.status) + self.assertEqual("duplicate", second.status) + self.assertEqual(first.batch_id, second.batch_id) + + files = self.connection.execute("SELECT COUNT(*) AS n FROM source_files").fetchone() + self.assertEqual(1, files["n"]) + sheets = self.connection.execute("SELECT COUNT(*) AS n FROM sheet_batches").fetchone() + self.assertEqual(1, sheets["n"]) + rows = self.connection.execute("SELECT COUNT(*) AS n FROM source_rows").fetchone() + self.assertEqual(first.batches[0].transactions.__len__(), rows["n"]) + + duplicate = self.connection.execute( + "SELECT status, duplicate_of_id FROM import_batches WHERE status = 'duplicate'" + ).fetchone() + self.assertEqual(first.batch_id, duplicate["duplicate_of_id"]) + + def test_repeated_upload_under_a_different_filename_is_still_duplicate(self) -> None: + first = self.import_sample() + second = import_statement( + self.connection, + self.storage, + "改名后的流水.xls", + SAMPLE_FILE.read_bytes(), + ) + self.assertEqual("duplicate", second.status) + self.assertEqual(first.batch_id, second.batch_id) + files = self.connection.execute("SELECT COUNT(*) AS n FROM source_files").fetchone() + self.assertEqual(1, files["n"]) + + def test_immutability_triggers_block_updates_and_deletes(self) -> None: + self.import_sample() + for statement in ( + "UPDATE source_rows SET income = '0'", + "DELETE FROM source_rows", + "UPDATE source_files SET sha256 = 'x'", + "DELETE FROM source_files", + "UPDATE sheet_batches SET bank_name = 'x'", + "DELETE FROM sheet_batches", + ): + with self.subTest(statement=statement): + with self.assertRaises(sqlite3.IntegrityError): + self.connection.execute(statement) + self.connection.rollback() + + def test_row_level_unique_constraint_blocks_duplicate_rows(self) -> None: + self.import_sample() + row = self.connection.execute( + "SELECT sheet_batch_id, source_row FROM source_rows LIMIT 1" + ).fetchone() + with self.assertRaises(sqlite3.IntegrityError): + self.connection.execute( + """ + INSERT INTO source_rows ( + sheet_batch_id, source_row, transaction_at, income, expense, created_at + ) VALUES (?, ?, '2026-01-01T00:00:00', '1', '0', '2026-01-01T00:00:00Z') + """, + (row["sheet_batch_id"], row["source_row"]), + ) + self.connection.rollback() + + +class FailedImportTests(PersistenceTestCase): + def _unknown_template_bytes(self) -> bytes: + workbook = Workbook() + sheet = workbook.active + sheet.title = "流水" + sheet.append(["日期", "金额", "备注"]) + sheet.append(["2026-01-01", "100.00", "测试"]) + target = Path(self.temp_dir.name) / "unknown.xlsx" + workbook.save(target) + return target.read_bytes() + + def test_parse_failure_keeps_exception_batch_without_rows(self) -> None: + result = import_statement( + self.connection, self.storage, "未知银行.xlsx", self._unknown_template_bytes() + ) + self.assertEqual("exception", result.status) + self.assertIsNotNone(result.message) + self.assertIn("未知银行.xlsx", result.message) + self.assertNotIn(result.sha256, result.message) + + batch = self.connection.execute( + "SELECT status FROM import_batches WHERE id = ?", (result.batch_id,) + ).fetchone() + self.assertEqual("exception", batch["status"]) + + exceptions = self.connection.execute( + "SELECT * FROM import_exceptions WHERE import_batch_id = ?", + (result.batch_id,), + ).fetchall() + self.assertEqual(1, len(exceptions)) + self.assertEqual("parse", exceptions[0]["stage"]) + + rows = self.connection.execute("SELECT COUNT(*) AS n FROM source_rows").fetchone() + self.assertEqual(0, rows["n"]) + + # The source file is still preserved as evidence for later diagnosis. + files = self.connection.execute("SELECT COUNT(*) AS n FROM source_files").fetchone() + self.assertEqual(1, files["n"]) + + def test_reupload_after_failure_is_tracked_as_duplicate(self) -> None: + content = self._unknown_template_bytes() + first = import_statement(self.connection, self.storage, "未知银行.xlsx", content) + second = import_statement(self.connection, self.storage, "未知银行.xlsx", content) + self.assertEqual("exception", first.status) + self.assertEqual("duplicate", second.status) + self.assertEqual(first.batch_id, second.batch_id) + + +if __name__ == "__main__": + unittest.main()