"""
文字询价字段抽取入口。

本文件职责：生产走 DeepSeek B；单测可注入 facts；无网时只认「键：值」结构化行，不猜港口。
禁止：用正则把「上海到洛杉矶」拆成起运目的；禁止改六态。
"""

from __future__ import annotations

import logging
import re
from typing import Any, Optional

from agent.schema.field_validate import QUOTE_DATE_SYNONYMS

logger = logging.getLogger(__name__)

# 报价日期同义词。长的在前，避免短词截断。货好时间、提货地址不在这里。
_QUOTE_DATE_LABEL_RE = "|".join(
    re.escape(name)
    for name in sorted(("报价日期",) + QUOTE_DATE_SYNONYMS, key=len, reverse=True)
)

_MODE_WORDS = (
    ("空运", "AIR"),
    ("海运", "SEA"),
    ("陆运", "LAND"),
)
# 未定运输方式时：整句只回 1/2/3 对应陆运/海运/空运。10件、1.件数 不算。
_MODE_INDEX = re.compile(r"^\s*([123１-３])\s*[\.、．]?\s*$")
_MODE_BY_INDEX = {
    "1": "LAND",
    "2": "SEA",
    "3": "AIR",
    "１": "LAND",
    "２": "SEA",
    "３": "AIR",
}

# 仅「键：值」一行一条，键必须是合同字段或展示名。报价日期含询价/走货/发货等同义词。
_LABEL_KEYS = (
    r"起运地|起运港|目的地|目的港|品名|件数|毛重|体积（CBM）|体积\(CBM\)|体积|重量（KG）|重量\(KG\)|重量|"
    r"包装方式|包装类型|货量|货物数量|整柜或拼柜|箱型箱量|"
    + _QUOTE_DATE_LABEL_RE
    + r"|运输分类|车型/数量|车型数量|"
    r"贸易条款|货好时间|货源地|提货地址|收货地址|送货地址|商品海关编码|海关编码|HS编码|HS|是否含油|是否含电|是否含磁|"
    r"客户名称|货值|是否为危险品|通关口岸（非必填）|通关口岸"
)
_LABEL_LINE = re.compile(rf"^({_LABEL_KEYS})\s*[:：]\s*(.+)$")
_LABEL_INLINE = re.compile(rf"({_LABEL_KEYS})\s*[:：]\s*(\S+)")
# 陆运核对卡非必填：销售可能写「通关口岸皇岗」或照抄「通关口岸（非必填）：皇岗」。长词优先。
_OPTIONAL_INQUIRY_LABELS = (
    "通关口岸（非必填）",
    "通关口岸",
    "货源地（非必填）",
    "货好时间（非必填）",
    "提货地址（非必填）",
    "收货地址（非必填）",
    "贸易条款（非必填）",
    "起运地（非必填）",
    "起运港（非必填）",
    "货源地",
    "货好时间",
    "提货地址",
    "收货地址",
    "送货地址",
    "贸易条款",
    "起运港",
)
# 图上常见「4. 希望送到…」：编号项不是询价字段，值收到这里为止。
_NUMBERED_ITEM = re.compile(r"(?m)^\s*\d+[\.、．]\s+\S")
# 补问复制清单：1.件数（必填）：10 ；下一行 (如：10）不算值。
_CLARIFY_EXAMPLE_LINE = re.compile(r"^\s*[（(]如\s*[:：]")
_CLARIFY_INDEX = re.compile(r"^\s*\d+[\.、．]\s*")
_CLARIFY_DECOR = re.compile(r"（(?:KG|CBM|kg|cbm|必填|非必填)）|\((?:KG|CBM|必填|非必填)\)", re.I)
_CLARIFY_NAMES = frozenset(
    {
        "起运地",
        "起运港",
        "目的地",
        "目的港",
        "品名",
        "货物品名",
        "件数",
        "毛重",
        "体积（CBM）",
        "体积",
        "重量（KG）",
        "重量",
        "包装方式",
        "包装类型",
        "货量",
        "货物数量",
        "整柜或拼柜",
        "箱型箱量",
        "报价日期",
        *QUOTE_DATE_SYNONYMS,
        "运输分类",
        "运输方式",
        "运输类型",
        "线路类别",
        "车型/数量",
        "车型数量",
        "贸易条款",
        "货源地",
        "货好时间",
        "提货地址",
        "收货地址",
        "送货地址",
        "通关口岸",
        "始发站",
    }
)
# 报价单表头和格子在同一行：只洗这些询价键，不碰提货/收货地址。
_SHEET_CLEAN_KEYS = frozenset(
    {
        "起运地",
        "起运港",
        "目的地",
        "目的港",
        "品名",
        "件数",
        "毛重",
        "体积（CBM）",
        "体积",
        "重量（KG）",
        "重量",
        "货量",
        "货物数量",
        "整柜或拼柜",
        "箱型箱量",
        "贸易条款",
        "货好时间",
        "运输分类",
        "报价日期",
        *QUOTE_DATE_SYNONYMS,
        "包装方式",
        "包装类型",
    }
)
# 报价单从这里往下是费用和条款，不是询价栏。切掉后再抽字段。
_QUOTE_BODY_MARKS = (
    "报价明细",
    "报价条件",
    "预估费用",
    "空舱费",
    "托运方",
    "承运方",
)
# 表头常被拉成「体    积」「计 费 重」。只收这些词中间的空格，不跨行。
_TIGHTEN_LABELS = tuple(
    sorted(
        {
            *_QUOTE_BODY_MARKS,
            "计费重",
            "时效要求",
            "派送地址",
            "提货地址",
            "收货地址",
            "货好时间",
            "贸易条款",
            "报价日期",
            "起运地",
            "起运港",
            "目的地",
            "目的港",
            "包装类型",
            "包装方式",
            "体积",
            "重量",
            "品名",
            "件数",
            "货源地",
            "运输方式",
        },
        key=len,
        reverse=True,
    )
)
_FEE_NAME = re.compile(r"^\*?[\u4e00-\u9fffA-Za-z0-9]{1,12}费$")
_SHEET_STOP = (
    "报价明细",
    "报价条件",
    "预估费用",
    "计费重",
    "时效要求",
    "要求运输时效",
    "Requested Transit",
    "费用汇总",
    "POD Local Charges",
    "提货地址",
    "Place of Receipt",
    "送货地址",
    "派送地址",
    "Place of Delivery",
    "要求货物抵达",
    "Requested Arrival",
)
_SHEET_EN_ALIAS = re.compile(r"^[/／][A-Za-z][A-Za-z ]*")
_SHEET_REQUIRED_HINT = re.compile(r"（[^）]*必填[^）]*）")
_SHEET_REQUIRED_HINT_EN = re.compile(
    r"\([^)]*(?:必填|Packing List|Quantity)[^)]*\)",
    re.I,
)
_SHEET_HEADER_NOISE = re.compile(
    r"(费用汇总|提货地址|送货地址|派送地址|要求货物抵达|要求运输时效|"
    r"Local Charges|Place of Receipt|Place of Delivery|Requested Arrival|Requested Transit)",
    re.I,
)
# 群里协同标签，冒号可省略：「客户名称张三」「货值 10万」。长词优先。
_COLLAB_LABELS = (
    "是否为危险品",
    "商品海关编码",
    "车型/数量",
    "包装方式",
    "客户名称",
    "海关编码",
    "贸易条款",
    "货好时间",
    "是否含油",
    "是否含电",
    "是否含磁",
    "HS编码",
    "货值",
    "HS",
)


