"""高级运费规则：Admin 服务层（CRUD、修订/回滚、模拟、legacy 转换、模式护栏）。

router.py 的每个路由体只做一件事——调用本模块的一个函数。这样做不只是好看：
本仓库的测试不起 DB、也不起 TestClient，逻辑留在路由体里就只能靠"断言源码文本"来测，
而那种测试在守卫被删掉之后依然是绿的（曾经真的发生过）。

── 硬规则 ──────────────────────────────────────────────────────────────────
1. 任何写路径都要：校验全部 ID 属于本租户 → 校验聚合不变式 → 同一事务里写修订
   → invalidate_rule_set(旧 cache_version) → commit。少一步就是脏数据或最长 300 秒的旧运费。
2. cache_version 只能由 SQL 自增，绝不接受客户端传来的值。
3. 金额一律租户本位币，且**从不接受客户端传来的运费**——模拟只收购物车事实。
4. 模型统一在函数体内从 app.core.models 懒加载：app.core.models 会反向 import 本插件的
   models，模块级 `from .models import ...` 在"先导入本插件"的场景下直接循环导入
   （同 cache.py / audit.py / plugins/tax/calculator.py）。
"""
from __future__ import annotations

import logging
from decimal import Decimal
from typing import Any, Iterable, Mapping, Sequence

from fastapi import HTTPException
from pydantic import ValidationError
from sqlalchemy import func, select
from sqlalchemy.orm import selectinload

from .cache import invalidate_rule_set, load_settings, settings_or_default
from .quote import RuleSet, build_quote_snapshot, canonical_strategy, quote_facts
from .schemas import AdvancedShippingRuleIn, check_tenant_rule_count
from .services import build_cart_facts, get_tenant_default_locale, resolve_locale_text

logger = logging.getLogger(__name__)

#: 强制模式的前置确认：必须各有一次"无地址"与"代表性地址"的成功模拟被确认。
ACK_NO_ADDRESS = "simulation_ack_no_address"
ACK_REPRESENTATIVE = "simulation_ack_representative"
REQUIRED_ACKS: tuple[str, ...] = (ACK_NO_ADDRESS, ACK_REPRESENTATIVE)

#: 软删除在最新修订上留下的记号。删除只是 enabled=0，靠它把"被删的"和"手动停用的"分开。
DELETED_NOTE = "deleted"

#: legacy 转换的幂等标记列。**刻意不放在 pricing JSON 里**：pricing 是客户端可写字段，
#: 一次 PUT /rules/{id} 省掉这个 key 就能把标记抹掉，下次转换会重复建规则、白烧配额。
#: _apply() 从不写这一列，所以任何客户端请求都改不到它。
LEGACY_CONVERSION_COLUMN = "from_legacy_conversion"

#: 转换预览必须原样带给操作者的告知 key（前端按 key 渲染 i18n 文案）。
NOTE_INHERIT_INCLUDES_SURCHARGES = "inherit_fee_already_includes_core_surcharges"
NOTE_CONVERSION_CREATES_DISABLED = "conversion_creates_disabled_rules_only"
NOTE_CONVERSION_DOES_NOT_ENFORCE = "conversion_does_not_switch_enforcement"


# ══════════════════════════════════════════════════════════════════════════
#  纯函数：可以脱离会话直接测
# ══════════════════════════════════════════════════════════════════════════

def assert_group_strategy(payload, siblings: Sequence[Any], *, exclude_rule_id: int | None = None) -> None:
    """同一 (shipping_method_id, delivery_mode, quote_group) 组内不许混聚合策略。

    引擎解析一组时用的是"优先级最高的成员的策略"（quote._group_key/_strategy），
    混策略不会报错，只会按其中一条静默定价——那是算错钱，比报错糟。所以在写库层拒掉。
    """
    if not payload.quote_group:
        return
    want = canonical_strategy(payload.aggregate_strategy)
    for row in siblings:
        if exclude_rule_id is not None and row.id == exclude_rule_id:
            continue
        if canonical_strategy(row.aggregate_strategy) != want:
            raise HTTPException(
                status_code=400,
                detail=(f"报价分组 {payload.quote_group} 内已存在聚合策略 "
                        f"{row.aggregate_strategy}（规则 #{row.id}），不允许混用 "
                        f"{payload.aggregate_strategy}"),
            )


def assert_mode_transition(current_mode: str, new_mode: str, *, acked: Iterable[str]) -> None:
    """切 enforced 前必须已确认过"无地址"与"代表性地址"两种成功模拟；回退 shadow 立即生效。"""
    if new_mode != "enforced" or current_mode == "enforced":
        return
    missing = [k for k in REQUIRED_ACKS if k not in set(acked)]
    if missing:
        raise HTTPException(
            status_code=400,
            detail=f"切换到强制模式前必须先确认模拟结果，缺少: {', '.join(missing)}",
        )


def assert_ack_kind(kind: str, address: Mapping[str, Any] | None) -> None:
    """无地址确认不许带地址，代表性地址确认必须带地址——否则两个 ack 可以用同一次模拟凑齐。"""
    if kind == "no_address" and address:
        raise HTTPException(status_code=400, detail="no_address 确认不允许携带地址")
    if kind == "representative" and not address:
        raise HTTPException(status_code=400, detail="representative 确认必须携带代表性地址")


def assert_ack_result(kind: str, result: Mapping[str, Any]) -> None:
    """代表性地址必须真的报出价来；无地址场景本就可能无报价，跑通即可确认。"""
    if kind == "representative" and not result.get("quotes"):
        raise HTTPException(
            status_code=400,
            detail="代表性地址模拟没有任何报价，强制模式会让该地址结不了账",
        )


