陆运多条线路按段展示 TMS 费用名和时效，选定后再写回该条成本。

Co-authored-by: Cursor <cursoragent@cursor.com>

inquiry-agent/agent/channel/h5/land_route_store.py

@@ -0,0 +1,121 @@
+"""
+陆运选线路 H5 的短命 token。
+
+本文件职责：token 绑定 sender + 工单 + 线路报价；URL 禁止 userid。
+优先 Redis；单测走内存。
+禁止：在本包查 TMS、改六态。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import secrets
+import threading
+import time
+from typing import Any, Optional
+
+from agent.channel.h5.copy_store import inquiry_form_base
+from agent.config import get_settings
+from agent.redis_coord.keys import TOKEN_COPY_TICKET
+
+logger = logging.getLogger(__name__)
+
+LAND_ROUTE_TTL_SECONDS = 7 * 24 * 60 * 60
+_KEY = TOKEN_COPY_TICKET + ":land_route"
+_MEM: dict[str, tuple[str, float]] = {}
+_MEM_LOCK = threading.Lock()
+
+
+def issue_land_route_token(
+    *,
+    sender_id: str,
+    work_order_no: str,
+    quote: dict[str, Any],
+    facts: dict[str, str] | None = None,
+    ttl_seconds: int = LAND_ROUTE_TTL_SECONDS,
+) -> str:
+    """签发选线路 token。副作用：写 Redis 或内存。"""
+    payload = json.dumps(
+        {
+            "sender_id": sender_id,
+            "work_order_no": work_order_no,
+            "quote": quote,
+            "facts": dict(facts or {}),
+        },
+        ensure_ascii=False,
+    )
+    token = secrets.token_urlsafe(24)
+    if _redis_set(token, payload, ttl_seconds):
+        return token
+    expires = time.time() + ttl_seconds
+    with _MEM_LOCK:
+        _MEM[token] = (payload, expires)
+    logger.info("land route token 走内存 wo=%s", work_order_no)
+    return token
+
+
+def lookup_land_route_token(token: str) -> Optional[dict[str, Any]]:
+    """读 token。过期或不存在返回 None。"""
+    raw = (token or "").strip()
+    if not raw:
+        return None
+    text = _redis_get(raw)
+    if text is None:
+        with _MEM_LOCK:
+            hit = _MEM.get(raw)
+            if not hit:
+                return None
+            payload, expires = hit
+            if expires < time.time():
+                _MEM.pop(raw, None)
+                return None
+            text = payload
+    try:
+        data = json.loads(text)
+    except json.JSONDecodeError:
+        return None
+    return data if isinstance(data, dict) else None
+
+
+def build_land_route_url(token: str) -> str:
+    """拼选线路页：{inquiry-form}/land-routes/{token}。"""
+    raw = (token or "").strip()
+    if not raw:
+        return ""
+    return f"{inquiry_form_base().rstrip('/')}/land-routes/{raw}"
+
+
+def _redis_set(token: str, payload: str, ttl: int) -> bool:
+    client = _try_redis()
+    if client is None:
+        return False
+    try:
+        client.raw.set(client.key(_KEY, token), payload, ex=ttl)
+        return True
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("land route token Redis 写入失败：%s", exc)
+        return False
+
+
+def _redis_get(token: str) -> Optional[str]:
+    client = _try_redis()
+    if client is None:
+        return None
+    try:
+        raw = client.raw.get(client.key(_KEY, token))
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("land route token Redis 读取失败：%s", exc)
+        return None
+    if raw is None:
+        return None
+    return raw.decode("utf-8") if isinstance(raw, bytes) else str(raw)
+
+
+def _try_redis():
+    try:
+        from agent.redis_coord import get_redis_runtime
+
+        return get_redis_runtime(get_settings()).client
+    except Exception:  # noqa: BLE001
+        return None