project/tbsg/start_run_ele.py

"""
饿了么闪购 — 调度任务入口（模仿 D:\\drug\\pdd\\start_run_pdd.py）
从调度系统拉取任务执行，不再手动传参。

用法: python start_run_ele.py <ADB序列号>
      （adb devices 查看序列号；不传参数启动会直接报错退出）

流程: 心跳注册 → 循环领任务 → step1打开App → step2搜索(品牌+药品名合并)
      → step3遍历采集(说明书/快照/资质/入库) → 上报结果 → 等600秒 → 领下一个任务
"""
import json
import random
import re
import sys
import time
from datetime import datetime
from pathlib import Path

import requests
import uiautomator2

sys.path.insert(0, str(Path(__file__).parent))

from steps.executor import SafeExecutor
from main1 import step1_open_app, step2_search, step3_swipe_and_enter
from commons.scheduler import CrawlerScheduler
from commons import err_log

# ── 配置 ────────────────────────────────────────────────
PLATFORM_ELE = 12          # 饿了么平台编号（与 db.py 一致）
LOOP_INTERVAL_MIN = 480    # 任务/无任务等待间隔下限（秒）
LOOP_INTERVAL_MAX = 720    # 上限；实际在两者间随机取（拟人，避免固定节奏被建模）
START_HOUR = 6             # 采集开始时间（该点之前不领任务；心跳保持在线）
API_HEADERS = {'X-Crawler-Token': 'zhijiayun_crawler_2026'}


def loop_interval() -> int:
    """本轮等待秒数：480~720 随机"""
    return random.randint(LOOP_INTERVAL_MIN, LOOP_INTERVAL_MAX)


def wait_until_start_hour():
    """当前时间早于 START_HOUR 点 → 等到整点再领任务（心跳线程不受影响，保持在线）"""
    now = datetime.now()
    if now.hour >= START_HOUR:
        return
    target = now.replace(hour=START_HOUR, minute=0, second=0, microsecond=0)
    wait_s = (target - now).total_seconds()
    print(f"[调度] 当前 {now.strftime('%H:%M')}，未到 {START_HOUR} 点开始时间，等待 {wait_s/3600:.1f} 小时...")
    if wait_s > 0:
        time.sleep(wait_s)

# 设备号通过命令行传入：python start_run_ele.py <ADB序列号>
DEVICE_ID = sys.argv[1].strip() if len(sys.argv) > 1 else ""

APP_PACKAGE = "me.ele"


# ── 账号压力统计(跨任务累计, 按天清零; 用于分析风控封号阈值) ──
_PRESSURE = {"day": None, "tasks": 0, "crawled": 0}


def _count_captcha_today(device_id: str) -> int:
    """当日验证码次数(main1 的验证码日志按日期开头计数)"""
    try:
        m = sys.modules.get("main1")
        if m is None or not hasattr(m, "_captcha_log_path"):
            return 0
        path = m._captcha_log_path(device_id)
        today = time.strftime("%Y-%m-%d")
        with open(path, encoding="utf-8") as f:
            return sum(1 for line in f if line.startswith(today))
    except Exception:
        return 0


def close_app(device_id: str):
    """任务结束（验证码超限停止/封号/正常完成/异常）后关闭饿了么App。
    下个任务 step1 会重新拉起；关闭失败不影响主流程。"""
    try:
        uiautomator2.connect(device_id).app_stop(APP_PACKAGE)
        print(f"[任务] 已关闭App({APP_PACKAGE})")
    except Exception as e:
        print(f"[任务] 关闭App失败(不影响流程): {e}")


def build_task_payload_from_api(task_dict: dict, device_row: str) -> dict:
    """从调度 API 返回的 task dict 构建采集可消费的 payload（照搬 PDD 逻辑）"""
    username = device_row
    device_id = task_dict['_device_id'] or username
    search_key = (
        f"{task_dict.get('product_brand', '')}"
        f"{task_dict.get('product_name', '')} "
        f"{task_dict.get('product_specs', '') or ''}"
    ).strip()
    # 规格列表：按 | 、 ， 分隔（用于规格过滤，照搬 PDD）
    spec_raw = task_dict.get('product_specs', '') or ''
    spec_list = [s.strip() for s in re.split(r'[|、，,\n\r]+', spec_raw) if s.strip()]
    collect_config_info = json.dumps({
        "sampling_cycle": task_dict.get('sampling_cycle', ''),
        "sampling_start_time": task_dict.get('sampling_start_time', 0),
        "sampling_end_time": task_dict.get('sampling_end_time', 0),
    }, ensure_ascii=False)
    return {
        "task_id": task_dict.get('id'),
        "search_key": search_key,
        "title_key": task_dict.get('product_name'),
        "brand": task_dict.get('product_brand', ''),
        "spec_list": spec_list,
        "product_specs": spec_raw,
        "enterprise_id": task_dict.get('company_id'),
        "collect_round": task_dict.get('collect_round'),
        "collect_equipment_account_id": task_dict.get('collect_equipment_account_id', 1),
        "collect_region_id": task_dict.get('collect_region_id', 1),
        "current_page": task_dict.get('current_page') or 0,   # 调度重派时给的页码（跨设备接力恢复）
        "collect_config_info": collect_config_info,
        "username": username,
        "device_id": device_id,
        "device_row": device_row,
    }


