605 lines
23 KiB
Python
605 lines
23 KiB
Python
"""揽收系统 HTTP API 封装 — 直接调用后端接口，不依赖浏览器。
|
||
|
||
依赖 Node.js token bridge 计算 $_ts 安全框架的 c93l5Fya 动态 token。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
from typing import Any
|
||
from urllib.parse import urlparse, urljoin
|
||
|
||
from log import debug
|
||
|
||
|
||
# ─── Token Bridge 管理 ────────────────────────────────────────────────
|
||
|
||
def _get_bridge_dir() -> str:
|
||
"""获取 token_bridge 目录路径（支持 PyInstaller 打包）。"""
|
||
if getattr(sys, "frozen", False):
|
||
base = sys._MEIPASS
|
||
else:
|
||
base = os.path.dirname(os.path.abspath(__file__))
|
||
return os.path.join(base, "token_bridge")
|
||
|
||
|
||
class TokenBridge:
|
||
"""管理 Node.js token 计算子进程（HTTP 通信）。"""
|
||
|
||
def __init__(self):
|
||
self._proc: subprocess.Popen | None = None
|
||
self._port: int = 0
|
||
self._base_url: str = ""
|
||
|
||
def start(self):
|
||
bridge_dir = _get_bridge_dir()
|
||
kwargs = dict(
|
||
cwd=bridge_dir,
|
||
stderr=subprocess.PIPE,
|
||
text=True, encoding="utf-8", errors="replace",
|
||
)
|
||
if sys.platform == "win32":
|
||
kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
|
||
self._proc = subprocess.Popen(["node", "server.js", "0"], **kwargs)
|
||
|
||
# 进程退出时强制杀子进程，防止 _MEIPASS 目录无法删除
|
||
import atexit as _atexit
|
||
_atexit.register(self.stop)
|
||
# 读取就绪信号（包含端口号）
|
||
line = self._proc.stderr.readline()
|
||
try:
|
||
msg = json.loads(line.strip())
|
||
if not msg.get("ready"):
|
||
raise RuntimeError(f"Bridge startup failed: {line}")
|
||
self._port = msg["port"]
|
||
self._base_url = f"http://127.0.0.1:{self._port}"
|
||
except json.JSONDecodeError:
|
||
raise RuntimeError(f"Bridge unexpected stderr: {line}")
|
||
|
||
# 后台线程持续读取 stderr 并输出到 log
|
||
import threading as _threading
|
||
def _read_stderr():
|
||
from log import trace
|
||
for line in self._proc.stderr:
|
||
trace(f"bridge stderr: {line.strip()}")
|
||
_threading.Thread(target=_read_stderr, daemon=True).start()
|
||
|
||
def stop(self):
|
||
if self._proc:
|
||
try:
|
||
self._proc.kill()
|
||
self._proc.wait(timeout=3)
|
||
except Exception:
|
||
pass
|
||
self._proc = None
|
||
self._port = 0
|
||
|
||
def _request(self, path: str, body: dict = None) -> dict:
|
||
"""向 bridge HTTP 服务器发送请求。"""
|
||
from log import debug, trace
|
||
import requests as _requests
|
||
url = f"{self._base_url}{path}"
|
||
trace(f"bridge -> {path} {str(body)[:100] if body else ''}")
|
||
try:
|
||
if body:
|
||
resp = _requests.post(url, json=body, timeout=30)
|
||
else:
|
||
resp = _requests.get(url, timeout=30)
|
||
resp.raise_for_status()
|
||
result = resp.json()
|
||
trace(f"bridge <- {str(result)[:100]}")
|
||
if "error" in result:
|
||
raise RuntimeError(f"Bridge error: {result['error']}")
|
||
return result
|
||
except _requests.RequestException as e:
|
||
# 读取 stderr
|
||
err = ""
|
||
try:
|
||
while True:
|
||
chunk = self._proc.stderr.readline()
|
||
if not chunk:
|
||
break
|
||
err += chunk
|
||
except Exception:
|
||
pass
|
||
if err:
|
||
raise RuntimeError(f"Bridge request failed: {e}, stderr={err[:300]}")
|
||
raise
|
||
|
||
def initialize(self, html: str, scripts: list[dict], base_url: str):
|
||
return self._request("/init", {
|
||
"html": html,
|
||
"scripts": scripts,
|
||
"baseUrl": base_url,
|
||
})
|
||
|
||
def compute_token(self, method: str, path: str) -> str:
|
||
result = self._request("/token", {
|
||
"method": method,
|
||
"path": path,
|
||
})
|
||
return result["token"]
|
||
|
||
def ping(self) -> bool:
|
||
try:
|
||
self._request("/ping")
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
# ─── HTTP 会话 & 页面资源管理 ─────────────────────────────────────────
|
||
|
||
class PickupSession:
|
||
"""管理 HTTP 会话（cookies）和页面资源加载。"""
|
||
|
||
def __init__(self, base_url: str):
|
||
self.base_url = base_url.rstrip("/")
|
||
parsed = urlparse(self.base_url)
|
||
self._origin = f"{parsed.scheme}://{parsed.netloc}"
|
||
# 使用 requests 保持 cookie 会话（同步，足够简单可靠）
|
||
import requests as _requests
|
||
self._http: _requests.Session = _requests.Session()
|
||
self._http.verify = False # 内部自签证书
|
||
import urllib3
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
self._http.headers.update({
|
||
"User-Agent": (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||
"Chrome/109.0.0.0 Safari/537.36"
|
||
),
|
||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||
"Cache-Control": "no-cache",
|
||
"Pragma": "no-cache",
|
||
})
|
||
self._scripts: list[dict] = []
|
||
self._html: str = ""
|
||
self._pickup_path: str = ""
|
||
self._ts_cd: str = ""
|
||
self._ts_nsd: int = 0
|
||
|
||
@property
|
||
def cookies_str(self) -> str:
|
||
"""将 requests cookies 转为 Cookie 请求头字符串。"""
|
||
pairs = []
|
||
for cookie in self._http.cookies:
|
||
pairs.append(f"{cookie.name}={cookie.value}")
|
||
return "; ".join(pairs)
|
||
|
||
def load_session(self, session_path: str) -> bool:
|
||
"""从文件加载之前保存的 cookies。返回是否加载成功。"""
|
||
import requests as _requests
|
||
try:
|
||
with open(session_path, "r") as f:
|
||
data = json.load(f)
|
||
seen = set()
|
||
for cookie_data in data.get("cookies", []):
|
||
key = (cookie_data["name"], cookie_data.get("domain", ""), cookie_data.get("path", "/"))
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
self._http.cookies.set(
|
||
cookie_data["name"],
|
||
cookie_data["value"],
|
||
domain=cookie_data.get("domain", ""),
|
||
path=cookie_data.get("path", "/"),
|
||
)
|
||
return len(seen) > 0
|
||
except (FileNotFoundError, json.JSONDecodeError, KeyError):
|
||
return False
|
||
|
||
def save_session(self, session_path: str):
|
||
"""保存 cookies 到文件。"""
|
||
data = {
|
||
"cookies": [
|
||
{"name": c.name, "value": c.value, "domain": c.domain, "path": c.path}
|
||
for c in self._http.cookies
|
||
]
|
||
}
|
||
with open(session_path, "w") as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
|
||
def _abs_url(self, path_or_url: str) -> str:
|
||
if path_or_url.startswith("http"):
|
||
return path_or_url
|
||
return urljoin(self._origin, path_or_url)
|
||
|
||
def fetch_page_resources(self) -> dict:
|
||
"""加载 pickup 页面及所需 JS 文件。返回 {html, scripts, base_url, nsd}。"""
|
||
import requests as _requests
|
||
|
||
pickup_path = urlparse(self.base_url).path
|
||
self._pickup_path = pickup_path
|
||
|
||
# 加载 pickup 页面（需要 Referer 满足安全框架要求）
|
||
portal_url = f"{self._origin}/portal/a"
|
||
debug(f"fetch_page: loading {self.base_url}, cookies={list(self._http.cookies.keys())}")
|
||
resp = self._http.get(
|
||
self.base_url,
|
||
headers={
|
||
"Referer": portal_url,
|
||
"Sec-Fetch-Dest": "iframe",
|
||
"Sec-Fetch-Mode": "navigate",
|
||
"Sec-Fetch-Site": "same-origin",
|
||
"Upgrade-Insecure-Requests": "1",
|
||
},
|
||
timeout=30,
|
||
)
|
||
debug(f"fetch_page: status={resp.status_code}")
|
||
if resp.status_code != 200:
|
||
debug(f"fetch_page: resp_headers={dict(resp.headers)}")
|
||
debug(f"fetch_page: resp_body={resp.text[:500]}")
|
||
resp.raise_for_status()
|
||
self._html = resp.text
|
||
|
||
# 从页面 HTML 中提取 $_ts 安全参数
|
||
self._ts_cd = ""
|
||
self._ts_nsd = 0
|
||
m = re.search(r'\$_ts\.cd="([^"]*)"', self._html)
|
||
if m:
|
||
self._ts_cd = m.group(1)
|
||
m = re.search(r'\$_ts\.nsd=(\d+)', self._html)
|
||
if m:
|
||
self._ts_nsd = int(m.group(1))
|
||
if not self._ts_cd or not self._ts_nsd:
|
||
raise RuntimeError("Failed to extract $_ts.cd/nsd from page")
|
||
|
||
# 2. 提取外部 JS 引用，只保留 security JS（cfd114a.js）
|
||
# 页面业务脚本（jQuery、main.js 等）会导致 jsdom 卡死，不需要
|
||
script_urls = re.findall(
|
||
r'<script[^>]+src="([^"]+)"[^>]*>', self._html, re.IGNORECASE
|
||
)
|
||
self._scripts = []
|
||
for url in script_urls:
|
||
if "cfd114a.js" not in url:
|
||
continue
|
||
abs_url = self._abs_url(url)
|
||
try:
|
||
script_resp = self._http.get(abs_url, timeout=15)
|
||
if script_resp.status_code == 200:
|
||
self._scripts.append({"url": url, "content": script_resp.text})
|
||
debug(f"fetch_page: loaded security JS {url}")
|
||
except _requests.RequestException:
|
||
debug(f"Failed to fetch script: {abs_url}")
|
||
|
||
if not self._scripts:
|
||
raise RuntimeError("Security JS (cfd114a.js) not found on page")
|
||
|
||
return {
|
||
"html": self._html,
|
||
"scripts": self._scripts,
|
||
"base_url": self.base_url,
|
||
"ts_cd": self._ts_cd,
|
||
"ts_nsd": self._ts_nsd,
|
||
}
|
||
|
||
def request(self, method: str, path: str, token: str,
|
||
data: dict[str, str] = None) -> dict[str, Any]:
|
||
"""向 pickup-web 后端发送 API 请求。
|
||
|
||
Args:
|
||
method: HTTP 方法 (GET/POST)
|
||
path: API 路径 (如 /pickup-web/a/common/customerautocomplete)
|
||
token: c93l5Fya 安全 token
|
||
data: POST 表单数据
|
||
|
||
Returns:
|
||
JSON 响应（如果是 JSON），否则返回 {"_text": response_text}
|
||
"""
|
||
full_url = self._abs_url(path)
|
||
if "?" in full_url:
|
||
full_url += f"&c93l5Fya={token}"
|
||
else:
|
||
full_url += f"?c93l5Fya={token}"
|
||
|
||
headers = {
|
||
"X-Requested-With": "XMLHttpRequest",
|
||
"Accept": "application/json, text/javascript, */*; q=0.01",
|
||
"Referer": self.base_url,
|
||
}
|
||
if data:
|
||
headers["Content-Type"] = "application/x-www-form-urlencoded; charset=UTF-8"
|
||
|
||
debug(f"API {method} {path}")
|
||
resp = self._http.request(
|
||
method, full_url,
|
||
headers=headers,
|
||
data=data or None,
|
||
timeout=30,
|
||
)
|
||
debug(f"API {method} {path} → {resp.status_code}")
|
||
if resp.status_code >= 400:
|
||
debug(f"API error body: {resp.text[:500]}")
|
||
resp.raise_for_status()
|
||
|
||
text = resp.text
|
||
if text and ("<html" in text[:300].lower() or "<!doctype" in text[:300].lower()):
|
||
raise RuntimeError("Session 已过期，返回了登录页面，请重新登录并获取 cookies")
|
||
|
||
content_type = resp.headers.get("Content-Type", "")
|
||
if "json" in content_type or text.strip().startswith(("[", "{")):
|
||
try:
|
||
return resp.json()
|
||
except ValueError:
|
||
pass
|
||
return {"_text": text}
|
||
|
||
|
||
# ─── 业务 API ─────────────────────────────────────────────────────────
|
||
|
||
class PickupAPI:
|
||
"""揽收系统业务 API 封装。
|
||
|
||
每个方法对应页面上的一个 AJAX 操作。
|
||
所有需要 token 的方法都通过 token_bridge 计算 c93l5Fya 参数。
|
||
"""
|
||
|
||
def __init__(self, session: PickupSession, bridge: TokenBridge):
|
||
self._sess = session
|
||
self._bridge = bridge
|
||
|
||
def _token(self, method: str, path: str) -> str:
|
||
return self._bridge.compute_token(method, path)
|
||
|
||
def _post(self, path: str, data: dict[str, str] = None) -> dict:
|
||
token = self._token("POST", path)
|
||
return self._sess.request("POST", path, token, data)
|
||
|
||
def _get(self, path: str) -> dict:
|
||
token = self._token("GET", path)
|
||
return self._sess.request("GET", path, token)
|
||
|
||
# ── 产品 ──────────────────────────────────────────────────
|
||
|
||
def query_product(self, keyword: str) -> list[dict]:
|
||
"""搜索产品名称，返回候选列表（data=id|bizProductNo, value=显示名）。"""
|
||
result = self._post(
|
||
"/pickup-web/a/postwaybilledit/queryproductmanager",
|
||
data={"no": "on", "query": keyword},
|
||
)
|
||
from log import trace
|
||
trace(f"query_product raw: {str(result)[:300]}")
|
||
suggestions = result.get("suggestions", []) if isinstance(result, dict) else []
|
||
# 解析 data 字段: "productId|bizProductNo"
|
||
parsed = []
|
||
for s in suggestions:
|
||
data_parts = s.get("data", "").split("|")
|
||
parsed.append({
|
||
"productId": data_parts[0] if len(data_parts) > 0 else "",
|
||
"bizProductNo": data_parts[1] if len(data_parts) > 1 else "",
|
||
"productName": s.get("value", ""),
|
||
})
|
||
return parsed
|
||
|
||
def select_product(self, biz_product_id: str, biz_product_no: str) -> dict:
|
||
"""选择产品后获取产品详情（模板、材料等）。"""
|
||
product_info = {"bizProductId": biz_product_id, "bizProductNo": biz_product_no}
|
||
|
||
# 获取模板（返回数组，取第一个）
|
||
template = self._post(
|
||
"/pickup-web/a/common/templatecodemodel",
|
||
data={"bizProductNo": biz_product_no},
|
||
)
|
||
tmpl = template[0] if isinstance(template, list) and template else template
|
||
if isinstance(tmpl, dict):
|
||
product_info.update(tmpl)
|
||
|
||
# 查询产品详情
|
||
detail = self._post(
|
||
"/pickup-web/a/common/queryproductbymanager",
|
||
data={
|
||
"bizProductId": biz_product_id,
|
||
"bizProductNo": biz_product_no,
|
||
},
|
||
)
|
||
if isinstance(detail, dict):
|
||
product_info.update(detail)
|
||
elif isinstance(detail, list) and detail:
|
||
product_info.update(detail[0])
|
||
|
||
# 查询物料信息（返回数组或 dict）
|
||
material = self._post(
|
||
"/pickup-web/a/common/findppoductmaterialid",
|
||
data={"productCode": biz_product_no},
|
||
)
|
||
mat = material[0] if isinstance(material, list) and material else material
|
||
if isinstance(mat, dict):
|
||
product_info["material"] = mat
|
||
|
||
return product_info
|
||
|
||
# ── 客户 ──────────────────────────────────────────────────
|
||
|
||
def query_customer(self, keyword: str) -> list[dict]:
|
||
"""搜索客户名称，返回候选列表（data=customerId, valueName=短名, value=全名）。"""
|
||
result = self._post(
|
||
"/pickup-web/a/common/customerautocomplete",
|
||
data={"query": keyword},
|
||
)
|
||
from log import trace
|
||
trace(f"query_customer raw: {str(result)[:300]}")
|
||
suggestions = result.get("suggestions", []) if isinstance(result, dict) else []
|
||
parsed = []
|
||
for s in suggestions:
|
||
parsed.append({
|
||
"customerId": s.get("data", ""),
|
||
"customerName": s.get("valueName", s.get("value", "")),
|
||
"customerFullName": s.get("value", ""),
|
||
})
|
||
return parsed
|
||
|
||
def select_customer(self, customer_id: str) -> dict:
|
||
"""选择客户后获取客户详情。"""
|
||
result = self._post(
|
||
"/pickup-web/a/common/customerins",
|
||
data={"customerId": customer_id},
|
||
)
|
||
if isinstance(result, dict):
|
||
return result
|
||
return {}
|
||
|
||
def find_sender_history(self, sender_no: str) -> dict:
|
||
"""根据寄件人编号查询实名历史。"""
|
||
result = self._post(
|
||
"/pickup-web/a/postwaybillargs/findrealnamehistorybysender",
|
||
data={"senderNo": sender_no},
|
||
)
|
||
if isinstance(result, dict):
|
||
return result
|
||
return {}
|
||
|
||
# ── 运单 ──────────────────────────────────────────────────
|
||
|
||
def query_waybill(self, waybill_no: str, ecommerce_no: str,
|
||
biz_product_no: str, payment_mode: str,
|
||
biz_product_id: str = "",
|
||
post_blend: str = "0",
|
||
lock_customer: str = "0") -> dict:
|
||
"""查询运单信息。"""
|
||
return self._post(
|
||
"/pickup-web/a/postwaybill/querypostwaybillmanager",
|
||
data={
|
||
"waybillNo": waybill_no,
|
||
"ecommerceNo": ecommerce_no,
|
||
"bizProductNo": biz_product_no,
|
||
"postBlend": post_blend,
|
||
"lockCustomer": lock_customer,
|
||
"paymentMode": payment_mode,
|
||
"bizProductId": biz_product_id,
|
||
},
|
||
)
|
||
|
||
# ── 地址 ──────────────────────────────────────────────────
|
||
|
||
def address_match_raw(self, receiver_addr: str) -> dict:
|
||
"""根据地址文本查询区划代码（addressMatch API）。"""
|
||
return self._post(
|
||
"/pickup-web/a/postwaybill/addressMatch",
|
||
data={"receiverAddr": receiver_addr},
|
||
)
|
||
|
||
def query_district(self, dist_code: str) -> dict:
|
||
"""根据区划代码查询地址基本信息。"""
|
||
return self._post(
|
||
"/pickup-web/a/postwaybilledit/addressbasicno",
|
||
data={"dto": dist_code},
|
||
)
|
||
|
||
def match_address(self, receiver_arrive_org_no: str,
|
||
base_product_no: str,
|
||
waybill_no: str,
|
||
receiver_addr: str,
|
||
sender_no: str = "",
|
||
transfer_type: str = "3",
|
||
contents_attribute: str = "3",
|
||
single_seal_flag: str = "0",
|
||
inter_flag: str = "0",
|
||
reserved29: str = "0",
|
||
safety_type: str = "0",
|
||
io_type: str = "10") -> dict:
|
||
"""匹配收件地址，获取寄达地网格等信息。"""
|
||
return self._post(
|
||
"/pickup-web/a/common/findGridPrammer",
|
||
data={
|
||
"receiverArriveOrgNo": receiver_arrive_org_no,
|
||
"baseProductNo": base_product_no,
|
||
"transferType": transfer_type,
|
||
"contentsAttribute": contents_attribute,
|
||
"singleSealFlag": single_seal_flag,
|
||
"interFlag": inter_flag,
|
||
"reserved29": reserved29,
|
||
"waybillNo": waybill_no,
|
||
"receiverAddr": receiver_addr,
|
||
"senderNo": sender_no,
|
||
"safetyType": safety_type,
|
||
"ioType": io_type,
|
||
},
|
||
)
|
||
|
||
# ── 校验 ──────────────────────────────────────────────────
|
||
|
||
def check_product_material(self, cargo_names: str, waybill_no: str,
|
||
product_code: str, dist_code: str) -> dict:
|
||
"""校验产品物料。"""
|
||
return self._post(
|
||
"/pickup-web/a/common/checkppoductmaterialidnew",
|
||
data={
|
||
"cargoNames": cargo_names,
|
||
"cargoOptions": "",
|
||
"waybillNo": waybill_no,
|
||
"productCode": product_code,
|
||
"distCode": dist_code,
|
||
},
|
||
)
|
||
|
||
def check_reach_area(self, product_code: str, dist_code: str,
|
||
cust_code: str, weight: str, waybill_no: str) -> dict:
|
||
"""校验可达区域。"""
|
||
return self._post(
|
||
"/pickup-web/a/common/reachareaverification",
|
||
data={
|
||
"productCode": product_code,
|
||
"distCode": dist_code,
|
||
"custCode": cust_code,
|
||
"weight": weight,
|
||
"waybillNo": waybill_no,
|
||
},
|
||
)
|
||
|
||
def check_important_address(self, waybill_no: str, addr: str) -> dict:
|
||
"""检查重要地址。"""
|
||
return self._post(
|
||
"/pickup-web/a/postwaybill/importantaddress",
|
||
data={"waybillNo": waybill_no, "queryAddr": addr},
|
||
)
|
||
|
||
def check_sensitive_customers(self, mobile_phones: str, waybill_no: str) -> dict:
|
||
"""检查敏感客户。"""
|
||
return self._post(
|
||
"/pickup-web/a/postwaybillargs/findsensitivecustomers",
|
||
data={"mobilePhone": mobile_phones, "waybillNo": waybill_no},
|
||
)
|
||
|
||
# ── 杂项查询 ──────────────────────────────────────────────
|
||
|
||
def query_ecommerce(self, query: str = "B") -> list:
|
||
"""查询电商平台列表。"""
|
||
return self._post(
|
||
"/pickup-web/a/postwaybill/queryecommerce",
|
||
data={"query": query},
|
||
)
|
||
|
||
def query_package_sort(self) -> list:
|
||
"""查询包装分类。"""
|
||
return self._post("/pickup-web/a/common/querypackagesort")
|
||
|
||
def query_deliver_date(self) -> dict:
|
||
"""查询可投递日期。"""
|
||
return self._post("/pickup-web/a/postwaybillargs/deliverdate")
|
||
|
||
def query_reason_temp_storage(self) -> dict:
|
||
"""查询暂存原因列表。"""
|
||
return self._post("/pickup-web/a/postwaybill/queryreasonfortemporarystorage")
|
||
|
||
def check_customer_warning(self) -> dict:
|
||
"""检查客户预警。"""
|
||
return self._post("/pickup-web/a/common/customerwarning")
|
||
|
||
def check_sheet_warning(self) -> dict:
|
||
"""检查面单预警。"""
|
||
return self._post("/pickup-web/a/postwaybill/sheetWarning")
|
||
|
||
# ── 提交 ──────────────────────────────────────────────────
|
||
|
||
def submit(self, form_data: dict[str, str]) -> dict:
|
||
"""提交揽收表单。返回 {"resCode": "", "resMess": "收寄成功", ...}。"""
|
||
return self._post(
|
||
"/pickup-web/a/postwaybill/isnotetosave",
|
||
data=form_data,
|
||
)
|