Files
auto-pickup-v2/pickup_api.py
T

616 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""揽收系统 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 到文件(保存白名单中的必需及可选 cookie)。
requests.Session 会自动合并服务端 Set-Cookie 响应头,
可能累积非必需 cookie。保存时只保留白名单中的 cookie。
"""
from login_helper import ALLOWED_COOKIE_NAMES
all_cookies = [
{"name": c.name, "value": c.value, "domain": c.domain, "path": c.path}
for c in self._http.cookies
]
saved_cookies = [c for c in all_cookies if c["name"] in ALLOWED_COOKIE_NAMES]
skipped = len(all_cookies) - len(saved_cookies)
data = {
"cookies": saved_cookies,
"total_in_session": len(all_cookies),
"saved": len(saved_cookies),
"skipped": skipped,
}
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,
)