butubb
97cec6e211
平台此前出了问题只能靠人盯页面。现在各组件统一走 `notifier.notify(事件, **字段)`， 推到企业微信 / 自建服务；**所有可通知点都登记进事件目录，默认全关，用户按 webhook 勾选**。 ## 架构（`core/notifier.py` + `core/notify_events.py`） 业务线程调 `notify()` → 只做内存操作（读配置快照/匹配订阅/入队）→ 返回； 后台 1 个 dispatcher（聚合 + 每 hook 限流 + 折叠摘要）+ 3 个 sender（真实 HTTP、退避重试） 负责真正发出去。硬约束：**notify 零 DB、零 HTTP、零阻塞、异常不冒泡**——所以任务线程里 可以直接调（不用 app_context、不用 try/except），但**必须放在所有 `with self._lock` 之外**。 - **事件目录 36 条**（任务批次/单设备/设备/Worker/业务/安装/系统/AI），支持 `task.*` 通配订阅； 语义分工避免重复告警：`worker.*` 是单次尝试级，`task.device.success/failed` 是唯一权威结论。 - **适配器可插拔**：`wecom`（markdown，按 4096 **字节**截断、超限不截半个汉字）+ `json`（模板占位符，替换值按 JSON 转义，保存前干跑校验）；钉钉/飞书留了插槽（前端置灰）。 - **防打爆四层**：聚合窗口（默认 30s，同批次合并成一条并带样本）→ 令牌桶限流（默认 18/分， 对齐企微硬限 20）→ 被限流的**折叠成摘要不丢弃** → 有界队列背压。取舍：失败通知最多延迟 一个窗口，换来群不被刷屏。 ## 安全与存储 - 配置只落 `app_meta.notify_webhooks` 一个键（**不建表** → 不涉及备份覆盖红线）。 - URL 本身就是凭据（企微 `?key=`）→ 接口回显/发送记录/日志一律 `mask_url()/scrub()`； 编辑时留空即不修改；secret 永不回显。DATA_MODEL 的明文凭据告警补上了这一条。 - 发送记录：内存环形缓冲 200 条（重启清空）+ 独立 `logs/notify.log`。 ## 接入点（每个都放在锁外、不改 return 顺序） task_manager（批次开始/结束用新增的 `_BatchTracker` 统一在 finally 计数、单设备成功/失败/ 离线/重试/停止/抢占/归还/cron 停止）、device_worker 心跳看门狗、generic 任务选择器连续失效、 apk 安装开始/完成、设备上下线（**状态沿检测**，只报新变化）、备份导出/恢复、经验巡检、 用户登录、服务启停。 ## 前端 系统 Tab 新增「通知 / Webhook」子分栏：多条 webhook 列表（URL 打码）+ 编辑弹窗（格式/URL/ 密钥/事件勾选树带 ★建议/聚合/限流/自定义模板/预览）+ 发送测试 + 发送记录。 ## 自测 - 进程内逻辑 10 组断言全绿：聚合合并、限流+折叠、无配置/全局关静默丢弃、未知事件、 内部异常不外泄、JSON 转义（标题含引号换行仍合法）、URL/异常消息脱敏、配置校验。 - 端到端（假 webhook 接收端）17 项断言全绿：真实事件投递（user.login / task.batch.no_device）、 企微请求体形状、**HTTP 200 + errcode 93000 判为失败**、500 重试 3 次、记录里 URL 打码。 - 韧性：webhook 指向黑洞地址时登录耗时 100~114ms（基线 107~133ms，**异步隔离生效**）； 配置写成坏 JSON 服务照常启动、通知静默不发、日志有 error（服务端实测后已复原）。 - 页面：系统 → 通知 面板/弹窗/36 个事件复选框/预览全部正常，无 JS 报错。 - 自测数据已清理（webhook、自建任务、写坏又复原的配置键）。 文档：新增 doc/NOTIFY.md（事件表/配置/格式约束/防刷屏/加事件三步骤/排障）并登记进 doc/README； API.md §2.11；DATA_MODEL 的 app_meta 键表与明文凭据告警；ARCHITECTURE 线程表/分层/扩展点； 根 README 功能索引与日志表。
615 lines
26 KiB
Python
615 lines
26 KiB
Python
"""系统数据备份导出/导入（数据库 + 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": "设备端安装记录",
|
||
}
|
||
_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)
|