"""高级运费规则：AI 起草（只读）与显式审核后应用（写库）。

── 铁律 ────────────────────────────────────────────────────────────────────
1. **起草不写库**：ai_draft 只读租户事实、只调模型、只跑内存里的模拟；
   一条规则、一条修订、一行配置都不落地。唯一的写是 consume_ai_quota 的额度计数
   （那是平台计费，不是运费配置）。
2. **不另开校验分支**：模型产出的规则一律过 schemas.AdvancedShippingRuleIn，
   与手工配置同一个契约。第二条校验路径迟早会和第一条飘开。
3. **应用走既有写路径**：ai_apply 逐条调 admin_service._insert_rule，
   于是自动继承租户归属校验、租户配额、聚合策略不变式、修订写入；
   收尾统一 _finish（按**旧** cache_version 失效缓存 + 一次 commit）。
   中途任何一条失败都直接抛异常、绝不 commit —— 整批要么全成要么全不成。
4. **AI 碰不到开关**：请求体里没有 mode，也没有插件启用字段；
   from_legacy_conversion 由 _insert_rule 默认 0，模型塞进来的同名 key 会被剥掉。
5. 金额一律租户本位币；模型不定义拒绝原因，reason key 由引擎按 REASON_KEYS 产出。
"""
from __future__ import annotations

import csv
import io
import json
import logging
import re
from decimal import Decimal
from typing import Any

from fastapi import HTTPException
from pydantic import BaseModel, Field, ValidationError

from app.core.ai_utils import call_ai, resolve_ai_extra
from app.core.services.ai_quota import consume_ai_quota

from . import admin_service as svc
from .cache import load_rule_set, settings_or_default
from .quote import RuleSet, RuleSpec, evaluate_rule_set, make_inherit_resolver
from .schemas import (
    MAX_POSTCODE_PATTERNS,
    MAX_RULES_PER_TENANT,
    MAX_SELECTOR_IDS,
    MAX_TIERS_PER_RULE,
    AdvancedShippingRuleIn,
)
from .templates import TEMPLATE_VERSION, TEMPLATES

logger = logging.getLogger(__name__)

#: 粘贴表格的硬上限。32 KiB 与规则 JSON 的上限同源；行/列上限挡住"整本工作簿贴进来"。
MAX_TABLE_BYTES = 32 * 1024
MAX_TABLE_ROWS = 400
MAX_TABLE_COLUMNS = 32

#: 一次起草最多产出多少条规则（租户总配额是 MAX_RULES_PER_TENANT，这里只限单批）。
MAX_AI_RULES_PER_DRAFT = 60
MAX_SAMPLE_CASES = 6
MAX_PROMPT_CHARS = 4000

#: 进 prompt 的目录/规则条目上限，避免大租户把上下文撑爆。
_CONTEXT_LIMIT = 100

#: 必跑的无地址场景标签。强制模式下没有兜底，"没地址时会怎样"必须每次都摆在操作者面前。
NO_ADDRESS_LABEL = "no_address"

#: 模型不许控制的内部字段：id 由数据库给，from_legacy_conversion 是转换幂等标记，
#: mode/tenant_id 根本不是规则的一部分。剥掉而不是报错——模型偶尔多嘴不该让整批作废。
_FORBIDDEN_RULE_KEYS = frozenset({
    "id", "tenant_id", "mode", "from_legacy_conversion", "current_revision_no", "retired",
})


# ══════════════════════════════════════════════════════════════════════════
#  纯函数：粘贴表格解析（stdlib csv，不引任何依赖、不接受文件上传）
# ══════════════════════════════════════════════════════════════════════════

def _trim_trailing_blanks(row: list[str]) -> list[str]:
    """去掉行尾的空单元格。表格导出常在每行末尾多一个逗号，那不是"列数不一致"。"""
    out = list(row)
    while out and not out[-1].strip():
        out.pop()
    return out


