"""
报价工作表映射 mode：上传时用千问产出 mappingJson。

本文件职责：加载映射分析提示词、拼用户消息、调千问、剥 Markdown、校验 JSON、
对照单元格清单做缺项检测；有缺口则多轮补跑（默认最多 3 轮），目标一次流程内齐全。
禁止：发明运价；在企微回调线程同步调用；出单阶段再调本 mode。
"""

from __future__ import annotations

import json
import logging
import re
from pathlib import Path
from typing import Any, Optional

from agent.llm.modes import LlmMode

logger = logging.getLogger(__name__)

MODE = LlmMode.SHEET_MAP

_PROMPT_CACHE: str | None = None

# 清单原文片段 → 白名单 code（按运输方式过滤；关键词避免过短英文误报）
_LABEL_TO_CODE_COMMON: list[tuple[tuple[str, ...], str]] = [
    (("报价单编号", "询价单号", "RFQ No", "Quote No"), "quote_no"),
    (
        (
            "报价日期",
            "询价日期",
            "走货日期",
            "询价时间",
            "走货时间",
            "发货时间",
            "发货日期",
            "出货日期",
            "出货时间",
            "提货时间",
            "提货日期",
            "Inquiry Date",
            "Quote Date",
        ),
        "quote_date",
    ),
    (("有效期至", "有效期", "Validity"), "valid_until"),
    (("客户名称", "Customer Name", "Customer"), "customer_name"),
    (("联系人", "Contact Person", "Contact Name"), "contact_name"),
    (("电话", "Tel", "Phone", "Mobile"), "contact_phone"),
    (("币别", "Currency"), "currency"),
    (("业务员", "Sales"), "sales_name"),
    (("品名", "Commodity", "Goods Description"), "commodity"),
]

_LABEL_TO_CODE_SEA: list[tuple[tuple[str, ...], str]] = [
    (("提货地址", "Place of Receipt"), "pickup_address"),
    (("送货地址", "派送地址", "Place of Delivery"), "delivery_address"),
    (("货值", "Cargo Value"), "cargo_value"),
    (("海关编码", "HS Code", "HS编码"), "hs_code"),
    (("油、电、磁", "油,电,磁", "油电磁"), "oil_elec_mag"),
    (("特殊说明", "特别说明", "Special Remarks"), "special_remarks"),
    (("要求货物到达", "Requested Arrival"), "eta_request"),
    (("运输时效", "Transit Time", "航程时间"), "transit_time"),
    (("船期", "Sailing Schedule"), "sailing_schedule"),
    (("开船日", "On Board Date"), "on_board_date"),
    (("船公司", "Shipping Line", "Shipping Company"), "shipping_line"),
    (("货好时间", "Cargo Ready"), "cargo_ready_date"),
    (("贸易条款", "Trade Terms", "Incoterms"), "trade_terms"),
    (("整柜或散货", "整柜或拼柜", "FCL or LCL"), "fcl_or_lcl"),
    (("箱型箱量", "箱型", "柜量", "Type & Qty"), "container_type_qty"),
    (("起运港", "Port of Loading", "POL"), "pol"),
    (("目的港", "Port of Discharge", "POD"), "pod"),
    (("货物数量", "Quantity"), "quantity"),
]

_LABEL_TO_CODE_AIR: list[tuple[tuple[str, ...], str]] = [
    (("起运地", "起运港", "Departure", "Origin Airport"), "origin"),
    (("目的地", "目的港", "Destination Airport"), "destination"),
    (("件数", "Pieces", "Pkgs"), "pieces"),
    (("毛重", "重量", "Gross Weight", "Weight"), "weight_kg"),
    (("体积", "Volume", "CBM"), "volume_cbm"),
    (("计费重", "Chargeable Weight"), "chargeable_weight_kg"),
    (("航班", "Flight"), "flight_no"),
    (("中转机场", "Transfer Airport"), "transfer_airport"),
    (("贸易条款", "Trade Terms", "Incoterms"), "trade_terms"),
    (("特殊说明", "特别说明", "Special Remarks"), "special_remarks"),
]

