ai-kefu-admin/app/platform/config_validation.py

"""配置校验管线(Phase 2 Task 6)——PUT /config/{group} 的分组校验。

validate_group(group, payload, context) -> list[str]:错误清单(中文,带
字段路径),空清单 = 通过;路由层把非空清单拼进 400 detail。

纪律:
- **键 allowlist**:各组未知键显性拒绝(与 wall_schema._KNOWN_FIELDS 同款
  防呆——配置编辑器时代拼错键 = 深合并静默忽略,这里前移到保存期报错)。
  wall 组例外:未知键交 compile_wall 管(它的报错自带 wall 侧字段路径)。
- **单一事实源**:各组合法键从 DEFAULT_CONFIG 派生(prompts 除外——
  DEFAULT 不含 prompts,取 ai_service.PROMPTS_KEYS 四件套)。
- **wall 组不重复造校验**:深拷贝 payload 交 compile_wall,复用全部既有
  校验(键 allowlist/占位符白名单/flags/正则编译/必填池)。
- **prompts 组**:四键字符串 + 大小上限(company_facts ≤50KB 软上限,
  spec §5.7——每轮整段拼入 system prompt,过大门线性放大 LLM 成本;
  其余三件 ≤100KB)。字节口径按 UTF-8 编码长度。
- **push.prefix 跨租户查重**(spec E1,external_key 前缀全局唯一防 CRM
  线索空间串号):context.other_prefixes 由路由层从 DB 各租户当前版本
  收集(排除自身;空串 = 未设置,不参战——出厂默认本就多租户同空)。
- 数值字段(None = "未设置"哨兵,与 DEFAULT_CONFIG 同款纪律):None 或
  范围内数值;bool 不是 int(isinstance 陷阱,显式排除)。
"""
import copy
from dataclasses import dataclass, field

from app.engine.services.ai_service import PROMPTS_KEYS
from app.engine.wall_schema import WallConfigError, compile_wall
from app.platform.defaults import DEFAULT_CONFIG

# 各组合法键 = DEFAULT_CONFIG 同组键(单一事实源);prompts 特例见上。
_ALLOWED_KEYS: dict[str, frozenset] = {
    group: frozenset(keys) for group, keys in DEFAULT_CONFIG.items()}
_ALLOWED_KEYS["prompts"] = frozenset(PROMPTS_KEYS)

# 合法组名(排序输出进错误信息,稳定可测)
CONFIG_GROUPS = frozenset(_ALLOWED_KEYS)

# prompts 大小上限(字节,UTF-8 口径):company_facts 50KB 软上限,其余 100KB
PROMPTS_SIZE_CAPS = {
    name: 50 * 1024 if name == "company_facts" else 100 * 1024
    for name in PROMPTS_KEYS}

_BRAND_STR_KEYS = ("title", "color", "avatar", "welcome", "placeholder",
                   "whatsapp", "email", "privacy_url")


@dataclass(frozen=True)
class ValidationContext:
    """跨租户校验上下文(路由层按请求现势收集,校验管线不自查 DB)。"""
    other_prefixes: frozenset = field(default_factory=frozenset)


def _type_name(value) -> str:
    return type(value).__name__


def _want_str(value, path: str, errors: list) -> None:
    if not isinstance(value, str):
        errors.append(f"{path}: 须为字符串(实际 {_type_name(value)})")


def _want_bool(value, path: str, errors: list) -> None:
    if not isinstance(value, bool):
        errors.append(f"{path}: 须为布尔(实际 {_type_name(value)})")


def _want_int(value, path: str, errors: list) -> None:
    # bool 是 int 的子类——True 会被 isinstance 放过,显式排除
    if not isinstance(value, int) or isinstance(value, bool):
        errors.append(f"{path}: 须为整数(实际 {_type_name(value)})")


def _want_str_list(value, path: str, errors: list) -> None:
    if not isinstance(value, list):
        errors.append(f"{path}: 须为字符串列表(实际 {_type_name(value)})")
        return
    for i, item in enumerate(value):
        if not isinstance(item, str):
            errors.append(f"{path}[{i}]: 项须为字符串(实际 {_type_name(item)})")


def _check_brand(payload: dict, ctx: ValidationContext, errors: list) -> None:
    for key in _BRAND_STR_KEYS:
        if key in payload:
            _want_str(payload[key], f"brand.{key}", errors)
    if "topics" in payload:
        _want_str_list(payload["topics"], "brand.topics", errors)


def _check_wall(payload: dict, ctx: ValidationContext, errors: list) -> None:
    """全量交 compile_wall(深拷贝防意外 mutation):键 allowlist/占位符/
    flags/正则编译/必填池全套复用,WallConfigError 转清单项。"""
    try:
        compile_wall(copy.deepcopy(payload))
    except WallConfigError as exc:
        errors.append(str(exc))


def _check_prompts(payload: dict, ctx: ValidationContext,
                   errors: list) -> None:
    for key in PROMPTS_KEYS:
        if key not in payload:
            errors.append(f"prompts.{key}: 缺少必填键(四件套须整组提交)")
            continue
        text = payload[key]
        if not isinstance(text, str):
            errors.append(f"prompts.{key}: 须为字符串(实际 {_type_name(text)})")
            continue
        size = len(text.encode("utf-8"))
        cap = PROMPTS_SIZE_CAPS[key]
        if size > cap:
            errors.append(
                f"prompts.{key}: 内容 {size} 字节超过上限 {cap} 字节"
                f"({cap // 1024}KB,spec §5.7——每轮拼入 system prompt,"
                f"请精炼)")


