107 lines
3.6 KiB
Python
107 lines
3.6 KiB
Python
# 1 号药城加入购物车。
|
||
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
from typing import Any
|
||
|
||
import requests
|
||
import urllib3
|
||
|
||
from cart_add import KIND_OK, KIND_SYSTEM, AddOutcome, classify_mall_message
|
||
from cart_match import CartOffer
|
||
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
|
||
GATEWAY = "https://gateway-b2b.fangkuaiyi.com"
|
||
YAOEX_ADD_URL = f"{GATEWAY}/small/shopCart/addShopCart"
|
||
|
||
|
||
# 按再搜到的 sku 把数量加入 1 号药城购物车。没有 sku 当系统失败，不改业务原因。
|
||
def add_yaoex_cart(auth, offer: CartOffer, qty: int, *, session: requests.Session | None = None) -> AddOutcome:
|
||
ids = _ids(offer)
|
||
if not ids["skuId"] and not ids["spuCode"]:
|
||
return AddOutcome(kind=KIND_SYSTEM, reason="缺少 sku")
|
||
payload = _form(
|
||
auth,
|
||
{
|
||
"skuId": ids["skuId"],
|
||
"spuCode": ids["spuCode"],
|
||
"sellerCode": ids["sellerCode"],
|
||
"amount": int(qty),
|
||
},
|
||
)
|
||
own = session is None
|
||
client = session or requests.Session()
|
||
try:
|
||
response = client.post(
|
||
YAOEX_ADD_URL,
|
||
data=payload,
|
||
headers=_headers(),
|
||
timeout=30,
|
||
verify=False,
|
||
)
|
||
response.raise_for_status()
|
||
return _outcome_from_json(response.json())
|
||
except Exception as exc:
|
||
return AddOutcome(kind=KIND_SYSTEM, reason=str(exc))
|
||
finally:
|
||
if own:
|
||
client.close()
|
||
|
||
|
||
# 从报价里取出加车要的 sku、spu 和商家编码。
|
||
def _ids(offer: CartOffer) -> dict[str, str]:
|
||
raw = offer.extra.get("raw") if isinstance(offer.extra.get("raw"), dict) else offer.extra
|
||
if not isinstance(raw, dict):
|
||
raw = {}
|
||
sku = str(raw.get("skuId") or raw.get("id") or offer.sku_id or "")
|
||
spu = str(raw.get("spuCode") or offer.sku_id or "")
|
||
seller = str(raw.get("sellerCode") or raw.get("vendorCode") or raw.get("shopCode") or "")
|
||
return {"skuId": sku, "spuCode": spu, "sellerCode": seller}
|
||
|
||
|
||
# 拼 1 号药城表单：带上登录 token，再叠上加车字段。
|
||
def _form(auth, 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": auth.token,
|
||
"userToken": auth.token,
|
||
"userId": getattr(auth, "user_id", ""),
|
||
"buyerCode": getattr(auth, "buyer_code", ""),
|
||
}
|
||
payload.update(extra)
|
||
return payload
|
||
|
||
|
||
# 1 号药城加车请求头。来源页是购物车。
|
||
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/cart/",
|
||
}
|
||
|
||
|
||
# 把加车回包收成成功、业务失败或系统失败。看不懂的回包当系统失败。
|
||
def _outcome_from_json(payload: object) -> AddOutcome:
|
||
if not isinstance(payload, dict):
|
||
return AddOutcome(kind=KIND_SYSTEM, reason="加购返回异常")
|
||
msg = str(payload.get("msg") or payload.get("message") or payload.get("errorMsg") or "")
|
||
code = payload.get("code")
|
||
success = payload.get("success")
|
||
if success is True or code in (0, "0", 200, "200"):
|
||
return AddOutcome(kind=KIND_OK)
|
||
if msg:
|
||
return classify_mall_message(msg)
|
||
return AddOutcome(kind=KIND_SYSTEM, reason="加购失败")
|