"""系统数据备份导出/导入（数据库 + APK 文件）。

**归档格式：zip（`users.db` + `manifest.json` + 可选 `apks/*.apk`）。**
SQLite 在这里的角色是**交换格式**，不是运行时数据库——这在迁到 MySQL 之后依然成立，
带来三个好处：
  1. 前端三个接口的请求/响应结构一行都不用改
  2. 校验逻辑（完整性 / 必需表 / schema 版本 / 未登记表反向自检）可以整体沿用
  3. 2026-09-13 之前导出的老备份继续可导入；也不依赖任何外部二进制
     （`sqlite3` 是 Python 标准库，python:slim 容器里现成）

导出：把**当前库**（MySQL 或 SQLite）整库读出来，写进一个临时 SQLite 文件 → zip。
      MySQL 下把这条连接提到 REPEATABLE READ，保证所有表读的是同一时刻。
导入：上传(zip/db) → 暂存校验预览 → 应用（先自动导出一份当前库到
      BACKUP_DIR/pre_restore_*.zip 作安全网，再把暂存归档移到 RESTORE_PENDING_DIR）
      → 下次启动 consume_pending_restore() 在**单个事务内**整库替换生效。

为什么仍然要重启生效：TaskManager / device_pool 把任务与设备**缓存在内存**里，
整库替换后内存副本全部失效。（SQLite 时代还有"Windows 无法替换被持有的文件"这层，
改用 MySQL 之后只剩内存态这一层。）

安全红线：全程只读现有库 / 只写自己目录下的文件；替换数据用单事务，失败回滚，
绝不留下半新半旧的库；绝不 kill-server、绝不 disconnect（与全项目一致）。
"""
import json
import os
import shutil
import sqlite3
import time
import uuid
import zipfile
from urllib.parse import quote

from sqlalchemy import create_engine, inspect, select, text

from config import (DATA_DIR, APK_DIR, BACKUP_DIR,
                    RESTORE_STAGING_DIR, RESTORE_PENDING_DIR)
from core.logger import get_logger
from core.models import CURRENT_SCHEMA_VERSION, db
from core import notifier

_log = get_logger("core.backup")

DB_FILE = os.path.join(DATA_DIR, "users.db")

# 判定「本平台备份库」的必需表（缺任何一张即拒绝导入）
REQUIRED_TABLES = ("app_meta", "user", "task_job", "device_group")
# 备份覆盖清单：**由模型元数据派生**，不再手工维护。
# 这样「新增持久化表必须登记进备份清单」这条红线从"靠人记"变成结构上不可能漏
# （2026-09-10 动作库 agent_action 就漏过：数据其实在快照里，只是清单没列，
#   导出预览里看不到那一行 → 被误判成"没备份"）。
SUMMARY_TABLES = tuple(sorted(t.name for t in db.metadata.tables.values()))
TABLE_LABELS = {
    "app_meta": "系统配置(app_meta)",
    "user": "用户", "device_group": "设备分组", "task_job": "任务计划",
    "custom_action": "自定义动作", "apk_file": "APK 记录", "device": "设备池",
    "pending_device": "待连接设备", "agent_conversation": "AI 会话",
    "agent_experience": "经验库", "experience_audit": "经验巡检",
    "agent_action": "动作库", "device_install_log": "设备端安装记录",
    "task_step_log": "任务步骤明细",
    "done_mark": "去重记录（已做过）",
    "device_account": "账号台账",
}
_STAGE_TTL = 1800  # 导入暂存有效期（秒）


class BackupError(Exception):
    """备份/导入相关可预期错误（msg 直接给前端展示）。"""


# ================== 通用工具 ==================
def _ts():
    return time.strftime("%Y%m%d_%H%M%S")


def _sqlite_uri(path):
    """把路径转成 sqlite URI（兼容含空格/中文/反斜杠的 Windows 路径）。"""
    return "file:{}".format(quote(os.path.abspath(path).replace("\\", "/"), safe="/:"))


def _readonly_uri(path):
    """sqlite 只读 URI。"""
    return _sqlite_uri(path) + "?mode=ro"


def _sqlite_engine_url(path):
    """SQLAlchemy 用的 SQLite URL（注意与 file: URI 不同，engine 不吃那个格式）。"""
    return "sqlite:///" + os.path.abspath(path).replace("\\", "/")


def _connect_readonly(path):
    con = sqlite3.connect(_readonly_uri(path), uri=True)
    con.text_factory = str  # 与平台一致按 UTF-8 读（表内文本均为 UTF-8）
    return con


