chore(t220): secure erp contract baseline

.gitignore

@@ -38,3 +38,16 @@ coverage.out
 
 # Logs
 *.log
+
+# Private ERP captures and query output
+*.har
+captcha.jpg
+captcha.png
+freight_detail.json
+erp-connector/.env*
+
+# Python connector outputs
+__pycache__/
+*.py[cod]
+.pytest_cache/
+.venv/

docs/current-state.md

@@ -5,7 +5,7 @@
 ## 当前快照
 
 - 日期：2026-07-28
-- 阶段：T-220 顺运宝 ERP 字段契约与凭证安全基线已领取
+- 阶段：T-220 已完成，T-221 顺运宝精确单号 Connector 已领取
 - Git：当前分支为 `main`；T-001 至 T-004、T-101 至 T-104、T-201 至 T-219
   均按文档提交、实现提交的顺序纳入历史
 - 生产代码：`android-buyer/` 已接入 Roubao Android 源码
@@ -151,8 +151,8 @@
 | `docs/tasks/T-217.md` | DONE | 重新核对已授权商品并选择 SKU/数量，停在最终提交前 |
 | `docs/tasks/T-218.md` | DONE | 单次提交围栏、最多一次提交动作和待付款订单唯一对账 |
 | `docs/tasks/T-219.md` | DONE | Admin 待付款提醒、Roubao 终态和跨端闭环验收 |
-| `docs/tasks/T-220.md` | DOING | 顺运宝 ERP 字段契约与凭证安全基线 |
-| `docs/tasks/T-221.md` | TODO | 顺运宝精确单号 Connector |
+| `docs/tasks/T-220.md` | DONE | 顺运宝 ERP 字段契约与凭证安全基线 |
+| `docs/tasks/T-221.md` | DOING | 顺运宝精确单号 Connector |
 | `docs/tasks/T-222.md` | TODO | 货运信息存储、API 与 Admin 页面 |
 | `docs/tasks/T-223.md` | TODO | 待采购需求提取与任务生成 |
 | `docs/tasks/T-224.md` | TODO | ERP 日期增量同步 |
@@ -167,9 +167,10 @@
 ## 任务摘要
 
 - 已完成：T-001 至 T-004、T-101 至 T-104、T-201 至 T-219。
-- 进行中：T-220 顺运宝 ERP 字段契约与凭证安全基线。
-- 下一步：依次完成 T-221 精确单号 Connector、T-222 货运模块、T-223 采购需求生成
-  和 T-224 日期增量同步，再进入 T-301 P0 UI 完整交互验收。
+- 已完成：另含 T-220 顺运宝 ERP 字段契约与凭证安全基线。
+- 进行中：T-221 顺运宝精确单号 Connector。
+- 下一步：完成 T-222 货运模块、T-223 采购需求生成和 T-224 日期增量同步，再进入
+  T-301 P0 UI 完整交互验收。
 
 ## 当前可运行内容
 

docs/integrations/shunyunbao-contract.md

@@ -0,0 +1,72 @@
+# 顺运宝 ERP 字段契约
+
+> 本文只记录协议结构和字段语义，不保存真实账号、Cookie、JWT、订单号、收件人、
+> 电话、地址、商品内容或原始响应。私有 HAR/响应只允许位于被 Git 忽略的本地目录。
+
+## 已确认请求链
+
+1. 同一 HTTP Session 获取 `GET /api/p/code1` 验证码。
+2. `POST /am/auth/login` 提交环境变量提供的账号/密码和受控验证码。
+3. `POST /am/stock/listTotal` 与 `/am/stock/list` 使用“全部单号”条件查询。
+4. `POST /am/stock/detail/listByStock?hist=0` 使用列表 `stock.id` 批量读取详情。
+
+本地 HAR 结构确认：
+
+- “全部单号”查询输入匹配列表 `code`，不保证等于 `orderCode`。
+- 列表 `id` 等于详情请求 `ids[]` 使用的 stock id，也等于详情头 `id`。
+- 详情头 `details[]` 是商品级集合，一个货运单可以有多条商品。
+- 日期区间使用 `t_stock.created`、`op=0`、`type=3`、`optType=0`；具体增量规则属于
+  T-224。
+
+接口来自 ERP 页面协议，不是已确认的开放 API。生产使用前必须确认厂商授权、请求
+频率、服务账号和个人信息处理规则。
+
+## 身份和规范字段
+
+### 货运单
+
+| 规范字段 | ERP 来源 | 规则 |
+| --- | --- | --- |
+| `external_stock_id` | `stock.id` / `detail.id` | 必填正整数，字符串化保存；唯一外部身份 |
+| `source_code` | `stock.code` | 受控展示；查询输入不作为主键 |
+| `platform_order_no` | `stock.orderCode` | 可空，不进入 URL/普通日志 |
+| `shop_name` | `detail.shopName`，回退 `stock.shopName` | 可空，限制长度 |
+| `source_created_at` | `detail.created`，回退 `stock.created` | 必须可解析为 ERP 本地时间 |
+| `order_status` | `stock.orderStatus` / `detail.status` | 原值字符串化，不猜测数字语义 |
+| `purchase_status` | `stock.purchaseStatus` | 原值字符串化，不自动决定采购 |
+| `is_canceled` | `stock.isCancel` | 只接受明确布尔/已确认枚举 |
+
+### 商品明细
+
+| 规范字段 | ERP 来源 | 规则 |
+| --- | --- | --- |
+| `external_item_id` | `detail.details[].id` | 必填正整数；在货运单内唯一 |
+| `title` | `productTitle`，可核对回退 `detailProductName` | 去首尾空白；空值进入复核 |
+| `product_spec` | `productSpec` | 原样保留，不由 VLM 改写 |
+| `sku` | 非空 `sku`，否则 `variationSku` | 仍需 Admin 复核；两者都空时阻塞 |
+| `quantity` | `productQty` | 必须为正整数，不能使用货运头总数量代替 |
+| `product_thumb_ref` | `productThumb` | 当前确认可能为数字 ERP 引用，不是 URL |
+| `purchase_status` | `purchaseStatus` | 原值字符串化，允许值另行确认 |
+
+`productThumb` 未解析为可鉴权图片内容前，商品状态必须是 `NEEDS_IMAGE`。系统不得把
+数字引用拼接成 URL、抓取未知主机或用其他商品图片补位。
+
+## 规范化与版本
+
+- canonical JSON 固定 UTF-8、字段顺序和空值规则；对最小化对象计算 SHA-256。
+- `(creator, SHUNYUNBAO, external_stock_id)` 唯一确定货运单。
+- `(freight_order_id, external_item_id)` 唯一确定商品。
+- hash 不变是幂等重放；hash 变化递增 revision，不覆盖已生成采购任务。
+- 同一响应含重复 external item id 且内容不一致时整批失败，不选择其中一条。
+- 缺失详情头、`details` 不是数组或任一商品没有外部 ID时整批失败，不保存半批。
+
+## 数据最小化
+
+以下字段不得进入本项目数据库、业务响应、普通日志、任务、VLM 或 Git：
+
+- `receiver`、`receiverTel`、`receiverAddr` 和其他收件个人信息。
+- ERP username/password、验证码、Cookie、JWT、权限和完整用户资料。
+- 未裁剪的 `stock`、`detail`、HAR、HTTP body 和 Redis session JSON。
+
+允许保存的审计信息是：source system、外部 ID、规范字段 hash、同步状态、稳定错误码、
+数量和时间。错误信息不得拼接 ERP 原始 body。