def rule_display_names(rows: Sequence[tuple[Any, int]]) -> list[str]:
    """(presentation, rule_id) → 可读名。没有文案时回落成 #id，绝不返回空串。"""
    names = []
    for presentation, rule_id in rows:
        name = resolve_locale_text((presentation or {}).get("name"), None, None)
        names.append(name or f"#{rule_id}")
    return names


def rule_snapshot(rule) -> dict:
    """规则的完整字段快照（修订表存的就是它，也是回滚的输入）。"""
    return {
        "shipping_method_id": rule.shipping_method_id,
        "quote_group": rule.quote_group,
        "aggregate_strategy": rule.aggregate_strategy,
        "delivery_mode": rule.delivery_mode,
        "presentation": rule.presentation or {},
        "conditions": rule.conditions or {},
        "pricing": rule.pricing or {},
        "priority": rule.priority,
        "enabled": bool(rule.enabled),
    }


def _zone_conditions(zone) -> tuple[dict | None, str]:
    """区域 → 条件。返回 (conditions, reason)；conditions 为 None 表示表达不出来。

    **legacy 的区域匹配是 OR，条件 DSL 是 AND**（shipping_calculator.py:304-320）：
    province_rules 命中**或** countries 命中都算命中。所以
    countries=["NZ"] + province_rules={"NZ":["Auckland"]} 在 legacy 里覆盖整个 NZ，
    照抄成 country ∧ province 只覆盖 Auckland——强制模式下所有非 Auckland 的 NZ 客户
    直接结不了账。这种 OR 形状一律拒绝转换，让操作者自己决定要哪一边。
    """
    if zone is None or zone.is_all_countries:
        return {}, ""
    countries = [str(c) for c in (zone.countries or [])]
    province_rules = {str(k): list(v or []) for k, v in (zone.province_rules or {}).items()}

    if countries and province_rules:
        return None, (
            f"区域同时配了 countries={countries} 与 province_rules={sorted(province_rules)}，"
            "legacy 的匹配是两者取或（整个国家都覆盖），条件规则是取且（只覆盖列出的省份）。"
            "请先把区域拆成「按国家」或「按省份」其中一种，再转换")
    if countries:
        # 只有国家：国家级规则天然覆盖它的所有省份，与 legacy 等价
        return {"country": {"include": countries}}, ""
    if len(province_rules) == 1:
        country, provinces = next(iter(province_rules.items()))
        if provinces:
            return {"country": {"include": [country]},
                    "province": {"include": sorted({str(p) for p in provinces})}}, ""
        return None, f"区域的 province_rules[{country}] 是空列表，legacy 永不命中此区域"
    if len(province_rules) > 1:
        return None, (
            f"区域的 province_rules 覆盖多个国家 {sorted(province_rules)}，"
            "摊平成 country ∧ province 会得到国家×省份的笛卡尔积（比 legacy 更宽）。"
            "请按国家拆分区域后再转换")
    return None, "区域既没有 countries 也没有 province_rules，legacy 永不命中此区域"


def _method_warnings(method) -> list[str]:
    """inherit 委派给 calculate_for_order，它**不**做可用性过滤——丢掉的门槛要显式说出来。"""
    warnings: list[str] = []
    if method.min_order_amount:
        warnings.append(
            f"min_order_amount={method.min_order_amount} 只在 legacy 的可用方案筛选里生效，"
            "inherit 不会再过滤，请补一条 subtotal.min 条件")
    if method.max_weight:
        warnings.append(
            f"max_weight={method.max_weight} 只在 legacy 的可用方案筛选里生效，"
            "inherit 不会再过滤，请补一条 weight_kg.max 条件")
    return warnings


def plan_legacy_conversion(*, methods: Sequence[Any], zones: Sequence[Any],
                           existing_rules: Sequence[Any], locale: str) -> dict:
    """幂等的转换计划：现行运费方案/区域 → 每个方案一条 inherit 规则。

    - 只看 is_active=1 的方案；区域缺失或已停用的方案落进 unmappable（覆盖范围表达不出来）。
    - 上次转换产出的规则靠 from_legacy_conversion **列**认出来 → already_converted，
      重跑不会产生第二条。人工手写的 inherit 规则没有这个标记，不会被误当成已转换。
      标记在列上而不在 pricing JSON 里，客户端改不到（见 LEGACY_CONVERSION_COLUMN）。
    - 产出的规则一律 enabled=False：转换只生成待复核草稿，绝不切换计费。
    """
    zones_by_id = {z.id: z for z in zones}
    converted_by_method: dict[int, int] = {}
    for rule in existing_rules:
        if getattr(rule, LEGACY_CONVERSION_COLUMN, 0):
            converted_by_method.setdefault(rule.shipping_method_id, rule.id)

    convertible: list[dict] = []
    already: list[dict] = []
    unmappable: list[dict] = []

    for method in methods:
        if not method.is_active:
            continue
        if method.id in converted_by_method:
            already.append({"shipping_method_id": method.id,
                            "rule_id": converted_by_method[method.id]})
            continue
        zone = None
        if method.zone_id is not None:
            zone = zones_by_id.get(method.zone_id)
            if zone is None or not zone.is_active:
                unmappable.append({
                    "shipping_method_id": method.id,
                    "method_name": method.name,
                    "reasons": [f"运费区域 {method.zone_id} 不存在或已停用，覆盖范围无法表达为条件"],
                })
                continue
        conditions, reason = _zone_conditions(zone)
        if conditions is None:
            unmappable.append({"shipping_method_id": method.id,
                               "method_name": method.name, "reasons": [reason]})
            continue
        convertible.append({
            "shipping_method_id": method.id,
            "method_name": method.name,
            "zone_id": method.zone_id,
            "warnings": _method_warnings(method),
            "rule": {
                "shipping_method_id": method.id,
                "quote_group": None,
                "aggregate_strategy": "independent",
                "delivery_mode": "standard",
                "presentation": {"name": {locale: method.name}},
                "conditions": conditions,
                "pricing": {"mode": "inherit"},
                "priority": 100,
                "enabled": False,
            },
        })

    return {
        "notes": [
            NOTE_INHERIT_INCLUDES_SURCHARGES,
            NOTE_CONVERSION_CREATES_DISABLED,
            NOTE_CONVERSION_DOES_NOT_ENFORCE,
        ],
        "convertible": convertible,
        "already_converted": already,
        "unmappable": unmappable,
    }


