qwen-image-skill/scripts/shared/config.py

from __future__ import annotations

import json
from copy import deepcopy
from pathlib import Path
from typing import Any


BASE_DIR = Path(__file__).resolve().parents[2]
CONFIG_PATH = BASE_DIR / "config.json"
EXAMPLE_CONFIG_PATH = BASE_DIR / "config.example.json"
DEFAULT_CONFIG: dict[str, Any] = {
    "provider": {
        "name": "qwen-compatible",
        "base_url": "https://api-inference.modelscope.cn/v1",
        "model": "qwen-image",
    },
    "generation": {
        "output_dir": "./outputs",
        "timeout_seconds": 300,
        "poll_interval_seconds": 3,
        "default_size": "1024x1024",
        "default_n": 1,
        "default_response_format": "b64_json",
        "default_quality": "standard",
    },
}


def load_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as handle:
        return json.load(handle)


def _merge_defaults(defaults: dict[str, Any], value: dict[str, Any]) -> dict[str, Any]:
    merged = deepcopy(defaults)
    for key, item in value.items():
        if isinstance(item, dict) and isinstance(merged.get(key), dict):
            merged[key] = _merge_defaults(merged[key], item)
        else:
            merged[key] = item
    return merged


def load_config() -> dict[str, Any]:
    if CONFIG_PATH.exists():
        return _merge_defaults(DEFAULT_CONFIG, load_json(CONFIG_PATH))
    return _merge_defaults(DEFAULT_CONFIG, load_json(EXAMPLE_CONFIG_PATH))


def resolve_output_dir(config: dict[str, Any]) -> Path:
    raw_output_dir = config.get("generation", {}).get("output_dir", "./outputs")
    output_dir = Path(raw_output_dir)
    if not output_dir.is_absolute():
        output_dir = BASE_DIR / output_dir
    output_dir.mkdir(parents=True, exist_ok=True)
    return output_dir


def require_provider_config(config: dict[str, Any]) -> dict[str, Any]:
    provider = config.get("provider", {})
    missing = [
        key
        for key in ("api_key",)
        if not str(provider.get(key, "")).strip()
    ]
    if missing:
        raise ValueError(
            "Missing provider config fields: "
            + ", ".join(missing)
            + ". Update config.json before running the skill."
        )
    return provider