def detect_transport_mode(text: str) -> str:
    """
    运输方式：空运/海运/陆运，或空运口语「飞」（南京飞吉隆坡）。

    未定方式时整句回 1/2/3 分别是陆运/海运/空运；「1.陆运」仍按文字认。
    写了海运/陆运以书面为准。只说飞就是空运，补问件数时不得再问运输方式。
    「10件」「1.件数」不是运输方式。
    """
    raw = text or ""
    found = [code for word, code in _MODE_WORDS if word in raw]
    if len(found) == 1:
        return found[0]
    if len(found) > 1:
        return ""
    hit = _MODE_INDEX.match(raw.strip())
    if hit:
        return _MODE_BY_INDEX.get(hit.group(1), "")
    if "飞" in raw:
        return "AIR"
    return ""


def parse_clarify_paste_facts(text: str) -> dict[str, str]:
    """
    认补问复制清单：编号 + 字段名 +（单位/必填）+ 冒号后的值。

    「(如：10）」整行丢掉。冒号后为空也不收，避免把示例当成已填。
    不猜港口，不改六态。
    """
    facts: dict[str, str] = {}
    for line in (text or "").splitlines():
        raw = (line or "").strip()
        if not raw or _CLARIFY_EXAMPLE_LINE.match(raw):
            continue
        raw = _CLARIFY_INDEX.sub("", raw, count=1)
        if "：" not in raw and ":" not in raw:
            continue
        if "：" in raw:
            name, val = raw.split("：", 1)
        else:
            name, val = raw.split(":", 1)
        name = _CLARIFY_DECOR.sub("", name).strip()
        val = (val or "").strip()
        if not name or name not in _CLARIFY_NAMES:
            continue
        if not val or _CLARIFY_EXAMPLE_LINE.match(val):
            continue
        facts[name] = clip_inquiry_value(val)
    return {key: val for key, val in facts.items() if val}


