inquiry_robot/inquiry-agent/agent/policy/land_text_flow.py

"""
陆运文字询价确定性流程。

本文件职责：三项选项各自必填（不联动）→ 按类型补货物 → 无工单号核对卡 →（确定后）建单查 TMS。
查价后半段复用海运拉群/报价单/成交，拉人按线路类别，禁止陆运岗全员。
禁止：字段齐了就建单；直连 TMS；把海运「出卡即查价」改掉。
调用：handler 在 Worker/inbox 线程；禁止回调线程同步跑。
"""

from __future__ import annotations

import logging
import threading
from typing import Any, Optional

from agent.ledger.memory_ledger import MemoryLedger
from agent.llm.extract_text import extract_inquiry_snapshot
from agent.policy import inquiry_copy as copy
from agent.policy.air_text_flow import FlowSession, ReplyFn
from agent.policy.sea_text_flow import SeaTextInquiryFlow
from agent.schema.field_validate import harvest_oral_measures, validate_required_fields
from agent.schema.land_options import (
    harvest_land_options,
    looks_like_confirm,
    option_prompt,
    option_prompt_needs_sample,
    sample_pair_allowed,
    validate_land_options,
)
from agent.schema.tms_land_query import assemble_land_query

logger = logging.getLogger(__name__)

_LAND_FRONT = frozenset({"need_mode", "need_land_options", "clarify", "wait_confirm", "idle", ""})


