2.0 京东

spiders/jd/jd_auto_crawl.py

@@ -10,13 +10,19 @@ from DrissionPage import ChromiumPage, ChromiumOptions
 import json
 import hashlib
 from commons.Logger import get_spider_logger
-from commons.conn_mysql import MySQLPoolOnline
+from commons.conn_mysql import MySQLPoolOn2
 from pipelines.drug_pipelines import DrugPipeline
 from commons.feishu_webhook import send_text
-from spiders.jd.jd_captcha import handle_jd_slider_captcha
 from oss_upload.oss_upload import AliyunOSSUploader
+import threading
+from commons.config import JD_DEVICE_ID
+
+import requests
+
 logger = get_spider_logger("jd")
 
+jd_log = get_spider_logger("jd_log")
+
 chrome_path = r"C:\Program Files\Google\Chrome\Application\chrome.exe"
 
 FETCH_TIMEOUT_FIRST = 5
@@ -35,11 +41,14 @@ return r.bottom > 80 && r.top < h - 40 && r.right > 0 && r.left < w;
 """
 
 
+
+
+
 class JdCrawlerV2:
-    def __init__(self, drug_dict=None):
+    def __init__(self, drug_dict=None,scheduler=None):
         self.driver = None
         self.register_signal_handler()
-        self.db = MySQLPoolOnline()
+        self.db = MySQLPoolOn2()
         self.ip = None
         self.account_name = None
         self.login_username = None
@@ -50,11 +59,20 @@ class JdCrawlerV2:
         self.ossuploader = AliyunOSSUploader()
         self.start_page = 1
         self.end_page = 1
+        self.report_data={}
+        self.heartbeat_interval = 30
+
+        self.scheduler = scheduler 
+
+
+
+
         if self.task_dict:
             self.get_product_data()
         self.success = True
         self.is_no_prodcut = 0
 
+
     def get_product_data(self):
         self.task_id = self.task_dict["id"]
         self.company_id = self.task_dict["company_id"]
@@ -70,8 +88,10 @@ class JdCrawlerV2:
         self.account_id = self.task_dict.get("collect_equipment_account_id", "")
         self.collect_region_id = self.task_dict.get("collect_region_id", "")
         self.collect_round = self.task_dict.get("collect_round", 1)
-        self.start_page = self._parse_page(self.task_dict.get("start_page"), 1)
+        self.start_page = self._parse_page(self.task_dict.get("current_page")+1, 1)
         self.end_page = self._parse_page(self.task_dict.get("end_page"), 15)
+        self.username = self.task_dict.get("username",'')
+        self.report_data = {'task_id':self.task_id,'platform':self.platform,'username':self.task_dict.get("username",JD_DEVICE_ID)}
 
     @staticmethod
     def _parse_page(value, default=1):
@@ -91,7 +111,7 @@ class JdCrawlerV2:
     def init_browser(self):
         co = ChromiumOptions().set_browser_path(chrome_path)
         debug_port = self._get_free_port()
-        co.set_user_data_path(f"./spiders/jd/{self.account_name}")
+        co.set_user_data_path(f"./spiders/jd/test/")
 
         co.set_local_port(debug_port)
         co.set_argument(f"--remote-debugging-port={debug_port}")
@@ -100,6 +120,15 @@ class JdCrawlerV2:
         co.set_argument("--disable-dev-shm-usage")
         co.set_argument("--no-first-run")  # 避免首次运行弹窗
         co.set_argument("--no-default-browser-check")  # 避免默认浏览器检查
+        #co.headless(True)
+        co.set_user_agent('Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36')
+            
+            # 3. 添加反检测启动参数
+        co.set_argument('--disable-blink-features=AutomationControlled')  # 禁用自动化控制标志
+        co.set_argument('--disable-gpu')  # 禁用 GPU（无头模式常用）
+        co.set_argument('--no-sandbox')  # 禁用沙箱
+        co.set_argument('--start-maximized')  # 启动时最大化窗口
+        co.set_argument('--disable-infobars')  # 禁用信息栏提示
 
         if self.ip:
             proxy = self.ip.strip()
@@ -118,6 +147,8 @@ class JdCrawlerV2:
         self._listen_started = True
         logger.info("已启动搜索接口监听")
 
+
+
     def register_signal_handler(self):
         def handler(signum, frame):
             print("\n⚠️ 程序退出")
@@ -312,8 +343,10 @@ class JdCrawlerV2:
             ele_xpath = "//div[@id='main_search_conter']//div[contains(@class,'_goodsContainer_')]/div[@data-sku=" + "'" + sku_id + "'" + "]"
             ele_screen = self.driver.ele("xpath="+ele_xpath)
             upload_key = hashlib.md5(item_url.encode("utf-8")).hexdigest()
-            snap_url = self._take_snapshot(upload_key,ele_screen)
 
+            #后续开启
+            #snap_url = self._take_snapshot(upload_key,ele_screen)
+            snap_url = ''
             try:
                 price = Decimal(str(low_price)).quantize(Decimal("0.00"))
             except (InvalidOperation, ValueError):
@@ -526,9 +559,7 @@ class JdCrawlerV2:
 
         time.sleep(random.uniform(3, 5))
 
-        if not handle_jd_slider_captcha(self.driver):
-            print("滑块验证码未通过")
-            return False
+
 
         return True
 
@@ -580,8 +611,32 @@ class JdCrawlerV2:
             return False
         self.sleep(2, 4)
         return True
+    
+    def _error_report(self,data):
+            self.report_data.update(data)
+            self.scheduler.stop()
+            self.scheduler.post_report(self.report_data)
+
+    def _success_report(self,data):
+            self.report_data.update(data)
+            self.scheduler.post_report(self.report_data)
+
+    def post_report(self,data):
+        url = "http://192.168.2.246:8080/api/collect_task/report"
+        print('传给返回接口的数据',data)
+        headers = {'X-Crawler-Token':'zhijiayun_crawler_2026'}
+        response = requests.post(url, json=data,headers =headers)
+        #记录日志
+        result = response.json()
+
+        if(result.get('code')!='success'):
+            logger.info(f'翻页回传：{result}')
+            self.scheduler.stop()
+            #用守护线程 直接杀死该进程
+        print(f'任务进度上传 {result}')
 
     def crawl(self):
+
         total = 0
         keyword = self._build_search_keyword()
 
@@ -590,12 +645,20 @@ class JdCrawlerV2:
 
         if self._is_logged_out():
             if not self.login_password or not self.login_username:
+                #这里要报账号异常错误，没登录账号
+                self._error_report({'is_finished':0,'need_reassign':1,'current_page':self.start_page,'exception_type':1})
+
                 return
             if not self._ensure_logged_in():
                 self.disable_account()
                 send_text(f"京东：{self.account_name}账号登录失败")
+                jd_log.info(f'出现要求登录:::::{self.driver.html}')
                 self.success = False
+
+                self._error_report({'is_finished':0,'need_reassign':1,'current_page':self.start_page,'exception_type':1})
+
                 return
+            
             self.driver.get("https://www.jd.com/", timeout=15)
             self.sleep(3, 5)
 
@@ -607,32 +670,46 @@ class JdCrawlerV2:
             f"https://search.jd.com/Search?keyword={kw}&enc=utf-8&wq={kw}", timeout=15
         )
         self.sleep(5, 8)
-
-        if self._check_page_blocked():
+        if self._check_page_blocked(): #禁止搜索
+            self._error_report({'is_finished':0,'need_reassign':1,'current_page':self.start_page,'exception_type':2})
             return
 
-        if not handle_jd_slider_captcha(self.driver, pause_listen=False):
-            logger.warning("进入搜索页后滑块验证码处理失败")
-            self.success = False
-            return
+        #这一步捕捉异常：
+        html = self.driver.html or ""
+        if "请刷新或切换账户试试" in html:
+            self._error_report({'is_finished':0,'need_reassign':1,'current_page':page_no,'exception_type':2})
+            return 
+
+
+        total_page = int(re.search(r'(\d+)', self.driver.ele('@class:_pagination_total_1gv36_61', timeout=2).text).group(1))
+        self.report_data.update({'total_pages':total_page})
+        
+
 
         if self.start_page > 1:
             if not self._jump_to_page(self.start_page):
                 logger.warning("跳页失败，将从第 1 页开始采集")
                 self.start_page = 1
 
+                self._error_report({'is_finished':0,'need_reassign':1,'current_page':self.start_page,'exception_type ':4})
+                return    #说明页数可能出错，直接返回结束任务且汇报
+
+
         logger.info(
             "采集页码范围: %s ~ %s（共 %s 页）",
             self.start_page,
-            self.end_page,
+            total_page,
             self.end_page - self.start_page + 1,
         )
 
-        for page_no in range(self.start_page, self.end_page + 1):
+        for page_no in range(self.start_page, total_page+1):
             if self._is_logged_out():
                 if not self._ensure_logged_in():
+                    self._error_report({'is_finished':0,'need_reassign':1,'current_page':page_no,'exception_type':1})
+
                     self.success = False
                     break
+
                 self.driver.get(
                     f"https://search.jd.com/Search?keyword={kw}&enc=utf-8&wq={kw}",
                     timeout=15,
@@ -641,17 +718,17 @@ class JdCrawlerV2:
                 if page_no > 1:
                     self._jump_to_page(page_no)
 
-            if not handle_jd_slider_captcha(self.driver, pause_listen=True):
-                logger.warning("滑块验证码处理失败，停止采集")
-                self.success = False
-                break
+
 
             if self._check_page_blocked():
                 break
 
+            self._success_report({'is_finished':0,'need_reassign':0,'current_page':page_no})
+
             logger.info("===== 正在爬取第 %s 页 =====", page_no)
             search_ele = self.driver.ele("xpath=//div[@id='search-condition']", timeout=10)
             if not search_ele:
+                self._error_report({'is_finished':0,'need_reassign':1,'current_page':page_no,'exception_type':2})
                 logger.warning("未找到搜索结果区域，停止采集")
                 break
 
@@ -660,28 +737,54 @@ class JdCrawlerV2:
             total += page_n
             logger.info("累计监听条数: %s", total)
 
+            if self.scheduler.end ==True:
+                logger.info('心跳失败')
+                self._error_report({'is_finished':0,'need_reassign':0,'current_page':page_no})
+                break
             if self.is_no_prodcut > 20:
+                
                 logger.info("连续无匹配商品过多，停止采集")
+                self._success_report({'is_finished':1,'need_reassign':0,'current_page':page_no})
                 break
 
             if page_no >= self.end_page:
+                self._success_report({'is_finished':1,'need_reassign':0,'current_page':page_no})
                 break
 
             next_btn = self.driver.ele("text=下一页", timeout=2)
             if not next_btn:
                 logger.info("没有下一页（未找到）")
+                self._success_report({'is_finished':1,'need_reassign':0,'current_page':page_no})
                 break
+
             cls_str = next_btn.attr("class") or ""
             if "disabled" in cls_str:
                 logger.info("没有下一页（已禁用）")
+                self._success_report({'is_finished':1,'need_reassign':0,'current_page':page_no})
+                                
                 break
 
             if not self._go_next_page(next_btn):
+                self._error_report({'is_finished':0,'need_reassign':1,'current_page':page_no,'exception_type':4})
                 break
+            if page_no==total_page:  #达到最大页数
+                self._success_report({'is_finished':1,'need_reassign':0,'current_page':page_no})                
+                break
+        self.scheduler.stop()
+        
+            
+
 
     def run(self):
         # 检测账号
-        if not self.get_account():
+
+        # 启动守护线程
+        # self._daemon_thread = threading.Thread(target=self._heartbeat_reporter)
+        # self._daemon_thread.daemon = True  # 守护线程，主线程退出时自动结束
+        # self._daemon_thread.start()
+
+        #if  self.get_account():
+        if 0:
             logger.info("==================当前无账号可用==================")
             self.success = False
             return self.pipeline.crawl_count, self.success
@@ -698,7 +801,7 @@ class JdCrawlerV2:
             self.success = False
             logger.exception("爬取异常: %s", e)
             self.sleep(3, 5)
-
+            self._error_report({'is_finished':0,'need_reassign':1,'current_page':self.start_page,'exception_type':3})
         finally:
             if self.driver:
                 self.driver.quit()

start_run_jd.py

@@ -2,7 +2,10 @@ import json
 import time
 import requests
 from commons.Logger import logger
-from commons.conn_mysql import MySQLPoolOnline, MySQLPool39
+
+from commons.scheduler import CrawlerScheduler
+
+from commons.conn_mysql import MySQLPoolOn2, MySQLPool39
 from spiders.jd.jd_auto_crawl import JdCrawlerV2
 import schedule
 from commons.feishu_webhook import send_text
@@ -14,10 +17,10 @@ platform_name = "京东"
 class JdMain:
     def __init__(self):
         # self.db_online = MySQLPool39()
-        self.db_online = MySQLPoolOnline()
+        self.db_online = MySQLPoolOn2()
         self.crawl_count = ""
         self.task_id = ""
-        self.task_dict = self.get_task()
+        #self.task_dict = self.get_task()
 
     def get_status(self, status):
         if status not in (2, 3, 4):
@@ -57,7 +60,7 @@ class JdMain:
                 ON t.`collect_equipment_account_id` = a.`id`
             WHERE a.`device_id` = %s
               AND t.`platform` = 2
-              AND t.`status` = 1
+              AND t.`status` = 3
             LIMIT 1
         """
         task_list = self.db_online.select_data(sql, (JD_DEVICE_ID,))