def build_simulation_result(rule_set: RuleSet, quote_set, facts, *, mode: str) -> dict:
    """模拟结果：全部候选报价 + 被拒原因 key + 无覆盖诊断 + 聚合拆解 + 币种 + 优惠策略。

    聚合拆解给的是"这组由哪些规则、按什么策略合出来的"。逐成员金额引擎不外传
    （RuleQuote 只带合计），需要的话得改引擎——不在本任务范围。
    """
    strategies = {r.id: r.aggregate_strategy for r in rule_set.rules}
    priorities = {r.id: r.priority for r in rule_set.rules}

    quotes = []
    for quote in quote_set.quotes:
        snapshot = build_quote_snapshot(quote, facts, mode=mode)
        snapshot["fee"] = str(quote.fee)
        snapshot["priority"] = quote.priority
        snapshot["description_map"] = dict(quote.description_map)
        snapshot["aggregate"] = {
            "strategy": strategies.get(quote.rule_ids[0], "independent") if quote.rule_ids else "independent",
            "members": [
                {"rule_id": rid, "revision_no": rev, "priority": priorities.get(rid)}
                for rid, rev in zip(quote.rule_ids, quote.revision_ids)
            ],
        }
        quotes.append(snapshot)

    reason_keys = list(quote_set.reason_keys)
    no_coverage = not quote_set.quotes
    return {
        "mode": mode,
        "currency": facts.base_currency,
        "delivery_mode": facts.delivery_mode,
        "operating_timezone": facts.operating_timezone,
        "quoted_at": facts.local_now.isoformat(),
        "quotes": quotes,
        "rejected": [
            {
                "shipping_method_id": q.shipping_method_id,
                "quote_group": q.quote_group,
                "delivery_mode": q.delivery_mode,
                "rule_ids": list(q.rule_ids),
                "reason_keys": list(q.reason_keys),
            }
            for q in quote_set.rejected
        ],
        "reason_keys": reason_keys,
        "no_coverage": no_coverage,
        "no_coverage_diagnostic": None if not no_coverage else {
            "enabled_rule_count": len(rule_set.rules),
            "delivery_mode": facts.delivery_mode,
            "reason_keys": reason_keys,
            "hint": "强制模式下没有兜底：没有任何规则出价就是结不了账",
        },
        "promotion_policy": {
            # pricing.free_shipping 命中时引擎把 fee 归零并打上 promo_free_shipping，
            # 不叠加、不再走 modifier/floor/cap。
            "free_shipping_zeroes_fee": True,
            "free_shipping_quotes": [
                list(q.rule_ids) for q in quote_set.quotes if q.promo_free_shipping
            ],
            "coupons_are_conditions_only": True,
        },
    }


# ══════════════════════════════════════════════════════════════════════════
#  会话辅助
# ══════════════════════════════════════════════════════════════════════════

async def ensure_settings(db, tenant_id: int):
    """读租户配置；没有就按 shadow 建一行（首次启用必须落在影子模式）。"""
    from app.core.models import AdvancedShippingSettings

    settings = await load_settings(db, tenant_id)
    if settings is None:
        settings = AdvancedShippingSettings(
            tenant_id=tenant_id, mode="shadow",
            operating_timezone="Pacific/Auckland", cache_version=1,
        )
        db.add(settings)
        await db.flush()
    return settings


async def _finish(db, tenant_id: int, settings) -> None:
    """收尾：同一事务里递增 cache_version 并删旧 key，然后提交。

    必须传**旧** cache_version：invalidate_rule_set 删的是旧 key，
    传新的等于没删，租户会看到最长 300 秒（CACHE_TTL）的旧运费。
    """
    await invalidate_rule_set(db, tenant_id, cache_version=settings.cache_version)
    await db.commit()


async def _owned_ids(db, model, tenant_id: int, ids: Iterable[int]) -> set[int]:
    ids = [int(i) for i in ids]
    if not ids:
        return set()
    rows = await db.execute(
        select(model.id).where(model.id.in_(ids), model.tenant_id == tenant_id)
    )
    return set(rows.scalars().all())


async def _assert_owned_method(db, tenant_id: int, method_id: int) -> None:
    from app.core.models.shipping import ShippingMethod

    if method_id not in await _owned_ids(db, ShippingMethod, tenant_id, [method_id]):
        raise HTTPException(status_code=404, detail=f"运费方案 {method_id} 不存在或不属于本租户")


def _hide_targets(payload, rule_id: int | None = None) -> list[int]:
    """去重 + 拒绝自隐藏。

    - 唯一约束是 (tenant_id, rule_id, hidden_rule_id)，[12, 12] 原样插入 = commit 时 500。
    - 自隐藏会让规则一命中就把自己藏掉（quote.py 的 hidden 集合），永远不出价。
    """
    ids = sorted({int(i) for i in payload.hide_rule_ids})
    if rule_id is not None and rule_id in ids:
        raise HTTPException(status_code=400, detail="规则不能隐藏自己")
    return ids


