Merge remote-tracking branch 'origin/master'

# Conflicts:
#	mt_V2/db.py

mt_V2/db.py

@@ -4,23 +4,26 @@ import threading
 import time
 import atexit
 from collections import deque
+
+
+# simple direct connector (used by some scripts)
 from pymysql.cursors import DictCursor
 
-# ============================================================
-# 数据库配置
-# ============================================================
-DB_HOST = "120.24.26.108"
-DB_PORT = 3307
-DB_USER = "collect_user"
-DB_PASSWORD = "collect123456"
-DB_NAME = "drug_retrieve"      # 正式库；测试库改 drug_retrieve_test
-DB_CHARSET = "utf8mb4"
-
-# ============================================================
-# 连接池
-# ============================================================
+def get_connection():
+    return pymysql.connect(
+        host="120.24.26.108",
+        port=3307,
+        user="collect_user",
+        password="collect123456",
+        database="drug_retrieve",
+        charset="utf8mb4",
+        autocommit=False,
+        cursorclass=DictCursor,
+    )
+
+
+# connection pool used by a_mt.py
 MYSQL_POOL = None
-POOL_LOCK = threading.Lock()
 
 
 class PooledConnection:
@@ -134,49 +137,39 @@ class MySQLConnectionPool:
             self._total = 0
 
 
-def _pool_kwargs():
-    """连接池参数"""
-    return {
-        "host": DB_HOST,
-        "port": DB_PORT,
-        "user": DB_USER,
-        "password": DB_PASSWORD,
-        "database": DB_NAME,
-        "charset": DB_CHARSET,
-        "cursorclass": pymysql.cursors.Cursor,
-    }
-
-
 def get_mysql():
-    """获取数据库连接（优先连接池，失败则直连）"""
     global MYSQL_POOL
-    with POOL_LOCK:
-        if MYSQL_POOL is None:
-            try:
-                MYSQL_POOL = MySQLConnectionPool(minconn=1, maxconn=20, **_pool_kwargs())
-            except Exception:
-                MYSQL_POOL = None
+    if 'MYSQL_POOL' not in globals() or MYSQL_POOL is None:
+        MYSQL_POOL = None
+        try:
+            MYSQL_POOL = MySQLConnectionPool(
+                minconn=1,
+                maxconn=20,
+                host="120.24.26.108",
+                port=3307,
+                user="root",
+                password="zhijiayun123456",
+                database="drug_retrieve",
+                charset="utf8mb4",
+                cursorclass=pymysql.cursors.Cursor,
+            )
+        except Exception:
+            MYSQL_POOL = None
     if MYSQL_POOL is None:
-        return pymysql.connect(**_pool_kwargs())
+        return pymysql.connect(
+            host="120.24.26.108",
+            port=3307,
+            user="root",
+            password="zhijiayun123456",
+            database="drug_retrieve",
+            charset="utf8mb4",
+            cursorclass=pymysql.cursors.Cursor,
+        )
     return MYSQL_POOL.getconn()
 
 
-def get_connection():
-    """直连数据库（兼容旧脚本，DictCursor 模式）"""
-    return pymysql.connect(
-        host=DB_HOST,
-        port=DB_PORT,
-        user=DB_USER,
-        password=DB_PASSWORD,
-        database=DB_NAME,
-        charset=DB_CHARSET,
-        autocommit=False,
-        cursorclass=DictCursor,
-    )
-
-
 # 在程序退出时确保池中所有物理连接被关闭
 try:
-    atexit.register(lambda: MYSQL_POOL.closeall() if MYSQL_POOL is not None else None)
+    atexit.register(lambda: MYSQL_POOL.closeall() if 'MYSQL_POOL' in globals() and MYSQL_POOL is not None else None)
 except Exception:
-    pass
+    pass

mt_V2/scheduler.py