@@ -70,11 +73,31 @@ class JdMain:
         print(task_dict)
         return task_dict
 
+    def get_task(self):
+        task_api = "http://192.168.2.246:8080/api/collect_task/pull"
+
+        headers = {'X-Crawler-Token':'zhijiayun_crawler_2026'}
+        
+        params = {
+        'platform': 2,
+        'username': JD_DEVICE_ID
+        }
+        response = requests.get(task_api, params=params,headers=headers)
+        result = response.json()
+        if result.get('code') =='success':
+            return result.get('data').get('task')
+        
+        logger.info(f'拉取任务：{result}')
+        print('拉取任务返回',result)
+
+
     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("心跳任务上报成功")
@@ -82,22 +105,49 @@ class JdMain:
             logger.info("心跳任务上报失败")
 
     def run(self):
-        self.task_dict = self.get_task()
-        if not self.task_dict:
-            logger.info(f"{platform_name}暂无任务")
-            return
 
+        spider_schedule =  CrawlerScheduler(JD_DEVICE_ID,2)
+        spider_schedule.start()
+        time.sleep(3)
+
+        while 1 :
+            print(f'not spider_schedule.end:{not spider_schedule.end}')
+            if(not spider_schedule.end):
+                print('进入工作')
+
+                self.task_dict = spider_schedule.get_task()
+                
+                #self.task_dict = {'id': 4, 'company_id': 0, 'product_name': '藿香正气合剂', 'product_specs': '', 'product_brand': '999', 'product_keyword': '', 'current_page': 0, 'start_offset': 0}
+
+                print(self.task_dict)
+
+
+
+                if not self.task_dict:
+                    logger.info(f"{platform_name}暂无任务")
+                    time.sleep(35)
+                    continue
+
+                
+                # self.get_status(2)
+                self.crawl_count, is_success = JdCrawlerV2(self.task_dict,spider_schedule).run()
+                spider_schedule.stop()
+            else:
+
+                spider_schedule.start()
+                print('休息')
+                time.sleep(35)
 