async def _assert_owned_rules(db, tenant_id: int, rule_ids: Sequence[int]) -> None:
    from app.core.models import AdvancedShippingRule

    if not rule_ids:
        return
    owned = await _owned_ids(db, AdvancedShippingRule, tenant_id, rule_ids)
    missing = sorted(set(int(i) for i in rule_ids) - owned)
    if missing:
        raise HTTPException(status_code=404, detail=f"规则 {missing} 不存在或不属于本租户")


async def _get_rule(db, tenant_id: int, rule_id: int):
    from app.core.models import AdvancedShippingRule

    row = (await db.execute(
        select(AdvancedShippingRule).where(
            AdvancedShippingRule.id == rule_id,
            AdvancedShippingRule.tenant_id == tenant_id,
        )
    )).scalar_one_or_none()
    if row is None:
        raise HTTPException(status_code=404, detail=f"规则 {rule_id} 不存在或不属于本租户")
    return row


async def _group_siblings(db, tenant_id: int, payload):
    from app.core.models import AdvancedShippingRule

    if not payload.quote_group:
        return []
    rows = await db.execute(
        select(AdvancedShippingRule).where(
            AdvancedShippingRule.tenant_id == tenant_id,
            AdvancedShippingRule.shipping_method_id == payload.shipping_method_id,
            AdvancedShippingRule.delivery_mode == payload.delivery_mode,
            AdvancedShippingRule.quote_group == payload.quote_group,
        )
    )
    return list(rows.scalars().all())


async def _rule_count(db, tenant_id: int) -> int:
    from app.core.models import AdvancedShippingRule

    result = await db.execute(
        select(func.count()).select_from(AdvancedShippingRule)
        .where(AdvancedShippingRule.tenant_id == tenant_id)
    )
    return int(result.scalar() or 0)


def _add_revision(db, tenant_id: int, rule, *, revision_no: int,
                  user_id: int | None, note: str | None) -> None:
    from app.core.models import AdvancedShippingRuleRevision

    db.add(AdvancedShippingRuleRevision(
        tenant_id=tenant_id, rule_id=rule.id, revision_no=revision_no,
        snapshot=rule_snapshot(rule), changed_by=user_id, change_note=note,
    ))


def _rule_out(rule, hide_rule_ids: Sequence[int] = (), *, retired: bool = False) -> dict:
    data = rule_snapshot(rule)
    data.update(id=rule.id, current_revision_no=rule.current_revision_no,
                hide_rule_ids=list(hide_rule_ids), retired=retired,
                from_legacy_conversion=bool(getattr(rule, LEGACY_CONVERSION_COLUMN, 0)))
    return data


async def _hide_ids(db, tenant_id: int, rule_id: int) -> list[int]:
    from app.core.models import AdvancedShippingRuleHide

    rows = await db.execute(
        select(AdvancedShippingRuleHide.hidden_rule_id).where(
            AdvancedShippingRuleHide.tenant_id == tenant_id,
            AdvancedShippingRuleHide.rule_id == rule_id,
        )
    )
    return sorted(rows.scalars().all())


def _apply(rule, payload) -> None:
    rule.shipping_method_id = payload.shipping_method_id
    rule.quote_group = payload.quote_group
    rule.aggregate_strategy = payload.aggregate_strategy
    rule.delivery_mode = payload.delivery_mode
    rule.presentation = payload.presentation.model_dump(exclude_none=True)
    rule.conditions = payload.conditions
    rule.pricing = payload.pricing
    rule.priority = payload.priority
    rule.enabled = int(payload.enabled)


# ══════════════════════════════════════════════════════════════════════════
#  规则 CRUD
# ══════════════════════════════════════════════════════════════════════════

async def list_rules(db, tenant_id: int) -> list[dict]:
    """列出规则。已删除（软停用）的规则带 retired=True。

    删除只是 enabled=0（修订表 RESTRICT，硬删会 500），所以"被删掉的"和"手动停用的"
    在库里长得一模一样。用最新修订的 change_note 区分，不为此再加一列。
    """
    from app.core.models import (
        AdvancedShippingRule, AdvancedShippingRuleHide, AdvancedShippingRuleRevision,
    )

    rules = (await db.execute(
        select(AdvancedShippingRule)
        .where(AdvancedShippingRule.tenant_id == tenant_id)
        .order_by(AdvancedShippingRule.priority, AdvancedShippingRule.id)
    )).scalars().all()
    hides = (await db.execute(
        select(AdvancedShippingRuleHide)
        .where(AdvancedShippingRuleHide.tenant_id == tenant_id)
    )).scalars().all()
    by_rule: dict[int, list[int]] = {}
    for hide in hides:
        by_rule.setdefault(hide.rule_id, []).append(hide.hidden_rule_id)

    retired: set[int] = set()
    disabled = [(r.id, r.current_revision_no) for r in rules if not r.enabled]
    if disabled:
        notes = await db.execute(
            select(AdvancedShippingRuleRevision.rule_id,
                   AdvancedShippingRuleRevision.change_note)
            .where(AdvancedShippingRuleRevision.tenant_id == tenant_id,
                   AdvancedShippingRuleRevision.rule_id.in_([i for i, _ in disabled]))
        )
        latest = {rid: rev for rid, rev in disabled}
        for rid, note in notes.all():
            if note == DELETED_NOTE and rid in latest:
                retired.add(rid)
    return [
        _rule_out(r, sorted(by_rule.get(r.id, [])), retired=r.id in retired)
        for r in rules
    ]