def parse_table_text(text: str) -> list[list[str]]:
    """把粘贴的 CSV/TSV 解析成矩形表；不合规直接抛 ValueError（调用方转 400）。

    表头可以是英文或中文，分隔符由 csv.Sniffer 猜（逗号/制表/分号/竖线），猜不出按逗号。
    **列数必须与表头一致**：一行多一个价格或少一个价格，静默补空会让整片郊区错价。
    """
    if not text or not text.strip():
        raise ValueError("粘贴的表格是空的")
    raw = text.encode("utf-8")
    if len(raw) > MAX_TABLE_BYTES:
        raise ValueError(f"粘贴的表格超过上限 {MAX_TABLE_BYTES} 字节（当前 {len(raw)}）")

    sample = text[:8192]
    try:
        dialect: Any = csv.Sniffer().sniff(sample, delimiters=",\t;|")
    except csv.Error:
        dialect = csv.excel

    rows = [
        _trim_trailing_blanks(row)
        for row in csv.reader(io.StringIO(text), dialect)
    ]
    rows = [row for row in rows if any(cell.strip() for cell in row)]
    if not rows:
        raise ValueError("粘贴的表格没有任何有效行")
    if len(rows) > MAX_TABLE_ROWS:
        raise ValueError(f"粘贴的表格超过上限 {MAX_TABLE_ROWS} 行（当前 {len(rows)}）")

    width = len(rows[0])
    if width > MAX_TABLE_COLUMNS:
        raise ValueError(f"粘贴的表格超过上限 {MAX_TABLE_COLUMNS} 列（当前 {width}）")
    for index, row in enumerate(rows):
        if len(row) != width:
            raise ValueError(
                f"第 {index + 1} 行有 {len(row)} 列，与表头的 {width} 列不一致，请先修正表格")
    return [[cell.strip() for cell in row] for row in rows]


def table_to_text(rows: list[list[str]]) -> str:
    """回填进 prompt 用的规范化表格（统一制表分隔，避免逗号在数值里再次歧义）。"""
    return "\n".join("\t".join(row) for row in rows)


# ══════════════════════════════════════════════════════════════════════════
#  纯函数：模型响应解析与校验
# ══════════════════════════════════════════════════════════════════════════

def extract_json_object(raw: str) -> dict:
    """从模型回复里取出唯一的 JSON 对象，容忍 markdown 代码块包裹。"""
    cleaned = re.sub(r"```(?:json)?\s*", "", raw or "").replace("```", "").strip()
    start, end = cleaned.find("{"), cleaned.rfind("}")
    if start == -1 or end <= start:
        raise ValueError("回复里没有 JSON 对象")
    try:
        data = json.loads(cleaned[start:end + 1])
    except json.JSONDecodeError as exc:
        raise ValueError(f"JSON 解析失败: {exc}") from exc
    if not isinstance(data, dict):
        raise ValueError("回复的顶层不是 JSON 对象")
    return data


def validate_draft_rules(
    raw_rules: Any,
) -> tuple[list[AdvancedShippingRuleIn], list[dict]]:
    """逐条过**手工配置同款**契约。返回 (合法 payload, 不合法条目及原因)。

    不合法的条目不抛异常、也绝不进草稿——原样回给操作者看，让他判断是重跑还是手改。
    """
    if not isinstance(raw_rules, list):
        raise ValueError("模型没有返回 rules 数组")
    if len(raw_rules) > MAX_AI_RULES_PER_DRAFT:
        raise ValueError(f"单次起草最多 {MAX_AI_RULES_PER_DRAFT} 条规则，收到 {len(raw_rules)} 条")

    valid: list[AdvancedShippingRuleIn] = []
    invalid: list[dict] = []
    for index, item in enumerate(raw_rules):
        if not isinstance(item, dict):
            invalid.append({"index": index, "error": "不是 JSON 对象", "raw": item})
            continue
        cleaned = {k: v for k, v in item.items() if k not in _FORBIDDEN_RULE_KEYS}
        try:
            valid.append(AdvancedShippingRuleIn(**cleaned))
        except (ValidationError, TypeError, ValueError) as exc:
            invalid.append({"index": index, "error": str(exc), "raw": cleaned})
    return valid, invalid