_LABEL_TO_CODE_LAND: list[tuple[tuple[str, ...], str]] = [
    (("始发地", "起运地", "发站", "Origin"), "origin"),
    (("目的地", "到站", "Destination"), "destination"),
    (("运输时效", "时效", "Transit Time"), "transit_time"),
    (("特殊说明", "特别说明"), "special_remarks"),
]

_SEA_FEE_MARKERS: list[tuple[tuple[str, ...], tuple[str, ...]]] = [
    # (清单关键词, 可匹配的 feeRegion id/title/tmsSource 子串)
    (
        ("始发港费用", "Origin Charges"),
        ("origin_charges", "polcost", "始发港", "origin"),
    ),
    (
        ("海运费用", "Ocean Freight"),
        ("ocean_freight", "oceancost", "海运费用", "ocean"),
    ),
    (
        ("目的港费用", "Destination Charges"),
        ("destination_charges", "podcost", "目的港", "destination"),
    ),
]


def _labels_for_mode(mode: str) -> list[tuple[tuple[str, ...], str]]:
    m = (mode or "sea").strip().lower()
    out = list(_LABEL_TO_CODE_COMMON)
    if m in ("sea", "海运"):
        out.extend(_LABEL_TO_CODE_SEA)
    elif m in ("air", "空运"):
        out.extend(_LABEL_TO_CODE_AIR)
    elif m in ("land", "陆运"):
        out.extend(_LABEL_TO_CODE_LAND)
    else:
        out.extend(_LABEL_TO_CODE_SEA)
    return out


def _label_hit(blob: str, labels: tuple[str, ...]) -> bool:
    """清单命中：中文标签直接包含；英文至少 4 字符，降低误报。"""
    for lab in labels:
        if not lab:
            continue
        if any("\u4e00" <= ch <= "\u9fff" for ch in lab):
            if lab in blob:
                return True
        elif len(lab) >= 4 and lab.lower() in blob.lower():
            return True
    return False


def mode_meta() -> dict[str, Any]:
    return {
        "mode": MODE.value,
        "provider": "qwen",
        "purpose": "sheet_map",
        "implemented": True,
    }


def _prompt_path() -> Path:
    return Path(__file__).resolve().parents[1] / "quote_templates" / "prompts" / "mapping_analyzer.txt"


def load_system_prompt() -> str:
    """加载映射分析器系统提示词（仓库内只读资源）。"""
    global _PROMPT_CACHE
    if _PROMPT_CACHE is not None:
        return _PROMPT_CACHE
    path = _prompt_path()
    text = path.read_text(encoding="utf-8")
    _PROMPT_CACHE = text
    return text


def _inventory_lines(payload: dict[str, Any]) -> list[str]:
    inventory = payload.get("cellInventory") or payload.get("cell_inventory") or []
    if isinstance(inventory, list):
        return [str(x) for x in inventory]
    return [str(inventory)]


def _inventory_blob(payload: dict[str, Any]) -> str:
    return "\n".join(_inventory_lines(payload))


