#!/usr/bin/env python3
"""Validate fast/slow/ambiguous label records without external dependencies."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any


POLICY_VERSION = "XA4.1-2026.6"

REASON_CODES_BY_LABEL: dict[str, set[str]] = {
    "快": {
        "DIRECT_ACTION",
        "DIRECT_DETERMINISTIC_QUERY",
        "EXPLICIT_DEVICE_MULTI_ACTION",
        "DIRECT_OPERATION_FLOW",
    },
    "慢": {
        "INTENT_INFERENCE",
        "SCENE_GOAL_MAPPING",
        "MULTI_STEP_PLANNING",
        "COMPLEX_REASONING",
        "TRUE_DISAMBIGUATION",
        "CONTEXT_REQUIRED",
        "MISSING_REQUIRED_SLOT",
        "AUTOMATION_SLOW_POLICY",
    },
    "模糊": {
        "OPEN_CHAT_OR_QA",
        "PRODUCT_KNOWLEDGE",
        "FAULT_DIAGNOSIS",
        "DEVICE_OR_ENV_STATE_QUERY",
        "POLICY_UNCERTAIN",
    },
}

DEFAULT_ROUTE_BY_LABEL = {
    "快": "快系统",
    "慢": "慢系统",
    "模糊": "慢系统",
}

LEGACY_LABELS = {
    "FAST_DIRECT",
    "SLOW_FILTER_RANK",
    "SLOW_DISAMBIGUATE",
    "SLOW_GOAL_STATE",
    "SLOW_WORKFLOW",
    "SLOW_CONTEXT_DEPENDENT",
    "SLOW_UNCLEAR",
    "SLOW_ADVANCED_CAPABILITY",
}

CONFIDENCE_VALUES = {"high", "medium", "low"}

REQUIRED_FIELDS = {
    "query",
    "label",
    "default_route",
    "primary_reason_code",
    "reason_codes",
    "reason",
    "evidence",
    "context_used",
    "fast_system_pending",
    "confidence",
    "review_required",
    "uncertainties",
    "policy_version",
}

OPTIONAL_FIELDS = {
    "id",
    "counterevidence",
    "domain_label",
    "candidate_domain_labels",
    "structural_signals",
    "legacy_eval_label",
}


def _is_nonempty_string(value: Any) -> bool:
    return isinstance(value, str) and bool(value.strip())


def _check_string_list(
    value: Any,
    *,
    field: str,
    errors: list[str],
    require_nonempty: bool = False,
    unique: bool = False,
) -> None:
    if not isinstance(value, list):
        errors.append(f"{field} 必须是数组")
        return
    if require_nonempty and not value:
        errors.append(f"{field} 至少包含一项")
    if any(not _is_nonempty_string(item) for item in value):
        errors.append(f"{field} 的每一项都必须是非空字符串")
    if unique and all(isinstance(item, str) for item in value):
        if len(value) != len(set(value)):
            errors.append(f"{field} 不允许重复项")


def validate_record(record: Any, index: int) -> dict[str, Any]:
    errors: list[str] = []
    warnings: list[str] = []

    if not isinstance(record, dict):
        return {
            "index": index,
            "valid": False,
            "errors": ["记录必须是 JSON 对象"],
            "warnings": [],
        }

    if "__parse_error__" in record:
        return {
            "index": index,
            "valid": False,
            "errors": [str(record["__parse_error__"])],
            "warnings": [],
        }

    missing = sorted(REQUIRED_FIELDS - set(record))
    if missing:
        errors.append(f"缺少必填字段: {', '.join(missing)}")

    unknown = sorted(set(record) - REQUIRED_FIELDS - OPTIONAL_FIELDS)
    if unknown:
        errors.append(f"存在未知字段: {', '.join(unknown)}")

    query = record.get("query")
    if not _is_nonempty_string(query):
        errors.append("query 必须是非空字符串")

    label = record.get("label")
    if label not in REASON_CODES_BY_LABEL:
        errors.append("label 必须是 快、慢、模糊 之一")

    default_route = record.get("default_route")
    expected_route = DEFAULT_ROUTE_BY_LABEL.get(label)
    if expected_route is not None and default_route != expected_route:
        errors.append(f"label={label} 时 default_route 必须是 {expected_route}")

    primary_reason_code = record.get("primary_reason_code")
    reason_codes = record.get("reason_codes")
    allowed_reasons = REASON_CODES_BY_LABEL.get(label, set())
    if primary_reason_code not in allowed_reasons:
        errors.append(
            f"primary_reason_code={primary_reason_code!r} 与 label={label!r} 不兼容"
        )

    if not isinstance(reason_codes, list):
        errors.append("reason_codes 必须是数组")
    else:
        if not reason_codes:
            errors.append("reason_codes 至少包含一项")
        if all(isinstance(code, str) for code in reason_codes):
            if len(reason_codes) != len(set(reason_codes)):
                errors.append("reason_codes 不允许重复")
        invalid_reason_codes = [
            code
            for code in reason_codes
            if not isinstance(code, str) or code not in allowed_reasons
        ]
        if invalid_reason_codes:
            rendered_codes = ", ".join(repr(code) for code in invalid_reason_codes)
            errors.append(f"reason_codes 包含与 label 不兼容的值: {rendered_codes}")
        if primary_reason_code not in reason_codes:
            errors.append("primary_reason_code 必须同时出现在 reason_codes 中")

    if not _is_nonempty_string(record.get("reason")):
        errors.append("reason 必须是非空字符串")

    _check_string_list(
        record.get("evidence"),
        field="evidence",
        errors=errors,
        require_nonempty=True,
    )

    if "counterevidence" in record:
        _check_string_list(
            record.get("counterevidence"),
            field="counterevidence",
            errors=errors,
        )

    if "domain_label" in record:
        domain_label = record.get("domain_label")
        if domain_label is not None and not _is_nonempty_string(domain_label):
            errors.append("domain_label 必须是非空字符串或 null")

    if "candidate_domain_labels" in record:
        _check_string_list(
            record.get("candidate_domain_labels"),
            field="candidate_domain_labels",
            errors=errors,
            unique=True,
        )

    structural_signals = record.get("structural_signals")
    if structural_signals is not None:
        if not isinstance(structural_signals, dict):
            errors.append("structural_signals 必须是对象")
        else:
            expected_keys = {"complex", "multi_instruction", "auto_task"}
            actual_keys = set(structural_signals)
            if actual_keys != expected_keys:
                errors.append(
                    "structural_signals 必须且只能包含 "
                    "complex、multi_instruction、auto_task"
                )
            for key in expected_keys:
                value = structural_signals.get(key)
                if value is not None and not isinstance(value, bool):
                    errors.append(f"structural_signals.{key} 必须是布尔值或 null")

    if not isinstance(record.get("context_used"), bool):
        errors.append("context_used 必须是布尔值")

    fast_system_pending = record.get("fast_system_pending")
    if fast_system_pending is not None and not isinstance(fast_system_pending, bool):
        errors.append("fast_system_pending 必须是布尔值或 null")
    if fast_system_pending is True and label != "快":
        errors.append("只有 label=快 时 fast_system_pending 才能为 true")

    legacy_eval_label = record.get("legacy_eval_label")
    if (
        "legacy_eval_label" in record
        and legacy_eval_label is not None
        and legacy_eval_label not in LEGACY_LABELS
    ):
        errors.append("legacy_eval_label 不是受支持的旧八类标签")
    if label == "模糊" and legacy_eval_label is not None:
        warnings.append("模糊档通常无法无损映射到旧二值标签，请确认迁移依据")

    confidence = record.get("confidence")
    if confidence not in CONFIDENCE_VALUES:
        errors.append("confidence 必须是 high、medium、low 之一")

    review_required = record.get("review_required")
    if not isinstance(review_required, bool):
        errors.append("review_required 必须是布尔值")
    if confidence == "low" and review_required is not True:
        errors.append("confidence=low 时 review_required 必须为 true")

    _check_string_list(
        record.get("uncertainties"),
        field="uncertainties",
        errors=errors,
    )
    uncertainties = record.get("uncertainties")
    if isinstance(uncertainties, list) and uncertainties and review_required is False:
        warnings.append("uncertainties 非空但 review_required=false，请确认不确定点不影响标签")

    if record.get("policy_version") != POLICY_VERSION:
        errors.append(f"policy_version 必须是 {POLICY_VERSION}")

    return {
        "index": index,
        "id": record.get("id"),
        "valid": not errors,
        "errors": errors,
        "warnings": warnings,
    }


def _load_records_from_file(path: Path) -> list[Any]:
    text = path.read_text(encoding="utf-8-sig")
    if path.suffix.lower() == ".jsonl":
        records: list[Any] = []
        for line_number, line in enumerate(text.splitlines(), start=1):
            if not line.strip():
                continue
            try:
                records.append(json.loads(line))
            except json.JSONDecodeError as exc:
                records.append(
                    {
                        "__parse_error__": (
                            f"JSONL 第 {line_number} 行解析失败: {exc.msg}"
                        )
                    }
                )
        return records

    payload = json.loads(text)
    if isinstance(payload, list):
        return payload
    if isinstance(payload, dict) and isinstance(payload.get("records"), list):
        return payload["records"]
    return [payload]


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="校验 XA4.1-2026.6 快慢分流打标 JSON/JSONL"
    )
    source_group = parser.add_mutually_exclusive_group(required=True)
    source_group.add_argument("--record", help="单条 JSON 对象字符串")
    source_group.add_argument("--file", type=Path, help="JSON 或 JSONL 文件路径")
    parser.add_argument(
        "--strict",
        action="store_true",
        help="把 warning 也视为校验失败",
    )
    return parser.parse_args()


def main() -> int:
    args = _parse_args()
    try:
        if args.record is not None:
            records = [json.loads(args.record)]
        else:
            records = _load_records_from_file(args.file)
    except (OSError, json.JSONDecodeError) as exc:
        result = {
            "valid": False,
            "total": 0,
            "invalid": 1,
            "warning_count": 0,
            "errors": [str(exc)],
            "results": [],
        }
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 1

    results = [validate_record(record, index) for index, record in enumerate(records)]
    invalid = sum(not result["valid"] for result in results)
    warning_count = sum(len(result["warnings"]) for result in results)
    valid = invalid == 0 and (not args.strict or warning_count == 0)
    summary = {
        "valid": valid,
        "total": len(records),
        "invalid": invalid,
        "warning_count": warning_count,
        "results": results,
    }
    print(json.dumps(summary, ensure_ascii=False, indent=2))
    return 0 if valid else 1


if __name__ == "__main__":
    sys.exit(main())