feat: update

booker_builtin.py

@@ -1,328 +1,361 @@
-import os
-import time
-import json
-import threading
-import random
-import traceback
-import redis
-from typing import List, Dict, Callable
-
-from vs_types import GroupConfig, VSPlgConfig, Task, VSQueryResult, AppointmentType
-from vs_plg_factory import VSPlgFactory 
-from toolkit.thread_pool import ThreadPool 
-from toolkit.vs_cloud_api import VSCloudApi
-from toolkit.backoff import ExponentialBackoff
-
-class BuiltinBookerGCO:
-    """
-    非绑定模式 (公共内置账号池):
-    - 只维护全局 target_instances 数量的实例。
-    - 所有实例热机等待，发现信号后临时去云端 Pop 订单。
-    """
-    def __init__(self, cfg: GroupConfig, redis_conf: Dict, logger: Callable[[str], None] = None):
-        self.m_cfg = cfg
-        self.m_factory = VSPlgFactory()
-        self.m_logger = logger
-        self.m_tasks: List[Task] = []
-        self.m_lock = threading.RLock()
-        self.m_stop_event = threading.Event()
-        self.redis_client = redis.Redis(**redis_conf)
-        self.m_pending_builtin = 0
-        
-        self.m_tracker_key = f"vs:worker:tasks_tracker:{self.m_cfg.identifier}"
-        self.group_backoff = ExponentialBackoff(base_delay=60.0, max_delay=10*60.0, factor=2.0)
-        self.task_backoff = ExponentialBackoff(base_delay=5*60.0, max_delay=2*60*60.0, factor=2.0)
-        self.m_last_spawn_time = 0.0
-        self.heartbeat_ttl = 300
-
-    def _log(self, message):
-        if self.m_logger:
-            self.m_logger(f'[BUILTIN-BOOKER] [{self.m_cfg.identifier}] {message}')
-
-    def start(self):
-        if not self.m_cfg.enable:
-            return
-        self._log("Starting Built-in Booker...")
-        plugin_name = self.m_cfg.plugin_config.plugin_name
-        class_name = "".join(part.title() for part in plugin_name.split('_'))
-        plugin_path = os.path.join(self.m_cfg.plugin_config.lib_path, self.m_cfg.plugin_config.plugin_bin)
-        self.m_factory.register_plugin(plugin_name, plugin_path, class_name)
-
-        threading.Thread(target=self._booking_trigger_loop, daemon=True).start()
-        threading.Thread(target=self._creator_loop, daemon=True).start()
-        threading.Thread(target=self._maintain_loop, daemon=True).start()
-
-    def stop(self):
-        self._log("Stopping Booker...")
-        self.m_stop_event.set()
-
-    def _get_redis_key(self, routing_key: str) -> str:
-        return f"vs:signal:{routing_key}"
-
-    def _maintain_loop(self):
-        self._log("Maintain loop started.")
-        while not self.m_stop_event.is_set():
-            time.sleep(1.0)
-            now = time.time()
-            
-            with self.m_lock:
-                tasks_to_check = list(self.m_tasks)
-            
-            if not tasks_to_check:
-                continue
-            
-            healthy_tasks = []
-            for t in tasks_to_check:
-                if now >= t.next_remote_ping:
-                    try:
-                        t.instance.keep_alive()
-                        if t.instance.health_check():
-                            healthy_tasks.append(t)
-                            next_delay = random.randint(180, 300) 
-                            t.next_remote_ping = now + next_delay
-                        else:
-                            self._log(f"♻️ Instance unhealthy. Will be removed.")
-                    except Exception as e:
-                        self._log(f"Instance keep-alive failed: {e}")
-                else:
-                    healthy_tasks.append(t)
-            
-            with self.m_lock:
-                self.m_tasks = [t for t in self.m_tasks if t in healthy_tasks]
-
-    def _booking_trigger_loop(self):
-        self._log("Trigger loop started.")
-        while not self.m_stop_event.is_set():
-            try:
-                time.sleep(1.0)
-                now = time.time()
-                for apt_type in self.m_cfg.appointment_types:
-                    redis_key = self._get_redis_key(apt_type.routing_key)
-                    raw_data = self.redis_client.get(redis_key)
-                    if not raw_data:
-                        continue
-                    
-                    try:
-                        data = json.loads(raw_data)
-                        query_result = VSQueryResult.model_validate(data['query_result'])
-                        query_result.apt_type = AppointmentType.model_validate(data['apt_type'])
-                    except Exception as parse_err:
-                        self._log(f"Data parsing error for {redis_key}: {parse_err}. Deleting corrupted signal.")
-                        self.redis_client.delete(redis_key)
-                        continue
-                    
-                    matching_tasks = []
-                    with self.m_lock:
-                        for task in self.m_tasks:
-                            if now < task.next_run or not task.book_allowed:
-                                continue
-                            if apt_type.routing_key not in task.acceptable_routing_keys:
-                                continue
-                            
-                            self._log(f"🚀 Triggering BOOK for {apt_type.routing_key}")
-                            task.next_run = now + self.m_cfg.booker.booking_cooldown
-                            matching_tasks.append(task)
-                    
-                    if matching_tasks:
-                        threads = []
-                        for task in matching_tasks:
-                            self._log(f"🚀 Triggering BOOK for {apt_type.routing_key}")
-                            t = threading.Thread(target=self._execute_book_job, args=(task, query_result))
-                            threads.append(t)
-                            t.start()
-                        
-                        for t in threads:
-                            t.join() 
-            except Exception as e:
-                self._log(f"Trigger loop error: {e}")
-                time.sleep(2)
-
-    def _execute_book_job(self, task: Task, query_result: VSQueryResult):
-        queue_name = f"auto.{query_result.apt_type.routing_key}"
-
-        task_id = None
-        task_data = None
-        booking_success = False
-        is_rate_limited = False
-        
-        try:
-            task_data = VSCloudApi.Instance().get_vas_task_pop(queue_name)
-            if not task_data:
-                return 
-            task_id = task_data['id']
-            order_id = task_data.get('order_id')
-            self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + self.heartbeat_ttl})
-                        
-            user_input = task_data.get('user_inputs', {})
-            book_res = task.instance.book(query_result, user_input)
-            
-            if book_res.success:
-                booking_success = True
-                self._log(f"✅ BOOK SUCCESS! Order: {order_id}")
-                
-                grab_info = {
-                    "account": book_res.account,
-                    "session_id": book_res.session_id,
-                    "urn": book_res.urn,
-                    "slot_date": book_res.book_date,
-                    "slot_time": book_res.book_time,
-                    "timestamp": int(time.time()),
-                    "payment_link": book_res.payment_link
-                }
-                VSCloudApi.Instance().update_vas_task(task_id, {"status": "grabbed", "grabbed_history": grab_info})
-                push_content = (
-                    f"🎉 【预定成功通知】\n"
-                    f"━━━━━━━━━━━━━━━\n"
-                    f"订单编号: {order_id}\n"
-                    f"预约账号: {book_res.account}\n"
-                    f"预约日期: {book_res.book_date}\n"
-                    f"预约时间: {book_res.book_time}\n"
-                    f"预约编号: {book_res.urn}\n"
-                    f"支付链接: {book_res.payment_link if book_res.payment_link else '无需支付/暂无'}\n"
-                    f"━━━━━━━━━━━━━━━\n"
-                )
-                VSCloudApi.Instance().push_weixin_text(push_content)
-                self.redis_client.zrem(self.m_tracker_key, task_id)
-                
-                # === 核心：成功次数判断 ===
-                task.successful_bookings += 1
-                max_b = self.m_cfg.booker.max_bookings_per_account
-                if max_b > 0 and task.successful_bookings >= max_b:
-                    self._log(f"Account reached max bookings ({max_b}). Destroying instance.")
-                    with self.m_lock:
-                        if task in self.m_tasks:
-                            self.m_tasks.remove(task)
-            else:
-                self._log(f"❌ BOOK FAILED for Order: {order_id}")
-
-        except Exception as e:
-            err_str = str(e)
-            self._log(f"Exception during booking: {err_str}")
-            rate_limited_indicators = [
-                "42901" in err_str,
-                "Rate limited" in err_str
-            ]
-            if any(rate_limited_indicators):
-                is_rate_limited = True
-                with self.m_lock:
-                    if task in self.m_tasks:
-                        self.m_tasks.remove(task)
-                if task_data and task_id is not None:
-                    task_meta = task_data.get('meta') or {} 
-                    t_fails = task_meta.get('booking_failures', 0) + 1
-                    task_meta['booking_failures'] = t_fails
-                    
-                    try:
-                        VSCloudApi.Instance().update_vas_task(task_id, {"meta": task_meta})
-                    except Exception as cloud_err:
-                        self._log(f"Failed to update task meta: {cloud_err}")
-                        
-                    t_cd = self.task_backoff.calculate(t_fails)
-                    self._log(f"⏳ Task={task_id} (Booking Attempt {t_fails}) suspended for {t_cd:.1f}s.")
-                    self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + t_cd})
-            
-        finally:
-           if not booking_success and task_id is not None and not is_rate_limited:
-                self.redis_client.zadd(self.m_tracker_key, {str(task_id): 0})
-                self._log(f"♻️ Task={task_id} normal failure. Instantly handed over to Sweeper.")
-                
-    def _creator_loop(self):
-        self._log("Creator loop started.")
-        spawn_interval = 10.0
-        group_cd_key = f"vs:group:cooldown:{self.m_cfg.identifier}"
-        while not self.m_stop_event.is_set():
-            time.sleep(2.0)
-            if self.redis_client.exists(group_cd_key):
-                continue
-            with self.m_lock:
-                current = len(self.m_tasks)
-                pending = self.m_pending_builtin
-                target = self.m_cfg.booker.target_instances
-            if (current + pending) < target:
-                now = time.time()
-                if now - self.m_last_spawn_time >= spawn_interval:
-                    self.m_last_spawn_time = now 
-                    self._spawn_worker()
-
-    def _spawn_worker(self):
-        with self.m_lock:
-            self.m_pending_builtin += 1
-            
-        def _job():
-            try:
-                plg_cfg = VSPlgConfig()
-                plg_cfg.debug = self.m_cfg.debug
-                plg_cfg.free_config = self.m_cfg.free_config
-                plg_cfg.session_max_life = self.m_cfg.session_max_life
-
-                if self.m_cfg.need_account:
-                    acc = VSCloudApi.Instance().get_next_account(self.m_cfg.booker.account_pool_id, self.m_cfg.booker.account_cd)
-                    plg_cfg.account.id = acc['id']
-                    plg_cfg.account.username = acc['username']
-                    plg_cfg.account.password = acc['password']
-
-                if self.m_cfg.need_proxy:
-                    proxy = VSCloudApi.Instance().get_next_proxy(self.m_cfg.proxy_pool, self.m_cfg.proxy_cd)
-                    plg_cfg.proxy.id = proxy['id']
-                    plg_cfg.proxy.ip = proxy['ip']
-                    plg_cfg.proxy.port = proxy['port']
-                    plg_cfg.proxy.proto = proxy['proto']
-                    plg_cfg.proxy.username = proxy['username']
-                    plg_cfg.proxy.password = proxy['password']
-
-                instance = self.m_factory.create(self.m_cfg.identifier, self.m_cfg.plugin_config.plugin_name)
-                instance.set_log(self.m_logger)
-                instance.set_config(plg_cfg)
-                instance.create_session()
-                
-                with self.m_lock:
-                    all_keys = [apt.routing_key for apt in self.m_cfg.appointment_types]
-                    self.m_tasks.append(
-                        Task(
-                            instance=instance,
-                            qw_cfg=self.m_cfg.query_wait,
-                            next_run=time.time(), 
-                            task_ref=None,
-                            acceptable_routing_keys=all_keys,
-                            source_queue="built-in",
-                            book_allowed=True,
-                            next_remote_ping = time.time() + random.randint(180, 300) 
-                        )
-                    )
-                    
-                    group_fail_key = f"vs:group:failures:{self.m_cfg.identifier}"
-                    self.redis_client.delete(group_fail_key)
-                    
-                self._log(f"+++ Built-in Booker spawned: {plg_cfg.account.username}")
-            except Exception as e:
-                err_str = str(e)
-                resource_not_found_indicators = [
-                    "40401" in err_str,
-                    "Account not found" in err_str,
-                    "Proxy not found" in err_str,
-                ]
-                if any(resource_not_found_indicators):
-                    return
-                
-                self._log(f"Spawn failed: {e}")
-                
-                rate_limited_indicators = [
-                    "42901" in err_str,
-                    "Rate limited" in err_str
-                ]
-                if any(rate_limited_indicators):
-                    group_fail_key = f"vs:group:failures:{self.m_cfg.identifier}"
-                    group_cd_key = f"vs:group:cooldown:{self.m_cfg.identifier}"
-                    
-                    # 更新全局（机器组）失败次数
-                    g_fails = self.redis_client.incr(group_fail_key)
-                    # 计算退避时间
-                    g_cd = self.group_backoff.calculate(g_fails)
-                    # 设置 Redis 全局冷却保护阀
-                    self.redis_client.set(group_cd_key, "1", ex=int(g_cd))
-                    self._log(f"📉 [Rate Limited] Group '{self.m_cfg.identifier}' failed {g_fails} times. Global Backoff: {g_cd:.1f}s.")
-
-            finally:
-                with self.m_lock:
-                    self.m_pending_builtin = max(0, self.m_pending_builtin - 1)
+import os
+import time
+import json
+import threading
+import random
+import redis
+from typing import List, Dict, Callable
+
+from vs_types import GroupConfig, VSPlgConfig, Task, VSQueryResult, AppointmentType
+from vs_plg_factory import VSPlgFactory 
+from toolkit.thread_pool import ThreadPool 
+from toolkit.vs_cloud_api import VSCloudApi
+from toolkit.backoff import ExponentialBackoff
+
+class BuiltinBookerGCO:
+    """
+    非绑定模式 (公共内置账号池):
+    - 只维护全局 target_instances 数量的实例。
+    - 所有实例热机等待，发现信号后临时去云端 Pop 订单。
+    """
+    def __init__(self, cfg: GroupConfig, redis_conf: Dict, logger: Callable[[str], None] = None):
+        self.m_cfg = cfg
+        self.m_factory = VSPlgFactory()
+        self.m_logger = logger
+        self.m_tasks: List[Task] = []
+        self.m_lock = threading.RLock()
+        self.m_stop_event = threading.Event()
+        self.redis_client = redis.Redis(**redis_conf)
+        self.m_pending_builtin = 0
+        
+        self.m_tracker_key = f"vs:worker:tasks_tracker:{self.m_cfg.identifier}"
+        self.group_backoff = ExponentialBackoff(base_delay=60.0, max_delay=10*60.0, factor=2.0)
+        self.task_backoff = ExponentialBackoff(base_delay=5*60.0, max_delay=2*60*60.0, factor=2.0)
+        self.m_last_spawn_time = 0.0
+        self.heartbeat_ttl = 300
+
+    def _log(self, message):
+        if self.m_logger:
+            self.m_logger(f'[BUILTIN-BOOKER] [{self.m_cfg.identifier}] {message}')
+
+    def start(self):
+        if not self.m_cfg.enable:
+            return
+        self._log("Starting Built-in Booker...")
+        plugin_name = self.m_cfg.plugin_config.plugin_name
+        class_name = "".join(part.title() for part in plugin_name.split('_'))
+        plugin_path = os.path.join(self.m_cfg.plugin_config.lib_path, self.m_cfg.plugin_config.plugin_bin)
+        self.m_factory.register_plugin(plugin_name, plugin_path, class_name)
+
+        threading.Thread(target=self._booking_trigger_loop, daemon=True).start()
+        threading.Thread(target=self._creator_loop, daemon=True).start()
+        threading.Thread(target=self._maintain_loop, daemon=True).start()
+
+    def stop(self):
+        self._log("Stopping Booker...")
+        self.m_stop_event.set()
+        self._cleanup_all_tasks("booker stop")
+
+    def _cleanup_task(self, task: Task, reason: str = ""):
+        try:
+            instance = getattr(task, 'instance', None)
+            if instance and hasattr(instance, 'cleanup'):
+                instance.cleanup()
+                self._log(f"🧹 Cleaned up built-in instance. Reason: {reason}")
+        except Exception as e:
+            self._log(f"Cleanup failed for built-in instance. Reason: {reason}. Error: {e}")
+
+    def _remove_task(self, task: Task, reason: str = "", cleanup: bool = True):
+        removed = False
+        with self.m_lock:
+            if task in self.m_tasks:
+                self.m_tasks.remove(task)
+                removed = True
+        if cleanup and removed:
+            self._cleanup_task(task, reason)
+        return removed
+
+    def _cleanup_all_tasks(self, reason: str = ""):
+        with self.m_lock:
+            tasks = list(self.m_tasks)
+            self.m_tasks.clear()
+        for task in tasks:
+            self._cleanup_task(task, reason)
+
+    def _get_redis_key(self, routing_key: str) -> str:
+        return f"vs:signal:{routing_key}"
+
+    def _maintain_loop(self):
+        self._log("Maintain loop started.")
+        while not self.m_stop_event.is_set():
+            time.sleep(1.0)
+            now = time.time()
+            
+            with self.m_lock:
+                tasks_to_check = list(self.m_tasks)
+            
+            if not tasks_to_check:
+                continue
+            
+            healthy_tasks = []
+            dead_tasks = []
+            for t in tasks_to_check:
+                if now >= t.next_remote_ping:
+                    try:
+                        t.instance.keep_alive()
+                        if t.instance.health_check():
+                            healthy_tasks.append(t)
+                            next_delay = random.randint(180, 300) 
+                            t.next_remote_ping = now + next_delay
+                        else:
+                            dead_tasks.append(t)
+                            self._log(f"♻️ Instance unhealthy. Will be removed.")
+                    except Exception as e:
+                        dead_tasks.append(t)
+                        self._log(f"Instance keep-alive failed: {e}")
+                else:
+                    healthy_tasks.append(t)
+            
+            if dead_tasks:
+                with self.m_lock:
+                    current_tasks = list(self.m_tasks)
+                    self.m_tasks = [t for t in self.m_tasks if t in healthy_tasks]
+                for t in dead_tasks:
+                    if t in current_tasks:
+                        self._cleanup_task(t, "unhealthy or keep-alive failed")
+            else:
+                with self.m_lock:
+                    self.m_tasks = [t for t in self.m_tasks if t in healthy_tasks]
+
+    def _booking_trigger_loop(self):
+        self._log("Trigger loop started.")
+        while not self.m_stop_event.is_set():
+            try:
+                time.sleep(1.0)
+                now = time.time()
+                for apt_type in self.m_cfg.appointment_types:
+                    redis_key = self._get_redis_key(apt_type.routing_key)
+                    raw_data = self.redis_client.get(redis_key)
+                    if not raw_data:
+                        continue
+                    
+                    try:
+                        data = json.loads(raw_data)
+                        query_result = VSQueryResult.model_validate(data['query_result'])
+                        query_result.apt_type = AppointmentType.model_validate(data['apt_type'])
+                    except Exception as parse_err:
+                        self._log(f"Data parsing error for {redis_key}: {parse_err}. Deleting corrupted signal.")
+                        self.redis_client.delete(redis_key)
+                        continue
+                    
+                    matching_tasks = []
+                    with self.m_lock:
+                        for task in self.m_tasks:
+                            if now < task.next_run or not task.book_allowed:
+                                continue
+                            if apt_type.routing_key not in task.acceptable_routing_keys:
+                                continue
+                            
+                            self._log(f"🚀 Triggering BOOK for {apt_type.routing_key}")
+                            task.next_run = now + self.m_cfg.booker.booking_cooldown
+                            matching_tasks.append(task)
+                    
+                    if matching_tasks:
+                        threads = []
+                        for task in matching_tasks:
+                            self._log(f"🚀 Triggering BOOK for {apt_type.routing_key}")
+                            t = threading.Thread(target=self._execute_book_job, args=(task, query_result))
+                            threads.append(t)
+                            t.start()
+                        
+                        for t in threads:
+                            t.join() 
+            except Exception as e:
+                self._log(f"Trigger loop error: {e}")
+                time.sleep(2)
+
+    def _execute_book_job(self, task: Task, query_result: VSQueryResult):
+        queue_name = f"auto.{query_result.apt_type.routing_key}"
+
+        task_id = None
+        task_data = None
+        booking_success = False
+        is_rate_limited = False
+        
+        try:
+            task_data = VSCloudApi.Instance().get_vas_task_pop(queue_name)
+            if not task_data:
+                return 
+            task_id = task_data['id']
+            order_id = task_data.get('order_id')
+            self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + self.heartbeat_ttl})
+                        
+            user_input = task_data.get('user_inputs', {})
+            book_res = task.instance.book(query_result, user_input)
+            
+            if book_res.success:
+                booking_success = True
+                self._log(f"✅ BOOK SUCCESS! Order: {order_id}")
+                
+                grab_info = {
+                    "account": book_res.account,
+                    "session_id": book_res.session_id,
+                    "urn": book_res.urn,
+                    "slot_date": book_res.book_date,
+                    "slot_time": book_res.book_time,
+                    "timestamp": int(time.time()),
+                    "payment_link": book_res.payment_link
+                }
+                VSCloudApi.Instance().update_vas_task(task_id, {"status": "grabbed", "grabbed_history": grab_info})
+                push_content = (
+                    f"🎉 【预定成功通知】\n"
+                    f"━━━━━━━━━━━━━━━\n"
+                    f"订单编号: {order_id}\n"
+                    f"预约账号: {book_res.account}\n"
+                    f"预约日期: {book_res.book_date}\n"
+                    f"预约时间: {book_res.book_time}\n"
+                    f"预约编号: {book_res.urn}\n"
+                    f"支付链接: {book_res.payment_link if book_res.payment_link else '无需支付/暂无'}\n"
+                    f"━━━━━━━━━━━━━━━\n"
+                )
+                VSCloudApi.Instance().push_weixin_text(push_content)
+                self.redis_client.zrem(self.m_tracker_key, task_id)
+                
+                # === 核心：成功次数判断 ===
+                task.successful_bookings += 1
+                max_b = self.m_cfg.booker.max_bookings_per_account
+                if max_b > 0 and task.successful_bookings >= max_b:
+                    self._log(f"Account reached max bookings ({max_b}). Destroying instance.")
+                    self._remove_task(task, "max bookings reached")
+            else:
+                self._log(f"❌ BOOK FAILED for Order: {order_id}")
+
+        except Exception as e:
+            err_str = str(e)
+            self._log(f"Exception during booking: {err_str}")
+            rate_limited_indicators = [
+                "42901" in err_str,
+                "Rate limited" in err_str
+            ]
+            if any(rate_limited_indicators):
+                is_rate_limited = True
+                self._remove_task(task, "booking rate limited")
+                if task_data and task_id is not None:
+                    task_meta = task_data.get('meta') or {} 
+                    t_fails = task_meta.get('booking_failures', 0) + 1
+                    task_meta['booking_failures'] = t_fails
+                    
+                    try:
+                        VSCloudApi.Instance().update_vas_task(task_id, {"meta": task_meta})
+                    except Exception as cloud_err:
+                        self._log(f"Failed to update task meta: {cloud_err}")
+                        
+                    t_cd = self.task_backoff.calculate(t_fails)
+                    self._log(f"⏳ Task={task_id} (Booking Attempt {t_fails}) suspended for {t_cd:.1f}s.")
+                    self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + t_cd})
+            
+        finally:
+           if not booking_success and task_id is not None and not is_rate_limited:
+                self.redis_client.zadd(self.m_tracker_key, {str(task_id): 0})
+                self._log(f"♻️ Task={task_id} normal failure. Instantly handed over to Sweeper.")
+                
+    def _creator_loop(self):
+        self._log("Creator loop started.")
+        spawn_interval = 10.0
+        group_cd_key = f"vs:group:cooldown:{self.m_cfg.identifier}"
+        while not self.m_stop_event.is_set():
+            time.sleep(2.0)
+            if self.redis_client.exists(group_cd_key):
+                continue
+            with self.m_lock:
+                current = len(self.m_tasks)
+                pending = self.m_pending_builtin
+                target = self.m_cfg.booker.target_instances
+            if (current + pending) < target:
+                now = time.time()
+                if now - self.m_last_spawn_time >= spawn_interval:
+                    self.m_last_spawn_time = now 
+                    self._spawn_worker()
+
+    def _spawn_worker(self):
+        with self.m_lock:
+            self.m_pending_builtin += 1
+            
+        def _job():
+            try:
+                plg_cfg = VSPlgConfig()
+                plg_cfg.debug = self.m_cfg.debug
+                plg_cfg.free_config = self.m_cfg.free_config
+                plg_cfg.session_max_life = self.m_cfg.session_max_life
+
+                if self.m_cfg.need_account:
+                    acc = VSCloudApi.Instance().get_next_account(self.m_cfg.booker.account_pool_id, self.m_cfg.booker.account_cd)
+                    plg_cfg.account.id = acc['id']
+                    plg_cfg.account.username = acc['username']
+                    plg_cfg.account.password = acc['password']
+
+                if self.m_cfg.need_proxy:
+                    proxy = VSCloudApi.Instance().get_next_proxy(self.m_cfg.proxy_pool, self.m_cfg.proxy_cd)
+                    plg_cfg.proxy.id = proxy['id']
+                    plg_cfg.proxy.ip = proxy['ip']
+                    plg_cfg.proxy.port = proxy['port']
+                    plg_cfg.proxy.proto = proxy['proto']
+                    plg_cfg.proxy.username = proxy['username']
+                    plg_cfg.proxy.password = proxy['password']
+
+                instance = self.m_factory.create(self.m_cfg.identifier, self.m_cfg.plugin_config.plugin_name)
+                instance.set_log(self.m_logger)
+                instance.set_config(plg_cfg)
+                instance.create_session()
+                
+                with self.m_lock:
+                    all_keys = [apt.routing_key for apt in self.m_cfg.appointment_types]
+                    self.m_tasks.append(
+                        Task(
+                            instance=instance,
+                            qw_cfg=self.m_cfg.query_wait,
+                            next_run=time.time(), 
+                            task_ref=None,
+                            acceptable_routing_keys=all_keys,
+                            source_queue="built-in",
+                            book_allowed=True,
+                            next_remote_ping = time.time() + random.randint(180, 300) 
+                        )
+                    )
+                    
+                    group_fail_key = f"vs:group:failures:{self.m_cfg.identifier}"
+                    self.redis_client.delete(group_fail_key)
+                    
+                self._log(f"+++ Built-in Booker spawned: {plg_cfg.account.username}")
+            except Exception as e:
+                err_str = str(e)
+                resource_not_found_indicators = [
+                    "40401" in err_str,
+                    "Account not found" in err_str,
+                    "Proxy not found" in err_str,
+                ]
+                if any(resource_not_found_indicators):
+                    return
+                
+                self._log(f"Spawn failed: {e}")
+                
+                rate_limited_indicators = [
+                    "42901" in err_str,
+                    "Rate limited" in err_str
+                ]
+                if any(rate_limited_indicators):
+                    group_fail_key = f"vs:group:failures:{self.m_cfg.identifier}"
+                    group_cd_key = f"vs:group:cooldown:{self.m_cfg.identifier}"
+                    
+                    # 更新全局（机器组）失败次数
+                    g_fails = self.redis_client.incr(group_fail_key)
+                    # 计算退避时间
+                    g_cd = self.group_backoff.calculate(g_fails)
+                    # 设置 Redis 全局冷却保护阀
+                    self.redis_client.set(group_cd_key, "1", ex=int(g_cd))
+                    self._log(f"📉 [Rate Limited] Group '{self.m_cfg.identifier}' failed {g_fails} times. Global Backoff: {g_cd:.1f}s.")
+
+            finally:
+                with self.m_lock:
+                    self.m_pending_builtin = max(0, self.m_pending_builtin - 1)
         ThreadPool.getInstance().enqueue(_job)

