|
|
@@ -0,0 +1,2112 @@
|
|
|
+# coding=utf-8
|
|
|
+"""
|
|
|
+小红书商品采集器
|
|
|
+==============
|
|
|
+调度/多线程/数据库框架模仿 PDD 模式,
|
|
|
+XHS 特有逻辑(弹窗检测、页面导航、数据提取)在本文件。
|
|
|
+
|
|
|
+使用方式:
|
|
|
+ 1. 配置下方变量区
|
|
|
+ 2. python main.py
|
|
|
+"""
|
|
|
+
|
|
|
+import sys
|
|
|
+import os
|
|
|
+from area_info.city_name_to_id import get_city
|
|
|
+import builtins
|
|
|
+import time
|
|
|
+import re
|
|
|
+import random
|
|
|
+import subprocess
|
|
|
+import datetime
|
|
|
+import threading
|
|
|
+import logging
|
|
|
+import json
|
|
|
+
|
|
|
+import requests
|
|
|
+import pymysql
|
|
|
+import uiautomator2 as u2
|
|
|
+import oss2
|
|
|
+import cv2
|
|
|
+import uuid
|
|
|
+from PIL import Image
|
|
|
+from pathlib import Path
|
|
|
+
|
|
|
+import xhs_config
|
|
|
+
|
|
|
+from commons.Logger import get_spider_logger
|
|
|
+from commons.feishu_webhook import send_text, send_error_card
|
|
|
+from commons.scheduler import CrawlerScheduler
|
|
|
+
|
|
|
+# 初始化统一日志(控制台 + logs/xhs_new2.log)
|
|
|
+_xhs_logger = get_spider_logger("xhs")
|
|
|
+logging.root.handlers = _xhs_logger.handlers
|
|
|
+logging.root.setLevel(_xhs_logger.level)
|
|
|
+
|
|
|
+# 保留原 print 的调用方式,但统一写入日志,避免到处翻控制台输出。
|
|
|
+_ORIGINAL_PRINT = builtins.print
|
|
|
+
|
|
|
+
|
|
|
+def print(*args, sep=" ", end="\n", file=None, flush=False):
|
|
|
+ message = sep.join(str(arg) for arg in args)
|
|
|
+ if end and end != "\n":
|
|
|
+ message = f"{message}{end}"
|
|
|
+ logged = False
|
|
|
+ try:
|
|
|
+ root_logger = logging.getLogger()
|
|
|
+ if root_logger.handlers:
|
|
|
+ root_logger.info(message.rstrip("\n"))
|
|
|
+ logged = True
|
|
|
+ except Exception:
|
|
|
+ logged = False
|
|
|
+ if not logged:
|
|
|
+ _ORIGINAL_PRINT(*args, sep=sep, end=end, file=file if file else sys.stdout, flush=flush)
|
|
|
+ if file not in (None, sys.stdout, sys.stderr):
|
|
|
+ _ORIGINAL_PRINT(*args, sep=sep, end=end, file=file, flush=flush)
|
|
|
+
|
|
|
+
|
|
|
+# 数据库连接池(与 PDD 同库:120.24.26.108:3307/drug_retrieve)
|
|
|
+from dbutils.pooled_db import PooledDB
|
|
|
+
|
|
|
+_db_pool = PooledDB(
|
|
|
+ creator=pymysql,
|
|
|
+ maxconnections=10,
|
|
|
+ mincached=2,
|
|
|
+ maxcached=5,
|
|
|
+ blocking=True,
|
|
|
+ charset="utf8mb4",
|
|
|
+ host="120.24.26.108",
|
|
|
+ port=3307,
|
|
|
+ user="root",
|
|
|
+ password="zhijiayun123456",
|
|
|
+ database="drug_retrieve",
|
|
|
+)
|
|
|
+
|
|
|
+
|
|
|
+def get_mysql():
|
|
|
+ """从连接池获取数据库连接(兼容旧接口)。调用方负责 conn.close() 归还连接。"""
|
|
|
+ return _db_pool.connection()
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================
|
|
|
+# 常量
|
|
|
+# ============================================================
|
|
|
+HEARTBEAT_INTERVAL_SECONDS = 60
|
|
|
+SCHEDULER_INTERVAL_SECONDS = 600
|
|
|
+MANUAL_SCHEDULER_INTERVAL_SECONDS = 300
|
|
|
+TASK_REPORT_URL = "http://pricesys.kailin.com.cn:8083/api/collect_task/report"
|
|
|
+TASK_HEARTBEAT_URL = "http://pricesys.kailin.com.cn:8083/api/collect_task/heartbeat"
|
|
|
+TASK_API_HEADERS = {'X-Crawler-Token': 'zhijiayun_crawler_2026'}
|
|
|
+XHS_PLATFORM_ID = 15
|
|
|
+XHS_PACKAGE = "com.xingin.xhs"
|
|
|
+TABLE_NAME = "retrieve_scrape_data"
|
|
|
+SHOP_TABLE_NAME = "retrieve_scrape_shop_info"
|
|
|
+TASK_STATUS_PENDING = 1
|
|
|
+DEVICE_STATUS_IDLE = 0
|
|
|
+DEFAULT_MAX_COUNTS_LIMIT = 500
|
|
|
+_DEFAULT_PATH = Path(__file__).parent.parent / "pdd" / "city.json"
|
|
|
+
|
|
|
+# 手动任务模式开关(True 时忽略数据库,任务和设备完全自己定)
|
|
|
+USE_MANUAL_TASKS = False
|
|
|
+MANUAL_TASKS = [
|
|
|
+ {
|
|
|
+ "device_id": "XOYPOZDADU79VGVG", # 必填:设备序列号
|
|
|
+ "search_key": "999 感冒灵胶囊 12",
|
|
|
+ "title_key": "感冒灵胶囊",
|
|
|
+ "spec_list": "12",
|
|
|
+ "brand": "999",
|
|
|
+ "save_search_key": "感冒灵胶囊",
|
|
|
+ "enterprise_id": 8,
|
|
|
+ "start_page": 0,
|
|
|
+ "end_page": 50,
|
|
|
+ "max_counts_limit": 500,
|
|
|
+ "sort": "默认",
|
|
|
+ "collect_round": 1,
|
|
|
+ "collect_equipment_account_id": 1,
|
|
|
+ "collect_region_id": 1,
|
|
|
+ },
|
|
|
+]
|
|
|
+
|
|
|
+# 内存占用状态(与 PDD 模式对齐)
|
|
|
+dispatch_lock = threading.Lock()
|
|
|
+running_task_ids = set()
|
|
|
+running_device_ids = set()
|
|
|
+worker_threads = {}
|
|
|
+scheduler_stop_event = threading.Event()
|
|
|
+scheduler_timer = None
|
|
|
+manual_dispatched_task_ids = set()
|
|
|
+
|
|
|
+# ============================================================
|
|
|
+# 工具函数
|
|
|
+# ============================================================
|
|
|
+
|
|
|
+def parse_optional_int(value, default=None):
|
|
|
+ if value in (None, ""):
|
|
|
+ return default
|
|
|
+ try:
|
|
|
+ return int(value)
|
|
|
+ except (TypeError, ValueError):
|
|
|
+ return default
|
|
|
+
|
|
|
+
|
|
|
+def get_adb_device_status_map():
|
|
|
+ try:
|
|
|
+ result = subprocess.run(
|
|
|
+ ["adb", "devices"],
|
|
|
+ capture_output=True,
|
|
|
+ text=True,
|
|
|
+ timeout=10,
|
|
|
+ )
|
|
|
+ except Exception as e:
|
|
|
+ logging.exception(f"读取 ADB 设备状态失败: {e}")
|
|
|
+ return {}
|
|
|
+ status_map = {}
|
|
|
+ for line in result.stdout.splitlines():
|
|
|
+ line = line.strip()
|
|
|
+ if not line or line.startswith("List of devices"):
|
|
|
+ continue
|
|
|
+ parts = line.split()
|
|
|
+ if len(parts) >= 2:
|
|
|
+ status_map[parts[0]] = parts[1]
|
|
|
+ return status_map
|
|
|
+
|
|
|
+
|
|
|
+def is_adb_device_online(device_id, adb_status_map=None):
|
|
|
+ device_id = str(device_id or "").strip()
|
|
|
+ if not device_id:
|
|
|
+ return False
|
|
|
+ if adb_status_map is None:
|
|
|
+ adb_status_map = get_adb_device_status_map()
|
|
|
+ return adb_status_map.get(device_id) == "device"
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================
|
|
|
+# API 上报
|
|
|
+# ============================================================
|
|
|
+
|
|
|
+def report_api(task_id, platform, username, page=0, is_finished=0,
|
|
|
+ crawled_count=0, total_pages=0):
|
|
|
+ """向调度中心上报任务状态(完全对齐 PDD _send_report 格式)。"""
|
|
|
+ if USE_MANUAL_TASKS:
|
|
|
+ logging.info(f"手动任务模式:跳过接口上报 task_id={task_id}")
|
|
|
+ return
|
|
|
+
|
|
|
+ params = {
|
|
|
+ "task_id": task_id,
|
|
|
+ "platform": str(platform),
|
|
|
+ "username": str(username) if username else "",
|
|
|
+ "current_page": page,
|
|
|
+ "total_pages": total_pages,
|
|
|
+ "crawled_count": crawled_count,
|
|
|
+ "is_finished": is_finished,
|
|
|
+ }
|
|
|
+ print(params)
|
|
|
+ try:
|
|
|
+ res = requests.post(TASK_REPORT_URL, json=params, headers=TASK_API_HEADERS, timeout=20)
|
|
|
+ print(res.text)
|
|
|
+ except Exception as e:
|
|
|
+ logging.exception(f"上报接口异常 task_id={task_id}: {e}")
|
|
|
+
|
|
|
+
|
|
|
+def report_heartbeat(task_id, page=None, device_id=None):
|
|
|
+ """上报任务心跳"""
|
|
|
+ if USE_MANUAL_TASKS:
|
|
|
+ return
|
|
|
+
|
|
|
+ params = {
|
|
|
+ "collect_task_allocate_id": task_id,
|
|
|
+ "heartbeat_time": int(time.time()),
|
|
|
+ "current_page": page if page is not None else 0,
|
|
|
+ "device_id": device_id if device_id else "",
|
|
|
+ }
|
|
|
+
|
|
|
+ try:
|
|
|
+ requests.post(TASK_HEARTBEAT_URL, json=params, headers=TASK_API_HEADERS, timeout=10)
|
|
|
+ logging.debug(f"心跳上报 task_id={task_id}, page={page}")
|
|
|
+ except Exception as e:
|
|
|
+ logging.warning(f"心跳上报失败: {e}")
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================
|
|
|
+# 任务 / 设备拉取
|
|
|
+# ============================================================
|
|
|
+
|
|
|
+def fetch_idle_xhs_devices():
|
|
|
+ """读取数据库里所有空闲的小红书设备;真实在线状态由 ADB 再二次判断。"""
|
|
|
+ conn = None
|
|
|
+ try:
|
|
|
+ conn = get_mysql()
|
|
|
+ with conn.cursor() as cursor:
|
|
|
+ sql = """
|
|
|
+ SELECT *
|
|
|
+ FROM retrieve_collect_equipment_account
|
|
|
+ WHERE platform = %s AND status = %s
|
|
|
+ ORDER BY id ASC
|
|
|
+ """
|
|
|
+ cursor.execute(sql, (XHS_PLATFORM_ID, DEVICE_STATUS_IDLE))
|
|
|
+ return cursor.fetchall()
|
|
|
+ except Exception as e:
|
|
|
+ logging.exception(f"读取空闲 XHS 设备失败: {e}")
|
|
|
+ return []
|
|
|
+ finally:
|
|
|
+ if conn:
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+
|
|
|
+def fetch_manual_task_payloads():
|
|
|
+ payloads = []
|
|
|
+ adb_status_map = get_adb_device_status_map()
|
|
|
+ for idx, task in enumerate(MANUAL_TASKS):
|
|
|
+ task_id = parse_optional_int(task.get("task_id"), 900000 + idx + 1)
|
|
|
+ if task_id in manual_dispatched_task_ids:
|
|
|
+ continue
|
|
|
+
|
|
|
+ search_key = str(task.get("search_key", "")).strip()
|
|
|
+ if not search_key:
|
|
|
+ logging.warning(f"跳过手动任务,缺少 search_key: {task}")
|
|
|
+ continue
|
|
|
+
|
|
|
+ device_id = str(task.get("device_id", "")).strip()
|
|
|
+ if not device_id:
|
|
|
+ logging.warning(f"跳过手动任务 {task_id},缺少 device_id")
|
|
|
+ continue
|
|
|
+
|
|
|
+ if not is_adb_device_online(device_id, adb_status_map):
|
|
|
+ logging.info(
|
|
|
+ f"手动任务 {task_id} 设备 {device_id} ADB 不在线,"
|
|
|
+ f"status={adb_status_map.get(device_id, 'missing')}"
|
|
|
+ )
|
|
|
+ continue
|
|
|
+
|
|
|
+ with dispatch_lock:
|
|
|
+ if device_id in running_device_ids:
|
|
|
+ logging.info(f"手动任务 {task_id} 设备 {device_id} 已被占用,本轮跳过")
|
|
|
+ continue
|
|
|
+ running_task_ids.add(task_id)
|
|
|
+ running_device_ids.add(device_id)
|
|
|
+
|
|
|
+ title_key = task.get("title_key")
|
|
|
+ save_search_key = task.get("save_search_key")
|
|
|
+ payloads.append({
|
|
|
+ "task_id": task_id,
|
|
|
+ "enterprise_id": parse_optional_int(task.get("enterprise_id"), 0),
|
|
|
+ "platform": XHS_PLATFORM_ID,
|
|
|
+ "title_key": title_key if title_key not in (None, "") else search_key,
|
|
|
+ "spec_list": task.get("spec_list", ""),
|
|
|
+ "brand": task.get("brand", ""),
|
|
|
+ "search_key": search_key,
|
|
|
+ "save_search_key": save_search_key if save_search_key not in (None, "") else search_key,
|
|
|
+ "start_page": parse_optional_int(task.get("start_page"), 0),
|
|
|
+ "end_page": parse_optional_int(task.get("end_page"), None),
|
|
|
+ "max_counts_limit": parse_optional_int(task.get("max_counts_limit"), DEFAULT_MAX_COUNTS_LIMIT),
|
|
|
+ "collect_config_info": task.get("collect_config_info", ""),
|
|
|
+ "collect_equipment_account_id": task.get("collect_equipment_account_id", ""),
|
|
|
+ "collect_region_id": task.get("collect_region_id", ""),
|
|
|
+ "collect_round": task.get("collect_round", ""),
|
|
|
+ "sort": task.get("sort", "默认"),
|
|
|
+ "device_id": device_id,
|
|
|
+ })
|
|
|
+ manual_dispatched_task_ids.add(task_id)
|
|
|
+
|
|
|
+ if payloads:
|
|
|
+ logging.info(f"手动任务模式启用,本轮分发 {len(payloads)} 条任务")
|
|
|
+ else:
|
|
|
+ logging.info("手动任务模式启用,但没有可分发的新任务")
|
|
|
+ return payloads
|
|
|
+
|
|
|
+
|
|
|
+def fetch_runnable_task_payloads():
|
|
|
+ """获取本轮可执行的任务 payload 列表。"""
|
|
|
+ if USE_MANUAL_TASKS:
|
|
|
+ logging.info("手动任务模式启用,跳过数据库任务分配")
|
|
|
+ return fetch_manual_task_payloads()
|
|
|
+
|
|
|
+ # 数据库模式(暂未启用,保留入口)
|
|
|
+ logging.info("数据库模式:当前没有待执行任务")
|
|
|
+ return []
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================
|
|
|
+# Worker 生命周期
|
|
|
+# ============================================================
|
|
|
+
|
|
|
+def cleanup_finished_workers():
|
|
|
+ with dispatch_lock:
|
|
|
+ dead_device_ids = [
|
|
|
+ device_id
|
|
|
+ for device_id, thread in worker_threads.items()
|
|
|
+ if not thread.is_alive()
|
|
|
+ ]
|
|
|
+ for device_id in dead_device_ids:
|
|
|
+ worker_threads.pop(device_id, None)
|
|
|
+ running_device_ids.discard(device_id)
|
|
|
+ orphan_device_ids = [
|
|
|
+ device_id
|
|
|
+ for device_id in running_device_ids
|
|
|
+ if device_id not in worker_threads
|
|
|
+ ]
|
|
|
+ for device_id in orphan_device_ids:
|
|
|
+ running_device_ids.discard(device_id)
|
|
|
+ if not worker_threads and running_task_ids:
|
|
|
+ logging.info(
|
|
|
+ f"cleanup: 无存活 worker,清空残留 running_task_ids: {running_task_ids}"
|
|
|
+ )
|
|
|
+ running_task_ids.clear()
|
|
|
+
|
|
|
+
|
|
|
+def has_active_workers():
|
|
|
+ cleanup_finished_workers()
|
|
|
+ with dispatch_lock:
|
|
|
+ return any(thread.is_alive() for thread in worker_threads.values())
|
|
|
+
|
|
|
+
|
|
|
+def all_manual_tasks_dispatched():
|
|
|
+ expected_task_ids = {
|
|
|
+ parse_optional_int(task.get("task_id"), 900000 + idx + 1)
|
|
|
+ for idx, task in enumerate(MANUAL_TASKS)
|
|
|
+ }
|
|
|
+ return expected_task_ids.issubset(manual_dispatched_task_ids)
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================
|
|
|
+# 线程入口
|
|
|
+# ============================================================
|
|
|
+
|
|
|
+def run_task_worker(task_payload):
|
|
|
+ """单个任务线程的主入口:通过 CrawlerScheduler 维持心跳,驱动 XHS 采集并上报。"""
|
|
|
+ task_id = task_payload["task_id"]
|
|
|
+ device_id = task_payload["device_id"]
|
|
|
+ username = task_payload.get("username", device_id)
|
|
|
+ search_key = str(task_payload.get("search_key", "")).strip()
|
|
|
+ xhs = None
|
|
|
+ should_send_finish_notice = False
|
|
|
+
|
|
|
+ # 每个设备线程拥有自己的 CrawlerScheduler 实例,负责心跳 + 上报
|
|
|
+ scheduler = CrawlerScheduler(username, str(XHS_PLATFORM_ID))
|
|
|
+ scheduler.start() # 启动后台心跳线程
|
|
|
+
|
|
|
+ try:
|
|
|
+ logging.info(f"[任务 {task_id}] 开始执行,设备: {device_id}")
|
|
|
+ print(task_payload)
|
|
|
+ xhs = XHS(
|
|
|
+ task_payload["search_key"],
|
|
|
+ device_id,
|
|
|
+ title_key=task_payload.get("title_key"),
|
|
|
+ spec_list=task_payload.get("spec_list"),
|
|
|
+ brand=task_payload.get("brand", ""),
|
|
|
+ save_search_key=task_payload.get("save_search_key"),
|
|
|
+ start_page=task_payload.get("start_page"),
|
|
|
+ end_page=task_payload.get("end_page"),
|
|
|
+ max_counts_limit=task_payload.get("max_counts_limit"),
|
|
|
+ collect_config_info=task_payload.get("collect_config_info", ""),
|
|
|
+ sort=task_payload.get("sort"),
|
|
|
+ platform=task_payload.get("platform"),
|
|
|
+ task_id=task_payload.get("task_id"),
|
|
|
+ enterprise_id=task_payload.get("enterprise_id"),
|
|
|
+ collect_round=task_payload.get("collect_round"),
|
|
|
+ collect_equipment_account_id=task_payload.get("collect_equipment_account_id", 1),
|
|
|
+ collect_region_id=task_payload.get("collect_region_id", 1),
|
|
|
+ username=task_payload.get("username"),
|
|
|
+ )
|
|
|
+ completed_normally = xhs.main()
|
|
|
+ if completed_normally:
|
|
|
+ logging.info(f"[任务 {task_id}] 执行完成,设备: {device_id}")
|
|
|
+ else:
|
|
|
+ logging.info(f"[任务 {task_id}] 已结束,设备: {device_id}")
|
|
|
+ should_send_finish_notice = True
|
|
|
+ except Exception as e:
|
|
|
+ end_page = task_payload.get("start_page")
|
|
|
+ err_msg = str(e)
|
|
|
+ if xhs is not None:
|
|
|
+ end_page = getattr(xhs, "page", end_page)
|
|
|
+ xhs.finish_task_abnormally(end_page, f"任务执行异常: {err_msg}")
|
|
|
+ else:
|
|
|
+ report_api(task_id, platform=XHS_PLATFORM_ID, username="unknown",
|
|
|
+ page=end_page, is_finished=0)
|
|
|
+ logging.exception(f"[任务 {task_id}] 执行异常,设备: {device_id},错误: {err_msg}")
|
|
|
+ send_error_card(
|
|
|
+ task_name=f"XHS任务(task_id={task_id}, device={device_id}, key={search_key})",
|
|
|
+ err_msg=err_msg,
|
|
|
+ mention_all=False,
|
|
|
+ )
|
|
|
+ should_send_finish_notice = True
|
|
|
+ finally:
|
|
|
+ scheduler.stop() # 停止心跳
|
|
|
+ # 释放内存占用标记
|
|
|
+ with dispatch_lock:
|
|
|
+ running_device_ids.discard(device_id)
|
|
|
+ worker_threads.pop(device_id, None)
|
|
|
+ if should_send_finish_notice:
|
|
|
+ scraped_count = getattr(xhs, "max_counts", 0) if xhs is not None else 0
|
|
|
+ notify_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
+ drug_name = str(task_payload.get("search_key", "")).strip()
|
|
|
+ send_text(
|
|
|
+ f"{notify_time} 通知:\n"
|
|
|
+ f"平台: 小红书, 药品: {drug_name}, 爬取数据: {scraped_count}条"
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================
|
|
|
+# 派单
|
|
|
+# ============================================================
|
|
|
+
|
|
|
+def dispatch_pending_tasks():
|
|
|
+ """执行一轮派单,把每个可运行任务绑定到对应设备线程。"""
|
|
|
+ cleanup_finished_workers()
|
|
|
+ task_payloads = fetch_runnable_task_payloads()
|
|
|
+ if not task_payloads:
|
|
|
+ return
|
|
|
+
|
|
|
+ for task_payload in task_payloads:
|
|
|
+ device_id = task_payload["device_id"]
|
|
|
+ try:
|
|
|
+ thread = threading.Thread(
|
|
|
+ target=run_task_worker,
|
|
|
+ args=(task_payload,),
|
|
|
+ daemon=True,
|
|
|
+ name=f"xhs-{device_id}",
|
|
|
+ )
|
|
|
+ with dispatch_lock:
|
|
|
+ worker_threads[device_id] = thread
|
|
|
+ thread.start()
|
|
|
+ logging.info(f"[任务 {task_payload['task_id']}] 已分发到设备 {device_id}")
|
|
|
+ except Exception:
|
|
|
+ with dispatch_lock:
|
|
|
+ running_task_ids.discard(task_payload["task_id"])
|
|
|
+ running_device_ids.discard(device_id)
|
|
|
+ worker_threads.pop(device_id, None)
|
|
|
+ raise
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================
|
|
|
+# 定时调度
|
|
|
+# ============================================================
|
|
|
+
|
|
|
+def schedule_dispatch(delay_seconds=SCHEDULER_INTERVAL_SECONDS, job_func=None):
|
|
|
+ """注册下一轮调度定时器。"""
|
|
|
+ global scheduler_timer
|
|
|
+ if scheduler_stop_event.is_set():
|
|
|
+ return
|
|
|
+ if job_func is None:
|
|
|
+ job_func = scheduled_dispatch_job
|
|
|
+ scheduler_timer = threading.Timer(delay_seconds, job_func)
|
|
|
+ scheduler_timer.daemon = False
|
|
|
+ scheduler_timer.name = "xhs-scheduler"
|
|
|
+ scheduler_timer.start()
|
|
|
+
|
|
|
+
|
|
|
+def scheduled_dispatch_job():
|
|
|
+ try:
|
|
|
+ dispatch_pending_tasks()
|
|
|
+ except Exception as e:
|
|
|
+ logging.exception(f"XHS 定时调度异常: {e}")
|
|
|
+ finally:
|
|
|
+ schedule_dispatch(SCHEDULER_INTERVAL_SECONDS)
|
|
|
+
|
|
|
+
|
|
|
+def manual_scheduled_dispatch_job():
|
|
|
+ try:
|
|
|
+ dispatch_pending_tasks()
|
|
|
+ except Exception as e:
|
|
|
+ logging.exception(f"XHS 手动任务定时调度异常: {e}")
|
|
|
+ finally:
|
|
|
+ if all_manual_tasks_dispatched() and not has_active_workers():
|
|
|
+ logging.info("手动任务模式:全部任务已完成,停止轮询并退出")
|
|
|
+ scheduler_stop_event.set()
|
|
|
+ return
|
|
|
+ schedule_dispatch(MANUAL_SCHEDULER_INTERVAL_SECONDS, manual_scheduled_dispatch_job)
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================
|
|
|
+# ====== 变量区结束 ======
|
|
|
+# ============================================================
|
|
|
+
|
|
|
+
|
|
|
+class XHS:
|
|
|
+ """小红书商品详情页采集器"""
|
|
|
+
|
|
|
+ SEP = "=" * 60
|
|
|
+ SLEEP_CLICK = 1.5
|
|
|
+ SLEEP_APP_START = 3
|
|
|
+ SLEEP_CLIPBOARD = 0.8
|
|
|
+
|
|
|
+ # 弹窗检测
|
|
|
+ POPUP_XPATHS = [
|
|
|
+ # 原有
|
|
|
+ '//*[@text="立即领取"]',
|
|
|
+ '//*[@text="送你 66周年庆礼券 啦 "]',
|
|
|
+ '//*[contains(@text, "可用")]',
|
|
|
+ # 优惠券弹窗
|
|
|
+ '//*[@text="恭喜领取成功 "]',
|
|
|
+ '//*[@text="立即使用"]',
|
|
|
+ '//*[@text="无门槛"]',
|
|
|
+ '//*[contains(@text, "后失效")]',
|
|
|
+ ]
|
|
|
+ POPUP_THRESHOLD = 3
|
|
|
+
|
|
|
+ # 列表页判定锚点
|
|
|
+ LIST_PAGE_ANCHORS = [
|
|
|
+ '//*[@text="搜索"]',
|
|
|
+ '//*[@text="全部"]',
|
|
|
+ '//*[@text="用户"]',
|
|
|
+ '//*[@text="商品"]',
|
|
|
+ '//*[@content-desc="全部删除"]',
|
|
|
+ ]
|
|
|
+ LIST_PAGE_HIT_THRESHOLD = 2
|
|
|
+
|
|
|
+ def __init__(self, search_key, device_id,
|
|
|
+ title_key=None, spec_list=None, brand="",
|
|
|
+ save_search_key=None, start_page=0, end_page=None,
|
|
|
+ max_counts_limit=None, collect_config_info="",
|
|
|
+ sort=None, platform=None, task_id=None,
|
|
|
+ enterprise_id=None, collect_round=None,
|
|
|
+ collect_equipment_account_id=None, collect_region_id=None,
|
|
|
+ username=None):
|
|
|
+ self.package_name = XHS_PACKAGE
|
|
|
+ self.device_id = device_id
|
|
|
+ self.d = None
|
|
|
+
|
|
|
+ # 任务参数
|
|
|
+ self.username = username # 调度系统账号名
|
|
|
+ self.enterprise_id = enterprise_id
|
|
|
+ self.task_id = task_id
|
|
|
+ self.platform = platform
|
|
|
+ self.sort = sort
|
|
|
+ self.search_key = str(search_key or "").strip()
|
|
|
+ self.title_key = title_key if title_key is not None else search_key
|
|
|
+ self.spec_list = self._normalize_rule_list(spec_list)
|
|
|
+ self.brand = str(brand or "").strip()
|
|
|
+ self.save_search_key = save_search_key or search_key
|
|
|
+ self.start_page = max(parse_optional_int(start_page, 0), 0)
|
|
|
+ self.end_page = max(parse_optional_int(end_page, 0), 0)
|
|
|
+ self.max_counts_limit = max_counts_limit
|
|
|
+ self.collect_config_info = collect_config_info or ""
|
|
|
+ self.collect_round = collect_round
|
|
|
+ self.collect_equipment_account_id = collect_equipment_account_id
|
|
|
+ self.collect_region_id = collect_region_id
|
|
|
+
|
|
|
+ # 运行状态
|
|
|
+ self.page = self.start_page
|
|
|
+ self.max_counts = 0
|
|
|
+ self.unrelated_data = 0
|
|
|
+ self.click_counts = 0
|
|
|
+ self.finish_reported = False
|
|
|
+ self.program_start_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
+ self._heartbeat_running = False
|
|
|
+
|
|
|
+ # 表名(使用 XHS 独立表)
|
|
|
+ self.table_name = TABLE_NAME
|
|
|
+ self.shop_table_name = SHOP_TABLE_NAME
|
|
|
+
|
|
|
+ self._city_lookup_cache = None
|
|
|
+
|
|
|
+ # OSS 快照(从 xhs_config 拿)
|
|
|
+ self.oss_config = {
|
|
|
+ "access_key_id": xhs_config.access_key_id,
|
|
|
+ "access_key_secret": xhs_config.access_key_secret,
|
|
|
+ "endpoint": xhs_config.endpoint,
|
|
|
+ "bucket_name": xhs_config.bucket_name,
|
|
|
+ "oss_prefix": xhs_config.oss_prefix,
|
|
|
+ }
|
|
|
+ self.address_region_index = self.build_address_region_index(_DEFAULT_PATH)
|
|
|
+
|
|
|
+ def _norm_region_name(self, name):
|
|
|
+ if name is None:
|
|
|
+ return ""
|
|
|
+ text = str(name).strip()
|
|
|
+ if not text:
|
|
|
+ return ""
|
|
|
+ return re.sub(r'(省|市|自治区|特别行政区)$', '', text)
|
|
|
+
|
|
|
+ def build_address_region_index(self, file_path):
|
|
|
+ pairs = []
|
|
|
+ try:
|
|
|
+ with open(file_path, 'r', encoding='utf-8') as f:
|
|
|
+ data = json.load(f)
|
|
|
+ for province in data:
|
|
|
+ p_name = str(province.get("name") or "").strip()
|
|
|
+ if not p_name:
|
|
|
+ continue
|
|
|
+ p_candidates = {p_name, self._norm_region_name(p_name)}
|
|
|
+ cities = province.get("sons") or []
|
|
|
+ if not cities:
|
|
|
+ pairs.append({
|
|
|
+ "province": p_name,
|
|
|
+ "city": "",
|
|
|
+ "province_candidates": p_candidates,
|
|
|
+ "city_candidates": set()
|
|
|
+ })
|
|
|
+ continue
|
|
|
+ for city in cities:
|
|
|
+ c_name = str(city.get("name") or "").strip()
|
|
|
+ c_candidates = {c_name, self._norm_region_name(c_name)} if c_name else set()
|
|
|
+ pairs.append({
|
|
|
+ "province": p_name,
|
|
|
+ "city": c_name,
|
|
|
+ "province_candidates": p_candidates,
|
|
|
+ "city_candidates": c_candidates
|
|
|
+ })
|
|
|
+ except Exception as e:
|
|
|
+ print(f"构建地址省市索引失败: {e}")
|
|
|
+ return pairs
|
|
|
+
|
|
|
+ def infer_region_from_address(self, address):
|
|
|
+ text = str(address or "").strip()
|
|
|
+ if not text:
|
|
|
+ return "", ""
|
|
|
+ for item in self.address_region_index:
|
|
|
+ if any(k and k in text for k in item["province_candidates"]):
|
|
|
+ city = item["city"]
|
|
|
+ if city and any(k and k in text for k in item["city_candidates"]):
|
|
|
+ return item["province"], city
|
|
|
+ for item in self.address_region_index:
|
|
|
+ if any(k and k in text for k in item["province_candidates"]):
|
|
|
+ return item["province"], item["city"]
|
|
|
+ return "", ""
|
|
|
+
|
|
|
+ def _load_city_lookup(self):
|
|
|
+ if self._city_lookup_cache is not None:
|
|
|
+ return self._city_lookup_cache
|
|
|
+
|
|
|
+ base_dir = os.path.dirname(os.path.abspath(__file__))
|
|
|
+ candidate_paths = [
|
|
|
+ os.path.join(base_dir, "..", "pdd", "city.json"),
|
|
|
+ ]
|
|
|
+
|
|
|
+ city_data = []
|
|
|
+ loaded_path = ""
|
|
|
+ for path in candidate_paths:
|
|
|
+ if not os.path.exists(path):
|
|
|
+ continue
|
|
|
+ try:
|
|
|
+ with open(path, "r", encoding="utf-8") as f:
|
|
|
+ candidate_data = json.load(f)
|
|
|
+ if isinstance(candidate_data, list):
|
|
|
+ city_data = candidate_data
|
|
|
+ loaded_path = path
|
|
|
+ break
|
|
|
+ except Exception as exc:
|
|
|
+ print(f"加载 city.json 失败: {path}, error={exc}")
|
|
|
+
|
|
|
+ province_to_id = {}
|
|
|
+ province_city_to_id = {}
|
|
|
+ global_city_matches = {}
|
|
|
+
|
|
|
+ for province in city_data:
|
|
|
+ province_norm = self._normalize_region_name(province.get("name", ""), "province")
|
|
|
+ province_id = parse_optional_int(province.get("id"), 0) or 0
|
|
|
+
|
|
|
+ if province_norm and province_id and province_norm not in province_to_id:
|
|
|
+ province_to_id[province_norm] = province_id
|
|
|
+
|
|
|
+ city_map = {}
|
|
|
+ for city in province.get("sons", []) or []:
|
|
|
+ city_norm = self._normalize_region_name(city.get("name", ""), "city")
|
|
|
+ city_id = parse_optional_int(city.get("id"), 0) or 0
|
|
|
+ if city_norm and city_id and city_norm not in city_map:
|
|
|
+ city_map[city_norm] = city_id
|
|
|
+ if city_norm and city_id and province_id:
|
|
|
+ global_city_matches.setdefault(city_norm, []).append((province_id, city_id, province_norm))
|
|
|
+
|
|
|
+ if province_norm and province_norm not in province_city_to_id:
|
|
|
+ province_city_to_id[province_norm] = city_map
|
|
|
+
|
|
|
+ if loaded_path:
|
|
|
+ print(f"已加载 city.json: {loaded_path}")
|
|
|
+ else:
|
|
|
+ print("未找到可用 city.json,省市ID将回填为 0")
|
|
|
+
|
|
|
+ self._city_lookup_cache = {
|
|
|
+ "province_to_id": province_to_id,
|
|
|
+ "province_city_to_id": province_city_to_id,
|
|
|
+ "global_city_matches": global_city_matches,
|
|
|
+ }
|
|
|
+ return self._city_lookup_cache
|
|
|
+
|
|
|
+ def _resolve_region_ids(self, province_name, city_name):
|
|
|
+ city_id, province_id, _, _ = get_city(city_name or province_name)
|
|
|
+ return province_id or 0, city_id or 0
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _normalize_region_name(name, level):
|
|
|
+ text = re.sub(r"\s+", "", str(name or ""))
|
|
|
+ text = text.replace(" ", "")
|
|
|
+ if not text:
|
|
|
+ return ""
|
|
|
+
|
|
|
+ if level == "province":
|
|
|
+ province_alias = {
|
|
|
+ "内蒙古自治区": "内蒙古",
|
|
|
+ "广西壮族自治区": "广西",
|
|
|
+ "宁夏回族自治区": "宁夏",
|
|
|
+ "新疆维吾尔自治区": "新疆",
|
|
|
+ "西藏自治区": "西藏",
|
|
|
+ "香港特别行政区": "香港",
|
|
|
+ "澳门特别行政区": "澳门",
|
|
|
+ }
|
|
|
+ if text in province_alias:
|
|
|
+ return province_alias[text]
|
|
|
+
|
|
|
+ for suffix in ("特别行政区", "维吾尔自治区", "回族自治区", "壮族自治区", "自治区", "省", "市"):
|
|
|
+ if text.endswith(suffix) and len(text) > len(suffix):
|
|
|
+ return text[:-len(suffix)]
|
|
|
+ return text
|
|
|
+
|
|
|
+ city_alias = {
|
|
|
+ "北京市": "北京",
|
|
|
+ "天津市": "天津",
|
|
|
+ "上海市": "上海",
|
|
|
+ "重庆市": "重庆",
|
|
|
+ }
|
|
|
+ if text in city_alias:
|
|
|
+ return city_alias[text]
|
|
|
+
|
|
|
+ for suffix in ("自治州", "地区", "盟", "州", "市"):
|
|
|
+ if text.endswith(suffix) and len(text) > len(suffix):
|
|
|
+ return text[:-len(suffix)]
|
|
|
+ return text
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 工具方法
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _normalize_rule_list(value):
|
|
|
+ if value is None:
|
|
|
+ return []
|
|
|
+ if isinstance(value, (list, tuple, set)):
|
|
|
+ raw_values = value
|
|
|
+ else:
|
|
|
+ raw_values = [value]
|
|
|
+ return [str(v).strip() for v in raw_values if str(v).strip()]
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _normalize_match_text(value):
|
|
|
+ return re.sub(r'\s+', '', str(value or '')).lower()
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def get_sleep_time():
|
|
|
+ return random.uniform(0.5, 1.0)
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def get_current_date():
|
|
|
+ return datetime.datetime.now().strftime('%Y/%m/%d')
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _strip_content_desc_prefix(text, prefix):
|
|
|
+ for sep in (",", ","):
|
|
|
+ text = text.replace(f"{prefix}{sep}", "")
|
|
|
+ return text.strip()
|
|
|
+
|
|
|
+ def _safe_get_attr(self, xpath, attr="text"):
|
|
|
+ try:
|
|
|
+ el = self.d.xpath(xpath)
|
|
|
+ if not el.exists:
|
|
|
+ return ""
|
|
|
+ if attr == "content_desc":
|
|
|
+ return (el.info.get('contentDescription') or "").strip()
|
|
|
+ return (el.text or "").strip()
|
|
|
+ except Exception:
|
|
|
+ return ""
|
|
|
+
|
|
|
+ def _scroll_until_found(self, xpath, max_swipes=5, direction="up", scale=0.3):
|
|
|
+ el = self.d.xpath(xpath)
|
|
|
+ if el.exists:
|
|
|
+ return el
|
|
|
+ for _ in range(max_swipes):
|
|
|
+ self.d.swipe_ext(direction, scale=scale)
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+ el = self.d.xpath(xpath)
|
|
|
+ if el.exists:
|
|
|
+ return el
|
|
|
+ return None
|
|
|
+
|
|
|
+ def _count_popup_elements(self):
|
|
|
+ return sum(1 for xp in self.POPUP_XPATHS if self.d.xpath(xp).exists)
|
|
|
+
|
|
|
+ def _dismiss_popup_if_exists(self):
|
|
|
+ count = self._count_popup_elements()
|
|
|
+ if count >= self.POPUP_THRESHOLD:
|
|
|
+ print(f" [弹窗检测] 命中 {count} 个特征,关闭弹窗")
|
|
|
+ self.d.press("back")
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+ return True
|
|
|
+ return False
|
|
|
+
|
|
|
+ def _find_with_popup_retry(self, find_func, step_name="", max_retries=2):
|
|
|
+ for attempt in range(max_retries):
|
|
|
+ result = find_func()
|
|
|
+ if result:
|
|
|
+ return result
|
|
|
+ if attempt < max_retries - 1:
|
|
|
+ print(f" [{step_name}] 未找到,关闭弹窗后重试 ({attempt + 1}/{max_retries - 1})")
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+ time.sleep(random.uniform(1.5, 2))
|
|
|
+ return None
|
|
|
+
|
|
|
+ def _match_any_keyword(self, text, keywords):
|
|
|
+ keyword_list = self._normalize_rule_list(keywords)
|
|
|
+ if not keyword_list:
|
|
|
+ return True
|
|
|
+ normalized = self._normalize_match_text(text)
|
|
|
+ return any(self._normalize_match_text(k) in normalized for k in keyword_list)
|
|
|
+
|
|
|
+ def is_on_list_page(self):
|
|
|
+ hits = sum(1 for xp in self.LIST_PAGE_ANCHORS if self.d.xpath(xp).exists)
|
|
|
+ return hits >= self.LIST_PAGE_HIT_THRESHOLD
|
|
|
+
|
|
|
+ def is_title_useful(self, title):
|
|
|
+ if self.title_key and not self._match_any_keyword(title, self.title_key):
|
|
|
+ print(f" 不匹配: 标题关键词「{self.title_key}」")
|
|
|
+ return False
|
|
|
+ if self.brand and not self._match_any_keyword(title, self.brand):
|
|
|
+ print(f" 不匹配: 品牌「{self.brand}」")
|
|
|
+ return False
|
|
|
+ if self.spec_list and not self._match_any_keyword(title, self.spec_list):
|
|
|
+ print(f" 不匹配: 规格「{self.spec_list}」")
|
|
|
+ return False
|
|
|
+ return True
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 设备连接 & App 控制
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def connect_device(self):
|
|
|
+ try:
|
|
|
+ self.d = u2.connect_usb(self.device_id)
|
|
|
+
|
|
|
+ # (已移除全局 ANR 监控,后台轮询会导致滑动时卡顿/抽搐,触发防跑封控)
|
|
|
+
|
|
|
+ self._restart_uiautomator_services()
|
|
|
+ print(f'[连接成功] 设备: {self.device_id}')
|
|
|
+ return True
|
|
|
+ except Exception as e:
|
|
|
+ print(f'[连接失败] {self.device_id}: {e}')
|
|
|
+ return False
|
|
|
+
|
|
|
+ def _restart_uiautomator_services(self):
|
|
|
+ stop_cmd = f'adb -s {self.device_id} shell /data/local/tmp/atx-agent server -d --stop'
|
|
|
+ start_cmd = f'adb -s {self.device_id} shell /data/local/tmp/atx-agent server -d'
|
|
|
+ subprocess.run(stop_cmd, capture_output=True, text=True, shell=True)
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+ subprocess.run(start_cmd, capture_output=True, text=True, shell=True)
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+
|
|
|
+ def start_app(self):
|
|
|
+ self.d.app_start(self.package_name)
|
|
|
+ time.sleep(self.SLEEP_APP_START)
|
|
|
+
|
|
|
+ def get_screen_size(self):
|
|
|
+ if not hasattr(self, '_cached_screen_w'):
|
|
|
+ info = self.d.info
|
|
|
+ self._cached_screen_w = info.get('displayWidth', 720)
|
|
|
+ self._cached_screen_h = info.get('displayHeight', 1640)
|
|
|
+ return self._cached_screen_w, self._cached_screen_h
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 页面导航
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def back_to_list_page(self, max_attempts=5):
|
|
|
+ for _ in range(max_attempts):
|
|
|
+ if self.is_on_list_page():
|
|
|
+ return True
|
|
|
+ self.d.press("back")
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+ return self.is_on_list_page()
|
|
|
+
|
|
|
+ def enter_search_page(self):
|
|
|
+ """首页 → 市集 → 搜索框 → 搜索 → 商品列表页"""
|
|
|
+ # 1. 点市集/购物/商城 (兼容不同版本的小红书)
|
|
|
+ el = None
|
|
|
+ for txt in ["购物", "市集", "商城"]:
|
|
|
+ tmp_el = self.d.xpath(f'//*[@text="{txt}"]')
|
|
|
+ if tmp_el.exists:
|
|
|
+ el = tmp_el
|
|
|
+ print(f"找到入口: {txt}")
|
|
|
+ break
|
|
|
+
|
|
|
+ if not el:
|
|
|
+ print("[错误] 未找到「市集/购物/商城」")
|
|
|
+ return False
|
|
|
+ el.click()
|
|
|
+ time.sleep(random.uniform(1.5, 2))
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+
|
|
|
+ # 2. 点搜索入口(弹窗遮挡时自动重试)
|
|
|
+ def _find_search_entry():
|
|
|
+ tvs = self.d.xpath('//android.widget.TextView').all()
|
|
|
+ for tv in tvs:
|
|
|
+ try:
|
|
|
+ b = tv.info.get('bounds', {})
|
|
|
+ if 150 <= b.get('top', 0) <= 180 and 200 <= b.get('bottom', 0) <= 250:
|
|
|
+ tv.click()
|
|
|
+ return True
|
|
|
+ except Exception:
|
|
|
+ continue
|
|
|
+ return False
|
|
|
+
|
|
|
+ if not self._find_with_popup_retry(_find_search_entry, step_name="搜索入口"):
|
|
|
+ print("[错误] 未找到搜索入口")
|
|
|
+ return False
|
|
|
+ time.sleep(random.uniform(1, 1.5))
|
|
|
+
|
|
|
+ # 3. 输入关键字
|
|
|
+ spec_str = " ".join(self.spec_list) if self.spec_list else ""
|
|
|
+ search_key = f"{self.brand} {self.title_key} {spec_str}".strip()
|
|
|
+ print(f" 搜索词: {search_key}")
|
|
|
+
|
|
|
+ def _find_edit_text():
|
|
|
+ edit = self.d.xpath('//android.widget.EditText')
|
|
|
+ if edit.exists:
|
|
|
+ edit.click()
|
|
|
+ return True
|
|
|
+ return False
|
|
|
+
|
|
|
+ if not self._find_with_popup_retry(_find_edit_text, step_name="搜索输入框"):
|
|
|
+ print("[错误] 未找到搜索输入框")
|
|
|
+ return False
|
|
|
+ time.sleep(0.5)
|
|
|
+ try:
|
|
|
+ self.d.send_keys(search_key, clear=True)
|
|
|
+ except Exception as e:
|
|
|
+ print(f" [警告] send_keys 失败 ({e}),尝试使用 set_text 兜底输入")
|
|
|
+ try:
|
|
|
+ # 兼容某些手机(或模拟器)上 uiautomator2 切换输入法报错的问题
|
|
|
+ self.d(className="android.widget.EditText").set_text(search_key)
|
|
|
+ except Exception as e2:
|
|
|
+ print(f" [错误] set_text 也失败: {e2}")
|
|
|
+ time.sleep(random.uniform(0.5, 1))
|
|
|
+
|
|
|
+ # 4. 点搜索按钮
|
|
|
+ def _find_search_btn():
|
|
|
+ btn = self.d.xpath('//*[@text="搜索"]')
|
|
|
+ if btn.exists:
|
|
|
+ btn.click()
|
|
|
+ return True
|
|
|
+ return False
|
|
|
+
|
|
|
+ if not self._find_with_popup_retry(_find_search_btn, step_name="搜索按钮"):
|
|
|
+ print("[错误] 未找到搜索按钮")
|
|
|
+ return False
|
|
|
+ time.sleep(random.uniform(1.5, 2))
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+
|
|
|
+ print(" 已进入商品列表页")
|
|
|
+ return True
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 商品卡片识别
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def get_product_cards(self):
|
|
|
+ cards = []
|
|
|
+ for idx in (1, 2, 3):
|
|
|
+ candidates = self.d.xpath(
|
|
|
+ f'(//androidx.recyclerview.widget.RecyclerView)[{idx}]/android.widget.FrameLayout'
|
|
|
+ ).all()
|
|
|
+ for c in candidates:
|
|
|
+ try:
|
|
|
+ bounds = c.info.get('bounds', {})
|
|
|
+ h = bounds.get('bottom', 0) - bounds.get('top', 0)
|
|
|
+ if h > 400:
|
|
|
+ cards = candidates
|
|
|
+ break
|
|
|
+ except Exception:
|
|
|
+ continue
|
|
|
+ if cards:
|
|
|
+ break
|
|
|
+ if not cards:
|
|
|
+ return []
|
|
|
+ visible = []
|
|
|
+ for card in cards:
|
|
|
+ try:
|
|
|
+ bounds = card.info.get('bounds', {})
|
|
|
+ top = bounds.get('top', 0)
|
|
|
+ bottom = bounds.get('bottom', 0)
|
|
|
+ h = bottom - top
|
|
|
+ if h > 400 and top >= 554 and bottom <= 2600:
|
|
|
+ visible.append(card)
|
|
|
+ except Exception:
|
|
|
+ continue
|
|
|
+ return visible
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 1. 提取标题(拿不到返回 "")
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def get_title(self):
|
|
|
+ print("\n[1/7] 提取标题...")
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+ try:
|
|
|
+ title = self._safe_get_attr(
|
|
|
+ '//*[contains(@content-desc, "商品名称")]', "content_desc"
|
|
|
+ )
|
|
|
+ if title:
|
|
|
+ title = self._strip_content_desc_prefix(title, "商品名称")
|
|
|
+ print(f" 标题={title}")
|
|
|
+ return title
|
|
|
+ text_els = self.d.xpath('//android.widget.TextView').all()
|
|
|
+ longest = ""
|
|
|
+ for el in text_els:
|
|
|
+ t = (el.text or "").strip()
|
|
|
+ if len(t) > len(longest):
|
|
|
+ longest = t
|
|
|
+ if longest and len(longest) > 5:
|
|
|
+ print(f" 标题(fallback)={longest}")
|
|
|
+ return longest
|
|
|
+ except Exception as e:
|
|
|
+ print(f" 标题提取异常: {e}")
|
|
|
+ print(" [失败] 未找到标题,返回空")
|
|
|
+ return ""
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 2. 提取价格(拿不到返回 None)
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def get_price(self):
|
|
|
+ print("\n[2/7] 提取价格...")
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+ try:
|
|
|
+ buy_btn = self.d.xpath('//*[@text="立即购买"]')
|
|
|
+ if not buy_btn.exists:
|
|
|
+ buy_btn = self.d.xpath('//*[@text="领券购买"]')
|
|
|
+ if not buy_btn.exists:
|
|
|
+ print(" [失败] 未找到「购买选项」,价格留空")
|
|
|
+ return None
|
|
|
+ buy_btn.click()
|
|
|
+ print(" 点击「购买」")
|
|
|
+ time.sleep(self.SLEEP_CLICK)
|
|
|
+
|
|
|
+ price_str = ""
|
|
|
+ price_xpath = '//*[contains(@content-desc, "到手价")]'
|
|
|
+ if self.d.xpath(price_xpath).exists:
|
|
|
+ price_str = self._safe_get_attr(price_xpath, "content_desc")
|
|
|
+ else:
|
|
|
+ fallback_xpath = '//*[contains(@content-desc, "¥")]'
|
|
|
+ if self.d.xpath(fallback_xpath).exists:
|
|
|
+ price_str = self._safe_get_attr(fallback_xpath, "content_desc")
|
|
|
+
|
|
|
+ self.d.press("back")
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+
|
|
|
+ if price_str:
|
|
|
+ match = re.search(r'¥([\d\.]+)', price_str)
|
|
|
+ if match:
|
|
|
+ price = float(match.group(1))
|
|
|
+ print(f" 价格={price} 元")
|
|
|
+ return price
|
|
|
+ except Exception as e:
|
|
|
+ print(f" 价格提取异常: {e}")
|
|
|
+ try:
|
|
|
+ self.d.press("back")
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ print(" [失败] 无法解析价格,返回 None")
|
|
|
+ return None
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 3. 提取规格(拿不到返回 "")
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def get_spec(self, current_price=None):
|
|
|
+ print("\n[3/7] 提取规格...")
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+ spec = ""
|
|
|
+ try:
|
|
|
+ # 必须打开购买弹窗才能看到完整的规格列表和价格绑定
|
|
|
+ buy_btn = self.d.xpath('//*[@text="立即购买"]')
|
|
|
+ if not buy_btn.exists:
|
|
|
+ buy_btn = self.d.xpath('//*[@text="领券购买"]')
|
|
|
+
|
|
|
+ if buy_btn.exists:
|
|
|
+ buy_btn.click()
|
|
|
+ time.sleep(self.SLEEP_CLICK)
|
|
|
+
|
|
|
+ if current_price is not None:
|
|
|
+ str_price = str(current_price)
|
|
|
+ if str_price.endswith('.0'):
|
|
|
+ str_price = str_price[:-2]
|
|
|
+
|
|
|
+ # 尝试两种不同的 UI 结构来匹配规格名称
|
|
|
+ xpath1 = f'//android.widget.TextView[contains(@text, "{str_price}")]/preceding-sibling::android.widget.TextView[1]'
|
|
|
+ xpath2 = f'//android.widget.TextView[contains(@text, "{str_price}")]/parent::*/preceding-sibling::android.widget.TextView[1]'
|
|
|
+
|
|
|
+ if self.d.xpath(xpath1).exists:
|
|
|
+ spec = self.d.xpath(xpath1).get_text()
|
|
|
+ elif self.d.xpath(xpath2).exists:
|
|
|
+ spec = self.d.xpath(xpath2).get_text()
|
|
|
+
|
|
|
+ if not spec:
|
|
|
+ # 兜底:尝试常见的规格前缀(比如“容量”、“颜色”、“规格”等)
|
|
|
+ fallback_labels = ["规格", "容量", "颜色", "尺码", "款式", "净含量", "类型", "套餐", "版本", "型号"]
|
|
|
+ for label in fallback_labels:
|
|
|
+ fallback_xpath = f'//android.widget.TextView[@text="{label}"]/parent::*/following-sibling::*//android.widget.TextView[1]'
|
|
|
+ if self.d.xpath(fallback_xpath).exists:
|
|
|
+ spec = self.d.xpath(fallback_xpath).get_text()
|
|
|
+ if spec:
|
|
|
+ break
|
|
|
+
|
|
|
+ # 无论成败,提取完都必须把弹窗关掉,以便后续抓取店铺
|
|
|
+ self.d.press("back")
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+
|
|
|
+ if spec:
|
|
|
+ print(f" 规格={spec}")
|
|
|
+ else:
|
|
|
+ print(" [失败] 未找到规格,留空")
|
|
|
+ except Exception as e:
|
|
|
+ print(f" 规格提取异常: {e}")
|
|
|
+ spec = ""
|
|
|
+ try:
|
|
|
+ self.d.press("back")
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+ except:
|
|
|
+ pass
|
|
|
+ return spec or ""
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 4. 提取店铺名(拿不到返回 "")
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def get_shop_name(self):
|
|
|
+ print("\n[4/7] 提取店铺名...")
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+ try:
|
|
|
+ self._scroll_until_found('//*[@text="进店"]', max_swipes=5)
|
|
|
+
|
|
|
+ # 尝试找 "进店" 按钮前面的文本节点作为店铺名
|
|
|
+ shop_name_node = self.d.xpath('//*[@text="进店"]/preceding-sibling::android.widget.TextView[1]')
|
|
|
+ if not shop_name_node.exists:
|
|
|
+ shop_name_node = self.d.xpath('//*[@text="进店"]/../preceding-sibling::*//android.widget.TextView[1]')
|
|
|
+
|
|
|
+ shop_name = ""
|
|
|
+ if shop_name_node.exists:
|
|
|
+ shop_name = shop_name_node.get_text()
|
|
|
+
|
|
|
+ # 如果没拿到,用关键字回退匹配
|
|
|
+ if not shop_name:
|
|
|
+ for keyword in ["旗舰店", "大药房", "专卖店", "专营店", "店"]:
|
|
|
+ shop_name = self._safe_get_attr(f'//*[contains(@content-desc, "{keyword}")]', "content_desc")
|
|
|
+ if not shop_name:
|
|
|
+ shop_name = self._safe_get_attr(f'//*[contains(@text, "{keyword}")]')
|
|
|
+ if shop_name:
|
|
|
+ break
|
|
|
+
|
|
|
+ if shop_name:
|
|
|
+ print(f" 店铺名={shop_name}")
|
|
|
+ else:
|
|
|
+ print(" [失败] 未找到店铺名,留空")
|
|
|
+ except Exception as e:
|
|
|
+ print(f" 店铺名提取异常: {e}")
|
|
|
+ shop_name = ""
|
|
|
+ return shop_name or ""
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 5. 提取公司名(拿不到返回 "")
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def _find_shop_avatar(self):
|
|
|
+ imgs = self.d.xpath('//android.widget.ImageView').all()
|
|
|
+ for img in imgs:
|
|
|
+ try:
|
|
|
+ bounds = img.info.get('bounds', {})
|
|
|
+ left = bounds.get('left', 0)
|
|
|
+ top = bounds.get('top', 0)
|
|
|
+ right = bounds.get('right', 0)
|
|
|
+ bottom = bounds.get('bottom', 0)
|
|
|
+ w = right - left
|
|
|
+ h = bottom - top
|
|
|
+ if 30 <= left <= 120 and 300 <= top <= 370 and abs(w - h) <= 30:
|
|
|
+ return img
|
|
|
+ except Exception:
|
|
|
+ continue
|
|
|
+ return None
|
|
|
+
|
|
|
+ def get_company_name(self):
|
|
|
+ print("\n[5/7] 提取公司名...")
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+ company = ""
|
|
|
+ try:
|
|
|
+ enter_btn = self._scroll_until_found('//*[@text="进店"]', max_swipes=5)
|
|
|
+ if not enter_btn:
|
|
|
+ print(" [失败] 未找到「进店」,公司名留空")
|
|
|
+ return ""
|
|
|
+
|
|
|
+ enter_btn.click()
|
|
|
+ time.sleep(random.uniform(1.5, 2))
|
|
|
+ print(" 点击「进店」")
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+
|
|
|
+ avatar = self._find_shop_avatar()
|
|
|
+ if avatar:
|
|
|
+ avatar.click()
|
|
|
+ time.sleep(random.uniform(1.5, 2))
|
|
|
+ print(" 点击「店铺头像」")
|
|
|
+ else:
|
|
|
+ notes_btn = self.d.xpath('//*[contains(@text, "篇笔记")]')
|
|
|
+ if not notes_btn.exists:
|
|
|
+ print(" [失败] 未找到店铺头像和「X篇笔记」,公司名留空")
|
|
|
+ self.d.press("back")
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+ return ""
|
|
|
+ notes_btn.click()
|
|
|
+ time.sleep(random.uniform(1.5, 2))
|
|
|
+ print(" 点击「X篇笔记」资质入口")
|
|
|
+
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+ company = self._safe_get_attr('//*[contains(@text, "公司")]')
|
|
|
+ if company:
|
|
|
+ print(f" 公司名={company}")
|
|
|
+ else:
|
|
|
+ print(" [无] 未找到公司名,留空")
|
|
|
+
|
|
|
+ self.d.press("back")
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+ self.d.press("back")
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+ except Exception as e:
|
|
|
+ print(f" 公司名提取异常: {e}")
|
|
|
+ company = ""
|
|
|
+ return company or ""
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 6. 提取商品链接(拿不到返回 "")
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def get_product_link(self):
|
|
|
+ print("\n[6/7] 提取商品链接...")
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+ try:
|
|
|
+ # 获取老的剪贴板
|
|
|
+ old_clipboard = (self.d.clipboard or "").strip()
|
|
|
+
|
|
|
+ share_btn = self.d.xpath('//*[@content-desc="分享商品"]')
|
|
|
+ if not share_btn.exists:
|
|
|
+ share_btn = self.d.xpath('//*[contains(@content-desc, "分享")]')
|
|
|
+
|
|
|
+ # 增加一个重试逻辑,防止页面卡顿导致还没加载出来
|
|
|
+ if not share_btn.exists:
|
|
|
+ time.sleep(2)
|
|
|
+ share_btn = self.d.xpath('//*[contains(@content-desc, "分享")]')
|
|
|
+
|
|
|
+ if not share_btn.exists:
|
|
|
+ print(" [失败] 未找到「分享商品」,链接留空")
|
|
|
+ return ""
|
|
|
+
|
|
|
+ share_btn.click()
|
|
|
+ time.sleep(1)
|
|
|
+ print(" 点击「分享商品」")
|
|
|
+
|
|
|
+ copy_btn = self.d.xpath('//*[@text="复制链接"]')
|
|
|
+ if copy_btn.exists:
|
|
|
+ copy_btn.click()
|
|
|
+ time.sleep(self.SLEEP_CLIPBOARD)
|
|
|
+
|
|
|
+ link = (self.d.clipboard or "").strip()
|
|
|
+
|
|
|
+ if link == old_clipboard and link != "":
|
|
|
+ print(" [警告] 剪贴板内容未改变,可能复制失败")
|
|
|
+ link = ""
|
|
|
+
|
|
|
+ if link:
|
|
|
+ match = re.search(r'https?://xhslink\.com/\S+', link)
|
|
|
+ if match:
|
|
|
+ link = match.group(0)
|
|
|
+ print(f" 商品链接={link}")
|
|
|
+ else:
|
|
|
+ print(" [失败] 剪贴板未更新,链接留空")
|
|
|
+ except Exception as e:
|
|
|
+ print(f" 链接提取异常: {e}")
|
|
|
+ link = ""
|
|
|
+ return link or ""
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 7. 提取批准文号(拿不到返回 "")
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def get_approval_number(self):
|
|
|
+ print("\n[7/7] 提取批准文号...")
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+ try:
|
|
|
+ w, h = self.d.window_size()
|
|
|
+ for _ in range(6):
|
|
|
+ self.d.swipe(w // 2, int(h * 0.88), w // 2, int(h * 0.50), duration=0.3)
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+ if self.d.xpath('//*[@text="批准文号"]').exists:
|
|
|
+ break
|
|
|
+
|
|
|
+ approval = self._safe_get_attr('//*[starts-with(@text, "国药准字")]')
|
|
|
+ if approval:
|
|
|
+ print(f" 批准文号={approval}")
|
|
|
+ else:
|
|
|
+ print(" [无] 未找到国药准字,留空")
|
|
|
+ except Exception as e:
|
|
|
+ print(f" 批准文号提取异常: {e}")
|
|
|
+ approval = ""
|
|
|
+ return approval or ""
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 去重 & 落库
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def data_is_exists(self, data):
|
|
|
+ required_keys = ['min_price', 'shop', 'scrape_date', 'platform']
|
|
|
+ if not all(key in data for key in required_keys):
|
|
|
+ return False
|
|
|
+ conn = None
|
|
|
+ try:
|
|
|
+ conn = get_mysql()
|
|
|
+ with conn.cursor() as cur:
|
|
|
+ query_sql = f"""
|
|
|
+ SELECT * FROM {self.table_name}
|
|
|
+ WHERE min_price = %s
|
|
|
+ AND store_name = %s
|
|
|
+ AND scrape_date = %s
|
|
|
+ AND platform_id = %s
|
|
|
+ AND collect_equipment_account_id = %s
|
|
|
+ AND collect_round = %s
|
|
|
+ """
|
|
|
+ cur.execute(query_sql, (
|
|
|
+ data['min_price'], data['shop'], data['scrape_date'],
|
|
|
+ data['platform'],
|
|
|
+ data.get('collect_equipment_account_id', 0),
|
|
|
+ data.get('collect_round', 0),
|
|
|
+ ))
|
|
|
+ return bool(cur.fetchone())
|
|
|
+ except Exception as e:
|
|
|
+ print(f"去重查询异常: {e}")
|
|
|
+ return False
|
|
|
+ finally:
|
|
|
+ if conn:
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ def shop_is_exists(self, data):
|
|
|
+ shop_value = str(data.get('shop') or '').strip()
|
|
|
+ if not shop_value:
|
|
|
+ return False
|
|
|
+ conn = None
|
|
|
+ try:
|
|
|
+ conn = get_mysql()
|
|
|
+ with conn.cursor() as cur:
|
|
|
+ cur.execute(
|
|
|
+ f"SELECT * FROM {self.shop_table_name} WHERE shop = %s LIMIT 1",
|
|
|
+ (shop_value,)
|
|
|
+ )
|
|
|
+ return bool(cur.fetchone())
|
|
|
+ except Exception as e:
|
|
|
+ print(f"店铺去重查询异常: {e}")
|
|
|
+ return False
|
|
|
+ finally:
|
|
|
+ if conn:
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ def get_province_city_from_shop(self, data):
|
|
|
+ shop_name = data.get('shop')
|
|
|
+ if not shop_name:
|
|
|
+ return
|
|
|
+ conn = None
|
|
|
+ try:
|
|
|
+ conn = get_mysql()
|
|
|
+ with conn.cursor() as cur:
|
|
|
+ cur.execute(
|
|
|
+ f"SELECT province, city, business_license_company "
|
|
|
+ f"FROM {self.shop_table_name} WHERE shop = %s AND platform = %s LIMIT 1",
|
|
|
+ (shop_name, XHS_PLATFORM_ID)
|
|
|
+ )
|
|
|
+ result = cur.fetchone()
|
|
|
+ if result:
|
|
|
+ data['province_name'] = result[0] or ''
|
|
|
+ data['city_name'] = result[1] or ''
|
|
|
+ data['company_name'] = result[2] or ''
|
|
|
+ else:
|
|
|
+ data['province_name'] = ''
|
|
|
+ data['city_name'] = ''
|
|
|
+ data['company_name'] = ''
|
|
|
+ except Exception as e:
|
|
|
+ print(f"查询省市信息失败: {e}")
|
|
|
+ data['province_name'] = ''
|
|
|
+ data['city_name'] = ''
|
|
|
+ data['company_name'] = ''
|
|
|
+ finally:
|
|
|
+ if conn:
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ def save_to_shop_database(self, data):
|
|
|
+ shop = str(data.get('shop') or '').strip()
|
|
|
+ if not shop:
|
|
|
+ print("保存店铺数据失败:shop 为空")
|
|
|
+ return False
|
|
|
+
|
|
|
+ now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
+ fields = (
|
|
|
+ shop, # company_id
|
|
|
+ shop, # shop_id
|
|
|
+ str(data.get('contact_address') or '').strip(),
|
|
|
+ str(data.get('qualification_number') or '').strip(),
|
|
|
+ str(data.get('business_license_company') or '').strip(),
|
|
|
+ str(data.get('business_license_address') or '').strip(),
|
|
|
+ str(data.get('store_url') or '').strip(),
|
|
|
+ str(data.get('scrape_date') or self.get_current_date()).strip(),
|
|
|
+ str(data.get('platform') or str(XHS_PLATFORM_ID)).strip(),
|
|
|
+ str(data.get('province_name') or data.get('province') or '').strip(),
|
|
|
+ str(data.get('city_name') or data.get('city') or '').strip(),
|
|
|
+ data.get('create_time') or now,
|
|
|
+ data.get('update_time') or now,
|
|
|
+ )
|
|
|
+
|
|
|
+ for attempt in range(5):
|
|
|
+ conn = None
|
|
|
+ try:
|
|
|
+ conn = get_mysql()
|
|
|
+ with conn.cursor() as cur:
|
|
|
+ cur.execute(f"""
|
|
|
+ INSERT IGNORE INTO {self.shop_table_name} (
|
|
|
+ shop,company_id,shop_id,contact_address, qualification_number,
|
|
|
+ business_license_company, business_license_address,
|
|
|
+ shop_url, scrape_date, platform,
|
|
|
+ province, city, create_time, update_time
|
|
|
+ ) VALUES (%s,%s,%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
|
+ """, (shop,) + fields)
|
|
|
+ conn.commit()
|
|
|
+ print("店铺数据存入数据库成功")
|
|
|
+ return True
|
|
|
+ except Exception as e:
|
|
|
+ print(f'保存店铺数据库异常 (尝试 {attempt + 1}/5): {e}')
|
|
|
+ if conn:
|
|
|
+ conn.rollback()
|
|
|
+ conn.close()
|
|
|
+ if attempt == 4:
|
|
|
+ return False
|
|
|
+ time.sleep(2)
|
|
|
+
|
|
|
+ def save_to_database(self, data):
|
|
|
+ """保存采集数据,拿不到的字段全部写空/0"""
|
|
|
+ print(f'保存数据到数据库:{data}')
|
|
|
+ now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
+
|
|
|
+ for attempt in range(5):
|
|
|
+ conn = None
|
|
|
+ try:
|
|
|
+ conn = get_mysql()
|
|
|
+ with conn.cursor() as cur:
|
|
|
+ cur.execute(f"""
|
|
|
+ INSERT IGNORE INTO {self.table_name} (
|
|
|
+ enterprise_id, platform_id, platform_item_id,
|
|
|
+ province_id, city_id,
|
|
|
+ province_name, city_name, area_info,
|
|
|
+ product_brand, product_name, product_specs, search_name,
|
|
|
+ collect_config_info, one_box_price,
|
|
|
+ manufacture_date, expiry_date, manufacturer, approval_number,
|
|
|
+ is_sold_out, online_posting_count, continuous_listing_count,
|
|
|
+ link_url, store_name, store_url,
|
|
|
+ shipment_province_id, shipment_province_name,
|
|
|
+ shipment_city_id, shipment_city_name,
|
|
|
+ company_name, qualification_number,
|
|
|
+ scrape_date, min_price, number, sales, inventory,
|
|
|
+ snapshot_url,
|
|
|
+ collect_equipment_account_id, insert_time, update_time,
|
|
|
+ collect_round, collect_region_id, task_id,
|
|
|
+ shop_id, company_id
|
|
|
+ ) VALUES (
|
|
|
+ %s,%s,%s, %s,%s, %s,%s,%s, %s,%s,%s,%s,
|
|
|
+ %s,%s, %s,%s,%s,%s, %s,%s,%s, %s,%s,%s,
|
|
|
+ %s,%s, %s,%s, %s,%s, %s,%s,%s,%s, %s,
|
|
|
+ %s,%s,%s, %s,%s,%s,%s,
|
|
|
+ %s,%s
|
|
|
+ )
|
|
|
+ """, (
|
|
|
+ data.get('enterprise_id', 0),
|
|
|
+ data.get('platform_id', XHS_PLATFORM_ID),
|
|
|
+ data.get('product_name', '')+data.get('store_name', '')+str(data.get('min_price', 0)),
|
|
|
+ data.get('province_id', 0),
|
|
|
+ data.get('city_id', 0),
|
|
|
+ data.get('province_name', ''),
|
|
|
+ data.get('city_name', ''),
|
|
|
+ data.get('area_info', ''),
|
|
|
+ data.get('product_brand', ''),
|
|
|
+ data.get('product_name', ''),
|
|
|
+ data.get('product_specs', ''),
|
|
|
+ data.get('search_name', ''),
|
|
|
+ data.get('collect_config_info', ''),
|
|
|
+ data.get('one_box_price', 0),
|
|
|
+ data.get('manufacture_date', ''),
|
|
|
+ data.get('expiry_date', ''),
|
|
|
+ data.get('manufacturer', ''),
|
|
|
+ data.get('approval_number', ''),
|
|
|
+ data.get('is_sold_out', 0),
|
|
|
+ data.get('online_posting_count', 1),
|
|
|
+ data.get('continuous_listing_count', 1),
|
|
|
+ data.get('link_url', ''),
|
|
|
+ data.get('store_name', ''),
|
|
|
+ data.get('store_url', ''),
|
|
|
+ data.get('shipment_province_id', 0),
|
|
|
+ data.get('shipment_province_name', ''),
|
|
|
+ data.get('shipment_city_id', 0),
|
|
|
+ data.get('shipment_city_name', ''),
|
|
|
+ data.get('company_name', ''),
|
|
|
+ data.get('qualification_number', ''),
|
|
|
+ data.get('scrape_date', ''),
|
|
|
+ data.get('min_price', 0),
|
|
|
+ data.get('number', 1),
|
|
|
+ data.get('sales', ''),
|
|
|
+ data.get('inventory', ''),
|
|
|
+ data.get('snapshot_url', ''),
|
|
|
+ data.get('collect_equipment_account_id', 0),
|
|
|
+ data.get('insert_time', now),
|
|
|
+ data.get('update_time', now),
|
|
|
+ data.get('collect_round', 0),
|
|
|
+ data.get('collect_region_id', 0),
|
|
|
+ data.get('task_id') or 0,
|
|
|
+ data.get('store_name', ''),
|
|
|
+ data.get('store_name', ''),
|
|
|
+ ))
|
|
|
+ conn.commit()
|
|
|
+ self.max_counts += 1
|
|
|
+ print(f"========== 设备 [{self.device_id}] 存入数据库成功,当前该设备总计采集数据: {self.max_counts} 条 ==========")
|
|
|
+ return True
|
|
|
+ except Exception as e:
|
|
|
+ print(f'保存数据库异常 (尝试 {attempt + 1}/5): {e}')
|
|
|
+ if conn:
|
|
|
+ conn.rollback()
|
|
|
+ conn.close()
|
|
|
+ if attempt == 4:
|
|
|
+ print("达到最大重试次数,保存失败")
|
|
|
+ return False
|
|
|
+ time.sleep(2)
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 快照功能(滚动截图 → 拼接 → 压缩 → 上传 OSS)
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def screenshot_and_upload_oss(self, screenshot_desc='xhs_product_screenshot'):
|
|
|
+ """1.滚动截图 → 2.压缩 → 3.上传OSS → 4.清理本地,返回 OSS URL 或 None"""
|
|
|
+ timestamp = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
|
|
|
+ unique_id = uuid.uuid4().hex[:8]
|
|
|
+ base_name = f"{screenshot_desc}_{self.device_id}_{timestamp}_{unique_id}"
|
|
|
+ scroll_path = f"./{base_name}_scroll.png"
|
|
|
+ compress_path = f"./{base_name}_compress.jpg"
|
|
|
+
|
|
|
+ if not hasattr(self, 'oss_bucket'):
|
|
|
+ self.oss_auth = oss2.Auth(
|
|
|
+ self.oss_config["access_key_id"],
|
|
|
+ self.oss_config["access_key_secret"]
|
|
|
+ )
|
|
|
+ self.oss_bucket = oss2.Bucket(
|
|
|
+ self.oss_auth,
|
|
|
+ self.oss_config["endpoint"],
|
|
|
+ self.oss_config["bucket_name"]
|
|
|
+ )
|
|
|
+
|
|
|
+ try:
|
|
|
+ screenshot_image = self._scroll_screenshot()
|
|
|
+ screenshot_image.save(scroll_path)
|
|
|
+ print(f"滚动截图完成: {scroll_path}")
|
|
|
+
|
|
|
+ if not self.compress_image(scroll_path, compress_path):
|
|
|
+ raise Exception("图片压缩失败")
|
|
|
+
|
|
|
+ oss_file_path = f"{self.oss_config['oss_prefix']}/{os.path.basename(compress_path)}"
|
|
|
+ self.oss_bucket.put_object_from_file(oss_file_path, compress_path)
|
|
|
+ oss_url = f"https://{self.oss_config['bucket_name']}.{self.oss_config['endpoint']}/{oss_file_path}"
|
|
|
+ print(f"OSS上传成功: {oss_url}")
|
|
|
+ return oss_url
|
|
|
+ except Exception as e:
|
|
|
+ print(f"快照/上传OSS失败: {e}")
|
|
|
+ return None
|
|
|
+ finally:
|
|
|
+ for f in [scroll_path, compress_path]:
|
|
|
+ if os.path.exists(f):
|
|
|
+ try:
|
|
|
+ os.remove(f)
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+
|
|
|
+ def _merge_screenshots(self, screens):
|
|
|
+ """垂直堆叠多张 PIL 截图"""
|
|
|
+ if len(screens) == 1:
|
|
|
+ return screens[0].convert('RGB')
|
|
|
+ rgb = [s.convert('RGB') for s in screens]
|
|
|
+ total_w = rgb[0].width
|
|
|
+ total_h = sum(s.height for s in rgb)
|
|
|
+ merged = Image.new('RGB', (total_w, total_h))
|
|
|
+ y = 0
|
|
|
+ for img in rgb:
|
|
|
+ merged.paste(img, (0, y))
|
|
|
+ y += img.height
|
|
|
+ return merged
|
|
|
+
|
|
|
+ def _scroll_screenshot(self, scroll_times=1):
|
|
|
+ """截首屏 → 滚动到「进店」→ 堆叠拼接"""
|
|
|
+ w, h = self.d.window_size()
|
|
|
+ screens = [self.d.screenshot()]
|
|
|
+
|
|
|
+ if self.d(textContains='进店').exists or self.d(textStartsWith='进店').exists:
|
|
|
+ return self._merge_screenshots(screens)
|
|
|
+
|
|
|
+ for i in range(scroll_times):
|
|
|
+ self.d.swipe(w // 2, int(h * 0.85), w // 2, int(h * 0.15),
|
|
|
+ duration=random.uniform(0.8, 1.5))
|
|
|
+ time.sleep(random.uniform(2.0, 4.0))
|
|
|
+ screens.append(self.d.screenshot())
|
|
|
+ if self.d(textContains='进店').exists or self.d(textStartsWith='进店').exists:
|
|
|
+ break
|
|
|
+
|
|
|
+ return self._merge_screenshots(screens)
|
|
|
+
|
|
|
+ def compress_image(self, input_path, output_path, target_size_kb=100,
|
|
|
+ max_quality=80, min_quality=20):
|
|
|
+ """cv2 压缩图片到 target_size_kb 以内"""
|
|
|
+ try:
|
|
|
+ img = cv2.imread(input_path)
|
|
|
+ if img is None:
|
|
|
+ return False
|
|
|
+ h, w = img.shape[:2]
|
|
|
+ img_rs = cv2.resize(img, (int(w * 0.7), int(h * 0.7)),
|
|
|
+ interpolation=cv2.INTER_AREA)
|
|
|
+
|
|
|
+ quality = max_quality
|
|
|
+ while quality >= min_quality:
|
|
|
+ cv2.imwrite(output_path, img_rs, [cv2.IMWRITE_JPEG_QUALITY, quality])
|
|
|
+ if os.path.getsize(output_path) / 1024 <= target_size_kb:
|
|
|
+ return True
|
|
|
+ quality -= 5
|
|
|
+
|
|
|
+ img_rs2 = cv2.resize(img_rs, (int(img_rs.shape[1] * 0.5),
|
|
|
+ int(img_rs.shape[0] * 0.5)),
|
|
|
+ cv2.INTER_AREA)
|
|
|
+ cv2.imwrite(output_path, img_rs2, [cv2.IMWRITE_JPEG_QUALITY, min_quality])
|
|
|
+ return True
|
|
|
+ except Exception as e:
|
|
|
+ print(f"图片压缩失败: {e}")
|
|
|
+ return False
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def is_max_count_reached(self):
|
|
|
+ return bool(self.max_counts_limit and self.max_counts >= self.max_counts_limit)
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 完成上报
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def finish_task_normally(self, end_page, reason):
|
|
|
+ if not self.finish_reported:
|
|
|
+ report_api(
|
|
|
+ self.task_id,
|
|
|
+ platform=self.platform,
|
|
|
+ username=getattr(self, 'username', ''),
|
|
|
+ page=end_page,
|
|
|
+ is_finished=1,
|
|
|
+ crawled_count=getattr(self, 'max_counts', 0),
|
|
|
+ total_pages=end_page,
|
|
|
+ )
|
|
|
+ self.finish_reported = True
|
|
|
+ print(reason)
|
|
|
+ return True
|
|
|
+
|
|
|
+ def finish_task_abnormally(self, end_page, reason, finish_status=0):
|
|
|
+ if not self.finish_reported:
|
|
|
+ report_api(
|
|
|
+ self.task_id,
|
|
|
+ platform=self.platform,
|
|
|
+ username=getattr(self, 'username', ''),
|
|
|
+ page=end_page,
|
|
|
+ is_finished=0,
|
|
|
+ crawled_count=getattr(self, 'max_counts', 0),
|
|
|
+ total_pages=end_page,
|
|
|
+ )
|
|
|
+ self.finish_reported = True
|
|
|
+ print(reason)
|
|
|
+ return False
|
|
|
+
|
|
|
+ def finish_task_with_max_count(self, end_page):
|
|
|
+ return self.finish_task_normally(
|
|
|
+ end_page,
|
|
|
+ f"达到最大采集数量 {self.max_counts_limit},已采集 {self.max_counts} 条,停止任务"
|
|
|
+ )
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 单商品详情采集
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def integrate_data(self):
|
|
|
+ """采集单个商品详情页所有数据并落库"""
|
|
|
+ print(self.SEP)
|
|
|
+ print(f" 小红书商品详情页采集")
|
|
|
+ print(f" 时间: {self.get_current_date()}")
|
|
|
+ print(f" 设备: {self.device_id}")
|
|
|
+ print(f" 目标: {self.title_key} | 品牌: {self.brand or '-'}")
|
|
|
+ print(self.SEP)
|
|
|
+
|
|
|
+ title = self.get_title() or ""
|
|
|
+ if title and not self.is_title_useful(title):
|
|
|
+ print(f" [跳过] 标题不匹配: {title[:40]}...")
|
|
|
+ self.unrelated_data += 1
|
|
|
+ return False
|
|
|
+
|
|
|
+ # 快照:刚进详情页,页面最干净,截取商品顶部→店铺区域
|
|
|
+ oss_url = ''
|
|
|
+ try:
|
|
|
+ result = self.screenshot_and_upload_oss()
|
|
|
+ if result:
|
|
|
+ oss_url = result
|
|
|
+ print(f"快照上传成功: {oss_url}")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"快照异常: {e}")
|
|
|
+
|
|
|
+ min_price = self.get_price()
|
|
|
+ spec = self.get_spec(min_price) or ""
|
|
|
+ shop_name = self.get_shop_name() or ""
|
|
|
+ company = self.get_company_name() or ""
|
|
|
+
|
|
|
+ link = self.get_product_link() or ""
|
|
|
+ if not link:
|
|
|
+ print(" [跳过] 未获取到商品链接,本条数据作废,不存入数据库")
|
|
|
+ return False
|
|
|
+
|
|
|
+ approval = self.get_approval_number() or ""
|
|
|
+
|
|
|
+ print(f"\n{self.SEP}")
|
|
|
+ print(" 采集结果汇总")
|
|
|
+ print(self.SEP)
|
|
|
+ print(f" 标题: {title}")
|
|
|
+ print(f" 价格: {min_price} 元" if min_price is not None else " 价格: (未取到)")
|
|
|
+ print(f" 规格: {spec}")
|
|
|
+ print(f" 店铺名: {shop_name}")
|
|
|
+ print(f" 公司名: {company}")
|
|
|
+ print(f" 批准文号: {approval}")
|
|
|
+ print(f" 商品链接: {link}")
|
|
|
+ print(self.SEP)
|
|
|
+
|
|
|
+ scrape_date = self.get_current_date()
|
|
|
+
|
|
|
+ # 去重
|
|
|
+ dup_data = {
|
|
|
+ 'min_price': min_price or 0,
|
|
|
+ 'shop': shop_name or '',
|
|
|
+ 'scrape_date': scrape_date,
|
|
|
+ 'platform': str(XHS_PLATFORM_ID),
|
|
|
+ 'collect_equipment_account_id': self.collect_equipment_account_id or 0,
|
|
|
+ 'collect_round': self.collect_round or 0,
|
|
|
+ }
|
|
|
+ if min_price is not None and self.data_is_exists(dup_data):
|
|
|
+ print('存在相同数据不入库')
|
|
|
+ return True
|
|
|
+
|
|
|
+ # 店铺表
|
|
|
+ province_name = ''
|
|
|
+ city_name = ''
|
|
|
+ company_name = company or ''
|
|
|
+
|
|
|
+ shop_data = {
|
|
|
+ 'shop': shop_name,
|
|
|
+ 'store_url': link,
|
|
|
+ 'scrape_date': scrape_date,
|
|
|
+ 'platform': XHS_PLATFORM_ID,
|
|
|
+ 'create_time': datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
|
+ 'update_time': datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
|
+ }
|
|
|
+
|
|
|
+ if shop_name and self.shop_is_exists(shop_data):
|
|
|
+ print("店铺数据已存在,进行省市回填")
|
|
|
+ self.get_province_city_from_shop(shop_data)
|
|
|
+ province_name = shop_data.get('province_name') or ''
|
|
|
+ city_name = shop_data.get('city_name') or ''
|
|
|
+ # 如果数据库里的为空,则保留本次新抓到的 company_name
|
|
|
+ company_name = shop_data.get('company_name') or company_name
|
|
|
+
|
|
|
+ # 如果之前入库时省市是空的,但我们现在有公司名,就顺手解析一次
|
|
|
+ if company_name and not province_name:
|
|
|
+ province_name, city_name = self.infer_region_from_address(company_name)
|
|
|
+ print(f"基于公司名补全省市: {province_name}, {city_name}")
|
|
|
+
|
|
|
+ elif shop_name:
|
|
|
+ print("店铺数据不存在,插入店铺表")
|
|
|
+ shop_data['business_license_company'] = company_name
|
|
|
+ if company_name:
|
|
|
+ p_name, c_name = self.infer_region_from_address(company_name)
|
|
|
+ shop_data['province_name'] = p_name
|
|
|
+ shop_data['city_name'] = c_name
|
|
|
+ province_name = p_name
|
|
|
+ city_name = c_name
|
|
|
+ print(f"基于公司名匹配到省市: {p_name}, {c_name}")
|
|
|
+ self.save_to_shop_database(shop_data)
|
|
|
+
|
|
|
+ # 获取省市ID
|
|
|
+ province_id, city_id = self._resolve_region_ids(province_name, city_name)
|
|
|
+
|
|
|
+ # 组装落库数据
|
|
|
+ now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
+ save_data = {
|
|
|
+ 'enterprise_id': self.enterprise_id or 0,
|
|
|
+ 'platform_id': XHS_PLATFORM_ID,
|
|
|
+ 'platform_item_id': '',
|
|
|
+ 'province_id': province_id,
|
|
|
+ 'city_id': city_id,
|
|
|
+ 'province_name': province_name,
|
|
|
+ 'city_name': city_name,
|
|
|
+ 'area_info': '',
|
|
|
+ 'product_brand': self.brand,
|
|
|
+ 'product_name': title,
|
|
|
+ 'product_specs': spec,
|
|
|
+ 'search_name': self.search_key,
|
|
|
+ 'collect_config_info': self.collect_config_info or '',
|
|
|
+ 'one_box_price': 0,
|
|
|
+ 'manufacture_date': '',
|
|
|
+ 'expiry_date': '',
|
|
|
+ 'manufacturer': '',
|
|
|
+ 'approval_number': approval,
|
|
|
+ 'is_sold_out': 0,
|
|
|
+ 'online_posting_count': 1,
|
|
|
+ 'continuous_listing_count': 1,
|
|
|
+ 'link_url': link,
|
|
|
+ 'store_name': shop_name,
|
|
|
+ 'store_url': '',
|
|
|
+ 'shipment_province_id': 0,
|
|
|
+ 'shipment_province_name': '',
|
|
|
+ 'shipment_city_id': 0,
|
|
|
+ 'shipment_city_name': '',
|
|
|
+ 'company_name': company_name,
|
|
|
+ 'qualification_number': '',
|
|
|
+ 'scrape_date': scrape_date,
|
|
|
+ 'min_price': min_price or 0,
|
|
|
+ 'number': 1,
|
|
|
+ 'sales': '',
|
|
|
+ 'inventory': '',
|
|
|
+ 'snapshot_url': oss_url or '',
|
|
|
+ 'collect_equipment_account_id': self.collect_equipment_account_id or 0,
|
|
|
+ 'insert_time': now,
|
|
|
+ 'update_time': now,
|
|
|
+ 'collect_round': self.collect_round or 0,
|
|
|
+ 'collect_region_id': self.collect_region_id or 0,
|
|
|
+ 'task_id': self.task_id or 0,
|
|
|
+ }
|
|
|
+
|
|
|
+ try:
|
|
|
+ from spiders.pdd.box_script import extract_quantity_and_unit
|
|
|
+ quantity, unit = extract_quantity_and_unit(f"{title} {spec}")
|
|
|
+ if quantity:
|
|
|
+ save_data['number'] = quantity
|
|
|
+ if not save_data['product_specs']:
|
|
|
+ save_data['product_specs'] = f"{quantity}{unit}"
|
|
|
+ print(f"✅ 智能提取到数量: {quantity}{unit}")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"智能提取数量失败: {e}")
|
|
|
+
|
|
|
+ self.save_to_database(save_data)
|
|
|
+ return True
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 首页检测 & 回退(参照 PDD 的 is_pdd_home_page / back_to_pdd_home_page)
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ XHS_HOME_ANCHORS = [
|
|
|
+ '//*[@text="首页"]',
|
|
|
+ '//*[@text="市集"]',
|
|
|
+ '//*[@content-desc="发布"]',
|
|
|
+ '//*[@text="消息"]',
|
|
|
+ '//*[@text="我"]',
|
|
|
+ ]
|
|
|
+ XHS_HOME_HIT_THRESHOLD = 3
|
|
|
+
|
|
|
+ def is_xhs_home_page(self):
|
|
|
+ """命中 >= 3 个底部 tab 锚点即认为在首页"""
|
|
|
+ try:
|
|
|
+ hits = sum(1 for xp in self.XHS_HOME_ANCHORS if self.d.xpath(xp).exists)
|
|
|
+ return hits >= self.XHS_HOME_HIT_THRESHOLD
|
|
|
+ except Exception:
|
|
|
+ return False
|
|
|
+
|
|
|
+ def is_xhs_foreground(self):
|
|
|
+ """前台包名 = 小红书 + 页面可见"""
|
|
|
+ try:
|
|
|
+ current_app = self.d.app_current() or {}
|
|
|
+ if current_app.get("package", "") != self.package_name:
|
|
|
+ return False
|
|
|
+ screen_on = bool((self.d.info or {}).get("screenOn", True))
|
|
|
+ if not screen_on:
|
|
|
+ return False
|
|
|
+ # 任一已知锚点可见就算在可操作页面
|
|
|
+ visible_anchors = self.XHS_HOME_ANCHORS + [
|
|
|
+ '//*[@text="进店"]',
|
|
|
+ '//*[@text="搜索"]',
|
|
|
+ '//*[contains(@content-desc, "商品名称")]',
|
|
|
+ ]
|
|
|
+ for xp in visible_anchors:
|
|
|
+ if self.d.xpath(xp).exists:
|
|
|
+ return True
|
|
|
+ return False
|
|
|
+ except Exception:
|
|
|
+ return False
|
|
|
+
|
|
|
+ def back_to_xhs_home_page(self, max_back_times=5):
|
|
|
+ """按 back 直到回到小红书首页,最多 max_back_times 次"""
|
|
|
+ for i in range(max_back_times):
|
|
|
+ if self.is_xhs_home_page():
|
|
|
+ print(f" 已回到小红书首页(第 {i} 次 back)")
|
|
|
+ return True
|
|
|
+ self.d.press("back")
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+ return self.is_xhs_home_page()
|
|
|
+
|
|
|
+ def is_xhs_package_foreground(self):
|
|
|
+ """只判断前台包名是否为小红书,不检查页面锚点"""
|
|
|
+ try:
|
|
|
+ return (self.d.app_current() or {}).get("package", "") == self.package_name
|
|
|
+ except Exception:
|
|
|
+ return False
|
|
|
+
|
|
|
+ # ============================================================
|
|
|
+ # 主循环
|
|
|
+ # ============================================================
|
|
|
+
|
|
|
+ def main(self):
|
|
|
+ completed_normally = False
|
|
|
+ current_page = self.page
|
|
|
+ consecutive_empty_pages = 0
|
|
|
+
|
|
|
+ if not self.connect_device():
|
|
|
+ self.finish_task_abnormally(current_page, "设备连接失败")
|
|
|
+ return False
|
|
|
+
|
|
|
+ # 统一入口准备:确保在首页后再走市集→搜索
|
|
|
+ if self.is_xhs_package_foreground():
|
|
|
+ print("小红书在前台,尝试回退到首页...")
|
|
|
+ self.back_to_xhs_home_page(max_back_times=5)
|
|
|
+ else:
|
|
|
+ print("小红书不在前台,重新启动...")
|
|
|
+ self.start_app()
|
|
|
+ time.sleep(5) # 给 App 充足加载时间
|
|
|
+ self.back_to_xhs_home_page(max_back_times=3)
|
|
|
+
|
|
|
+ # 最终确认:如果还是不在首页,再启动一次
|
|
|
+ if not self.is_xhs_home_page():
|
|
|
+ print("仍未在首页,重新启动再试...")
|
|
|
+ self.start_app()
|
|
|
+ time.sleep(5)
|
|
|
+ self.back_to_xhs_home_page(max_back_times=3)
|
|
|
+
|
|
|
+ if not self.enter_search_page():
|
|
|
+ self.finish_task_abnormally(current_page, "进入搜索页失败")
|
|
|
+ return False
|
|
|
+
|
|
|
+ report_api(self.task_id, platform=self.platform, username=getattr(self, 'username', ''),
|
|
|
+ page=self.page, is_finished=0)
|
|
|
+
|
|
|
+ self._heartbeat_running = True
|
|
|
+
|
|
|
+ def heartbeat_worker():
|
|
|
+ while self._heartbeat_running:
|
|
|
+ time.sleep(HEARTBEAT_INTERVAL_SECONDS)
|
|
|
+ if self._heartbeat_running:
|
|
|
+ report_heartbeat(self.task_id, getattr(self, 'page', 0), device_id=self.device_id)
|
|
|
+
|
|
|
+ threading.Thread(target=heartbeat_worker, daemon=True).start()
|
|
|
+
|
|
|
+ try:
|
|
|
+ for _ in range(300):
|
|
|
+ print(f"\n{'=' * 60}")
|
|
|
+ print(f" 第 {current_page + 1} 页")
|
|
|
+ print(f"{'=' * 60}")
|
|
|
+
|
|
|
+ if self.unrelated_data > 15:
|
|
|
+ completed_normally = self.finish_task_normally(
|
|
|
+ current_page, '连续超过15个不达标,结束采集')
|
|
|
+ break
|
|
|
+
|
|
|
+ if self.is_max_count_reached():
|
|
|
+ completed_normally = self.finish_task_with_max_count(current_page)
|
|
|
+ break
|
|
|
+
|
|
|
+ cards = self.get_product_cards()
|
|
|
+ print(f" 当前页 {len(cards)} 个可见商品")
|
|
|
+
|
|
|
+ if not cards:
|
|
|
+ consecutive_empty_pages += 1
|
|
|
+ if consecutive_empty_pages >= 3:
|
|
|
+ completed_normally = self.finish_task_normally(
|
|
|
+ current_page, "连续空页达到阈值")
|
|
|
+ break
|
|
|
+ else:
|
|
|
+ consecutive_empty_pages = 0
|
|
|
+
|
|
|
+ for i, card in enumerate(cards):
|
|
|
+ if self.unrelated_data > 15 or self.is_max_count_reached():
|
|
|
+ break
|
|
|
+
|
|
|
+ print(f"\n [{i + 1}/{len(cards)}]")
|
|
|
+ try:
|
|
|
+ card.click()
|
|
|
+ except Exception as e:
|
|
|
+ print(f" [失败] 点击失败: {e}")
|
|
|
+ continue
|
|
|
+ time.sleep(self.SLEEP_CLICK)
|
|
|
+ self._dismiss_popup_if_exists()
|
|
|
+
|
|
|
+ if self.is_on_list_page():
|
|
|
+ print(" [跳过] 点击后仍在列表页")
|
|
|
+ continue
|
|
|
+
|
|
|
+ try:
|
|
|
+ self.integrate_data()
|
|
|
+ except Exception as e:
|
|
|
+ print(f" [异常] 采集详情失败: {e}")
|
|
|
+
|
|
|
+ if not self.back_to_list_page():
|
|
|
+ print(" [警告] 未能回到列表页")
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+
|
|
|
+ if self.end_page and current_page >= self.end_page:
|
|
|
+ completed_normally = self.finish_task_normally(
|
|
|
+ current_page, f"已到结束页 {self.end_page}")
|
|
|
+ break
|
|
|
+
|
|
|
+ print(f"\n 翻到第 {current_page + 2} 页...")
|
|
|
+ self.d.swipe_ext("up", scale=0.5)
|
|
|
+ time.sleep(self.get_sleep_time())
|
|
|
+ current_page += 1
|
|
|
+ self.page = current_page
|
|
|
+
|
|
|
+ # 同步实时状态给后端,更新 UI 上的页数和已采集数量
|
|
|
+ report_api(self.task_id, platform=self.platform, username=getattr(self, 'username', ''), page=self.page, is_finished=0, crawled_count=self.max_counts)
|
|
|
+
|
|
|
+ finally:
|
|
|
+ self._heartbeat_running = False
|
|
|
+
|
|
|
+ if completed_normally is None:
|
|
|
+ completed_normally = False
|
|
|
+ if not completed_normally and not self.finish_reported:
|
|
|
+ self.finish_task_abnormally(current_page, "采集流程异常结束")
|
|
|
+
|
|
|
+ print(f"\n 遍历完成,共处理 {current_page + 1} 页,采集 {self.max_counts} 条")
|
|
|
+ return completed_normally
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================
|
|
|
+# 启动
|
|
|
+# ============================================================
|
|
|
+
|
|
|
+def main():
|
|
|
+ """启动调度器入口(模仿 PDD 模式)。"""
|
|
|
+ interval_seconds = MANUAL_SCHEDULER_INTERVAL_SECONDS if USE_MANUAL_TASKS else SCHEDULER_INTERVAL_SECONDS
|
|
|
+ logging.info(f"XHS 调度器启动,轮询间隔 {interval_seconds} 秒")
|
|
|
+ dispatch_pending_tasks()
|
|
|
+ if USE_MANUAL_TASKS:
|
|
|
+ logging.info(f"手动任务模式:启动定时轮询,间隔 {MANUAL_SCHEDULER_INTERVAL_SECONDS} 秒")
|
|
|
+ if all_manual_tasks_dispatched() and not has_active_workers():
|
|
|
+ logging.info("手动任务模式:全部任务已完成,进程退出")
|
|
|
+ return
|
|
|
+ schedule_dispatch(MANUAL_SCHEDULER_INTERVAL_SECONDS, manual_scheduled_dispatch_job)
|
|
|
+ scheduler_stop_event.wait()
|
|
|
+ return
|
|
|
+ schedule_dispatch(SCHEDULER_INTERVAL_SECONDS)
|
|
|
+ scheduler_stop_event.wait()
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == '__main__':
|
|
|
+ main()
|