def build_user_message(payload: dict[str, Any], *, gaps: list[str] | None = None) -> str:
    """
    组装用户消息：运输方式 + 文件名 + 工作表 + 单元格清单。
    gaps 非空时追加补全指令（第二轮）。
    """
    mode = str(payload.get("transportMode") or payload.get("transport_mode") or "sea").strip()
    mode_cn = {"sea": "海运", "air": "空运", "land": "陆运"}.get(mode, mode)
    file_name = str(payload.get("fileName") or payload.get("file_name") or "")
    sheet = str(payload.get("sheet") or "")
    lines = _inventory_lines(payload)
    max_lines = 2200
    truncated_note = ""
    if len(lines) > max_lines:
        lines = lines[:max_lines]
        truncated_note = f"\n（单元格清单已截断至前 {max_lines} 行）\n"
    body = "\n".join(lines)
    gap_block = ""
    if gaps:
        gap_lines = "\n".join(f"- {g}" for g in gaps)
        gap_block = (
            "\n【补跑缺口】上一轮 mapping 缺少以下项，请输出完整 mappingJson（不要只回增量）：\n"
            f"{gap_lines}\n"
        )
    return (
        f"运输方式：{mode_cn}\n"
        f"模板文件名：{file_name}\n"
        f"工作表名：{sheet}\n"
        f"单元格清单：地址 | 原文 | 合并范围 | 填充色\n"
        f"{body}"
        f"{truncated_note}"
        f"{gap_block}\n"
        "输出前自检 fields 与 feeRegions/routeRegions 是否覆盖清单中的可见业务标签，"
        "再只输出 mappingJson。"
    )


def strip_json_fence(raw: str) -> str:
    """去掉模型可能包的 Markdown 代码围栏。"""
    text = (raw or "").strip()
    if not text:
        return ""
    fence = re.match(r"^```(?:json)?\s*([\s\S]*?)\s*```$", text, re.IGNORECASE)
    if fence:
        return fence.group(1).strip()
    # 偶发前后解释：取第一个 { 到最后一个 }
    start = text.find("{")
    end = text.rfind("}")
    if start >= 0 and end > start:
        return text[start : end + 1]
    return text


def parse_mapping_json(raw: str) -> dict[str, Any]:
    """
    解析模型正文为 mapping 对象。

    成功：{ok:True, mapping:dict}
    失败：{ok:False, error:str}
    """
    cleaned = strip_json_fence(raw)
    if not cleaned:
        return {"ok": False, "error": "模型返回空内容"}
    try:
        data = json.loads(cleaned)
    except json.JSONDecodeError as exc:
        return {"ok": False, "error": f"mappingJson 不是合法 JSON: {exc}"}
    if not isinstance(data, dict):
        return {"ok": False, "error": "mappingJson 不是对象"}
    if not isinstance(data.get("fields"), dict):
        data["fields"] = data.get("fields") if isinstance(data.get("fields"), dict) else {}
    data.setdefault("schemaVersion", "quote-template-mapping-v1")
    return {"ok": True, "mapping": data}


def assess_mapping_gaps(payload: dict[str, Any], mapping: dict[str, Any] | None) -> list[str]:
    """
    对照单元格清单，列出 mapping 明显缺口（中文说明，供补跑用户消息）。

    不打外网。按运输方式过滤字段词典；海运另检查三段费用区。
    """
    blob = _inventory_blob(payload)
    if not blob.strip():
        return []
    mapping = mapping or {}
    fields = mapping.get("fields") if isinstance(mapping.get("fields"), dict) else {}
    field_codes = {str(k) for k in fields.keys()}
    fee_regions = mapping.get("feeRegions") if isinstance(mapping.get("feeRegions"), list) else []
    fee_blob = " ".join(
        str(r.get("id") or "")
        + " "
        + str(r.get("title") or "")
        + " "
        + str(r.get("tmsSource") or "")
        for r in fee_regions
        if isinstance(r, dict)
    ).lower()
    route_regions = mapping.get("routeRegions") if isinstance(mapping.get("routeRegions"), list) else []

    gaps: list[str] = []
    mode = str(payload.get("transportMode") or payload.get("transport_mode") or "sea").strip().lower()
    # 字段匹配用：去掉费用段标题，避免「目的港费用」误报成 pod 等字段缺口
    field_blob = blob
    if mode in ("sea", "海运"):
        for markers, matchers in _SEA_FEE_MARKERS:
            if not any(m in blob for m in markers):
                continue
            if not any(x in fee_blob for x in matchers):
                gaps.append(f"费用区缺失：清单有「{markers[0]}」，须建对应 feeRegion")
            for m in markers:
                field_blob = field_blob.replace(m, " ")
    elif mode in ("land", "陆运"):
        # 陆运主表常见项目/线路矩阵：清单有「项目」或车型价列但无 routeRegions
        if (
            ("项目" in blob or "车型" in blob or "3T" in blob or "拼车" in blob)
            and not route_regions
        ):
            gaps.append("线路区缺失：陆运表须建 routeRegions 承接线路/车型报价")

    # 陆运已建 routeRegions 时，始发地/目的地/时效常在线路矩阵列头，不再强制 fields
    land_route_covered = mode in ("land", "陆运") and bool(route_regions)
    land_skip_codes = {"origin", "destination", "transit_time"} if land_route_covered else set()

    for labels, code in _labels_for_mode(mode):
        if code in field_codes or code in land_skip_codes:
            continue
        if _label_hit(field_blob, labels):
            gaps.append(f"字段缺失：清单有「{labels[0]}」→ 应映射 {code}")
    return gaps