def resolve_draft_shipping_method_ids(
    raw_rules: Any, shipping_methods: list[dict[str, Any]],
) -> tuple[Any, list[str]]:
    """Resolve an AI-returned method name or template marker only when unambiguous."""
    if not isinstance(raw_rules, list):
        return raw_rules, []

    methods = [(int(method["id"]), str(method.get("name") or "").strip().casefold())
               for method in shipping_methods if method.get("id") is not None]
    names: dict[str, list[int]] = {}
    for method_id, name in methods:
        if name:
            names.setdefault(name, []).append(method_id)

    resolved: list[Any] = []
    warnings: list[str] = []
    for index, item in enumerate(raw_rules):
        if not isinstance(item, dict) or not isinstance(item.get("shipping_method_id"), str):
            resolved.append(item)
            continue
        value = item["shipping_method_id"].strip()
        matches = names.get(value.casefold(), [])
        method_id = matches[0] if len(matches) == 1 else (
            methods[0][0] if value == "zone_delivery" and len(methods) == 1 else None
        )
        if method_id is None:
            resolved.append(item)
            continue
        resolved_item = dict(item)
        resolved_item["shipping_method_id"] = method_id
        resolved.append(resolved_item)
        warnings.append(f"草稿规则 #{index + 1} 的配送方案已自动绑定为 ID {method_id}。")
    return resolved, warnings


def draft_rule_specs(payloads: list[AdvancedShippingRuleIn]) -> tuple[RuleSpec, ...]:
    """草稿 → 引擎形态。id 用负数：草稿还没有主键，负号让模拟结果里一眼看出"这条是草稿"。"""
    return tuple(
        RuleSpec(
            id=-(index + 1),
            shipping_method_id=p.shipping_method_id,
            delivery_mode=p.delivery_mode,
            quote_group=p.quote_group,
            aggregate_strategy=p.aggregate_strategy,
            presentation=p.presentation.model_dump(exclude_none=True),
            conditions=p.conditions,
            pricing=p.pricing,
            priority=p.priority,
            revision_no=0,
            enabled=bool(p.enabled),
        )
        for index, p in enumerate(payloads)
    )


class AiSampleItem(BaseModel):
    product_id: int
    variant_id: int | None = None
    qty: Decimal = Field(..., gt=0)
    line_total: Decimal = Field(..., ge=0)


class AiSampleCase(BaseModel):
    """模型提出（或操作者补充）的样例购物车。**没有 fee 字段**——运费永远由引擎算。"""
    label: str = Field("", max_length=120)
    address: dict[str, Any] | None = None
    cart_subtotal: Decimal = Field(Decimal("0"), ge=0)
    cart_total: Decimal = Field(Decimal("0"), ge=0)
    delivery_mode: str = "standard"
    items: list[AiSampleItem] = Field(default_factory=list, max_length=20)
    payment_method: str | None = None
    coupon_codes: list[str] = Field(default_factory=list, max_length=10)
    customer_group_ids: list[int] = Field(default_factory=list, max_length=20)


def no_address_case() -> AiSampleCase:
    return AiSampleCase(label=NO_ADDRESS_LABEL, address=None)


def parse_sample_cases(raw: Any) -> tuple[list[AiSampleCase], list[str]]:
    """无地址场景永远排在第一位，模型给的样例接在后面（非法样例只警告、不作废整批）。"""
    cases = [no_address_case()]
    warnings: list[str] = []
    for index, item in enumerate(list(raw or [])[:MAX_SAMPLE_CASES]):
        try:
            case = AiSampleCase(**item)
        except (ValidationError, TypeError) as exc:
            warnings.append(f"样例 #{index + 1} 不合法，已跳过: {exc}")
            continue
        if case.label == NO_ADDRESS_LABEL:
            case = case.model_copy(update={"label": f"{NO_ADDRESS_LABEL}-{index + 1}"})
        cases.append(case)
    return cases, warnings


# ══════════════════════════════════════════════════════════════════════════
#  提示词
# ══════════════════════════════════════════════════════════════════════════

