chore: 初始化仓库（排除 adb 二进制文件以解决推送超时）
This commit is contained in:
@@ -0,0 +1,734 @@
|
||||
"""通用任务管理框架：任务类型注册 + 设备分组 + 任务计划 + 定时调度 + 重试 + 持久化。
|
||||
|
||||
设计目标：可扩展，未来加非抖音任务只需注册新的 Task 类。
|
||||
|
||||
核心概念：
|
||||
TaskType — 任务类型（如"抖音养号"），可注册，含默认参数和 worker 工厂
|
||||
DeviceGroup — 设备分组，持久化到 SQLite（core.models.DeviceGroup）
|
||||
TaskJob — 任务计划（类型+目标+参数+调度+重试），持久化到 SQLite（core.models.TaskJob）
|
||||
TaskManager — 统管调度器、分组、任务、运行实例、状态
|
||||
|
||||
定时：APScheduler BackgroundScheduler，cron 触发后为每个目标设备起 worker。
|
||||
支持三种调度模式：
|
||||
once — 手动/立即执行，不注册 cron
|
||||
cron — 仅注册启动 cron（到点启动 worker）
|
||||
cron_stop — 注册启动 cron + 停止 cron（到点启动，到点停止）
|
||||
重试：单设备 worker 失败后按 retry.max_attempts 重试，间隔 retry.delay。
|
||||
并发：同一 serial 同时只允许一个 worker，避免冲突。
|
||||
"""
|
||||
import os
|
||||
import json
|
||||
import time
|
||||
import uuid
|
||||
import threading
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
from apscheduler.schedulers.background import BackgroundScheduler
|
||||
from apscheduler.triggers.cron import CronTrigger
|
||||
|
||||
from config import DATA_DIR
|
||||
from core.logger import get_logger
|
||||
from core.models import db, DeviceGroup as GroupRow, TaskJob as JobRow
|
||||
from .stf_client import STFClient, DeviceOfflineError
|
||||
from .adb_helper import get_foreground_app, adb_connect_light, adb_disconnect
|
||||
from .device_worker import (
|
||||
get_all_worker_status, _update_status, _remove_worker,
|
||||
_WORKERS, _WORKERS_LOCK,
|
||||
start_watchdog, stop_watchdog,
|
||||
)
|
||||
from tasks import list_task_types, get_task_class
|
||||
|
||||
_log = get_logger("core.tm")
|
||||
|
||||
|
||||
# ================== 设备分组 ==================
|
||||
class DeviceGroup:
|
||||
def __init__(self, name, serials=None, description=""):
|
||||
self.name = name
|
||||
self.serials = serials or []
|
||||
self.description = description
|
||||
|
||||
def to_dict(self):
|
||||
return {"name": self.name, "serials": self.serials, "description": self.description}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, d):
|
||||
return cls(d["name"], d.get("serials", []), d.get("description", ""))
|
||||
|
||||
|
||||
# ================== 任务计划 ==================
|
||||
class TaskJob:
|
||||
"""一个任务计划：什么任务、跑哪些设备、什么参数、何时跑、失败重试策略。"""
|
||||
|
||||
def __init__(self, job_id=None, name="", task_type="douyin_nurture",
|
||||
target=None, params=None, schedule=None, retry=None, enabled=True):
|
||||
self.id = job_id or uuid.uuid4().hex[:8]
|
||||
self.name = name
|
||||
self.task_type = task_type
|
||||
# target: {"mode": "all"|"group"|"serial", "group_name": "", "serial": ""}
|
||||
self.target = target or {"mode": "all"}
|
||||
self.params = params or {}
|
||||
self.schedule = schedule or {"mode": "once"} # {"mode":"once"|"cron", "cron":"0 9 * * *"}
|
||||
self.retry = retry or {"max_attempts": 1, "delay": 60}
|
||||
self.enabled = enabled
|
||||
|
||||
def to_dict(self):
|
||||
return {"id": self.id, "name": self.name, "task_type": self.task_type,
|
||||
"target": self.target, "params": self.params,
|
||||
"schedule": self.schedule, "retry": self.retry, "enabled": self.enabled}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, d):
|
||||
return cls(d["id"], d["name"], d["task_type"], d.get("target"),
|
||||
d.get("params"), d.get("schedule"), d.get("retry"), d.get("enabled", True))
|
||||
|
||||
def resolve_serials(self, manager):
|
||||
"""根据 target 解析出实际要跑的 serial 列表。"""
|
||||
mode = self.target.get("mode", "all")
|
||||
if mode == "serial":
|
||||
return [self.target["serial"]]
|
||||
if mode == "group":
|
||||
g = manager.groups.get(self.target.get("group_name"))
|
||||
return list(g.serials) if g else []
|
||||
# all：返回所有空闲设备
|
||||
try:
|
||||
return [d["serial"] for d in manager.stf.list_free_devices()]
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
# ================== 前台 App 扫描器 ==================
|
||||
class _ForegroundScanner:
|
||||
"""前台 App 扫描器：手动触发，获取所有在线设备的前台 App。
|
||||
|
||||
设计原则：**不打扰设备**，扫描过程不会让设备退出当前 App。
|
||||
|
||||
策略（按设备状态区分获取逻辑）：
|
||||
- worker 运行中设备：用已有 remote_adb_url 直接查询（已有 adb 连接，无额外开销）
|
||||
- 空闲设备：直接 adb connect <serial> → 查询 → adb disconnect（绕过 STF，不打扰设备）
|
||||
- 被别人占用的设备：标记 "他人占用"
|
||||
- STF occupy/release 会打扰设备（可能退回桌面），绝不使用
|
||||
|
||||
serial 格式为 IP:5555（设备本身的 adb 网络地址），可直接 adb connect。
|
||||
adb connect/disconnect 只建立/断开调试连接，不影响设备 UI。
|
||||
dumpsys window 只读取窗口状态，不执行任何操作。
|
||||
"""
|
||||
|
||||
def __init__(self, stf):
|
||||
self.stf = stf
|
||||
self._cache = {} # serial -> app_name
|
||||
self._cache_lock = threading.Lock()
|
||||
self._scanning = threading.Event() # 标记是否正在扫描
|
||||
self._last_scan_time = 0 # 上次扫描完成时间戳
|
||||
self._last_scan_count = 0 # 上次扫描设备数
|
||||
|
||||
def stop(self):
|
||||
"""兼容 TaskManager.shutdown() 调用。"""
|
||||
pass
|
||||
|
||||
def get(self, serial):
|
||||
"""读取缓存的前台 App。"""
|
||||
with self._cache_lock:
|
||||
return self._cache.get(serial)
|
||||
|
||||
@property
|
||||
def is_scanning(self):
|
||||
return self._scanning.is_set()
|
||||
|
||||
@property
|
||||
def last_scan_time(self):
|
||||
return self._last_scan_time
|
||||
|
||||
def scan_once(self):
|
||||
"""手动触发一次扫描（后台线程执行，不阻塞调用者）。
|
||||
|
||||
如果已有扫描在进行中，直接返回 False。
|
||||
"""
|
||||
if self._scanning.is_set():
|
||||
return False
|
||||
t = threading.Thread(target=self._scan_all, name="fg-scan-once", daemon=True)
|
||||
t.start()
|
||||
return True
|
||||
|
||||
def _scan_all(self):
|
||||
"""扫描所有在线设备（后台线程执行）。
|
||||
|
||||
按设备归属分四类处理：
|
||||
1. worker 运行中：用已有 remote_adb_url 直接查询（无额外开销）
|
||||
2. 自己账户占用但无 worker：调用 STF remoteConnect 获取隧道查询
|
||||
（不 occupy/release，不打扰设备 UI）
|
||||
3. 完全空闲设备（using=False）：尝试轻量 adb connect serial
|
||||
（单次尝试，不 kill-server，不影响其他 worker）
|
||||
4. 被他人占用：标记 "(他人占用)"
|
||||
"""
|
||||
self._scanning.set()
|
||||
_log.info("前台 App 扫描已启动")
|
||||
try:
|
||||
all_devices = self.stf.list_all_devices()
|
||||
except Exception as e:
|
||||
_log.error("前台 App 扫描: 获取设备列表失败: %s", e)
|
||||
self._scanning.clear()
|
||||
return
|
||||
|
||||
# 自己账户已占用的设备列表（用于区分"自己占用"vs"他人占用"）
|
||||
try:
|
||||
my_serials = {d["serial"] for d in self.stf.list_my_devices()}
|
||||
except Exception:
|
||||
my_serials = set()
|
||||
|
||||
worker_status = {w["serial"]: w for w in get_all_worker_status()}
|
||||
|
||||
# 分类设备
|
||||
have_conn = {} # serial -> remote_adb_url（worker 运行中，已有 adb 连接）
|
||||
my_owned = [] # 自己账户占用但无 running worker，需 remoteConnect 获取隧道
|
||||
free_serials = [] # 完全空闲设备，尝试轻量 adb connect
|
||||
skip_serials = [] # 被他人占用
|
||||
|
||||
for dev in all_devices:
|
||||
if not dev.get("present"):
|
||||
continue
|
||||
serial = dev.get("serial", "")
|
||||
if not serial:
|
||||
continue
|
||||
w = worker_status.get(serial, {})
|
||||
url = w.get("remote_adb_url")
|
||||
if url and w.get("status") in ("running", "connecting"):
|
||||
# 1. 设备正在执行任务，已有 adb 连接，直接查询
|
||||
have_conn[serial] = url
|
||||
elif serial in my_serials:
|
||||
# 2. 自己账户占用但无 worker，可安全 remoteConnect（不打扰设备）
|
||||
my_owned.append(serial)
|
||||
elif dev.get("using"):
|
||||
# 4. 被他人占用
|
||||
skip_serials.append(serial)
|
||||
else:
|
||||
# 3. 完全空闲，尝试轻量 adb connect（不 kill-server）
|
||||
free_serials.append(serial)
|
||||
|
||||
results = {}
|
||||
_log.info("前台 App 扫描分类: 运行中=%d, 自己占用=%d, 空闲=%d, 他人占用=%d",
|
||||
len(have_conn), len(my_owned), len(free_serials), len(skip_serials))
|
||||
|
||||
# 1. worker 运行中设备：用已有 remote_adb_url 查询（并发 10）
|
||||
if have_conn:
|
||||
with ThreadPoolExecutor(max_workers=min(10, len(have_conn))) as pool:
|
||||
futures = {pool.submit(get_foreground_app, url): s
|
||||
for s, url in have_conn.items()}
|
||||
for fut in as_completed(futures, timeout=15):
|
||||
s = futures[fut]
|
||||
try:
|
||||
results[s] = fut.result()
|
||||
except Exception:
|
||||
results[s] = None
|
||||
|
||||
# 2. 自己占用设备：remoteConnect 获取隧道 → 查询 → 断开（并发 5，不打扰设备）
|
||||
if my_owned:
|
||||
with ThreadPoolExecutor(max_workers=min(5, len(my_owned))) as pool:
|
||||
futures = {pool.submit(self._scan_my_owned, s): s
|
||||
for s in my_owned}
|
||||
for fut in as_completed(futures, timeout=30):
|
||||
s = futures[fut]
|
||||
try:
|
||||
results[s] = fut.result()
|
||||
except Exception:
|
||||
results[s] = None
|
||||
|
||||
# 3. 空闲设备：轻量 adb connect serial → 查询 → disconnect（并发 10）
|
||||
if free_serials:
|
||||
with ThreadPoolExecutor(max_workers=min(10, len(free_serials))) as pool:
|
||||
futures = {pool.submit(self._scan_free, s): s
|
||||
for s in free_serials}
|
||||
for fut in as_completed(futures, timeout=20):
|
||||
s = futures[fut]
|
||||
try:
|
||||
results[s] = fut.result()
|
||||
except Exception:
|
||||
results[s] = None
|
||||
|
||||
# 4. 被别人占用的设备
|
||||
for s in skip_serials:
|
||||
results[s] = "(他人占用)"
|
||||
|
||||
# 更新缓存
|
||||
with self._cache_lock:
|
||||
for s, app in results.items():
|
||||
self._cache[s] = app
|
||||
|
||||
self._last_scan_time = time.time()
|
||||
self._last_scan_count = len(results)
|
||||
self._scanning.clear()
|
||||
_log.info("前台 App 扫描完成: %d 台设备", len(results))
|
||||
|
||||
def _scan_my_owned(self, serial):
|
||||
"""自己账户占用的设备：通过 STF remoteConnect 获取隧道查询。
|
||||
|
||||
不调用 occupy/release，只建立/断开 ADB 隧道，不打扰设备 UI。
|
||||
"""
|
||||
try:
|
||||
url = self.stf.remote_connect(serial)
|
||||
if not url:
|
||||
return None
|
||||
try:
|
||||
if not adb_connect_light(url):
|
||||
return None
|
||||
return get_foreground_app(url)
|
||||
finally:
|
||||
adb_disconnect(url)
|
||||
self.stf.remote_disconnect(serial)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def _scan_free(self, serial):
|
||||
"""空闲设备：轻量 adb connect serial 查询（单次尝试，不 kill-server）。
|
||||
|
||||
serial 通常为 IP:5555，若设备 5555 端口未开放则快速失败。
|
||||
绝不重试、绝不 kill-server，避免影响正在运行的 worker。
|
||||
"""
|
||||
try:
|
||||
if not adb_connect_light(serial):
|
||||
return "(不可达)"
|
||||
try:
|
||||
return get_foreground_app(serial)
|
||||
finally:
|
||||
adb_disconnect(serial)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
# ================== 任务管理器 ==================
|
||||
class TaskManager:
|
||||
def __init__(self, stf_client=None, app=None):
|
||||
self.stf = stf_client or STFClient()
|
||||
self.app = app # Flask app，用于 db context
|
||||
self.scheduler = BackgroundScheduler(timezone="Asia/Shanghai")
|
||||
self.scheduler.start()
|
||||
start_watchdog() # 启动心跳看门狗
|
||||
self.groups = {} # name -> DeviceGroup(内存业务对象)
|
||||
self.jobs = {} # id -> TaskJob(内存业务对象)
|
||||
self._running = {} # serial -> {"worker", "job_id", "started_at", "attempt"}
|
||||
self._lock = threading.Lock()
|
||||
self._fg_scanner = _ForegroundScanner(self.stf)
|
||||
self._load()
|
||||
|
||||
def _db(self):
|
||||
"""获取 db context。task_manager 在后台线程跑，需自己推 app context。"""
|
||||
if self.app is None:
|
||||
raise RuntimeError("TaskManager 未关联 Flask app，无法访问数据库")
|
||||
return self.app.app_context()
|
||||
|
||||
# ---- 持久化（SQLite，替代旧 JSON 文件）----
|
||||
def _load(self):
|
||||
"""从数据库加载分组和任务到内存。"""
|
||||
try:
|
||||
with self._db():
|
||||
for row in GroupRow.query.all():
|
||||
g = DeviceGroup(row.name, row.get_serials(), row.description or "")
|
||||
self.groups[g.name] = g
|
||||
dirty = False # 标记是否有任务参数需要写回
|
||||
for row in JobRow.query.all():
|
||||
j = TaskJob(job_id=row.id, name=row.name, task_type=row.task_type,
|
||||
target=row.get_target(), params=row.get_params(),
|
||||
schedule=row.get_schedule(), retry=row.get_retry(),
|
||||
enabled=row.enabled)
|
||||
# 数据规整：移除已废弃的 action 配置（如抖音的 comment 已删除）
|
||||
if j.task_type == "douyin_nurture":
|
||||
actions = j.params.get("actions", {})
|
||||
if "comment" in actions:
|
||||
del actions["comment"]
|
||||
row.set_params(j.params)
|
||||
dirty = True
|
||||
_log.info(f"任务 {j.name}({j.id}): 已移除废弃的 comment 配置")
|
||||
self.jobs[j.id] = j
|
||||
if dirty:
|
||||
db.session.commit()
|
||||
_log.info("已将规整后的任务参数写回数据库")
|
||||
_log.info(f"从数据库加载 {len(self.groups)} 个分组, {len(self.jobs)} 个任务")
|
||||
except Exception as e:
|
||||
_log.error(f"从数据库加载失败: {e}")
|
||||
# 恢复启用的 cron 任务
|
||||
for job in self.jobs.values():
|
||||
if job.enabled and job.schedule.get("mode") in ("cron", "cron_stop"):
|
||||
self._add_cron(job)
|
||||
|
||||
def _save_groups(self):
|
||||
"""保存所有分组到数据库（upsert）。"""
|
||||
with self._db():
|
||||
for g in self.groups.values():
|
||||
row = GroupRow.query.filter_by(name=g.name).first()
|
||||
if not row:
|
||||
row = GroupRow(name=g.name)
|
||||
db.session.add(row)
|
||||
row.set_serials(g.serials)
|
||||
row.description = g.description
|
||||
db.session.commit()
|
||||
|
||||
def _save_jobs(self):
|
||||
"""保存所有任务到数据库（upsert）。"""
|
||||
with self._db():
|
||||
for j in self.jobs.values():
|
||||
row = JobRow.query.get(j.id)
|
||||
if not row:
|
||||
row = JobRow(id=j.id)
|
||||
db.session.add(row)
|
||||
row.name = j.name
|
||||
row.task_type = j.task_type
|
||||
row.set_target(j.target)
|
||||
row.set_params(j.params)
|
||||
row.set_schedule(j.schedule)
|
||||
row.set_retry(j.retry)
|
||||
row.enabled = j.enabled
|
||||
db.session.commit()
|
||||
|
||||
# ---- 分组管理 ----
|
||||
def add_group(self, name, serials, description=""):
|
||||
g = DeviceGroup(name, serials, description)
|
||||
self.groups[name] = g
|
||||
self._save_groups()
|
||||
return g
|
||||
|
||||
def update_group(self, name, serials=None, description=None):
|
||||
g = self.groups.get(name)
|
||||
if not g:
|
||||
return None
|
||||
if serials is not None:
|
||||
g.serials = serials
|
||||
if description is not None:
|
||||
g.description = description
|
||||
self._save_groups()
|
||||
return g
|
||||
|
||||
def delete_group(self, name):
|
||||
if name in self.groups:
|
||||
del self.groups[name]
|
||||
self._save_groups()
|
||||
return True
|
||||
return False
|
||||
|
||||
# ---- 任务计划管理 ----
|
||||
def add_job(self, name, task_type, target, params, schedule, retry, enabled=True):
|
||||
job = TaskJob(name=name, task_type=task_type, target=target, params=params,
|
||||
schedule=schedule, retry=retry, enabled=enabled)
|
||||
self.jobs[job.id] = job
|
||||
if enabled and schedule.get("mode") in ("cron", "cron_stop"):
|
||||
self._add_cron(job)
|
||||
self._save_jobs()
|
||||
return job
|
||||
|
||||
def update_job(self, job_id, **fields):
|
||||
job = self.jobs.get(job_id)
|
||||
if not job:
|
||||
return None
|
||||
# 更新前先移除旧 cron
|
||||
was_cron = job.enabled and job.schedule.get("mode") in ("cron", "cron_stop")
|
||||
if was_cron:
|
||||
self._remove_cron(job_id)
|
||||
for k, v in fields.items():
|
||||
if v is not None and hasattr(job, k):
|
||||
setattr(job, k, v)
|
||||
if job.enabled and job.schedule.get("mode") in ("cron", "cron_stop"):
|
||||
self._add_cron(job)
|
||||
self._save_jobs()
|
||||
return job
|
||||
|
||||
def delete_job(self, job_id):
|
||||
self._remove_cron(job_id)
|
||||
if job_id in self.jobs:
|
||||
del self.jobs[job_id]
|
||||
self._save_jobs()
|
||||
return True
|
||||
return False
|
||||
|
||||
def toggle_job(self, job_id, enabled):
|
||||
job = self.jobs.get(job_id)
|
||||
if not job:
|
||||
return None
|
||||
job.enabled = enabled
|
||||
if enabled and job.schedule.get("mode") in ("cron", "cron_stop"):
|
||||
self._add_cron(job)
|
||||
else:
|
||||
self._remove_cron(job_id)
|
||||
self._save_jobs()
|
||||
return job
|
||||
|
||||
# ---- APScheduler 集成 ----
|
||||
def _cron_id(self, job_id, kind="start"):
|
||||
"""生成 cron job id。kind: start=启动 cron，stop=停止 cron。"""
|
||||
return f"job_{job_id}_{kind}"
|
||||
|
||||
def _add_cron(self, job):
|
||||
"""注册定时任务。支持三种模式：
|
||||
- once: 不注册 cron
|
||||
- cron: 只注册启动 cron（到点启动 worker）
|
||||
- cron_stop: 注册启动 cron + 停止 cron（到点启动，到点停止）
|
||||
"""
|
||||
mode = job.schedule.get("mode", "once")
|
||||
if mode not in ("cron", "cron_stop"):
|
||||
return
|
||||
|
||||
start_cron = job.schedule.get("cron", "")
|
||||
if start_cron:
|
||||
try:
|
||||
trigger = CronTrigger.from_crontab(start_cron)
|
||||
except Exception as e:
|
||||
_log.error(f"启动 cron 解析失败 {start_cron}: {e}")
|
||||
else:
|
||||
self.scheduler.add_job(self._on_cron_trigger, trigger, args=[job.id],
|
||||
id=self._cron_id(job.id, "start"),
|
||||
replace_existing=True)
|
||||
_log.info(f"已注册启动定时 {job.name}({job.id}) cron={start_cron}")
|
||||
|
||||
if mode == "cron_stop":
|
||||
stop_cron = job.schedule.get("stop_cron", "")
|
||||
if stop_cron:
|
||||
try:
|
||||
trigger = CronTrigger.from_crontab(stop_cron)
|
||||
except Exception as e:
|
||||
_log.error(f"停止 cron 解析失败 {stop_cron}: {e}")
|
||||
else:
|
||||
self.scheduler.add_job(self._on_cron_stop, trigger, args=[job.id],
|
||||
id=self._cron_id(job.id, "stop"),
|
||||
replace_existing=True)
|
||||
_log.info(f"已注册停止定时 {job.name}({job.id}) stop_cron={stop_cron}")
|
||||
|
||||
def _remove_cron(self, job_id):
|
||||
"""移除该任务的所有 cron job（启动 + 停止）。"""
|
||||
for kind in ("start", "stop"):
|
||||
try:
|
||||
self.scheduler.remove_job(self._cron_id(job_id, kind))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _on_cron_trigger(self, job_id):
|
||||
"""cron 触发：启动该任务的所有目标设备 worker。"""
|
||||
job = self.jobs.get(job_id)
|
||||
if not job:
|
||||
return
|
||||
_log.info(f"定时触发: {job.name}({job.id})")
|
||||
self._run_job(job)
|
||||
|
||||
def _on_cron_stop(self, job_id):
|
||||
"""停止 cron 触发：停止该任务正在运行的所有 worker。
|
||||
|
||||
只停本 job 启动的 worker，不影响其他任务。
|
||||
"""
|
||||
job = self.jobs.get(job_id)
|
||||
if not job:
|
||||
return
|
||||
with self._lock:
|
||||
items = [(s, info) for s, info in self._running.items()
|
||||
if info.get("job_id") == job_id]
|
||||
stopped = []
|
||||
for serial, info in items:
|
||||
w = info.get("worker")
|
||||
if w and w.is_alive():
|
||||
w.stop()
|
||||
stopped.append(serial)
|
||||
_log.info(f"定时停止 {job.name}({job.id}): 停止 {len(stopped)} 台设备 {stopped}")
|
||||
|
||||
def run_job_now(self, job_id):
|
||||
"""立即执行任务（手动触发）。"""
|
||||
job = self.jobs.get(job_id)
|
||||
if not job:
|
||||
return {"ok": False, "error": "任务不存在"}
|
||||
# 在独立线程跑，不阻塞调用方
|
||||
t = threading.Thread(target=self._run_job, args=(job,), daemon=True)
|
||||
t.start()
|
||||
return {"ok": True, "msg": f"任务 {job.name} 已触发"}
|
||||
|
||||
def _run_job(self, job):
|
||||
"""执行任务：为每个目标设备起 worker（含重试循环）。"""
|
||||
serials = job.resolve_serials(self)
|
||||
if not serials:
|
||||
_log.warning(f"任务 {job.name} 无可用设备")
|
||||
return
|
||||
task_cls = get_task_class(job.task_type)
|
||||
if not task_cls:
|
||||
_log.error(f"未知任务类型: {job.task_type}")
|
||||
return
|
||||
task = task_cls()
|
||||
max_attempts = max(1, job.retry.get("max_attempts", 1))
|
||||
delay = job.retry.get("delay", 60)
|
||||
|
||||
for serial in serials:
|
||||
# 每台设备一个重试循环线程，互不影响
|
||||
t = threading.Thread(target=self._run_with_retry,
|
||||
args=(task, serial, job, max_attempts, delay), daemon=True)
|
||||
t.start()
|
||||
|
||||
def _run_with_retry(self, task, serial, job, max_attempts, delay):
|
||||
"""单设备任务执行 + 重试。
|
||||
|
||||
异常分类：
|
||||
DeviceOfflineError — 设备掉线，立即放弃不重试（换设备也没用）
|
||||
其他异常 — 按 max_attempts 重试
|
||||
"""
|
||||
for attempt in range(1, max_attempts + 1):
|
||||
# 同一 serial 同时只能一个 worker
|
||||
with self._lock:
|
||||
if serial in self._running:
|
||||
_log.warning(f"{serial} 已有任务在跑，跳过 (job={job.name})")
|
||||
return
|
||||
self._running[serial] = {"job_id": job.id, "started_at": time.time(),
|
||||
"attempt": attempt, "task_type": job.task_type}
|
||||
_update_status(serial, task_job=job.name, attempt=attempt,
|
||||
max_attempts=max_attempts)
|
||||
|
||||
worker = None
|
||||
try:
|
||||
worker = task.create_worker(self.stf, serial, job.params)
|
||||
with self._lock:
|
||||
self._running[serial]["worker"] = worker
|
||||
_log.info(f"{serial} 开始任务 {job.name} (第{attempt}/{max_attempts}次)")
|
||||
worker.start()
|
||||
worker.join() # 等待 worker 结束
|
||||
# worker 正常结束（done 或被 stop）
|
||||
with self._lock:
|
||||
self._running.pop(serial, None)
|
||||
# 判断是否成功：看 status
|
||||
with _WORKERS_LOCK:
|
||||
st = _WORKERS.get(serial, {}).get("status")
|
||||
# 清除 task_job 标记，避免前端误判"运行中"
|
||||
_update_status(serial, task_job="")
|
||||
if st == "done":
|
||||
_log.info(f"{serial} 任务 {job.name} 成功完成")
|
||||
return
|
||||
if st == "released" and attempt == 1:
|
||||
# 被手动停止，不重试
|
||||
return
|
||||
_log.warning(f"{serial} 任务未成功(status={st})")
|
||||
except DeviceOfflineError as e:
|
||||
# 设备掉线，立即放弃，不重试
|
||||
_log.error(f"{serial} 设备离线，放弃任务 {job.name}: {e}")
|
||||
with self._lock:
|
||||
self._running.pop(serial, None)
|
||||
_update_status(serial, status="failed", task_job="",
|
||||
last_error=f"设备离线: {e}")
|
||||
return
|
||||
except Exception as e:
|
||||
_log.error(f"{serial} 执行异常: {e}", exc_info=True)
|
||||
with self._lock:
|
||||
self._running.pop(serial, None)
|
||||
_update_status(serial, task_job="")
|
||||
|
||||
if attempt < max_attempts:
|
||||
# 检查是否是端口耗尽类临时错误，需要更长退避等端口释放
|
||||
with _WORKERS_LOCK:
|
||||
err = _WORKERS.get(serial, {}).get("last_error", "")
|
||||
if err.startswith("[transient]"):
|
||||
# Windows TCP 端口耗尽，TIME_WAIT 默认 2-4 分钟，等 120 秒
|
||||
extra_delay = max(delay, 120)
|
||||
_log.info(f"{serial} ADB 连接临时错误（端口耗尽），{extra_delay}s 后重试 ({attempt+1}/{max_attempts})")
|
||||
time.sleep(extra_delay)
|
||||
else:
|
||||
_log.info(f"{serial} {delay}s 后重试 ({attempt+1}/{max_attempts})")
|
||||
time.sleep(delay)
|
||||
|
||||
_log.error(f"{serial} 任务 {job.name} 重试耗尽，放弃")
|
||||
_update_status(serial, status="failed", last_error=f"{job.name} 重试{max_attempts}次失败")
|
||||
|
||||
# ---- 运行控制 ----
|
||||
def stop_device(self, serial):
|
||||
"""停止指定设备的 worker。"""
|
||||
with self._lock:
|
||||
info = self._running.get(serial)
|
||||
if not info:
|
||||
return False
|
||||
w = info.get("worker")
|
||||
if w and w.is_alive():
|
||||
w.stop()
|
||||
return True
|
||||
return False
|
||||
|
||||
def stop_all(self):
|
||||
"""停止所有运行中的 worker。"""
|
||||
with self._lock:
|
||||
items = list(self._running.items())
|
||||
stopped = []
|
||||
for serial, info in items:
|
||||
w = info.get("worker")
|
||||
if w and w.is_alive():
|
||||
w.stop()
|
||||
stopped.append(serial)
|
||||
return stopped
|
||||
|
||||
def get_running(self):
|
||||
with self._lock:
|
||||
return {s: dict(v) for s, v in self._running.items()}
|
||||
|
||||
# ---- 状态查询（带缓存，避免 STF 请求阻塞前端）----
|
||||
_status_cache = None # (timestamp, data, error)
|
||||
_status_cache_lock = threading.Lock()
|
||||
_STATUS_CACHE_TTL = 5.0 # 缓存 5 秒，前端 5 秒刷新刚好命中
|
||||
|
||||
def get_status(self):
|
||||
"""综合状态：STF 设备池 + 本地 worker + 运行中的任务。
|
||||
|
||||
带 5 秒缓存：STF 请求慢时避免每次 /api/status 都打 STF 阻塞 Flask。
|
||||
worker 状态实时读（内存，无 IO），不受缓存影响。
|
||||
"""
|
||||
with self._status_cache_lock:
|
||||
if self._status_cache:
|
||||
ts, cached, err = self._status_cache
|
||||
if time.time() - ts < self._STATUS_CACHE_TTL:
|
||||
if err:
|
||||
return None, err
|
||||
# 用缓存的设备列表 + 实时 worker 状态重新组装
|
||||
return self._merge_status(cached), None
|
||||
# 缓存过期或不存在，重新拉 STF
|
||||
try:
|
||||
all_devices = self.stf.list_all_devices()
|
||||
except Exception as e:
|
||||
with self._status_cache_lock:
|
||||
self._status_cache = (time.time(), None, str(e))
|
||||
return None, f"获取设备列表失败: {e}"
|
||||
with self._status_cache_lock:
|
||||
self._status_cache = (time.time(), all_devices, None)
|
||||
return self._merge_status(all_devices), None
|
||||
|
||||
def _merge_status(self, all_devices):
|
||||
"""用 STF 设备列表 + 实时 worker 状态 + 前台 App 组装返回结果。"""
|
||||
worker_status = {w["serial"]: w for w in get_all_worker_status()}
|
||||
running = self.get_running()
|
||||
|
||||
result = []
|
||||
for dev in all_devices:
|
||||
serial = dev.get("serial", "")
|
||||
owner = dev.get("owner")
|
||||
w = worker_status.get(serial, {})
|
||||
r = running.get(serial, {})
|
||||
result.append({
|
||||
"serial": serial,
|
||||
"model": dev.get("model") or w.get("model", "") or dev.get("product", ""),
|
||||
"device_name": dev.get("name") or "",
|
||||
"present": dev.get("present", False),
|
||||
"ready": dev.get("ready", False),
|
||||
"stf_occupied": dev.get("using", False),
|
||||
"owner": owner.get("name", "") if owner else "",
|
||||
"worker_status": w.get("status", "idle"),
|
||||
"foreground_app": self._fg_scanner.get(serial),
|
||||
# 通用进度字段（任意 app 通用，前端统一解析展示）
|
||||
# 结构：{"done": int, "total": int, "unit": str, "action_counts": dict}
|
||||
"progress": w.get("progress", {}),
|
||||
"current_action": w.get("current_action", ""),
|
||||
"last_error": w.get("last_error", ""),
|
||||
"running_job": r.get("job_id", ""),
|
||||
"task_job": w.get("task_job", ""),
|
||||
"attempt": r.get("attempt", 0),
|
||||
})
|
||||
return result
|
||||
|
||||
def list_all_serials(self):
|
||||
"""返回 STF 上所有在线设备的 serial 列表（供分组表单勾选用）。
|
||||
|
||||
复用 get_status 缓存，避免开页面时阻塞。
|
||||
"""
|
||||
devices, err = self.get_status()
|
||||
if err or not devices:
|
||||
return []
|
||||
return [d["serial"] for d in devices if d.get("present")]
|
||||
|
||||
def shutdown(self):
|
||||
self.stop_all()
|
||||
self._fg_scanner.stop()
|
||||
stop_watchdog()
|
||||
self.scheduler.shutdown(wait=False)
|
||||
Reference in New Issue
Block a user