def remove_quiet(path):
    """尽力删除文件/目录（发送完成清理等场景，忽略不存在）。"""
    try:
        if os.path.isdir(path):
            shutil.rmtree(path, ignore_errors=True)
        elif os.path.exists(path):
            os.remove(path)
    except OSError:
        pass


def _ensure_dir(path):
    os.makedirs(path, exist_ok=True)


def _list_tables(con):
    rows = con.execute(
        "SELECT name FROM sqlite_master WHERE type='table'").fetchall()
    return {r[0] for r in rows}


def _read_schema_version(con):
    try:
        row = con.execute(
            "SELECT value FROM app_meta WHERE key='schema_version'").fetchone()
        return int(row[0]) if row and row[0] else 0
    except Exception:
        return 0


def _table_rows(con, table):
    try:
        return con.execute('SELECT COUNT(*) FROM "%s"' % table).fetchone()[0]
    except Exception:
        return None


def _summary_info(con):
    """表行数摘要（只列出存在的表）。"""
    rows = []
    for t in SUMMARY_TABLES:
        n = _table_rows(con, t)
        if n is not None:
            rows.append({"table": t, "label": TABLE_LABELS.get(t, t), "rows": n})
    return rows


def _prune_old_exports(max_age=3600):
    """清理过期的导出临时 zip（下载完成后的 call_on_close 在 Windows 上可能
    因文件锁删不掉，这里按时间兜底清理；保留近 1 小时的便于失败重试）。"""
    now = time.time()
    if not os.path.isdir(BACKUP_DIR):
        return
    for n in os.listdir(BACKUP_DIR):
        if n.startswith("export_") and n.endswith(".zip"):
            p = os.path.join(BACKUP_DIR, n)
            try:
                if now - os.path.getmtime(p) > max_age:
                    os.remove(p)
            except OSError:
                pass


def _prune_stale_staging():
    """清理超时未应用的暂存目录（TTL 后自动删除）。"""
    now = time.time()
    if not os.path.isdir(RESTORE_STAGING_DIR):
        return
    for name in os.listdir(RESTORE_STAGING_DIR):
        p = os.path.join(RESTORE_STAGING_DIR, name)
        try:
            if os.path.isdir(p) and now - os.path.getmtime(p) > _STAGE_TTL:
                remove_quiet(p)
        except OSError:
            pass


def _current_env_label():
    """当前库登记的环境标签（dev/prod）与库唯一标识（未登记则空）。"""
    from core import db_config
    try:
        return db_config.meta_get("deployment_env") or "", db_config.meta_get("deployment_id") or ""
    except Exception:
        return "", ""