_SYSTEM_PROMPT = """\
你是电商后台的运费规则起草助手。只输出一个 JSON 对象，不要输出解释文字或 markdown。

## 输出契约
{
  "summary": "给运营人员看的本轮自然语言说明（中文，说明改了什么和仍需确认什么）",
  "rules": [ 规则对象, ... ],
  "assumptions": ["你为了产出这批规则所做的假设（中文，逐条）"],
  "questions": ["需要操作者确认才能定稿的问题"],
  "warnings": ["你自己觉得有风险的地方"],
  "sample_cases": [{"label":"场景名","address":{...},"cart_subtotal":"20.00","cart_total":"20.00","delivery_mode":"standard"}]
}

## 规则对象
{
  "shipping_method_id": 必须是下面列出的物流方案 id 之一，必须写 JSON 整数，绝不能写名称、模板 key（例如 zone_delivery）或其他字符串,
  "quote_group": null 或字符串（同组按 aggregate_strategy 合并成一个报价）,
  "aggregate_strategy": "independent|sum|lowest|highest|average|all_required",
  "delivery_mode": "standard|express|pickup",
  "presentation": {"name": {"en-NZ": "英文名", "zh-CN": "中文名"}},
  "conditions": { ... },
  "pricing": { ... },
  "priority": 0-100000 的整数（小的先算）,
  "enabled": true 或 false
}
presentation.name 必须是语言映射对象，**禁止裸字符串**。

## conditions 可用键
- country / province / city / district：{"include": [...]} 或 {"exclude": [...]}
- postcode_patterns：{"include": ["9*", ...]}，最多 %(postcodes)d 条
- subtotal / total / weight_kg / volume_cm3 / unit_quantity / line_count：
  {"min": 数字, "max": 数字}（min 闭、max 开；不需要的边写 null）
- category_ids / product_ids / brand_ids / variant_ids / customer_group_ids：id 数组，最多 %(selectors)d 个
- delivery_mode: ["pickup"] 之类；weekdays: [1..7]；hours: [{"start":9,"end":18}]
- payment_methods / coupon_codes：字符串数组
- **package_count 不支持**，任何情况下都不要使用。

## pricing 可用形状
- {"mode":"flat","amount":"8.50"}
- {"mode":"flat","amount":"0","free_shipping":true}
- {"mode":"inherit"}（沿用核心运费方案算出的价，可再叠 {"modifier":{"add":"6.00"}}）
- {"mode":"tiered_weight","calculation":"single","tiers":[{"start":0,"end":5,"amount":"8.00"}, ...]}
- {"mode":"tiered_dimensional_weight","dimensional_divisor":5000,"tiers":[...]}
- tiers 每条规则最多 %(tiers)d 段；末段 end 用 null 表示不封顶，可带 "block": 5 或 "proportional": true

## 新西兰地址的关键事实（弄错会让每条规则静默错配范围）
运单地址里 city 是大城市名（如 "Auckland"、"Wellington"），
**suburb（郊区）落在 district 字段**，不是 city。
所以"Ponsonby 收 7.5 纽币"必须写成 {"city":{"include":["Auckland"]},"district":{"include":["Ponsonby"]}}，
写成 city.include=["Ponsonby"] 会永远不命中。表格里的 Suburb 列一律映射到 district。

## 起草纪律
- 一张有上百行的报价表，请**按价格档位归并**：同一价格的所有郊区合成一条规则，
  district.include 里列出这些郊区，而不是一行一条规则。整批最多 %(max_rules)d 条。
- 金额一律用字符串，币种是租户本位币 %(currency)s，不要写货币符号、不要换算。
- 时间条件按租户运营时区 %(timezone)s 判定。
- 拿不准的地方写进 questions，不要自己编：编出来的运费是要向真实客户收钱的。
- 新起草的规则一律 "enabled": false，是否启用由操作者决定。
- 你**不能**切换插件模式、不能启用插件、不能写 id 或任何内部标记字段。

## 本租户现有事实
可用物流方案（shipping_method_id: 名称）：
%(methods)s
运费区域：
%(zones)s
商品分类（id: 名称）：
%(categories)s
品牌（id: 名称）：
%(brands)s
现有高级运费规则：
%(rules)s
可参考的内置模板（模板版本 %(template_version)s）：
%(templates)s
"""


