""" 小红书 (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()