"""抖音养号任务定义。 本文件自包含所有抖音养号参数（观看、滑动、操作等），不依赖 core 的业务配置。 抖音专属操作（点赞/评论）在 actions/ 子包里，xpath 只适用于抖音。 修改养号行为：改下面 DEFAULT_PARAMS 默认值，或通过前端任务参数覆盖。 修改养号逻辑：改 DouyinNurtureWorker.run_task。 加新抖音操作：在 actions/ 下建 .py，继承 BaseAction + @register_action。 """ import time import random from tasks.base import BaseTask, register_task from .actions import list_action_types, get_action_class from core.device_worker import BaseWorker, _update_status from core.u2_helper import ensure_app_running, wait_for_app_home from core.logger import get_logger _log = get_logger("task.douyin") # ================== 抖音养号参数（本任务专属，写在这里） ================== DOUYIN_PKG = "com.ss.android.ugc.aweme" DEFAULT_PARAMS = { # 观看 "watch_count": 80, # 观看视频数量（0=不限数量，仅按 max_duration 终止） "watch_min": 5.0, # 单个视频最短观看秒数 "watch_max": 35.0, # 单个视频最长观看秒数 # 运行时长终止（通用，0=不限时，按 watch_count 终止） # 两个条件哪个先到就停；都为 0 则永不停止（需手动停止） "max_duration": 0, # 最大运行时长（秒），如 1800=30分钟，0=不限时 # 滑动 "swipe_min": 0.25, # 上滑手势最短时长(秒) "swipe_max": 0.50, # 上滑手势最长时长(秒) "gap_min": 1.0, # 视频间隔最短秒数 "gap_max": 3.0, # 视频间隔最长秒数 # 操作（每个操作有 enabled 开关 + 自身参数） "actions": { "like": { "enabled": True, "params": {"rate": 0.3, "method": "by_element"}, }, }, } class DouyinNurtureWorker(BaseWorker): """抖音养号 worker：看视频 + 按配置执行操作（点赞/评论等），被退出自动重连。 只需实现 run_task(d)，STF 占用/释放、u2 连接、异常、状态上报由基类处理。 操作参数优先级：任务 params.actions > 本文件 DEFAULT_PARAMS.actions。 """ def __init__(self, serial, params=None): super().__init__(serial, params) p = {**DEFAULT_PARAMS, **(self.params or {})} self.watch_count = int(p["watch_count"]) self.watch_min = float(p["watch_min"]) self.watch_max = float(p["watch_max"]) # 通用运行时长终止（0=不限时） self.max_duration = int(p.get("max_duration", 0)) self.swipe_min = float(p["swipe_min"]) self.swipe_max = float(p["swipe_max"]) self.gap_min = float(p["gap_min"]) self.gap_max = float(p["gap_max"]) # 操作配置：{action_type: {"enabled":bool, "params":{...}}} self.actions_cfg = p.get("actions", {}) # 实例化所有启用的操作 self._actions = [] for atype, cfg in self.actions_cfg.items(): if not cfg.get("enabled"): continue cls = get_action_class(atype) if cls: self._actions.append(cls()) # 启动日志：打印终止条件 + 启用的 action，便于排查 stop_conds = [] if self.watch_count > 0: stop_conds.append(f"数量={self.watch_count}") if self.max_duration > 0: stop_conds.append(f"时长={self.max_duration}s") if not stop_conds: stop_conds.append("无（需手动停止）") _log.info(f"[{serial}] 终止条件: {', '.join(stop_conds)}") if self._actions: summary = ", ".join( f"{a.action_type}(rate={self.actions_cfg.get(a.action_type, {}).get('params', {}).get('rate', '?')})" for a in self._actions ) _log.info(f"[{serial}] 启用操作: {summary}") else: _log.warning(f"[{serial}] 未启用任何操作，整个任务期间不会执行点赞") def run_task(self, d): """抖音养号主逻辑。d 是 u2.Device，已由基类连好。 终止条件（哪个先到就停）： - watch_count > 0：看完指定数量视频 - max_duration > 0：达到最大运行时长 - 两者都为 0：永不停止，需手动停止 """ def is_home(d): return (d(descriptionContains="拍摄").exists or d(descriptionContains="首页").exists) # 启动运行时长计时（基类通用方法） self._start_timer() d.app_start(DOUYIN_PKG, wait=True) if not wait_for_app_home(d, DOUYIN_PKG, is_home, timeout=40): self.set_action("首页加载超时，继续尝试") watched = 0 action_counts = {a.action_type: 0 for a in self._actions} # 初始化通用进度上报 # total=0 时前端不显示百分比，只显示已看数量 + 运行时长 total = self.watch_count if self.watch_count > 0 else 0 self.set_progress(done=0, total=total, unit="视频", action_counts=action_counts) while not self.stopped(): # 终止条件检查 if self.watch_count > 0 and watched >= self.watch_count: break if self.is_time_up(): _log.info(f"[{self.serial}] 达到最大运行时长 {self.max_duration}s，停止") break if not ensure_app_running(d, DOUYIN_PKG): _update_status(self.serial, status="error", last_error="抖音连续重启失败，放弃该设备") return _update_status(self.serial, douyin_running=True, last_error="") watch = random.uniform(self.watch_min, self.watch_max) # 动作描述包含运行时长，便于前端观察 elapsed = self.elapsed() time_info = f"，已运行 {elapsed//60}m{elapsed%60}s" if elapsed > 0 else "" self.set_action(f"观看视频 {watched+1}，{watch:.0f}s{time_info}") time.sleep(watch) # 检测当前是否为可操作的正常视频页（有右侧操作栏 = 点赞/评论按钮） # 营销广告、直播、异常页通常没有这些元素，直接滑过去不做操作 is_normal = self._is_normal_video(d) if is_normal: # 正常视频：执行所有启用的操作（看完视频后、滑动前） if not self._actions: _log.info(f"[{self.serial}] 视频{watched+1}: 正常视频但无启用操作") for action in self._actions: if self.stopped(): break cfg = self.actions_cfg.get(action.action_type, {}) params = {**action.default_params, **cfg.get("params", {})} rate = params.get("rate", "?") try: ok = action.execute(d, params, self) _log.info(f"[{self.serial}] 视频{watched+1}: {action.action_type} " f"rate={rate} execute={ok}") if ok: action_counts[action.action_type] += 1 time.sleep(random.uniform(0.5, 1.5)) except Exception as e: _log.error(f"[{self.serial}] 视频{watched+1}: 操作 {action.action_type} " f"异常: {e}") else: self.set_action(f"跳过广告/直播 {watched+1}") _log.info(f"[{self.serial}] 视频{watched+1}: 非正常视频页，跳过操作") if self.stopped(): break d.swipe(500, 1000, 500, 300, random.uniform(self.swipe_min, self.swipe_max)) time.sleep(random.uniform(self.gap_min, self.gap_max)) watched += 1 # 上报通用进度（含运行时长，前端展示） elapsed = self.elapsed() self.set_progress(done=watched, total=total, unit="视频", action_counts=dict(action_counts), elapsed=elapsed) # 终止原因汇总 stop_reason = "" if self.stopped(): stop_reason = "（手动停止）" elif self.is_time_up(): stop_reason = f"（达到时长 {self.max_duration}s）" elif self.watch_count > 0 and watched >= self.watch_count: stop_reason = f"（达到数量 {self.watch_count}）" summary = f"完成 {watched} 个视频" + "".join( f"，{k} {v}次" for k, v in action_counts.items() if v ) + stop_reason _update_status(self.serial, douyin_running=False, current_action=summary) @staticmethod def _is_normal_video(d): """判断当前是否为可操作的正常视频页。 正常视频页右侧有操作栏（点赞/评论/分享按钮）。 营销广告、直播、异常页通常没有这些元素 → 返回 False，直接滑过去。 检测任一特征元素存在即认为是正常视频页。 定位优先级：description 精确 → descriptionContains 模糊 → resourceId 列表 → 兜底 每次返回 False 时记日志，便于定位"为什么操作没执行"。 """ try: # 1. description 精确匹配（最稳） for desc in ("点赞", "未点赞", "评论", "未评论"): if d(description=desc).exists: return True # 2. descriptionContains 模糊匹配（适配不同版本文案） for kw in ("赞", "评论"): if d(descriptionContains=kw).exists: return True # 3. resourceId 列表（扩充候选，覆盖更多抖音版本） for rid in ("com.ss.android.ugc.aweme:id/aky", "com.ss.android.ugc.aweme:id/akq", "com.ss.android.ugc.aweme:id/d-like-view-icon", "com.ss.android.ugc.aweme:id/d-comment-view-icon", "com.ss.android.ugc.aweme:id/h3r"): if d(resourceId=rid).exists: return True # 所有特征元素都没找到 → 非正常视频页（广告/直播/异常） _log.info(f"_is_normal_video=False: 未找到点赞/评论特征元素") return False except Exception as e: _log.warning(f"_is_normal_video 异常: {e}") return False @register_task class DouyinNurtureTask(BaseTask): """抖音养号：看视频 + 按配置执行操作（点赞/评论等），被退出自动重连。""" task_type = "douyin_nurture" name = "抖音养号" description = "自动观看抖音视频，按配置执行点赞/评论等操作，被退出自动重连" default_params = dict(DEFAULT_PARAMS) @classmethod def list_action_types(cls): """返回本任务支持的抖音操作列表。""" return list_action_types() @classmethod def get_action_class(cls, action_type): """按 action_type 取抖音操作类。""" return get_action_class(action_type) def create_worker(self, serial, params): merged = {**DEFAULT_PARAMS, **(params or {})} # actions 字段深合并（参数级，保留前端没传的操作默认值） # 确保即使前端只传 actions.like.enabled，like.params.rate 也能拿到默认 0.3 default_actions = DEFAULT_PARAMS["actions"] merged_actions = params.get("actions", {}) if params else {} for atype, dflt in default_actions.items(): if atype not in merged_actions: # 前端完全没传这个操作 → 用默认 merged_actions[atype] = dflt else: # 前端传了 → 参数级深合并 cfg = merged_actions[atype] merged_cfg = {} for k in ("enabled", "params"): merged_cfg[k] = cfg.get(k, dflt.get(k)) # params 再深合并一层 merged_params = dict(dflt.get("params", {})) merged_params.update(cfg.get("params", {})) merged_cfg["params"] = merged_params merged_actions[atype] = merged_cfg merged["actions"] = merged_actions return DouyinNurtureWorker(serial, params=merged)