def merge_mapping_prefer_existing(
    old_mapping: dict[str, Any] | None,
    new_mapping: dict[str, Any] | None,
    *,
    cell_inventory: list[str] | None = None,
) -> dict[str, Any]:
    """
    重新解析合并：费用/线路区按 id 旧优先；fields 对照单元格清单智能取舍。

    规则：旧无→用新；新无→留旧；相同→留旧；旧坐标仍贴标签→留旧；
    旧坐标对不上（清单里能看到该标签）→改用新坐标。无清单时保守旧优先。
    """
    new_mapping = dict(new_mapping or {})
    old_mapping = old_mapping if isinstance(old_mapping, dict) else {}
    out = dict(new_mapping)

    new_fields = new_mapping.get("fields") if isinstance(new_mapping.get("fields"), dict) else {}
    old_fields = old_mapping.get("fields") if isinstance(old_mapping.get("fields"), dict) else {}
    new_meta = new_mapping.get("fieldMeta") if isinstance(new_mapping.get("fieldMeta"), dict) else {}
    old_meta = old_mapping.get("fieldMeta") if isinstance(old_mapping.get("fieldMeta"), dict) else {}
    meta_out: dict[str, Any] = dict(new_meta)

    codes = set(str(k) for k in old_fields.keys()) | set(str(k) for k in new_fields.keys())
    fields: dict[str, Any] = {}
    inv = list(cell_inventory or [])
    for code in codes:
        old_cell = str(old_fields.get(code) or "").strip()
        new_cell = str(new_fields.get(code) or "").strip()
        kept_old = False
        if not old_cell:
            chosen = new_cell
        elif not new_cell:
            chosen = old_cell
            kept_old = True
        elif old_cell.lower() == new_cell.lower():
            chosen = old_cell
            kept_old = True
        elif inv and _field_mapping_still_aligned(code, old_cell, inv):
            chosen = old_cell
            kept_old = True
        elif inv:
            chosen = new_cell
            logger.info("映射坐标纠偏 code=%s old=%s -> new=%s", code, old_cell, new_cell)
        else:
            chosen = old_cell
            kept_old = True
        if chosen:
            fields[code] = chosen
        if kept_old and isinstance(old_meta.get(code), dict):
            meta_out[code] = dict(old_meta[code])
    out["fields"] = fields
    if meta_out:
        out["fieldMeta"] = meta_out

    out["feeRegions"] = _merge_regions_by_id(
        old_mapping.get("feeRegions"), new_mapping.get("feeRegions")
    )
    out["routeRegions"] = _merge_regions_by_id(
        old_mapping.get("routeRegions"), new_mapping.get("routeRegions")
    )
    return out


