Nexus/server/utils/validation_errors_zh.py

"""Translate Pydantic / FastAPI validation errors to Chinese for API 422 responses."""

from __future__ import annotations

import re
from typing import Any

# Pydantic v2 default messages (exact match)
_MSG_EXACT: dict[str, str] = {
    "Field required": "必填",
    "Input should be a valid string": "必须是字符串",
    "Input should be a valid integer": "必须是整数",
    "Input should be a valid number": "必须是数字",
    "Input should be a valid boolean": "必须是布尔值",
    "Input should be a valid dictionary": "必须是对象",
    "Input should be a valid list": "必须是数组",
    "JSON decode error": "JSON 格式无效",
    "Input should be a valid dictionary or object to extract fields from": "请求体必须是 JSON 对象",
    "Input should be a valid UUID": "必须是有效的 UUID",
    "Input should be a valid datetime": "必须是有效的日期时间",
    "Input should be a valid date": "必须是有效的日期",
    "Input should be a valid time": "必须是有效的时间",
    "Input should be a valid url": "必须是有效的 URL",
    "Input should be a valid email address": "必须是有效的邮箱",
}

# Regex patterns for parameterized Pydantic messages (legacy / string fields)
_MSG_PATTERNS: list[tuple[re.Pattern[str], str]] = [
    (re.compile(r"^String should have at least (\d+) characters?$"), r"至少需要 \1 个字符"),
    (re.compile(r"^String should have at most (\d+) characters?$"), r"最多 \1 个字符"),
    (re.compile(r"^String should match pattern '(.+)'$"), r"格式不符合要求"),
    (re.compile(r"^Input should be greater than or equal to (\d+)"), r"不能小于 \1"),
    (re.compile(r"^Input should be less than or equal to (\d+)"), r"不能大于 \1"),
    (re.compile(r"^Input should be greater than (\d+)"), r"必须大于 \1"),
    (re.compile(r"^Input should be less than (\d+)"), r"必须小于 \1"),
    (re.compile(r"^List should have at least (\d+) items?(?: .*)?$"), r"至少需要 \1 项"),
    (re.compile(r"^List should have at most (\d+) items?(?: after validation, not (\d+))?$"), r"最多 \1 项"),
]


def translate_validation_msg(msg: str, err_type: str = "", ctx: dict[str, Any] | None = None) -> str:
    """Return Chinese message when a known translation exists."""
    ctx = ctx or {}
    if err_type == "too_long" and ctx.get("field_type") == "List":
        max_len = ctx.get("max_length")
        actual = ctx.get("actual_length")
        if max_len is not None and actual is not None:
            return f"最多 {max_len} 项，当前 {actual} 项"
        if max_len is not None:
            return f"最多 {max_len} 项"
    if err_type == "too_short" and ctx.get("field_type") == "List":
        min_len = ctx.get("min_length")
        actual = ctx.get("actual_length")
        if min_len is not None and actual is not None:
            return f"至少需要 {min_len} 项，当前 {actual} 项"
        if min_len is not None:
            return f"至少需要 {min_len} 项"
    if err_type == "missing":
        return "必填"
    if not msg:
        return msg
    if msg in _MSG_EXACT:
        return _MSG_EXACT[msg]
    for pattern, repl in _MSG_PATTERNS:
        if pattern.match(msg):
            out = pattern.sub(repl, msg)
            # too_long list with "not N" suffix — append actual count when present
            m = re.search(r"not (\d+)$", msg)
            if m and "最多" in out and "当前" not in out:
                return f"{out}，当前 {m.group(1)} 项"
            return out
    if msg.startswith("Value error, "):
        return msg[len("Value error, "):]
    if err_type == "string_type":
        return "必须是字符串"
    if err_type in ("int_type", "int_parsing"):
        return "必须是整数"
    if err_type in ("float_type", "float_parsing"):
        return "必须是数字"
    if err_type == "bool_parsing":
        return "必须是 true 或 false"
    if err_type == "less_than_equal" and ctx.get("le") is not None:
        return f"不能大于 {ctx['le']}"
    if err_type == "greater_than_equal" and ctx.get("ge") is not None:
        return f"不能小于 {ctx['ge']}"
    if err_type == "less_than" and ctx.get("lt") is not None:
        return f"必须小于 {ctx['lt']}"
    if err_type == "greater_than" and ctx.get("gt") is not None:
        return f"必须大于 {ctx['gt']}"
    return msg


_LOC_PREFIX_SKIP = frozenset({"query", "body", "path", "header", "cookie"})

_FIELD_ZH: dict[str, str] = {
    "per_page": "每页条数",
    "page": "页码",
    "limit": "条数",
    "offset": "偏移量",
    "server_ids": "服务器",
    "source_path": "源路径",
    "target_path": "目标路径",
    "sync_mode": "同步模式",
    "batch_id": "批次 ID",
    "batch_size": "批次大小",
    "concurrency": "并发数",
    "value": "值",
    "username": "用户名",
    "password": "密码",
    "name": "名称",
    "cron_expr": "Cron 表达式",
    "fire_at": "执行时间",
}


def translate_loc_zh(loc: Any) -> str:
    """Human-readable Chinese field path (no query/body prefixes)."""
    if not isinstance(loc, (list, tuple)):
        return "参数"
    parts: list[str] = []
    for piece in loc:
        if isinstance(piece, str) and piece in _LOC_PREFIX_SKIP:
            continue
        if isinstance(piece, int):
            parts.append(f"第 {piece + 1} 项")
            continue
        key = str(piece)
        parts.append(_FIELD_ZH.get(key, key))
    return ".".join(parts) if parts else "参数"


def translate_validation_errors(errors: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Clone Pydantic error dicts with translated ``msg`` and ``loc_zh``."""
    translated: list[dict[str, Any]] = []
    for err in errors:
        ctx = err.get("ctx") if isinstance(err.get("ctx"), dict) else None
        item = {k: v for k, v in err.items() if k != "ctx"}
        item["msg"] = translate_validation_msg(
            str(err.get("msg", "")),
            str(err.get("type", "")),
            ctx,
        )
        item["loc_zh"] = translate_loc_zh(err.get("loc"))
        translated.append(item)
    return translated