"""药九九统一入口 — 根据任务 snapshot_collect_status 字段自动切换普通/快照模式。 对接 CrawlerScheduler API(拉取任务/心跳/进度上报), snapshot_collect_status=0 → 快照模式(Playwright截图),非0 → 普通API模式。 """ import random import time import json from spiders.yaojiujiu.yyjzt_crawl import YyjztSpider from spiders.yaojiujiu.yyjzt_snapshot import YyjztSnapshotSpider from commons.scheduler import CrawlerScheduler from commons.conn_mysql import MySQLPoolOn2 from commons.Logger import logger PLATFORM_NAME = "药九九" PLATFORM_ID = 7 def get_device_config(): """从 accounts_platform 获取药九九账号(platform=7, status=1)。""" db = MySQLPoolOn2() sql = """SELECT `name`, `phone`, `password` FROM `accounts_platform` WHERE `platform`=7 AND `status`=1 ORDER BY `cookie_timestamp` ASC LIMIT 1""" rows = db.select_data(sql) if not rows: logger.error("无可用药九九账号(platform=7, status=1)") return None account = rows[0] logger.info("获取到药九九账号: name=%s, phone=%s", account["name"], account["phone"]) return account["phone"], account["password"], account["name"] def _build_task_dict(api_task, device_id, account_password, account_name, scheduler): """组装 spider 需要的 task_dict。""" current_page = api_task.get("current_page", 0) task_dict = { "id": api_task.get("id"), "company_id": api_task.get("company_id"), "product_name": api_task.get("product_name", ""), "product_specs": api_task.get("product_specs", ""), "product_brand": api_task.get("product_brand", ""), "product_keyword": api_task.get("product_keyword", ""), "current_page": current_page, "start_page": current_page if current_page > 0 else 1, "end_page": 0, "scheduler": scheduler, "_device_id": device_id, "_platform": str(PLATFORM_ID), "collect_task_id": str(api_task.get("id", "")), "snapshot_collect_status": api_task.get("snapshot_collect_status"), "sampling_cycle": "", "sampling_start_time": "", "sampling_end_time": "", "collect_equipment_account_id": "", "collect_region_id": "", "collect_equipment_id": "", "collect_round": 1, # 登录账号信息(spider 内检查登录状态用) "account_phone": device_id, "account_password": account_password, "account_name": account_name, } return task_dict def _report_progress(scheduler, task_id, is_finished, crawled_count=0, current_page=0, total_pages=0, exception_type=None, remark=None): data = { "task_id": task_id, "platform": str(PLATFORM_ID), "username": scheduler.username, "current_page": current_page, "is_finished": is_finished, "total_pages": total_pages, } data["crawled_count"] = crawled_count if exception_type is not None: data["exception_type"] = exception_type if remark: data["remark"] = remark try: scheduler.post_report(data) logger.info("任务上报成功: is_finished=%s, crawled_count=%s", is_finished, crawled_count) except Exception as e: logger.warning("任务上报失败: %s", e) def run_one_task(scheduler, device_id, account_password, account_name): """拉取一个任务,根据 snapshot_collect_status 选 spider 执行并上报。""" api_task = scheduler.get_task() if not api_task: logger.info("%s 暂无任务", PLATFORM_NAME) return task_id = api_task.get("id", "") task_dict = _build_task_dict(api_task, device_id, account_password, account_name, scheduler) log_dict = {k: v for k, v in task_dict.items() if k != "scheduler"} logger.info("拉取到任务: %s", json.dumps(log_dict, ensure_ascii=False)) # 根据 snapshot_collect_status 决定走普通爬虫还是快照爬虫 snapshot_status = task_dict.get("snapshot_collect_status", 0) if not snapshot_status: spider_cls = YyjztSnapshotSpider logger.info("任务 %s 启用快照模式", task_id) else: spider_cls = YyjztSpider logger.info("任务 %s 使用普通模式", task_id) try: result = spider_cls(task_dict).run() except Exception as e: logger.exception("任务 %s 执行异常: %s", task_id, e) _report_progress(scheduler, task_id, is_finished=0, crawled_count=0, exception_type=5, remark=str(e)[:200]) return if len(result) == 3: crawl_count, is_success, total_pages = result else: crawl_count, is_success = result total_pages = 0 if is_success: _report_progress(scheduler, task_id, is_finished=1, crawled_count=crawl_count, current_page=total_pages, total_pages=total_pages) else: _report_progress(scheduler, task_id, is_finished=0, crawled_count=crawl_count, current_page=total_pages, total_pages=total_pages) if __name__ == "__main__": # 1. 获取账号 result = get_device_config() if result: device_id, account_password, account_name = result else: device_id, account_password, account_name = "YYJZT", None, None # 2. 启动调度器 scheduler = CrawlerScheduler(DEVICE_ID=device_id, platform=str(PLATFORM_ID)) scheduler.start() time.sleep(2) logger.info("药九九调度器已启动, device_id=%s", device_id) # 3. 循环执行 idle_seconds = random.randint(180, 300) logger.info("药九九循环任务已启动, 间隔 %s 秒", idle_seconds) while True: try: logger.info("开始执行药九九爬虫") run_one_task(scheduler, device_id, account_password, account_name) logger.info("药九九爬虫执行完成") except Exception as e: logger.error("药九九爬虫执行失败: %s", e, exc_info=True) time.sleep(idle_seconds)