def parse_labeled_facts(text: str) -> dict[str, str]:
    """
    只采集「字段名：值」，整行优先；同一句里的标签也收。不猜港口。

    补问复制清单（1.件数（必填）：10）优先。
    报价单常把表头和格子写在一行（起运港/POL（必填）  纽约港），
    冒号对不上时再按表头切一刀，值只留格子。
    """
    facts: dict[str, str] = parse_clarify_paste_facts(text)
    raw = "\n".join(
        line
        for line in (text or "").splitlines()
        if not _CLARIFY_EXAMPLE_LINE.match(line.strip())
    )
    for line in raw.splitlines():
        shown = line.strip()
        m = _LABEL_LINE.match(shown)
        if m and not str(facts.get(m.group(1)) or "").strip():
            facts[m.group(1)] = clip_sheet_field_value(m.group(2).strip())
    if not facts:
        for m in _LABEL_INLINE.finditer(raw):
            facts[m.group(1)] = clip_sheet_field_value(m.group(2).strip())
    from agent.schema.field_validate import normalize_facts

    taken = set(normalize_facts(facts))
    for key, val in harvest_sheet_inquiry_cells(raw).items():
        if not val:
            continue
        mapped = next(iter(normalize_facts({key: val})), "")
        if mapped and mapped in taken:
            continue
        if str(facts.get(key) or "").strip():
            continue
        facts[key] = val
        if mapped:
            taken.add(mapped)
    return {key: val for key, val in facts.items() if val}


def _merge_spoken_collab(text: str, facts: dict[str, str]) -> dict[str, str]:
    """
    私聊已经说到的协同项写进事实，进群才能预填。

    不覆盖已有询价键。不拿去补问、不挡查价。
    """
    from agent.schema.field_validate import (
        harvest_oral_collab,
        harvest_spoken_trade_term,
        normalize_facts,
    )

    merged = dict(facts or {})
    extra = dict(harvest_oral_collab(text))
    extra.update(harvest_spoken_trade_term(text))
    extra.update(normalize_facts(harvest_collab_labels(text)))
    for key, val in extra.items():
        if val and not str(merged.get(key) or "").strip():
            merged[key] = val
    return merged


def prepare_sheet_oral(text: str) -> str:
    """
    报价单只留询价栏。

    费用明细、报价条件、落款整段丢掉。表头里被空格拉开的「体    积」收回成「体积」，
    否则上一栏会把后面的报价条件整段吞进去。计费重不是毛重，单独拿掉。
    副作用：无。
    """
    body = _tighten_sheet_labels(text or "")
    cut = len(body)
    for mark in _QUOTE_BODY_MARKS:
        pos = body.find(mark)
        if 0 <= pos < cut:
            cut = pos
    body = body[:cut]
    body = re.sub(r"计费重\s*[:：]?\s*[0-9.]+\s*(?:KG|KGS|kg|公斤)?", " ", body)
    return body.strip()


def _tighten_sheet_labels(text: str) -> str:
    """已知表头中间只允许空格。不跨行，避免把两行并成一个词。"""
    body = text or ""
    gap = r"[ \t\u3000]{0,6}"
    for label in _TIGHTEN_LABELS:
        if len(label) < 2:
            continue
        pat = re.compile(
            r"(?<![\u4e00-\u9fffA-Za-z0-9])" + gap.join(re.escape(ch) for ch in label)
        )
        body = pat.sub(label, body)
    return body


def _drop_fee_place_values(facts: dict[str, str]) -> dict[str, str]:
    """起运地/目的地收成「操作费、文件费」时丢掉。那是费用栏，不是港口。"""
    out = dict(facts or {})
    for key in ("起运港", "起运地", "目的港", "目的地"):
        val = str(out.get(key) or "").strip()
        if not val:
            continue
        bits = [p.strip(" *：:\t") for p in re.split(r"[\n/、,，]+", val) if p.strip()]
        if bits and all(_FEE_NAME.match(p) for p in bits):
            out.pop(key, None)
    return out