booker_order.py

@@ -1,377 +1,404 @@
-import os
-import time
-import json
-import threading
-import random
-import traceback
-import redis
-from typing import List, Dict, Callable
-
-from vs_types import GroupConfig, VSPlgConfig, Task, VSQueryResult, AppointmentType
-from vs_plg_factory import VSPlgFactory 
-from toolkit.thread_pool import ThreadPool 
-from toolkit.vs_cloud_api import VSCloudApi
-from toolkit.backoff import ExponentialBackoff
-
-class OrderBookerGCO:
-    """
-    绑定模式 (订单自带账号):
-    - 按城市队列维护热机配额。
-    - 绝对的 1 对 1 关系：一个实例绑定一个云端订单。
-    - 预订成功后，实例立即销毁。
-    """
-    def __init__(self, cfg: GroupConfig, redis_conf: Dict, logger: Callable[[str], None] = None):
-        self.m_cfg = cfg
-        self.m_factory = VSPlgFactory()
-        self.m_logger = logger
-        self.m_tasks: List[Task] = []
-        self.m_lock = threading.RLock()
-        self.m_stop_event = threading.Event()
-        self.redis_client = redis.Redis(**redis_conf)
-        self.m_pending_order_by_queue: Dict[str, int] = {}
-        
-        self.m_tracker_key = f"vs:worker:tasks_tracker:{self.m_cfg.identifier}"
-        self.queue_backoff = ExponentialBackoff(base_delay=1*60.0, max_delay=10*60.0, factor=2.0)
-        self.account_backoff = ExponentialBackoff(base_delay=5*60.0, max_delay=2*60*60.0, factor=2.0)
-        self.m_last_spawn_time = 0.0
-        self.heartbeat_ttl = 300
-
-    def _log(self, message):
-        if self.m_logger:
-            self.m_logger(f'[ORDER-BOOKER] [{self.m_cfg.identifier}] {message}')
-
-    def start(self):
-        if not self.m_cfg.enable:
-            return
-        self._log("Starting Order Booker...")
-        plugin_name = self.m_cfg.plugin_config.plugin_name
-        class_name = "".join(part.title() for part in plugin_name.split('_'))
-        plugin_path = os.path.join(self.m_cfg.plugin_config.lib_path, self.m_cfg.plugin_config.plugin_bin)
-        self.m_factory.register_plugin(plugin_name, plugin_path, class_name)
-
-        threading.Thread(target=self._booking_trigger_loop, daemon=True).start()
-        threading.Thread(target=self._creator_loop, daemon=True).start()
-        threading.Thread(target=self._maintain_loop, daemon=True).start()
-
-    def stop(self):
-        self._log("Stopping Booker...")
-        self.m_stop_event.set()
-
-    def _get_redis_key(self, routing_key: str) -> str:
-        return f"vs:signal:{routing_key}"
-            
-    def _maintain_loop(self):
-        self._log("Maintain loop started.")
-        heartbeat_interval = 60
-        while not self.m_stop_event.is_set():
-            for _ in range(heartbeat_interval):
-                if self.m_stop_event.is_set():
-                    return
-                time.sleep(1.0)
-            
-            with self.m_lock:
-                tasks_to_check = list(self.m_tasks)
-                
-            if not tasks_to_check:
-                continue
-            
-            healthy_tasks = []
-            dead_tasks = []
-            now = time.time()
-            
-            for t in tasks_to_check:
-                if now >= t.next_remote_ping:
-                    try:
-                        t.instance.keep_alive()
-                        if t.instance.health_check(): 
-                            healthy_tasks.append(t)
-                            next_delay = random.randint(180, 300) 
-                            t.next_remote_ping = now + next_delay
-                            self._log(f"🛡️ Task={t.task_ref} keep-alive success. Next ping in {next_delay}s.")
-                        else:
-                            dead_tasks.append(t)
-                            self._log(f"♻️ Instance for task={t.task_ref} unhealthy.")
-                    except Exception as e:
-                        dead_tasks.append(t)
-                        self._log(f"♻️ Instance for task={t.task_ref} keep-alive failed: {e}.")
-                else:
-                    healthy_tasks.append(t)
-            
-            if healthy_tasks:
-                try:
-                    pipeline = self.redis_client.pipeline()
-                    new_deadline = time.time() + self.heartbeat_ttl
-                    for t in healthy_tasks:
-                        if t.task_ref is not None:
-                            pipeline.zadd(self.m_tracker_key, {str(t.task_ref): new_deadline})
-                    pipeline.execute()
-                    self._log(f"💓 Heartbeat sent. Renewed {len(healthy_tasks)} tasks.")
-                except Exception as e:
-                    self._log(f"Redis Heartbeat update failed: {e}")
-
-            if dead_tasks:
-                try:
-                    pipeline = self.redis_client.pipeline()
-                    for t in dead_tasks:
-                        if t.task_ref is not None:
-                            pipeline.zadd(self.m_tracker_key, {str(t.task_ref): 0})
-                    pipeline.execute()
-                    self._log(f"🗑️ Handed over {len(dead_tasks)} dead tasks to Sweeper.")
-                except Exception as e:
-                    pass
-            
-            with self.m_lock:
-                self.m_tasks = [t for t in self.m_tasks if t in healthy_tasks]
-
-    def _booking_trigger_loop(self):
-        self._log("Trigger loop started.")
-        while not self.m_stop_event.is_set():
-            try:
-                time.sleep(1.0)
-                now = time.time()
-                for apt_type in self.m_cfg.appointment_types:
-                    redis_key = self._get_redis_key(apt_type.routing_key)
-                    raw_data = self.redis_client.get(redis_key)
-                    if not raw_data:
-                        continue
-                    try:
-                        data = json.loads(raw_data)
-                        query_result = VSQueryResult.model_validate(data['query_result'])
-                        query_result.apt_type = AppointmentType.model_validate(data['apt_type'])
-                    except Exception as parse_err:
-                        self._log(f"Data parsing error for {redis_key}: {parse_err}. Deleting corrupted signal.")
-                        self.redis_client.delete(redis_key)
-                        continue
-                    
-                    matching_tasks = []
-                    with self.m_lock:
-                        for task in self.m_tasks:
-                            if now < task.next_run or not task.book_allowed:
-                                continue
-                            if apt_type.routing_key not in task.acceptable_routing_keys:
-                                continue
-                            
-                            task.next_run = now + self.m_cfg.booker.booking_cooldown
-                            matching_tasks.append(task)
-                            
-                    if matching_tasks:
-                        threads = []
-                        for task in matching_tasks:
-                            self._log(f"🚀 Triggering BOOK for {apt_type.routing_key} | Order Ref: {task.task_ref}")
-                            t = threading.Thread(target=self._execute_book_job, args=(task, query_result))
-                            threads.append(t)
-                            t.start()
-                        
-                        for t in threads:
-                            t.join() 
-                    
-            except Exception as e:
-                self._log(f"Trigger loop error: {e}")
-                time.sleep(2)
-
-    def _execute_book_job(self, task: Task, query_result: VSQueryResult):
-        task_id = task.task_ref
-        task_data = None
-
-        try:
-            task_data = VSCloudApi.Instance().get_vas_task(task_id)
-            if not task_data or task_data.get('status') in ['grabbed', 'pause', 'completed', 'cancelled']:
-                self._log(f"Bound Task={task_id} is no longer valid or already processed. Removing instance.")
-                with self.m_lock:
-                    if task in self.m_tasks:
-                        self.m_tasks.remove(task)
-                self.redis_client.zrem(self.m_tracker_key, task_id)
-                return
-            
-            order_id = task_data.get('order_id')
-            user_input = task_data.get('user_inputs', {})
-            book_res = task.instance.book(query_result, user_input)
-
-            if book_res.success:
-                self._log(f"✅ BOOK SUCCESS! Order: {order_id}. Destroying instance.")
-                grab_info = {
-                    "account": book_res.account,
-                    "session_id": book_res.session_id,
-                    "urn": book_res.urn,
-                    "slot_date": book_res.book_date,
-                    "slot_time": book_res.book_time,
-                    "timestamp": int(time.time()),
-                    "payment_link": book_res.payment_link
-                }
-                VSCloudApi.Instance().update_vas_task(task_id, {"status": "grabbed", "grabbed_history": grab_info})
-                push_content = (
-                    f"🎉 【预定成功通知】\n"
-                    f"━━━━━━━━━━━━━━━\n"
-                    f"订单编号: {order_id}\n"
-                    f"预约账号: {book_res.account}\n"
-                    f"预约日期: {book_res.book_date}\n"
-                    f"预约时间: {book_res.book_time}\n"
-                    f"预约编号: {book_res.urn}\n"
-                    f"支付链接: {book_res.payment_link if book_res.payment_link else '无需支付/暂无'}\n"
-                    f"━━━━━━━━━━━━━━━\n"
-                )
-                VSCloudApi.Instance().push_weixin_text(push_content)
-                self.redis_client.zrem(self.m_tracker_key, task_id)
-                
-                with self.m_lock:
-                    if task in self.m_tasks:
-                        self.m_tasks.remove(task)
-            else:
-                self._log(f"❌ BOOK FAILED for Order: {order_id}. Will retry on next signal.")
-
-        except Exception as e:
-            err_str = str(e)
-            self._log(f"Exception during booking: {err_str}")
-            rate_limited_indicators = [
-                "42901" in err_str,
-                "Rate limited" in err_str
-            ]
-            if any(rate_limited_indicators):
-                with self.m_lock:
-                    if task in self.m_tasks:
-                        self.m_tasks.remove(task)
-                if task_data and task_id is not None:
-                    task_meta = task_data.get('meta', {})
-                    t_fails = task_meta.get('booking_failures', 0) + 1
-                    task_meta['booking_failures'] = t_fails
-                    
-                    try:
-                        VSCloudApi.Instance().update_vas_task(task_id, {"meta": task_meta})
-                    except Exception as cloud_err:
-                        self._log(f"Failed to update task meta: {cloud_err}")
-                        
-                    t_cd = self.task_backoff.calculate(t_fails)
-                    self._log(f"⏳ Task={task_id} (Booking Attempt {t_fails}) suspended for {t_cd:.1f}s.")
-                    self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + t_cd})
-
-    def _creator_loop(self):
-        self._log("Creator loop started.")
-        spawn_interval = 10.0
-        while not self.m_stop_event.is_set():
-            time.sleep(2.0)
-            for apt in self.m_cfg.appointment_types:
-                r_key = apt.routing_key
-                
-                queue_cd_key = f"vs:queue:cooldown:{r_key}"
-                if self.redis_client.exists(queue_cd_key):
-                    continue
-                
-                with self.m_lock:
-                    active = sum(1 for t in self.m_tasks if getattr(t, 'source_queue', '') == r_key)
-                    pending = self.m_pending_order_by_queue.get(r_key, 0)
-                    target = self.m_cfg.booker.target_instances
-                
-                if (active + pending) < target:
-                    now = time.time()
-                    if now - self.m_last_spawn_time >= spawn_interval:
-                        self.m_last_spawn_time = now 
-                        self._spawn_worker(r_key)
-                        break
-
-    def _spawn_worker(self, target_routing_key: str):
-        with self.m_lock: 
-            self.m_pending_order_by_queue[target_routing_key] = self.m_pending_order_by_queue.get(target_routing_key, 0) + 1
-            
-        def _job():
-            success = False
-            task_id = None
-            is_rate_limited = False
-            
-            try:
-                queue_name = f"auto.{target_routing_key}"
-                task_data = VSCloudApi.Instance().get_vas_task_pop(queue_name)
-                if not task_data:
-                    return 
-                
-                task_id = task_data['id']
-                
-                self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + self.heartbeat_ttl})
-                user_inputs = task_data.get('user_inputs', {})
-                
-                plg_cfg = VSPlgConfig()
-                plg_cfg.debug = self.m_cfg.debug
-                plg_cfg.free_config = self.m_cfg.free_config
-                plg_cfg.session_max_life = self.m_cfg.session_max_life
-                plg_cfg.account.username = user_inputs.get("username", "")
-                plg_cfg.account.password = user_inputs.get("password", "")
-                if not plg_cfg.account.username:
-                    return
-                
-                acceptable_keys = [target_routing_key]
-                if self.m_cfg.need_proxy:
-                    proxy = VSCloudApi.Instance().get_next_proxy(self.m_cfg.proxy_pool, self.m_cfg.proxy_cd)
-                    plg_cfg.proxy.id = proxy['id']
-                    plg_cfg.proxy.ip = proxy['ip']
-                    plg_cfg.proxy.port = proxy['port']
-                    plg_cfg.proxy.proto = proxy['proto']
-                    plg_cfg.proxy.username = proxy['username']
-                    plg_cfg.proxy.password = proxy['password']
-
-                instance = self.m_factory.create(self.m_cfg.identifier, self.m_cfg.plugin_config.plugin_name)
-                instance.set_log(self.m_logger)
-                instance.set_config(plg_cfg)
-                instance.create_session() 
-                
-                with self.m_lock:
-                    self.m_tasks.append(
-                        Task(
-                            instance=instance,
-                            qw_cfg=self.m_cfg.query_wait,
-                            next_run=time.time(), 
-                            task_ref=task_id,
-                            acceptable_routing_keys=acceptable_keys, 
-                            source_queue=target_routing_key,
-                            book_allowed=True,
-                            next_remote_ping=time.time() + random.randint(180, 300) 
-                        )
-                    )
-                    queue_fail_key = f"vs:queue:failures:{target_routing_key}"
-                    self.redis_client.delete(queue_fail_key)                    
-                success = True
-                self._log(f"+++ Order Booker spawned: {plg_cfg.account.username} (Target: {acceptable_keys})")
-            except Exception as e:
-                err_str = str(e)
-                resource_not_found_indicators = [
-                    "40401" in err_str,
-                    "Account not found" in err_str,
-                    "Proxy not found" in err_str
-                ]
-                if any(resource_not_found_indicators):
-                    return
-                
-                self._log(f"Order Booker spawn failed: {e}")
-                
-                rate_limited_indicators = [
-                    "42901" in err_str,
-                    "Rate limited" in err_str
-                ]
-                if any(rate_limited_indicators):
-                    is_rate_limited = True
-                    queue_fail_key = f"vs:queue:failures:{target_routing_key}"
-                    queue_cd_key = f"vs:queue:cooldown:{target_routing_key}"
-                    q_fails = self.redis_client.incr(queue_fail_key)
-                    q_cd = self.queue_backoff.calculate(q_fails)
-                    self.redis_client.set(queue_cd_key, "1", ex=int(q_cd))
-                    self._log(f"📉 [Rate Limited] Queue '{target_routing_key}' failed {q_fails} times. Global Backoff: {q_cd:.1f}s.")
-                    if task_id is not None:
-                        task_meta = task_data.get('meta') or {}
-                        t_fails = task_meta.get('spawn_failures', 0) + 1
-                        task_meta['spawn_failures'] = t_fails
-                        
-                        try:
-                            VSCloudApi.Instance().update_vas_task(task_id, {"meta": task_meta})
-                        except Exception as cloud_err:
-                            self._log(f"Failed to update task meta: {cloud_err}")
-                        
-                        t_cd = self.account_backoff.calculate(t_fails)
-                        self._log(f"⏳ Task={task_id} (Attempt {t_fails}) suspended for {t_cd:.1f}s.")
-                        self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + t_cd})       
-            finally:
-                with self.m_lock: 
-                    self.m_pending_order_by_queue[target_routing_key] = max(0, self.m_pending_order_by_queue[target_routing_key] - 1)
-                
-                # 创建/登录失败，调用安全归还函数
-                if not success and task_id is not None and not is_rate_limited:
-                    self.redis_client.zadd(self.m_tracker_key, {str(task_id): 0})
-                    self._log(f"♻️ Task={task_id} failed normal spawn. Instantly handed over to Sweeper.")
+import os
+import time
+import json
+import threading
+import random
+import redis
+from typing import List, Dict, Callable
+
+from vs_types import GroupConfig, VSPlgConfig, Task, VSQueryResult, AppointmentType
+from vs_plg_factory import VSPlgFactory 
+from toolkit.thread_pool import ThreadPool 
+from toolkit.vs_cloud_api import VSCloudApi
+from toolkit.backoff import ExponentialBackoff
+
+class OrderBookerGCO:
+    """
+    绑定模式 (订单自带账号):
+    - 按城市队列维护热机配额。
+    - 绝对的 1 对 1 关系：一个实例绑定一个云端订单。
+    - 预订成功后，实例立即销毁。
+    """
+    def __init__(self, cfg: GroupConfig, redis_conf: Dict, logger: Callable[[str], None] = None):
+        self.m_cfg = cfg
+        self.m_factory = VSPlgFactory()
+        self.m_logger = logger
+        self.m_tasks: List[Task] = []
+        self.m_lock = threading.RLock()
+        self.m_stop_event = threading.Event()
+        self.redis_client = redis.Redis(**redis_conf)
+        self.m_pending_order_by_queue: Dict[str, int] = {}
+        
+        self.m_tracker_key = f"vs:worker:tasks_tracker:{self.m_cfg.identifier}"
+        self.queue_backoff = ExponentialBackoff(base_delay=1*60.0, max_delay=10*60.0, factor=2.0)
+        self.account_backoff = ExponentialBackoff(base_delay=5*60.0, max_delay=2*60*60.0, factor=2.0)
+        self.m_last_spawn_time = 0.0
+        self.heartbeat_ttl = 300
+
+    def _log(self, message):
+        if self.m_logger:
+            self.m_logger(f'[ORDER-BOOKER] [{self.m_cfg.identifier}] {message}')
+
+    def start(self):
+        if not self.m_cfg.enable:
+            return
+        self._log("Starting Order Booker...")
+        plugin_name = self.m_cfg.plugin_config.plugin_name
+        class_name = "".join(part.title() for part in plugin_name.split('_'))
+        plugin_path = os.path.join(self.m_cfg.plugin_config.lib_path, self.m_cfg.plugin_config.plugin_bin)
+        self.m_factory.register_plugin(plugin_name, plugin_path, class_name)
+
+        threading.Thread(target=self._booking_trigger_loop, daemon=True).start()
+        threading.Thread(target=self._creator_loop, daemon=True).start()
+        threading.Thread(target=self._maintain_loop, daemon=True).start()
+
+    def stop(self):
+        self._log("Stopping Booker...")
+        self.m_stop_event.set()
+        self._cleanup_all_tasks("booker stop")
+
+    def _cleanup_task(self, task: Task, reason: str = ""):
+        try:
+            instance = getattr(task, 'instance', None)
+            if instance and hasattr(instance, 'cleanup'):
+                instance.cleanup()
+                self._log(f"🧹 Cleaned up instance for task={getattr(task, 'task_ref', None)}. Reason: {reason}")
+        except Exception as e:
+            self._log(f"Cleanup failed for task={getattr(task, 'task_ref', None)}. Reason: {reason}. Error: {e}")
+
+    def _remove_task(self, task: Task, reason: str = "", cleanup: bool = True):
+        removed = False
+        with self.m_lock:
+            if task in self.m_tasks:
+                self.m_tasks.remove(task)
+                removed = True
+        if cleanup and removed:
+            self._cleanup_task(task, reason)
+        return removed
+
+    def _cleanup_all_tasks(self, reason: str = ""):
+        with self.m_lock:
+            tasks = list(self.m_tasks)
+            self.m_tasks.clear()
+        for task in tasks:
+            self._cleanup_task(task, reason)
+
+    def _get_redis_key(self, routing_key: str) -> str:
+        return f"vs:signal:{routing_key}"
+            
+    def _maintain_loop(self):
+        self._log("Maintain loop started.")
+        heartbeat_interval = 60
+        while not self.m_stop_event.is_set():
+            for _ in range(heartbeat_interval):
+                if self.m_stop_event.is_set():
+                    return
+                time.sleep(1.0)
+            
+            with self.m_lock:
+                tasks_to_check = list(self.m_tasks)
+                
+            if not tasks_to_check:
+                continue
+            
+            healthy_tasks = []
+            dead_tasks = []
+            now = time.time()
+            
+            for t in tasks_to_check:
+                if now >= t.next_remote_ping:
+                    try:
+                        t.instance.keep_alive()
+                        if t.instance.health_check(): 
+                            healthy_tasks.append(t)
+                            next_delay = random.randint(180, 300) 
+                            t.next_remote_ping = now + next_delay
+                            self._log(f"🛡️ Task={t.task_ref} keep-alive success. Next ping in {next_delay}s.")
+                        else:
+                            dead_tasks.append(t)
+                            self._log(f"♻️ Instance for task={t.task_ref} unhealthy.")
+                    except Exception as e:
+                        dead_tasks.append(t)
+                        self._log(f"♻️ Instance for task={t.task_ref} keep-alive failed: {e}.")
+                else:
+                    healthy_tasks.append(t)
+            
+            if healthy_tasks:
+                try:
+                    pipeline = self.redis_client.pipeline()
+                    new_deadline = time.time() + self.heartbeat_ttl
+                    for t in healthy_tasks:
+                        if t.task_ref is not None:
+                            pipeline.zadd(self.m_tracker_key, {str(t.task_ref): new_deadline})
+                    pipeline.execute()
+                    self._log(f"💓 Heartbeat sent. Renewed {len(healthy_tasks)} tasks.")
+                except Exception as e:
+                    self._log(f"Redis Heartbeat update failed: {e}")
+
+            if dead_tasks:
+                try:
+                    pipeline = self.redis_client.pipeline()
+                    for t in dead_tasks:
+                        if t.task_ref is not None:
+                            pipeline.zadd(self.m_tracker_key, {str(t.task_ref): 0})
+                    pipeline.execute()
+                    self._log(f"🗑️ Handed over {len(dead_tasks)} dead tasks to Sweeper.")
+                except Exception as e:
+                    pass
+            
+            if dead_tasks:
+                with self.m_lock:
+                    current_tasks = list(self.m_tasks)
+                    self.m_tasks = [t for t in self.m_tasks if t in healthy_tasks]
+                for t in dead_tasks:
+                    if t in current_tasks:
+                        self._cleanup_task(t, "unhealthy or keep-alive failed")
+            else:
+                with self.m_lock:
+                    self.m_tasks = [t for t in self.m_tasks if t in healthy_tasks]
+
+    def _booking_trigger_loop(self):
+        self._log("Trigger loop started.")
+        while not self.m_stop_event.is_set():
+            try:
+                time.sleep(1.0)
+                now = time.time()
+                for apt_type in self.m_cfg.appointment_types:
+                    redis_key = self._get_redis_key(apt_type.routing_key)
+                    raw_data = self.redis_client.get(redis_key)
+                    if not raw_data:
+                        continue
+                    try:
+                        data = json.loads(raw_data)
+                        query_result = VSQueryResult.model_validate(data['query_result'])
+                        query_result.apt_type = AppointmentType.model_validate(data['apt_type'])
+                    except Exception as parse_err:
+                        self._log(f"Data parsing error for {redis_key}: {parse_err}. Deleting corrupted signal.")
+                        self.redis_client.delete(redis_key)
+                        continue
+                    
+                    matching_tasks = []
+                    with self.m_lock:
+                        for task in self.m_tasks:
+                            if now < task.next_run or not task.book_allowed:
+                                continue
+                            if apt_type.routing_key not in task.acceptable_routing_keys:
+                                continue
+                            
+                            task.next_run = now + self.m_cfg.booker.booking_cooldown
+                            matching_tasks.append(task)
+                            
+                    if matching_tasks:
+                        threads = []
+                        for task in matching_tasks:
+                            self._log(f"🚀 Triggering BOOK for {apt_type.routing_key} | Order Ref: {task.task_ref}")
+                            t = threading.Thread(target=self._execute_book_job, args=(task, query_result))
+                            threads.append(t)
+                            t.start()
+                        
+                        for t in threads:
+                            t.join() 
+                    
+            except Exception as e:
+                self._log(f"Trigger loop error: {e}")
+                time.sleep(2)
+
+    def _execute_book_job(self, task: Task, query_result: VSQueryResult):
+        task_id = task.task_ref
+        task_data = None
+
+        try:
+            task_data = VSCloudApi.Instance().get_vas_task(task_id)
+            if not task_data or task_data.get('status') in ['grabbed', 'pause', 'completed', 'cancelled']:
+                self._log(f"Bound Task={task_id} is no longer valid or already processed. Removing instance.")
+                self._remove_task(task, "bound task no longer valid")
+                self.redis_client.zrem(self.m_tracker_key, task_id)
+                return
+            
+            order_id = task_data.get('order_id')
+            user_input = task_data.get('user_inputs', {})
+            book_res = task.instance.book(query_result, user_input)
+
+            if book_res.success:
+                self._log(f"✅ BOOK SUCCESS! Order: {order_id}. Destroying instance.")
+                grab_info = {
+                    "account": book_res.account,
+                    "session_id": book_res.session_id,
+                    "urn": book_res.urn,
+                    "slot_date": book_res.book_date,
+                    "slot_time": book_res.book_time,
+                    "timestamp": int(time.time()),
+                    "payment_link": book_res.payment_link
+                }
+                VSCloudApi.Instance().update_vas_task(task_id, {"status": "grabbed", "grabbed_history": grab_info})
+                push_content = (
+                    f"🎉 【预定成功通知】\n"
+                    f"━━━━━━━━━━━━━━━\n"
+                    f"订单编号: {order_id}\n"
+                    f"预约账号: {book_res.account}\n"
+                    f"预约日期: {book_res.book_date}\n"
+                    f"预约时间: {book_res.book_time}\n"
+                    f"预约编号: {book_res.urn}\n"
+                    f"支付链接: {book_res.payment_link if book_res.payment_link else '无需支付/暂无'}\n"
+                    f"━━━━━━━━━━━━━━━\n"
+                )
+                VSCloudApi.Instance().push_weixin_text(push_content)
+                self.redis_client.zrem(self.m_tracker_key, task_id)
+                self._remove_task(task, "booking success")
+            else:
+                self._log(f"❌ BOOK FAILED for Order: {order_id}. Will retry on next signal.")
+
+        except Exception as e:
+            err_str = str(e)
+            self._log(f"Exception during booking: {err_str}")
+            rate_limited_indicators = [
+                "42901" in err_str,
+                "Rate limited" in err_str
+            ]
+            if any(rate_limited_indicators):
+                self._remove_task(task, "booking rate limited")
+                if task_data and task_id is not None:
+                    task_meta = task_data.get('meta', {})
+                    t_fails = task_meta.get('booking_failures', 0) + 1
+                    task_meta['booking_failures'] = t_fails
+                    
+                    try:
+                        VSCloudApi.Instance().update_vas_task(task_id, {"meta": task_meta})
+                    except Exception as cloud_err:
+                        self._log(f"Failed to update task meta: {cloud_err}")
+                        
+                    t_cd = self.task_backoff.calculate(t_fails)
+                    self._log(f"⏳ Task={task_id} (Booking Attempt {t_fails}) suspended for {t_cd:.1f}s.")
+                    self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + t_cd})
+
+    def _creator_loop(self):
+        self._log("Creator loop started.")
+        spawn_interval = 10.0
+        while not self.m_stop_event.is_set():
+            time.sleep(2.0)
+            for apt in self.m_cfg.appointment_types:
+                r_key = apt.routing_key
+                
+                queue_cd_key = f"vs:queue:cooldown:{r_key}"
+                if self.redis_client.exists(queue_cd_key):
+                    continue
+                
+                with self.m_lock:
+                    active = sum(1 for t in self.m_tasks if getattr(t, 'source_queue', '') == r_key)
+                    pending = self.m_pending_order_by_queue.get(r_key, 0)
+                    target = self.m_cfg.booker.target_instances
+                
+                if (active + pending) < target:
+                    now = time.time()
+                    if now - self.m_last_spawn_time >= spawn_interval:
+                        self.m_last_spawn_time = now 
+                        self._spawn_worker(r_key)
+                        break
+
+    def _spawn_worker(self, target_routing_key: str):
+        with self.m_lock: 
+            self.m_pending_order_by_queue[target_routing_key] = self.m_pending_order_by_queue.get(target_routing_key, 0) + 1
+            
+        def _job():
+            success = False
+            task_id = None
+            is_rate_limited = False
+            
+            try:
+                queue_name = f"auto.{target_routing_key}"
+                task_data = VSCloudApi.Instance().get_vas_task_pop(queue_name)
+                if not task_data:
+                    return 
+                
+                task_id = task_data['id']
+                
+                self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + self.heartbeat_ttl})
+                user_inputs = task_data.get('user_inputs', {})
+                
+                plg_cfg = VSPlgConfig()
+                plg_cfg.debug = self.m_cfg.debug
+                plg_cfg.free_config = self.m_cfg.free_config
+                plg_cfg.session_max_life = self.m_cfg.session_max_life
+                plg_cfg.account.username = user_inputs.get("username", "")
+                plg_cfg.account.password = user_inputs.get("password", "")
+                if not plg_cfg.account.username:
+                    return
+                
+                acceptable_keys = [target_routing_key]
+                if self.m_cfg.need_proxy:
+                    proxy = VSCloudApi.Instance().get_next_proxy(self.m_cfg.proxy_pool, self.m_cfg.proxy_cd)
+                    plg_cfg.proxy.id = proxy['id']
+                    plg_cfg.proxy.ip = proxy['ip']
+                    plg_cfg.proxy.port = proxy['port']
+                    plg_cfg.proxy.proto = proxy['proto']
+                    plg_cfg.proxy.username = proxy['username']
+                    plg_cfg.proxy.password = proxy['password']
+
+                instance = self.m_factory.create(self.m_cfg.identifier, self.m_cfg.plugin_config.plugin_name)
+                instance.set_log(self.m_logger)
+                instance.set_config(plg_cfg)
+                instance.create_session() 
+                
+                with self.m_lock:
+                    self.m_tasks.append(
+                        Task(
+                            instance=instance,
+                            qw_cfg=self.m_cfg.query_wait,
+                            next_run=time.time(), 
+                            task_ref=task_id,
+                            acceptable_routing_keys=acceptable_keys, 
+                            source_queue=target_routing_key,
+                            book_allowed=True,
+                            next_remote_ping=time.time() + random.randint(180, 300) 
+                        )
+                    )
+                    queue_fail_key = f"vs:queue:failures:{target_routing_key}"
+                    self.redis_client.delete(queue_fail_key)                    
+                success = True
+                self._log(f"+++ Order Booker spawned: {plg_cfg.account.username} (Target: {acceptable_keys})")
+            except Exception as e:
+                err_str = str(e)
+                resource_not_found_indicators = [
+                    "40401" in err_str,
+                    "Account not found" in err_str,
+                    "Proxy not found" in err_str
+                ]
+                if any(resource_not_found_indicators):
+                    return
+                
+                self._log(f"Order Booker spawn failed: {e}")
+                
+                rate_limited_indicators = [
+                    "42901" in err_str,
+                    "Rate limited" in err_str
+                ]
+                if any(rate_limited_indicators):
+                    is_rate_limited = True
+                    queue_fail_key = f"vs:queue:failures:{target_routing_key}"
+                    queue_cd_key = f"vs:queue:cooldown:{target_routing_key}"
+                    q_fails = self.redis_client.incr(queue_fail_key)
+                    q_cd = self.queue_backoff.calculate(q_fails)
+                    self.redis_client.set(queue_cd_key, "1", ex=int(q_cd))
+                    self._log(f"📉 [Rate Limited] Queue '{target_routing_key}' failed {q_fails} times. Global Backoff: {q_cd:.1f}s.")
+                    if task_id is not None:
+                        task_meta = task_data.get('meta') or {}
+                        t_fails = task_meta.get('spawn_failures', 0) + 1
+                        task_meta['spawn_failures'] = t_fails
+                        
+                        try:
+                            VSCloudApi.Instance().update_vas_task(task_id, {"meta": task_meta})
+                        except Exception as cloud_err:
+                            self._log(f"Failed to update task meta: {cloud_err}")
+                        
+                        t_cd = self.account_backoff.calculate(t_fails)
+                        self._log(f"⏳ Task={task_id} (Attempt {t_fails}) suspended for {t_cd:.1f}s.")
+                        self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + t_cd})       
+            finally:
+                with self.m_lock: 
+                    self.m_pending_order_by_queue[target_routing_key] = max(0, self.m_pending_order_by_queue[target_routing_key] - 1)
+                
+                # 创建/登录失败，调用安全归还函数
+                if not success and task_id is not None and not is_rate_limited:
+                    self.redis_client.zadd(self.m_tracker_key, {str(task_id): 0})
+                    self._log(f"♻️ Task={task_id} failed normal spawn. Instantly handed over to Sweeper.")
         ThreadPool.getInstance().enqueue(_job)

