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 功能索引与日志表。
94 lines
3.9 KiB
Python
94 lines
3.9 KiB
Python
"""系统管理 API（数据备份导出/导入）。仅管理员。
|
|
|
|
导出：POST /api/system/backup/export → 返回 zip 附件下载
|
|
预览：POST /api/system/backup/preview → multipart 上传 zip/db，返回暂存 token + 表清单
|
|
应用：POST /api/system/backup/apply → 校验后落「重启生效」恢复任务
|
|
|
|
导入涉及整库替换，只允许管理员操作（web/auth.admin_required）。
|
|
"""
|
|
from io import BytesIO
|
|
|
|
from flask import Blueprint, jsonify, request, send_file
|
|
from flask_login import current_user
|
|
|
|
from web.auth import admin_required
|
|
from core import system_backup as sb
|
|
from core.logger import get_logger
|
|
|
|
_log = get_logger("web.system")
|
|
|
|
bp = Blueprint("system", __name__)
|
|
|
|
|
|
@bp.route("/api/system/backup/export", methods=["POST"])
|
|
@admin_required
|
|
def api_system_backup_export():
|
|
"""生成导出 zip 并作为附件返回（含 users.db 快照 + manifest.json + 可选 apks）。"""
|
|
data = request.json or {}
|
|
include_apk = bool(data.get("include_apk", True))
|
|
try:
|
|
zip_path, fname, manifest = sb.create_export(include_apk=include_apk)
|
|
except sb.BackupError as e:
|
|
return jsonify({"ok": False, "error": str(e)}), 400
|
|
except Exception as e:
|
|
_log.error(f"导出备份失败: {e}")
|
|
return jsonify({"ok": False, "error": f"导出失败: {e}"}), 500
|
|
|
|
# 读入内存后立即删除磁盘文件再发送——Windows 上 send_file 流式发送时文件
|
|
# 句柄可能到 close 仍未释放，call_on_close 删不掉会残留；改 BytesIO 发送
|
|
# 则可确定性清理（导出为一次性下载，无需保留服务端副本）。
|
|
with open(zip_path, "rb") as f:
|
|
data = f.read()
|
|
sb.remove_quiet(zip_path)
|
|
try:
|
|
from core import notifier
|
|
notifier.notify("system.backup.exported", filename=fname, size=len(data),
|
|
tables=len(manifest.get("tables") or []),
|
|
include_apk=include_apk,
|
|
user=getattr(current_user, "username", ""))
|
|
except Exception:
|
|
pass # 通知失败不影响导出
|
|
resp = send_file(BytesIO(data), as_attachment=True, download_name=fname,
|
|
mimetype="application/zip")
|
|
return resp
|
|
|
|
|
|
@bp.route("/api/system/backup/preview", methods=["POST"])
|
|
@admin_required
|
|
def api_system_backup_preview():
|
|
"""上传备份文件（zip/db）→ 暂存并校验 → 返回预览（表行数/schema/告警）。"""
|
|
f = request.files.get("file")
|
|
if not f or not f.filename:
|
|
return jsonify({"ok": False, "error": "未选择文件"}), 400
|
|
try:
|
|
token, info = sb.stage_upload(f)
|
|
except sb.BackupError as e:
|
|
return jsonify({"ok": False, "error": str(e)}), 400
|
|
except Exception as e:
|
|
_log.error(f"备份预览失败: {e}")
|
|
return jsonify({"ok": False, "error": f"解析文件失败: {e}"}), 500
|
|
_log.info(f"备份预览: {info.get('file_name')} "
|
|
f"(schema v{info.get('schema_version')}) token={token}")
|
|
return jsonify({"ok": True, "token": token, "preview": info})
|
|
|
|
|
|
@bp.route("/api/system/backup/apply", methods=["POST"])
|
|
@admin_required
|
|
def api_system_backup_apply():
|
|
"""确认应用导入：自动备份当前库 → 落待生效恢复任务（重启后生效）。
|
|
|
|
force_env_mismatch 默认 False：备份来源环境与当前库环境不符时拒绝导入，
|
|
要跨环境（如把生产数据导进 dev 库）必须显式传 true。
|
|
"""
|
|
body = request.json or {}
|
|
token = body.get("token", "")
|
|
force = bool(body.get("force_env_mismatch", False))
|
|
try:
|
|
res = sb.apply_restore(str(token).strip(), force_env_mismatch=force)
|
|
except sb.BackupError as e:
|
|
return jsonify({"ok": False, "error": str(e)}), 400
|
|
except Exception as e:
|
|
_log.error(f"应用导入失败: {e}")
|
|
return jsonify({"ok": False, "error": f"应用导入失败: {e}"}), 500
|
|
return jsonify({"ok": True, **res})
|