async def create_rule(db, tenant_id: int, payload: AdvancedShippingRuleIn, *,
                      user_id: int | None = None, note: str | None = None) -> dict:
    settings = await ensure_settings(db, tenant_id)
    rule = await _insert_rule(db, tenant_id, payload, user_id=user_id, note=note)
    await _finish(db, tenant_id, settings)
    # 去重后的边才是库里真实写入的那份，回传原始 payload 会和后续 list 的结果对不上
    return _rule_out(rule, _hide_targets(payload))


async def _insert_rule(db, tenant_id: int, payload: AdvancedShippingRuleIn, *,
                       user_id: int | None, note: str | None,
                       from_legacy_conversion: bool = False):
    """建一条规则 + 首个修订 + 隐藏边。不提交、不失效缓存——由调用方收尾。"""
    from app.core.models import AdvancedShippingRule, AdvancedShippingRuleHide

    await _assert_owned_method(db, tenant_id, payload.shipping_method_id)
    try:
        check_tenant_rule_count(await _rule_count(db, tenant_id))
    except ValueError as exc:                    # 配额是 400，不是 500
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    hide_ids = _hide_targets(payload)
    await _assert_owned_rules(db, tenant_id, hide_ids)
    assert_group_strategy(payload, await _group_siblings(db, tenant_id, payload))

    rule = AdvancedShippingRule(tenant_id=tenant_id, current_revision_no=1,
                                from_legacy_conversion=int(from_legacy_conversion))
    _apply(rule, payload)
    db.add(rule)
    await db.flush()

    _add_revision(db, tenant_id, rule, revision_no=1, user_id=user_id, note=note)
    for hidden_id in hide_ids:
        db.add(AdvancedShippingRuleHide(
            tenant_id=tenant_id, rule_id=rule.id, hidden_rule_id=hidden_id))
    return rule


async def update_rule(db, tenant_id: int, rule_id: int, payload: AdvancedShippingRuleIn, *,
                      user_id: int | None = None, note: str | None = None) -> dict:
    from sqlalchemy import delete as sa_delete

    from app.core.models import AdvancedShippingRuleHide

    settings = await ensure_settings(db, tenant_id)
    rule = await _get_rule(db, tenant_id, rule_id)
    await _assert_owned_method(db, tenant_id, payload.shipping_method_id)
    hide_ids = _hide_targets(payload, rule_id)
    await _assert_owned_rules(db, tenant_id, hide_ids)
    assert_group_strategy(payload, await _group_siblings(db, tenant_id, payload),
                          exclude_rule_id=rule_id)

    _apply(rule, payload)
    rule.current_revision_no = rule.current_revision_no + 1
    _add_revision(db, tenant_id, rule, revision_no=rule.current_revision_no,
                  user_id=user_id, note=note)

    await db.execute(sa_delete(AdvancedShippingRuleHide).where(
        AdvancedShippingRuleHide.tenant_id == tenant_id,
        AdvancedShippingRuleHide.rule_id == rule_id,
    ))
    for hidden_id in hide_ids:
        db.add(AdvancedShippingRuleHide(
            tenant_id=tenant_id, rule_id=rule_id, hidden_rule_id=hidden_id))

    await _finish(db, tenant_id, settings)
    return _rule_out(rule, hide_ids)


async def delete_rule(db, tenant_id: int, rule_id: int, *, user_id: int | None = None) -> dict:
    """删除 = 停用 + 记一条修订。

    ponytail: 修订表是 RESTRICT 外键（审计轨迹不可随规则消失），硬删规则必然撞 FK 报 500。
    上限是"规则行永远留在 200 条配额里"；真需要彻底清理，得先设计修订归档再改成硬删。
    """
    settings = await ensure_settings(db, tenant_id)
    rule = await _get_rule(db, tenant_id, rule_id)
    rule.enabled = 0
    rule.current_revision_no = rule.current_revision_no + 1
    _add_revision(db, tenant_id, rule, revision_no=rule.current_revision_no,
                  user_id=user_id, note=DELETED_NOTE)
    hide_ids = await _hide_ids(db, tenant_id, rule_id)
    await _finish(db, tenant_id, settings)
    return _rule_out(rule, hide_ids, retired=True)


# ══════════════════════════════════════════════════════════════════════════
#  修订与回滚
# ══════════════════════════════════════════════════════════════════════════

async def list_revisions(db, tenant_id: int, rule_id: int, *,
                         limit: int = 50, offset: int = 0) -> list[dict]:
    """修订只增不删（delete_rule 还会再加一条），所以必须分页：
    一条每天编辑的规则两年后会在单个响应里吐出七百份完整 JSON 快照。"""
    from app.core.models import AdvancedShippingRuleRevision

    rows = (await db.execute(
        select(AdvancedShippingRuleRevision)
        .where(AdvancedShippingRuleRevision.tenant_id == tenant_id,
               AdvancedShippingRuleRevision.rule_id == rule_id)
        .order_by(AdvancedShippingRuleRevision.revision_no.desc())
        .limit(min(max(int(limit), 1), 200)).offset(max(int(offset), 0))
    )).scalars().all()
    return [
        {"revision_no": r.revision_no, "snapshot": r.snapshot, "changed_by": r.changed_by,
         "change_note": r.change_note,
         "created_at": r.created_at.isoformat() if r.created_at else None}
        for r in rows
    ]