def extract_inquiry_snapshot(
    text: str,
    *,
    injected_facts: Optional[dict[str, Any]] = None,
    injected_mode: str = "",
    b_result: Optional[dict[str, Any]] = None,
    allow_b: bool = False,
    chat_fn: Optional[Any] = None,
) -> dict[str, Any]:
    """
    抽出业务线 + 字段快照。

    优先 injected_*（单测）；其次合并标签行与 DeepSeek B（口语）。
    标签行覆盖 B 的同名字段。B 未出现的字段保持空，不本地猜港口。
    件数/毛重/体积若原话已带 TMS 单位（件/KGS/CBM），空字段可机读补上。
    """
    from agent.schema.field_validate import harvest_oral_collab, harvest_oral_measures, normalize_facts

    if injected_facts is not None:
        mode = (injected_mode or detect_transport_mode(text) or "").upper()
        facts = harvest_oral_measures(text, normalize_facts(dict(injected_facts)))
        facts = _merge_spoken_collab(text, facts)
        facts = _merge_optional_inquiry_labels(text, facts)
        for key in ("货好时间", "提货地址", "收货地址", "货源地", "贸易条款"):
            if str(facts.get(key) or "").strip():
                facts[key] = clip_inquiry_value(str(facts[key]))
        facts = _drop_fee_place_values(_clip_sheet_inquiry_facts(facts))
        subtype = str(facts.get("运输类型") or "").strip()
        return {
            "business_line": mode,
            "land_subtype": subtype,
            "facts": facts,
            "source": "injected",
        }
    # 附件常是整张报价单。先丢掉费用明细和条款，再抽询价栏。
    text = prepare_sheet_oral(text)
    mode = detect_transport_mode(text)
    facts = parse_labeled_facts(text)
    source = "labeled_or_mode"
    b_payload = b_result
    if b_payload is None and (allow_b or chat_fn is not None):
        from agent.llm.mode_extract_fields import invoke_extract_fields

        b_payload = invoke_extract_fields(
            text, allow_network=allow_b, chat_fn=chat_fn
        )
    if isinstance(b_payload, dict) and not b_payload.get("ok"):
        err = str(b_payload.get("error") or "").strip()
        if err not in {"b_network_disabled", "extract_fields_shell_not_implemented"} and (
            allow_b or chat_fn is not None
        ):
            return {
                "business_line": mode,
                "land_subtype": "",
                "facts": {},
                "source": "llm_tech_fail",
                "tech_fail": True,
                "tech_error": err or "LLM_FAIL",
            }
    if b_payload and b_payload.get("ok"):
        merged = dict(b_payload.get("facts") or {})
        merged.update(facts)
        facts = merged
        # 运输方式只认原话里的空运/海运/陆运、1/2/3、飞。
        # 模型常把「郑州-吉隆坡」猜成空运，不能拿来默认。
        source = "deepseek_b"
    facts = harvest_oral_measures(text, normalize_facts(facts))
    facts = _merge_spoken_collab(text, facts)
    facts = _merge_optional_inquiry_labels(text, facts)
    for key in ("货好时间", "提货地址", "收货地址", "货源地", "贸易条款"):
        if str(facts.get(key) or "").strip():
            facts[key] = clip_inquiry_value(str(facts[key]))
    facts = _drop_fee_place_values(_clip_sheet_inquiry_facts(facts))
    land_subtype = str((b_payload or {}).get("land_subtype") or facts.get("运输类型") or "").strip()
    if land_subtype and not str(facts.get("运输类型") or "").strip():
        facts["运输类型"] = land_subtype
    if not mode and facts.get("起运港") and facts.get("目的港") and "飞" in (text or ""):
        mode = "AIR"
    logger.info("extract.snapshot mode=%s keys=%s src=%s", mode or "-", sorted(facts), source)
    return {
        "business_line": mode,
        "land_subtype": land_subtype,
        "facts": facts,
        "source": source,
    }


def harvest_collab_labels(text: str) -> dict[str, str]:
    """
    群里协同标签：有无冒号都收。

    「客户名称张三 包装方式编织袋 货值 10万」切成三项。
    只说「纸箱」没有标签则不收。值取到下一个标签之前。
    「(如：随时可提）」整行丢掉，避免补问示例被当成已填。
    """
    kept = "\n".join(
        line
        for line in (text or "").splitlines()
        if not _CLARIFY_EXAMPLE_LINE.match(line.strip())
    )
    return _harvest_labels(kept, _COLLAB_LABELS)