docs/tasks/T-220.md

@@ -4,7 +4,7 @@ title: 顺运宝 ERP 字段契约与凭证安全基线
 phase: 2
 deps:
   - T-219
-status: DOING
+status: DONE
 created: 2026-07-28
 context_ref: 78dc595
 work_branch: null
@@ -50,10 +50,10 @@ ERP 详情的 `details[]` 才是商品级来源；一个货运头可能有多个
 
 ## 验收要点
 
-- [ ] Git 跟踪内容不含 ERP 账号、密码、Cookie、JWT、真实单号和货运 PII。
-- [ ] 单文件脚本无硬编码凭证/订单号，不再向 stdout 输出完整响应。
-- [ ] 字段契约覆盖一货运头多商品、重复同步、缺详情、缺图片和来源更新。
-- [ ] 可用静态命令证明脚本可编译，安全扫描和仓库根验证通过；不访问线上 ERP。
+- [x] Git 跟踪内容不含 ERP 账号、密码、Cookie、JWT、真实单号和货运 PII。
+- [x] 单文件脚本无硬编码凭证/订单号，不再向 stdout 输出完整响应。
+- [x] 字段契约覆盖一货运头多商品、重复同步、缺详情、缺图片和来源更新。
+- [x] 可用静态命令证明脚本可编译，安全扫描和仓库根验证通过；不访问线上 ERP。
 
 ## 边界
 
@@ -66,3 +66,11 @@ ERP 详情的 `details[]` 才是商品级来源；一个货运头可能有多个
 
 - 2026-07-28：在 T-219 `78dc595` 后领取。静态读取本机单文件和独立 ERP 参考工程；
   只提取协议结构和字段名，没有执行登录、线上查询或写操作。
+- 2026-07-28：任务合约提交 `cc847c9`。新增
+  `docs/integrations/shunyunbao-contract.md`，冻结 stock/detail/item 身份、最小字段、
+  canonical hash、PII 排除和缺图边界；私有 HAR/响应没有复制进仓库。
+- 2026-07-28：`shunyunbaoerp_single.py` 已移除硬编码账号、密码和订单号，改为环境
+  变量/交互输入；验证码与结果默认写 `.local/shunyunbao/`，stdout 不再输出 OCR
+  文本或完整货运响应。`.gitignore` 增加 HAR、验证码、查询结果和 Python 输出。
+- 2026-07-28：`python -m py_compile .\shunyunbaoerp_single.py`、离线 payload smoke、
+  Python AST 凭证/订单字面量扫描和根 `.\init.ps1` 通过。未访问线上 ERP。

docs/tasks/T-221.md

@@ -4,7 +4,7 @@ title: 顺运宝精确单号 Connector
 phase: 2
 deps:
   - T-220
-status: TODO
+status: DOING
 created: 2026-07-28
 context_ref: 78dc595
 work_branch: null
@@ -57,4 +57,4 @@ Go 后端不应在 Gin 请求里启动 Python，也不应保存 ERP 密码、Coo
 
 ## 执行记录
 
-- 2026-07-28：任务合约已冻结，等待 T-220。
+- 2026-07-28：任务合约已冻结；T-220 完成后领取。

shunyunbaoerp_single.py