class LandTextInquiryFlow(SeaTextInquiryFlow):
    """
    陆运私聊 Owner。

    前半段与海运不同：核对通过前不建单。后半段套海运建群，匹配改走线路类别。
    """

    bookmark_kind = "land"

    def on_text(
        self,
        *,
        sender_id: str,
        text: str,
        reply: ReplyFn,
        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,
        inbound_raw: Optional[dict[str, Any]] = None,
    ) -> str:
        """处理一句陆运私聊。返回阶段名。无 sender 拒绝。"""
        if not (sender_id or "").strip():
            return "reject_no_sender"
        sess = self.session_for_action(sender_id, inbound_raw, text=text)
        if (not sess or sess.phase != "wait_confirm") and looks_like_confirm(text):
            restored = self.restore_confirm_session(sender_id)
            if restored:
                sess = restored
            else:
                retry = self.restore_tms_retry_session(sender_id)
                if retry:
                    sess = retry
        sess = self._detach_frozen_bookmark(sess, sender_id)
        phase = (sess.phase if sess else "") or ""
        btn = self.match_button(text, sess)
        if btn:
            return self.on_button(
                sender_id=sender_id,
                action=btn,
                reply=reply,
                card_meta=inbound_raw,
            )
        if sess and sess.phase == "tms_tech" and (sess.work_order_no or "").strip():
            if looks_like_confirm(text):
                logger.info("land.tms_retry wo=%s sender=%s", sess.work_order_no, sender_id)
                return self._quote_existing_land(sess, reply)
        if sess and sess.phase == "wait_confirm" and looks_like_confirm(text):
            # 确定查价不再抽字段，避免「确定」被模型改成空运/空 facts。
            logger.info("land.confirm sender=%s", sender_id)
            return self._confirm_or_hold(
                sender_id=sender_id,
                facts=dict(sess.facts),
                text=text,
                reply=reply,
                sess=sess,
            )
        # 未成交后下一句任意文字即原因。不能走下面的 new_over_old，否则会当新询价丢掉。
        if (
            sess
            and sess.phase == "wait_lost_reason"
            and not (injected_facts and injected_mode)
        ):
            return self._capture_lost_reason(sess, text, reply)
        if sess:
            from agent.policy.quote_adjust_ops import try_private_adjust

            adjusted = try_private_adjust(self, sess, text, reply)
            if adjusted:
                return adjusted
        if sess and phase not in _LAND_FRONT:
            # 查价后半段书签还在时，销售又发新陆运：只走陆运前半段。
            # 禁止再 super 去海运 on_text，海运会把 LAND 转回来，递归到死、一句都不回。
            logger.info(
                "land.new_over_old phase=%s wo=%s",
                phase,
                sess.work_order_no or "-",
            )
            sess = None

        snap = extract_inquiry_snapshot(
            text,
            injected_facts=injected_facts,
            injected_mode=injected_mode,
            b_result=b_result,
            allow_b=allow_b,
            chat_fn=chat_fn,
        )
        mode = (snap.get("business_line") or "").upper()
        snap_facts = dict(snap.get("facts") or {})
        facts = dict(snap_facts)
        # DeepSeek B 把陆运类型放在 land_subtype，不在货物 facts 里。
        subtype = str(snap.get("land_subtype") or facts.get("运输类型") or "").strip()
        if subtype and not str(facts.get("运输类型") or "").strip():
            facts["运输类型"] = subtype
        option_keys = ("运输类型", "线路类别", "运输分类")
        base = dict(sess.facts) if sess else {}
        cargo = {k: v for k, v in facts.items() if k not in option_keys}
        merged = dict(base)
        merged.update(cargo)
        facts = harvest_oral_measures(text, merged)
        # 选项只从原话+能对上枚举的模型值收；禁止模型「陆运」盖掉已选的国内运输拼车。
        facts = harvest_land_options(text, facts)
        b_opt = " ".join(
            str(x).strip()
            for x in (
                snap.get("land_subtype"),
                *(snap_facts.get(k) for k in option_keys),
            )
            if str(x or "").strip()
        )
        if b_opt:
            facts = harvest_land_options(b_opt, facts)

        if not mode:
            mode = "LAND" if (sess and (sess.business_line or "").upper() == "LAND") else ""
        if not mode:
            reply(copy.ASK_TRANSPORT)
            self._save(
                FlowSession(
                    sender_id=sender_id,
                    thread_id=sender_id,
                    phase="need_mode",
                    business_line="",
                    facts=facts,
                    immutable_text=sess.immutable_text if sess and sess.immutable_text else text,
                )
            )
            return "need_mode"

        if mode != "LAND":
            if mode == "SEA":
                from agent.policy.sea_text_flow import get_sea_text_flow

                # 抽成海运后丢掉陆运选项草稿，下一句补字段不能再发陆运样例图。
                self.drop_current_bookmark(sender_id)
                return get_sea_text_flow().on_text(
                    sender_id=sender_id,
                    text=text,
                    reply=reply,
                    injected_facts=facts,
                    injected_mode=mode,
                    b_result=b_result,
                    allow_b=allow_b,
                    chat_fn=chat_fn,
                    inbound_raw=inbound_raw,
                )
            from agent.policy.air_text_flow import get_air_text_flow

            self.drop_current_bookmark(sender_id)
            return get_air_text_flow().on_text(
                sender_id=sender_id,
                text=text,
                reply=reply,
                injected_facts=facts,
                injected_mode=mode,
                b_result=b_result,
                allow_b=allow_b,
                chat_fn=chat_fn,
                inbound_raw=inbound_raw,
            )

        return self._advance_front(
            sender_id=sender_id,
            facts=facts,
            text=text,
            reply=reply,
            sess=sess,
        )

    def _reply_land_options(
        self,
        reply: ReplyFn,
        facts: dict[str, str],
        *,
        message: str = "",
    ) -> None:
        """
        选项补问：先发文字清单，需要样例时再发表格图。

        图片失败由出站回退文字样例。不在本方法上传素材。
        """
        body = (message or "").strip() or option_prompt(current=facts)
        reply(body)
        if option_prompt_needs_sample(current=facts):
            reply(copy.land_sample_image_fallback(), copy.land_sample_image_payload())

    def _confirm_or_hold(
        self,
        *,
        sender_id: str,
        facts: dict[str, str],
        text: str,
        reply: ReplyFn,
        sess: FlowSession,
    ) -> str:
        """确定意图：字段仍齐则建单查价。"""
        check = self._land_ready(facts)
        if not check["ready"]:
            return self._advance_front(
                sender_id=sender_id,
                facts=facts,
                text=text,
                reply=reply,
                sess=sess,
            )
        return self._create_and_quote_land(
            sender_id=sender_id,
            facts=check["facts"],
            text=sess.immutable_text or text,
            reply=reply,
        )

    def _create_and_quote_land(
        self,
        *,
        sender_id: str,
        facts: dict[str, str],
        text: str,
        reply: ReplyFn,
    ) -> str:
        """
        销售确定之后才建陆运工单并查 TMS。

        核对卡阶段没有工单号。运输分类写入工单 facts，不进入 assemble 出站。
        """
        import uuid

        tms_ready = assemble_land_query(facts)
        if not tms_ready.get("ok"):
            return self._advance_front(
                sender_id=sender_id,
                facts=facts,
                text=text,
                reply=reply,
                sess=self.session_of(sender_id),
            )
        self._save(
            FlowSession(
                sender_id=sender_id,
                thread_id=sender_id,
                phase="quoting",
                business_line="LAND",
                facts=facts,
                immutable_text=text,
            )
        )
        created = self._ledger.create_ticket(
            sender_id=sender_id,
            business_line="LAND",
            facts=facts,
            idempotency_key=f"create:{sender_id}:{uuid.uuid4().hex}",
        )
        no = str(created.get("work_order_no") or "").strip()
        logger.info("land.create wo=%s sender=%s", no or "-", sender_id)
        if not no:
            reply("建单失败，请稍后把刚才那条再发一次。陆运字段已经记下。")
            return "ledger_fail"
        self._clear_parked_confirm(sender_id)
        first, history_wo = self._resolve_similar(
            created, facts=facts, current_no=no, business_line="LAND"
        )
        # 与海运一样先出询价确认（工单号+字段），再查价；禁止只回一句「工单已创建」。
        out_id = reply(
            copy.inquiry_card(
                work_order_no=no,
                facts=facts,
                first_or_same=first,
                transport_mode="陆运",
            )
        ) or ""
        self._wait_inquiry_sent(str(out_id), work_order_no=no)

        tms = self._ledger.query_tms(work_order_no=no, facts=facts, business_line="LAND")
        sess = FlowSession(
            sender_id=sender_id,
            thread_id=sender_id,
            business_line="LAND",
            facts=facts,
            work_order_no=no,
            first_or_same=first,
            history_work_order_no=history_wo,
            immutable_text=text,
            wait_version=1,
        )
        return self._apply_land_tms(
            sess=sess,
            tms=tms,
            reply=reply,
            retrying=False,
        )

    def _quote_existing_land(self, sess: FlowSession, reply: ReplyFn) -> str:
        """工单已在：确定后只重查 TMS，禁止再开一单。"""
        no = (sess.work_order_no or "").strip()
        facts = dict(sess.facts or {})
        if not no or not facts:
            reply(copy.LAND_CONFIRM_EXPIRED)
            return "land_confirm_expired"
        reply(copy.LAND_TMS_RETRYING.format(wo=no))
        tms = self._ledger.query_tms(work_order_no=no, facts=facts, business_line="LAND")
        return self._apply_land_tms(sess=sess, tms=tms, reply=reply, retrying=True)

    def _apply_land_tms(
        self,
        *,
        sess: FlowSession,
        tms: dict[str, Any],
        reply: ReplyFn,
        retrying: bool,
    ) -> str:
        """查价回包落到有价/无价/技术失败。失败时把工单号记进待重查。"""
        _ = retrying
        no = (sess.work_order_no or "").strip()
        if not tms.get("ok"):
            reply("TMS 查询失败，请稍后重试。工单已创建：" + no)
            sess.phase = "tms_tech"
            self._save(sess)
            self._park_tms_retry(sess)
            return "tms_tech"
        self._clear_parked_confirm(sess.sender_id)
        if not tms.get("has_price"):
            self._emit_tms_miss(reply, no)
            sess.phase = "tms_miss"
            sess.status = "询价中"
            sess.allowed = (copy.BTN_PULL_COLLAB,)
            self._save(sess)
            return "tms_miss"

        quote = tms["quote"]
        up = self._ledger.upsert_quote(work_order_no=no, quote=quote, to_status="已报价")
        if not up.get("ok"):
            reply("主账写入报价失败，不能当作已报价。")
            sess.phase = "ledger_fail"
            self._save(sess)
            return "ledger_fail"
        sess.quote = quote
        copy.remember_tms_quote(sess, quote)
        sess.status = "已报价"
        sess.phase = "wait_collab"
        sess.allowed = (copy.BTN_PULL_COLLAB, copy.BTN_SKIP_COLLAB)
        sess.wait_version += 1
        self._save(sess)
        self._emit_sea_hit(
            reply,
            work_order_no=no,
            quote=quote,
            first_or_same=sess.first_or_same,
            history_work_order_no=sess.history_work_order_no,
            status="已报价",
            business_line=sess.business_line or "LAND",
        )
        return "wait_collab"

    def _land_ready(self, facts: dict[str, str]) -> dict[str, Any]:
        """选项+货物是否可出核对卡。"""
        opt = validate_land_options(facts)
        merged = dict(facts)
        merged.update(opt.get("normalized") or {})
        if opt.get("errors"):
            return {"ready": False, "opt": opt, "facts": merged, "stage": "options_error"}
        if not opt.get("ok"):
            return {"ready": False, "opt": opt, "facts": merged, "stage": "options_missing"}
        land_type = (merged.get("运输类型") or "").strip()
        check = validate_required_fields(
            facts=merged, business_line="LAND", land_subtype=land_type
        )
        cargo = dict(check["facts"])
        cargo.update({k: merged[k] for k in ("运输类型", "线路类别", "运输分类") if merged.get(k)})
        tms = assemble_land_query(cargo)
        cargo = dict(tms.get("facts") or cargo)
        cargo.update({k: merged[k] for k in ("运输类型", "线路类别", "运输分类") if merged.get(k)})
        missing = list(check.get("missing") or [])
        for key in tms.get("missing") or []:
            if key not in missing and key not in {"运输类型", "线路类别"}:
                missing.append(key)
        if not check["ok"] or not tms.get("ok"):
            return {
                "ready": False,
                "opt": opt,
                "facts": cargo,
                "stage": "clarify",
                "missing": missing,
                "land_type": land_type,
            }
        return {"ready": True, "facts": cargo, "opt": opt, "land_type": land_type}

    def _advance_front(
        self,
        *,
        sender_id: str,
        facts: dict[str, str],
        text: str,
        reply: ReplyFn,
        sess: Optional[FlowSession],
    ) -> str:
        """选项与货物往前推，出补问或核对卡。不建单。"""
        state = self._land_ready(facts)
        facts = dict(state["facts"])
        immutable = sess.immutable_text if sess and sess.immutable_text else text
        if state.get("ready"):
            reply(copy.land_confirm_card(facts))
            self._save(
                FlowSession(
                    sender_id=sender_id,
                    thread_id=sender_id,
                    phase="wait_confirm",
                    business_line="LAND",
                    facts=facts,
                    immutable_text=immutable,
                    wait_version=(sess.wait_version + 1) if sess else 1,
                )
            )
            self._park_confirm(sender_id, facts=facts, immutable_text=immutable)
            return "wait_confirm"
        if state.get("stage") == "options_error":
            err = (state["opt"].get("errors") or [{}])[0]
            self._reply_land_options(
                reply,
                facts,
                message=str(err.get("message") or ""),
            )
            self._save(
                FlowSession(
                    sender_id=sender_id,
                    thread_id=sender_id,
                    phase="need_land_options",
                    business_line="LAND",
                    facts=facts,
                    immutable_text=immutable,
                    wait_version=(sess.wait_version + 1) if sess else 1,
                )
            )
            self._clear_parked_confirm(sender_id)
            return "need_land_options"
        if state["stage"] == "options_missing":
            keys = list((state.get("opt") or {}).get("missing") or [])
            logger.info("land.options_missing missing=%s", keys)
            new_sess = FlowSession(
                sender_id=sender_id,
                thread_id=sender_id,
                phase="need_land_options",
                business_line="LAND",
                facts=facts,
                immutable_text=immutable,
                wait_version=(sess.wait_version + 1) if sess else 1,
            )
            from agent.policy.handoff_human import inherit_pre_create_ticket, note_clarify_round

            if sess is not None:
                new_sess.clarify_missing_sig = sess.clarify_missing_sig
                new_sess.clarify_retry = sess.clarify_retry
                new_sess.handoff_offer = sess.handoff_offer
                new_sess.handoff_committed = sess.handoff_committed
                new_sess.virtual_work_order_no = sess.virtual_work_order_no
                inherit_pre_create_ticket(sess, new_sess)
            decision = note_clarify_round(new_sess, keys)
            self._reply_land_options(
                reply,
                facts,
                message=copy.with_clarify_progress(
                    option_prompt(current=facts), new_sess.clarify_retry
                ),
            )
            if decision == "handoff":
                self._clear_parked_confirm(sender_id)
                return self._offer_handoff_now(
                    sender_id=sender_id,
                    reason="clarify",
                    reply=reply,
                    sess=new_sess,
                )
            self._save(new_sess)
            self._clear_parked_confirm(sender_id)
            return "need_land_options"
        new_sess = FlowSession(
            sender_id=sender_id,
            thread_id=sender_id,
            phase="clarify",
            business_line="LAND",
            facts=facts,
            immutable_text=immutable,
            wait_version=(sess.wait_version + 1) if sess else 1,
        )
        phase = self._clarify_or_handoff(
            sess=sess,
            keys=list(state.get("missing") or []),
            new_sess=new_sess,
            reply=reply,
            clarify_text=copy.ask_land_fields(
                facts=facts,
                missing_keys=state.get("missing") or [],
                land_subtype=state.get("land_type") or "",
            ),
        )
        self._clear_parked_confirm(sender_id)
        return phase

    def restore_confirm_session(self, sender_id: str) -> Optional[FlowSession]:
        """
        核对卡书签：先看进程内，没有再从短协调找回。

        双进程抢 inbox 或 HTTP 刚重启时，销售手里的核对卡还在，内存可能是空的。
        """
        hit = self.session_of(sender_id)
        if (
            hit
            and (hit.business_line or "").upper() == "LAND"
            and hit.phase == "wait_confirm"
        ):
            return hit
        from agent.redis_coord.pending_land_confirm import load_pending_land_confirm

        parked = load_pending_land_confirm(sender_id)
        if str(parked.get("phase") or "wait_confirm") == "tms_tech":
            return None
        facts_in = parked.get("facts") if isinstance(parked.get("facts"), dict) else {}
        facts = {str(k): str(v).strip() for k, v in dict(facts_in).items() if str(v).strip()}
        if not facts:
            return None
        sess = FlowSession(
            sender_id=sender_id,
            thread_id=sender_id,
            phase="wait_confirm",
            business_line="LAND",
            facts=facts,
            immutable_text=str(parked.get("immutable_text") or ""),
            wait_version=int(parked.get("wait_version") or 1),
        )
        self._save(sess)
        logger.info("land.confirm restored sender=%s", sender_id)
        return sess

    def restore_tms_retry_session(self, sender_id: str) -> Optional[FlowSession]:
        """
        查价失败后等确定重查。进程内存没有时从短协调找回工单号+字段。
        """
        hit = self.session_of(sender_id)
        if (
            hit
            and (hit.business_line or "").upper() == "LAND"
            and hit.phase == "tms_tech"
            and (hit.work_order_no or "").strip()
        ):
            return hit
        from agent.redis_coord.pending_land_confirm import load_pending_land_confirm

        parked = load_pending_land_confirm(sender_id)
        if str(parked.get("phase") or "") != "tms_tech":
            return None
        no = str(parked.get("work_order_no") or "").strip()
        facts_in = parked.get("facts") if isinstance(parked.get("facts"), dict) else {}
        facts = {str(k): str(v).strip() for k, v in dict(facts_in).items() if str(v).strip()}
        if not no or not facts:
            return None
        sess = FlowSession(
            sender_id=sender_id,
            thread_id=sender_id,
            phase="tms_tech",
            business_line="LAND",
            facts=facts,
            work_order_no=no,
            immutable_text=str(parked.get("immutable_text") or ""),
            wait_version=int(parked.get("wait_version") or 1),
        )
        self._save(sess)
        logger.info("land.tms_retry restored sender=%s wo=%s", sender_id, no)
        return sess

    def _park_tms_retry(self, sess: FlowSession) -> None:
        """查价失败：记下工单号，下一句确定只重查不建单。"""
        from agent.redis_coord.pending_land_confirm import save_pending_land_confirm

        save_pending_land_confirm(
            sess.sender_id,
            facts=dict(sess.facts or {}),
            immutable_text=sess.immutable_text or "",
            wait_version=int(sess.wait_version or 1),
            phase="tms_tech",
            work_order_no=sess.work_order_no or "",
        )

    def _park_confirm(
        self,
        sender_id: str,
        *,
        facts: dict[str, str],
        immutable_text: str,
    ) -> None:
        """出核对卡时记下字段。失败只打日志，不挡出卡。"""
        from agent.redis_coord.pending_land_confirm import save_pending_land_confirm

        save_pending_land_confirm(
            sender_id,
            facts=facts,
            immutable_text=immutable_text,
            phase="wait_confirm",
        )

    def _clear_parked_confirm(self, sender_id: str) -> None:
        """离开核对卡或已建单：清掉待确定，避免旧卡「确定」再开一单。"""
        from agent.redis_coord.pending_land_confirm import clear_pending_land_confirm

        clear_pending_land_confirm(sender_id)

    def _pull_collab_group(
        self,
        sess: FlowSession,
        reply: ReplyFn,
        card_meta: Optional[dict[str, Any]] = None,
    ) -> str:
        """
        点「拉产品进群协同」：先核填写样例，再按线路拉人建群。

        类型+线路对不上样例：不建群，告诉销售重新开单。按钮保持可点。
        对得上则走海运同一套建群。
        """
        facts = dict(sess.facts or {})
        land_type = str(facts.get("运输类型") or "").strip()
        route = str(facts.get("线路类别") or "").strip()
        if not sample_pair_allowed(land_type, route):
            logger.info(
                "land.group.reject pair type=%s route=%s wo=%s",
                land_type or "-",
                route or "-",
                sess.work_order_no or "-",
            )
            reply(copy.land_group_pair_rejected())
            return sess.phase or "wait_collab"
        return super()._pull_collab_group(sess, reply, card_meta=card_meta)

    def _match_sea_staff(self, sess: FlowSession) -> list[dict[str, Any]]:
        """按本单线路类别整词匹配产品·陆运，禁止全员。"""
        matcher = getattr(self._ledger, "match_land_staff", None)
        if not callable(matcher):
            return []
        route = str((sess.facts or {}).get("线路类别") or "").strip()
        try:
            rows = matcher(route_category=route) or []
        except Exception:
            logger.exception("匹配陆运产品人员失败 wo=%s route=%s", sess.work_order_no, route)
            return []
        return [dict(x) for x in rows if isinstance(x, dict)]

    def _group_no_staff_copy(self) -> str:
        """本线路没有陆运产品岗，不建空群。"""
        return copy.land_group_no_staff()

    def _group_brief_mode(self, sess: FlowSession) -> str:
        _ = sess
        return "陆运"


_FLOW: Optional[LandTextInquiryFlow] = None
_FLOW_GUARD = threading.Lock()


def get_land_text_flow() -> LandTextInquiryFlow:
    """进程内一份陆运流程。"""
    global _FLOW
    with _FLOW_GUARD:
        if _FLOW is None:
            _FLOW = LandTextInquiryFlow()
        return _FLOW


def reset_land_text_flow_for_test() -> LandTextInquiryFlow:
    """单测重置全局陆运实例。"""
    global _FLOW
    with _FLOW_GUARD:
        _FLOW = LandTextInquiryFlow(MemoryLedger())
        return _FLOW