cmroubao/shunyunbaoerp_single.py

"""
顺运宝 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()