history_store.py

"""NekoCardReader v3 数据档案层。

对现有 UI 保持 v2 API 兼容，同时使用 SQLite 提供：
- 按卡累计、稳定去重与每次扫描快照；
- 原始块归档接口（冻结的读卡层未来暴露 raw bytes 后即可接入）；
- 未解析记录修正、路线几何缓存、应用状态；
- history_v2.json 自动迁移与 SQLite 在线备份。
"""
from __future__ import annotations

import datetime
import json
import os
import shutil
import sqlite3
import threading
from pathlib import Path

from felica_reader import Transaction
import paths

SCHEMA_VERSION = 3
PARSER_VERSION = "v0.2-frozen-core"
COMBINED = "__all__"

_STORE = paths.data_path("history_v2.json")       # 兼容旧代码/迁移源
_DB = paths.data_path("history_v3.sqlite3")
_LOCK = threading.RLock()
_SCHEMA_LOCK = threading.Lock()
_SCHEMA_READY = set()

class _ClosingConnection(sqlite3.Connection):
    """Commit/rollback like sqlite3.Connection, then release the Windows handle."""

    def __exit__(self, exc_type, exc_value, traceback):
        try:
            return super().__exit__(exc_type, exc_value, traceback)
        finally:
            self.close()


def _open_database(path: str, timeout: float = 15) -> sqlite3.Connection:
    return sqlite3.connect(path, timeout=timeout, factory=_ClosingConnection)


def _now() -> str:
    return datetime.datetime.now().isoformat(timespec="microseconds")


def _default_label(idm: str) -> str:
    if not idm or idm == "unknown":
        return "未知卡"
    return f"卡 …{idm[-4:].upper()}"


def _key(d: dict):
    lid = d.get("log_id", 0)
    if lid:
        return ("id", int(lid))
    return ("f", d.get("date_str", ""), d.get("in_station", ""),
            d.get("out_station", ""), d.get("balance", 0), d.get("amount", 0))


def _key_text(d: dict) -> str:
    return json.dumps(_key(d), ensure_ascii=False, separators=(",", ":"))


def _parse_date(s: str):
    if not s:
        return None
    # 中国记录可能带时间，排序/日历只取日期部分。
    s = s.strip()[:10]
    for fmt in ("%Y/%m/%d", "%Y-%m-%d", "%m/%d/%Y", "%Y年%m月%d日", "%Y.%m.%d"):
        try:
            return datetime.datetime.strptime(s, fmt).date()
        except ValueError:
            pass
    return None


def _sort_key(d: dict):
    dt = _parse_date(d.get("date_str", "") or "")
    return (dt or datetime.date.min, int(d.get("log_id", 0)))


def _dict_to_tx(d: dict, index: int) -> Transaction:
    return Transaction(
        index=index,
        date=_parse_date(d.get("date_str", "") or ""),
        date_str=d.get("date_str", "") or "",
        in_station=d.get("in_station", "") or "",
        in_company=d.get("in_company", "") or "",
        out_station=d.get("out_station", "") or "",
        out_company=d.get("out_company", "") or "",
        memo=d.get("memo", "") or "",
        balance=int(d.get("balance", 0)),
        amount=int(d.get("amount", 0)),
        in_commute=bool(d.get("in_commute", False)),
        out_commute=bool(d.get("out_commute", False)),
        log_id=int(d.get("log_id", 0)),
        country=d.get("country", "日本") or "日本",
        region=d.get("region", "") or "",
        city=d.get("city", "") or "",
        line=d.get("line", "") or "",
        in_lat=d.get("in_lat"), in_lng=d.get("in_lng"),
        out_lat=d.get("out_lat"), out_lng=d.get("out_lng"),
    )


def _to_tx_list(records: list[dict]) -> list[Transaction]:
    ordered = sorted(records, key=_sort_key, reverse=True)
    return [_dict_to_tx(d, i) for i, d in enumerate(ordered)]


