project/xhs/start_run_xhs.py

"""
小红书 (XHS) 平台采集入口
调度、数据库、飞书通知全部走 commons。
"""
import os
import re
import sys
import random
import time
import subprocess
from datetime import datetime

# xhs main.py 内部 import xhs_config 需要 spiders/xhs 在 path 里
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "spiders", "xhs"))

from commons.scheduler import CrawlerScheduler
from commons.Logger import get_spider_logger
from commons.feishu_webhook import send_text, send_error_card
from commons.conn_mysql import MySQLPoolOn2
from spiders.xhs.main import XHS, XHS_PLATFORM_ID

logger = get_spider_logger("xhs_runner")

PLATFORM_NAME = "小红书"
IDLE_SECONDS_MIN = 30
IDLE_SECONDS_MAX = 90


def get_local_adb_devices():
    """获取当前电脑连接的所有 ADB 设备序列号"""
    try:
        output = subprocess.check_output(['adb', 'devices'], stderr=subprocess.STDOUT).decode('utf-8', errors='ignore')
        serials = []
        for line in output.splitlines():
            parts = line.strip().split()
            if len(parts) == 2 and parts[1] == 'device':
                serials.append(parts[0])
        return serials
    except Exception:
        return []

def get_device_from_db():
    """从 DB 拿一个小红书空闲设备，返回 (username, device_id)
    如果发现有设备卡在 status=1 但物理连接在当前电脑上，自动将其重置为 0
    """
    db = MySQLPoolOn2()
    
    # 1. 优先尝试直接获取空闲设备
    rows = db.select_data(
        "SELECT * FROM retrieve_collect_equipment_account WHERE platform = %s AND status = 0 ORDER BY id ASC",
        (XHS_PLATFORM_ID,),
    )
    if rows:
        row = rows[0]
        values = list(row.values()) if isinstance(row, dict) else row
        username = values[4] if len(values) > 4 else None
        device_id = values[11] if len(values) > 11 else username
        return username, device_id

    # 2. 如果没有空闲设备，查找是否由于断电/断网导致设备状态卡在 1（死锁）
    stuck_rows = db.select_data(
        "SELECT * FROM retrieve_collect_equipment_account WHERE platform = %s AND status = 1 ORDER BY id ASC",
        (XHS_PLATFORM_ID,),
    )
    if stuck_rows:
        local_serials = get_local_adb_devices()
        for row in stuck_rows:
            values = list(row.values()) if isinstance(row, dict) else row
            username = values[4] if len(values) > 4 else None
            device_id = values[11] if len(values) > 11 else username
            
            # 如果卡死的设备确实插在这台电脑上，说明是前一次运行崩溃留下的僵尸状态
            if device_id and device_id in local_serials:
                logger.warning(f"发现本地设备 {username}({device_id}) 状态异常卡在 1 (运行中)！")
                logger.warning("正在自动将其强制重置为 0 (空闲) 以恢复运行...")
                try:
                    db.update_data(
                        "UPDATE retrieve_collect_equipment_account SET status = 0 WHERE platform = %s AND account_number = %s",
                        (XHS_PLATFORM_ID, username)
                    )
                    logger.info("重置成功！已恢复该设备状态。")
                    return username, device_id
                except Exception as e:
                    logger.error(f"尝试自动重置设备死锁失败: {e}")
                    
    return None, None


def notify_result(task, crawl_count, success):
    """发送飞书通知"""
    try:
        drug_name = task.get("search_key", task.get("product_name", ""))
        notice_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        text_msg = (
            f"{notice_time} 通知:\n"
            f"平台: {PLATFORM_NAME}, 关键词: {drug_name}, 爬取数据: {crawl_count}条"
        )
        if success:
            send_text(text_msg)
        else:
            send_error_card(
                task_name=drug_name,
                err_msg=f"task_id={task.get('task_id')}, crawl_count={crawl_count}",
                mention_all=False,
            )
    except Exception as e:
        logger.warning("飞书通知发送失败：%s", e)


