import os import json import hashlib import time from typing import Any, Tuple, Optional, Callable from toolkit.vs_cloud_api import VSCloudApi def compute_config_hash(config_data: Any) -> str: """计算 JSON 配置数据的 MD5 哈希,用于快速对比判断配置是否发生变更""" if config_data is None: return "" serialized = json.dumps(config_data, sort_keys=True, ensure_ascii=False) return hashlib.md5(serialized.encode('utf-8')).hexdigest() def fetch_cloud_config(config_key: str) -> Any: """ 使用内置 VSCloudApi 从云端读取指定 config_key 的动态配置。 支持节点专有 key (如 sentinel_config:node01),若无专有配置则自动回退至全局 key (如 sentinel_config)。 """ try: data = VSCloudApi.Instance().get_dynamic_config(config_key) if data is not None: return data except Exception: pass # 若针对节点的特定 Key 未配置或获取失败,自动回退到全局 Key if ":" in config_key: base_key = config_key.split(":")[0] return VSCloudApi.Instance().get_dynamic_config(base_key) return None def save_local_cache(path: str, data: Any) -> bool: """将从云端获取到的最新配置保存到本地缓存文件路径""" try: dir_name = os.path.dirname(path) if dir_name and not os.path.exists(dir_name): os.makedirs(dir_name, exist_ok=True) tmp_path = path + ".tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) os.replace(tmp_path, path) return True except Exception as e: print(f"[CloudConfig] Failed to save local cache to {path}: {e}") return False def load_local_config(path: str) -> Any: """从本地文件读取配置""" if not os.path.exists(path): return None with open(path, "r", encoding="utf-8") as f: return json.load(f) def load_remote_or_cache( config_key: str, local_cache_path: str, use_cloud: bool = True, logger: Optional[Callable[[str], None]] = None ) -> Tuple[Any, str, bool]: """ 核心拉取与 Fallback 逻辑: - 若开启了 use_cloud(默认):尝试调用 VSCloudApi 获取,成功则写入本地缓存文件;失败则 Fallback 读取本地缓存。 - 若未开启 use_cloud(显式传了 -c/--config):直接读取本地配置文件。 """ def log_info(msg: str): if logger: logger(f"[CloudConfig] {msg}") else: print(f"[CloudConfig] {msg}") def log_warn(msg: str): if logger: logger(f"[CloudConfig][WARNING] {msg}") else: print(f"[CloudConfig][WARNING] {msg}") if use_cloud: try: log_info(f"Fetching cloud configuration key '{config_key}' via VSCloudApi...") data = fetch_cloud_config(config_key) if data is not None: config_hash = compute_config_hash(data) log_info(f"Successfully fetched cloud config '{config_key}' (hash: {config_hash[:8]})") save_local_cache(local_cache_path, data) return data, config_hash, True else: log_warn(f"Cloud config key '{config_key}' returned empty/null data.") except Exception as e: log_warn(f"Failed to fetch cloud config '{config_key}' via VSCloudApi: {e}") log_warn(f"Fallback to local cache file: {local_cache_path}") local_data = load_local_config(local_cache_path) if local_data is None: log_warn(f"Local config file {local_cache_path} does not exist or is empty.") local_data = [] if "booker" in local_cache_path or "sentinel" in local_cache_path else {} config_hash = compute_config_hash(local_data) return local_data, config_hash, False