def harvest_inquiry_optional_labels(text: str) -> dict[str, str]:
    """
    陆运询价非必填标签：通关口岸。

    核对卡写「通关口岸（非必填）：」，销售可能照抄或只写「通关口岸皇岗」。
    有无冒号都收。不猜口岸名。
    """
    return _harvest_labels(text, _OPTIONAL_INQUIRY_LABELS)


def clip_inquiry_value(raw: str) -> str:
    """
    字段值收到下一个编号项为止。

    图上常在货好时间后面接着「4. 希望送到…」「6. 送货地址」。
    那些不是询价字段，不能进补问。地址里的「602、603」没有点号后空格，会留下。
    """
    text = (raw or "").strip()
    if not text:
        return ""
    text = re.sub(r"[（(]如\s*[:：][^）)]*[）)]?\s*$", "", text).strip()
    if _CLARIFY_EXAMPLE_LINE.match(text):
        return ""
    hit = _NUMBERED_ITEM.search(text)
    if hit and hit.start() > 0:
        text = text[: hit.start()]
    text = re.sub(r"(?:\n|\r)+\s*\d+[\.、．]\s*$", "", text)
    text = text.strip()
    # 复制补问清单时空着的下一行是「4.提货地址」。切到下一栏后只剩编号，不能当成货好时间。
    if re.fullmatch(r"\d+[\.、．]", text):
        return ""
    return text


def clip_sheet_field_value(raw: str) -> str:
    """
    报价单格子：去掉 /POL、（必填说明）、Terms 和同一行下一列表头，只留紧挨的值。

    测服手工海运单：起运港/POL（必填）  纽约港  目的港…  巴生西港  费用汇总 → 纽约港 / 巴生西港。
    取第一格有效值，不要最后一格（最后一格常是费用汇总、提货地址）。
    地址字段不要走这里。
    """
    text = clip_inquiry_value(raw)
    if not text:
        return ""
    text = _SHEET_EN_ALIAS.sub("", text)
    text = _SHEET_REQUIRED_HINT.sub("", text)
    text = _SHEET_REQUIRED_HINT_EN.sub("", text)
    text = re.sub(r"^\((?:KG|CBM)\)\s*[:：]?\s*", "", text, flags=re.I)
    if "必填" in text:
        text = re.sub(r"^[^）\n]*必填[^）\n]*）\s*", "", text)
    text = re.sub(r"^[\s:：]+", "", text)
    for label in _SHEET_CLEAN_KEYS:
        idx = 0
        while True:
            pos = text.find(label, idx)
            if pos < 0:
                break
            if pos > 0 and _label_is_token(text, pos, label):
                text = text[:pos]
                break
            idx = pos + len(label)
    for stop in _SHEET_STOP:
        pos = text.find(stop)
        if pos >= 0:
            text = text[:pos]
    text = text.strip().strip("：:/*")
    parts = [p.strip().strip("：:/*") for p in re.split(r"[\t]+|\s{2,}", text) if p.strip().strip("：:/*")]
    parts = [p for p in parts if p and not _SHEET_HEADER_NOISE.search(p)]
    if not parts:
        return ""
    # 格子里只剩下一个字段名（体积、时效要求）时，这一栏是空的，不能把字段名当成值。
    # 只剩「4.」也是空的：那是下一行「4.提货地址」被切掉字段名后留下的编号。
    token = parts[0].strip().strip("：:")
    if token in _SHEET_CLEAN_KEYS or token in _TIGHTEN_LABELS:
        return ""
    if re.fullmatch(r"\d+[\.、．]", token):
        return ""
    return parts[0].strip()


def harvest_sheet_inquiry_cells(text: str) -> dict[str, str]:
    """
    按询价字段名切报价单原文。标签必须是独立词，不能切「指定起运港口」的中间。

    一行两个字段时，值收到下一个字段名为止。副作用：无。
    """
    raw = text or ""
    hits: list[tuple[int, str]] = []
    for label in _SHEET_CLEAN_KEYS:
        start = 0
        while True:
            pos = raw.find(label, start)
            if pos < 0:
                break
            if _label_is_token(raw, pos, label):
                hits.append((pos, label))
            start = pos + len(label)
    if not hits:
        return {}
    hits.sort()
    kept: list[tuple[int, str]] = []
    for pos, label in hits:
        if kept and kept[-1][0] == pos and len(label) <= len(kept[-1][1]):
            continue
        if kept and kept[-1][0] == pos:
            kept[-1] = (pos, label)
            continue
        kept.append((pos, label))
    out: dict[str, str] = {}
    for i, (pos, label) in enumerate(kept):
        begin = pos + len(label)
        end = kept[i + 1][0] if i + 1 < len(kept) else len(raw)
        val = clip_sheet_field_value(raw[begin:end])
        if val:
            out[label] = val
    return out


