import os import sys import time import json import argparse from typing import Dict from vs_types import GroupConfig from gco_wrapper import GCOWrapper from logger_setup import setup_app_logger from configure import REDIS_CFG from booker import BuiltinBookerGCO, OrderBookerGCO from utils.cloud_config import load_remote_or_cache, compute_config_hash, fetch_cloud_config, save_local_cache def get_gco_class_for_booker(cfg: GroupConfig): if cfg.booker.account_source == "order": return OrderBookerGCO return BuiltinBookerGCO def sync_booker_groups( groups_conf: list, active_wrappers: Dict[str, GCOWrapper], active_gco_classes: Dict[str, type], app_logger ): new_groups_by_id: Dict[str, GroupConfig] = {} for item in groups_conf: cfg = GroupConfig.from_json(item) new_groups_by_id[cfg.identifier] = cfg current_ids = list(active_wrappers.keys()) for gid in current_ids: wrapper = active_wrappers[gid] new_cfg = new_groups_by_id.get(gid) if not new_cfg or not new_cfg.enable: app_logger.info(f"Stopping disabled/removed Booker group [{gid}]...") try: wrapper.stop() except Exception as e: app_logger.error(f"Error stopping group [{gid}]: {e}") del active_wrappers[gid] if gid in active_gco_classes: del active_gco_classes[gid] for gid, cfg in new_groups_by_id.items(): if not cfg.enable: continue target_gco_class = get_gco_class_for_booker(cfg) if gid in active_wrappers: if active_gco_classes.get(gid) != target_gco_class: app_logger.info(f"Mode changed for Booker group [{gid}]. Restarting group with new class...") try: active_wrappers[gid].stop() except Exception as e: app_logger.error(f"Error stopping group [{gid}] for class switch: {e}") try: wrapper = GCOWrapper( gco_class=target_gco_class, gco_cfg=cfg, redis_conf=REDIS_CFG ) wrapper.load() wrapper.start() active_wrappers[gid] = wrapper active_gco_classes[gid] = target_gco_class except Exception as e: app_logger.error(f"Failed to restart group [{gid}]: {e}") else: app_logger.info(f"Updating configuration for active Booker group [{gid}]...") try: active_wrappers[gid].update_config(cfg) except Exception as e: app_logger.error(f"Error updating config for group [{gid}]: {e}") else: mode_str = "ORDER (Bound)" if target_gco_class == OrderBookerGCO else "BUILT-IN (Unbound)" app_logger.info(f"Starting new wrapper for Booker group [{gid}] (Mode: {mode_str})...") try: wrapper = GCOWrapper( gco_class=target_gco_class, gco_cfg=cfg, redis_conf=REDIS_CFG ) wrapper.load() wrapper.start() active_wrappers[gid] = wrapper active_gco_classes[gid] = target_gco_class except Exception as e: app_logger.error(f"Failed to start group [{gid}]: {e}") def main(): parser = argparse.ArgumentParser(description="Booker Runner") parser.add_argument( "-c", "--config", type=str, required=False, default=None, help="Path to local config.json. If specified, disables cloud config and runs with local file." ) parser.add_argument( "--poll-interval", type=float, required=False, default=30.0, help="Interval in seconds to poll cloud config (default: 30.0s)" ) args = parser.parse_args() # 逻辑 1:默认开启云端配置,除非命令行指定了 -c/--config 本地文件路径 if args.config is not None: config_path = args.config use_cloud = False else: config_path = "config/config_booker.json" use_cloud = True # 逻辑 2:直接从环境变量获取 node_id,默认 node01 node_id = os.getenv("NODE_ID") or "node01" config_key = f"booker_config:{node_id}" poll_interval = args.poll_interval app_logger = setup_app_logger("Booker") app_logger.info(f"Booker Logger is ready! (Node ID: '{node_id}', Cloud Mode: {use_cloud})") groups_conf, current_hash, loaded_from_remote = load_remote_or_cache( config_key=config_key, local_cache_path=config_path, use_cloud=use_cloud, logger=app_logger.info ) active_wrappers: Dict[str, GCOWrapper] = {} active_gco_classes: Dict[str, type] = {} sync_booker_groups(groups_conf, active_wrappers, active_gco_classes, app_logger) app_logger.info( f"Successfully initialized Booker with {len(active_wrappers)} active group(s)." ) if use_cloud: app_logger.info(f"Cloud config active for key '{config_key}'. Polling every {poll_interval}s.") last_poll_time = time.time() try: while True: time.sleep(1) now = time.time() if use_cloud and (now - last_poll_time >= poll_interval): last_poll_time = now try: new_conf = fetch_cloud_config(config_key) if new_conf is not None: new_hash = compute_config_hash(new_conf) if new_hash != current_hash: app_logger.info( f"Cloud config change detected (hash: {current_hash[:8]} -> {new_hash[:8]}). Auto-reloading..." ) save_local_cache(config_path, new_conf) sync_booker_groups(new_conf, active_wrappers, active_gco_classes, app_logger) current_hash = new_hash app_logger.info(f"Auto-reload completed. Currently {len(active_wrappers)} group(s) running.") except Exception as e: app_logger.warning(f"Error polling cloud config key '{config_key}': {e}. Retrying next cycle.") except KeyboardInterrupt: app_logger.info("Shutting down Bookers...") for gid, wrapper in list(active_wrappers.items()): try: wrapper.stop() except Exception as e: app_logger.error(f"Error stopping group [{gid}]: {e}") if __name__ == "__main__": main()