B-39: 数据库持久化与不可变导入

- 版本化 SQLite 迁移（向前/回滚）与连接助手。 - 导入管线：上传文件按 SHA-256 内容哈希不可变保存，文件/行/批次幂等；重复上传返回 duplicate 并复用已有批次，不产生第二份事实。 - 每条规范化源行反查文件、工作表、原始行号与模板版本。 - 解析失败保留 exception/failed 批次与诊断，不产生已确认交易。 - 决策记录见 docs/decisions/002-persistence.md。
2026-08-16 01:48:53 +08:00
parent c292fb791d
commit 763a940683
7 changed files with 994 additions and 0 deletions
@@ -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 <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` 自动化验证。
@@ -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())
@@ -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
+    )
@@ -39,3 +39,4 @@ class StatementBatch:
    period_end: date | None
    transactions: tuple[NormalizedTransaction, ...]
    warnings: tuple[str, ...]
+    template_version: int = 1
@@ -190,6 +190,7 @@ def _parse_sheet(source: Path, sheet: RawSheet) -> StatementBatch:
        else None,
        transactions=tuple(transactions),
        warnings=tuple(warnings),
+        template_version=template.version,
    )


@@ -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:
@@ -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()