# ================== 导出：当前库 → SQLite 归档 ==================
def dump_to_sqlite(dest_path):
    """把当前数据库整库导出成一个 SQLite 文件（备份归档格式）。

    一致性：MySQL 下把这条连接提到 REPEATABLE READ，让所有表读同一个时间点
    （应用全局是 READ COMMITTED，事务提交后自增/被改的行会"漂移"）。
    SQLite 下不需要——单文件库天然一致。
    """
    _ensure_dir(os.path.dirname(os.path.abspath(dest_path)))
    remove_quiet(dest_path)
    remove_quiet(dest_path + "-wal")
    remove_quiet(dest_path + "-shm")

    out_eng = create_engine(_sqlite_engine_url(dest_path))
    is_mysql = db.engine.dialect.name == "mysql"
    conn = db.engine.connect()
    if is_mysql:
        conn = conn.execution_options(isolation_level="REPEATABLE READ")

    dumped = []
    try:
        # 归档库用同一套模型建表（mysql_* 表选项在 SQLite 下被忽略）
        db.metadata.create_all(out_eng)
        src_cols = {}
        insp = inspect(conn)
        for table in db.metadata.sorted_tables:
            try:
                src_cols[table.name] = {c["name"] for c in insp.get_columns(table.name)}
            except Exception:
                src_cols[table.name] = set()

        with out_eng.begin() as out:
            for table in db.metadata.sorted_tables:
                have = src_cols.get(table.name) or set()
                cols = [c for c in table.columns if c.name in have]
                if not cols:
                    _log.warning(f"导出跳过 {table.name}：源库没有这张表")
                    continue
                try:
                    rows = [dict(r._mapping)
                            for r in conn.execute(select(*cols))]
                except Exception as e:
                    _log.error(f"导出读取 {table.name} 失败: {e}")
                    raise BackupError(f"导出失败：读取 {table.name} 出错（{e}）")
                if rows:
                    out.execute(table.insert(), rows)
                dumped.append((table.name, len(rows)))
    finally:
        conn.close()
        out_eng.dispose()

    # 归档库收尾：落回单文件 + 补"空值不参与唯一"的唯一索引。
    # SQLAlchemy 建的 sqlite 连接会被 PRAGMA 监听器设成 WAL，留下 -wal/-shm；
    # zip 里只打包主文件，所以必须先 checkpoint 回 DELETE 模式再删附属文件。
    try:
        con = sqlite3.connect(dest_path)
        try:
            con.execute("PRAGMA journal_mode=DELETE")
        except sqlite3.Error as e:
            _log.warning(f"归档库切回单文件模式失败: {e}")
        for name, sql in (("ux_device_name",
                           "CREATE UNIQUE INDEX IF NOT EXISTS ux_device_name "
                           "ON device(name) WHERE name IS NOT NULL AND name <> ''"),
                          ("ux_device_fingerprint",
                           "CREATE UNIQUE INDEX IF NOT EXISTS ux_device_fingerprint "
                           "ON device(fingerprint) WHERE fingerprint IS NOT NULL "
                           "AND fingerprint <> ''")):
            try:
                con.execute(sql)
            except Exception as e:
                _log.warning(f"归档库建索引 {name} 失败（数据可能有重复）: {e}")
        con.commit()
        con.close()
    except Exception as e:
        _log.warning(f"归档库收尾失败（不影响导出）: {e}")
    remove_quiet(dest_path + "-wal")
    remove_quiet(dest_path + "-shm")

    _log.info("已导出到 SQLite 归档: %s（%d 张表）",
              os.path.basename(dest_path), len(dumped))
    return dumped


# 兼容旧名（历史代码/脚本里叫 snapshot_db）
snapshot_db = dump_to_sqlite


def create_export(include_apk=True):
    """生成导出 zip，返回 (zip_path, filename, manifest)。"""
    _ensure_dir(BACKUP_DIR)
    _prune_old_exports()
    base = os.path.join(BACKUP_DIR, f"export_{_ts()}")
    snap = base + ".snapshot.db"
    dump_to_sqlite(snap)

    apk_meta = []
    if include_apk and os.path.isdir(APK_DIR):
        for fname in sorted(os.listdir(APK_DIR)):
            if fname.lower().endswith(".apk"):
                fp = os.path.join(APK_DIR, fname)
                try:
                    apk_meta.append({"name": fname, "size": os.path.getsize(fp)})
                except OSError:
                    pass

    env_label, db_id = _current_env_label()
    zip_path = base + ".zip"
    con = _connect_readonly(snap)
    try:
        manifest = {
            "format": "auto_control_backup",
            "version": 2,
            "created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
            "schema_version": _read_schema_version(con),
            "db_backend": db.engine.dialect.name,   # mysql / sqlite
            "deployment_env": env_label,            # 来源环境（跨环境导入时告警）
            "source_db_id": db_id,                  # 来源库唯一标识
            "include_apk": include_apk,
            "tables": _summary_info(con),
            "apks": apk_meta,
        }
        # 覆盖自检：登记在册的业务表若在快照里缺失（新增功能忘了登记 / 建表失败），
        # 显式告警并写进 manifest——避免"以为备份了其实没有"（备份覆盖红线）。
        _present = {t["table"] for t in manifest["tables"]}
        _missing = [t for t in SUMMARY_TABLES if t not in _present]
        if _missing:
            manifest["coverage_missing"] = _missing
            _log.warning(f"备份覆盖检查：以下登记表未纳入快照，请确认是否应备份: {_missing}")
    finally:
        con.close()

    try:
        with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as z:
            z.write(snap, arcname="users.db")
            for a in apk_meta:
                z.write(os.path.join(APK_DIR, a["name"]), arcname="apks/" + a["name"])
            z.writestr("manifest.json", json.dumps(
                manifest, ensure_ascii=False, indent=2))
    finally:
        remove_quiet(snap)
    return zip_path, os.path.basename(zip_path), manifest