# 与主账 FILL_BY_LABEL 对齐的常用码→标签（智能合并校验用）
_FIELD_LABELS: dict[str, tuple[str, ...]] = {
    "quote_no": ("RFQ No", "报价单编号", "报价编号"),
    "quote_date": (
        "报价日期",
        "询价日期",
        "走货日期",
        "询价时间",
        "走货时间",
        "发货时间",
        "发货日期",
        "出货日期",
        "出货时间",
        "提货时间",
        "提货日期",
        "Inquiry Date",
    ),
    "pol": ("起运港", "POL"),
    "pod": ("目的港", "POD"),
    "commodity": ("品名", "货物品名", "Commodity"),
    "quantity": ("货物数量", "货量"),
    "fcl_or_lcl": ("整柜或拼柜",),
    "container_type_qty": ("箱型箱量",),
    "valid_until": ("报价有效期", "有效期至"),
    "total_amount": ("费用总计", "报价合计"),
    "origin": ("始发地", "始发站", "起运地"),
    "destination": ("目的地",),
}


def _parse_inventory_line(line: str) -> tuple[str, str, str] | None:
    """地址 | 原文 | 合并 | 色 → (full_addr, text, merge)。"""
    parts = [p.strip() for p in str(line or "").split("|")]
    if len(parts) < 2:
        return None
    return parts[0], parts[1], parts[2] if len(parts) > 2 else "-"


def _col_index(letter: str) -> int:
    n = 0
    for ch in (letter or "").upper():
        if "A" <= ch <= "Z":
            n = n * 26 + (ord(ch) - ord("A") + 1)
    return max(n - 1, 0)


def _split_addr(full: str) -> tuple[str, int, int] | None:
    """Sheet!B3 → (sheet, row0, col0)。"""
    text = (full or "").strip()
    if "!" not in text:
        return None
    sheet, addr = text.split("!", 1)
    m = re.match(r"^\$?([A-Za-z]+)\$?(\d+)$", addr.strip())
    if not m:
        return None
    return sheet, int(m.group(2)) - 1, _col_index(m.group(1))


def _merge_end_col(merge: str, fallback_col: int) -> int:
    """A3:C3 → 末列 index；无合并返回 fallback。"""
    m = (merge or "").strip()
    if not m or m == "-":
        return fallback_col
    # 可能是 A3:C3 或 Sheet!A3:C3
    if "!" in m:
        m = m.split("!", 1)[1]
    if ":" not in m:
        return fallback_col
    end = m.split(":", 1)[1].strip()
    mm = re.match(r"^\$?([A-Za-z]+)\$?(\d+)$", end)
    if not mm:
        return fallback_col
    return _col_index(mm.group(1))


def _text_looks_like_label(text: str, labels: tuple[str, ...]) -> bool:
    norm = (text or "").replace("：", ":").strip()
    if not norm:
        return False
    primary = norm.split("/")[0].strip()
    upper = norm.upper()
    for lb in labels:
        if not lb:
            continue
        if norm.lower().startswith(lb.lower()) or primary.lower().startswith(lb.lower()):
            return True
        if len(lb) >= 3 and lb.upper() in upper:
            return True
    return False


def _field_mapping_still_aligned(code: str, cell_ref: str, inventory: list[str]) -> bool:
    """
    旧坐标是否仍可贴标签。

    quote_no：贴着 RFQ 的 B3 常是 Port to Port / 工作天单选，不算对齐。
    """
    labels = _FIELD_LABELS.get(str(code), ())
    if not labels:
        return True
    parsed = _split_addr(cell_ref)
    if not parsed:
        return False
    sheet, row, col = parsed
    rows: list[tuple[str, int, int, str, str]] = []
    mapped_text = ""
    for line in inventory:
        one = _parse_inventory_line(line)
        if not one:
            continue
        addr, text, merge = one
        sp = _split_addr(addr)
        if not sp or sp[0] != sheet:
            continue
        rows.append((addr, sp[1], sp[2], text, merge))
        if sp[1] == row and sp[2] == col:
            mapped_text = text
            if _text_looks_like_label(text, labels):
                return True
    if _is_unsuitable_value_cell(mapped_text):
        return False
    if str(code) == "quote_no" and mapped_text.strip() and not _looks_like_quote_no(mapped_text):
        return False
    saw = False
    for _addr, r, c, text, merge in rows:
        if not _text_looks_like_label(text, labels):
            continue
        saw = True
        if r == row and c == col:
            return True
        end_col = _merge_end_col(merge, c)
        if r == row:
            gap = col - end_col
            if 1 <= gap <= 3 and (not mapped_text.strip() or _looks_like_quote_no(mapped_text)):
                return True
            if str(code) == "quote_no" and gap > 3 and (
                not mapped_text.strip() or _looks_like_quote_no(mapped_text)
            ):
                return True
    return not saw


