updata

mt_V2/commons/Logger.py

@@ -3,7 +3,24 @@ from logging.handlers import TimedRotatingFileHandler
 import os
 from datetime import datetime
 
-LOG_FORMAT = "%(asctime)s [%(levelname)s] %(message)s"
+# 当前设备ID（由入口脚本通过 set_device_id 设置，日志格式里引用）
+CURRENT_DEVICE_ID = "unknown"
+
+
+def set_device_id(device_id):
+    """入口脚本调用：设置当前设备ID，之后所有日志自动带上"""
+    global CURRENT_DEVICE_ID
+    CURRENT_DEVICE_ID = str(device_id or "unknown")
+
+
+class DeviceIdFilter(logging.Filter):
+    """运行时读取模块级 CURRENT_DEVICE_ID，注入每条日志"""
+    def filter(self, record):
+        record.device_id = CURRENT_DEVICE_ID
+        return True
+
+
+LOG_FORMAT = "%(asctime)s [%(levelname)s] [%(device_id)s] %(message)s"
 
 
 def _build_logger(name: str, log_file: str) -> logging.Logger:
@@ -20,9 +37,12 @@ def _build_logger(name: str, log_file: str) -> logging.Logger:
     file_handler = logging.FileHandler(log_file, encoding="utf-8")
     file_handler.setLevel(logging.DEBUG)
 
+    _device_filter = DeviceIdFilter()
     formatter = logging.Formatter(LOG_FORMAT)
     console_handler.setFormatter(formatter)
     file_handler.setFormatter(formatter)
+    console_handler.addFilter(_device_filter)
+    file_handler.addFilter(_device_filter)
 
     logger_obj.addHandler(console_handler)
     logger_obj.addHandler(file_handler)

mt_V2/db.py

@@ -4,26 +4,23 @@ import threading
 import time
 import atexit
 from collections import deque
-
-
-# simple direct connector (used by some scripts)
 from pymysql.cursors import DictCursor
 
-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
+# ============================================================
+# 数据库配置
+# ============================================================
+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"
+
+# ============================================================
+# 连接池
+# ============================================================
 MYSQL_POOL = None
+POOL_LOCK = threading.Lock()
 
 
 class PooledConnection:
@@ -137,39 +134,49 @@ 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
-    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
+    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 is None:
-        return pymysql.connect(
-            host="120.24.26.108",
-            port=3307,
-            user="root",
-            password="zhijiayun123456",
-            database="drug_retrieve",
-            charset="utf8mb4",
-            cursorclass=pymysql.cursors.Cursor,
-        )
+        return pymysql.connect(**_pool_kwargs())
     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' in globals() and MYSQL_POOL is not None else None)
+    atexit.register(lambda: MYSQL_POOL.closeall() if MYSQL_POOL is not None else None)
 except Exception:
-    pass
+    pass

mt_V2/logger.py

@@ -1,30 +1,36 @@
 # mt_spider/logger.py
 import os
 import logging
-from logging.handlers import TimedRotatingFileHandler
+from datetime import datetime
 
 LOG_DIR = os.path.join(os.path.dirname(__file__), '..', 'logs')
 os.makedirs(LOG_DIR, exist_ok=True)
 
+
 def setup_logger(name: str = "mt_spider"):
-    logger = logging.getLogger()
+    """初始化日志，每个 name 独立的日志文件，每天一个新文件，不旋转（避免 Windows 多进程文件锁冲突）"""
+    logger = logging.getLogger(name)
     logger.setLevel(logging.INFO)
     fmt = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
 
     console = logging.StreamHandler()
     console.setFormatter(fmt)
 
-    file_handler = TimedRotatingFileHandler(
-        filename=os.path.join(LOG_DIR, f'{name}.log'),
-        when='midnight',
-        interval=1,
-        encoding='utf-8',
-        backupCount=0  # 不删除历史
-    )
-    file_handler.suffix = "%Y-%m-%d"
+    today = datetime.now().strftime('%Y-%m-%d')
+    log_file = os.path.join(LOG_DIR, f'{name}_{today}.log')
+    file_handler = logging.FileHandler(log_file, encoding='utf-8')
     file_handler.setFormatter(fmt)
 
-    if not logger.handlers:          # 防止重复
+    if not logger.handlers:
         logger.addHandler(console)
         logger.addHandler(file_handler)
+
+    # 同时给根 logger 也加上文件 handler（保证 logging.info() 直接调用也能写文件）
+    root = logging.getLogger()
+    root.setLevel(logging.INFO)
+    # 防止重复添加
+    existing_files = {h.baseFilename for h in root.handlers if hasattr(h, 'baseFilename')}
+    if log_file not in existing_files:
+        root.addHandler(file_handler)
+
     return logger

mt_V2/main.py