def _check_contact(payload: dict, ctx: ValidationContext,
                   errors: list) -> None:
    for key in ("cc_completion", "email_light_ask"):
        if key in payload:
            _want_bool(payload[key], f"contact.{key}", errors)
    if "template_languages" in payload:
        _want_str_list(payload["template_languages"],
                       "contact.template_languages", errors)


def _check_email(payload: dict, ctx: ValidationContext, errors: list) -> None:
    if "sender_name" in payload:
        _want_str(payload["sender_name"], "email.sender_name", errors)
    if "smtp" in payload and not (payload["smtp"] is None
                                  or isinstance(payload["smtp"], dict)):
        errors.append(f"email.smtp: 须为映射或 null(实际 "
                      f"{_type_name(payload['smtp'])})")
    if "notify_emails" in payload:
        value = payload["notify_emails"]
        if not isinstance(value, list):
            errors.append(f"email.notify_emails: 须为列表(实际 "
                          f"{_type_name(value)})")
            return
        for i, item in enumerate(value):
            if not isinstance(item, str) or "@" not in item:
                errors.append(f"email.notify_emails[{i}]: 项须为含 @ 的邮箱"
                              f"地址(实际 {item!r})")


def _check_push(payload: dict, ctx: ValidationContext, errors: list) -> None:
    if "enabled" in payload:
        _want_bool(payload["enabled"], "push.enabled", errors)
    for key in ("url", "api_key", "prefix"):
        if key in payload:
            _want_str(payload[key], f"push.{key}", errors)
    if "timeout" in payload:
        _want_positive_number(payload["timeout"], "push.timeout", errors)
    if "retry_backoff_base" in payload:
        _want_positive_number(payload["retry_backoff_base"],
                              "push.retry_backoff_base", errors)
    if "max_attempts" in payload:
        value = payload["max_attempts"]
        if value is not None and (not isinstance(value, int)
                                  or isinstance(value, bool)
                                  or not 1 <= value <= 10):
            errors.append(f"push.max_attempts: 须为 1-10 整数或 null(实际 "
                          f"{value!r})")
    # spec E1: external_key 前缀全局唯一(空串 = 未设置,不参战)
    prefix = payload.get("prefix")
    if isinstance(prefix, str) and prefix and prefix in ctx.other_prefixes:
        errors.append(f"push.prefix: {prefix!r} 已被其他租户占用"
                      f"(external_key 前缀全局唯一)")


def _want_positive_number(value, path: str, errors: list) -> None:
    if value is None:
        return
    if (not isinstance(value, (int, float)) or isinstance(value, bool)
            or value <= 0):
        errors.append(f"{path}: 须为正数或 null(实际 {value!r})")


def _check_ai(payload: dict, ctx: ValidationContext, errors: list) -> None:
    for key in ("provider", "model"):
        if key in payload:
            _want_str(payload[key], f"ai.{key}", errors)
    for key in ("api_key", "base_url"):
        if key in payload:
            value = payload[key]
            if value is not None and not isinstance(value, str):
                errors.append(f"ai.{key}: 须为字符串或 null(实际 "
                              f"{_type_name(value)})")


def _check_engine(payload: dict, ctx: ValidationContext,
                  errors: list) -> None:
    for key in ("rate_limit_per_hour", "ws_conn_limit"):
        if key in payload:
            value = payload[key]
            _want_int(value, f"engine.{key}", errors)
            if isinstance(value, int) and not isinstance(value, bool) \
                    and value < 1:
                errors.append(f"engine.{key}: 须为正整数(实际 {value})")
    if "context_seed_keys" in payload:
        value = payload["context_seed_keys"]
        if not isinstance(value, dict):
            errors.append(f"engine.context_seed_keys: 须为映射(实际 "
                          f"{_type_name(value)})")
            return
        for k, v in value.items():
            if not isinstance(v, str):
                errors.append(f"engine.context_seed_keys.{k}: 须为字符串"
                              f"(实际 {_type_name(v)})")


def _check_misc(payload: dict, ctx: ValidationContext, errors: list) -> None:
    if "name_to_region_extras" in payload:
        value = payload["name_to_region_extras"]
        if not isinstance(value, dict):
            errors.append(f"misc.name_to_region_extras: 须为映射(实际 "
                          f"{_type_name(value)})")
            return
        for k, v in value.items():
            if not isinstance(v, str):
                errors.append(f"misc.name_to_region_extras.{k}: 须为字符串"
                              f"(实际 {_type_name(v)})")


_GROUP_CHECKERS = {
    "brand": _check_brand,
    "wall": _check_wall,
    "contact": _check_contact,
    "email": _check_email,
    "push": _check_push,
    "ai": _check_ai,
    "engine": _check_engine,
    "misc": _check_misc,
    "prompts": _check_prompts,
}


def validate_group(group: str, payload, context: ValidationContext) -> list:
    """校验单个配置组的 payload(整组替换语义:只查出现键的类型/形状,
    缺省键由消费侧深合并出厂默认补)。返回错误清单(中文,带字段路径),
    空清单 = 通过。未知组名/非对象 payload 也走清单(路由统一转 400)。"""
    checker = _GROUP_CHECKERS.get(group)
    if checker is None:
        return [f"未知配置组: {group!r}(合法: "
                f"{', '.join(sorted(CONFIG_GROUPS))})"]
    if not isinstance(payload, dict):
        return [f"{group}: 须为 JSON 对象(实际 {_type_name(payload)})"]
    errors: list = []
    if group != "wall":    # wall 的未知键交 compile_wall(报错自带其字段路径)
        unknown = set(payload) - _ALLOWED_KEYS[group]
        if unknown:
            errors.append(f"{group}: 未知键 {', '.join(sorted(unknown))}"
                          f"(合法: {', '.join(sorted(_ALLOWED_KEYS[group]))})")
    checker(payload, context, errors)
    return errors