_SCHEMA = """
CREATE TABLE IF NOT EXISTS meta (
    key TEXT PRIMARY KEY,
    value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS cards (
    card_id TEXT PRIMARY KEY,
    label TEXT NOT NULL,
    country TEXT NOT NULL DEFAULT '',
    created_at TEXT NOT NULL,
    last_seen_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS transactions (
    card_id TEXT NOT NULL REFERENCES cards(card_id) ON DELETE CASCADE,
    txn_key TEXT NOT NULL,
    sort_date TEXT NOT NULL DEFAULT '',
    log_id INTEGER NOT NULL DEFAULT 0,
    record_json TEXT NOT NULL,
    first_seen_at TEXT NOT NULL,
    last_seen_at TEXT NOT NULL,
    PRIMARY KEY(card_id, txn_key)
);
CREATE INDEX IF NOT EXISTS idx_tx_card_date ON transactions(card_id, sort_date DESC, log_id DESC);
CREATE TABLE IF NOT EXISTS scans (
    scan_id INTEGER PRIMARY KEY AUTOINCREMENT,
    card_id TEXT NOT NULL REFERENCES cards(card_id) ON DELETE CASCADE,
    read_at TEXT NOT NULL,
    record_count INTEGER NOT NULL,
    added_count INTEGER NOT NULL,
    parser_version TEXT NOT NULL,
    snapshot_json TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_scans_card_time ON scans(card_id, read_at DESC);
CREATE TABLE IF NOT EXISTS raw_blocks (
    raw_id INTEGER PRIMARY KEY AUTOINCREMENT,
    card_id TEXT NOT NULL REFERENCES cards(card_id) ON DELETE CASCADE,
    scan_id INTEGER REFERENCES scans(scan_id) ON DELETE SET NULL,
    system_code TEXT NOT NULL DEFAULT '',
    service_code TEXT NOT NULL DEFAULT '',
    block_no INTEGER NOT NULL,
    raw_hex TEXT NOT NULL,
    parser_version TEXT NOT NULL,
    archived_at TEXT NOT NULL,
    UNIQUE(card_id, system_code, service_code, block_no, raw_hex)
);
CREATE TABLE IF NOT EXISTS station_overrides (
    card_id TEXT NOT NULL,
    txn_key TEXT NOT NULL,
    field TEXT NOT NULL,
    value TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    PRIMARY KEY(card_id, txn_key, field),
    FOREIGN KEY(card_id, txn_key) REFERENCES transactions(card_id, txn_key) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS route_cache (
    cache_key TEXT PRIMARY KEY,
    source TEXT NOT NULL,
    data_version TEXT NOT NULL,
    geometry_json TEXT NOT NULL,
    distance_km REAL NOT NULL DEFAULT 0,
    updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS app_state (
    key TEXT PRIMARY KEY,
    value_json TEXT NOT NULL,
    updated_at TEXT NOT NULL
);
"""


def _load_legacy() -> dict:
    if not os.path.exists(_STORE):
        return {"cards": {}}
    try:
        with open(_STORE, "r", encoding="utf-8") as f:
            data = json.load(f)
    except (OSError, json.JSONDecodeError):
        return {"cards": {}}
    if isinstance(data, list):
        return {"cards": {"unknown": {"label": _default_label("unknown"), "records": data}}}
    return data if isinstance(data, dict) and isinstance(data.get("cards"), dict) else {"cards": {}}


def _ensure_schema(con: sqlite3.Connection):
    con.executescript(_SCHEMA)
    con.execute("INSERT OR REPLACE INTO meta(key,value) VALUES('schema_version',?)",
                (str(SCHEMA_VERSION),))
    con.commit()


def _migrate_legacy(con: sqlite3.Connection):
    done = con.execute("SELECT value FROM meta WHERE key='legacy_v2_migrated'").fetchone()
    if done:
        return
    data = _load_legacy()
    cards = data.get("cards", {})
    now = _now()
    try:
        for card_id, card in cards.items():
            records = card.get("records", []) or []
            country = next((r.get("country") for r in records if r.get("country")), "")
            con.execute(
                "INSERT OR IGNORE INTO cards(card_id,label,country,created_at,last_seen_at) VALUES(?,?,?,?,?)",
                (card_id, card.get("label") or _default_label(card_id), country, now, now))
            for record in records:
                con.execute(
                    "INSERT OR IGNORE INTO transactions"
                    "(card_id,txn_key,sort_date,log_id,record_json,first_seen_at,last_seen_at) "
                    "VALUES(?,?,?,?,?,?,?)",
                    (card_id, _key_text(record), record.get("date_str", ""),
                     int(record.get("log_id", 0)), json.dumps(record, ensure_ascii=False), now, now))
        con.execute("INSERT OR REPLACE INTO meta(key,value) VALUES('legacy_v2_migrated',?)", (now,))
        con.commit()
        if cards and os.path.exists(_STORE):
            backup = _STORE + ".migrated-v0.3.bak"
            if not os.path.exists(backup):
                shutil.copy2(_STORE, backup)
    except Exception:
        con.rollback()
        raise