# ================== 校验 ==================
def validate_backup(db_file):
    """校验备份库可读、是本平台库。返回 (ok, info|error_msg)。"""
    if not os.path.exists(db_file):
        return False, "缺少数据库文件 users.db"
    try:
        con = _connect_readonly(db_file)
    except sqlite3.Error as e:
        return False, f"无法打开数据库: {e}"
    try:
        integrity = con.execute("PRAGMA integrity_check").fetchone()[0]
        if integrity != "ok":
            return False, f"数据库完整性校验失败: {integrity}"
        tables = _list_tables(con)
        missing = [t for t in REQUIRED_TABLES if t not in tables]
        if missing:
            return False, f"不是本平台备份库（缺少必需表: {', '.join(missing)}）"
        schema_version = _read_schema_version(con)
        missing_optional = [t for t in SUMMARY_TABLES
                            if t not in REQUIRED_TABLES and t not in tables]
        warnings = []
        if schema_version < CURRENT_SCHEMA_VERSION:
            warnings.append(
                f"备份 schema 较旧（v{schema_version} < 当前 v{CURRENT_SCHEMA_VERSION}），"
                f"缺失的表/列按当前结构补空，不会报错但那些数据回不来")
        elif schema_version > CURRENT_SCHEMA_VERSION:
            warnings.append(
                f"备份 schema 较新（v{schema_version} > 当前 v{CURRENT_SCHEMA_VERSION}），"
                f"当前平台版本可能读不了新增列/表，建议先升级再恢复")
        if missing_optional:
            warnings.append("备份缺少部分可选表（" + "、".join(
                TABLE_LABELS.get(t, t) for t in missing_optional)
                + "），导入后这些表为空")
        # 反向自检：备份里有"未登记"的表 → 提醒把它纳入覆盖清单（红线）
        extra_tables = sorted(t for t in tables
                              if t not in SUMMARY_TABLES and not t.startswith("sqlite_"))
        if extra_tables:
            warnings.append("备份含未登记的其它表（" + "、".join(extra_tables)
                            + "）——如属业务数据，请登记进 core/models.py 的模型与 "
                              "core/system_backup.py 的 TABLE_LABELS")
        warnings.append("备份为全量数据：含用户口令哈希、AI 配置里的 API Key 等敏感信息，请妥善保管")
        return True, {
            "integrity": integrity,
            "schema_version": schema_version,
            "current_schema_version": CURRENT_SCHEMA_VERSION,
            "tables": _summary_info(con),
            "missing_optional": missing_optional,
            "extra_tables": extra_tables,
            "warnings": warnings,
        }
    except sqlite3.Error as e:
        return False, f"读取数据库失败: {e}"
    finally:
        con.close()


# ================== 导入：暂存 + 预览 ==================
def _extract_zip(zip_path, staging_dir):
    """解压备份 zip：取顶层 users.db、可选 apks/*.apk 与 manifest.json（防路径穿越）。"""
    db_dest = os.path.join(staging_dir, "users.db")
    apk_dest = os.path.join(staging_dir, "apks")
    got_db = False
    with zipfile.ZipFile(zip_path) as z:
        for info in z.infolist():
            if info.is_dir():
                continue
            name = info.filename.replace("\\", "/")
            base = os.path.basename(name)
            if name == "users.db" and base == "users.db":
                z.extract(info, staging_dir)  # 已确认顶层名字，无穿越
                got_db = True
            elif name == "manifest.json":
                with z.open(info) as src, open(
                        os.path.join(staging_dir, "manifest.json"), "wb") as dst:
                    shutil.copyfileobj(src, dst)
            elif name.startswith("apks/") and base.lower().endswith(".apk"):
                _ensure_dir(apk_dest)
                with z.open(info) as src, open(
                        os.path.join(apk_dest, base), "wb") as dst:
                    shutil.copyfileobj(src, dst)
    if not got_db:
        raise BackupError("zip 内未找到 users.db（顶层）")
    return db_dest


def _read_staged_manifest(staging_dir):
    p = os.path.join(staging_dir, "manifest.json")
    if not os.path.exists(p):
        return {}
    try:
        with open(p, encoding="utf-8") as f:
            return json.load(f) or {}
    except Exception:
        return {}


