193 lines
6.1 KiB
Python
193 lines
6.1 KiB
Python
"""
|
|
Scene Configuration Loader
|
|
场景配置加载器 - 支持平台特定配置覆盖
|
|
|
|
功能:
|
|
1. 从统一配置文件 scene_configs.json 加载配置
|
|
2. 合并默认配置和平台特定覆盖配置
|
|
3. 支持深度合并策略 (平台配置覆盖默认配置的同名场景)
|
|
4. 支持场景移除 (通过 null 值标记)
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Dict, Any, Optional
|
|
from copy import deepcopy
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 配置文件路径
|
|
CONFIG_FILE = Path(__file__).parent / "scene_configs.json"
|
|
|
|
|
|
def _expand_env_vars(value):
|
|
"""递归展开 ${ENV_VAR_NAME} 格式的环境变量引用"""
|
|
if isinstance(value, str):
|
|
def _replacer(match):
|
|
env_value = os.environ.get(match.group(1))
|
|
if env_value is None:
|
|
return match.group(0)
|
|
return env_value
|
|
return re.sub(r'\$\{([^}]+)\}', _replacer, value)
|
|
elif isinstance(value, dict):
|
|
return {k: _expand_env_vars(v) for k, v in value.items()}
|
|
elif isinstance(value, list):
|
|
return [_expand_env_vars(v) for v in value]
|
|
return value
|
|
|
|
|
|
def deep_merge(base: Dict, override: Dict) -> Dict:
|
|
"""
|
|
深度合并两个字典,override 中的值会覆盖 base 中的同名键
|
|
|
|
Args:
|
|
base: 基础配置字典
|
|
override: 覆盖配置字典
|
|
|
|
Returns:
|
|
合并后的字典
|
|
|
|
特殊处理:
|
|
- 如果 override 中的值为 None,则从结果中移除该键
|
|
- 对于嵌套字典,递归进行深度合并
|
|
- 对于列表,直接覆盖(不合并列表内容)
|
|
"""
|
|
result = deepcopy(base)
|
|
|
|
for key, value in override.items():
|
|
# 特殊处理: None 值表示移除该场景
|
|
if value is None:
|
|
result.pop(key, None)
|
|
continue
|
|
|
|
# 如果两者都是字典,递归合并
|
|
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
|
|
result[key] = deep_merge(result[key], value)
|
|
else:
|
|
# 否则直接覆盖
|
|
result[key] = deepcopy(value)
|
|
|
|
return result
|
|
|
|
|
|
def load_scene_config(platform: str = "android") -> Dict[str, Any]:
|
|
"""
|
|
加载场景配置,自动合并默认配置和平台覆盖配置
|
|
|
|
Args:
|
|
platform: 平台名称 (android/ios/web/windows)
|
|
|
|
Returns:
|
|
合并后的配置字典,包含 keywords 和 instructions
|
|
|
|
Raises:
|
|
FileNotFoundError: 配置文件不存在
|
|
json.JSONDecodeError: 配置文件格式错误
|
|
KeyError: 缺少必要的配置项
|
|
|
|
Example:
|
|
>>> config = load_scene_config("android")
|
|
>>> print(config["keywords"]["login"])
|
|
['登录', '登陆', 'login', ...]
|
|
"""
|
|
# 1. 检查配置文件是否存在
|
|
if not CONFIG_FILE.exists():
|
|
raise FileNotFoundError(f"配置文件不存在: {CONFIG_FILE}")
|
|
|
|
# 2. 加载统一配置文件
|
|
try:
|
|
with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
|
|
all_configs = json.load(f)
|
|
except json.JSONDecodeError as e:
|
|
logger.error(f"配置文件格式错误: {e}")
|
|
raise
|
|
|
|
# 3. 验证配置文件结构
|
|
if "default" not in all_configs:
|
|
raise KeyError("配置文件缺少 'default' 节点")
|
|
|
|
# 4. 获取默认配置
|
|
default_config = all_configs["default"]
|
|
|
|
# 验证默认配置完整性
|
|
if "keywords" not in default_config or "instructions" not in default_config:
|
|
raise KeyError("默认配置缺少 'keywords' 或 'instructions' 字段")
|
|
|
|
# 5. 获取平台覆盖配置 (如果存在)
|
|
platform_config = all_configs.get(platform, {})
|
|
|
|
# 6. 深度合并配置 (平台配置优先)
|
|
merged_config = {
|
|
"keywords": deep_merge(
|
|
default_config.get("keywords", {}),
|
|
platform_config.get("keywords", {})
|
|
),
|
|
"instructions": deep_merge(
|
|
default_config.get("instructions", {}),
|
|
platform_config.get("instructions", {})
|
|
),
|
|
"step_limits": deep_merge(
|
|
default_config.get("step_limits", {}),
|
|
platform_config.get("step_limits", {})
|
|
)
|
|
}
|
|
|
|
# 7. 展开环境变量引用 (格式: ${VAR_NAME})
|
|
merged_config = _expand_env_vars(merged_config)
|
|
|
|
# 8. 记录日志
|
|
scenes = list(merged_config["keywords"].keys())
|
|
if platform_config.get("keywords") or platform_config.get("instructions") or platform_config.get("step_limits"):
|
|
overridden_scenes = (
|
|
set(platform_config.get("keywords", {}).keys())
|
|
| set(platform_config.get("instructions", {}).keys())
|
|
| set(platform_config.get("step_limits", {}).keys())
|
|
)
|
|
logger.info(f"[SceneConfig] 平台 '{platform}' 加载配置成功, 场景数: {len(scenes)}, 覆盖场景: {overridden_scenes}")
|
|
else:
|
|
logger.info(f"[SceneConfig] 平台 '{platform}' 使用默认配置, 场景数: {len(scenes)}")
|
|
|
|
return merged_config
|
|
|
|
|
|
def get_available_scenes(platform: str = "android") -> list:
|
|
"""
|
|
获取指定平台可用的场景列表
|
|
|
|
Args:
|
|
platform: 平台名称
|
|
|
|
Returns:
|
|
场景名称列表
|
|
"""
|
|
config = load_scene_config(platform)
|
|
return list(config["keywords"].keys())
|
|
|
|
|
|
# 测试代码 (仅在直接运行此模块时执行)
|
|
if __name__ == "__main__":
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
print("=" * 60)
|
|
print("场景配置加载器测试")
|
|
print("=" * 60)
|
|
|
|
# 测试默认配置加载
|
|
print("\n[测试 1] 加载 Android 平台默认配置:")
|
|
android_config = load_scene_config("android")
|
|
print(f" 可用场景: {list(android_config['keywords'].keys())}")
|
|
print(f" login 关键词数量: {len(android_config['keywords']['login'])}")
|
|
|
|
# 测试其他平台
|
|
for platform in ["ios", "web", "windows"]:
|
|
print(f"\n[测试 2] 加载 {platform} 平台配置:")
|
|
config = load_scene_config(platform)
|
|
print(f" 可用场景: {list(config['keywords'].keys())}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("测试完成!")
|
|
print("=" * 60)
|