coordinate/main_booker.py

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()