def _is_unsuitable_value_cell(text: str) -> bool:
    t = (text or "").replace("：", ":").strip()
    if not t:
        return False
    lower = t.lower()
    if any(x in t for x in ("○", "●", "⊙", "□", "☑")):
        return True
    if "port to port" in lower or "door to door" in lower:
        return True
    if "工作天" in t or "工作日" in t or "回复" in t:
        return True
    if "其它" in t and ("天" in t or "回复" in t or "Port" in t):
        return True
    return False


def _looks_like_quote_no(text: str) -> bool:
    t = (text or "").strip()
    if not t or len(t) > 40:
        return False
    import re as _re

    if _re.match(r"(?i)WO\d{6,}", t) or _re.match(r"(?i)W0\d{6,}", t):
        return True
    if _re.match(r"(?i)RFQ[-_]?[A-Z0-9]+", t):
        return True
    return bool(_re.match(r"[A-Z]{1,5}[-_]?\d{8,}", t))


def _merge_regions_by_id(old_arr: Any, new_arr: Any) -> list[dict[str, Any]]:
    """区域按 id 合并：旧整段保留，新 id 追加。"""
    out: list[dict[str, Any]] = []
    seen: set[str] = set()
    for src in (old_arr, new_arr):
        if not isinstance(src, list):
            continue
        for item in src:
            if not isinstance(item, dict):
                continue
            rid = str(item.get("id") or "").strip()
            if rid and rid in seen:
                continue
            if rid:
                seen.add(rid)
            out.append(dict(item))
    return out


def invoke_shell(*, payload: dict[str, Any]) -> dict[str, Any]:
    """
    无网/单测入口：若带 assistant_json 则直接解析；否则返回未实现。
    """
    raw = payload.get("assistant_json") or payload.get("content")
    if raw is not None:
        parsed = parse_mapping_json(str(raw))
        parsed["stub"] = True
        parsed["mode"] = MODE.value
        return parsed
    return {
        "ok": False,
        "stub": True,
        "mode": MODE.value,
        "error": "sheet_map_shell_needs_assistant_json_or_network",
        "echo_keys": sorted(payload.keys()),
    }


def _chat_once(
    *,
    chat: Any,
    payload: dict[str, Any],
    gaps: list[str] | None = None,
) -> dict[str, Any]:
    messages = [
        {"role": "system", "content": load_system_prompt()},
        {"role": "user", "content": build_user_message(payload, gaps=gaps)},
    ]
    result = chat(mode=MODE, messages=messages, temperature=0.0)
    if not getattr(result, "ok", False):
        err = str(getattr(result, "error", "") or "qwen_failed")
        logger.warning("sheet_map 失败 versionId=%s err=%s", payload.get("versionId"), err)
        return {
            "ok": False,
            "error": _humanize_llm_error(err),
            "mapping": None,
            "provider": getattr(result, "provider", "qwen"),
        }
    parsed = parse_mapping_json(str(result.content or ""))
    if not parsed.get("ok"):
        return {
            "ok": False,
            "error": str(parsed.get("error") or "解析 mapping 失败"),
            "mapping": None,
            "raw_preview": str(result.content or "")[:240],
        }
    return {
        "ok": True,
        "mapping": parsed["mapping"],
        "provider": getattr(result, "provider", "qwen"),
    }