config/config.json

@@ -9,7 +9,7 @@
         {
             "identifier": "vfs.ie.nl",
             "debug": false,
-            "enable": true,
+            "enable": false,
             "need_account": true,
             "need_proxy": true,
             "proxy_pool": ["proxy-cheap", "decodo"],
@@ -453,7 +453,7 @@
         {
             "identifier": "vfs.ie.at",
             "debug": false,
-            "enable": true,
+            "enable": false,
             "need_account": true,
             "need_proxy": true,
             "proxy_pool": ["proxy-cheap", "decodo"],
@@ -651,7 +651,7 @@
         {
             "identifier": "vfs.ie.hu",
             "debug": false,
-            "enable": true,
+            "enable": false,
             "need_account": true,
             "need_proxy": true,
             "proxy_pool": ["proxy-cheap", "decodo"],
@@ -974,14 +974,14 @@
             "enable": true,
             "need_account": true,
             "need_proxy": true,
-            "proxy_pool": ["proxy-cheap", "decodo"],
+            "proxy_pool": ["local"],
             "proxy_cd": 300,
             "session_max_life": 1800,
             "sentinel": {
                 "account_source": "built-in",
-                "account_pool_id": "gb.fr.sentinel",
+                "account_pool_id": "tls.gb.fr.sentinel",
                 "target_instances": 1,
-                "account_cd": 1800,
+                "account_cd": 300,
                 "signal_ttl": 30
             },
             "booker": {
@@ -1016,18 +1016,7 @@
             "free_config": {
                 "tls_url": "https://visas-fr.tlscontact.com/en-us/country/gb/vac/gbLON2fr",
                 "location": "London",
-                "capsolver_key": "CAP-5441DD341DD3CC2FAEF0BE6FE493EE9A",
-                "apt_config": {
-                    "code": "gbLON2fr",
-                    "country": "gb",
-                    "mission": "fr",
-                    "city": "London"
-                },
-                "interest_month": "06-2026",
-                "target_labels": [
-                    "",
-                    "pta"
-                ]
+                "capsolver_key": "CAP-5441DD341DD3CC2FAEF0BE6FE493EE9A"
             }
         },
         {
@@ -1088,7 +1077,7 @@
         {
             "identifier": "e-konsulat.ie.pl",
             "debug": false,
-            "enable": true,
+            "enable": false,
             "need_account": false,
             "need_proxy": true,
             "proxy_pool": ["proxy-cheap", "decodo"],
@@ -1339,7 +1328,7 @@
         {
             "identifier": "greekemba.ie.gr",
             "debug": false,
-            "enable": true,
+            "enable": false,
             "need_account": true,
             "need_proxy": true,
             "proxy_pool": ["proxy-cheap", "decodo"],

main_booker.py

@@ -1,29 +1,51 @@
 import time
 import json
+import argparse
+
 from vs_types import GroupConfig
 from gco_wrapper import GCOWrapper
 from logger_setup import setup_app_logger
 
-# 导入拆分后的两个类
 from booker_builtin import BuiltinBookerGCO
 from booker_order import OrderBookerGCO
 
+
+def load_config(path):
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
 def main():
+    # ===== 1️⃣ 命令行参数 =====
+    parser = argparse.ArgumentParser(description="Booker Runner")
+
+    parser.add_argument(
+        "-c", "--config",
+        type=str,
+        required=True,
+        help="Path to config.json"
+    )
+
+    args = parser.parse_args()
+
+    # ===== 2️⃣ 日志 =====
     app_logger = setup_app_logger("Booker")
-    app_logger.info("Booker Logger is ready! All VSC macros are hooked.")
+    app_logger.info("Booker Logger is ready!")
 
-    with open("config/config.json", "r", encoding="utf-8") as f:
-        cfg_data = json.load(f)
+    # ===== 3️⃣ 加载配置 =====
+    cfg_data = load_config(args.config)
 
     redis_conf = cfg_data.get('redis')
     groups_conf = cfg_data.get('group_list')
+
     wrappers = []
 
+    # ===== 4️⃣ 启动逻辑 =====
     for item in groups_conf:
         cfg = GroupConfig.from_json(item)
-        if not cfg.enable: continue
-            
-        # === 核心：根据配置自动选择实现类 ===
+        if not cfg.enable:
+            continue
+
         if cfg.booker.account_source == "order":
             gco_class = OrderBookerGCO
             app_logger.info(f"[{cfg.identifier}] Mode: ORDER (Bound)")
@@ -31,18 +53,27 @@ def main():
             gco_class = BuiltinBookerGCO
             app_logger.info(f"[{cfg.identifier}] Mode: BUILT-IN (Unbound)")
 
-        wrapper = GCOWrapper(gco_class=gco_class, gco_cfg=cfg, redis_conf=redis_conf)
+        wrapper = GCOWrapper(
+            gco_class=gco_class,
+            gco_cfg=cfg,
+            redis_conf=redis_conf
+        )
+
         wrapper.load()
         wrapper.start()
         wrappers.append(wrapper)
-    
-    app_logger.info(f"Successfully started {len(wrappers)} Booker groups. Press Ctrl+C to stop.")
-    
+
+    app_logger.info(f"Started {len(wrappers)} Booker groups.")
+
+    # ===== 5️⃣ 保持运行 =====
     try:
-        while True: time.sleep(1)
+        while True:
+            time.sleep(1)
     except KeyboardInterrupt:
-        app_logger.info("Shutting down Bookers...")
-        for wrapper in wrappers: wrapper.stop()
+        app_logger.info("Shutting down...")
+        for wrapper in wrappers:
+            wrapper.stop()
+
 
 if __name__ == "__main__":
     main()

main_sentinel.py

@@ -1,39 +1,68 @@
 import time
 import json
+import argparse
+
 from vs_types import GroupConfig
 from gco_wrapper import GCOWrapper
 from sentinel import SentinelGCO
 from logger_setup import setup_app_logger
 
+
+def load_config(path: str):
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
 def main():
+    # ===== 1️⃣ 命令行参数 =====
+    parser = argparse.ArgumentParser(description="Sentinel Runner")
+
+    parser.add_argument(
+        "-c", "--config",
+        type=str,
+        required=True,
+        help="Path to config.json"
+    )
+
+    args = parser.parse_args()
+
+    # ===== 2️⃣ logger =====
     app_logger = setup_app_logger("Sentinel")
     app_logger.info("Sentinel Logger is ready! All VSC macros are hooked.")
 
-    # 读取配置文件
-    with open("config/config.json", "r", encoding="utf-8") as f:
-        cfg_data = json.load(f)
+    # ===== 3️⃣ 读取配置 =====
+    cfg_data = load_config(args.config)
 
     redis_conf = cfg_data.get('redis')
     groups_conf = cfg_data.get('group_list')
 
-    # 用于保存所有启动的 wrapper，方便后续退出时清理
     wrappers = []
 
-    # 遍历 JSON 数组中的每一个组别配置
+    # ===== 4️⃣ 启动 groups =====
     for item in groups_conf:
         cfg = GroupConfig.from_json(item)
+
         if not cfg.enable:
             app_logger.info(f"Group [{cfg.identifier}] is disabled. Skipping.")
             continue
-            
+
         app_logger.info(f"Starting wrapper for group [{cfg.identifier}]...")
-        wrapper = GCOWrapper(gco_class=SentinelGCO, gco_cfg=cfg, redis_conf=redis_conf)
+
+        wrapper = GCOWrapper(
+            gco_class=SentinelGCO,
+            gco_cfg=cfg,
+            redis_conf=redis_conf
+        )
+
         wrapper.load()
         wrapper.start()
         wrappers.append(wrapper)
-    
-    app_logger.info(f"Successfully started {len(wrappers)} Sentinel groups. Press Ctrl+C to stop.")
-    
+
+    app_logger.info(
+        f"Successfully started {len(wrappers)} Sentinel groups. Press Ctrl+C to stop."
+    )
+
+    # ===== 5️⃣ keep alive =====
     try:
         while True:
             time.sleep(1)
@@ -42,5 +71,6 @@ def main():
         for wrapper in wrappers:
             wrapper.stop()
 
+
 if __name__ == "__main__":
     main()

plugins/ita_plugin.py

@@ -157,56 +157,15 @@ class ItaPlugin(IVSPlg):
 
         co.headless(False) 
         co.set_argument('--no-sandbox')
-        # co.set_argument('--disable-gpu')
+        co.set_argument('--disable-gpu')
         co.set_argument('--disable-dev-shm-usage')
         co.set_argument('--window-size=1920,1080')
         co.set_argument('--disable-blink-features=AutomationControlled')
-        co.set_argument('--ignore-gpu-blocklist') # 忽略无显卡黑名单
-        co.set_argument('--enable-webgl')         # 强制开启 WebGL
-        co.set_argument('--use-gl=angle')         # 使用 ANGLE 渲染后端
-        co.set_argument('--use-angle=swiftshader')# 强制使用 CPU 进行 3D 渲染 (这步最关键！)
         co.set_argument(f"--fingerprint={specific_fp.get('seed')}")
         co.set_argument(f"--fingerprint-platform={specific_fp.get('platform')}")
         co.set_argument(f"--fingerprint-brand={specific_fp.get('brand')}")
         try:
             self.page = ChromiumPage(co)
-            if self.config.debug:
-                self.page.get('https://example.com')
-                js_script = """
-                function getFingerprint() {
-                    let webglVendor = 'Unknown';
-                    let webglRenderer = 'Unknown';
-                    try {
-                        let canvas = document.createElement('canvas');
-                        let gl = canvas.getContext('webgl') || canvas.getContext('experimental-webgl');
-                        if (gl) {
-                            let debugInfo = gl.getExtension('WEBGL_debug_renderer_info');
-                            if (debugInfo) {
-                                webglVendor = gl.getParameter(debugInfo.UNMASKED_VENDOR_WEBGL);
-                                webglRenderer = gl.getParameter(debugInfo.UNMASKED_RENDERER_WEBGL);
-                            }
-                        }
-                    } catch(e) {}
-
-                    return {
-                        "User-Agent": navigator.userAgent,
-                        "Platform": navigator.userAgentData ? navigator.userAgentData.platform : navigator.platform,
-                        "Brands": navigator.userAgentData ? navigator.userAgentData.brands.map(b => b.brand).join(', ') : 'Not Supported',
-                        "CPU Cores": navigator.hardwareConcurrency,
-                        "Language": navigator.language,
-                        "Timezone": Intl.DateTimeFormat().resolvedOptions().timeZone,
-                        "WebGL Vendor": webglVendor,
-                        "WebGL Renderer": webglRenderer
-                    };
-                }
-                return getFingerprint();
-                """
-
-                fp_data = self.page.run_js(js_script)
-                self._log("================ 预检浏览器指纹数据 ================")
-                self._log(json.dumps(fp_data, indent=4, ensure_ascii=False))
-                self._log("====================================================")
-            
             login_url = f"{self._host}/Home"
             self._log(f"Navigating to {login_url}")
             self.page.get(login_url)

plugins/pol_plugin.py

@@ -20,7 +20,6 @@ from vs_plg import IVSPlg
 from vs_types import VSPlgConfig, AppointmentType, VSQueryResult, VSBookResult, AvailabilityStatus, TimeSlot, DateAvailability, NotFoundError, PermissionDeniedError, RateLimiteddError, SessionExpiredOrInvalidError, BizLogicError 
 from toolkit.vs_cloud_api import VSCloudApi
 from toolkit.proxy_tunnel import ProxyTunnel
-from utils.fingerprint_utils import FingerprintGenerator
 
 class BrowserResponse:
     def __init__(self, result_dict):
@@ -138,25 +137,13 @@ class PolPlugin(IVSPlg):
                 co.set_argument(f'--proxy-server={proxy_str}')
         else:
             self._log("[WARN] No proxy configured!")
-
-        fingerprint_gen = FingerprintGenerator()
-        specific_fp = fingerprint_gen.generate(self.config.account.username)
-        self._log(f'browser fingerprint={specific_fp}')
         
         co.headless(False)
         co.set_argument('--no-sandbox')
-        # co.set_argument('--disable-gpu')
+        co.set_argument('--disable-gpu')
         co.set_argument('--disable-dev-shm-usage') 
         co.set_argument('--window-size=1920,1080')
         co.set_argument('--disable-blink-features=AutomationControlled')
-        co.set_argument('--ignore-certificate-errors')
-        co.set_argument('--ignore-gpu-blocklist') # 忽略无显卡黑名单
-        co.set_argument('--enable-webgl')         # 强制开启 WebGL
-        co.set_argument('--use-gl=angle')         # 使用 ANGLE 渲染后端
-        co.set_argument('--use-angle=swiftshader')# 强制使用 CPU 进行 3D 渲染 (这步最关键！)
-        co.set_argument(f"--fingerprint={specific_fp.get('seed')}")
-        co.set_argument(f"--fingerprint-platform={specific_fp.get('platform')}")
-        co.set_argument(f"--fingerprint-brand={specific_fp.get('brand')}")
         try:
             self.page = ChromiumPage(co)
             url_home = "https://secure.e-konsulat.gov.pl"

plugins/tls_plugin.py

@@ -10,7 +10,6 @@ from datetime import datetime
 from typing import List, Dict, Optional, Any, Callable
 from urllib.parse import urljoin, urlparse, urlencode, parse_qs
 
-# DrissionPage 核心
 from DrissionPage import ChromiumPage, ChromiumOptions
 
 from vs_plg import IVSPlg
@@ -92,7 +91,7 @@ class TlsPlugin(IVSPlg):
         except SessionExpiredOrInvalidError as e:
             self.is_healthy = False
         except Exception as e:
-            pass
+            self._log(f"Unexpected error in keep_alive: {e}")
 
     def health_check(self) -> bool:
         if not self.is_healthy:
@@ -159,23 +158,26 @@ class TlsPlugin(IVSPlg):
                 co.set_argument(f'--proxy-server={proxy_str}')
         else:
             self._log("[WARN] No proxy configured!")
-            
-        fingerprint_gen = FingerprintGenerator()
-        specific_fp = fingerprint_gen.generate(self.config.account.username)
-        self._log(f'browser fingerprint={specific_fp}')
+
+
+        specific_fp = FingerprintGenerator().generate(self.config.account.username)
+        fp_seed = specific_fp.get("seed")
+        fp_platform = specific_fp.get("platform")
+        fp_brand = specific_fp.get("brand")
+        self._log(f'browser fingerprint seed={fp_seed}')
 
         co.headless(False)
         co.set_argument('--no-sandbox')
         co.set_argument('--disable-dev-shm-usage')
         co.set_argument('--window-size=1920,1080')
         co.set_argument('--disable-blink-features=AutomationControlled')
-        co.set_argument('--ignore-gpu-blocklist')
-        co.set_argument('--enable-webgl')
-        co.set_argument('--use-gl=angle')
-        co.set_argument('--use-angle=swiftshader')
-        co.set_argument(f"--fingerprint={specific_fp.get('seed')}")
-        co.set_argument(f"--fingerprint-platform={specific_fp.get('platform')}")
-        co.set_argument(f"--fingerprint-brand={specific_fp.get('brand')}")
+        # co.set_argument('--ignore-gpu-blocklist')
+        # co.set_argument('--enable-webgl')
+        # co.set_argument('--use-gl=angle')
+        # co.set_argument('--use-angle=swiftshader')
+        co.set_argument(f"--fingerprint={fp_seed}")
+        co.set_argument(f"--fingerprint-platform={fp_platform}")
+        co.set_argument(f"--fingerprint-brand={fp_brand}")
         
         try:
             self.page = ChromiumPage(co)
@@ -224,7 +226,7 @@ class TlsPlugin(IVSPlg):
             time.sleep(5)
             
             cf_bypasser = CloudflareBypasser(self.page, log=True)
-            if not cf_bypasser.bypass(max_retry=15):
+            if not cf_bypasser.bypass(max_retry=6):
                 raise BizLogicError("Cloudflare bypass timeout")
             time.sleep(3)
             cf_bypasser.handle_waiting_room()
@@ -239,7 +241,7 @@ class TlsPlugin(IVSPlg):
             init_y = random.randint(10, viewport_height - 10)
             self.mouse.move(init_x, init_y) 
 
-            max_steps = 20 
+            max_steps = 10 
             session_created = False
             has_submitted_login = False
             
@@ -248,6 +250,14 @@ class TlsPlugin(IVSPlg):
                 current_url = self.page.url
                 self._log(f"--- [Router Step {step+1}] Current URL: {current_url} ---")
                 
+                cloudflare_blocked_indicators = [
+                    "Sorry, you have been blocked" in self.page.html,
+                    "You are being rate limited" in self.page.html,
+                    "Cloudflare Ray ID" in self.page.html
+                ]
+                if any(cloudflare_blocked_indicators):
+                    raise BizLogicError(message="Blocked by Cloudflare WAF. Need to change IP or browser fingerprint.")
+                
                 # 状态 1：到达终极目标页面 (成功退出条件)
                 if "appointment-booking" in current_url or self.page.ele('tag:button@text():Book your appointment', timeout=1):
                     btn_selector = 'tag:button@text():Book your appointment'            
@@ -266,13 +276,25 @@ class TlsPlugin(IVSPlg):
                 if any(no_applicant_indicators):
                     raise BizLogicError(message="No applicant added. Cannot proceed to booking.") 
                 
-                # 状态 3：首页/登录入口页 -> 需要点击进入登录
-                if self.page.ele("tag:a@@href:login", timeout=1) and not self.page.ele('tag:label@@text():Email', timeout=1):
-                    self._log("State: Login Portal. Clicking login link...")
-                    login_link = self.page.ele("tag:a@@href:login")
-                    self.mouse.human_click_ele(login_link)
-                    time.sleep(3)
-                    continue
+                if current_url == tls_url:
+                    # 状态 3：首页/登录入口页 -> 需要点击进入登录
+                    if self.page.ele("tag:a@@href:login", timeout=1) and not self.page.ele('tag:label@@text():Email', timeout=1):
+                        self._log("State: Login Portal. Clicking login link...")
+                        login_link = self.page.ele("tag:a@@href:login")
+                        self.mouse.human_click_ele(login_link)
+                        time.sleep(3)
+                        continue
+                    if self.page.ele("tag:svg@@data-testid=user-button", timeout=1):
+                        self._log("State: Already login, logout now...")
+                        user_btn = self.page.ele("tag:svg@@data-testid=user-button")
+                        self.mouse.human_click_ele(user_btn)
+                        time.sleep(1.5)
+                        logout_btn = self.page.ele("#logout")
+                        self.mouse.human_click_ele(logout_btn)
+                        time.sleep(1.5)
+                        self.page.get(tls_url)
+                        time.sleep(3)
+                        continue
                 
                 # 状态 4：真正的登录表单页
                 if self.page.ele('tag:label@@text():Email', timeout=1) and not has_submitted_login:
@@ -385,51 +407,12 @@ class TlsPlugin(IVSPlg):
     def query(self, apt_type: AppointmentType) -> VSQueryResult:
         res = VSQueryResult()
         res.success = False
-        interest_month = self.free_config.get("interest_month", time.strftime("%m-%Y"))
-        
-        target_date_obj = datetime.strptime(interest_month, "%m-%Y")
-        target_month_text = target_date_obj.strftime("%B %Y")
-        target_year = target_date_obj.year
-        target_month_num = target_date_obj.month
-        
-        slots = []
-        current_selected_ele = self.page.ele('@data-testid=btn-current-month-available')
-        current_month_text = current_selected_ele.text.strip() if current_selected_ele else ""
-
-        is_on_target_month = (current_month_text.lower() == target_month_text.lower())
-
-        if not is_on_target_month:
-            self._log(f"Current is '{current_month_text}', navigating to '{target_month_text}'...")
-            reached_target = False
-            for step in range(12):
-                current_ele = self.page.ele('@data-testid=btn-current-month-available', timeout=2)
-                if current_ele and current_ele.text.strip().lower() == target_month_text.lower():
-                    self._log(f"✅ Successfully navigated to target month: '{target_month_text}'!")
-                    reached_target = True
-                    break
-                
-                next_btn = self.page.ele('@data-testid=btn-next-month-available', timeout=2)
-                
-                if next_btn and next_btn.tag.lower() == 'button':
-                    self._log(f"Clicking next month: {next_btn.text.strip()} ...")
-                    next_btn.click(by_js=True)
-                    time.sleep(2.5)
-                else:
-                    self._log("⚠️ Reached the end of the calendar or 'Next Month' is disabled.")
-                    break
 
-            if not reached_target:
-                self._log(f"❌ Could not navigate to target month: {target_month_text}. Stop parsing.")
-                res.success = False
-                res.availability_status = AvailabilityStatus.NoneAvailable
-                return res
-            self._log("Extracting slots from DOM using robust data-testid features...")
-            slots = self._scan_dom_for_slots(target_year, target_month_num)
-        else:
-            self._log(f"Already on '{target_month_text}'. Executing silent JS fetch...") 
-            resp = self._perform_request("GET", self.page.url, retry_count=1)
-            self._check_page_is_session_expired_or_invalid('Book your appointment', resp.text)
-            slots = self._parse_appointment_slots(resp.text)
+        slots = []
+        self._log(f"Executing silent JS fetch...") 
+        resp = self._perform_request("GET", self.page.url, retry_count=1)
+        self._check_page_is_session_expired_or_invalid('Book your appointment', resp.text)
+        slots = self._parse_appointment_slots(resp.text)
         
         if slots:
             res.success = True
@@ -451,8 +434,8 @@ class TlsPlugin(IVSPlg):
             res.success = False
             res.availability_status = AvailabilityStatus.NoneAvailable
         return res
-
-    def book(self, slot_info: VSQueryResult, user_inputs: Dict = None) -> VSBookResult:
+    
+    def book_bak(self, slot_info: VSQueryResult, user_inputs: Dict = None) -> VSBookResult:
         res = VSBookResult()
         res.success = False
         
@@ -497,87 +480,223 @@ class TlsPlugin(IVSPlg):
         selected_label = selected_slot["label"]
 
         self._log(f"Found {len(all_possible_slots)} valid slots. selected slot: {selected_date} {selected_time.time} {selected_label}")
+        self.page.listen.start('/workflow/appointment-booking', method='POST')
+        page_url = self.page.url
+        location = self.travel_group.get('location')
+        mapper = {
+            "London": "gbLON2fr",
+            "Dublin": "ieDUB2fr"
+        }
+        location_id = mapper.get(location)
+        form_group_id =self.travel_group.get('group_number')
         
-        # ================== 新增：随机选择预订模式 ==================
-        book_mode = random.choice([1, 2])
-        self._log(f"Using booking mode: {book_mode}")
+        api_token = self.free_config.get("capsolver_key", "")
+        rc_params = {
+            "type": "ReCaptchaV3Task",
+            "page": page_url,
+            "action": "book", 
+            "siteKey": "6LcTpXcfAAAAAM3VojNhyV-F1z92ADJIvcSZ39Y9",
+            "apiToken": api_token,
+            "proxy": True
+        }
+        g_token = self._solve_recaptcha(rc_params)
+
+        ACTION_ID = "6033ac5e6e4ac04f59a4b74a9c5dd312876dd46bd9"
         
-        if book_mode == 1:
-            rand_x = random.randint(300, 800)
-            rand_y = random.randint(400, 700)
-            self._log(f"Mode 1: Moving mouse to ({rand_x}, {rand_y}) and clicking, select slot")
-            self.mouse.click(rand_x, rand_y, humanize=True)
+        js_script = f"""
+        (async function() {{
+            const url = "{page_url}";
+            const groupId = "{form_group_id}";
+            const locationId = "{location_id}";
+            const selectedDate = "{selected_date}";
+            const selectedTime = "{selected_time.time}";
+            const selectedLabel = "{selected_label}";
+            const captchaToken = "{g_token}";
+            const actionId = "{ACTION_ID}";
             
-            js_update_form = f"""
+            let castleToken = '';
             try {{
-                const buttons = Array.from(document.querySelectorAll('button[type="submit"]'));
-                const submitBtn = buttons.find(btn => {{
-                    return btn.textContent.trim().toLowerCase().includes('book your appointment');
-                }});
-                if (!submitBtn) return 'Submit button not found';
-                const form = submitBtn.closest('form');
-                if (!form) return 'Correct form not found';
-                function setReactValue(input, value) {{
-                    if (!input) return;
-                    input.value = value;
-                    input.dispatchEvent(new Event('input', {{ bubbles: true }}));
-                    input.dispatchEvent(new Event('change', {{ bubbles: true }}));
+                const castleModule = window.__webpack_require__(93773);
+                if (castleModule && castleModule.createRequestToken) {{
+                    castleToken = await castleModule.createRequestToken();
+                    console.log('Castle token obtained:', castleToken);
                 }}
-                setReactValue(form.querySelector('input[name="date"]'), '{selected_date}');
-                setReactValue(form.querySelector('input[name="time"]'), '{selected_time.time}');
-                setReactValue(form.querySelector('input[name="appointmentLabel"]'), '{selected_label}');
-                submitBtn.removeAttribute('disabled');
-                submitBtn.classList.remove('opacity-50', 'cursor-not-allowed'); 
-                return 'form_updated';
-            }} catch (e) {{
-                return e.toString();
+            }} catch(e) {{
+                console.error('Failed to get Castle token:', e);
             }}
-            """
-            update_res = self.page.run_js(js_update_form)
-            self._log(f"Mode 1: Form update triggered: {update_res}")
             
-            if update_res != 'form_updated':
-                raise BizLogicError(message=f"Failed to update form in Mode 1: {update_res}")
+            let routerStateTree = '';
+            try {{
+                const routerModule = window.__webpack_require__(11807);
+                const routerState = routerModule.getCurrentAppRouterState();
+                if (routerState && routerState.tree) {{
+                    const prepareModule = window.__webpack_require__(16378);
+                    routerStateTree = prepareModule.prepareFlightRouterStateForRequest(routerState.tree);
+                    console.log('Router state tree obtained');
+                }}
+            }} catch(e) {{
+                console.error('Failed to get router state:', e);
+            }}
+            
+            const formData = new FormData();
+            formData.append('1_formGroupId', groupId);
+            formData.append('1_lang', 'en-us');
+            formData.append('1_process', 'APPOINTMENT');
+            formData.append('1_location', locationId);
+            formData.append('1_date', selectedDate);
+            formData.append('1_time', selectedTime);
+            formData.append('1_appointmentLabel', selectedLabel);
+            formData.append('1_castleRequestToken', castleToken);
+            formData.append('1_captchaToken', captchaToken);
+            formData.append('0', '[{{"status":"IDLE"}},"$K1"]');
+
+            const headers = {{
+                'Accept': 'text/x-component',
+                'Next-Action': actionId,
+                'Next-Router-State-Tree': routerStateTree,
+            }};
             
-            submit_btn = self.page.ele('tag:button@@type=submit@@text():Book your appointment')
-            if not submit_btn:
-                raise BizLogicError(message="Submit button not found for mouse click")
+            const response = await fetch(url, {{
+                method: 'POST',
+                headers: headers,
+                body: formData,
+                credentials: 'include'
+            }});
+            
+            const text = await response.text();
+            const responseHeaders = {{}};
+            response.headers.forEach((value, key) => {{
+                responseHeaders[key] = value;
+            }});
+            const result = {{
+                status: response.status,
+                body: text,
+                headers: responseHeaders,
+                url: response.url,
+                redirected: response.redirected,
+                ok: response.ok
+            }};
+            return result;
+        }})();
+        """
+        
+        self._log("Submitting booking request via JS Fetch...")
+        self.page.run_js(js_script)
+        
+        packet = self.page.listen.wait(timeout=10)
+        if not packet:
+            raise BizLogicError(message='Listening data failed')
+        
+        self.page.listen.stop()
+        self._log(f"URL: {packet.url}")
+        self._log(f"POST Body: {packet.request.postData}")
+        self._log(f"POST Stat: {packet.response.status}")
+        self._log(f"POST head: {packet.response.headers}")
+        self._log(f"POST Resp: {packet.response.raw_body}")
+        redirect_location = packet.response.headers.get('location', '') or ''
+        appointment_confirmation_indicators = [
+            "order-summary" in redirect_location,
+            "partner-services" in redirect_location,
+            "appointment-confirmation" in redirect_location,
+        ]
+        if any(appointment_confirmation_indicators):
+            self._log(f"Booking Success!")
+            res.success = True
+            res.book_date = selected_date
+            res.book_time = selected_time.time
+            return res
+        return res
+
+    def book(self, slot_info: VSQueryResult, user_inputs: Dict = None) -> VSBookResult:
+        res = VSBookResult()
+        res.success = False
+        
+        exp_start = user_inputs.get('expected_start_date', '')
+        exp_end = user_inputs.get('expected_end_date', '')
+        support_pta = user_inputs.get('support_pta', True)
+
+        target_labels = ['']
+        if support_pta:
+            target_labels.append('pta')
+
+        available_dates_str =[
+            da.date.strftime("%Y-%m-%d")
+            for da in slot_info.availability if da.date
+        ]
+        
+        valid_dates_list = self._filter_dates(available_dates_str, exp_start, exp_end)
+        if not valid_dates_list:
+            raise NotFoundError(message="No dates match user constraints")
+        
+        all_possible_slots =[]
+        for da in slot_info.availability:
+            if not da.date:
+                continue
                 
-            self._log("Mode 1: Moving mouse to submit button and clicking")
-            self.mouse.human_click_ele(submit_btn)
-            inject_res = 'clicked'
+            date_str = da.date.strftime("%Y-%m-%d")
+            if date_str in valid_dates_list:
+                for t in da.times:
+                    if t.label in target_labels:
+                        all_possible_slots.append({
+                            "date": date_str,
+                            "time_obj": t,
+                            "label": t.label
+                        })
+        
+        if not all_possible_slots:
+            raise NotFoundError(message="No suitable slot found (after label filtering)")
 
-        else:
-            js_inject_and_click = f"""
-            try {{
-                const buttons = Array.from(document.querySelectorAll('button[type="submit"]'));
-                const submitBtn = buttons.find(btn => {{
-                    return btn.textContent.trim().toLowerCase().includes('book your appointment');
-                }});
-                if (!submitBtn) return 'Submit button not found';
-                const form = submitBtn.closest('form');
-                if (!form) return 'Correct form not found';
-                function setReactValue(input, value) {{
-                    if (!input) return;
-                    input.value = value;
-                    input.dispatchEvent(new Event('input', {{ bubbles: true }}));
-                    input.dispatchEvent(new Event('change', {{ bubbles: true }}));
-                }}
-                setReactValue(form.querySelector('input[name="date"]'), '{selected_date}');
-                setReactValue(form.querySelector('input[name="time"]'), '{selected_time.time}');
-                setReactValue(form.querySelector('input[name="appointmentLabel"]'), '{selected_label}');
-                submitBtn.removeAttribute('disabled');
-                submitBtn.click();
-                return 'clicked';
-            }} catch (e) {{
-                return e.toString();
+        selected_slot = random.choice(all_possible_slots)
+        selected_date = selected_slot["date"]
+        selected_time = selected_slot["time_obj"]
+        selected_label = selected_slot["label"]
+
+        self._log(f"Found {len(all_possible_slots)} valid slots. selected slot: {selected_date} {selected_time.time} {selected_label}")
+        self.page.listen.start('/workflow/appointment-booking', method='POST')
+        js_update_form = f"""
+        try {{
+            const buttons = Array.from(document.querySelectorAll('button[type="submit"]'));
+            const submitBtn = buttons.find(btn => {{
+                return btn.textContent.trim().toLowerCase().includes('book your appointment');
+            }});
+            if (!submitBtn) return 'Submit button not found';
+            const form = submitBtn.closest('form');
+            if (!form) return 'Correct form not found';
+            function setReactValue(input, value) {{
+                if (!input) return;
+                input.value = value;
             }}
-            """
-            inject_res = self.page.run_js(js_inject_and_click)
-            self._log(f"Mode 2: Form submission triggered: {inject_res}")
+            setReactValue(form.querySelector('input[name="date"]'), '{selected_date}');
+            setReactValue(form.querySelector('input[name="time"]'), '{selected_time.time}');
+            setReactValue(form.querySelector('input[name="appointmentLabel"]'), '{selected_label}');
+            submitBtn.removeAttribute('disabled');
+            submitBtn.classList.remove('opacity-50', 'cursor-not-allowed'); 
+            return 'form_updated';
+        }} catch (e) {{
+            return e.toString();
+        }}
+        """
+        update_res = self.page.run_js(js_update_form)
+        self._log(f"Form update triggered: {update_res}")
         
-        if inject_res != 'clicked':
-            raise BizLogicError(message="Failed to inject form or click the submit button")
+        if update_res != 'form_updated':
+            raise BizLogicError(message=f"Failed to update form: {update_res}")
+        
+        submit_btn = self.page.ele('tag:button@@type=submit@@text():Book your appointment')
+        if not submit_btn:
+            raise BizLogicError(message="Submit button not found for mouse click")
+            
+        self._log("Moving mouse to submit button and clicking")
+        self.mouse.human_click_ele(submit_btn)
+            
+        packet = self.page.listen.wait(timeout=10)
+        if not packet:
+            raise BizLogicError(message='Listening data failed')
+        
+        self.page.listen.stop()
+        self._log(f"URL: {packet.url}")
+        self._log(f"POST Body: {packet.request.postData}")
+        self._log(f"POST Resp: {packet.response.body}")
 
         self._log("Waiting for Next.js to process the form submission...")
         for _ in range(10):
@@ -612,80 +731,6 @@ class TlsPlugin(IVSPlg):
                 pass
         return res
 
-    def _scan_dom_for_slots(self, target_year: int, target_month_num: int) -> list[dict]:
-        """
-        DOM-based slot scanning — 结合区块结构与类名/属性推断标签
-        """
-        slots = []
-        
-        day_blocks_xpath = '//div[p and div//button[contains(@data-testid, "slot")]]'
-        day_blocks = self.page.eles(f'xpath:{day_blocks_xpath}')
-        
-        for block in day_blocks:
-            p_ele = block.ele('tag:p')
-            if not p_ele: continue
-            
-            day_match = re.search(r'\d+', p_ele.text)
-            if not day_match: continue
-            day_str = day_match.group()
-            full_date = f"{target_year}-{target_month_num:02d}-{int(day_str):02d}"
-            
-            btn_selectors = [
-                'xpath:.//button[starts-with(@data-testid, "btn-available-slot")]',
-                'xpath:.//button[contains(@class, "available")]',
-                'xpath:.//div[contains(@class, "time-slot") and not(contains(@class, "unavailable"))]'
-            ]
-            
-            available_elements = []
-            for sel in btn_selectors:
-                elements = block.eles(sel)
-                if elements:
-                    available_elements.extend(elements)
-                    break
-            seen_times = set()
-            for el in available_elements:
-                classes = (el.attr("class") or "").lower()
-                test_id = (el.attr("data-testid") or "").lower()
-                combined_attrs = f"{classes} {test_id}"
-                if "disabled" in combined_attrs or "unavailable" in combined_attrs:
-                    continue
-                time_match = re.search(r'\d{2}:\d{2}', el.html)
-                if not time_match: continue
-                time_str = time_match.group()
-                
-                if time_str in seen_times:
-                    continue
-                seen_times.add(time_str)
-                label = ''  
-                if 'prime' in combined_attrs and 'weekend' in combined_attrs:
-                    label = 'ptaw'
-                elif 'prime' in combined_attrs or 'premium' in combined_attrs:
-                    label = 'pta'
-                elif any(k in combined_attrs for k in ['regular', 'standard', 'default']):
-                    label = ''
-                else:
-                    label = '' 
-                
-                slots.append({
-                    'date': full_date,
-                    'time': time_str,
-                    'label': label,
-                    'source': 'dom'
-                })
-                
-        return slots
-
-    def _get_proxy_url(self):
-        # 构造代理
-        proxy_url = ""
-        if self.config.proxy.ip:
-            s = self.config.proxy
-            if s.username:
-                proxy_url = f"{s.proto}://{s.username}:{s.password}@{s.ip}:{s.port}"
-            else:
-                proxy_url = f"{s.proto}://{s.ip}:{s.port}"
-        return proxy_url
-
     def _perform_request(self, method, url, headers=None, data=None, json_data=None, params=None, retry_count=0):
         """
         在浏览器上下文中注入 JS 执行 Fetch
@@ -815,17 +860,14 @@ class TlsPlugin(IVSPlg):
         if params.get("action"):
             task["pageAction"] = params.get("action")
             
-        # if params.get("proxy"):
-        #     p = urlparse(params.get("proxy"))
-        #     task["proxyType"] = p.proto
-        #     task["proxyAddress"] = p.hostname
-        #     task["proxyPort"] = p.port
-        #     if p.username:
-        #         task["proxyLogin"] = p.username
-        #         task["proxyPassword"] = p.password
-            
-        # 注意：使用 DrissionPage 后，通常是 ProxyLess 模式
-        # 除非你想让 Capsolver 也用同样的代理（通常不需要，除非风控极严）
+        if params.get("proxy"):
+            p = self.config.proxy
+            task["proxyType"] = p.proto
+            task["proxyAddress"] = p.ip
+            task["proxyPort"] = p.port
+            if p.username:
+                task["proxyLogin"] = p.username
+                task["proxyPassword"] = p.password
         
         payload = {"clientKey": key, "task": task}
         import requests as req # 局部引用，避免混淆
@@ -873,17 +915,16 @@ class TlsPlugin(IVSPlg):
                 d_str = day.get('day')
                 for s in day.get('slots', []):
                     labels = s.get('labels', [])
-                    lbl = ""
-                    # 简化逻辑：TLS label 列表
+                    lbl = None
                     if 'pta' in labels: lbl = 'pta'
                     elif 'ptaw' in labels: lbl = 'ptaw'
-                    elif '' in labels or not labels: lbl = ''
-                    
-                    slots.append({
-                        'date': d_str,
-                        'time': s.get('time'),
-                        'label': lbl
-                    })
+                    elif '' in labels: lbl = ''
+                    if lbl is not None:
+                        slots.append({
+                            'date': d_str,
+                            'time': s.get('time'),
+                            'label': lbl
+                        })
         return slots
   
     def _check_page_is_session_expired_or_invalid(self, keyword, html: str) -> bool:
@@ -923,7 +964,7 @@ class TlsPlugin(IVSPlg):
         # 1. 关闭浏览器
         if self.page:
             try:
-                self.page.quit() # 这会关闭 Chrome 进程
+                self.page.quit(force=True) # 这会关闭 Chrome 进程
             except Exception:
                 pass # 忽略已关闭的错误
             self.page = None

plugins/vfs_plugin.py

@@ -25,7 +25,6 @@ from vs_types import VSPlgConfig, AppointmentType, VSQueryResult, VSBookResult,
 from toolkit.vs_cloud_api import VSCloudApi
 from toolkit.proxy_tunnel import ProxyTunnel
 from utils.cloudflare_bypass_for_scraping import CloudflareBypasser
-from utils.fingerprint_utils import FingerprintGenerator
 
 
 VFS_PUBLIC_KEY_PEM = """-----BEGIN PUBLIC KEY-----
@@ -238,63 +237,14 @@ class VfsPlugin(IVSPlg):
         else:
             self._log("[WARN] No proxy configured!")
             
-        fingerprint_gen = FingerprintGenerator()
-        specific_fp = fingerprint_gen.generate(self.config.account.username)
-        self._log(f'browser fingerprint={specific_fp}')
-            
         co.headless(False) 
         co.set_argument('--no-sandbox')
         # co.set_argument('--disable-gpu')
         co.set_argument('--disable-dev-shm-usage') 
         co.set_argument('--window-size=1920,1080')
         co.set_argument('--disable-blink-features=AutomationControlled')
-        co.set_argument('--ignore-gpu-blocklist') # 忽略无显卡黑名单
-        co.set_argument('--enable-webgl')         # 强制开启 WebGL
-        co.set_argument('--use-gl=angle')         # 使用 ANGLE 渲染后端
-        co.set_argument('--use-angle=swiftshader')# 强制使用 CPU 进行 3D 渲染 (这步最关键！)
-        co.set_argument(f"--fingerprint={specific_fp.get('seed')}")
-        co.set_argument(f"--fingerprint-platform={specific_fp.get('platform')}")
-        co.set_argument(f"--fingerprint-brand={specific_fp.get('brand')}")
         try:
             self.page = ChromiumPage(co)
-            
-            if self.config.debug:
-                self.page.get('https://example.com')
-                js_script = """
-                function getFingerprint() {
-                    let webglVendor = 'Unknown';
-                    let webglRenderer = 'Unknown';
-                    try {
-                        let canvas = document.createElement('canvas');
-                        let gl = canvas.getContext('webgl') || canvas.getContext('experimental-webgl');
-                        if (gl) {
-                            let debugInfo = gl.getExtension('WEBGL_debug_renderer_info');
-                            if (debugInfo) {
-                                webglVendor = gl.getParameter(debugInfo.UNMASKED_VENDOR_WEBGL);
-                                webglRenderer = gl.getParameter(debugInfo.UNMASKED_RENDERER_WEBGL);
-                            }
-                        }
-                    } catch(e) {}
-
-                    return {
-                        "User-Agent": navigator.userAgent,
-                        "Platform": navigator.userAgentData ? navigator.userAgentData.platform : navigator.platform,
-                        "Brands": navigator.userAgentData ? navigator.userAgentData.brands.map(b => b.brand).join(', ') : 'Not Supported',
-                        "CPU Cores": navigator.hardwareConcurrency,
-                        "Language": navigator.language,
-                        "Timezone": Intl.DateTimeFormat().resolvedOptions().timeZone,
-                        "WebGL Vendor": webglVendor,
-                        "WebGL Renderer": webglRenderer
-                    };
-                }
-                return getFingerprint();
-                """
-
-                fp_data = self.page.run_js(js_script)
-                self._log("================ 预检浏览器指纹数据 ================")
-                self._log(json.dumps(fp_data, indent=4, ensure_ascii=False))
-                self._log("====================================================")
-            
             mission = self.free_config.get("mission_code", "")
             country = self.free_config.get("country_code", "")
             lang = self.free_config.get("language", "en")

requirements.txt

@@ -1,6 +1,4 @@
 DrissionPage>=4.0.0
-camoufox
-camoufox[geoip]
 bs4
 redis
 cryptography
@@ -10,7 +8,6 @@ easyocr
 fastapi
 numpy
 pydantic
-requests-toolbelt
 requests
 uvicorn
 psutil

sentinel.py

@@ -58,6 +58,28 @@ class SentinelGCO:
     def stop(self):
         self._log("Stopping Sentinel...")
         self.m_stop_event.set()
+        with self.m_lock:
+            tasks_to_cleanup = list(self.m_tasks)
+            self.m_tasks.clear()
+        for task in tasks_to_cleanup:
+            self._cleanup_task(task, "sentinel stopped")
+
+    def _cleanup_task(self, task: Task, reason: str):
+        try:
+            if task and task.instance and hasattr(task.instance, "cleanup"):
+                self._log(f"Cleaning up sentinel instance. reason={reason}")
+                task.instance.cleanup()
+        except Exception as e:
+            self._log(f"Cleanup failed. reason={reason}, error={e}")
+
+    def _remove_task(self, task: Task, reason: str):
+        removed = False
+        with self.m_lock:
+            if task in self.m_tasks:
+                self.m_tasks.remove(task)
+                removed = True
+        if removed:
+            self._cleanup_task(task, reason)
 
     def _get_redis_key(self, routing_key: str) -> str:
         return f"vs:signal:{routing_key}"
@@ -72,8 +94,30 @@ class SentinelGCO:
                 now = time.time()
                 
                 with self.m_lock:
-                    active_tasks = [t for t in self.m_tasks if t.instance.health_check()]
-                    self.m_tasks = active_tasks
+                    tasks_to_check = list(self.m_tasks)
+
+                active_tasks = []
+                dead_tasks = []
+                for t in tasks_to_check:
+                    try:
+                        if t.instance.health_check():
+                            active_tasks.append(t)
+                        else:
+                            dead_tasks.append(t)
+                    except Exception as e:
+                        dead_tasks.append(t)
+                        self._log(f"Health check failed: {e}")
+
+                if dead_tasks:
+                    with self.m_lock:
+                        current_tasks = list(self.m_tasks)
+                        self.m_tasks = [t for t in self.m_tasks if t in active_tasks]
+                    for t in dead_tasks:
+                        if t in current_tasks:
+                            self._cleanup_task(t, "health check failed")
+                else:
+                    with self.m_lock:
+                        self.m_tasks = [t for t in self.m_tasks if t in active_tasks]
                 
                 for task in active_tasks:
                     if now < task.next_run:
@@ -155,6 +199,8 @@ class SentinelGCO:
             self.m_pending_builtin += 1
             
         def _job():
+            instance = None
+            success = False
             try:
                 plg_cfg = VSPlgConfig()
                 plg_cfg.debug = self.m_cfg.debug
@@ -191,6 +237,7 @@ class SentinelGCO:
                     group_fail_key = f"vs:group:failures:{self.m_cfg.identifier}"
                     self.redis_client.delete(group_fail_key)
                 
+                success = True
                 self._log(f"+++ Sentinel spawned: {plg_cfg.account.username}")
 
             except Exception as e:
@@ -219,6 +266,12 @@ class SentinelGCO:
                     self._log(f"📉 [Rate Limited] Sentinel Spawn failed {g_fails} times. Global Backoff: {g_cd:.1f}s.")
                     
             finally:
+                if not success and instance is not None:
+                    try:
+                        if hasattr(instance, "cleanup"):
+                            instance.cleanup()
+                    except Exception as e:
+                        self._log(f"Cleanup failed after spawn failure: {e}")
                 with self.m_lock:
                     self.m_pending_builtin = max(0, self.m_pending_builtin - 1)
 

tls_registration_bot.py

@@ -41,7 +41,7 @@ def generate_random_account_detail() -> Dict:
         "pool_name": "tls.gb.fr.sentinel",
         "email": "user@gmail-app.com",
         "pwd": "Visafly@111",
-        "location": "London",
+        "location": "Wandsworth (London)",
         "visa_type": "Short stay (<90 days) - Tourism",
         "travel_purpose": "Tourism / Private visit",
         # FRA1LO20260411910
@@ -78,7 +78,7 @@ def generate_random_account_detail() -> Dict:
         gender = "Male" if gender_raw == "male" else "Female"
         birthday_raw = result.get("dob", {}).get("date", "")
         birthday = birthday_raw[:10] if birthday_raw else default_payload["birthday"]
-        city = result.get("location", {}).get("city") or default_payload["location"]
+        city = default_payload["location"]
         state = result.get("location", {}).get("state") or default_payload["province_residence"]
         country = result.get("location", {}).get("country") or default_payload["nationality"]
         phone_raw = result.get("cell") or result.get("phone") or default_payload["phone_number"]
@@ -412,10 +412,10 @@ class TlsRegistrator:
             time.sleep(3)
         if not self.page.wait.ele_displayed(btn_selector, timeout=10):
             raise BizLogicError(message=f"Can't find selector={btn_selector}")
-        # recpatchav2_token = ""
-        # if self.page.ele('.g-recaptcha') or self.page.ele('xpath://iframe[contains(@src, "recaptcha")]'):
-        #     self._log("Solving ReCaptcha...")
-        #     recpatchav2_token = self.solve_captcha(self.page.url, "ReCaptchaV2TaskProxyLess", "6LcDpXcfAAAAAM7wOEsF_38DNsL20tTvPTKxpyn0")
+        recpatchav2_token = ""
+        if self.page.ele('.g-recaptcha') or self.page.ele('xpath://iframe[contains(@src, "recaptcha")]'):
+            self._log("Solving ReCaptcha...")
+            recpatchav2_token = self.solve_captcha(self.page.url, "ReCaptchaV2TaskProxyLess", "6LcDpXcfAAAAAM7wOEsF_38DNsL20tTvPTKxpyn0")
         
         input_ele = self.page.ele('tag:label@@text():Email').next()
         self.mouse.human_click_ele(input_ele)
@@ -429,14 +429,14 @@ class TlsRegistrator:
         time.sleep(random.uniform(0.2, 0.6))
         self.keyboard.type_text(password, humanize=True)
         
-        # if recpatchav2_token:
-        #     inject_recpatchav2_token_js = f"""
-        #     var g = document.getElementById('g-recaptcha-response');
-        #     if(g) {{ g.value = "{recpatchav2_token}"; }}
-        #     """
-        #     self._log("Inject ReCaptchaV2 Token via JS...")
-        #     self.page.run_js(inject_recpatchav2_token_js)
-        #     time.sleep(random.uniform(0.5, 1.0))
+        if recpatchav2_token:
+            inject_recpatchav2_token_js = f"""
+            var g = document.getElementById('g-recaptcha-response');
+            if(g) {{ g.value = "{recpatchav2_token}"; }}
+            """
+            self._log("Inject ReCaptchaV2 Token via JS...")
+            self.page.run_js(inject_recpatchav2_token_js)
+            time.sleep(random.uniform(0.5, 1.0))
         
         self._log("Submitting Login...")
         time.sleep(random.uniform(0.3, 0.8))
@@ -455,37 +455,37 @@ class TlsRegistrator:
         time.sleep(5)
 
         # 解析 Dashboard 提取 Group ID
-        # self._log("Parsing Dashboard for Travel Group...")
-        # html = self.page.html
-        # js_pattern = r'\\"travelGroups\\":\s*(\[.*?\]),\\"availableCountriesToCreateGroups'
-        # js_match = re.search(js_pattern, html, re.DOTALL)
+        self._log("Parsing Dashboard for Travel Group...")
+        html = self.page.html
+        js_pattern = r'\\"travelGroups\\":\s*(\[.*?\]),\\"availableCountriesToCreateGroups'
+        js_match = re.search(js_pattern, html, re.DOTALL)
         
-        # groups = []
-        # if js_match:
-        #     json_str = js_match.group(1).replace(r'\"', '"')
-        #     groups = json.loads(json_str)
-
-        # travel_group = None
-        # for g in groups:
-        #     if g.get('vacName', '').lower() == location.lower():
-        #         travel_group = g
-        #         break
+        groups = []
+        if js_match:
+            json_str = js_match.group(1).replace(r'\"', '"')
+            groups = json.loads(json_str)
+
+        travel_group = None
+        for g in groups:
+            if g.get('vacName', '').lower() == location.lower():
+                travel_group = g
+                break
         
-        # if not travel_group:
-        #     raise BizLogicError(message=f"Travel Group not found for {location}")
+        if not travel_group:
+            raise BizLogicError(message=f"Travel Group not found for {location}")
         
-        # formgroup_id = travel_group.get('formGroupId')
-        # self._log(f"Waiting for group button to render: {formgroup_id}")
-        # btn_selector = f'tag:a@@data-testid=btn-select-group'            
-        # self._log(f"Select group_id={formgroup_id}...")
-        # self.mouse.human_click_ele(self.page.ele(btn_selector))
+        formgroup_id = travel_group.get('formGroupId')
+        self._log(f"Waiting for group button to render: {formgroup_id}")
+        btn_selector = f'tag:a@@data-testid=btn-select-group'            
+        self._log(f"Select group_id={formgroup_id}...")
+        self.mouse.human_click_ele(self.page.ele(btn_selector))
         
-        # self._log("Waiting for url redirect...")
-        # self.page.wait.url_change('travel-groups', exclude=True, timeout=45)
-        # time.sleep(2)
+        self._log("Waiting for url redirect...")
+        self.page.wait.url_change('travel-groups', exclude=True, timeout=45)
+        time.sleep(2)
 
-        # if "travel-groups" in self.page.url or "auth" in self.page.url:
-        #     raise BizLogicError(message="Redirect to service-level Failed!")
+        if "travel-groups" in self.page.url or "auth" in self.page.url:
+            raise BizLogicError(message="Redirect to service-level Failed!")
         
         btn_selector_add = 'tag:button@@data-testid=btn-add-applicant'
         btn_selector_max = 'tag:button@@data-testid=btn-max-number-of-applicants'
@@ -730,9 +730,9 @@ def register_worker(proxy_config, tls_url, capsolver_key):
 def main():
     # ================= 命令行参数解析 =================
     parser = argparse.ArgumentParser(description="TLS 批量注册机")
-    parser.add_argument("-n", "--concurrency", type=int, default=5, help="最大并发数 (N)")
-    parser.add_argument("-m", "--target", type=int, default=10, help="最大成功注册数 (M)")
-    parser.add_argument("-p", "--pool", type=str, default="isp_all", help="代理池名称")
+    parser.add_argument("-n", "--concurrency", type=int, default=1, help="最大并发数 (N)")
+    parser.add_argument("-m", "--target", type=int, default=1, help="最大成功注册数 (M)")
+    parser.add_argument("-p", "--pool", type=str, default="local", help="代理池名称")
     parser.add_argument("-u", "--url", type=str, default="https://visas-fr.tlscontact.com/en-us/country/gb/vac/gbLON2fr", help="TLS 目标网址")
     args = parser.parse_args()
 

toolkit/vs_cloud_api.py

@@ -7,6 +7,7 @@ from datetime import datetime
 from typing import Dict, Any, List, Optional
 from vs_types import NotFoundError, PermissionDeniedError, RateLimiteddError, SessionExpiredOrInvalidError, BizLogicError 
 from vs_log_macros import VSC_ERROR, VSC_INFO, VSC_WARN, VSC_DEBUG
+from tools.clash_api import switch_next_node
 
 class VSCloudApi:
     """
@@ -330,6 +331,25 @@ class VSCloudApi:
         else:
             raise BizLogicError(message=f"Get next account biz error: {result.get('message')}")
         
+    # def get_next_proxy(
+    #     self,
+    #     pools: List[str],
+    #     proxy_cd: int = 60
+    # ):
+    #     local = {
+    #         "pool_name": "local",
+    #         "proto": "http",
+    #         "ip": "127.0.0.1",
+    #         "port": 7890,
+    #         "username": "",
+    #         "password": "",
+    #         "time_zone": "Europe/Dublin",
+    #         "id": 213
+    #     }
+    #     node = switch_next_node()
+    #     VSC_INFO('-', f'proxy node={node}')
+    #     return local
+        
     def get_next_proxy(
         self,
         pools: List[str],

tools/clash_api.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+import requests
+import json
+
+# -------- 配置部分 --------
+API_URL = "http://127.0.0.1:9097"  # Clash 本地 API
+API_KEY = "esZnx8"                 # Clash API 密钥
+GROUP_NAME = "♻️ 手动切换"               # 要轮换的策略组名称
+# ---------------------------
+
+headers = {
+    "Authorization": f"Bearer {API_KEY}",
+    "Content-Type": "application/json"
+}
+
+def get_proxies():
+    """获取策略组和节点信息"""
+    resp = requests.get(f"{API_URL}/proxies", headers=headers, timeout=5)
+    resp.raise_for_status()
+    return resp.json()
+
+def switch_next_node():
+    proxies = get_proxies().get('proxies')
+    
+    if GROUP_NAME not in proxies:
+        print(f"策略组 '{GROUP_NAME}' 不存在")
+        return
+    
+    group = proxies[GROUP_NAME]
+    all_nodes = group['all']
+    current_node = group['now']
+    
+    # 找到下一个节点
+    try:
+        idx = all_nodes.index(current_node)
+        next_idx = (idx + 1) % len(all_nodes)
+        next_node = all_nodes[next_idx]
+    except ValueError:
+        # 当前节点不在列表里，选择第一个
+        next_node = all_nodes[0]
+    
+    # 切换节点（注意 API 路径修改）
+    data = {"name": next_node}
+    resp = requests.put(f"{API_URL}/proxies/{GROUP_NAME}", headers=headers, json=data, timeout=5)
+    resp.raise_for_status()
+    
+    print(f"[OK] 策略组 '{GROUP_NAME}' 切换完成")
+    print(f"    当前节点: {current_node} → {next_node}")
+    return next_node
+
+if __name__ == "__main__":
+    switch_next_node()

utils/fingerprint_utils.py

@@ -1,20 +1,17 @@
 import hashlib
 from typing import Dict
 
+
 class FingerprintGenerator:
     def __init__(self):
-        # 按现实世界市场份额设计的权重分配（数组长度为100，代表百分比）
-        self.platforms = (
-            ['windows'] * 70 +  # 70% 概率生成 Windows
-            ['macos'] * 25 +    # 25% 概率生成 macOS
-            ['linux'] * 5       # 5%  概率生成 Linux
-        )
-        
+        # platform 只保留 windows
+        self.platforms = ['windows']
+
+        # brand 只保留 Chrome 和 Edge
+        # Chrome 70%，Edge 30%
         self.brands = (
-            ['Chrome'] * 70 +   # 70% 概率生成 Chrome
-            ['Edge'] * 20 +     # 20% 概率生成 Edge
-            ['Opera'] * 5 +     # 5%  概率生成 Opera
-            ['Vivaldi'] * 5     # 5%  概率生成 Vivaldi
+            ['Chrome'] * 70 +
+            ['Edge'] * 30
         )
 
     def generate(self, username: str) -> Dict:
@@ -22,12 +19,16 @@ class FingerprintGenerator:
         根据用户名生成固定的指纹配置
         """
         md5_hash = hashlib.md5(username.encode('utf-8')).hexdigest()
+
         seed = int(md5_hash[:8], 16) & 0x7FFFFFFF
-        platform_index = int(md5_hash[8:12], 16) % len(self.platforms)
-        platform = self.platforms[platform_index]
+
+        # platform 固定 windows
+        platform = 'windows'
+
+        # brand 根据 hash 固定生成
         brand_index = int(md5_hash[12:16], 16) % len(self.brands)
         brand = self.brands[brand_index]
-        
+
         return {
             "username": username,
             "seed": seed,
@@ -40,41 +41,38 @@ class FingerprintGenerator:
         直接生成用于启动浏览器的命令行参数列表
         """
         fp = self.generate(username)
+
         return [
             f"--fingerprint={fp['seed']}",
             f"--fingerprint-platform={fp['platform']}",
             f"--fingerprint-brand={fp['brand']}"
         ]
 
+
 # ================= 测试代码 =================
 if __name__ == "__main__":
     fp_tool = FingerprintGenerator()
 
-    # 测试用户 A
-    user_a = fp_tool.generate("alice_2024")
-    print(f"User: alice_2024 -> Seed: {user_a['seed']}, OS: {user_a['platform']}, Browser: {user_a['brand']}")
-
-    # 测试用户 B
-    user_b = fp_tool.generate("bob_crypto")
-    print(f"User: bob_crypto -> Seed: {user_b['seed']}, OS: {user_b['platform']}, Browser: {user_b['brand']}")
-    
-    # 测试用户 C
-    user_c = fp_tool.generate("jerry_crypto")
-    print(f"User: bob_crypto -> Seed: {user_c['seed']}, OS: {user_c['platform']}, Browser: {user_c['brand']}")
-    
-    # 测试用户 D
-    user_d = fp_tool.generate("luna_crypto")
-    print(f"User: bob_crypto -> Seed: {user_d['seed']}, OS: {user_d['platform']}, Browser: {user_d['brand']}")
-    
-    # 测试用户 E
-    user_e = fp_tool.generate("jone_crypto")
-    print(f"User: bob_crypto -> Seed: {user_e['seed']}, OS: {user_e['platform']}, Browser: {user_e['brand']}")
-    
-    # 测试用户 F
-    user_f = fp_tool.generate("tom_crypto")
-    print(f"User: bob_crypto -> Seed: {user_f['seed']}, OS: {user_f['platform']}, Browser: {user_f['brand']}")
-    
+    users = [
+        "alice_2024",
+        "bob_crypto",
+        "jerry_crypto",
+        "luna_crypto",
+        "jone_crypto",
+        "tom_crypto"
+    ]
+
+    for username in users:
+        fp = fp_tool.generate(username)
+        print(
+            f"User: {username} -> "
+            f"Seed: {fp['seed']}, "
+            f"OS: {fp['platform']}, "
+            f"Browser: {fp['brand']}"
+        )
+
     # 模拟获取命令行参数
     args = fp_tool.get_cli_args("alice_2024")
+
     print("\n生成的命令行参数:")
     print(" ".join(args))

utils/network_interceptor.py

@@ -0,0 +1,223 @@
+# import time
+# import json
+# from DrissionPage import ChromiumPage
+# from network_interceptor import DrissionFetchInterceptor  # 导入刚才封装的工具类
+
+# # 初始化浏览器
+# page = ChromiumPage()
+
+# # 实例化拦截器
+# interceptor = DrissionFetchInterceptor(page)
+
+# # ==========================================
+# # 场景 1：动态修改 Request Body
+# # ==========================================
+# def custom_request_handler(url, original_post_data):
+#     print(f"原始请求体: {original_post_data}")
+#     # 你可以对 original_post_data 解析并修改，或者直接返回全新的数据
+    
+#     # 假设接口需要 JSON，你可以直接返回一个 Python 字典，工具类会自动将其转为 JSON 和 Base64
+#     new_request_data = {
+#         "username": "admin", 
+#         "password": "modified_password_by_tool",
+#         "timestamp": time.time()
+#     }
+#     return new_request_data
+
+# # 告诉拦截器，凡是包含 "api/submit" 的请求，发出去前先经过 custom_request_handler 处理
+# interceptor.intercept_request("api/submit", custom_request_handler)
+
+
+# # ==========================================
+# # 场景 2：动态修改 / 伪造 Response Body
+# # ==========================================
+# def custom_response_handler(url, original_body):
+#     print(f"收到服务器原始响应: {original_body[:100]}...") # 打印前100个字符
+    
+#     # 我们基于原有的响应数据，注入一些自己想要伪造的数据
+#     try:
+#         data = json.loads(original_body)
+#         data['message'] = "这是被DrissionPage拦截并篡改的Mock数据!"
+#         data['vip_status'] = True
+#         return data  # 直接返回字典即可
+#     except json.JSONDecodeError:
+#         # 如果不是JSON，直接返回一个新字符串
+#         return '{"code": 200, "message": "全量替换的Mock数据"}'
+
+# # 告诉拦截器，凡是包含 "api/data" 的响应，接收前先经过 custom_response_handler 处理
+# interceptor.intercept_response("api/data", custom_response_handler)
+
+# # ==========================================
+# # 启动拦截并开始业务测试
+# # ==========================================
+# interceptor.start()
+
+# print("拦截器已生效，开始访问页面...")
+# page.get('https://example.com') # 替换为测试的目标网址
+
+# # 模拟业务等待
+# time.sleep(100)
+
+# # 如果不需要拦截了，可以调用 stop (可选)
+# # interceptor.stop()
+
+
+
+
+
+import base64
+import json
+from typing import Callable, Union, Dict, Any
+
+class DrissionFetchInterceptor:
+    """
+    DrissionPage Fetch 请求/响应拦截修改器
+    基于 Chrome DevTools Protocol (CDP) 的 Fetch 域
+    """
+
+    def __init__(self, page):
+        self.page = page
+        self.patterns = []
+        self._request_handlers = {}
+        self._response_handlers = {}
+        self._is_running = False
+
+    def intercept_request(self, url_keyword: str, handler: Callable):
+        """
+        添加拦截 Request Body 的规则
+        :param url_keyword: URL中包含的关键字 (例如 'api/submit')
+        :param handler: 回调函数，接收 (url, original_post_data)，需返回新的 post_data 字符串或字典
+        """
+        self.patterns.append({
+            'urlPattern': f'*{url_keyword}*',
+            'requestStage': 'Request'
+        })
+        self._request_handlers[url_keyword] = handler
+
+    def intercept_response(self, url_keyword: str, handler: Callable):
+        """
+        添加拦截 Response Body 的规则
+        :param url_keyword: URL中包含的关键字 (例如 'api/data')
+        :param handler: 回调函数，接收 (url, original_body)，需返回新的 body 字符串或字典
+        """
+        self.patterns.append({
+            'urlPattern': f'*{url_keyword}*',
+            'requestStage': 'Response'
+        })
+        self._response_handlers[url_keyword] = handler
+
+    def start(self):
+        """启动拦截器"""
+        if not self.patterns:
+            print("[Interceptor] 没有配置任何拦截规则。")
+            return
+
+        # 开启 Fetch 域并应用规则
+        self.page.run_cdp('Fetch.enable', patterns=self.patterns)
+        # 绑定核心回调
+        self.page.driver.set_callback('Fetch.requestPaused', self._on_request_paused)
+        self._is_running = True
+        print(f"[Interceptor] 已启动，共 {len(self.patterns)} 条规则生效。")
+
+    def stop(self):
+        """停止拦截器"""
+        if self._is_running:
+            self.page.run_cdp('Fetch.disable')
+            self.page.driver.set_callback('Fetch.requestPaused', None)
+            self._is_running = False
+            print("[Interceptor] 已停止。")
+
+    def _on_request_paused(self, **kwargs):
+        """底层的 CDP 暂停事件路由器"""
+        request_id = kwargs.get('requestId')
+        request = kwargs.get('request', {})
+        url = request.get('url')
+        response_status = kwargs.get('responseStatusCode')
+
+        # === 阶段 1: 拦截并修改 Response (服务器返回后) ===
+        if response_status:
+            for keyword, handler in self._response_handlers.items():
+                if keyword in url:
+                    self._handle_response_modification(kwargs, handler)
+                    return
+            
+            # 没匹配上，放行
+            self.page.run_cdp('Fetch.continueRequest', requestId=request_id)
+            return
+
+        # === 阶段 2: 拦截并修改 Request (发往服务器前) ===
+        for keyword, handler in self._request_handlers.items():
+            if keyword in url:
+                self._handle_request_modification(kwargs, handler)
+                return
+        
+        # 没匹配上，放行
+        self.page.run_cdp('Fetch.continueRequest', requestId=request_id)
+
+    def _handle_request_modification(self, kwargs: Dict, handler: Callable):
+        """处理 Request 的修改逻辑"""
+        request_id = kwargs.get('requestId')
+        request = kwargs.get('request', {})
+        url = request.get('url')
+        original_post_data = request.get('postData', '')
+
+        try:
+            # 调用用户自定义的处理函数
+            new_data = handler(url, original_post_data)
+            
+            # 如果返回的是字典，自动转成 JSON 字符串
+            if isinstance(new_data, dict):
+                new_data = json.dumps(new_data)
+
+            if new_data is not None:
+                encoded_body = base64.b64encode(str(new_data).encode('utf-8')).decode('utf-8')
+                self.page.run_cdp('Fetch.continueRequest', requestId=request_id, postData=encoded_body)
+                print(f"[Interceptor] 成功修改 Request Body -> {url}")
+                return
+        except Exception as e:
+            print(f"[Interceptor] 修改 Request 时发生错误: {e}")
+        
+        # 如果发生异常或未返回新数据，兜底原样放行，防止浏览器卡死
+        self.page.run_cdp('Fetch.continueRequest', requestId=request_id)
+
+    def _handle_response_modification(self, kwargs: Dict, handler: Callable):
+        """处理 Response 的修改逻辑"""
+        request_id = kwargs.get('requestId')
+        url = kwargs.get('request', {}).get('url')
+        response_status = kwargs.get('responseStatusCode', 200)
+        # CDP 中 headers 的格式是一个字典列表 [{'name': 'Content-Type', 'value': '...'}, ...]
+        headers = kwargs.get('responseHeaders', [])
+
+        original_body = ""
+        try:
+            # 尝试获取原始 Response Body
+            res = self.page.run_cdp('Fetch.getResponseBody', requestId=request_id)
+            original_body = res.get('body', '')
+            if res.get('base64Encoded'):
+                original_body = base64.b64decode(original_body).decode('utf-8')
+        except Exception:
+            pass # 有些请求（如图片/流）可能获取不到body
+
+        try:
+            # 调用用户自定义的处理函数
+            new_data = handler(url, original_body)
+
+            # 如果返回的是字典，自动转成 JSON 字符串
+            if isinstance(new_data, dict):
+                new_data = json.dumps(new_data)
+
+            if new_data is not None:
+                encoded_body = base64.b64encode(str(new_data).encode('utf-8')).decode('utf-8')
+                # 修改响应体必须使用 Fetch.fulfillRequest
+                self.page.run_cdp('Fetch.fulfillRequest', 
+                                  requestId=request_id, 
+                                  responseCode=response_status,
+                                  responseHeaders=headers,
+                                  body=encoded_body)
+                print(f"[Interceptor] 成功修改 Response Body -> {url}")
+                return
+        except Exception as e:
+            print(f"[Interceptor] 修改 Response 时发生错误: {e}")
+
+        # 兜底：如果报错，原样放行请求
+        self.page.run_cdp('Fetch.continueRequest', requestId=request_id)