diff --git a/.gitignore b/.gitignore index 5493782..d2af3af 100644 --- a/.gitignore +++ b/.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/ diff --git a/docs/current-state.md b/docs/current-state.md index 2fd57e7..ddc05ea 100644 --- a/docs/current-state.md +++ b/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 完整交互验收。 ## 当前可运行内容 diff --git a/docs/integrations/shunyunbao-contract.md b/docs/integrations/shunyunbao-contract.md new file mode 100644 index 0000000..7cae96a --- /dev/null +++ b/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。 diff --git a/docs/tasks/T-220.md b/docs/tasks/T-220.md index 2c407e5..7bb1615 100644 --- a/docs/tasks/T-220.md +++ b/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。 diff --git a/docs/tasks/T-221.md b/docs/tasks/T-221.md index 6647fbb..1f077e0 100644 --- a/docs/tasks/T-221.md +++ b/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 完成后领取。 diff --git a/shunyunbaoerp_single.py b/shunyunbaoerp_single.py new file mode 100644 index 0000000..d18780a --- /dev/null +++ b/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()