async def rollback_rule(db, tenant_id: int, rule_id: int, revision_no: int, *,
                        user_id: int | None = None) -> dict:
    """回滚 = 用旧快照写一条**新**修订，不改也不删任何历史修订。

    **隐藏边不回滚**：rule_snapshot 不含 hide_rule_ids。隐藏边的另一端可能已经被删除
    （hides 是 CASCADE），照抄旧边会指向不存在的规则。回滚只还原规则本体，
    隐藏边请在规则列表里单独调整。from_legacy_conversion 列同样不受回滚影响。
    """
    from app.core.models import AdvancedShippingRuleRevision

    settings = await ensure_settings(db, tenant_id)
    rule = await _get_rule(db, tenant_id, rule_id)
    revision = (await db.execute(
        select(AdvancedShippingRuleRevision).where(
            AdvancedShippingRuleRevision.tenant_id == tenant_id,
            AdvancedShippingRuleRevision.rule_id == rule_id,
            AdvancedShippingRuleRevision.revision_no == revision_no,
        )
    )).scalar_one_or_none()
    if revision is None:
        raise HTTPException(status_code=404, detail=f"规则 {rule_id} 没有修订 {revision_no}")

    # 老快照可能不再满足当前契约（字段收紧过），必须重新过一遍校验再落库
    try:
        payload = AdvancedShippingRuleIn(**(revision.snapshot or {}))
    except ValidationError as exc:
        raise HTTPException(status_code=400, detail=f"修订 {revision_no} 不满足当前规则契约: {exc}")

    # 方案可能已被删/转给别的租户，回滚同样要重新校验归属与聚合不变式
    await _assert_owned_method(db, tenant_id, payload.shipping_method_id)
    assert_group_strategy(payload, await _group_siblings(db, tenant_id, payload),
                          exclude_rule_id=rule_id)

    _apply(rule, payload)
    rule.current_revision_no = rule.current_revision_no + 1
    _add_revision(db, tenant_id, rule, revision_no=rule.current_revision_no,
                  user_id=user_id, note=f"rollback to revision {revision_no}")
    await _finish(db, tenant_id, settings)
    return _rule_out(rule)


# ══════════════════════════════════════════════════════════════════════════
#  配置、影子事件、模板
# ══════════════════════════════════════════════════════════════════════════

async def read_settings(db, tenant_id: int) -> dict:
    """GET 不写库：没有配置行就直接返回默认值（首次启用固定 shadow），不建行也不 commit。"""
    settings = await load_settings(db, tenant_id)
    if settings is None:
        return {"mode": "shadow", "operating_timezone": "Pacific/Auckland",
                "cache_version": 1, "acknowledged_simulations": [],
                "enforcement_ready": False}
    acked = await _acknowledged(db, tenant_id, cache_version=settings.cache_version)
    return {
        "mode": settings.mode,
        "operating_timezone": settings.operating_timezone,
        "cache_version": settings.cache_version,
        "acknowledged_simulations": sorted(acked),
        "enforcement_ready": all(k in acked for k in REQUIRED_ACKS),
    }


async def _acknowledged(db, tenant_id: int, *, cache_version: int | None = None) -> set[str]:
    """已确认的模拟种类。

    传了 cache_version 就只认**当前**规则版本下确认的那些：cache_version 在每次规则/配置
    写入时自增，所以"确认完再加一条规则把无地址覆盖打穿，然后直接切强制"这条路被堵住了。
    """
    from app.core.models import AdvancedShippingShadowEvent

    rows = await db.execute(
        select(AdvancedShippingShadowEvent.event_type,
               AdvancedShippingShadowEvent.payload).where(
            AdvancedShippingShadowEvent.tenant_id == tenant_id,
            AdvancedShippingShadowEvent.event_type.in_(REQUIRED_ACKS),
        )
    )
    return {
        event_type for event_type, payload in rows.all()
        if cache_version is None or (payload or {}).get("cache_version") == cache_version
    }


async def write_settings(db, tenant_id: int, payload) -> dict:
    """改模式/时区。cache_version 只读，客户端传什么都忽略；时区缺省则保持不变。"""
    settings = await ensure_settings(db, tenant_id)
    acked = (await _acknowledged(db, tenant_id, cache_version=settings.cache_version)
             if payload.mode == "enforced" else set())
    assert_mode_transition(settings.mode, payload.mode, acked=acked)

    settings.mode = payload.mode
    timezone = getattr(payload, "operating_timezone", None)
    if timezone:                     # 只改模式的 PUT 不该顺手把租户时区重置成默认值
        settings.operating_timezone = timezone
    # 这里**不**调 _finish：mode/operating_timezone 都不参与缓存的 RuleSet（cache.load_rule_set），
    # 失效纯属浪费；更要命的是 ack 绑在 cache_version 上，白白自增会把刚做完的确认作废，
    # 于是"存时区 → 存模式"两次调用的前端永远切不到 enforced。
    await db.commit()
    # cache_version 不回传：_finish 那类 ORM UPDATE 之后这个属性是 V 还是 V+1 取决于
    # SQLAlchemy 的同步策略，而 ack 正是按它对齐的——让客户端重新 GET，不给一个可能错的数。
    return {"mode": settings.mode, "operating_timezone": settings.operating_timezone}


async def list_shadow_events(db, tenant_id: int, *, limit: int = 50, offset: int = 0) -> list[dict]:
    """影子事件只读——本模块**不**提供任何 update/delete 入口。

    排除 simulation_ack_* ：那些是模式切换的确认记录，不是影子比对结果。
    混在一起的话，Admin 的差异表会把 shipping_method_id=0 的哨兵行当成运费分歧展示。
    """
    from app.core.models import AdvancedShippingShadowEvent

    rows = (await db.execute(
        select(AdvancedShippingShadowEvent)
        .where(AdvancedShippingShadowEvent.tenant_id == tenant_id,
               AdvancedShippingShadowEvent.event_type.notin_(tuple(REQUIRED_ACKS)))
        .order_by(AdvancedShippingShadowEvent.id.desc())
        .limit(min(max(int(limit), 1), 200)).offset(max(int(offset), 0))
    )).scalars().all()
    return [
        {"id": e.id, "event_type": e.event_type, "shipping_method_id": e.shipping_method_id,
         "rule_id": e.rule_id, "quote_group": e.quote_group,
         "legacy_amount": None if e.legacy_amount is None else str(e.legacy_amount),
         "candidate_amount": None if e.candidate_amount is None else str(e.candidate_amount),
         "payload": e.payload,
         "created_at": e.created_at.isoformat() if e.created_at else None}
        for e in rows
    ]