def _fmt_pairs(rows: list[tuple[Any, Any]]) -> str:
    if not rows:
        return "  （无）"
    return "\n".join(f"  {a}: {b}" for a, b in rows[:_CONTEXT_LIMIT])


def build_system_prompt(context: dict) -> str:
    return _SYSTEM_PROMPT % {
        "postcodes": MAX_POSTCODE_PATTERNS,
        "selectors": MAX_SELECTOR_IDS,
        "tiers": MAX_TIERS_PER_RULE,
        "max_rules": min(MAX_AI_RULES_PER_DRAFT, MAX_RULES_PER_TENANT),
        "currency": context["currency"],
        "timezone": context["operating_timezone"],
        "methods": _fmt_pairs([(m["id"], m["name"]) for m in context["shipping_methods"]]),
        "zones": _fmt_pairs([(z["id"], f"{z['name']} countries={z['countries']} "
                                      f"provinces={z['province_rules']}")
                             for z in context["zones"]]),
        "categories": _fmt_pairs([(c["id"], c["name"]) for c in context["categories"]]),
        "brands": _fmt_pairs([(b["id"], b["name"]) for b in context["brands"]]),
        "rules": _fmt_pairs([(r["id"], json.dumps(r, ensure_ascii=False))
                             for r in context["existing_rules"]]),
        "template_version": TEMPLATE_VERSION,
        "templates": _fmt_pairs([(k, json.dumps(
            {"conditions": t["conditions"], "pricing": t["pricing"]}, ensure_ascii=False))
            for k, t in TEMPLATES.items()]),
    }


def build_user_prompt(prompt: str, rows: list[list[str]] | None) -> str:
    parts = []
    if prompt:
        parts.append(f"操作者的要求：\n{prompt.strip()[:MAX_PROMPT_CHARS]}")
    if rows:
        parts.append(f"操作者粘贴的报价表（共 {len(rows)} 行，第 1 行是表头，制表符分隔）：\n"
                     f"{table_to_text(rows)}")
    if not parts:
        raise ValueError("请至少提供一段文字要求或一张粘贴的报价表")
    parts.append("请按系统提示的 JSON 契约输出。")
    return "\n\n".join(parts)


def build_revision_prompt(rules: list[AdvancedShippingRuleIn], instruction: str) -> str:
    """把已校验的当前草稿和下一句指令交给模型；不保存会话。"""
    return (
        "当前已审核但尚未应用的完整规则草稿：\n"
        f"{json.dumps([rule.model_dump(mode='json') for rule in rules], ensure_ascii=False)}\n\n"
        f"本轮修改要求：\n{instruction.strip()[:MAX_PROMPT_CHARS]}\n\n"
        "请输出修改后的完整 rules 数组，不要只输出差异；同时给出 summary。"
    )


# ══════════════════════════════════════════════════════════════════════════
#  会话：起草（只读）
# ══════════════════════════════════════════════════════════════════════════

async def _ai_extra(db, tenant_id: int) -> dict:
    from app.core.models.tenant_settings import TenantSettings
    from sqlalchemy import select

    row = (await db.execute(
        select(TenantSettings).where(TenantSettings.tenant_id == tenant_id)
    )).scalar_one_or_none()
    extra = (row.extra or {}) if row else {}
    if not extra.get("ai_enabled"):
        raise HTTPException(status_code=400, detail="AI 功能未启用，请在系统设置 → AI 配置中开启")
    return await resolve_ai_extra(db, tenant_id, extra)