def _label_is_token(raw: str, pos: int, label: str) -> bool:
    """
    标签必须是独立词。

    「指定起运港口」中间的起运港不要。重量(KG)/体积(CBM) 留给更长标签。
    """
    if pos > 0:
        prev = raw[pos - 1]
        if prev.isalnum() or "\u4e00" <= prev <= "\u9fff":
            return False
    end = pos + len(label)
    nxt = raw[end : end + 6]
    if label in {"重量", "体积"} and (nxt.startswith("(") or nxt.startswith("（")):
        return False
    return True


def _clip_sheet_inquiry_facts(facts: dict[str, str]) -> dict[str, str]:
    """已抽出的询价键再洗一遍表头。不改地址。"""
    out = dict(facts or {})
    for key in list(out):
        if key not in _SHEET_CLEAN_KEYS:
            continue
        val = str(out.get(key) or "").strip()
        if not val:
            continue
        out[key] = clip_sheet_field_value(val)
    return {key: val for key, val in out.items() if str(val or "").strip()}


def _merge_optional_inquiry_labels(text: str, facts: dict[str, str]) -> dict[str, str]:
    """把原话里的通关口岸、货好时间、提货/收货地址写进询价事实。有值才覆盖。"""
    from agent.schema.field_validate import normalize_facts

    merged = dict(facts or {})
    extra = normalize_facts(harvest_inquiry_optional_labels(text))
    for key, val in extra.items():
        cleaned = clip_inquiry_value(val)
        if key in _SHEET_CLEAN_KEYS:
            cleaned = clip_sheet_field_value(cleaned)
        if not cleaned:
            continue
        old = str(merged.get(key) or "").strip()
        if old and cleaned.startswith(old) and len(cleaned) > len(old):
            continue
        if old and ("必填" in cleaned or "\n" in cleaned or len(cleaned) > len(old) * 3):
            continue
        merged[key] = cleaned
    return merged


def _harvest_labels(text: str, labels: tuple[str, ...]) -> dict[str, str]:
    """按标签切原话。同一起点只留最长标签。"""
    raw = text or ""
    hits: list[tuple[int, str]] = []
    for label in labels:
        start = 0
        while True:
            pos = raw.find(label, start)
            if pos < 0:
                break
            if _label_is_token(raw, pos, label):
                hits.append((pos, label))
            start = pos + len(label)
    if not hits:
        return {}
    hits.sort()
    # 同一起点只留最长标签，避免 HS 抢 HS编码、通关口岸抢「通关口岸（非必填）」。
    kept: list[tuple[int, str]] = []
    for pos, label in hits:
        if kept and kept[-1][0] == pos and len(label) <= len(kept[-1][1]):
            continue
        if kept and kept[-1][0] == pos:
            kept[-1] = (pos, label)
            continue
        kept.append((pos, label))
    out: dict[str, str] = {}
    for i, (pos, label) in enumerate(kept):
        begin = pos + len(label)
        end = kept[i + 1][0] if i + 1 < len(kept) else len(raw)
        chunk = raw[begin:end]
        chunk = re.sub(r"^[\s:：,，、]+", "", chunk)
        chunk = re.sub(r"[\s,，、]+$", "", chunk)
        if chunk:
            out[label] = chunk
    return out


def extract_collab_fields(
    text: str,
    *,
    allow_b: bool = False,
    injected: Optional[dict[str, str]] = None,
    allowed_keys: Optional[tuple[str, ...]] = None,
) -> dict[str, str]:
    """
    群里协同字段：标签（冒号可省）+ 口语含油/含电/含磁/危险品、货好时间。

    allowed_keys 传入则只返回本单该有的项。不传则海运协同五项 ∪ 贸易条款（陆运跨境仍要）∪ 陆运协同键，给意图分类。
    闲聊、只说「纸箱」不进这里。商品海关编码 / HS 折成 HS编码。
    """
    from agent.schema.field_validate import (
        ALL_COLLAB_KEYS,
        harvest_oral_collab,
        normalize_facts,
    )

    del allow_b
    keys = tuple(allowed_keys) if allowed_keys is not None else ALL_COLLAB_KEYS
    if injected:
        return {
            k: str(v).strip()
            for k, v in normalize_facts(dict(injected)).items()
            if k in keys and str(v or "").strip()
        }
    labeled = normalize_facts(parse_labeled_facts(text))
    labeled.update(normalize_facts(harvest_collab_labels(text)))
    oral = harvest_oral_collab(text)
    if "是否为危险品" in oral:
        if allowed_keys is not None and "是否为危险品" not in keys:
            oral.pop("是否为危险品", None)
        elif allowed_keys is None:
            raw = text or ""
            if "普货" in raw and not any(
                w in raw for w in ("危险", "非危", "有危", "含危", "是否为危险品")
            ):
                oral.pop("是否为危险品", None)
    merged = dict(oral)
    merged.update({k: v for k, v in labeled.items() if v})
    out: dict[str, str] = {}
    for key in keys:
        val = (merged.get(key) or "").strip()
        if val:
            out[key] = val
    return out


