cmbot/src/services/config_service.py

import json
import logging

logger = logging.getLogger(__name__)

DEFAULT_CONFIG = {
    "output_format": "PNG",   # PNG or JPG
    "output_quality": 95,     # JPG quality 1-95
    "output_dir": "",         # empty = use app output/ dir
    "last_garment_dir": "",
    "last_print_dir": "",
    "last_template": "",       # name of the last-selected template
    "last_batch_mode": "full_combo",  # BatchMode value of the last-used mode
    "last_tab": 0,             # last-selected workflow tab (0=添加印花, 1=AI 穿搭)
    "update_source": "",       # HTTP(S) manifest source (empty = no update check)
    "update_user": "",         # HTTP Basic Auth user (empty = anonymous)
    "update_pass": "",         # HTTP Basic Auth password (empty = anonymous)
    # AI 穿搭（docs/11 §11）：上次路径与批量设置并入 app_config（密钥不在此，见 ai_models.json）
    "outfit_excel": "",
    "outfit_output_dir": "",
    "outfit_model": "",            # last-selected model name
    "outfit_concurrency": 1,
    "outfit_request_interval": 2.0,
    "outfit_task_cooldown": 1.0,
    "outfit_retry_count": 2,
    "outfit_resolution": "1K",
    "outfit_quality": "均衡",
    "outfit_retry_failed": False,
    "outfit_prompt_name": "默认",  # last-selected 话术模板 name (docs/11 §7.2)
    "title_model": "GPT-5.5 文本",  # 标题模型名字（对应 ai_models.json 的 name，§17.3 配置定名）
}

_CONFIG_FILENAME = "app_config.json"
_AI_MODELS_FILENAME = "ai_models.json"
_OUTFIT_PROMPT_FILENAME = "outfit_prompt.txt"        # legacy single prompt (migrated)
_OUTFIT_PROMPTS_FILENAME = "outfit_prompts.json"     # multi named templates (§7.2)
_TITLE_PROMPT_FILENAME = "title_prompt.txt"          # single 标题生成提示词 (§17.3)
DEFAULT_OUTFIT_PROMPT_NAME = "默认"

# Default outfit prompt (docs/11 §7). Seeded into outfit_prompts.json on first run.
DEFAULT_OUTFIT_PROMPT = (
    "为商品「{title}」生成人物上身实穿效果图：真人模特正面穿着这件衣服，"
    "完整保留款式、版型、颜色与印花图案，自然光、纯色棚拍背景，"
    "电商主图风格，不加文字与促销标签。"
)

# Default title prompt (docs/11 §17.3, batch + comma-separated). Used when title_prompt.txt absent.
# 数量由用户改写（如「生成 10 条」）；一次请求返回多条、按逗号拆分后按序回填各行 A（§17.1/§19.23）。
DEFAULT_TITLE_PROMPT = (
    "请生成 10 条适合台湾蝦皮电商的中文女装商品标题，突出卖点与适穿场景，"
    "每条控制在 30 字以内。"
    "如果你需要思考或说明，请全部写在最前面；思考完毕后单独一行输出 ===TITLES===，"
    "其后只放所有标题、标题之间用逗号「，」分隔。"
    "===TITLES=== 之后不要出现任何非标题文字，"
    "不要使用 Markdown 表格、不要序号/編號/字元數、不要引号或表情。"
)


def load_config():
    """
    从 JSON 文件加载应用配置。

    返回默认配置与文件配置的合并字典。如果配置文件不存在、损坏或无法读取，
    函数不会抛出异常，而是记录日志并返回默认配置。

    Returns:
        dict: 合并后的配置字典
    """
    from services.file_service import get_config_path
    config_file = get_config_path(_CONFIG_FILENAME)

    if not config_file.exists():
        logger.info("Config file not found, using defaults: %s", config_file)
        return dict(DEFAULT_CONFIG)

    try:
        with open(str(config_file), encoding="utf-8") as f:
            data = json.load(f)
        if not isinstance(data, dict):
            raise ValueError("Config root is not a JSON object")
        merged = dict(DEFAULT_CONFIG)
        merged.update(data)
        logger.info("Config loaded from %s", config_file)
        return merged
    except (json.JSONDecodeError, ValueError) as exc:
        logger.warning("Config file damaged (%s), using defaults: %s", exc, config_file)
        return dict(DEFAULT_CONFIG)
    except OSError as exc:
        logger.warning("Config file unreadable (%s), using defaults: %s", exc, config_file)
        return dict(DEFAULT_CONFIG)