async def load_draft_context(db, tenant_id: int) -> dict:
    """起草只读这些：本租户启用中的物流方案/区域、目录、本位币、运营时区、现有规则。

    一律带 tenant_id 过滤——起草是 AI 输入，跨租户读进去等于把别家的价目表喂给模型。
    """
    from sqlalchemy import select

    from app.core.models.brand import Brand
    from app.core.models.category import Category
    from app.core.models.shipping import ShippingMethod
    from app.core.models.shipping_zone import ShippingZone

    settings = await settings_or_default(db, tenant_id)   # 只读：没有配置行也不建
    methods = (await db.execute(
        select(ShippingMethod)
        .where(ShippingMethod.tenant_id == tenant_id, ShippingMethod.is_active == 1)
        .order_by(ShippingMethod.sort_order, ShippingMethod.id)
    )).scalars().all()
    zones = (await db.execute(
        select(ShippingZone)
        .where(ShippingZone.tenant_id == tenant_id, ShippingZone.is_active == 1)
    )).scalars().all()
    categories = (await db.execute(
        select(Category.id, Category.name)
        .where(Category.tenant_id == tenant_id, Category.is_active == 1)
        .order_by(Category.id).limit(_CONTEXT_LIMIT)
    )).all()
    brands = (await db.execute(
        select(Brand.id, Brand.name)
        .where(Brand.tenant_id == tenant_id, Brand.is_active == 1)
        .order_by(Brand.id).limit(_CONTEXT_LIMIT)
    )).all()
    currency = await svc._base_currency(db, tenant_id)
    existing = await svc.list_rules(db, tenant_id)

    return {
        # settings 原样带出：草稿模拟必须用**租户的**运营时区判定 weekdays/hours，
        # 再查一次不但多一趟 IO，配置行缺失时还会静默退回默认时区（判错时段条件）。
        "settings": settings,
        "currency": currency,
        "operating_timezone": settings.operating_timezone,
        "mode": settings.mode,
        "cache_version": settings.cache_version,
        "shipping_methods": [{"id": m.id, "name": m.name, "zone_id": m.zone_id}
                             for m in methods],
        "zones": [{"id": z.id, "name": z.name, "countries": list(z.countries or []),
                   "province_rules": dict(z.province_rules or {})} for z in zones],
        "categories": [{"id": c_id, "name": name} for c_id, name in categories],
        "brands": [{"id": b_id, "name": name} for b_id, name in brands],
        "existing_rules": [
            {"id": r["id"], "shipping_method_id": r["shipping_method_id"],
             "delivery_mode": r["delivery_mode"], "quote_group": r["quote_group"],
             "aggregate_strategy": r["aggregate_strategy"], "priority": r["priority"],
             "enabled": r["enabled"], "name": (r["presentation"] or {}).get("name")}
            for r in existing[:_CONTEXT_LIMIT]
        ],
    }


async def _simulate_cases(db, tenant_id: int, cases: list[AiSampleCase], *,
                          settings, base_currency: str,
                          draft_specs: tuple[RuleSpec, ...] | None) -> list[dict]:
    """跑模拟。draft_specs 为 None 表示"跑库里已持久化的规则"（应用后的确认）；
    否则把草稿并入当前已启用规则集，看的是"照这份草稿写下去会怎样"。

    走的是与手工模拟完全相同的引擎（evaluate_rule_set）与结果构造（build_simulation_result），
    另起一套只会和结账/手工模拟飘开。
    """
    from app.core.services.shipping_calculator import ShippingCalculator

    rule_set: RuleSet | None = None
    if draft_specs is not None:
        stored = await load_rule_set(db, tenant_id)       # 不走缓存：草稿要看最新的库内事实
        rule_set = RuleSet(rules=stored.rules + draft_specs, hides=stored.hides)

    out: list[dict] = []
    for case in cases:
        if rule_set is None:
            result = await svc.simulate(db, tenant_id, case)
        else:
            facts = await svc.build_sim_facts(db, tenant_id, case, settings=settings,
                                              base_currency=base_currency)
            quote_set = await evaluate_rule_set(
                rule_set, facts,
                inherit_resolver=make_inherit_resolver(ShippingCalculator(db, tenant_id), facts),
            )
            result = svc.build_simulation_result(rule_set, quote_set, facts, mode=settings.mode)
        out.append({"label": case.label or NO_ADDRESS_LABEL, "result": result})
    return out