def extract_air_collab_fields(
    text: str,
    *,
    allow_b: bool = False,
    injected: Optional[dict[str, str]] = None,
) -> dict[str, str]:
    """
    空运群只认是否含电、是否含磁。

    复用海运口语收割，再滤成这两项。
    """
    keys = ("是否含电", "是否含磁")
    if injected:
        return {
            k: str(v).strip()
            for k, v in (injected or {}).items()
            if k in keys and str(v or "").strip()
        }
    got = extract_collab_fields(text, allow_b=allow_b)
    return {k: got[k] for k in keys if str(got.get(k) or "").strip()}


_RE_MONEY = re.compile(
    r"(?:USD|CNY|RMB|美金|美元|人民币)?\s*(\d+(?:\.\d+)?)\s*(?:USD|CNY|RMB|美金|美元|人民币|元)?"
    r"|(?:USD|CNY|RMB|美金|美元)\s*(\d+(?:\.\d+)?)",
    re.I,
)
_RE_FEE_LINE = re.compile(
    r"(?P<name>[\u4e00-\u9fffA-Za-z][\u4e00-\u9fffA-Za-z0-9/()（）+＋\-－]{0,24}费|[A-Za-z]{2,10})"
    r"\s*(?:改为|改成|调整为|调整成)?"
    r"\s*[:：=]?"
    r"\s*(?:(?P<ccy1>USD|CNY|RMB|美金|美元|人民币)\s+)?"
    r"(?P<amount>\d+(?:\.\d+)?)"
    r"\s*(?P<ccy2>USD|CNY|RMB|美金|美元|人民币|元)?",
    re.I,
)
# 「文件发改为：500」漏了「费」、中间还有冒号，不能丢掉。
_RE_FEE_CHANGE = re.compile(
    r"(?P<name>[\u4e00-\u9fff]{2,16}?)"
    r"\s*(?:改为|改成|调整为|调整成)"
    r"\s*[:：=]?"
    r"\s*(?P<amount>\d+(?:\.\d+)?)"
    r"\s*(?P<ccy>USD|CNY|RMB|美金|美元|人民币|元)?",
    re.I,
)
_SKIP_FEE_NAMES = frozenset({"usd", "cny", "rmb", "美金", "美元", "人民币", "元"})
# 航线常写「空运费 单价CNY99，数量100，金额CNY199」。单价和金额不是同一个数，必须认金额。
_RE_FEE_PRICED = re.compile(
    r"(?P<name>[\u4e00-\u9fffA-Za-z][\u4e00-\u9fffA-Za-z0-9/()（）+＋\-－]{0,24}费)"
    r"\s*单价\s*"
    r"(?:(?P<unit_ccy>USD|CNY|RMB|美金|美元|人民币)\s*)?"
    r"(?P<unit>\d+(?:\.\d+)?)"
    r"(?:\s*[，,、]\s*数量\s*(?P<qty>\d+(?:\.\d+)?))?"
    r"(?:\s*[，,、]\s*金额\s*"
    r"(?:(?P<amt_ccy>USD|CNY|RMB|美金|美元|人民币)\s*)?"
    r"(?P<amount>\d+(?:\.\d+)?))?",
    re.I,
)
_FULLWIDTH_NUM = str.maketrans("０１２３４５６７８９．", "0123456789.")


def _ascii_amount(raw: str) -> str:
    """全角数字折成半角，方便和模板金额列对齐。"""
    return (raw or "").translate(_FULLWIDTH_NUM).strip()


def _currency_of(text: str, hint: str = "") -> str:
    blob = f"{hint} {text}".upper()
    if any(x in blob for x in ("CNY", "RMB", "人民币", "元")):
        return "CNY"
    if any(x in blob for x in ("USD", "美金", "美元")):
        return "USD"
    return hint or "USD"