def _connect() -> sqlite3.Connection:
    db_path = os.path.abspath(_DB)
    os.makedirs(os.path.dirname(db_path), exist_ok=True)
    con = _open_database(db_path)
    con.row_factory = sqlite3.Row
    con.execute("PRAGMA foreign_keys=ON")
    con.execute("PRAGMA journal_mode=WAL")
    # 过去每个只读查询都会重复执行整份 CREATE TABLE 脚本和迁移检查；在 UI
    # 一次刷新打开十几个连接时，这部分反而成了主要卡顿。每个数据库路径只做一次。
    if db_path not in _SCHEMA_READY:
        with _SCHEMA_LOCK:
            if db_path not in _SCHEMA_READY:
                _ensure_schema(con)
                _migrate_legacy(con)
                _SCHEMA_READY.add(db_path)
    return con


def _records(con: sqlite3.Connection, card_id: str | None = None) -> list[dict]:
    if card_id is None:
        rows = con.execute("SELECT record_json FROM transactions").fetchall()
    else:
        rows = con.execute("SELECT record_json FROM transactions WHERE card_id=?", (card_id,)).fetchall()
    out = []
    for row in rows:
        try:
            out.append(json.loads(row["record_json"]))
        except (TypeError, json.JSONDecodeError):
            pass
    return out


def merge_and_load(idm: str, transactions: list[Transaction],
                   label: str | None = None) -> list[Transaction]:
    merged, _added = merge_and_load_with_stats(idm, transactions, label)
    return merged


def merge_and_load_with_stats(idm: str, transactions: list[Transaction],
                              label: str | None = None) -> tuple[list[Transaction], int]:
    """合并并返回 (累计记录, 本次新增数)，同时永久保存本次解析快照。"""
    card_id = str(idm or "unknown").strip().lower()
    now = _now()
    incoming = {}
    for tx in transactions:
        record = tx.to_dict()
        incoming[_key_text(record)] = record
    country = next((r.get("country") for r in incoming.values() if r.get("country")), "")

    with _LOCK, _connect() as con:
        con.execute(
            "INSERT INTO cards(card_id,label,country,created_at,last_seen_at) VALUES(?,?,?,?,?) "
            "ON CONFLICT(card_id) DO UPDATE SET label=excluded.label, "
            "country=CASE WHEN excluded.country<>'' THEN excluded.country ELSE cards.country END, "
            "last_seen_at=excluded.last_seen_at",
            (card_id, label or _default_label(card_id), country, now, now))
        old_keys = {r[0] for r in con.execute(
            "SELECT txn_key FROM transactions WHERE card_id=?", (card_id,)).fetchall()}
        overrides = {}
        for row in con.execute(
                "SELECT txn_key,field,value FROM station_overrides WHERE card_id=?", (card_id,)):
            overrides.setdefault(row["txn_key"], {})[row["field"]] = row["value"]
        for txn_key, record in incoming.items():
            record.update(overrides.get(txn_key, {}))
            con.execute(
                "INSERT INTO transactions"
                "(card_id,txn_key,sort_date,log_id,record_json,first_seen_at,last_seen_at) "
                "VALUES(?,?,?,?,?,?,?) ON CONFLICT(card_id,txn_key) DO UPDATE SET "
                "sort_date=excluded.sort_date,log_id=excluded.log_id,"
                "record_json=excluded.record_json,last_seen_at=excluded.last_seen_at",
                (card_id, txn_key, record.get("date_str", ""), int(record.get("log_id", 0)),
                 json.dumps(record, ensure_ascii=False), now, now))
        added = len(set(incoming) - old_keys)
        con.execute(
            "INSERT INTO scans(card_id,read_at,record_count,added_count,parser_version,snapshot_json) "
            "VALUES(?,?,?,?,?,?)",
            (card_id, now, len(incoming), added, PARSER_VERSION,
             json.dumps(list(incoming.values()), ensure_ascii=False)))
        merged = _to_tx_list(_records(con, card_id))
    try:
        auto_backup_daily()
    except Exception:
        pass
    return merged, added