def stage_upload(file_storage):
    """保存上传备份并校验，返回 (token, info)。失败抛 BackupError。"""
    _prune_stale_staging()
    orig_name = file_storage.filename or "backup"
    token = uuid.uuid4().hex[:12]
    staging_dir = os.path.join(RESTORE_STAGING_DIR, token)
    _ensure_dir(staging_dir)
    low = (orig_name or "").lower()
    try:
        raw_path = os.path.join(staging_dir, "upload" + ("." + low.rsplit(".", 1)[1] if "." in low else ""))
        file_storage.save(raw_path)
        if low.endswith(".zip"):
            _extract_zip(raw_path, staging_dir)
            db_file = os.path.join(staging_dir, "users.db")
            remove_quiet(raw_path)
        elif low.endswith(".db"):
            db_file = os.path.join(staging_dir, "users.db")
            os.replace(raw_path, db_file)
        else:
            raise BackupError("仅支持 .zip（平台导出）或 .db（备份库文件）")
        ok, info = validate_backup(db_file)
        if not ok:
            raise BackupError(info)
        info["file_name"] = orig_name
        info["file_size"] = os.path.getsize(db_file)

        # 跨环境提示（防混库第 5 层）：备份来自哪个环境、当前连的是哪个环境
        man = _read_staged_manifest(staging_dir)
        info["source_env"] = man.get("deployment_env") or ""
        info["source_db_id"] = man.get("source_db_id") or ""
        info["source_backend"] = man.get("db_backend") or "sqlite(旧版备份)"
        cur_env, _ = _current_env_label()
        info["current_env"] = cur_env
        if info["source_env"] and cur_env and info["source_env"] != cur_env:
            info["warnings"].append(
                f"该备份来自 {info['source_env']} 环境，而当前连接的是 {cur_env} 库 —— "
                f"导入会覆盖当前全部数据。确认这是你要的操作再继续（跨环境导入默认被拒绝）。")
        # 暂存目录里留住 manifest，apply 时还要用它做环境校验
        return token, info
    except BackupError:
        remove_quiet(staging_dir)
        raise
    except Exception as e:
        remove_quiet(staging_dir)
        raise BackupError(f"暂存上传文件失败: {e}")


# ================== 导入：应用（落待生效任务） ==================
def apply_restore(token, force_env_mismatch=False):
    """校验 token → 自动备份当前库（zip 安全网）→ 把暂存归档移到 restore_pending。

    返回 dict {ok, backup_name, message}；失败抛 BackupError。token 形如 12 位 hex。
    """
    if not token or len(token) != 12 or not all(c in "0123456789abcdef" for c in token):
        raise BackupError("无效的导入标识")
    staging_dir = os.path.join(RESTORE_STAGING_DIR, token)
    db_file = os.path.join(staging_dir, "users.db")
    if not os.path.exists(db_file):
        raise BackupError("导入会话已失效，请重新上传预览")
    ok, info = validate_backup(db_file)
    if not ok:
        raise BackupError(info)

    # 跨环境硬停：默认不许把 prod 的备份灌进 dev 库（反之亦然）
    man = _read_staged_manifest(staging_dir)
    src_env = man.get("deployment_env") or ""
    cur_env, _ = _current_env_label()
    if src_env and cur_env and src_env != cur_env and not force_env_mismatch:
        raise BackupError(
            f"备份来自 {src_env} 环境，当前库是 {cur_env} 环境 —— 已拒绝导入。\n"
            f"如确需跨环境恢复，请在确认页勾选「允许跨环境导入」后重试。")

    # 1) 自动备份当前库（安全网，可回滚；导出成 zip，能直接再导入回来）
    _ensure_dir(BACKUP_DIR)
    try:
        zip_path, _, _ = create_export(include_apk=True)
        backup_name = f"pre_restore_{_ts()}.zip"
        shutil.move(zip_path, os.path.join(BACKUP_DIR, backup_name))
    except Exception as e:
        raise BackupError(f"导入前自动备份当前库失败，已中止导入: {e}")

    # 2) 落待生效恢复任务（旧未消费任务被覆盖——上一份已无意义）
    if os.path.isdir(RESTORE_PENDING_DIR):
        remove_quiet(RESTORE_PENDING_DIR)
    _ensure_dir(RESTORE_PENDING_DIR)
    os.replace(db_file, os.path.join(RESTORE_PENDING_DIR, "users.db"))
    for extra in ("manifest.json",):
        src = os.path.join(staging_dir, extra)
        if os.path.exists(src):
            shutil.move(src, os.path.join(RESTORE_PENDING_DIR, extra))
    apk_src = os.path.join(staging_dir, "apks")
    if os.path.isdir(apk_src) and os.listdir(apk_src):
        shutil.move(apk_src, os.path.join(RESTORE_PENDING_DIR, "apks"))
    remove_quiet(staging_dir)

    schema_version = info.get("schema_version", 0)
    _log.info(f"导入已落恢复任务: schema v{schema_version}，"
              f"重启 web_server 后生效；当前库已自动备份 {backup_name}")
    return {
        "ok": True,
        "backup_name": backup_name,
        "message": "恢复任务已生成：当前库已自动备份，重启 web_server 后即应用导入的数据",
    }