def list_templates() -> list[dict]:
    from .templates import TEMPLATES, TEMPLATE_VERSION

    return [{"template_version": TEMPLATE_VERSION, **dict(t)} for t in TEMPLATES.values()]


# ══════════════════════════════════════════════════════════════════════════
#  模拟
# ══════════════════════════════════════════════════════════════════════════

async def _base_currency(db, tenant_id: int) -> str:
    from app.core.services.currency_service import CurrencyService

    currency = await CurrencyService.get_default(db, tenant_id)
    return currency.code if currency else "NZD"


async def build_sim_facts(db, tenant_id: int, payload, *, settings,
                          base_currency: str | None = None):
    """购物车事实（模拟专用入口）：校验商品/规格属于本租户，然后交给结账同一个构建器。

    base_currency 允许由调用方预取（AI 起草要连跑多个场景，不该每个场景查一次币种）；
    缺省时就地查，保持 simulate 原有的查询顺序（商品 → 币种）。
    """
    from app.core.models.product import Product, ProductVariant

    product_ids = sorted({int(i.product_id) for i in payload.items})
    products: dict[int, Any] = {}
    if product_ids:
        rows = (await db.execute(
            select(Product)
            .where(Product.id.in_(product_ids), Product.tenant_id == tenant_id)
            .options(selectinload(Product.categories))
        )).scalars().all()
        products = {p.id: p for p in rows}
    missing = [i for i in product_ids if i not in products]
    if missing:
        raise HTTPException(status_code=400, detail=f"商品 {missing} 不存在或不属于本租户")

    variant_ids = sorted({int(i.variant_id) for i in payload.items if i.variant_id})
    variants: dict[int, Any] = {}
    if variant_ids:
        rows = (await db.execute(
            select(ProductVariant).where(
                ProductVariant.id.in_(variant_ids),
                ProductVariant.tenant_id == tenant_id,
            )
        )).scalars().all()
        variants = {v.id: v for v in rows}

    if base_currency is None:
        base_currency = await _base_currency(db, tenant_id)

    return build_cart_facts(
        items=list(payload.items),
        products=products,
        variants=variants,
        line_totals=[Decimal(str(i.line_total)) for i in payload.items],
        cart_subtotal=Decimal(str(payload.cart_subtotal)),
        cart_total=Decimal(str(payload.cart_total)),
        address=payload.address or None,
        customer_group_ids=list(payload.customer_group_ids or ()),
        payment_method=payload.payment_method,
        coupon_codes=list(payload.coupon_codes or ()),
        base_currency=base_currency,
        delivery_mode=payload.delivery_mode,
        operating_timezone=settings.operating_timezone,
    )


async def simulate(db, tenant_id: int, payload) -> dict:
    """跑一次模拟：**与结账同一个事实构建器、同一个引擎**（services.build_cart_facts + quote）。

    只接受购物车事实，绝不接受客户端传来的运费。
    """
    settings = await settings_or_default(db, tenant_id)
    facts = await build_sim_facts(db, tenant_id, payload, settings=settings)

    # 与结账走同一个组合入口（quote.quote_facts）：模拟和真实结账不能各算各的。
    rule_set, quote_set = await quote_facts(
        db, tenant_id, facts, cache_version=settings.cache_version,
    )
    return build_simulation_result(rule_set, quote_set, facts, mode=settings.mode)


async def acknowledge_simulation(db, tenant_id: int, payload, *, user_id: int | None = None) -> dict:
    """跑一次模拟并把结果确认下来。两种 ack 齐了才允许切 enforced。

    ack 里记下当时的 cache_version：任何规则/配置写入都会让它自增，于是"确认完再改规则、
    然后直接切强制"会因为版本不匹配而被 _acknowledged 过滤掉，必须重新确认。
    """
    from app.core.models import AdvancedShippingShadowEvent

    settings = await ensure_settings(db, tenant_id)
    assert_ack_kind(payload.kind, payload.address)
    result = await simulate(db, tenant_id, payload)
    assert_ack_result(payload.kind, result)

    event_type = ACK_NO_ADDRESS if payload.kind == "no_address" else ACK_REPRESENTATIVE
    selected = result["quotes"][0] if result["quotes"] else None
    db.add(AdvancedShippingShadowEvent(
        tenant_id=tenant_id,
        shipping_method_id=selected["shipping_method_id"] if selected else 0,
        rule_id=(selected["rule_ids"][0] if selected and selected["rule_ids"] else None),
        quote_group=selected["quote_group"] if selected else None,
        event_type=event_type,
        candidate_amount=Decimal(selected["fee"]) if selected else None,
        # 只记脱敏摘要：不落地址、邮编、优惠券码（同 audit.build_shadow_payload 的口径）
        payload={"kind": payload.kind, "currency": result["currency"],
                 "reason_keys": result["reason_keys"],
                 "quote_count": len(result["quotes"]),
                 "cache_version": settings.cache_version,
                 "acknowledged_by": user_id},
    ))
    await db.commit()
    acked = await _acknowledged(db, tenant_id, cache_version=settings.cache_version)
    return {"acknowledged": sorted(acked),
            "enforcement_ready": all(k in acked for k in REQUIRED_ACKS),
            "result": result}