async def ai_draft(db, tenant_id: int, body) -> dict:
    """起草：读租户事实 → 调模型 → 过手工契约 → 跑模拟 → 返回草稿。**一行都不写库。**"""
    rows: list[list[str]] | None = None
    table_text = getattr(body, "table_text", None)
    if table_text:
        try:
            rows = parse_table_text(table_text)           # 先校验表格，不合法不烧额度
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc

    prompt = (getattr(body, "prompt", "") or "").strip()
    try:
        user_prompt = build_user_prompt(prompt, rows)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

    extra = await _ai_extra(db, tenant_id)
    await consume_ai_quota(db, tenant_id)

    context = await load_draft_context(db, tenant_id)
    raw = await call_ai(
        user_prompt=user_prompt,
        extra=extra,
        system_prompt=build_system_prompt(context),
        max_tokens=6144,
        timeout=110,
    )

    try:
        data = extract_json_object(raw)
    except ValueError as exc:
        raise HTTPException(
            status_code=422, detail=f"AI 返回格式无法解析：{exc}。原始回复：{(raw or '')[:300]}"
        ) from exc

    try:
        raw_rules, method_warnings = resolve_draft_shipping_method_ids(
            data.get("rules"), context["shipping_methods"])
        valid, invalid = validate_draft_rules(raw_rules)
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=f"AI 返回的规则不可用：{exc}") from exc

    cases, case_warnings = parse_sample_cases(data.get("sample_cases"))
    warnings = [str(w) for w in (data.get("warnings") or [])] + method_warnings + case_warnings
    if any(p.hide_rule_ids for p in valid):
        warnings.append("草稿里含隐藏边，隐藏边指向的是已存在的规则 id，草稿模拟不会体现它们")
    if invalid:
        warnings.append(f"有 {len(invalid)} 条规则不满足规则契约，已排除在草稿之外")

    simulations = await _simulate_cases(
        db, tenant_id, cases, settings=context["settings"], base_currency=context["currency"],
        draft_specs=draft_rule_specs(valid),
    )

    return {
        "summary": str(data.get("summary") or f"已生成 {len(valid)} 条待审核运费规则。"),
        "rules": [p.model_dump(mode="json") for p in valid],
        "invalid_rules": invalid,
        "assumptions": [str(a) for a in (data.get("assumptions") or [])],
        "questions": [str(q) for q in (data.get("questions") or [])],
        "warnings": warnings,
        "sample_cases": [c.model_dump(mode="json") for c in cases],
        "simulations": simulations,
        "table_row_count": len(rows) if rows else 0,
        "currency": context["currency"],
        "operating_timezone": context["operating_timezone"],
        "mode": context["mode"],
        "persisted": False,
    }


