# 架构详解（ARCHITECTURE） > 适用读者：要改后端 / 前端 / 任务引擎的开发者。 > 相关文档：[DATA_MODEL.md](DATA_MODEL.md)（表结构）、[API.md](API.md)（接口清单）、[DEVELOPMENT.md](DEVELOPMENT.md)（流程与红线）。 > 文中引用为 `文件:行号`，以当前代码为准；**行号会随改动漂移，函数名与常量名是稳定锚点**。 --- ## 1. 总览 ### 1.1 分层 ``` ┌──────────────────────────────────────────────────────────────────────┐ │ 表现层 templates/admin/*.html + static/admin/*.js（单页应用，无框架） │ │ 7 个顶级 Tab；轮询 / SSE / MJPEG 三类实时通道 │ └───────────────────────────┬──────────────────────────────────────────┘ │ fetch JSON / SSE / MJPEG / 表单 ┌───────────────────────────▼──────────────────────────────────────────┐ │ Web 层 web/（10 个 Flask 蓝图，全部 url_prefix 为空） │ │ auth 鉴权 · monitor 状态与设备操作 · tasks 任务 · admin 用户与日志 │ │ tools 运维 · devices 设备池 · apks 应用 · tailscale · agent AI 控制台 │ │ system 备份 │ └───────────────────────────┬──────────────────────────────────────────┘ │ 直接函数调用（共享对象由 web/context.py 注入） ┌───────────────────────────▼──────────────────────────────────────────┐ │ 领域层 task_manager（调度） device_worker（执行） device_pool（池） │ │ system_backup（备份） apk_manager（应用） device_discovery │ └───────────────────────────┬──────────────────────────────────────────┘ │ ┌───────────────────────────▼──────────────────────────────────────────┐ │ 基础层 adb_helper · u2_helper · uiauto_helper · ocr · clipboard │ │ notifier（通知分发：队列/聚合/限流/适配器） │ │ step_log（步骤明细：队列 + 批量落库 + 保留期清理） │ │ models（SQLite）· logger · config │ └──────────────────────────────────────────────────────────────────────┘ ▲ ┌───────────────────────────┴──────────────────────────────────────────┐ │ 任务定义层 tasks/（BaseTask 注册表 + generic/ 通用步骤引擎） │ └───────────────────────────┬──────────────────────────────────────────┘ ▲ ▲ ┌───────────────────────────┴────────┐ ┌───────────┴──────────────────┐ │ MCP Server（mcp_server/，:8033） │ │ AI Agent（mcp_agent/） │ │ 20 个 de_* 工具，供外部 AI 调用 │ │ OpenAI 兼容模型 → MCP 工具 │ └────────────────────────────────────┘ └──────────────────────────────┘ ``` ### 1.2 各层职责边界 | 层 | 做什么 | 不做什么 | |----|--------|---------| | 表现层 | 渲染、交互、轮询/流式拉取、按权限隐藏入口 | 不校验权限（只隐藏）；不做业务判断 | | Web 层 | 参数校验、鉴权装饰器、JSON 序列化、调用领域层 | 不直接操作 adb/u2（`monitor` 的看屏/截图除外，那本身就是"设备操作"） | | 领域层 | 调度、并发、重试、状态机、持久化 | 不感知 HTTP | | 基础层 | adb / u2 / OCR / 数据库 / 日志的原子能力 | 不含业务规则 | | 任务定义层 | 任务类型注册 + 具体任务执行逻辑 | 不感知调度与设备获取（`BaseWorker` 已封装） | **装配方向**：`web_server.py` 是唯一组装点；`web/context.py` 注入 `mgr` / `apk_mgr` / `device_pool`，避免 Web 层与领域层循环 import。 --- ## 2. 启动与装配顺序 理解启动顺序很关键——**很多副作用发生在 import 期**。 ### 2.1 阶段 A：import 期副作用（`web_server.py:6-25`） | 顺序 | 触发 | 副作用 | |------|------|--------| | 1 | `from core.task_manager import TaskManager` | 链式触发：`config.py` 模块体 → **读取根目录 `.env`**（`os.environ.setdefault`）；`core/logger.py` → 创建 `logs/`；`tasks/__init__.py` → **任务类型注册**（`@register_task` 在 import 期执行） | | 2 | `get_logger` / `core.models` | 得到 `db` / `init_db` / `User` | | 3 | `from config import WEB_HOST, WEB_PORT` | 读取端口常量 | > `.env` 是"import config 的副作用"，因此 `web_server.py:30` 读 `WEB_SECRET_KEY` 时它已生效。 ### 2.2 阶段 B~D：Flask 与领域对象 | 阶段 | 位置 | 做了什么 | |------|------|---------| | **B** | `:28-58` | `Flask(__name__)`；会话密钥（`.env` 的 `WEB_SECRET_KEY`，缺失则随机生成并 warning）；`TEMPLATES_AUTO_RELOAD=True`；**数据库目标由 `core/db_config` 装配**（`.env` 的 `DEPLOY_ENV`/`DB_*` → URI + 引擎参数），配置错直接 `SystemExit(2)`；`LoginManager` + `login_view="auth.login"` | | **C** | `:60-67` | **恢复任务消费** `consume_pending_restore()`。SQLite 时代它必须在 engine 首次打开 `users.db` **之前**（Windows 无法替换被持有的文件）；改用 MySQL 后这一步的语义会变成"启动期事务替换"，见 §7 | | **D** | `:69-88` | `init_db(app)`（建表 → 补列 → 版本账本 → 唯一索引 → 默认管理员 → 旧 JSON 迁移）→ `notifier.init_app`（通知 dispatcher/sender 线程）与 `step_log.init_app`（步骤明细写线程）→ **恢复任务消费** `consume_pending_restore()` → **库环境标签校验 + 启动横幅**（`db_config.verify_deployment_label/print_banner`，不符拒绝启动）；`device_pool.init_app`（**起一次性线程**，3s 后采集型号）；`device_discovery.init_app`（**起常驻扫描线程**）；`TaskManager(app=app)`（APScheduler + 看门狗 + 从库加载分组/任务 + 重注册 cron）；`ApkManager(app=app)` | ### 2.3 阶段 E~G：蓝图、巡检调度器、真正启动 | 阶段 | 位置 | 做了什么 | |------|------|---------| | **E** | `:64-67` | `context.init(...)`；`register_blueprints(app)`（10 个蓝图）；`agent_api.set_app(app)`（供后台线程推 app context） | | **F** | 同上附近 | **第二个独立 APScheduler**：`CronTrigger(hour=3, minute=47)` 挂经验库巡检、`hour=4, minute=13` 挂步骤明细清理（`_purge_step_log`，自建 app context）；失败仅 warning | | **G** | `__main__` | `_ensure_uiauto_running()`（拉起 uiautodev:20242，写 `data/uiauto.pid`，`atexit` 清理）→ `_preconnect_pool_devices()`（后台并发 connect 池内网络设备）→ `_purge_step_log_async()`（后台清理超期步骤明细）→ `_run_server()`（候选端口依次 bind：`0.0.0.0:18050` → `127.0.0.1:18050` → `127.0.0.1:18051..18055`）；退出时 `notifier.shutdown()` + `step_log.shutdown()` + `mgr.shutdown()` + `device_discovery.shutdown()` + 停 uiautodev | > ⚠️ **阶段 A~F 在 import 期就会起线程/调度器**，只有 uiautodev 拉起与预连接在 `__main__` 分支。以 WSGI 方式 import 本模块会得到"半个启动"的进程——本地调试请直接 `python web_server.py`。 --- ## 3. 线程与并发模型 ### 3.1 常驻线程一览 | 名称 | 启动位置 | 职责 | 周期 | |------|---------|------|------| | Flask 请求线程 | `app.run(threaded=True)` | 每请求一线程 | — | | **任务调度器** `BackgroundScheduler` | `TaskManager.__init__` | cron 触发 / 停止任务 | 按 cron | | **巡检调度器** `BackgroundScheduler` | `web_server.py` | 经验库 AI 巡检 | 每日 03:47 | | `worker-watchdog` | `start_watchdog()` | `running/connecting` 心跳超时 → 标 `error` | 30s 检查 / 120s 阈值 | | `device-discovery` | `device_discovery.init_app` | 网段扫描 + 断联重连 | 首轮延迟 15s，之后 interval（默认 60s） | | `_refresh_models_bg` | `device_pool.init_app` | 启动后采集全部在线设备型号 | 一次性（3s 后） | | `BaseWorker` × N | `TaskManager._run_with_retry` | 单设备任务执行 | 任务期 | | `_run_with_retry` × N | 同上 | 单设备重试循环 | 任务期 | | `fg-scan-once` | `_ForegroundScanner.scan_once` | 前台 App 扫描 | 手动触发，`Event` 防重入 | | `apk-install` | `ApkManager.install` | 并发 5 台安装 APK | 安装期，全局单任务 | | Agent 执行线程 | `web/agent_api.py` | AI 控制台一轮会话 | 按需 | | 巡检手动线程 | `web/agent_api.py` | 手动触发巡检 | 按需 | | **通知 dispatcher** | `core/notifier.init_app` | 通知聚合 + 每 hook 限流 + 折叠摘要 | 常驻 1 个 | | **通知 sender ×3** | 同上 | 真实发 webhook（退避重试、环形记录） | 常驻 3 个 | | **步骤明细写线程** | `core/step_log.init_app` | 批量落库 `task_step_log`（队列满丢弃并计数） | 常驻 1 个 | 两个 APScheduler 相互独立，时区均固定 `Asia/Shanghai`。 ### 3.2 锁与并发保护 | 锁 | 位置 | 保护对象 | 说明 | |----|------|---------|------| | `_ADB_LOCK` | `core/adb_helper.py` | adb connect 串行化 | connect 很快，串行不影响整体并发；不覆盖长命令 | | `_WORKERS_LOCK` | `core/device_worker.py` | 全局设备状态表 `_WORKERS` | "检查+更新"在同一锁内，避免与任务启动竞态 | | `TaskManager._lock` | `core/task_manager.py` | `_running` / `_stop_requested` | **抢占时禁止在锁内调 `stop_device`（死锁）** | | `_engine_lock` | `core/ocr.py` | OCR 引擎懒加载 + 推理串行 | 推理 0.2~0.5s，锁开销可忽略 | | `_scan_lock` / `_stop_event` | `core/device_discovery.py` | 定时/手动扫描互斥 | `acquire(blocking=False)` | | `_status_cache_lock` | `core/task_manager.py` | 状态缓存（TTL 5s） | 避免 `/api/status` 每次都查库 + adb | **无锁部分**：`device_pool` 与 `models` 不持显式锁，依赖"每次操作独立 app context" + 数据库自身的并发控制（MySQL 下是 InnoDB 行锁 + READ COMMITTED，回退 SQLite 时是 WAL + `busy_timeout=5000`）。 ### 3.3 错峰与心跳 - **错峰启动**：`_START_STAGGER_SEC = 0.2`，第 i 台设备延迟 `i × 0.2s` 启动（100 台 ≈ 20s 铺开），避免批量触发时的 adb 连接风暴。 - **心跳看门狗**：任何状态写入都会刷新 `last_heartbeat`；`running/connecting` 设备超过 120s 无心跳 → `status="error"` + `last_error="心跳超时…"`。长耗时的业务循环必须周期性 `self.heartbeat()`（`set_action` / `set_progress` 也会刷新）。 - **单实例约束**：同一 serial 同时只有一个 worker（`TaskManager._running`）；重复触发同任务跳过，不同任务未开抢占也跳过。 --- ## 4. 设备生命周期 ### 4.0 设备身份：名称 + 指纹（2026-09-11） 设备池原以 **serial（IP）当身份**，设备一换 IP 旧记录就成了连不上的"僵尸条目"，分组与 serial 模式的任务还吊着死地址。现在拆成三层： | 概念 | 是否稳定 | 作用 | |------|---------|------| | `name`（名称，**必填唯一**） | 稳定 | 人可读身份；分组/任务/日志按名称认设备 | | `fingerprint`（`ro.serialno`） | 稳定 | **机器识别**：认出"这是同一台设备" | | `serial`（IP:5555 / USB 序号） | **可变** | 当前连接地址 | **认领**（`device_pool.claim_device` 自动 / `relocate_device` 人工）：指纹命中或人工指认后， 把旧记录迁到新地址（名称/型号/备注/启用状态/添加时间全保留）并同步引用。 > ⚠️ **引用同步必须同时改库与内存**：分组、任务在 `TaskManager` 里还有一份内存副本， > **调度用的是内存对象**——只改库不重启不生效（表现为"分组里少一台、任务仍跑向旧地址"）。 > 因此 `device_pool` 迁址后回调 `TaskManager.sync_device_serial`，由装配层用 > `device_pool.set_move_hook(...)` 注册；`device_pool` 不能反向 import `task_manager`（循环依赖）。 ### 4.1 入池（三条路径） | 路径 | 入口 | 过程 | |------|------|------| | 手工添加 | `POST /api/devices/pool/add` | `device_pool.add_device`（upsert）→ 有 `:` 则 `adb connect` → 后台采集型号 | | 自动发现确认 | `POST /api/devices/discovery/confirm` | 扫描写 `pending_device` → 确认后 `add_device` + 删 pending + connect + 采型号 | | 启动预连接 | `_preconnect_pool_devices()` | 进程启动时并发 connect 池内网络设备（仅 `IP:5555`） | > 任何一条入池路径在设备可连时都会读取**设备指纹**；指纹命中池中已有设备 = 同一台换了地址 > → 走**认领**（§4.0），不新增记录。 断联设备的自动重连由发现线程每轮执行（只重连 `IP:5555`）。 ### 4.2 可用性判定 ``` list_configured() 设备池中 enabled=True 的 serial list_online() 本机 adb devices 中 state=device（池内有 USB 设备时并查 220 远程 adb server） list_ready() 两者交集 ← 调度 "all" 模式取这个 ``` ### 4.3 执行期状态字段 设备状态存在内存注册表 `_WORKERS[serial]`（**不落库，重启即清零**）： | 字段 | 含义 | |------|------| | `status` | `idle` / `connecting` / `running` / `done` / `error` / `released` / `failed` | | `serial` / `model` / `device_name` | 标识 | | `task_job` / `attempt` / `max_attempts` | 当前任务与重试进度 | | `current_action` / `progress` | 当前动作与进度（前端直接渲染） | | `last_error` / `last_warning` | 最近错误 / 选择器健康告警 | | `last_heartbeat` / `end_time` | 看门狗与时长上限 | | `present` / `ready` | 是否在线（对外 `/api/status` 字段） | ### 4.4 状态迁移 **Worker 侧**（`BaseWorker.run`）： ``` connecting ──获取设备──▶ u2 连接 ──▶ running ──▶ setup ──▶ run_task ──▶ teardown │ 未被 stop ──▶ done │ 异常 ──▶ error（DeviceOfflineError 单独分类，不重试） finally ──▶ 仅当仍为 running/connecting 时置 released ``` > `finally` 里的状态判断是为了**不覆盖业务结果**（`done`/`error` 必须保留）。 **调度侧**（`_run_with_retry`）：worker 结束后读 `status` → `done` 即成功返回；否则按 `max_attempts` 重试（`[transient]` 错误额外加长退避）→ 重试耗尽置 `status="failed"`，并把**真实失败原因**拼进 `last_error`（截断 200 字符）。 ### 4.5 释放 `STFDevice.release()` 是**空实现**——直连模式下**绝不 disconnect**（共享 adb transport 红线）。类名 `STFDevice` / `STFError` 是 STF 时代的历史命名，功能上已与 STF 无关。 --- ## 5. 任务调度链路 ### 5.1 完整调用链 ``` ① 注册 add_job / update_job / toggle_job → _add_cron：scheduler.add_job(_on_cron_trigger, CronTrigger.from_crontab(...), id=f"job_{id}_start", replace_existing=True) （cron_stop 额外注册 job_{id}_stop；启动时 _load() 为 enabled 任务重注册） ② 触发 APScheduler → _on_cron_trigger(job_id) → 任务存在？→ _in_run_window(schedule)？→ _run_job(job) ③ 解析 job.resolve_serials(self) # all=池内在线 / group=分组∩池 / serial=指定 → 为空则 warning 返回 ④ 任务类 get_task_class(job.task_type) → 未知则 error 返回（run_job_now 会直接报错） → task_cls()；max_attempts = max(1, retry.max_attempts) ⑤ 铺开 对每台设备起线程 _run_with_retry(task, serial, job, ..., idx * 0.2s) ⑥ 单设备 _run_with_retry： sleep(错峰) → 停止检查 → 单实例/抢占判定 → 登记 _running[serial] → _update_status(task_job=..., attempt=...) → worker = task.create_worker(serial, job.params) → worker.start() → worker.join() → 读 status：done 成功；否则重试（[transient] 退避 max(delay,120)s） → 耗尽：status="failed" + last_error（含真实原因） → finally：清停止标志；本任务若是抢占任务则归还设备（重跑被抢占任务） ⑦ 上报 worker 内 _update_status → 内存注册表 → TaskManager.get_status（5s 缓存） → _merge_status（合并池信息、清理陈旧条目） → GET /api/status → 前端 5s 轮询渲染 ⑧ 停止 stop_device / stop_all → _stop_requested.add + worker.stop()（置 Event） 业务循环检查 self.stopped() ⑨ 下次运行时间 next_run_of → _next_run_time（考虑运行窗口，最多向后探测 200 次） ``` ### 5.2 抢占机制 任务参数 `preempt=true` 时：`all` 模式目标集合变为"全部在线池内设备"（含正在跑的）；遇到设备已被占用时在**锁外**调 `stop_device` 并最多等 30s 接管；本任务结束后自动重新启动被抢占的任务（`preempted_job` 必须定义在重试循环外，否则归还信息会丢）。 --- ## 6. 前端架构 ### 6.1 单页应用 - 主页面 `templates/admin/monitor.html`：一个内联 `