def _parse_fee_rows(text: str) -> list[dict[str, str]]:
    """
    按行抽费用：码头操作费100CNY、报关费改为300CNY。

    只认带「费」的中文名或短英文费用码，避免把工单号当金额。
    """
    raw = _ascii_amount(text or "")
    rows: list[dict[str, str]] = []
    seen: set[str] = set()

    def _add(name: str, amount: str, ccy: str, **extra: str) -> None:
        if not name or not amount or name.lower() in _SKIP_FEE_NAMES:
            return
        # 「WO202609160001」会被 [A-Za-z]{2,10}+数字 抽成费用名 WO，群里发工单号不能当报价。
        if name.upper() == "WO":
            return
        if name in seen:
            return
        seen.add(name)
        row = {
            "name": name,
            "amount": amount,
            "currency": _currency_of(raw, ccy),
        }
        for key, val in extra.items():
            if val:
                row[key] = val
        rows.append(row)

    for hit in _RE_FEE_PRICED.finditer(raw):
        # 有「金额」用金额；只有单价时才用单价。禁止把单价误当成这一行的总金额。
        amount = (hit.group("amount") or "").strip() or (hit.group("unit") or "").strip()
        ccy = hit.group("amt_ccy") or hit.group("unit_ccy") or ""
        _add(
            (hit.group("name") or "").strip(),
            _ascii_amount(amount),
            ccy,
            unit_price=_ascii_amount(hit.group("unit") or ""),
            quantity=_ascii_amount(hit.group("qty") or ""),
        )
    for hit in _RE_FEE_LINE.finditer(raw):
        _add(
            (hit.group("name") or "").strip(),
            _ascii_amount(hit.group("amount") or ""),
            hit.group("ccy1") or hit.group("ccy2") or "",
        )
    for hit in _RE_FEE_CHANGE.finditer(raw):
        _add(
            (hit.group("name") or "").strip(),
            _ascii_amount(hit.group("amount") or ""),
            hit.group("ccy") or "",
        )
    return rows


def extract_product_quote(text: str, *, injected: Optional[dict[str, Any]] = None) -> dict[str, Any]:
    """
    产品文字报价：先按费用行抽，再退回合计/币别。注入优先。

    认不出价格返回空 dict，调用方回「没有识别到报价」。
    群里常见「码头操作费100CNY / 报关费改为300CNY」，不能只把第一笔数当成海运费。
    「空运费 单价…金额…」必须保留空运费，金额用「金额」后面的数，不用单价。
    """
    if isinstance(injected, dict) and (injected.get("total") or injected.get("fee_rows")):
        out = dict(injected)
        out.setdefault("source_label", "产品报价")
        return out
    raw = (text or "").strip()
    if not raw:
        return {}
    stripped = raw.replace("@询价小助手", "").replace("询价小助手", "")
    compact = stripped.replace(" ", "").replace("　", "")
    if re.fullmatch(r"WO\d{12}", compact, re.I):
        return {}
    rows = _parse_fee_rows(raw)
    if rows:
        ccy = rows[0].get("currency") or _currency_of(raw)
        total = rows[0]["amount"]
        if len({x.get("currency") for x in rows}) == 1:
            try:
                total = str(sum(float(x["amount"]) for x in rows))
            except ValueError:
                total = rows[0]["amount"]
        return {
            "total": total,
            "currency": ccy,
            "source_label": "产品报价",
            "raw_text": raw,
            "fee_rows": rows,
            "fee_lines": list(rows),
        }
    hit = _RE_MONEY.search(_ascii_amount(raw))
    if not hit:
        return {}
    amount = _ascii_amount(hit.group(1) or hit.group(2) or "")
    if not amount:
        return {}
    ccy = _currency_of(raw)
    # 句里已经点了费用名（空运费）时，不能退回默认海运费。
    named = re.search(
        r"([\u4e00-\u9fffA-Za-z][\u4e00-\u9fffA-Za-z0-9/()（）+＋\-－]{0,24}费)",
        raw,
    )
    fee_name = (named.group(1) if named else "") or "海运费"
    if fee_name in {"费用"}:
        fee_name = "海运费"
    return {
        "total": amount,
        "currency": ccy,
        "source_label": "产品报价",
        "raw_text": raw,
        "fee_rows": [{"name": fee_name, "amount": amount, "currency": ccy}],
        "fee_lines": [{"name": fee_name, "amount": amount, "currency": ccy}],
    }