@@ -773,7 +773,7 @@ class SpiderMonitor(threading.Thread):
 
 
 class MTScreenshot:
-    def __init__(self, d, oss_config, search_key, title_key, scroll_times=1, compress_quality=7, resize_ratio=0.8,device_id=None,
+    def __init__(self, d, oss_config, search_key, title_key, scroll_times=2, compress_quality=7, resize_ratio=0.8,device_id=None,
                  monitor=None):
         self.device_id = device_id
         # 接收外部已连接好的u2设备实例
@@ -993,6 +993,10 @@ class MTScreenshot:
 
 
 class MT:
+    # 封号检测：类变量，跨实例共享，不因重建实例而重置
+    open_page_fail_count = 0
+    open_page_fail_threshold = 6
+
     def __init__(
             self,
             key,
@@ -1079,6 +1083,7 @@ class MT:
         self.request_error_threshold = REQUEST_ERROR_STOP_THRESHOLD
         self.product_link_missing_count = 0
         self.product_link_missing_threshold = PRODUCT_LINK_STOP_THRESHOLD
+        # 封号检测：open_product_list_page 连续失败计数（类变量，跨实例共享，不因重建实例而重置）
         # 风控卡死检测："加载更多"按钮检测
         self.load_more_check_rounds = 4
         self.load_more_check_interval = 30
@@ -1354,12 +1359,18 @@ class MT:
         normalized_title_key = normalize_match_text(self.title_key)
         normalized_brand = normalize_match_text(self.brand)
 
+        # 1. 药品名必须在标题里
         if normalized_title_key != "" and normalized_title_key not in normalized_title:
             print(f"当前商品名称:{product_title} 不包含{self.title_key}关键字")
             return False
-        if normalized_brand != "" and normalized_brand not in normalized_title:
-            print(f"当前商品名称:{product_title} 不包含{self.brand}品牌")
+
+        # 2. 把药品名从标题里去掉后再检查品牌
+        #    （避免品牌是通用名子串的干扰，如品牌"午时" ⊂ 药名"午时茶颗粒"）
+        check_target = normalized_title.replace(normalized_title_key, "") if normalized_title_key else normalized_title
+        if normalized_brand != "" and normalized_brand not in check_target:
+            print(f"当前商品名称:{product_title} 去掉药名后不包含{self.brand}品牌")
             return False
+
         if not self.is_link_spec_useful(product_title):
             print(f"当前商品名称:{product_title} 不包含{self.spec_list}品规")
             return False
@@ -3617,6 +3628,7 @@ class MT:
             return "open_product_list_page"
         if step_name == "open_product_list_page":
             self.safe_exec(self.open_product_list_page)
+            MT.open_page_fail_count = 0  # 成功则重置封号计数
             return "collect_single_product"
         if step_name == "collect_single_product":
             has_next = self.safe_exec(self.collect_single_product)
@@ -3657,6 +3669,15 @@ class MT:
                     print(f'{current_step} 执行异常: {e}')
                     time.sleep(3)
                     step_failures[current_step] += 1
+                    # 封号检测：open_product_list_page 连续失败
+                    if current_step == "open_product_list_page":
+                        MT.open_page_fail_count += 1
+                        print(f"⚠️ open_product_list_page 连续失败 {MT.open_page_fail_count}/{MT.open_page_fail_threshold} 次")
+                        if MT.open_page_fail_count >= MT.open_page_fail_threshold:
+                            msg = f"疑似账号被封禁：open_product_list_page 连续失败 {self.open_page_fail_threshold} 次"
+                            logging.error(msg)
+                            self.finish_task_abnormally(self.get_current_page_no(), msg)
+                            raise CollectionStopError(msg)
                     retry_limit = self.workflow_retry_limit.get(current_step)
                     next_action = self.workflow_error_action.get(current_step)
                     record_restart_reason(
@@ -3683,8 +3704,6 @@ class MT:
             self.monitor.join()
 
 
-
-
 def fetch_task_from_scheduler(scheduler, device_id):
     """从已有的调度器获取一个任务，转换为 device_list 兼容的格式。没有任务返回 None。"""
     task = scheduler.get_task()
@@ -3891,11 +3910,22 @@ def run_device(device_id, scheduler=None):
         scheduler.stop()
 
 
+class DeviceIdLogFilter(logging.Filter):
+    """给每条日志注入当前设备ID（运行时读取全局 DEVICE_ID，终端传参后自动生效）"""
+    def filter(self, record):
+        record.device_id = DEVICE_ID
+        return True
+
+
 def main():
     logging.basicConfig(
         level=logging.INFO,
-        format='%(asctime)s [%(threadName)s] %(levelname)s: %(message)s'
+        format='%(asctime)s [%(threadName)s] [%(device_id)s] %(levelname)s: %(message)s'
     )
+    # 给所有 root handler 挂上设备ID过滤器
+    _device_filter = DeviceIdLogFilter()
+    for _h in logging.getLogger().handlers:
+        _h.addFilter(_device_filter)
 
     # 终端传入设备 ID：python main.py T4VK4LM7AAUOV8AY
     global DEVICE_ID
@@ -3903,6 +3933,13 @@ def main():
         DEVICE_ID = sys.argv[1].strip()
         logging.info(f"使用终端传入设备: {DEVICE_ID}")
 
+    # 同步设备ID到 commons.Logger（scheduler 等模块的日志带上设备ID）
+    try:
+        from commons.Logger import set_device_id
+        set_device_id(DEVICE_ID)
+    except Exception:
+        pass
+
     # 自动模式：全局只创建一个调度器 + 一个心跳线程，所有任务复用
     scheduler = None
     if not MANUAL_MODE:
@@ -3983,4 +4020,3 @@ device_list = {
 if __name__ == '__main__':
     main()
 
-

mt_V2/yzm.py

@@ -1,5 +1,4 @@
 
-
 import base64
 import io
 import math
@@ -153,9 +152,50 @@ def _save_debug_screenshot(d, captcha_type, tag="full"):
         return None
 
 
+def _compress_to_size(image_path, max_bytes=1024 * 1024):
+    """
+    将图片压缩到指定大小以内（默认 1MB）。
+    策略：转 RGB → 循环降低 JPEG 质量 → 仍超则缩小尺寸。
+    返回压缩后的字节流，不修改原图。
+    """
+    img = Image.open(image_path)
+    if img.mode != 'RGB':
+        img = img.convert('RGB')
+
+    quality = 90
+    while quality >= 30:
+        buf = io.BytesIO()
+        img.save(buf, format='JPEG', quality=quality)
+        if buf.tell() <= max_bytes:
+            return buf.getvalue()
+        quality -= 15
+
+    # 质量到 30 还超 → 缩小尺寸（每次 0.8 倍）
+    scale = 0.8
+    while True:
+        w, h = img.size
+        img = img.resize((max(1, int(w * scale)), max(1, int(h * scale))), Image.LANCZOS)
+        buf = io.BytesIO()
+        img.save(buf, format='JPEG', quality=80)
+        if buf.tell() <= max_bytes:
+            return buf.getvalue()
+        scale -= 0.1
+        if scale <= 0.4:
+            break
+
+    return buf.getvalue()
+
+
 def post_api(image_path, captcha_type, extra=None, direction=None, label_image_path=None, timeout=20):
-    with open(image_path, 'rb') as f:
-        image_base64 = base64.b64encode(f.read()).decode()
+    raw = open(image_path, 'rb').read()
+    # 超过 1MB 先压缩（打码平台限制）
+    if len(raw) > 1024 * 1024:
+        print(f"[yzm] 原图 {len(raw) / 1024 / 1024:.1f}MB 超过1MB，压缩中...")
+        compressed = _compress_to_size(image_path)
+        image_base64 = base64.b64encode(compressed).decode()
+        print(f"[yzm] 压缩后 {len(compressed) / 1024:.0f}KB")
+    else:
+        image_base64 = base64.b64encode(raw).decode()
 
     data = {
         "token": API_TOKEN,
@@ -180,7 +220,11 @@ def post_api(image_path, captcha_type, extra=None, direction=None, label_image_p
 
 def parse_points(response, y_offset=0):
     tuple_points = []
-    data = response.get("data", {}).get("data", "")
+    raw_data = response.get("data", {})
+    # API 失败时 data 可能是 [] 而非 dict，防御性处理
+    if not isinstance(raw_data, dict):
+        return tuple_points
+    data = raw_data.get("data", "")
     if not data:
         return tuple_points
 
@@ -205,7 +249,10 @@ def verify(image_path, captcha_type):
     if config["result"] == "points":
         return parse_points(response, y_offset=config["y_offset"])
 
-    return response.get("data", {}).get("data")
+    raw_data = response.get("data", {})
+    if isinstance(raw_data, dict):
+        return raw_data.get("data")
+    return None
 
 
 def crop_image_xy(

requirements.txt

@@ -1,14 +1,20 @@
-requests>=2.28
+requests==2.32.3
 DrissionPage>=4.0
-Pillow>=10.0
+Pillow==11.1.0
 pandas>=2.0
-pymysql>=1.0
+pymysql==1.1.1
 DBUtils>=3.0
 lxml>=4.9
-oss2>=2.18
+oss2
 pycryptodome>=3.19
 python-dotenv>=1.0
 schedule>=1.2
 playwright>=1.40
 curl_cffi>=0.7
 openpyxl>=3.1
+uiautomator2==3.2.9
+opencv-python==4.11.0.86
+numpy==2.0.2
+baidu-aip==4.16.13
+pywin32==310
+chardet