91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
"""Read YYB's official order list after a payment is completed."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from curl_cffi import requests
|
|
|
|
ORDER_LIST_URL = (
|
|
"https://ydd.yyb.qq.com/trpc.wesee_live.private_domain_creator_shop_svr."
|
|
"private_domain_creator_shop_svr/GetPrivateDomainOrderList"
|
|
)
|
|
USER_AGENT = (
|
|
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36"
|
|
)
|
|
|
|
|
|
def get_official_orders(cookies: dict[str, str], count: int = 20) -> dict[str, Any]:
|
|
"""Fetch the same completed-order list used by YYB's success page."""
|
|
response = requests.post(
|
|
ORDER_LIST_URL,
|
|
json={"count": count, "breakpoint": 0, "type": 1},
|
|
headers={
|
|
"Accept": "application/json, text/plain, */*",
|
|
"Origin": "https://m.yyb.qq.com",
|
|
"Referer": "https://m.yyb.qq.com/boc-mall/goods-mall/product-order-success",
|
|
"User-Agent": USER_AGENT,
|
|
},
|
|
cookies=cookies,
|
|
impersonate="chrome",
|
|
timeout=30,
|
|
)
|
|
if response.status_code != 200:
|
|
raise RuntimeError(f"订单状态查询失败: HTTP {response.status_code}")
|
|
try:
|
|
document = response.json()
|
|
except ValueError as exc:
|
|
raise RuntimeError("订单状态查询返回非 JSON") from exc
|
|
if not isinstance(document, dict):
|
|
raise RuntimeError("订单状态查询响应格式异常")
|
|
if document.get("ret_code") not in (None, 0, "0"):
|
|
raise RuntimeError(
|
|
f"订单状态查询失败: {document.get('ret_code')} {document.get('ret_msg', '')}"
|
|
)
|
|
if not isinstance(document.get("list", []), list):
|
|
raise RuntimeError("订单状态查询响应缺少 list")
|
|
return document
|
|
|
|
|
|
def order_ids(document: dict[str, Any]) -> set[str]:
|
|
"""Return the stable order IDs visible in an order-list response."""
|
|
return {
|
|
str(item["order_id"])
|
|
for item in document.get("list", [])
|
|
if isinstance(item, dict) and item.get("order_id") not in (None, "")
|
|
}
|
|
|
|
|
|
def is_finished(item: dict[str, Any]) -> bool:
|
|
"""YYB's completed-order representation observed on product-order-success."""
|
|
return item.get("is_finished") is True and item.get("status") in (1, "1")
|
|
|
|
|
|
def order_completion_states(document: dict[str, Any]) -> dict[str, bool]:
|
|
"""Snapshot whether each currently visible order is completed."""
|
|
return {
|
|
str(item["order_id"]): is_finished(item)
|
|
for item in document.get("list", [])
|
|
if isinstance(item, dict) and item.get("order_id") not in (None, "")
|
|
}
|
|
|
|
|
|
def find_completed_order(document: dict[str, Any], previous_states: dict[str, bool]) -> dict[str, Any] | None:
|
|
"""Find an order that appeared or transitioned to completed after the QR display."""
|
|
for item in document.get("list", []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
order_id = str(item.get("order_id", ""))
|
|
if order_id and is_finished(item) and not previous_states.get(order_id, False):
|
|
return item
|
|
return None
|
|
|
|
|
|
def completion_summary(item: dict[str, Any]) -> dict[str, Any]:
|
|
"""Persist only the state needed to identify a confirmed completion."""
|
|
return {
|
|
"order_id": str(item.get("order_id", "")),
|
|
"is_finished": item.get("is_finished"),
|
|
"status": item.get("status"),
|
|
}
|