# ══════════════════════════════════════════════════════════════════════════
#  legacy 转换（显式两步，绝不挂在插件启用的副作用上）
# ══════════════════════════════════════════════════════════════════════════

async def _load_conversion_inputs(db, tenant_id: int):
    from app.core.models import AdvancedShippingRule
    from app.core.models.shipping import ShippingMethod
    from app.core.models.shipping_zone import ShippingZone

    methods = (await db.execute(
        select(ShippingMethod).where(ShippingMethod.tenant_id == tenant_id)
        .order_by(ShippingMethod.sort_order, ShippingMethod.id)
    )).scalars().all()
    zones = (await db.execute(
        select(ShippingZone).where(ShippingZone.tenant_id == tenant_id)
    )).scalars().all()
    rules = (await db.execute(
        select(AdvancedShippingRule).where(AdvancedShippingRule.tenant_id == tenant_id)
    )).scalars().all()
    return list(methods), list(zones), list(rules)


async def preview_legacy_conversion(db, tenant_id: int) -> dict:
    settings = await settings_or_default(db, tenant_id)   # 预览是只读，不建配置行
    methods, zones, rules = await _load_conversion_inputs(db, tenant_id)
    locale = await get_tenant_default_locale(db, tenant_id) or "en-NZ"
    plan = plan_legacy_conversion(methods=methods, zones=zones, existing_rules=rules,
                                  locale=locale)
    plan["mode"] = settings.mode
    return plan


async def apply_legacy_conversion(db, tenant_id: int, *, user_id: int | None = None) -> dict:
    """把预览里的可转换项落成**禁用**规则 + 修订。不切换 mode，重跑不产生重复规则。"""
    settings = await ensure_settings(db, tenant_id)
    methods, zones, rules = await _load_conversion_inputs(db, tenant_id)
    locale = await get_tenant_default_locale(db, tenant_id) or "en-NZ"
    plan = plan_legacy_conversion(methods=methods, zones=zones, existing_rules=rules,
                                  locale=locale)

    created: list[int] = []
    for item in plan["convertible"]:
        payload = AdvancedShippingRuleIn(**item["rule"])
        rule = await _insert_rule(
            db, tenant_id, payload, user_id=user_id, from_legacy_conversion=True,
            note=f"legacy conversion of shipping method {item['shipping_method_id']}")
        created.append(rule.id)

    if created:
        await _finish(db, tenant_id, settings)
    else:
        await db.commit()
    return {"created": created, "already_converted": plan["already_converted"],
            "unmappable": plan["unmappable"], "notes": plan["notes"], "mode": settings.mode}


# ══════════════════════════════════════════════════════════════════════════
#  给核心运费区域删除用的引用检查
# ══════════════════════════════════════════════════════════════════════════

async def zone_reference_names(db, tenant_id: int, zone_id: int) -> list[str]:
    """哪些**启用中**的高级规则依赖这个运费区域。

    规则不直接指向区域，它指向 shipping_method；方案的 zone_id 才是区域。
    删掉区域会让这些规则的 inherit 定价换一个区域重新匹配（静默算错价），所以要拦。
    """
    from app.core.models import AdvancedShippingRule
    from app.core.models.shipping import ShippingMethod

    rows = await db.execute(
        select(AdvancedShippingRule.presentation, AdvancedShippingRule.id)
        .join(ShippingMethod, ShippingMethod.id == AdvancedShippingRule.shipping_method_id)
        .where(AdvancedShippingRule.tenant_id == tenant_id,
               AdvancedShippingRule.enabled == 1,
               ShippingMethod.tenant_id == tenant_id,
               ShippingMethod.zone_id == zone_id)
        .order_by(AdvancedShippingRule.id)
    )
    return rule_display_names(rows.all())


async def method_reference_names(db, tenant_id: int, method_id: int) -> list[str]:
    """哪些**启用中**的高级规则直接挂在这个物流方案上。

    advanced_shipping_rules.shipping_method_id 没有外键（插件表不该硬绑核心表），
    所以删方案不会连带清理规则：那些规则会继续参与报价、inherit 解析到一个不存在的方案，
    而且 _assert_owned_method 会让 update/rollback 永远 404——操作者连修都修不了。
    """
    from app.core.models import AdvancedShippingRule

    rows = await db.execute(
        select(AdvancedShippingRule.presentation, AdvancedShippingRule.id)
        .where(AdvancedShippingRule.tenant_id == tenant_id,
               AdvancedShippingRule.enabled == 1,
               AdvancedShippingRule.shipping_method_id == method_id)
        .order_by(AdvancedShippingRule.id)
    )
    return rule_display_names(rows.all())


__all__ = [
    "ACK_NO_ADDRESS", "ACK_REPRESENTATIVE", "LEGACY_CONVERSION_COLUMN",
    "NOTE_CONVERSION_CREATES_DISABLED", "NOTE_CONVERSION_DOES_NOT_ENFORCE",
    "NOTE_INHERIT_INCLUDES_SURCHARGES", "REQUIRED_ACKS",
    "acknowledge_simulation", "apply_legacy_conversion", "assert_ack_kind", "assert_ack_result",
    "assert_group_strategy", "assert_mode_transition", "build_sim_facts",
    "build_simulation_result", "create_rule",
    "delete_rule", "ensure_settings", "list_revisions", "list_rules", "list_shadow_events",
    "list_templates", "plan_legacy_conversion", "preview_legacy_conversion", "read_settings",
    "rollback_rule", "rule_display_names", "rule_snapshot", "simulate", "update_rule",
    "write_settings", "zone_reference_names",
]