def list_cards() -> list[tuple]:
    with _LOCK, _connect() as con:
        rows = con.execute(
            "SELECT c.card_id,c.label,COUNT(t.txn_key) AS n FROM cards c "
            "LEFT JOIN transactions t ON t.card_id=c.card_id "
            "GROUP BY c.card_id,c.label ORDER BY c.last_seen_at DESC").fetchall()
        return [(r["card_id"], r["label"] or _default_label(r["card_id"]), int(r["n"])) for r in rows]


def card_overviews() -> list[dict]:
    """一次连接返回卡片墙/余额汇总所需数据，避免逐卡反复打开 SQLite。"""
    with _LOCK, _connect() as con:
        rows = con.execute(
            "SELECT c.card_id,c.label,c.country,c.last_seen_at,t.record_json "
            "FROM cards c LEFT JOIN transactions t ON t.card_id=c.card_id "
            "ORDER BY c.last_seen_at DESC,t.sort_date DESC,t.log_id DESC").fetchall()
    grouped = {}
    order = []
    for row in rows:
        card_id = row["card_id"]
        if card_id not in grouped:
            grouped[card_id] = {"card_id": card_id,
                                "label": row["label"] or _default_label(card_id),
                                "count": 0, "country": row["country"] or "", "balance": 0}
            order.append(card_id)
        if not row["record_json"]:
            continue
        item = grouped[card_id]; item["count"] += 1
        try:
            record = json.loads(row["record_json"])
        except (TypeError, json.JSONDecodeError):
            continue
        if not item["country"]:
            item["country"] = record.get("country", "") or ""
        if not item["balance"]:
            item["balance"] = int(record.get("balance", 0) or 0)
    return [grouped[card_id] for card_id in order]


def delete_card(idm: str) -> bool:
    with _LOCK, _connect() as con:
        cur = con.execute("DELETE FROM cards WHERE card_id=?", (str(idm).lower(),))
        return cur.rowcount > 0


def clear_all():
    with _LOCK, _connect() as con:
        con.execute("DELETE FROM cards")


def load_card(idm: str) -> list[Transaction]:
    with _LOCK, _connect() as con:
        return _to_tx_list(_records(con, str(idm).lower()))


def load_combined() -> list[Transaction]:
    with _LOCK, _connect() as con:
        return _to_tx_list(_records(con))


def set_label(idm: str, label: str):
    with _LOCK, _connect() as con:
        con.execute("UPDATE cards SET label=? WHERE card_id=?", (label, str(idm).lower()))


def list_scans(card_id: str | None = None, limit: int = 100) -> list[dict]:
    with _LOCK, _connect() as con:
        if card_id:
            rows = con.execute(
                "SELECT * FROM scans WHERE card_id=? ORDER BY read_at DESC LIMIT ?",
                (card_id.lower(), limit)).fetchall()
        else:
            rows = con.execute("SELECT * FROM scans ORDER BY read_at DESC LIMIT ?", (limit,)).fetchall()
        return [dict(r) for r in rows]


def undo_last_scan(card_id: str | None = None) -> dict:
    """撤销最近一次读取中首次出现的记录；此前已有的重复记录不会被删除。"""
    with _LOCK, _connect() as con:
        if card_id:
            scan = con.execute(
                "SELECT * FROM scans WHERE card_id=? ORDER BY scan_id DESC LIMIT 1",
                (card_id.lower(),)).fetchone()
        else:
            scan = con.execute("SELECT * FROM scans ORDER BY scan_id DESC LIMIT 1").fetchone()
        if not scan:
            return {"removed": 0, "scan": None}
        cur = con.execute(
            "DELETE FROM transactions WHERE card_id=? AND first_seen_at=?",
            (scan["card_id"], scan["read_at"]))
        con.execute("DELETE FROM scans WHERE scan_id=?", (scan["scan_id"],))
        return {"removed": max(0, cur.rowcount), "scan": dict(scan)}


def archive_raw_blocks(card_id: str, blocks: list[dict], scan_id: int | None = None) -> int:
    """归档 raw_hex；当前冻结读卡层未暴露原始块，因此接口先就位。"""
    now = _now(); added = 0
    with _LOCK, _connect() as con:
        for block in blocks:
            cur = con.execute(
                "INSERT OR IGNORE INTO raw_blocks"
                "(card_id,scan_id,system_code,service_code,block_no,raw_hex,parser_version,archived_at) "
                "VALUES(?,?,?,?,?,?,?,?)",
                (card_id.lower(), scan_id, block.get("system_code", ""), block.get("service_code", ""),
                 int(block.get("block_no", 0)), block.get("raw_hex", ""),
                 block.get("parser_version", PARSER_VERSION), now))
            added += max(0, cur.rowcount)
    return added


