124 lines
3.6 KiB
Python
124 lines
3.6 KiB
Python
"""1 号药城订单导出与详情。"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import time
|
||
from datetime import date, datetime, timedelta
|
||
from typing import Any
|
||
|
||
import requests
|
||
import urllib3
|
||
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
|
||
GATEWAY = "https://gateway-b2b.fangkuaiyi.com"
|
||
EXPORT_URL = f"{GATEWAY}/api/order/export"
|
||
DETAIL_URL = f"{GATEWAY}/api/order/detail"
|
||
|
||
|
||
def day_range(day: date | None = None, days: int = 0) -> tuple[str, str]:
|
||
end = day or date.today()
|
||
start = end - timedelta(days=max(days, 0))
|
||
return (
|
||
datetime.combine(start, datetime.min.time()).strftime("%Y-%m-%d 00:00:00"),
|
||
f"{end.isoformat()} 23:59:59",
|
||
)
|
||
|
||
|
||
def export_orders(
|
||
token: str,
|
||
*,
|
||
begin: str,
|
||
end: str,
|
||
session: requests.Session | None = None,
|
||
) -> bytes:
|
||
params = {
|
||
"queryHistory": "false",
|
||
"productName": "",
|
||
"supplyName": "",
|
||
"flowId": "",
|
||
"createBeginTime": begin,
|
||
"createEndTime": end,
|
||
"orderStatus": 0,
|
||
}
|
||
own = session is None
|
||
client = session or requests.Session()
|
||
try:
|
||
response = client.post(
|
||
EXPORT_URL,
|
||
data=_form(token, {"jsonParams": json.dumps(params, ensure_ascii=False)}),
|
||
headers=_headers(),
|
||
timeout=180,
|
||
verify=False,
|
||
)
|
||
response.raise_for_status()
|
||
if "json" in (response.headers.get("content-type") or "") and response.content[:2] != b"PK":
|
||
raise RuntimeError(f"导出失败：{response.text[:300]}")
|
||
if response.content[:8] != b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" and response.content[:2] != b"PK":
|
||
head = response.content[:80]
|
||
raise RuntimeError(f"导出不是 Excel：{head!r}")
|
||
return response.content
|
||
finally:
|
||
if own:
|
||
client.close()
|
||
|
||
|
||
def fetch_order_details(
|
||
token: str,
|
||
order_nos: list[str],
|
||
*,
|
||
session: requests.Session | None = None,
|
||
) -> dict[str, dict[str, Any]]:
|
||
own = session is None
|
||
client = session or requests.Session()
|
||
details: dict[str, dict[str, Any]] = {}
|
||
try:
|
||
for order_no in order_nos:
|
||
response = client.post(
|
||
DETAIL_URL,
|
||
data=_form(token, {"jsonParams": json.dumps({"orderId": order_no})}),
|
||
headers={
|
||
**_headers(),
|
||
"clientInfo": json.dumps({"clientAppVersionCode": 6996}),
|
||
},
|
||
timeout=60,
|
||
verify=False,
|
||
)
|
||
response.raise_for_status()
|
||
payload = response.json()
|
||
data = payload.get("data") if isinstance(payload, dict) else None
|
||
if not isinstance(data, dict):
|
||
data = payload if isinstance(payload, dict) else {}
|
||
details[order_no] = data
|
||
return details
|
||
finally:
|
||
if own:
|
||
client.close()
|
||
|
||
|
||
def _form(token: str, extra: dict[str, Any]) -> dict[str, Any]:
|
||
payload = {
|
||
"traderName": "yaoex_pc",
|
||
"trader": "pc",
|
||
"closesignature": "yes",
|
||
"signature_method": "md5",
|
||
"signature": "****",
|
||
"timestamp": str(int(time.time() * 1000)),
|
||
"token": token,
|
||
"userToken": token,
|
||
}
|
||
payload.update(extra)
|
||
return payload
|
||
|
||
|
||
def _headers() -> dict[str, str]:
|
||
return {
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"X-Requested-With": "XMLHttpRequest",
|
||
"X-Request-Agent": "Axios",
|
||
"traderName": "yaoex_pc",
|
||
"trader": "pc",
|
||
"Origin": "https://mall.yaoex.com",
|
||
"Referer": "https://mall.yaoex.com/v2/orderList/",
|
||
}
|