def invoke_sheet_map(
    payload: dict[str, Any],
    *,
    allow_network: bool = False,
    chat_fn: Optional[Any] = None,
    timeout_seconds: float = 240.0,
    max_rounds: int = 3,
) -> dict[str, Any]:
    """
    调用千问产出 mappingJson；对照清单有缺口则补跑，最多 max_rounds 轮（默认 3），
    目标一次流程内尽量齐全，缺口写进 mappingGaps。

    allow_network=False 且无 chat_fn：不打外网。
    副作用：可能出站 HTTPS；应在 Worker HTTP 槽内调用。
    """
    if chat_fn is None and not allow_network:
        return {"ok": False, "error": "sheet_map_network_disabled", "mapping": None}

    from agent.llm.http_client import LlmHttpClient

    if chat_fn is None:
        client = LlmHttpClient.from_settings()
        client.timeout_seconds = float(timeout_seconds)
        chat = client.chat
    else:
        chat = chat_fn

    rounds_cap = max(1, min(int(max_rounds or 3), 4))
    first = _chat_once(chat=chat, payload=payload, gaps=None)
    if not first.get("ok"):
        return first

    mapping = first["mapping"]
    rounds = 1
    remaining = assess_mapping_gaps(payload, mapping)
    # 有缺口就补跑，直到齐、轮次用尽、或本轮未减少缺口
    while remaining and rounds < rounds_cap:
        logger.info(
            "sheet_map gaps versionId=%s round=%s n=%s sample=%s",
            payload.get("versionId"),
            rounds,
            len(remaining),
            remaining[:5],
        )
        nxt = _chat_once(chat=chat, payload=payload, gaps=remaining)
        rounds += 1
        if not (nxt.get("ok") and nxt.get("mapping")):
            logger.warning(
                "sheet_map 补跑失败，保留上一轮 versionId=%s err=%s",
                payload.get("versionId"),
                nxt.get("error"),
            )
            break
        mapping = nxt["mapping"]
        after = assess_mapping_gaps(payload, mapping)
        if len(after) >= len(remaining):
            # 未继续收敛则停止，避免空转烧额度
            logger.info(
                "sheet_map gaps 未收敛 versionId=%s before=%s after=%s stop",
                payload.get("versionId"),
                len(remaining),
                len(after),
            )
            remaining = after
            break
        remaining = after

    # 缺口落进 mappingJson；齐全时清空 AI 自报未映射，避免列表角标被说明性标签抬高
    if isinstance(mapping, dict):
        mapping["mappingGaps"] = list(remaining)
        if not remaining:
            mapping["unmappedRequired"] = []
            mapping["unmappedLabels"] = []
        elif not isinstance(mapping.get("unmappedLabels"), list):
            mapping["unmappedLabels"] = []
    logger.info(
        "sheet_map ok versionId=%s rounds=%s fields=%s feeRegions=%s routeRegions=%s remain_gaps=%s",
        payload.get("versionId"),
        rounds,
        len((mapping or {}).get("fields") or {}),
        len((mapping or {}).get("feeRegions") or []),
        len((mapping or {}).get("routeRegions") or []),
        len(remaining),
    )
    return {
        "ok": True,
        "mapping": mapping,
        "provider": first.get("provider", "qwen"),
        "rounds": rounds,
        "gaps_after": remaining,
    }


def _humanize_llm_error(err: str) -> str:
    e = (err or "").strip()
    if e == "llm_data_usage_not_confirmed":
        return "未确认 LLM 数据用途（LLM_DATA_USAGE_CONFIRMED），无法调用千问"
    if e == "qwen_not_configured":
        return "千问未配置（QWEN_API_KEY / QWEN_BASE_URL）"
    if e.startswith("http_"):
        return f"千问 HTTP 失败: {e}"
    return f"千问调用失败: {e[:180]}"