def run_task_worker(task_payload: dict, scheduler: CrawlerScheduler):
    """执行单个任务：打开App → 搜索 → 遍历采集 → 上报结果"""
    task_id = task_payload["task_id"]
    device_id = task_payload["device_id"]
    search_key = str(task_payload.get("search_key", "")).strip()
    keyword = str(task_payload.get("title_key") or search_key).strip()
    brand = str(task_payload.get("brand", "")).strip()
    t0 = time.time()
    report = {
        "task_id": task_id,
        "platform": scheduler.platform,
        "username": scheduler.username,
    }

    try:
        print(f"[任务 {task_id}] 开始执行，设备: {device_id}，搜索: {search_key}")
        # 重置停止标志（上个任务的验证码/封号停止不能污染本任务）
        _m = sys.modules.get("main1")
        if _m is not None:
            for _flag, _val in (("CAPTCHA_ABORTED", False), ("CAPTCHA_ABORT_REASON", ""),
                                ("ACCOUNT_ABORTED", False), ("CURRENT_PAGE", 0),
                                ("WHITE_SCREEN_REASSIGN", False), ("WHITE_SCREEN_REASSIGN_REASON", ""),
                                ("_LAST_SKIP_WHITE", False)):
                if hasattr(_m, _flag):
                    setattr(_m, _flag, _val)
        scheduler.limit_reached = False   # 上个任务的限额错误不污染本任务
        scheduler.limit_msg = ""
        ex = SafeExecutor(device_id)

        # step1: 打开 App
        if not step1_open_app(ex):
            print(f"[任务 {task_id}] step1 打开App失败")
            _kicked = bool(getattr(sys.modules.get("main1"), "ACCOUNT_ABORTED", False))
            err_log.log_error(device_id, "step1_open_app_failed",
                              message="打开App失败" + ("（账号被踢/封号）" if _kicked else ""),
                              extra={"task_id": task_id, "search_key": search_key, "account_aborted": _kicked})
            if getattr(sys.modules.get("main1"), "ACCOUNT_ABORTED", False):
                # 登录页xpath存在 = 账号被踢/封号
                report.update({"is_finished": 0, "need_reassign": 1, "exception_type": 5,
                               "crawled_count": 0, "current_page": 0, "remark": "账号被踢/封号"})
                print(f"[任务 {task_id}] 账号被踢/封号（登录页），回告调度停止")
            else:
                report.update({"is_finished": 0, "need_reassign": 1, "exception_type": 1, "crawled_count": 0})
            scheduler.post_report(report)
            return

        # step2: 搜索（品牌+药品名合并后的搜索词）
        if not step2_search(ex, search_key):
            print(f"[任务 {task_id}] step2 搜索失败")
            err_log.log_error(device_id, "step2_search_failed", message="搜索失败",
                              extra={"task_id": task_id, "search_key": search_key})
            report.update({"is_finished": 0, "need_reassign": 1, "exception_type": 2, "crawled_count": 0})
            scheduler.post_report(report)
            return

        # step3: 遍历采集（品牌+药品名过滤，任务字段随结果入库；逐页回告 current_page）
        visited = step3_swipe_and_enter(ex, keyword, brand, task_payload, scheduler)
        _main1_mod = sys.modules.get("main1")
        if getattr(_main1_mod, "ACCOUNT_ABORTED", False):
            # 账号被踢/封号：回告调度，需重新指派/换账号
            report.update({"is_finished": 0, "need_reassign": 1, "exception_type": 5,
                           "crawled_count": len(visited), "current_page": getattr(_main1_mod, "CURRENT_PAGE", 0), "remark": "账号被踢/封号"})
            print(f"[任务 {task_id}] 账号被踢/封号，回告调度停止")
        elif getattr(_main1_mod, "CAPTCHA_ABORTED", False):
            # 验证码频率超限（一天≥8次等）：回告调度
            _reason = getattr(_main1_mod, "CAPTCHA_ABORT_REASON", "") or "验证码频繁出现"
            report.update({"is_finished": 0, "need_reassign": 1, "exception_type": 5,
                           "crawled_count": len(visited), "current_page": getattr(_main1_mod, "CURRENT_PAGE", 0), "remark": _reason})
            print(f"[任务 {task_id}] {_reason}，回告调度停止")
        elif getattr(_main1_mod, "WHITE_SCREEN_REASSIGN", False):
            # 商品页反复白屏（重启恢复超限）：回告调度重派给其他设备
            _reason = getattr(_main1_mod, "WHITE_SCREEN_REASSIGN_REASON", "") or "页面持续白屏"
            report.update({"is_finished": 0, "need_reassign": 1,
                           "exception_type": getattr(_main1_mod, "WHITE_EXCEPTION_TYPE", 5),
                           "crawled_count": len(visited), "current_page": getattr(_main1_mod, "CURRENT_PAGE", 0),
                           "remark": _reason})
            print(f"[任务 {task_id}] {_reason}，回告调度重派")
        else:
            report.update({"is_finished": 1, "need_reassign": 0, "crawled_count": len(visited), "current_page": getattr(_main1_mod, "CURRENT_PAGE", 0)})
        if getattr(scheduler, "limit_reached", False):
            # 平台已释放任务（限额等）→ 不再回告（回告只会再次返回error）
            print(f"[任务 {task_id}] 平台已释放任务（{getattr(scheduler, 'limit_msg', '')}），跳过最终回告")
        else:
            scheduler.post_report(report)
        print(f"[任务 {task_id}] 执行完成，采集 {len(visited)} 条")
    except Exception as e:
        print(f"[任务 {task_id}] 执行异常: {e}")
        err_log.log_error(device_id, "task_exception", exc=e,
                          extra={"task_id": task_id, "search_key": search_key, "keyword": keyword})
        report.update({"is_finished": 0, "need_reassign": 1, "exception_type": 5, "crawled_count": 0, "current_page": getattr(sys.modules.get("main1"), "CURRENT_PAGE", 0)})
        scheduler.post_report(report)
    finally:
        # ── 账号压力日志: 每任务一条累计统计, 与 account_kicked 事件对照分析风控阈值 ──
        try:
            today = time.strftime("%Y-%m-%d")
            if _PRESSURE["day"] != today:
                _PRESSURE["day"] = today
                _PRESSURE["tasks"] = 0
                _PRESSURE["crawled"] = 0
            _PRESSURE["tasks"] += 1
            _PRESSURE["crawled"] += report.get("crawled_count", 0) or 0
            if report.get("is_finished") == 1:
                tag = "finished"
            elif "封号" in str(report.get("remark", "")):
                tag = "account_kicked"
            elif "验证码" in str(report.get("remark", "")):
                tag = "captcha_limit"
            elif getattr(scheduler, "limit_reached", False):
                tag = "limit_reached"
            else:
                tag = "failed"
            err_log.log_pressure(device_id, {
                "task_id": task_id, "result": tag,
                "crawled_today": _PRESSURE["crawled"], "tasks_today": _PRESSURE["tasks"],
                "captcha_today": _count_captcha_today(device_id),
                "runtime_min": round((time.time() - t0) / 60, 1),
                "search": search_key[:20],
            })
        except Exception:
            pass
        # 无论哪种方式结束（验证码超限/封号/正常完成/异常），都关掉App再进入等待
        close_app(device_id)