async def ai_revise(db, tenant_id: int, body) -> dict:
    """对内存中的已审核草稿进行一轮替代式修订；绝不写运费配置。"""
    instruction = (getattr(body, "instruction", "") or "").strip()
    if not instruction:
        raise HTTPException(status_code=400, detail="请填写本轮修改要求")
    try:
        current, invalid_current = validate_draft_rules(list(getattr(body, "rules", []) or []))
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=f"当前草稿不可用：{exc}") from exc
    if invalid_current or not current:
        raise HTTPException(status_code=400, detail="当前草稿包含无效规则，不能继续修改")

    extra = await _ai_extra(db, tenant_id)
    await consume_ai_quota(db, tenant_id)
    context = await load_draft_context(db, tenant_id)
    raw = await call_ai(
        user_prompt=build_revision_prompt(current, instruction),
        extra=extra,
        system_prompt=build_system_prompt(context),
        max_tokens=6144,
        timeout=110,
    )
    try:
        data = extract_json_object(raw)
        raw_rules, method_warnings = resolve_draft_shipping_method_ids(
            data.get("rules"), context["shipping_methods"])
        valid, invalid = validate_draft_rules(raw_rules)
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=f"AI 返回的修订草稿不可用：{exc}") from exc

    cases, case_warnings = parse_sample_cases(
        getattr(body, "sample_cases", None) or data.get("sample_cases"))
    warnings = [str(w) for w in (data.get("warnings") or [])] + method_warnings + case_warnings
    if invalid:
        warnings.append(f"有 {len(invalid)} 条规则不满足规则契约，未进入当前草稿")
    if not valid:
        raise HTTPException(status_code=422, detail="AI 本轮没有生成可用规则；已保留上一版草稿")
    simulations = await _simulate_cases(
        db, tenant_id, cases, settings=context["settings"], base_currency=context["currency"],
        draft_specs=draft_rule_specs(valid),
    )
    return {
        "summary": str(data.get("summary") or "已按本轮要求更新草稿，请审核规则与模拟结果。"),
        "rules": [p.model_dump(mode="json") for p in valid],
        "invalid_rules": invalid,
        "assumptions": [str(a) for a in (data.get("assumptions") or [])],
        "questions": [str(q) for q in (data.get("questions") or [])],
        "warnings": warnings,
        "sample_cases": [c.model_dump(mode="json") for c in cases],
        "simulations": simulations,
        "currency": context["currency"], "operating_timezone": context["operating_timezone"],
        "mode": context["mode"], "persisted": False,
    }


# ══════════════════════════════════════════════════════════════════════════
#  会话：应用（写库，走既有写路径）
# ══════════════════════════════════════════════════════════════════════════

async def ai_apply(db, tenant_id: int, body, *, user_id: int | None = None) -> dict:
    """把**操作者审核过的**规则整批落库，然后回一份写入后的模拟。

    - 不再调模型：审核的是请求体里的这几条，重新问一次可能得到别的东西。
    - 先把整批过一遍契约，再逐条 _insert_rule（继承归属/配额/聚合不变式/修订）。
    - 全程只在末尾 _finish 一次：中途任何一条抛异常都不会 commit，事务整体回滚。
    """
    note_suffix = (getattr(body, "change_note", None) or "").strip()

    try:
        payloads, invalid = validate_draft_rules(list(getattr(body, "rules", []) or []))
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    if invalid:
        raise HTTPException(
            status_code=400,
            detail=f"规则 {[i['index'] for i in invalid]} 不满足规则契约：{invalid[0]['error']}")
    if not payloads:
        raise HTTPException(status_code=400, detail="没有可应用的规则")

    settings = await svc.ensure_settings(db, tenant_id)
    created: list[int] = []
    for index, payload in enumerate(payloads):
        note = f"ai apply #{index + 1}" + (f": {note_suffix}" if note_suffix else "")
        rule = await svc._insert_rule(db, tenant_id, payload, user_id=user_id, note=note[:500])
        created.append(rule.id)
    await svc._finish(db, tenant_id, settings)     # 旧 cache_version 失效 + 唯一一次 commit

    cases, _warnings = parse_sample_cases(
        [c.model_dump() if isinstance(c, BaseModel) else c
         for c in (getattr(body, "sample_cases", None) or [])])
    simulations = await _simulate_cases(
        db, tenant_id, cases, settings=settings,
        base_currency=await svc._base_currency(db, tenant_id), draft_specs=None)

    # mode 原样回显：AI 路径从不改它，操作者要切强制得走 PUT /settings 并先确认两种模拟。
    return {"created": created, "mode": settings.mode, "simulations": simulations}


__all__ = [
    "MAX_AI_RULES_PER_DRAFT", "MAX_SAMPLE_CASES", "MAX_TABLE_BYTES", "MAX_TABLE_COLUMNS",
    "MAX_TABLE_ROWS", "NO_ADDRESS_LABEL", "AiSampleCase", "AiSampleItem",
    "ai_apply", "ai_draft", "ai_revise", "build_revision_prompt", "build_system_prompt", "build_user_prompt", "draft_rule_specs",
    "extract_json_object", "load_draft_context", "no_address_case", "parse_sample_cases",
    "parse_table_text", "table_to_text", "validate_draft_rules",
]