def list_unresolved(limit: int = 500, card_id: str | None = None) -> list[dict]:
    """列出疑似应为行程、但缺少一端站名的数据，供修复工作台使用。"""
    out = []
    with _LOCK, _connect() as con:
        if card_id and card_id != COMBINED:
            rows = con.execute(
                "SELECT card_id,txn_key,record_json FROM transactions WHERE card_id=? "
                "ORDER BY sort_date DESC LIMIT ?", (card_id.lower(), max(limit * 5, limit))).fetchall()
        else:
            rows = con.execute(
                "SELECT card_id,txn_key,record_json FROM transactions ORDER BY sort_date DESC LIMIT ?",
                (max(limit * 5, limit),)).fetchall()
    for row in rows:
        try:
            rec = json.loads(row["record_json"])
        except json.JSONDecodeError:
            continue
        if rec.get("memo"):
            continue
        if not rec.get("in_station") or not rec.get("out_station"):
            out.append({"card_id": row["card_id"], "txn_key": row["txn_key"], "record": rec})
        if len(out) >= limit:
            break
    return out


_FIXABLE_FIELDS = {"in_station", "out_station", "in_company", "out_company", "line", "city", "region"}


def apply_station_fix(card_id: str, txn_key: str, **fields) -> bool:
    clean = {k: str(v or "") for k, v in fields.items() if k in _FIXABLE_FIELDS and v is not None}
    if not clean:
        return False
    now = _now()
    with _LOCK, _connect() as con:
        row = con.execute(
            "SELECT record_json FROM transactions WHERE card_id=? AND txn_key=?",
            (card_id.lower(), txn_key)).fetchone()
        if not row:
            return False
        rec = json.loads(row["record_json"]); rec.update(clean)
        con.execute("UPDATE transactions SET record_json=?,last_seen_at=? WHERE card_id=? AND txn_key=?",
                    (json.dumps(rec, ensure_ascii=False), now, card_id.lower(), txn_key))
        for field, value in clean.items():
            con.execute(
                "INSERT INTO station_overrides(card_id,txn_key,field,value,updated_at) VALUES(?,?,?,?,?) "
                "ON CONFLICT(card_id,txn_key,field) DO UPDATE SET value=excluded.value,updated_at=excluded.updated_at",
                (card_id.lower(), txn_key, field, value, now))
        return True


def route_cache_get(cache_key: str) -> dict | None:
    with _LOCK, _connect() as con:
        row = con.execute("SELECT * FROM route_cache WHERE cache_key=?", (cache_key,)).fetchone()
        if not row:
            return None
        try:
            return {"geometry": json.loads(row["geometry_json"]),
                    "distance_km": float(row["distance_km"]),
                    "source": row["source"], "data_version": row["data_version"]}
        except json.JSONDecodeError:
            return None


def route_cache_get_many(cache_keys) -> dict[str, dict]:
    keys = list(dict.fromkeys(cache_keys))
    if not keys:
        return {}
    out = {}
    with _LOCK, _connect() as con:
        for start in range(0, len(keys), 500):
            batch = keys[start:start + 500]
            marks = ",".join("?" for _ in batch)
            for row in con.execute(f"SELECT * FROM route_cache WHERE cache_key IN ({marks})", batch):
                try:
                    out[row["cache_key"]] = {
                        "geometry": json.loads(row["geometry_json"]),
                        "distance_km": float(row["distance_km"]),
                        "source": row["source"], "data_version": row["data_version"]}
                except json.JSONDecodeError:
                    pass
    return out


def route_cache_set(cache_key: str, geometry, distance_km: float,
                    source: str = "", data_version: str = "v1"):
    with _LOCK, _connect() as con:
        con.execute(
            "INSERT INTO route_cache(cache_key,source,data_version,geometry_json,distance_km,updated_at) "
            "VALUES(?,?,?,?,?,?) ON CONFLICT(cache_key) DO UPDATE SET "
            "source=excluded.source,data_version=excluded.data_version,geometry_json=excluded.geometry_json,"
            "distance_km=excluded.distance_km,updated_at=excluded.updated_at",
            (cache_key, source, data_version, json.dumps(geometry, ensure_ascii=False), distance_km, _now()))