def main():
    if not DEVICE_ID:
        raise SystemExit("用法: python start_run_ele.py <ADB序列号>  （adb devices 查看序列号）")

    scheduler = CrawlerScheduler(DEVICE_ID, str(PLATFORM_ELE))
    scheduler.start()
    time.sleep(2)

    print(f"饿了么闪购采集器启动，设备: {DEVICE_ID}，等待间隔 {LOOP_INTERVAL_MIN}~{LOOP_INTERVAL_MAX} 秒随机")
    while True:
        try:
            wait_until_start_hour()   # 早于6点不领任务（心跳线程保持在线）
            # 先带设备id发一次心跳（注册/激活账号），再领任务（PDD 同款做法）
            requests.post(
                scheduler.heartbeat_url,
                json={"platform": scheduler.platform, "username": scheduler.username},
                headers=API_HEADERS,
                timeout=5,
                verify=False,
            )
            task = scheduler.get_task()
            if task is None:
                _iv = loop_interval()
                print(f"当前没有可执行的任务，{_iv} 秒后重试...")
                time.sleep(_iv)
                continue

            print(f"获取到任务: task_id={task.get('id')}, product={task.get('product_name')}")
            task['_device_row'] = DEVICE_ID
            task['_device_id'] = DEVICE_ID

            # 限额探针：执行前先发一次进度回告（页码=调度给的接力页，crawled=0）。
            # 平台返回 error（如今日限额已到/任务已释放）→ 直接跳过本轮，
            # 省掉开App/搜索/跨设备接力滑页的无用功；返回 success 则照常执行。
            scheduler.limit_reached = False
            scheduler.limit_msg = ""
            scheduler.post_report({
                "task_id": task.get('id'),
                "platform": scheduler.platform,
                "username": scheduler.username,
                "is_finished": 0,
                "need_reassign": 0,
                "current_page": task.get('current_page') or 0,
                "crawled_count": 0,
            })
            if scheduler.limit_reached:
                print(f"[探针] 平台仍限额（{scheduler.limit_msg}），本轮跳过执行")
                time.sleep(loop_interval())
                continue

            run_task_worker(build_task_payload_from_api(task, DEVICE_ID), scheduler)

            _iv = loop_interval()
            print(f"任务完成，等待 {_iv} 秒后开始下一轮...")
            time.sleep(_iv)
        except Exception as e:
            print(f"本轮任务异常: {e}")
            err_log.log_error(DEVICE_ID, "loop_exception", exc=e)
            time.sleep(loop_interval())


if __name__ == '__main__':
    main()
    # d = uiautomator2.connect("RG5LFYT8UKK7BI95")
    # d.screenshot('screenshot.png')