coordinate/booker_order.py

import os
import time
import json
import threading
import random
import redis
from typing import List, Dict, Callable, Any, Optional

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_com = redis.Redis(**redis_conf)
        self.redis_sub = redis.Redis(**redis_conf)
        self.m_pending_order_by_queue: Dict[str, int] = {}        
        self.m_last_spawn_times: Dict[str, float] = {}
        self.m_task_data_cache: Dict[str, dict] = {}
        
        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.task_backoff = ExponentialBackoff(base_delay=10, max_delay=30*60.0, factor=2.0)
        self.heartbeat_ttl = 2*60.0

    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()
        threading.Thread(target=self._cache_refresh_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
            task_id = str(getattr(task, 'task_ref', ''))
            self.m_task_data_cache.pop(task_id, None)
                
        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()
            self.m_task_data_cache.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_com.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_com.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 _cache_refresh_loop(self):
        self._log("Cache refresh loop started.")
        refresh_interval = 15*60
        
        while not self.m_stop_event.is_set():
            for _ in range(refresh_interval):
                if self.m_stop_event.is_set():
                    return
                time.sleep(1.0)
            with self.m_lock:
                task_ids = list(self.m_task_data_cache.keys())
            if not task_ids:
                continue
            for tid in task_ids:
                if self.m_stop_event.is_set():
                    break
                try:
                    fresh_data = VSCloudApi.Instance().get_vas_task(tid)
                    if fresh_data:
                        with self.m_lock:
                            if tid in self.m_task_data_cache:
                                self.m_task_data_cache[tid] = fresh_data
                except Exception:
                    pass
                time.sleep(0.5)

    def _booking_trigger_loop(self):
        self._log("Pub/Sub Trigger loop started.")
        channel_to_routing_key = {}
        for apt in self.m_cfg.appointment_types:
            channel = self._get_redis_key(apt.routing_key)
            channel_to_routing_key[channel] = apt.routing_key
            
        if not channel_to_routing_key:
            self._log("No appointment types configured. Exiting trigger loop.")
            return
            
        pubsub = None
        while not self.m_stop_event.is_set():
            try:
                if pubsub is None:
                    pubsub = self.redis_sub.pubsub(ignore_subscribe_messages=False)
                    channels_to_sub = list(channel_to_routing_key.keys())
                    self._log(f"⏳ Sending SUBSCRIBE command to Redis for: {channels_to_sub}")
                    pubsub.subscribe(*channels_to_sub)
                message = pubsub.get_message(timeout=5.0)
                if not message:
                    continue
                channel = message['channel']
                if isinstance(channel, bytes):
                    channel = channel.decode('utf-8')
                if message['type'] == 'subscribe':
                    active_subs = message['data']
                    self._log(f"📡 [Redis ACK] Successfully subscribed to: {channel} (Active connection subs: {active_subs})")
                    continue
                if message['type'] != 'message':
                    continue
                raw_data = message['data']
                if isinstance(raw_data, bytes):
                    raw_data = raw_data.decode('utf-8')
                routing_key = channel_to_routing_key.get(channel)
                if not routing_key:
                    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 channel {channel}: {parse_err}")
                    continue
                now = time.time()
                matching_tasks = []
                with self.m_lock:
                    for task in self.m_tasks:
                        if now < task.next_run or not task.book_allowed:
                            continue
                        if 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:
                    for task in matching_tasks:
                        self._log(f"🚀 Triggering BOOK for {routing_key} | Order Ref: {task.task_ref}")
                        t = threading.Thread(
                            target=self._execute_book_job, 
                            args=(task, query_result), 
                            daemon=True
                        )
                        t.start()
            except Exception as e:
                self._log(f"Trigger loop pub/sub error: {e}")
                if pubsub:
                    try:
                        pubsub.close()
                    except:
                        pass
                    pubsub = None
                time.sleep(2)
        if pubsub:
            pubsub.close()
            self._log("Pub/Sub connection closed.")

    def _execute_book_job(self, task: Task, query_result: VSQueryResult):
        task_id = task.task_ref
        task_data = None

        try:
            with self.m_lock:
                task_data = self.m_task_data_cache.get(str(task_id))
            
            if not task_data:
                self._log(f"Cache miss for {task_id}, fetching from cloud...")
                task_data = VSCloudApi.Instance().get_vas_task(str(task_id))
                if task_data:
                    with self.m_lock:
                        self.m_task_data_cache[str(task_id)] = task_data

            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_com.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
                }
                
                def _update_cloud_success():
                    try:
                        VSCloudApi.Instance().update_vas_task(str(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)
                    except Exception as e:
                        self._log(f"Failed to update success state to cloud: {e}")
                
                ThreadPool.getInstance().enqueue(_update_cloud_success)

                self.redis_com.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
                    
                    def _update_cloud_meta():
                        try:
                            VSCloudApi.Instance().update_vas_task(str(task_id), {"meta": task_meta})
                        except Exception as cloud_err:
                            self._log(f"Failed to update task meta: {cloud_err}")
                    ThreadPool.getInstance().enqueue(_update_cloud_meta)   
                    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_com.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)
            now = time.time()
            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_com.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:
                    last_spawn = self.m_last_spawn_times.get(r_key, 0.0)
                    if now - last_spawn >= spawn_interval:
                        self.m_last_spawn_times[r_key] = now 
                        self._spawn_worker(r_key)

    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']
                
                with self.m_lock:
                    self.m_task_data_cache[str(task_id)] = task_data
                
                self.redis_com.zadd(self.m_tracker_key, {str(task_id): time.time() + 5*60.0})
                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_com.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_com.incr(queue_fail_key)
                    q_cd = self.queue_backoff.calculate(q_fails)
                    self.redis_com.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(str(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_com.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_com.zadd(self.m_tracker_key, {str(task_id): 0})
                    self._log(f"♻️ Task={task_id} failed normal spawn. Instantly handed over to Sweeper.")
                    
                    with self.m_lock:
                        self.m_task_data_cache.pop(str(task_id), None)
                        
        ThreadPool.getInstance().enqueue(_job)