def route_cache_set_many(entries):
    rows = list(entries)
    if not rows:
        return
    now = _now()
    with _LOCK, _connect() as con:
        con.executemany(
            "INSERT INTO route_cache(cache_key,source,data_version,geometry_json,distance_km,updated_at) "
            "VALUES(?,?,?,?,?,?) ON CONFLICT(cache_key) DO UPDATE SET "
            "source=excluded.source,data_version=excluded.data_version,geometry_json=excluded.geometry_json,"
            "distance_km=excluded.distance_km,updated_at=excluded.updated_at",
            [(e["cache_key"], e.get("source", ""), e.get("data_version", "v1"),
              json.dumps(e.get("geometry"), ensure_ascii=False), float(e.get("distance_km", 0)), now)
             for e in rows])


def app_state_get(key: str, default=None):
    with _LOCK, _connect() as con:
        row = con.execute("SELECT value_json FROM app_state WHERE key=?", (key,)).fetchone()
        if not row:
            return default
        try:
            return json.loads(row["value_json"])
        except json.JSONDecodeError:
            return default


def app_state_set(key: str, value):
    with _LOCK, _connect() as con:
        con.execute(
            "INSERT INTO app_state(key,value_json,updated_at) VALUES(?,?,?) "
            "ON CONFLICT(key) DO UPDATE SET value_json=excluded.value_json,updated_at=excluded.updated_at",
            (key, json.dumps(value, ensure_ascii=False), _now()))


def database_stats() -> dict:
    with _LOCK, _connect() as con:
        def count(table):
            return int(con.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0])
        db_size = sum(os.path.getsize(p) for p in (_DB, _DB + "-wal", _DB + "-shm")
                      if os.path.exists(p))
        return {"schema_version": SCHEMA_VERSION, "database": _DB,
                "cards": count("cards"), "transactions": count("transactions"),
                "scans": count("scans"), "raw_blocks": count("raw_blocks"),
                "station_overrides": count("station_overrides"),
                "route_cache": count("route_cache"),
                "size_bytes": db_size}


def create_backup(destination: str) -> str:
    destination = os.path.abspath(destination)
    os.makedirs(os.path.dirname(destination), exist_ok=True)
    with _LOCK, _connect() as src, _open_database(destination) as dst:
        src.backup(dst)
    return destination


def auto_backup_daily(keep: int = 14) -> str:
    backup_dir = os.path.join(os.path.dirname(os.path.abspath(_DB)), "backups")
    os.makedirs(backup_dir, exist_ok=True)
    destination = os.path.join(backup_dir, f"auto_{datetime.date.today():%Y%m%d}.sqlite3")
    if not os.path.exists(destination):
        create_backup(destination)
    backups = sorted(Path(backup_dir).glob("auto_*.sqlite3"), key=lambda p: p.stat().st_mtime,
                     reverse=True)
    for old in backups[max(1, keep):]:
        try:
            old.unlink()
        except OSError:
            pass
    return destination


def restore_backup(source: str) -> str:
    source = os.path.abspath(source)
    if not os.path.exists(source):
        raise FileNotFoundError(source)
    with _open_database(source) as probe:
        tables = {r[0] for r in probe.execute("SELECT name FROM sqlite_master WHERE type='table'")}
        if not {"cards", "transactions", "meta"}.issubset(tables):
            raise ValueError("不是有效的 NekoCardReader v3 备份")
    with _LOCK, _open_database(source) as src, _open_database(_DB) as dst:
        src.backup(dst)
    return _DB


def export_json(destination: str) -> str:
    with _LOCK, _connect() as con:
        cards = {}
        for row in con.execute("SELECT card_id,label FROM cards"):
            cards[row["card_id"]] = {"label": row["label"], "records": _records(con, row["card_id"])}
        payload = {"schema_version": SCHEMA_VERSION, "exported_at": _now(), "cards": cards,
                   "scans": [dict(r) for r in con.execute("SELECT * FROM scans ORDER BY read_at")]}
    with open(destination, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
    return destination


if __name__ == "__main__":
    print(json.dumps(database_stats(), ensure_ascii=False, indent=2))