"""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"), }