@@ -0,0 +1,140 @@
+import time
+import requests
+from commons.Logger import get_spider_logger
+import threading
+import urllib3
+
+# 禁用SSL警告
+urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
+logger = get_spider_logger('scheduler')
+
+
+class CrawlerScheduler:
+    """爬虫任务调度器"""
+
+    def __init__(self, DEVICE_ID, platform, heartbeat_interval=30):
+        """
+        初始化调度器
+
+        Args:
+            platform: 平台名称
+            heartbeat_url: 心跳上报URL
+            heartbeat_interval: 心跳间隔时间（秒）
+        """
+
+        self.username = DEVICE_ID
+        self.platform = platform
+
+        self._lock = threading.Lock()
+
+        # ⭐ 关键修改：将 http:// 改为 https://
+        self.heartbeat_url = 'https://120.24.26.108:8083/api/collect_task/heartbeat'
+        self.heartbeat_interval = heartbeat_interval
+        self.end = False
+        self.heartbeat_thread = None
+
+    def _heartbeat_reporter(self):
+        """守护线程：只负责上报心跳"""
+        headers = {'X-Crawler-Token': 'zhijiayun_crawler_2026'}
+        while True:
+            try:
+                # 使用 HTTPS，并忽略证书验证
+                response = requests.post(
+                    self.heartbeat_url,
+                    json={
+                        "platform": self.platform,
+                        "username": self.username
+                    },
+                    headers=headers,
+                    timeout=3,
+                    verify=False   # 忽略SSL证书验证
+                )
+
+                print(f"[心跳] 发送成功: {response.status_code}")
+                result = response.json()
+
+                if self.end or result.get('code') != 'success':
+                    logger.error(f'心跳回传：{result}')
+                    self.set_flag(True)
+                    break
+
+                logger.info(f'心跳回传：{result}')
+                self.set_flag(False)
+
+                time.sleep(self.heartbeat_interval)
+
+            except Exception as e:
+                logger.error(e)
+                print(e)
+                time.sleep(5)
+
+    def start(self):
+        """启动调度器"""
+        self.set_flag(False)
+        self.heartbeat_thread = threading.Thread(
+            target=self._heartbeat_reporter,
+            daemon=True
+        )
+        self.heartbeat_thread.start()
+
+    def stop(self):
+        """停止调度器"""
+        self.set_flag(True)
+        logger.info('心跳停止')
+
+    def get_task(self):
+        try:
+            # ⭐ 关键修改：将 http:// 改为 https://
+            task_api = "https://120.24.26.108:8083/api/collect_task/pull"
+            headers = {'X-Crawler-Token': 'zhijiayun_crawler_2026'}
+            params = {
+                'platform': self.platform,
+                'username': self.username
+            }
+            response = requests.get(
+                task_api,
+                params=params,
+                headers=headers,
+                timeout=5,
+                verify=False   # 忽略SSL证书验证
+            )
+            result = response.json()
+            logger.info(f'拉取任务：{result}')
+            if result.get('code') == 'success':
+                return result.get('data').get('task')
+
+            print('拉取任务返回', result)
+
+        except Exception as e:
+            logger.error('获取任务报错', e)
+
+    def post_report(self, data):
+        try:
+            # ⭐ 关键修改：将 http:// 改为 https://
+            url = "https://120.24.26.108:8083/api/collect_task/report"
+            print('传给返回接口的数据', data)
+            logger.info(f'report上传数据：{data}')
+            headers = {'X-Crawler-Token': 'zhijiayun_crawler_2026'}
+            response = requests.post(
+                url,
+                json=data,
+                headers=headers,
+                timeout=5,
+                verify=False   # 忽略SSL证书验证
+            )
+            result = response.json()
+            logger.info(result)
+
+            if result.get('code') != 'success':
+                logger.error(f'翻页回传结果不成功：{result}')
+                self.stop()
+            print(f'任务进度上传 {result}')
+            return result
+        except Exception as e:
+            logger.error(e)
+            return None
+
+    def set_flag(self, value):
+        with self._lock:
+            self.end = value