def save_config(data):
    """Save config dict to JSON. Logs error on failure, does not raise."""
    from services.file_service import get_config_path
    config_file = get_config_path(_CONFIG_FILENAME)

    try:
        config_file.parent.mkdir(parents=True, exist_ok=True)
        with open(str(config_file), "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        logger.info("Config saved to %s", config_file)
    except OSError as exc:
        logger.error("Failed to save config to %s: %s", config_file, exc)


def _seed_factory_config_if_missing(filename):
    """Copy a packaged config template to user config only when missing."""
    from shutil import copy2
    from services.file_service import get_app_dir, get_config_path

    target = get_config_path(filename)
    if target.exists():
        return False

    source = get_app_dir() / "config" / filename
    if not source.exists():
        return False

    try:
        target.parent.mkdir(parents=True, exist_ok=True)
        copy2(str(source), str(target))
        logger.info("Seeded factory config from %s to %s", source, target)
        return True
    except OSError as exc:
        logger.warning(
            "Failed to seed factory config from %s to %s: %s",
            source,
            target,
            exc,
        )
        return False


def _factory_config_path(filename):
    """Return the packaged factory config path under the program root."""
    from services.file_service import get_app_dir
    return get_app_dir() / "config" / filename


def _load_json_file(path):
    with open(str(path), encoding="utf-8-sig") as f:
        return json.load(f)


def _model_list_from_data(data):
    models = data.get("models") if isinstance(data, dict) else data
    return models if isinstance(models, list) else None


def _model_name(model):
    if not isinstance(model, dict):
        return ""
    return str(model.get("name", "")).strip()


def _append_missing_title_model(models_file):
    """Append the configured title model from factory ai_models.json if missing.

    Existing user models and API keys are never overwritten. This only handles
    the upgrade case where a user already has ai_models.json but lacks the new
    app_config.title_model entry (docs/11 §6.1).
    """
    title_model_name = str(load_config().get("title_model", "")).strip()
    if not title_model_name or not models_file.exists():
        return False

    try:
        user_data = _load_json_file(models_file)
    except (json.JSONDecodeError, ValueError, OSError) as exc:
        logger.warning("AI models file unreadable (%s): %s", exc, models_file)
        return False

    user_models = _model_list_from_data(user_data)
    if user_models is None:
        logger.warning("AI models file has no model list: %s", models_file)
        return False
    if any(_model_name(model) == title_model_name for model in user_models):
        return False

    factory_file = _factory_config_path(_AI_MODELS_FILENAME)
    if not factory_file.exists():
        logger.info("Factory AI models template not found: %s", factory_file)
        return False

    try:
        factory_data = _load_json_file(factory_file)
    except (json.JSONDecodeError, ValueError, OSError) as exc:
        logger.warning("Factory AI models template unreadable (%s): %s", exc, factory_file)
        return False

    factory_models = _model_list_from_data(factory_data) or []
    title_model = None
    for model in factory_models:
        if _model_name(model) == title_model_name:
            title_model = dict(model)
            break
    if title_model is None:
        logger.info(
            "Factory AI models template has no title model named %s: %s",
            title_model_name,
            factory_file,
        )
        return False

    try:
        user_models.append(title_model)
        with open(str(models_file), "w", encoding="utf-8") as f:
            json.dump(user_data, f, ensure_ascii=False, indent=2)
        logger.info("Appended missing title model %s to %s", title_model_name, models_file)
        return True
    except OSError as exc:
        logger.warning("Failed to append title model to %s: %s", models_file, exc)
        return False


def load_ai_models():
    """Load AI model configs from ai_models.json.

    Returns a list of dicts (each with a ``name`` plus AiModelConfig fields).
    Missing/corrupt file → empty list (the UI then prompts the admin to add one).
    Keys live only in ~/.cmbot and are never committed (docs/11 §11).
    """
    from services.file_service import get_config_path
    _seed_factory_config_if_missing(_AI_MODELS_FILENAME)
    models_file = get_config_path(_AI_MODELS_FILENAME)
    _append_missing_title_model(models_file)
    if not models_file.exists():
        logger.info("AI models file not found: %s", models_file)
        return []
    try:
        data = _load_json_file(models_file)
    except (json.JSONDecodeError, ValueError, OSError) as exc:
        logger.warning("AI models file unreadable (%s): %s", exc, models_file)
        return []

    models = _model_list_from_data(data)
    if not isinstance(models, list):
        logger.warning("AI models file has no model list: %s", models_file)
        return []
    return [m for m in models if isinstance(m, dict)]


def load_outfit_prompt():
    """Return the saved outfit prompt template, or the built-in default."""
    from services.file_service import get_config_path
    _seed_factory_config_if_missing(_OUTFIT_PROMPT_FILENAME)
    prompt_file = get_config_path(_OUTFIT_PROMPT_FILENAME)
    if not prompt_file.exists():
        return DEFAULT_OUTFIT_PROMPT
    try:
        with open(str(prompt_file), encoding="utf-8-sig") as f:
            return f.read()
    except OSError as exc:
        logger.warning("Outfit prompt unreadable (%s): %s", exc, prompt_file)
        return DEFAULT_OUTFIT_PROMPT


def save_outfit_prompt(text):
    """Persist the outfit prompt template (utf-8, no BOM). Does not raise."""
    from services.file_service import get_config_path
    prompt_file = get_config_path(_OUTFIT_PROMPT_FILENAME)
    try:
        prompt_file.parent.mkdir(parents=True, exist_ok=True)
        with open(str(prompt_file), "w", encoding="utf-8") as f:
            f.write(text)
        logger.info("Outfit prompt saved to %s", prompt_file)
    except OSError as exc:
        logger.error("Failed to save outfit prompt to %s: %s", prompt_file, exc)


def load_title_prompt():
    """Return the saved 标题生成提示词, or the built-in default (docs/11 §17.3)."""
    from services.file_service import get_config_path
    _seed_factory_config_if_missing(_TITLE_PROMPT_FILENAME)
    prompt_file = get_config_path(_TITLE_PROMPT_FILENAME)
    if not prompt_file.exists():
        return DEFAULT_TITLE_PROMPT
    try:
        with open(str(prompt_file), encoding="utf-8-sig") as f:
            text = f.read()
        return text if text.strip() else DEFAULT_TITLE_PROMPT
    except OSError as exc:
        logger.warning("Title prompt unreadable (%s): %s", exc, prompt_file)
        return DEFAULT_TITLE_PROMPT


def save_title_prompt(text):
    """Persist the 标题生成提示词 (utf-8, no BOM). Does not raise."""
    from services.file_service import get_config_path
    prompt_file = get_config_path(_TITLE_PROMPT_FILENAME)
    try:
        prompt_file.parent.mkdir(parents=True, exist_ok=True)
        with open(str(prompt_file), "w", encoding="utf-8") as f:
            f.write(text)
        logger.info("Title prompt saved to %s", prompt_file)
    except OSError as exc:
        logger.error("Failed to save title prompt to %s: %s", prompt_file, exc)


def _normalize_prompts(data):
    """Keep only valid {name, text} entries (non-empty name, string text)."""
    if not isinstance(data, list):
        return []
    out = []
    for item in data:
        if isinstance(item, dict):
            name = str(item.get("name", "")).strip()
            text = item.get("text", "")
            if name and isinstance(text, str):
                out.append({"name": name, "text": text})
    return out


def load_outfit_prompts():
    """Load named 话术 templates (docs/11 §7.2); always returns >= 1.

    Missing/corrupt → migrate the legacy outfit_prompt.txt into a single
    「默认」template, or seed it from DEFAULT_OUTFIT_PROMPT.
    """
    from services.file_service import get_config_path
    prompts_file = get_config_path(_OUTFIT_PROMPTS_FILENAME)
    if prompts_file.exists():
        try:
            with open(str(prompts_file), encoding="utf-8-sig") as f:
                prompts = _normalize_prompts(json.load(f))
            if prompts:
                return prompts
        except (json.JSONDecodeError, ValueError, OSError) as exc:
            logger.warning("Outfit prompts unreadable (%s): %s", exc, prompts_file)

    # Seed / migrate (load_outfit_prompt reads the legacy txt or the default).
    return [{"name": DEFAULT_OUTFIT_PROMPT_NAME, "text": load_outfit_prompt()}]


def save_outfit_prompts(prompts):
    """Persist named 话术 templates as JSON (utf-8, no BOM). Does not raise."""
    from services.file_service import get_config_path
    prompts_file = get_config_path(_OUTFIT_PROMPTS_FILENAME)
    try:
        prompts_file.parent.mkdir(parents=True, exist_ok=True)
        with open(str(prompts_file), "w", encoding="utf-8") as f:
            json.dump(list(prompts), f, ensure_ascii=False, indent=2)
        logger.info("Outfit prompts saved to %s", prompts_file)
    except OSError as exc:
        logger.error("Failed to save outfit prompts to %s: %s", prompts_file, exc)