@@ -0,0 +1,971 @@
+"""
+顺运宝 ERP 单文件示例
+
+功能：
+1. 获取 ERP 验证码图片。
+2. 调用 http://127.0.0.1:8000/ocr 识别验证码。
+3. 登录 ERP，并在同一个 requests.Session 中保留 Cookie。
+4. 获取当前用户个人信息，检测 Cookie 会话是否仍处于登录状态。
+5. 把 Cookie、登录 JWT 和最小用户标识缓存到 Redis。
+6. 下次启动时恢复 Redis 会话并通过个人信息接口验证。
+7. 使用货运管理页面的“全部单号”条件搜索订单。
+8. 使用搜索结果的 id 获取货运详情和商品明细。
+
+安装依赖：
+    python -m pip install requests redis
+
+运行：
+    python shunyunbaoerp_single.py
+
+也可以提前设置账号密码，避免每次输入：
+    $env:SHUNYUNBAO_USERNAME = "your_username"
+    $env:SHUNYUNBAO_PASSWORD = "your_password"
+    $env:SHUNYUNBAO_ORDER_NUMBER = "your_order_number"
+
+验证码和查询结果默认写入被 Git 忽略的 .local/shunyunbao/。
+"""
+
+from __future__ import annotations
+
+import base64
+import getpass
+import hashlib
+import json
+import mimetypes
+import os
+import time
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlparse
+
+import requests
+
+try:
+    import redis
+except ImportError:
+    redis = None
+
+
+class ShunYunBaoERP:
+    """一个实例代表一个顺运宝 ERP 登录会话。"""
+
+    # HAR 中货运列表请求携带的 72 个列定义。
+    # 每一项依次是：表名、数据库列名、响应字段名、是否别名、表别名。
+    COLUMN_SPECS = [
+        ("t_stock", "created", "created", 0, "t"),
+        ("t_stock", "order_code", "orderCode", 0, "t"),
+        ("t_stock", "printer", "printer", 0, "t"),
+        ("t_stock", "weight_time", "weightTime", 0, "t"),
+        ("t_stock", "weight_inputer", "weightInputer", 0, "t"),
+        ("t_stock", "pkg_time", "pkgTime", 0, "t"),
+        ("t_stock", "code", "code", 0, "t"),
+        ("t_stock", "status", "status", 0, "t"),
+        ("t_stock", "order_status", "orderStatus", 0, "t"),
+        ("t_stock", "purchase_status", "purchaseStatus", 0, "t"),
+        ("t_stock", "exp_code", "expCode", 0, "t"),
+        ("t_stock", "exp_page_code", "expPageCode", 0, "t"),
+        ("t_stock", "exp_allow_print", "expAllowPrint", 0, "t"),
+        ("t_stock", "exp_page_status", "expPageStatus", 0, "t"),
+        ("t_stock", "page_id", "pageId", 0, "t"),
+        ("t_stock", "upload_time", "uploadTime", 0, "t"),
+        ("t_stock", "pay_time", "payTime", 0, "t"),
+        ("t_stock", "shop_day_to_ship", "shopDayToShip", 0, "t"),
+        ("t_stock", "remark9", "tsremark9", 1, "t"),
+        ("t_stock", "remark9", "remark9", 0, "t"),
+        ("t_stock", "detail_qty", "detailQty", 0, "t"),
+        ("t_stock", "order_qty", "orderQty", 0, "t"),
+        ("t_stock_detail", "inner_exp_code", "innerExpCode", 0, "t7"),
+        ("t_stock_detail", "shelf_code", "shelfCode", 0, "t7"),
+        ("t_stock", "shelf_code", "tsshelfCode", 1, "t"),
+        ("t_stock", "store_type", "storeType", 0, "t"),
+        ("t_stock", "order_bag_code", "orderBagCode", 0, "t"),
+        ("t_stock", "weight_cust_pkg", "weightCustPkg", 0, "t"),
+        ("t_stock", "weight_consign", "weightConsign", 0, "t"),
+        ("t_stock", "amt_order", "amtOrder", 0, "t"),
+        ("t_stock_append", "offline_amount", "offlineAmount", 0, "t8"),
+        ("t_stock_append", "escrow_amount", "escrowAmount", 0, "t8"),
+        ("t_stock", "exp_cod", "expCod", 0, "t"),
+        ("t_stock", "exp_company", "expCompany", 0, "t"),
+        ("t_stock", "transport", "transport", 0, "t"),
+        ("t_stock", "order_origin", "orderOrigin", 0, "t"),
+        ("t_stock", "exp_out_type", "expOutType", 0, "t"),
+        ("t_stock", "order_platform", "orderPlatform", 0, "t"),
+        ("t_stock", "exp_pkg_type", "expPkgType", 0, "t"),
+        ("t_stock", "exp_pkg_code", "expPkgCode", 0, "t"),
+        ("t_stock", "exp_batch", "expBatch", 0, "t"),
+        ("t_stock", "exp_ti_huo", "expTiHuo", 0, "t"),
+        ("t_stock", "print_time", "printTime", 0, "t"),
+        ("t_stock", "receiver", "receiver", 0, "t"),
+        ("t_stock", "receiver_tel", "receiverTel", 0, "t"),
+        ("t_stock", "receiver_addr", "receiverAddr", 0, "t"),
+        ("t_stock", "receiver_shop_name", "receiverShopName", 0, "t"),
+        ("t_stock", "receiver_shop_code", "receiverShopCode", 0, "t"),
+        ("t_stock", "product_name", "productName", 0, "t"),
+        ("t_stock", "is_cancel", "isCancel", 0, "t"),
+        ("t_stock", "err_msg", "errMsg", 0, "t"),
+        ("t_stock", "remark2", "remark2", 0, "t"),
+        ("t_stock", "remark1", "remark1", 0, "t"),
+        ("t_stock", "note", "note", 0, "t"),
+        ("t_store", "name", "name", 0, "t1"),
+        ("sys_user", "fullname", "sufullname", 1, "t3"),
+        ("sys_user", "dept_label_path", "deptLabelPath", 0, "t3"),
+        ("t_stock", "shop_name", "shopName", 0, "t"),
+        ("t_shop", "shop_id", "shopId", 0, "t6"),
+        ("t_stock", "remark4", "remark4", 0, "t"),
+        ("t_stock", "remark7", "remark7", 0, "t"),
+        ("t_stock", "package_time", "packageTime", 0, "t"),
+        ("t_stock", "packer", "packer", 0, "t"),
+        ("t_stock", "pack_type", "packType", 0, "t"),
+        ("t_stock", "track_status", "trackStatus", 0, "t"),
+        ("t_stock", "track_desc", "trackDesc", 0, "t"),
+        ("t_stock", "err_status", "errStatus", 0, "t"),
+        ("t_stock", "err_time", "errTime", 0, "t"),
+        ("t_stock", "err_msg", "tserrMsg", 1, "t"),
+        ("t_stock", "remark2", "tsremark2", 1, "t"),
+        ("t_stock", "remark1", "tsremark1", 1, "t"),
+        ("t_stock", "product_volume_str", "productVolumeStr", 0, "t"),
+    ]
+
+    def __init__(
+        self,
+        base_url: str = "https://www.shunyunbaoerp.com",
+        ocr_url: str = "http://127.0.0.1:8000/ocr",
+        redis_url: str | None = None,
+    ) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.ocr_url = ocr_url
+        self.timeout = (5, 30)  # 连接超时 5 秒，读取超时 30 秒。
+        self.page_size = 20
+        self.max_matches = 100
+        self.redis_url = redis_url or os.getenv(
+            "SHUNYUNBAO_REDIS_URL",
+            "redis://127.0.0.1:6379/0",
+        )
+        self.redis_key_prefix = os.getenv(
+            "SHUNYUNBAO_REDIS_KEY_PREFIX",
+            "shunyunbaoerp:session",
+        )
+        self.redis_session_ttl = int(
+            os.getenv("SHUNYUNBAO_REDIS_SESSION_TTL", "86400")
+        )
+        if self.redis_session_ttl <= 0:
+            raise ValueError("SHUNYUNBAO_REDIS_SESSION_TTL 必须大于 0")
+        self.redis_enabled = os.getenv(
+            "SHUNYUNBAO_REDIS_ENABLED",
+            "1",
+        ).lower() not in {"0", "false", "no", "off"}
+        self.redis_client: Any | None = None
+
+        # ERP 的验证码、登录和查询必须使用同一个 Session。
+        self.erp_session = requests.Session()
+        self.erp_session.headers.update(
+            {
+                "Accept": "application/json, text/plain, */*",
+                "Origin": self.base_url,
+                "Referer": f"{self.base_url}/sys/login",
+                "User-Agent": "Mozilla/5.0 shunyunbaoerp-single-script",
+                "X-Requested-With": "XMLHttpRequest",
+            }
+        )
+
+        # OCR 在本机运行，单独使用一个 Session，避免携带 ERP 的 Cookie。
+        # trust_env=False 可以防止系统代理把 127.0.0.1 请求转发到代理服务器。
+        self.ocr_session = requests.Session()
+        self.ocr_session.trust_env = False
+
+        self.user: dict[str, Any] | None = None
+        self.token: str | None = None
+
+    def get_captcha(self, save_path: str = "captcha.jpg") -> str:
+        """获取验证码并保存到文件，返回文件的绝对路径。"""
+
+        url = f"{self.base_url}/api/p/code1"
+        response = self.erp_session.get(
+            url,
+            params={"_": int(time.time() * 1000)},
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        self._capture_refreshed_token(response)
+
+        content_type = response.headers.get("Content-Type", "")
+        if not content_type.startswith("image/"):
+            raise RuntimeError(f"验证码接口没有返回图片，Content-Type={content_type!r}")
+
+        path = Path(save_path).resolve()
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(response.content)
+        return str(path)
+
+    def ocr_captcha(self, image_path: str) -> str:
+        """调用本地 OCR 接口识别验证码。
+
+        等价 curl：
+        curl --request POST --url http://127.0.0.1:8000/ocr \
+             --header "Content-Type: multipart/form-data" \
+             --form "file=@captcha.jpg"
+
+        requests 会自动生成带 boundary 的 Content-Type，所以不要手动设置该请求头。
+        """
+
+        path = Path(image_path)
+        if not path.is_file():
+            raise FileNotFoundError(f"验证码图片不存在: {path}")
+
+        mime_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
+        with path.open("rb") as image_file:
+            response = self.ocr_session.post(
+                self.ocr_url,
+                files={"file": (path.name, image_file, mime_type)},
+                timeout=self.timeout,
+            )
+        response.raise_for_status()
+
+        # 已验证的本地接口响应为：
+        # {"code": 200, "message": "Success", "data": "验证码文字"}
+        try:
+            body = response.json()
+        except ValueError:
+            # 同时兼容直接返回纯文本的 OCR 服务。
+            text = response.text
+        else:
+            if not isinstance(body, dict):
+                raise RuntimeError("OCR 响应 JSON 不是对象")
+            if body.get("code") not in (None, 0, 200, "0", "200"):
+                raise RuntimeError(
+                    f"OCR 识别失败: {body.get('message') or body.get('msg') or body}"
+                )
+
+            data = body.get("data")
+            if isinstance(data, str):
+                text = data
+            elif isinstance(data, dict):
+                text = str(
+                    data.get("text")
+                    or data.get("result")
+                    or data.get("code")
+                    or ""
+                )
+            else:
+                text = str(
+                    body.get("text")
+                    or body.get("result")
+                    or body.get("message")
+                    or ""
+                )
+
+        # 当前验证码由英文字母/数字组成，过滤 OCR 可能带回的空格和标点。
+        captcha_code = "".join(char for char in text.strip() if char.isalnum())
+        if not captcha_code:
+            raise RuntimeError(f"OCR 没有识别出验证码，原始响应: {response.text[:200]}")
+        return captcha_code
+
+    def login(self, username: str, password: str, captcha_code: str) -> dict[str, Any]:
+        """提交用户名、密码和验证码登录 ERP。"""
+
+        username = username.strip()
+        captcha_code = captcha_code.strip()
+        if not username:
+            raise ValueError("用户名不能为空")
+        if not password:
+            raise ValueError("密码不能为空")
+        if not captcha_code:
+            raise ValueError("验证码不能为空")
+
+        data = self._post_json(
+            "/am/auth/login",
+            {
+                "username": username,
+                "password": password,
+                "code": captcha_code,
+            },
+        )
+        if not isinstance(data, dict) or not isinstance(data.get("user"), dict):
+            raise RuntimeError("登录响应中缺少 user 数据")
+
+        self.user = data["user"]
+        self.token = data.get("token")
+        return self.user
+
+    def get_user_info(
+        self,
+        user_id: int | str | None = None,
+    ) -> dict[str, Any]:
+        """获取个人信息。
+
+        shunyunbaoerp_userinfo.har 中的接口是：
+            GET /am/user/get?id=<登录响应中的 user.id>
+
+        user_id 不传时，自动使用 login() 保存的当前用户 id。
+        """
+
+        if user_id is None and self.user is not None:
+            user_id = self.user.get("id")
+        if user_id in (None, ""):
+            raise ValueError("没有用户 id，请先登录或手动传入 user_id")
+
+        try:
+            normalized_user_id = int(user_id)
+        except (TypeError, ValueError) as exc:
+            raise ValueError("user_id 必须是正整数") from exc
+        if normalized_user_id <= 0:
+            raise ValueError("user_id 必须是正整数")
+
+        data = self._get_json(
+            "/am/user/get",
+            params={"id": normalized_user_id},
+        )
+        if not isinstance(data, dict):
+            raise RuntimeError("个人信息接口返回的 data 不是对象")
+        if data.get("id") is None or data.get("username") in (None, ""):
+            raise RuntimeError("个人信息中缺少 id 或 username")
+        return data
+
+    def is_logged_in(
+        self,
+        user_id: int | str | None = None,
+    ) -> bool:
+        """通过获取个人信息检测当前 Session 是否已经登录。
+
+        返回值：
+        - True：成功取得同一用户的个人信息。
+        - False：没有用户 id，或服务端明确返回未登录/登录过期。
+
+        网络超时、服务器 5xx、响应格式错误等情况不会返回 False，
+        而是抛出异常，避免把网络故障误认为退出登录。
+        """
+
+        if user_id is None and self.user is not None:
+            user_id = self.user.get("id")
+        if user_id in (None, ""):
+            return False
+
+        old_username = self.user.get("username") if self.user else None
+        try:
+            user_info = self.get_user_info(user_id)
+        except requests.HTTPError as exc:
+            status_code = exc.response.status_code if exc.response is not None else None
+            if status_code in (401, 403):
+                self.user = None
+                self.token = None
+                return False
+            raise
+        except RuntimeError as exc:
+            message = str(exc)
+            logged_out_markers = ("code=-2", "未登录", "登录过期")
+            if any(marker in message for marker in logged_out_markers):
+                self.user = None
+                self.token = None
+                return False
+            raise
+
+        same_id = str(user_info.get("id")) == str(user_id)
+        same_username = (
+            old_username is None
+            or str(user_info.get("username")) == str(old_username)
+        )
+        if not same_id or not same_username:
+            self.user = None
+            self.token = None
+            return False
+
+        # userinfo 中的 token 是另一种用户 API token，不是登录响应的 JWT。
+        # 检测登录状态时不允许它覆盖 self.token。
+        self.user = dict(user_info)
+        self.user.pop("token", None)
+        return True
+
+    def save_session_to_redis(self, username: str) -> bool:
+        """把 Cookie、登录 JWT 和最小用户标识保存到 Redis。
+
+        使用 JSON 而不是 pickle，避免反序列化任意代码风险。
+        Redis 只是缓存：保存失败不会让已经登录的 ERP 会话失效。
+        """
+
+        if not self.redis_enabled:
+            return False
+        if self.user is None or self.user.get("id") is None:
+            raise RuntimeError("没有可保存的登录用户，请先登录")
+
+        cookies = []
+        for cookie in self.erp_session.cookies:
+            if not self._cookie_belongs_to_erp(cookie.domain):
+                continue
+            cookies.append(
+                {
+                    "name": cookie.name,
+                    "value": cookie.value,
+                    "domain": cookie.domain
+                    or (urlparse(self.base_url).hostname or ""),
+                    "path": cookie.path or "/",
+                    "secure": bool(cookie.secure),
+                    "expires": cookie.expires,
+                    "rest": {
+                        str(key): str(value)
+                        for key, value in getattr(cookie, "_rest", {}).items()
+                    },
+                }
+            )
+
+        if not cookies:
+            print("警告：ERP Session 中没有可保存的 Cookie")
+            return False
+
+        ttl = self._session_ttl()
+        if ttl <= 0:
+            self.delete_session_from_redis(username)
+            return False
+
+        # 只保存恢复登录所需的最小用户信息，不缓存完整个人资料。
+        session_data = {
+            "version": 1,
+            "savedAt": int(time.time()),
+            "baseUrl": self.base_url,
+            "user": {
+                "id": self.user.get("id"),
+                "username": self.user.get("username") or username,
+            },
+            "token": self.token,
+            "cookies": cookies,
+        }
+
+        try:
+            client = self._get_redis_client()
+            client.setex(
+                self._redis_key(username),
+                ttl,
+                json.dumps(session_data, ensure_ascii=False, separators=(",", ":")),
+            )
+            return True
+        except Exception as exc:
+            if self._is_redis_error(exc):
+                print(f"警告：Redis 会话保存失败，将继续使用当前内存会话: {exc}")
+                return False
+            raise
+
+    def load_session_from_redis(self, username: str) -> bool:
+        """从 Redis 读取 Cookie、JWT 和最小用户标识到当前实例。
+
+        本函数只负责恢复，不代表会话一定有效；调用后还必须执行
+        is_logged_in() 或 restore_login_from_redis()。
+        """
+
+        if not self.redis_enabled:
+            return False
+        try:
+            client = self._get_redis_client()
+            raw_data = client.get(self._redis_key(username))
+        except Exception as exc:
+            if self._is_redis_error(exc):
+                print(f"警告：Redis 不可用，将改用 OCR 登录: {exc}")
+                return False
+            raise
+        if not raw_data:
+            return False
+
+        try:
+            session_data = json.loads(raw_data)
+        except (TypeError, json.JSONDecodeError):
+            self.delete_session_from_redis(username)
+            return False
+
+        if (
+            not isinstance(session_data, dict)
+            or session_data.get("version") != 1
+            or session_data.get("baseUrl") != self.base_url
+            or not isinstance(session_data.get("user"), dict)
+            or not isinstance(session_data.get("cookies"), list)
+        ):
+            self.delete_session_from_redis(username)
+            return False
+
+        cached_user = session_data["user"]
+        if str(cached_user.get("username")) != username:
+            self.delete_session_from_redis(username)
+            return False
+
+        self.erp_session.cookies.clear()
+        restored_cookie_count = 0
+        now = int(time.time())
+        for item in session_data["cookies"]:
+            if not isinstance(item, dict):
+                continue
+            name = item.get("name")
+            value = item.get("value")
+            domain = item.get("domain")
+            expires = item.get("expires")
+            if not name or value is None or not self._cookie_belongs_to_erp(domain):
+                continue
+            if expires is not None and int(expires) <= now:
+                continue
+
+            cookie_arguments: dict[str, Any] = {
+                "name": str(name),
+                "value": str(value),
+                "path": str(item.get("path") or "/"),
+                "secure": bool(item.get("secure")),
+                "rest": item.get("rest") if isinstance(item.get("rest"), dict) else {},
+            }
+            if domain:
+                cookie_arguments["domain"] = str(domain)
+            if expires is not None:
+                cookie_arguments["expires"] = int(expires)
+            cookie = requests.cookies.create_cookie(**cookie_arguments)
+            self.erp_session.cookies.set_cookie(cookie)
+            restored_cookie_count += 1
+
+        if restored_cookie_count == 0:
+            self.delete_session_from_redis(username)
+            return False
+
+        self.user = {
+            "id": cached_user.get("id"),
+            "username": cached_user.get("username"),
+        }
+        cached_token = session_data.get("token")
+        self.token = cached_token if isinstance(cached_token, str) else None
+        return True
+
+    def restore_login_from_redis(self, username: str) -> bool:
+        """恢复 Redis 会话，并调用个人信息接口验证是否仍然登录。"""
+
+        if not self.load_session_from_redis(username):
+            return False
+
+        # is_logged_in 会区分“明确未登录”和“网络故障”：
+        # 明确未登录返回 False；网络故障继续抛错，不盲目触发 OCR 登录。
+        if self.is_logged_in():
+            # 个人信息请求可能刷新 Cookie，验证成功后续期 Redis 缓存。
+            self.save_session_to_redis(username)
+            return True
+
+        self.delete_session_from_redis(username)
+        self.erp_session.cookies.clear()
+        self.user = None
+        self.token = None
+        return False
+
+    def delete_session_from_redis(self, username: str) -> bool:
+        """删除指定 ERP 用户的 Redis 会话缓存。"""
+
+        if not self.redis_enabled:
+            return False
+        try:
+            client = self._get_redis_client()
+            return bool(client.delete(self._redis_key(username)))
+        except Exception as exc:
+            if self._is_redis_error(exc):
+                print(f"警告：删除 Redis 会话失败: {exc}")
+                return False
+            raise
+
+    def login_with_redis_or_ocr(
+        self,
+        username: str,
+        password: str | None = None,
+        max_attempts: int = 5,
+        captcha_path: str = "captcha.jpg",
+    ) -> dict[str, Any]:
+        """启动时优先恢复 Redis 会话，无效时才执行 OCR 登录。"""
+
+        username = username.strip()
+        if not username:
+            raise ValueError("用户名不能为空")
+
+        if self.restore_login_from_redis(username):
+            print(f"Redis 会话有效，已恢复登录用户: {username}")
+            if self.user is None:
+                raise RuntimeError("Redis 会话验证成功但用户信息为空")
+            return self.user
+
+        print("Redis 中没有有效会话，开始获取验证码并登录 ERP")
+        if not password:
+            password = os.getenv("SHUNYUNBAO_PASSWORD") or getpass.getpass(
+                "ERP 密码: "
+            )
+        if not password:
+            raise ValueError("密码不能为空")
+        user = self.login_with_ocr(
+            username,
+            password,
+            max_attempts=max_attempts,
+            captcha_path=captcha_path,
+        )
+        self.save_session_to_redis(username)
+        return user
+
+    def login_with_ocr(
+        self,
+        username: str,
+        password: str,
+        max_attempts: int = 5,
+        captcha_path: str = "captcha.jpg",
+    ) -> dict[str, Any]:
+        """自动执行“获取验证码 → OCR → 登录”，失败时刷新验证码重试。"""
+
+        last_error: Exception | None = None
+        for attempt in range(1, max_attempts + 1):
+            try:
+                image_path = self.get_captcha(captcha_path)
+                captcha_code = self.ocr_captcha(image_path)
+                # HAR 中验证码长度为 4。长度明显不对时不请求登录，直接刷新图片。
+                if len(captcha_code) != 4:
+                    raise RuntimeError(
+                        f"OCR 结果长度为 {len(captcha_code)}，预期为 4"
+                    )
+
+                user = self.login(username, password, captcha_code)
+                if not self.is_logged_in(user.get("id")):
+                    raise RuntimeError("登录接口成功，但个人信息接口检测为未登录")
+                print(f"ERP 登录成功，用户: {user.get('username', '')}")
+                return user
+            except (requests.RequestException, RuntimeError) as exc:
+                last_error = exc
+                print(f"第 {attempt} 次登录未成功: {exc}")
+
+        raise RuntimeError(
+            f"连续 {max_attempts} 次未能登录，请检查 OCR、账号和密码。"
+            f"最后错误: {last_error}"
+        )
+
+    def build_order_search_payload(
+        self,
+        order_number: str,
+        start: int = 0,
+        page_index: int = 1,
+    ) -> dict[str, Any]:
+        """构造货运管理页面的单号搜索请求体。"""
+
+        order_number = order_number.strip()
+        if not order_number:
+            raise ValueError("订单号不能为空")
+
+        columns = [
+            {
+                "tableName": table_name,
+                "colName": column_name,
+                "fieldName": field_name,
+                "hasAlias": has_alias,
+                "tableAlias": table_alias,
+            }
+            for (
+                table_name,
+                column_name,
+                field_name,
+                has_alias,
+                table_alias,
+            ) in self.COLUMN_SPECS
+        ]
+
+        return {
+            "history": 0,
+            "length": self.page_size,
+            "start": start,
+            "pageTotal": 0,
+            "pageIndex": page_index,
+            "store": False,
+            "columns": columns,
+            "queries": [
+                {
+                    "dvalue": order_number,
+                    "tableName": "t_stock",
+                    "colName": "allcode",
+                    "op": 6,
+                    "type": 0,
+                    "tableAlias": "t",
+                    "optType": 1,
+                }
+            ],
+        }
+
+    def search_order(self, order_number: str) -> list[dict[str, Any]]:
+        """在货运管理中按单号搜索，返回货运列表记录。"""
+
+        first_payload = self.build_order_search_payload(order_number)
+        total_value = self._post_json("/am/stock/listTotal", first_payload)
+        try:
+            total = int(total_value or 0)
+        except (TypeError, ValueError) as exc:
+            raise RuntimeError(f"listTotal 返回的总数无效: {total_value!r}") from exc
+
+        if total == 0:
+            return []
+        if total > self.max_matches:
+            raise RuntimeError(
+                f"单号搜索返回 {total} 条，超过安全上限 {self.max_matches}，"
+                "请确认输入的是完整单号"
+            )
+
+        rows: list[dict[str, Any]] = []
+        for start in range(0, total, self.page_size):
+            page_index = start // self.page_size + 1
+            payload = self.build_order_search_payload(
+                order_number,
+                start=start,
+                page_index=page_index,
+            )
+            page_data = self._post_json("/am/stock/list", payload)
+            if not isinstance(page_data, dict) or not isinstance(
+                page_data.get("list"), list
+            ):
+                raise RuntimeError("货运列表响应缺少 data.list")
+            rows.extend(page_data["list"])
+
+        return rows[:total]
+
+    def get_freight_details_by_ids(
+        self,
+        stock_ids: list[int],
+    ) -> list[dict[str, Any]]:
+        """使用货运列表 id 批量获取货运详情。"""
+
+        if not stock_ids:
+            return []
+
+        details: list[dict[str, Any]] = []
+        # 一次最多请求 100 个 id，避免请求体过大。
+        for start in range(0, len(stock_ids), 100):
+            ids = stock_ids[start : start + 100]
+            data = self._post_json(
+                "/am/stock/detail/listByStock",
+                {"ids": ids},
+                params={"hist": 0},
+            )
+            if not isinstance(data, dict) or not isinstance(data.get("list"), list):
+                raise RuntimeError("货运详情响应缺少 data.list")
+            details.extend(data["list"])
+        return details
+
+    def search_freight_details(self, order_number: str) -> dict[str, Any]:
+        """完成“搜索单号 → 获取详情”，返回合并后的完整结果。"""
+
+        if not self.is_logged_in():
+            raise RuntimeError("当前 ERP 会话未登录或已经过期，请重新登录")
+
+        stock_rows = self.search_order(order_number)
+        if not stock_rows:
+            return {
+                "query": order_number,
+                "count": 0,
+                "records": [],
+            }
+
+        stock_ids: list[int] = []
+        for row in stock_rows:
+            if row.get("id") is None:
+                raise RuntimeError("货运列表中存在缺少 id 的记录")
+            stock_ids.append(int(row["id"]))
+
+        detail_rows = self.get_freight_details_by_ids(stock_ids)
+        details_by_id = {
+            str(detail["id"]): detail
+            for detail in detail_rows
+            if detail.get("id") is not None
+        }
+
+        records = [
+            {
+                "stock": stock,
+                "detail": details_by_id.get(str(stock["id"])),
+            }
+            for stock in stock_rows
+        ]
+        return {
+            "query": order_number,
+            "count": len(records),
+            "records": records,
+        }
+
+    def save_result(
+        self,
+        result: dict[str, Any],
+        save_path: str = ".local/shunyunbao/freight_detail.json",
+    ) -> str:
+        """把私有查询结果保存到本地忽略目录。"""
+
+        path = Path(save_path).resolve()
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(
+            json.dumps(result, ensure_ascii=False, indent=2, default=str),
+            encoding="utf-8",
+        )
+        return str(path)
+
+    def close(self) -> None:
+        """关闭 ERP、OCR 和 Redis 的连接池。"""
+
+        self.erp_session.close()
+        self.ocr_session.close()
+        if self.redis_client is not None and hasattr(self.redis_client, "close"):
+            self.redis_client.close()
+
+    def run(self) -> None:
+        """使用环境变量或交互输入执行一次受控查询。"""
+
+        username = (
+            os.getenv("SHUNYUNBAO_USERNAME")
+            or input("ERP 用户名: ").strip()
+        )
+        order_number = (
+            os.getenv("SHUNYUNBAO_ORDER_NUMBER")
+            or input("完整订单号: ").strip()
+        )
+        if not username:
+            raise ValueError("ERP 用户名不能为空")
+        if not order_number:
+            raise ValueError("完整订单号不能为空")
+
+        try:
+            # 只有 Redis 会话不存在或失效时，才会读取/询问 ERP 密码并 OCR 登录。
+            self.login_with_redis_or_ocr(username)
+            result = self.search_freight_details(order_number)
+            # 查询过程中服务端可能刷新 Cookie，结束前再保存一次。
+            self.save_session_to_redis(username)
+            result_path = self.save_result(result)
+
+            print(f"查询完成，共找到 {result['count']} 条记录。")
+            print(f"完整结果已保存到: {result_path}")
+        finally:
+            self.close()
+
+    def _post_json(
+        self,
+        path: str,
+        json_body: dict[str, Any],
+        params: dict[str, Any] | None = None,
+    ) -> Any:
+        """发送 ERP POST 请求并取出统一响应中的 data。"""
+
+        response = self.erp_session.post(
+            f"{self.base_url}{path}",
+            params=params,
+            json=json_body,
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        self._capture_refreshed_token(response)
+
+        try:
+            body = response.json()
+        except ValueError as exc:
+            raise RuntimeError(f"{path} 返回的不是 JSON") from exc
+
+        if not isinstance(body, dict):
+            raise RuntimeError(f"{path} 的 JSON 顶层不是对象")
+        if body.get("status") is not True:
+            message = body.get("msg") or "ERP 请求失败"
+            code = body.get("code")
+            raise RuntimeError(f"{path} 请求失败: {message}，code={code}")
+        if "data" not in body:
+            raise RuntimeError(f"{path} 成功响应中缺少 data")
+        return body["data"]
+
+    def _get_redis_client(self) -> Any:
+        """延迟创建 Redis 客户端，避免 Redis 缓存影响模块导入。"""
+
+        if self.redis_client is not None:
+            return self.redis_client
+        if redis is None:
+            raise RuntimeError(
+                "未安装 redis 包，请执行: python -m pip install redis"
+            )
+
+        parsed = urlparse(self.redis_url)
+        is_local = parsed.hostname in {"127.0.0.1", "localhost", "::1"}
+        if parsed.scheme != "rediss" and not is_local:
+            print("安全警告：远程 Redis 应使用 rediss:// TLS 连接")
+        if not is_local and parsed.password is None:
+            print("安全警告：远程 Redis 应配置 ACL 用户名和密码")
+
+        self.redis_client = redis.Redis.from_url(
+            self.redis_url,
+            decode_responses=True,
+            socket_connect_timeout=2,
+            socket_timeout=2,
+            health_check_interval=30,
+        )
+        return self.redis_client
+
+    def _redis_key(self, username: str) -> str:
+        """使用用户名哈希作为 Redis key，避免 key 中直接出现账号。"""
+
+        username_hash = hashlib.sha256(username.encode("utf-8")).hexdigest()[:24]
+        return f"{self.redis_key_prefix}:{username_hash}"
+
+    def _session_ttl(self) -> int:
+        """计算 Redis TTL，不超过配置值和登录 JWT 剩余时间。"""
+
+        ttl = self.redis_session_ttl
+        if not self.token or self.token.count(".") != 2:
+            return ttl
+
+        try:
+            payload = self.token.split(".")[1]
+            payload += "=" * (-len(payload) % 4)
+            claims = json.loads(
+                base64.urlsafe_b64decode(payload.encode("ascii")).decode("utf-8")
+            )
+            expires_at = int(claims["exp"])
+        except (
+            KeyError,
+            TypeError,
+            ValueError,
+            UnicodeDecodeError,
+            json.JSONDecodeError,
+        ):
+            return ttl
+        return max(0, min(ttl, expires_at - int(time.time())))
+
+    def _cookie_belongs_to_erp(self, cookie_domain: str | None) -> bool:
+        """只允许保存和恢复属于当前 ERP 主机的 Cookie。"""
+
+        erp_host = (urlparse(self.base_url).hostname or "").lower()
+        domain = (cookie_domain or erp_host).lstrip(".").lower()
+        return erp_host == domain or erp_host.endswith(f".{domain}")
+
+    def _capture_refreshed_token(self, response: requests.Response) -> None:
+        """保存服务端可能通过响应头刷新的登录 JWT。"""
+
+        refreshed_token = response.headers.get("X-Requested-With")
+        if isinstance(refreshed_token, str) and refreshed_token.count(".") == 2:
+            self.token = refreshed_token
+
+    @staticmethod
+    def _is_redis_error(error: Exception) -> bool:
+        """判断 redis-py 抛出的连接、超时等 Redis 异常。"""
+
+        return redis is not None and isinstance(error, redis.RedisError)
+
+    def _get_json(
+        self,
+        path: str,
+        params: dict[str, Any] | None = None,
+    ) -> Any:
+        """发送 ERP GET 请求并取出统一响应中的 data。"""
+
+        response = self.erp_session.get(
+            f"{self.base_url}{path}",
+            params=params,
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        self._capture_refreshed_token(response)
+
+        try:
+            body = response.json()
+        except ValueError as exc:
+            raise RuntimeError(f"{path} 返回的不是 JSON") from exc
+
+        if not isinstance(body, dict):
+            raise RuntimeError(f"{path} 的 JSON 顶层不是对象")
+        if body.get("status") is not True:
+            message = body.get("msg") or "ERP 请求失败"
+            code = body.get("code")
+            raise RuntimeError(f"{path} 请求失败: {message}，code={code}")
+        if "data" not in body:
+            raise RuntimeError(f"{path} 成功响应中缺少 data")
+        return body["data"]
+
+
+if __name__ == "__main__":
+    ShunYunBaoERP().run()