淘宝 京东 药房网

spider/requirements.txt

@@ -1,14 +0,0 @@
-requests>=2.28
-DrissionPage>=4.0
-Pillow>=10.0
-pandas>=2.0
-pymysql>=1.0
-DBUtils>=3.0
-lxml>=4.9
-oss2>=2.18
-pycryptodome>=3.19
-python-dotenv>=1.0
-schedule>=1.2
-playwright>=1.40
-curl_cffi>=0.7
-openpyxl>=3.1

spider/snapshot_jd.py

@@ -1,123 +0,0 @@
-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)

spider/snapshot_taobao.py

@@ -1,125 +0,0 @@
-import json
-import random
-import time
-import requests
-from commons.Logger import logger
-from commons.conn_mysql import MySQLPoolOnline, MySQLPool39
-from spiders.taobao.snapshot_taobao_crawl2 import TaobaoCrawl
-from commons.scheduler import CrawlerScheduler
-from commons.feishu_webhook import send_text
-from commons.config import TB_DEVICE_ID
-
-platform_name = "淘宝"
-
-
-class TaobaoMain:
-    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[:500])
-        except Exception as e:
-            logger.warning(f"状态上报失败: {e}")
-
-    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(f"心跳任务上报失败{str(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 a.`device_id` = %s
-              AND t.`platform` = 1
-              AND t.`status` = 1
-              AND t.`snapshot_collect_status` = 0
-            LIMIT 1
-        """
-        task_list = self.db_online.select_data(sql, (TB_DEVICE_ID,))
-
-        if not task_list:
-            return {}
-
-        task_dict = task_list[0]
-        self.task_id = task_dict["id"]
-        return task_dict
-
-    def run(self):
-
-        spider_schedule = CrawlerScheduler(TB_DEVICE_ID, 1)
-        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()
-
-                if not self.task_dict:
-                    logger.info(f"{platform_name}暂无任务")
-                    time.sleep(35)
-                    continue
-
-                print(self.task_dict)
-                self.task_dict.update({'collect_equipment_account_id': TB_DEVICE_ID})
-                self.crawl_count, is_success, self.driver, self.cumulative_pages, self.cumulative_items, self.cumulative_stored, self.cumulative_skipped = TaobaoCrawl(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(30)
-
-
-if __name__ == '__main__':
-    # 每10分钟执行一次
-    while True:
-        TaobaoMain().run()
-        interval_time = random.randint(1200, 1800)
-        logger.info(f"程序睡眠{interval_time}秒后继续执行")
-        time.sleep(interval_time)

spider/start_run_jd.py

@@ -1,165 +0,0 @@
-import json
-import time
-import requests
-from commons.Logger import logger
-
-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
-import random
-from commons.config import JD_DEVICE_ID
-platform_name = "京东"
-
-
-class JdMain:
-    def __init__(self):
-        # self.db_online = MySQLPool39()
-        self.db_online = MySQLPoolOn2()
-        self.crawl_count = ""
-        self.task_id = ""
-        #self.task_dict = self.get_task()
-
-    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 a.`device_id` = %s
-              AND t.`platform` = 2
-              AND t.`status` = 3
-            LIMIT 1
-        """
-        task_list = self.db_online.select_data(sql, (JD_DEVICE_ID,))
-        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 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("心跳任务上报成功")
-        except Exception as e:
-            logger.info("心跳任务上报失败")
-
-    def run(self):
-
-        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)
-
-            
-            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)
-
-
-if __name__ == '__main__':
-    # task_dict= {"id": 1622, 'collect_task_id': 4596, 'company_id': 8, 'product_name': '小儿氨酚烷胺颗粒',
-    # 'product_specs': '', 'product_keyword': '', 'product_brand': '可复美', 'sampling_cycle': 1,
-    # 'sampling_start_time': 1778083200, 'sampling_end_time': 1778342399, 'collect_equipment_account_id': 15,
-    # 'collect_region_id': 0, 'collect_equipment_id': 25, 'collect_round': 2,"start_page":4,"end_page":10}
-    # JdCrawlerV2(task_dict).run()
-    # 每10分钟执行一次
-    while True:
-        JdMain().run()
-        interval_time = random.randint(60, 180)
-        interval_time = 60
-        logger.info(f"程序睡眠{interval_time}秒后继续执行")
-        time.sleep(interval_time)

spider/start_run_taobao.py

@@ -1,126 +0,0 @@
-import random
-import time
-import requests
-from commons.Logger import logger
-from commons.conn_mysql import MySQLPoolOnline
-from spiders.taobao.taobao_crawl import TaobaoCrawl
-from commons.scheduler import CrawlerScheduler
-from commons.config import TB_DEVICE_ID
-
-platform_name = "淘宝"
-TB_DEVICE_ID = '17'
-
-class TaobaoMain:
-    def __init__(self):
-        # self.db_online = MySQLPool39()
-        self.db_online = MySQLPoolOnline()
-        self.crawl_count = ""
-        self.task_id = ""
-        self.task_dict = None
-
-    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[:500])
-        except Exception as e:
-            logger.warning(f"状态上报失败: {e}")
-
-    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(f"心跳任务上报失败{str(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 a.`device_id` = %s
-              AND t.`platform` = 1
-              AND t.`status` = 1
-            LIMIT 1
-        """
-        task_list = self.db_online.select_data(sql, (TB_DEVICE_ID,))
-
-        if not task_list:
-            return {}
-
-        task_dict = task_list[0]
-        self.task_id = task_dict["id"]
-        return task_dict
-
-    def run(self):
-
-        spider_schedule =  CrawlerScheduler(TB_DEVICE_ID,1)
-        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}
-
-
-
-                if not self.task_dict:
-
-
-                    logger.info(f"{platform_name}暂无任务")
-                    time.sleep(35)
-                    continue
-
-                print(self.task_dict)
-                self.task_dict.update({'collect_equipment_account_id':TB_DEVICE_ID})
-                # self.get_status(2)
-                self.crawl_count, is_success = TaobaoCrawl(self.task_dict,spider_schedule).run()
-                spider_schedule.stop()
-            else:
-
-                spider_schedule.start()
-                print('休息')
-                time.sleep(35)
-
-            
-            time.sleep(30)
-
-
-if __name__ == '__main__':
-    # 每10分钟执行一次
-    while True:
-        TaobaoMain().run()
-        interval_time = random.randint(1200, 1800)
-        logger.info(f"程序睡眠{interval_time}秒后继续执行")
-        time.sleep(interval_time)