-        self.get_status(2)
-        self.crawl_count, is_success = JdCrawlerV2(self.task_dict).run()
+            
+            time.sleep(30)
 
-        self.heartbeat_task()
-        # send_text(
-        #     f"""{str(time.strftime("%Y-%m-%d %H:%M:%S"))} 通知:\n平台: {platform_name}, 药品: {self.task_dict.get("product_brand") + " " + self.task_dict.get("product_name") + " " + self.task_dict.get("product_specs")},  爬取数据: {self.crawl_count}条""")
-        if is_success:
-            self.get_status(3)
-        else:
-            self.get_status(4)
+        # self.heartbeat_task()
+        # # send_text(
+        # #     f"""{str(time.strftime("%Y-%m-%d %H:%M:%S"))} 通知:\n平台: {platform_name}, 药品: {self.task_dict.get("product_brand") + " " + self.task_dict.get("product_name") + " " + self.task_dict.get("product_specs")},  爬取数据: {self.crawl_count}条""")
+        # if is_success:
+        #     self.get_status(3)
+        # else:
+        #     self.get_status(4)
 
 
 if __name__ == '__main__':
@@ -109,6 +159,7 @@ if __name__ == '__main__':
     # 每10分钟执行一次
     while True:
         JdMain().run()
-        interval_time = random.randint(1200, 1800)
+        interval_time = random.randint(60, 180)
+        interval_time = 60
         logger.info(f"程序睡眠{interval_time}秒后继续执行")
         time.sleep(interval_time)