# ================== 启动消费（init_db 之后、TaskManager 之前调用）==================
def _pending_db_path():
    """待生效恢复的归档路径（兼容旧版扁平布局）。"""
    p = os.path.join(RESTORE_PENDING_DIR, "users.db")
    return p if os.path.exists(p) else None


def consume_pending_restore():
    """把 restore_pending/users.db 的内容在**单个事务内**整库替换进来。

    与 SQLite 时代的"换文件"不同：现在是 DELETE 全表 + 分块 INSERT，全程一个事务——
    任何一步失败就 rollback，当前数据保持原样（比换文件更安全）。

    必须在 init_db() 之后调用（表已建好、app context 可用）；失败不阻塞启动：
    坏归档挪到 BACKUP_DIR/restore_failed_*/ 并继续用当前数据。
    返回是否真的应用了恢复。
    """
    pending_db = _pending_db_path()
    if not pending_db:
        return False
    # 消费前轻量复验（防止意外损坏的文件把主库换掉）
    ok, msg = validate_backup(pending_db)
    if not ok:
        fail_dir = os.path.join(BACKUP_DIR, f"restore_failed_{_ts()}")
        _ensure_dir(fail_dir)
        shutil.move(pending_db, os.path.join(fail_dir, "users.db"))
        remove_quiet(RESTORE_PENDING_DIR)
        _log.error(f"恢复任务校验失败，已搁置到 {fail_dir}，继续使用当前库: {msg}")
        notifier.notify("system.backup.restore_failed", error=str(msg)[:200],
                        fail_dir=fail_dir)
        return False

    try:
        applied = _apply_archive_to_db(pending_db)
    except Exception as e:
        _log.error(f"应用备份恢复失败，已回滚（当前数据未改动）: {e}，"
                   f"归档保留在 {RESTORE_PENDING_DIR} 供排查")
        return False

    # 合并 apks（覆盖同名）
    apk_src = os.path.join(RESTORE_PENDING_DIR, "apks")
    if os.path.isdir(apk_src):
        _ensure_dir(APK_DIR)
        for fname in os.listdir(apk_src):
            src = os.path.join(apk_src, fname)
            dst = os.path.join(APK_DIR, fname)
            if os.path.exists(dst):
                os.remove(dst)
            shutil.move(src, dst)
    remove_quiet(RESTORE_PENDING_DIR)
    schema_version = msg.get("schema_version", "?") if isinstance(msg, dict) else "?"
    _log.info(f"已应用备份恢复（单事务替换）: {applied}，schema v{schema_version}，apks 已合并")
    # 通知在"下一次启动"才发得出去（恢复本身是重启生效的），文案里说明白
    notifier.notify("system.backup.restored", applied_rows=applied,
                    schema_version=schema_version)
    return True


def _apply_archive_to_db(db_file, chunk=500):
    """把 SQLite 归档里的数据整库写进当前数据库，单事务、失败回滚。

    用 DELETE 而不是 TRUNCATE：TRUNCATE 是 DDL，会隐式提交，破坏原子性。
    """
    con = _connect_readonly(db_file)
    arch_tables = _list_tables(con)
    stats = []
    try:
        # 单事务：MySQL/SQLite 都支持事务内 DELETE + INSERT
        with db.engine.begin() as dst:
            for table in reversed(db.metadata.sorted_tables):
                dst.execute(table.delete())
            for table in db.metadata.sorted_tables:
                if table.name not in arch_tables:
                    _log.warning(f"归档里没有 {table.name}，该表导入后为空")
                    continue
                have = {r[1] for r in con.execute(
                    'PRAGMA table_info("%s")' % table.name)}
                cols = [c for c in table.columns if c.name in have]
                if not cols:
                    _log.warning(f"归档表 {table.name} 没有可用列，跳过")
                    continue
                sel = "SELECT {} FROM \"{}\"".format(
                    ", ".join('"%s"' % c.name for c in cols), table.name)
                rows = con.execute(sel).fetchall()
                names = [c.name for c in cols]
                for i in range(0, len(rows), chunk):
                    batch = [dict(zip(names, r)) for r in rows[i:i + chunk]]
                    dst.execute(table.insert(), batch)
                stats.append(f"{table.name}={len(rows)}")
    finally:
        con.close()
    return " ".join(stats)