| 1 | import copy |
| 2 | import logging |
| 3 | import os |
| 4 | from pathlib import Path |
| 5 | from typing import Any, Dict, Optional |
| 6 | |
| 7 | import yaml |
| 8 | |
| 9 | logger = logging.getLogger(__name__) |
| 10 | |
| 11 | BASE_DIR = Path(__file__).resolve().parent |
| 12 | CONFIG_PATH = BASE_DIR / "config.yaml" |
| 13 | CONFIG_EXAMPLE_PATH = BASE_DIR / "config.yaml.example" |
| 14 | |
| 15 | DEFAULT_CONFIG: Dict[str, Any] = { |
| 16 | "project_name": "Video-Claw", |
| 17 | "server": { |
| 18 | "host": "127.0.0.1", |
| 19 | "port": 8000, |
| 20 | "log_level": "INFO", |
| 21 | "access_log": False, |
| 22 | }, |
| 23 | "api_providers": { |
| 24 | "common": { |
| 25 | "print_model_input": False, |
| 26 | "proxy": "", |
| 27 | }, |
| 28 | "openai": { |
| 29 | "api_key": "", |
| 30 | "base_url": "https://api.openai.com/v1", |
| 31 | "enable_proxy": False, |
| 32 | }, |
| 33 | "gemini": { |
| 34 | "api_key": "", |
| 35 | "base_url": "https://generativelanguage.googleapis.com/v1beta", |
| 36 | "enable_proxy": False, |
| 37 | }, |
| 38 | "deepseek": { |
| 39 | "api_key": "", |
| 40 | "base_url": "https://api.deepseek.com/v1", |
| 41 | "enable_proxy": False, |
| 42 | }, |
| 43 | "dashscope": { |
| 44 | "api_key": "", |
| 45 | "base_url": "https://dashscope.aliyuncs.com/api/v1", |
| 46 | "enable_proxy": False, |
| 47 | }, |
| 48 | "ark": { |
| 49 | "api_key": "", |
| 50 | "base_url": "https://ark.cn-beijing.volces.com/api/v3", |
| 51 | "enable_proxy": False, |
| 52 | }, |
| 53 | "kling": { |
| 54 | "base_url": "https://api-beijing.klingai.com", |
| 55 | "api_key": "", |
| 56 | "enable_proxy": False, |
| 57 | }, |
| 58 | }, |
| 59 | "models": { |
| 60 | "llm": "qwen3.5-plus", |
| 61 | "vlm": "qwen3.5-plus", |
| 62 | "image_it2i": "doubao-seedream-5-0-260128", |
| 63 | "image_t2i": "doubao-seedream-5-0-260128", |
| 64 | "video": "wan2.7-i2v", |
| 65 | "video_first_frame": "wan2.7-i2v", |
| 66 | "video_start_end": "wan2.7-i2v", |
| 67 | "video_reference": "wan2.7-r2v", |
| 68 | }, |
| 69 | "generation": { |
| 70 | "style": "realistic", |
| 71 | "video_ratio": "16:9", |
| 72 | "video_resolution": "720P", |
| 73 | "video_generation_mode": "first_frame", |
| 74 | }, |
| 75 | } |
| 76 | |
| 77 | |
| 78 | def _deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]: |
| 79 | merged = copy.deepcopy(base) |
| 80 | for key, value in (override or {}).items(): |
| 81 | if isinstance(value, dict) and isinstance(merged.get(key), dict): |
| 82 | merged[key] = _deep_merge(merged[key], value) |
| 83 | else: |
| 84 | merged[key] = value |
| 85 | return merged |
| 86 | |
| 87 | |
| 88 | def _get(data: Dict[str, Any], path: str, default: Any = None) -> Any: |
| 89 | current: Any = data |
| 90 | for part in path.split("."): |
| 91 | if not isinstance(current, dict) or part not in current: |
| 92 | return default |
| 93 | current = current[part] |
| 94 | return current |
| 95 | |
| 96 | |
| 97 | def _coerce_config(data: Dict[str, Any]) -> Dict[str, Any]: |
| 98 | clean = _deep_merge(DEFAULT_CONFIG, data) |
| 99 | clean.pop("llm", None) |
| 100 | raw_server = data.get("server", {}) if isinstance(data, dict) else {} |
| 101 | |
| 102 | legacy_models = data.get("models", {}) if isinstance(data, dict) else {} |
| 103 | if isinstance(legacy_models, dict): |
| 104 | for legacy_key in ("style", "video_ratio", "video_resolution"): |
| 105 | # Legacy config compatibility: older config.yaml stored generation settings under models.*. |
| 106 | if legacy_key in legacy_models and not _get(data, f"generation.{legacy_key}"): |
| 107 | clean.setdefault("generation", {})[legacy_key] = legacy_models[legacy_key] |
| 108 | clean["models"].pop(legacy_key, None) |
| 109 | if legacy_models.get("video") and not any( |
| 110 | legacy_models.get(key) for key in ("video_first_frame", "video_start_end", "video_reference") |
| 111 | ): |
| 112 | # Legacy config compatibility: older configs had one models.video instead of mode-specific video models. |
| 113 | clean["models"]["video_first_frame"] = legacy_models["video"] |
| 114 | # Legacy config compatibility: models.eval was never used by runtime agents; keep it out after load. |
| 115 | clean["models"].pop("eval", None) |
| 116 | |
| 117 | server = clean["server"] |
| 118 | server["host"] = str(server.get("host") or DEFAULT_CONFIG["server"]["host"]) |
| 119 | try: |
| 120 | server["port"] = int(server.get("port")) |
| 121 | except (TypeError, ValueError): |
| 122 | server["port"] = DEFAULT_CONFIG["server"]["port"] |
| 123 | server["log_level"] = _normalize_log_level( |
| 124 | server.get("log_level") if isinstance(raw_server, dict) and "log_level" in raw_server else None, |
| 125 | server.get("debug"), |
| 126 | ) |
| 127 | server.pop("debug", None) |
| 128 | server["access_log"] = _as_bool(server.get("access_log")) |
| 129 | server.pop("admin_password", None) |
| 130 | |
| 131 | common = clean["api_providers"]["common"] |
| 132 | for key in ("local_proxy", "http_proxy", "https_proxy"): |
| 133 | common.pop(key, None) |
| 134 | common["print_model_input"] = _as_bool(common.get("print_model_input")) |
| 135 | common["proxy"] = str(common.get("proxy") or "") |
| 136 | |
| 137 | if isinstance(clean["models"].get("llm"), dict): |
| 138 | clean["models"]["llm"] = clean["models"]["llm"].get("model") or DEFAULT_CONFIG["models"]["llm"] |
| 139 | |
| 140 | for key, value in clean["models"].items(): |
| 141 | if isinstance(value, dict): |
| 142 | for sub_key, sub_value in value.items(): |
| 143 | value[sub_key] = "" if sub_value is None else str(sub_value) |
| 144 | else: |
| 145 | clean["models"][key] = "" if value is None else str(value) |
| 146 | |
| 147 | for key, value in clean["generation"].items(): |
| 148 | clean["generation"][key] = "" if value is None else str(value) |
| 149 | |
| 150 | for provider, values in clean["api_providers"].items(): |
| 151 | if provider == "common": |
| 152 | continue |
| 153 | for key, value in values.items(): |
| 154 | if key == "enable_proxy": |
| 155 | values[key] = _as_bool(value) |
| 156 | else: |
| 157 | values[key] = "" if value is None else str(value) |
| 158 | |
| 159 | return clean |
| 160 | |
| 161 | |
| 162 | def _as_bool(value: Any) -> bool: |
| 163 | if isinstance(value, bool): |
| 164 | return value |
| 165 | return str(value).strip().lower() in {"1", "true", "yes", "on"} |
| 166 | |
| 167 | |
| 168 | def _normalize_log_level(value: Any, legacy_debug: Any = None) -> str: |
| 169 | allowed = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"} |
| 170 | if legacy_debug is not None and (value is None or str(value).strip() == ""): |
| 171 | return "DEBUG" if _as_bool(legacy_debug) else "INFO" |
| 172 | normalized = str(value or DEFAULT_CONFIG["server"]["log_level"]).strip().upper() |
| 173 | return normalized if normalized in allowed else DEFAULT_CONFIG["server"]["log_level"] |
| 174 | |
| 175 | |
| 176 | def load_config() -> Dict[str, Any]: |
| 177 | if not CONFIG_PATH.exists(): |
| 178 | source = CONFIG_EXAMPLE_PATH if CONFIG_EXAMPLE_PATH.exists() else None |
| 179 | if source: |
| 180 | with source.open("r", encoding="utf-8") as f: |
| 181 | loaded = yaml.safe_load(f) or {} |
| 182 | return _coerce_config(loaded) |
| 183 | return copy.deepcopy(DEFAULT_CONFIG) |
| 184 | |
| 185 | with CONFIG_PATH.open("r", encoding="utf-8") as f: |
| 186 | loaded = yaml.safe_load(f) or {} |
| 187 | if not isinstance(loaded, dict): |
| 188 | raise ValueError("backend/config.yaml must contain a YAML mapping.") |
| 189 | return _coerce_config(loaded) |
| 190 | |
| 191 | |
| 192 | def save_config(values: Dict[str, Any]) -> Dict[str, Any]: |
| 193 | clean = _coerce_config(values) |
| 194 | with CONFIG_PATH.open("w", encoding="utf-8") as f: |
| 195 | yaml.safe_dump(clean, f, allow_unicode=True, sort_keys=False) |
| 196 | return clean |
| 197 | |
| 198 | |
| 199 | CONFIG_VALUES = load_config() |
| 200 | |
| 201 | |
| 202 | class Config: |
| 203 | CONFIG = CONFIG_VALUES |
| 204 | |
| 205 | HOST = _get(CONFIG, "server.host") |
| 206 | PORT = _get(CONFIG, "server.port") |
| 207 | LOG_LEVEL = _get(CONFIG, "server.log_level") |
| 208 | DEBUG = LOG_LEVEL == "DEBUG" |
| 209 | ACCESS_LOG = _get(CONFIG, "server.access_log") |
| 210 | |
| 211 | PRINT_MODEL_INPUT = _get(CONFIG, "api_providers.common.print_model_input") |
| 212 | PROXY = _get(CONFIG, "api_providers.common.proxy") |
| 213 | |
| 214 | OPENAI_API_KEY = _get(CONFIG, "api_providers.openai.api_key") |
| 215 | OPENAI_BASE_URL = _get(CONFIG, "api_providers.openai.base_url") |
| 216 | OPENAI_ENABLE_PROXY = _get(CONFIG, "api_providers.openai.enable_proxy") |
| 217 | GEMINI_API_KEY = _get(CONFIG, "api_providers.gemini.api_key") |
| 218 | GOOGLE_GEMINI_BASE_URL = _get(CONFIG, "api_providers.gemini.base_url") |
| 219 | GEMINI_ENABLE_PROXY = _get(CONFIG, "api_providers.gemini.enable_proxy") |
| 220 | DEEPSEEK_API_KEY = _get(CONFIG, "api_providers.deepseek.api_key") |
| 221 | DEEPSEEK_BASE_URL = _get(CONFIG, "api_providers.deepseek.base_url") |
| 222 | DEEPSEEK_ENABLE_PROXY = _get(CONFIG, "api_providers.deepseek.enable_proxy") |
| 223 | DASHSCOPE_API_KEY = _get(CONFIG, "api_providers.dashscope.api_key") |
| 224 | DASHSCOPE_BASE_URL = _get(CONFIG, "api_providers.dashscope.base_url") |
| 225 | DASHSCOPE_ENABLE_PROXY = _get(CONFIG, "api_providers.dashscope.enable_proxy") |
| 226 | ARK_API_KEY = _get(CONFIG, "api_providers.ark.api_key") |
| 227 | ARK_BASE_URL = _get(CONFIG, "api_providers.ark.base_url") |
| 228 | ARK_ENABLE_PROXY = _get(CONFIG, "api_providers.ark.enable_proxy") |
| 229 | KLING_API_KEY = _get(CONFIG, "api_providers.kling.api_key") |
| 230 | KLING_BASE_URL = _get(CONFIG, "api_providers.kling.base_url") |
| 231 | KLING_ENABLE_PROXY = _get(CONFIG, "api_providers.kling.enable_proxy") |
| 232 | |
| 233 | LLM_API_KEY = DASHSCOPE_API_KEY |
| 234 | LLM_BASE_URL = "" |
| 235 | LLM_MODEL = _get(CONFIG, "models.llm") |
| 236 | VLM_MODEL = _get(CONFIG, "models.vlm") |
| 237 | IMAGE_IT2I_MODEL = _get(CONFIG, "models.image_it2i") |
| 238 | IMAGE_T2I_MODEL = _get(CONFIG, "models.image_t2i") |
| 239 | VIDEO_MODEL = _get(CONFIG, "models.video") |
| 240 | VIDEO_FIRST_FRAME_MODEL = _get(CONFIG, "models.video_first_frame") |
| 241 | VIDEO_START_END_MODEL = _get(CONFIG, "models.video_start_end") |
| 242 | VIDEO_REFERENCE_MODEL = _get(CONFIG, "models.video_reference") |
| 243 | VIDEO_RATIO = _get(CONFIG, "generation.video_ratio") |
| 244 | VIDEO_RESOLUTION = _get(CONFIG, "generation.video_resolution") |
| 245 | VIDEO_GENERATION_MODE = _get(CONFIG, "generation.video_generation_mode") |
| 246 | STYLE = _get(CONFIG, "generation.style") |
| 247 | |
| 248 | BASE_DIR = str(BASE_DIR) |
| 249 | CODE_DIR = os.path.join(BASE_DIR, "code") |
| 250 | RESULT_DIR = os.path.join(CODE_DIR, "result") |
| 251 | TEMP_DIR = os.path.join(BASE_DIR, "temp") |
| 252 | SESSION_DIR = os.path.join(CODE_DIR, "data", "sessions") |
| 253 | TASK_DIR = os.path.join(CODE_DIR, "data", "tasks") |
| 254 | TASK_RESULT_DIR = os.path.join(RESULT_DIR, "task") |
| 255 | |
| 256 | @classmethod |
| 257 | def as_dict(cls) -> Dict[str, Any]: |
| 258 | return copy.deepcopy(cls.CONFIG) |
| 259 | |
| 260 | @classmethod |
| 261 | def provider_proxy(cls, provider: str) -> str: |
| 262 | provider_config = _get(cls.CONFIG, f"api_providers.{provider}", {}) |
| 263 | if not isinstance(provider_config, dict) or not _as_bool(provider_config.get("enable_proxy")): |
| 264 | return "" |
| 265 | return cls.PROXY or "" |
| 266 | |
| 267 | @classmethod |
| 268 | def requests_proxies(cls, provider: str) -> Optional[Dict[str, str]]: |
| 269 | proxy = cls.provider_proxy(provider) |
| 270 | if not proxy: |
| 271 | return None |
| 272 | return {"http": proxy, "https": proxy} |
| 273 | |
| 274 | @classmethod |
| 275 | def update_config(cls, values: Dict[str, Any]) -> Dict[str, Any]: |
| 276 | clean = save_config(values) |
| 277 | cls.CONFIG = clean |
| 278 | |
| 279 | cls.HOST = _get(clean, "server.host") |
| 280 | cls.PORT = _get(clean, "server.port") |
| 281 | cls.LOG_LEVEL = _get(clean, "server.log_level") |
| 282 | cls.DEBUG = cls.LOG_LEVEL == "DEBUG" |
| 283 | cls.ACCESS_LOG = _get(clean, "server.access_log") |
| 284 | |
| 285 | cls.PRINT_MODEL_INPUT = _get(clean, "api_providers.common.print_model_input") |
| 286 | cls.PROXY = _get(clean, "api_providers.common.proxy") |
| 287 | |
| 288 | cls.OPENAI_API_KEY = _get(clean, "api_providers.openai.api_key") |
| 289 | cls.OPENAI_BASE_URL = _get(clean, "api_providers.openai.base_url") |
| 290 | cls.OPENAI_ENABLE_PROXY = _get(clean, "api_providers.openai.enable_proxy") |
| 291 | cls.GEMINI_API_KEY = _get(clean, "api_providers.gemini.api_key") |
| 292 | cls.GOOGLE_GEMINI_BASE_URL = _get(clean, "api_providers.gemini.base_url") |
| 293 | cls.GEMINI_ENABLE_PROXY = _get(clean, "api_providers.gemini.enable_proxy") |
| 294 | cls.DEEPSEEK_API_KEY = _get(clean, "api_providers.deepseek.api_key") |
| 295 | cls.DEEPSEEK_BASE_URL = _get(clean, "api_providers.deepseek.base_url") |
| 296 | cls.DEEPSEEK_ENABLE_PROXY = _get(clean, "api_providers.deepseek.enable_proxy") |
| 297 | cls.DASHSCOPE_API_KEY = _get(clean, "api_providers.dashscope.api_key") |
| 298 | cls.DASHSCOPE_BASE_URL = _get(clean, "api_providers.dashscope.base_url") |
| 299 | cls.DASHSCOPE_ENABLE_PROXY = _get(clean, "api_providers.dashscope.enable_proxy") |
| 300 | cls.ARK_API_KEY = _get(clean, "api_providers.ark.api_key") |
| 301 | cls.ARK_BASE_URL = _get(clean, "api_providers.ark.base_url") |
| 302 | cls.ARK_ENABLE_PROXY = _get(clean, "api_providers.ark.enable_proxy") |
| 303 | cls.KLING_API_KEY = _get(clean, "api_providers.kling.api_key") |
| 304 | cls.KLING_BASE_URL = _get(clean, "api_providers.kling.base_url") |
| 305 | cls.KLING_ENABLE_PROXY = _get(clean, "api_providers.kling.enable_proxy") |
| 306 | |
| 307 | cls.LLM_API_KEY = cls.DASHSCOPE_API_KEY |
| 308 | cls.LLM_BASE_URL = "" |
| 309 | cls.LLM_MODEL = _get(clean, "models.llm") |
| 310 | cls.VLM_MODEL = _get(clean, "models.vlm") |
| 311 | cls.IMAGE_IT2I_MODEL = _get(clean, "models.image_it2i") |
| 312 | cls.IMAGE_T2I_MODEL = _get(clean, "models.image_t2i") |
| 313 | cls.VIDEO_MODEL = _get(clean, "models.video") |
| 314 | cls.VIDEO_FIRST_FRAME_MODEL = _get(clean, "models.video_first_frame") |
| 315 | cls.VIDEO_START_END_MODEL = _get(clean, "models.video_start_end") |
| 316 | cls.VIDEO_REFERENCE_MODEL = _get(clean, "models.video_reference") |
| 317 | cls.VIDEO_RATIO = _get(clean, "generation.video_ratio") |
| 318 | cls.VIDEO_RESOLUTION = _get(clean, "generation.video_resolution") |
| 319 | cls.VIDEO_GENERATION_MODE = _get(clean, "generation.video_generation_mode") |
| 320 | cls.STYLE = _get(clean, "generation.style") |
| 321 | return cls.as_dict() |
| 322 | |
| 323 | @classmethod |
| 324 | def check_dirs(cls): |
| 325 | data_dir = os.path.join(cls.CODE_DIR, "data") |
| 326 | for directory in [ |
| 327 | cls.CODE_DIR, |
| 328 | data_dir, |
| 329 | cls.SESSION_DIR, |
| 330 | cls.TASK_DIR, |
| 331 | cls.RESULT_DIR, |
| 332 | cls.TASK_RESULT_DIR, |
| 333 | cls.TEMP_DIR, |
| 334 | ]: |
| 335 | if not os.path.exists(directory): |
| 336 | os.makedirs(directory, exist_ok=True) |
| 337 | logger.info("Created directory: %s", directory) |
| 338 | |
| 339 | |
| 340 | Config.check_dirs() |
| 341 | settings = Config() |
| 342 |