def update_device_status(username, platform_id, new_status):
    """更新设备状态：0=空闲 1=运行中"""
    try:
        db = MySQLPoolOn2()
        sql = "UPDATE retrieve_collect_equipment_account SET status = %s WHERE username = %s AND platform = %s"
        db.update_data(sql, (new_status, username, platform_id))
        logger.info(f"==== 已将设备 {username} 的状态修改为 {new_status} ====")
    except Exception as e:
        logger.error(f"更新设备状态失败: {e}")


def main_loop():
    """循环拉取任务并执行采集"""
    username, device_id = get_device_from_db()
    if not username:
        logger.error("小红书 无可用设备，退出")
        return

    # 【1】获取到空闲设备后，立刻把数据库里的状态改成 1（运行中）
    update_device_status(username, XHS_PLATFORM_ID, 1)

    try:
        scheduler = CrawlerScheduler(username, str(XHS_PLATFORM_ID))
        idle_seconds = random.randint(IDLE_SECONDS_MIN, IDLE_SECONDS_MAX)
        logger.info("小红书采集启动，设备=%s adb=%s 每轮间隔 %s 秒", username, device_id, idle_seconds)

        while True:
            task_id = None
            try:
                scheduler.start()
                task = scheduler.get_task() or {}
                if not task:
                    logger.info("%s 暂无任务，等待 %s 秒后重试", PLATFORM_NAME, idle_seconds)
                    time.sleep(idle_seconds)
                    continue

                task_id = task.get("id")
                # API 字段映射（模仿 PDD build_task_payload_from_api）
                brand = task.get("product_brand", "") or ""
                title_key = task.get("product_name", "") or ""
                spec_raw = task.get("product_specs", "") or ""
                spec_list = [s.strip() for s in re.split(r'[|、，,\n\r]+', spec_raw) if s.strip()]
                search_key = f"{brand}{title_key} {spec_raw}".strip()
                save_search_key = task.get("product_keyword") or search_key
                logger.info("开始执行%s采集任务 task_id=%s keyword=%s", PLATFORM_NAME, task_id, search_key)

                xhs = XHS(
                    search_key=search_key,
                    device_id=device_id,
                    title_key=title_key,
                    spec_list=spec_list,
                    brand=brand,
                    save_search_key=save_search_key,
                    start_page=0,  # 取消恢复历史进度，取到任务就从第一个商品开始
                    end_page=task.get("end_page"),
                    max_counts_limit=task.get("max_counts_limit"),
                    collect_config_info=task.get("collect_config_info", ""),
                    sort=task.get("sort"),
                    platform=XHS_PLATFORM_ID,
                    task_id=task_id,
                    enterprise_id=task.get("company_id"),
                    collect_round=task.get("collect_round"),
                    collect_equipment_account_id=task.get("collect_equipment_account_id"),
                    collect_region_id=task.get("collect_region_id"),
                    username=username,
                )
                completed = xhs.main()
                crawl_count = getattr(xhs, 'max_counts', 0)
                logger.info("%s采集任务完成 task_id=%s count=%s success=%s",
                            PLATFORM_NAME, task_id, crawl_count, completed)

                notify_result(task, crawl_count, completed)

            except Exception as e:
                logger.error("%s 采集任务执行失败: %s", PLATFORM_NAME, e, exc_info=True)

            time.sleep(idle_seconds)

    except KeyboardInterrupt:
        # 【2】专门捕获手动断开 (如在终端按下 Ctrl+C 等操作)
        logger.info("检测到程序被手动停止(Ctrl+C)，准备释放设备...")
    except Exception as e:
        # 捕获其他意料之外的严重崩溃
        logger.error("主程序发生崩溃异常: %s", e)
    finally:
        # 【3】不管是因为报错退出、还是人为强行断开，这里都会尽最大努力执行，改回 0
        # 增加重试机制，防止断网瞬间改不回去
        for attempt in range(3):
            try:
                update_device_status(username, XHS_PLATFORM_ID, 0)
                break
            except Exception as e:
                logger.error(f"释放设备状态失败 (尝试 {attempt+1}/3): {e}")
                time.sleep(2)
        logger.info("设备已安全释放，状态恢复为 0。退出程序。")



if __name__ == '__main__':
    main_loop()