import json import time import requests from commons.Logger import logger from commons.conn_mysql import MySQLPoolOnline, MySQLPool39 from spiders.jd.jd_auto_crawl_snap2 import JdCrawlerV2 from commons.scheduler import CrawlerScheduler from commons.feishu_webhook import send_text import random from commons.config import JD_DEVICE_ID platform_name = "京东" class JdMain: def __init__(self): # self.db_online = MySQLPool39() self.db_online = MySQLPoolOnline() self.crawl_count = "" self.task_id = "" self.task_dict = None self.driver = None self.cumulative_pages = 0 self.cumulative_items = 0 self.cumulative_stored = 0 self.cumulative_skipped = 0 def get_status(self, status): if status not in (2, 3, 4): logger.warning(f"未知状态值: {status}, 跳过状态上报") return if status == 2: parmas = { "collect_task_allocate_id": self.task_id, "status": status, "finish_status": 0, "start_time": int(time.time()) } if status == 3: parmas = { "collect_task_allocate_id": self.task_id, "status": status, "finish_status": 1, "real_count": self.crawl_count, "end_time": int(time.time()), } if status == 4: parmas = { "collect_task_allocate_id": self.task_id, "status": status, "finish_status": 0, "end_time": int(time.time())} # url = "http://scheduletest.dfwy.tech/api/collect_equipment_execute/result_report" url = "http://scheduleapi.findit.ltd/api/collect_equipment_execute/result_report" try: res = requests.get(url, params=parmas, timeout=20) res.raise_for_status() logger.info("状态上报: %s", res.text) except Exception as e: logger.warning(f"状态上报失败: {e}") def get_task(self): """获取当前设备绑定的京东待执行快照任务。""" sql = """ SELECT t.* FROM `retrieve_collect_task_allocate` t INNER JOIN `retrieve_collect_equipment_account` a ON t.`collect_equipment_account_id` = a.`id` WHERE t.`platform` = 2 AND t.`status` = 1 AND t.`snapshot_collect_status` = 0 LIMIT 1 """ task_list = self.db_online.select_data(sql) print(task_list) if not task_list: return {} task_dict = task_list[0] self.task_id = task_dict["id"] print(task_dict) return task_dict def heartbeat_task(self): url = "https://scheduleapi.findit.ltd/api/collect_equipment_execute/heartbeat" params = { "collect_task_allocate_id": self.task_id, } try: res = requests.get(url, params=params, timeout=20) logger.info("心跳任务上报成功") except Exception as e: logger.info("心跳任务上报失败") def run(self): spider_schedule = CrawlerScheduler(JD_DEVICE_ID, 2) spider_schedule.start() time.sleep(3) while 1: if not spider_schedule.end: self.task_dict = spider_schedule.get_task() if not self.task_dict: logger.info(f"{platform_name}暂无任务") time.sleep(35) continue self.task_id = self.task_dict.get("id", "") self.crawl_count, is_success, self.driver, self.cumulative_pages, self.cumulative_items, self.cumulative_stored, self.cumulative_skipped = JdCrawlerV2(self.task_dict, spider_schedule, self.driver, self.cumulative_pages, self.cumulative_items, self.cumulative_stored, self.cumulative_skipped).run() spider_schedule.stop() else: spider_schedule.start() print('休息') time.sleep(35) time.sleep(10) if __name__ == '__main__': while True: JdMain().run() interval_time = random.randint(180, 300) logger.info(f"程序睡眠{interval_time}秒后继续执行") time.sleep(interval_time)