211 lines
9.7 KiB
Python
211 lines
9.7 KiB
Python
#!/usr/bin/env python3
|
||
"""用已登录的 YYB CK 选择和平精英点券档位、区服和角色。"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import random
|
||
import re
|
||
import sys
|
||
import time
|
||
import uuid
|
||
from pathlib import Path
|
||
from urllib.parse import unquote, urlencode
|
||
|
||
from curl_cffi import requests
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
from pyvm.login_profile import midas_login_params # noqa: E402
|
||
|
||
YYB_APP_ID = 52575843
|
||
SOURCE_ID = "24292013"
|
||
PRODUCTS_URL = "https://ydd.yyb.qq.com/new_direct_buy_shop/GetTokenMod"
|
||
CMALL_URL = "https://storeapi.pay.qq.com/api/unipay/{offer_id}/cmall_query"
|
||
PLATFORMS = {
|
||
"android": {
|
||
"label": "Android",
|
||
"query_platform": "pc_android",
|
||
"cmall_pf": "mds_storeopen_qb-__mds_myappjp_PC_aW9zd2hpdGVsaX0-android",
|
||
"order_pf": "mds_myappjp-__mds_myappjp_PC_aW9zd2hpdGVsaX0-android",
|
||
},
|
||
"ios": {
|
||
"label": "iOS",
|
||
"query_platform": "pc_ios",
|
||
"cmall_pf": "mds_storeopen_qb-__mds_myappjp_PC_aW9zd2hpdGVsaX0-iap",
|
||
"order_pf": "mds_myappjp-__mds_myappjp_PC_aW9zd2hpdGVsaX0-iap",
|
||
},
|
||
}
|
||
UA = (
|
||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36"
|
||
)
|
||
|
||
|
||
def load_cookies(path: Path) -> dict[str, str]:
|
||
session = json.loads(path.read_text(encoding="utf-8"))
|
||
cookies = {str(key): str(value) for key, value in session.get("cookies", {}).items() if value}
|
||
if not cookies.get("openid") or not cookies.get("accesstoken"):
|
||
raise ValueError("会话缺少 openid/accesstoken，请先执行 login-wechat.py")
|
||
return cookies
|
||
|
||
|
||
def product_options(cookies: dict[str, str], platform: str) -> list[dict[str, str | int]]:
|
||
response = requests.post(
|
||
PRODUCTS_URL,
|
||
json={"platform": PLATFORMS[platform]["query_platform"], "source_id": SOURCE_ID,
|
||
"yyb_app_id": YYB_APP_ID},
|
||
headers={"Accept": "application/json, text/plain, */*", "Origin": "https://m.yyb.qq.com",
|
||
"Referer": "https://m.yyb.qq.com/boc-mall/goods-mall/", "User-Agent": UA},
|
||
cookies=cookies, impersonate="chrome", timeout=30,
|
||
)
|
||
if response.status_code != 200:
|
||
raise RuntimeError(f"点券商品查询失败: HTTP {response.status_code}")
|
||
document = response.json()
|
||
if document.get("code") not in (None, 0):
|
||
raise RuntimeError(f"点券商品查询失败: {document.get('code')} {document.get('message', '')}")
|
||
products = document.get("token_mod", {}).get("products", [])
|
||
result: list[dict[str, str | int]] = []
|
||
for item in products:
|
||
product = item.get("product", {}) if isinstance(item, dict) else {}
|
||
match = re.fullmatch(r"(\d+)点券", str(product.get("product_name", "")))
|
||
if not match or str(product.get("status")) != "20":
|
||
continue
|
||
result.append({
|
||
"points": int(match.group(1)),
|
||
"product_id": str(product.get("product_id", "")),
|
||
"price_fen": int(product.get("price", 0)),
|
||
"offer_id": str(product.get("res_offer_id", "")),
|
||
"name": str(product.get("product_name", "")),
|
||
})
|
||
result.sort(key=lambda item: int(item["points"]))
|
||
if not result or any(not item["product_id"] or not item["offer_id"] for item in result):
|
||
raise RuntimeError("点券商品响应缺少 product_id 或 offer_id")
|
||
return result
|
||
|
||
|
||
class Cmall:
|
||
def __init__(self, cookies: dict[str, str], offer_id: str, platform: str) -> None:
|
||
self.cookies = cookies
|
||
self.offer_id = offer_id
|
||
self.platform = platform
|
||
self.session_token = f"{str(uuid.uuid4()).upper()}{int(time.time() * 1000)}"
|
||
|
||
def query(self, cmd: str, **extra: str) -> dict:
|
||
login = midas_login_params(self.cookies)
|
||
params = {
|
||
"from_h5": "1", "pf": PLATFORMS[self.platform]["cmall_pf"], "r": str(random.random()), "cmd": cmd,
|
||
"session_token": self.session_token,
|
||
"pfkey": "pfkey", "webversion": "", **extra,
|
||
**login,
|
||
}
|
||
# 当前商城页面将查询参数放在 URL 上，但请求方法为 POST 且没有 body。
|
||
response = requests.post(
|
||
CMALL_URL.format(offer_id=self.offer_id) + "?" + urlencode(params),
|
||
headers={"Origin": "https://z.iwan.yyb.qq.com", "Referer": "https://z.iwan.yyb.qq.com/",
|
||
"User-Agent": UA},
|
||
cookies=self.cookies, impersonate="chrome", timeout=30,
|
||
)
|
||
if response.status_code != 200:
|
||
raise RuntimeError(f"游戏数据查询失败: HTTP {response.status_code}")
|
||
document = response.json()
|
||
if document.get("ret") not in (0, "0"):
|
||
raise RuntimeError(f"游戏数据查询失败: {document.get('ret')} {document.get('msg', '')}")
|
||
return document
|
||
|
||
def zones(self) -> list[dict[str, str]]:
|
||
response = self.query("14", use_currency_offerid="1")
|
||
zones = response.get("zone_list", [])
|
||
return [{"zone_id": str(item.get("zone_id", "")), "name": str(item.get("zone_name", ""))}
|
||
for item in zones if isinstance(item, dict) and item.get("zone_id")]
|
||
|
||
def roles(self, zone_id: str) -> list[dict[str, str]]:
|
||
response = self.query("15", use_currency_offerid="1", zoneid=zone_id)
|
||
roles = response.get("role_info") or response.get("role_list") or []
|
||
# QQ 平台和平精英的 area 与 zoneid 不同(如 area=2, zoneid=1)；
|
||
# PlaceOrder 校验角色时需要该分区信息，随角色一并返回。
|
||
partition = response.get("partition_info") or {}
|
||
return [{
|
||
"role_id": str(item.get("role_id", "")),
|
||
"name": unquote(str(item.get("role_name", ""))),
|
||
"ban_status": str(item.get("ban_status", "")),
|
||
"area": str(partition.get("area", "")),
|
||
"partition": str(partition.get("partition", "")),
|
||
"platid": str(partition.get("platid", "")),
|
||
} for item in roles if isinstance(item, dict) and item.get("role_id")]
|
||
|
||
|
||
def choose(label: str, options: list[dict], display) -> dict:
|
||
if not options:
|
||
raise RuntimeError(f"没有可选择的{label}")
|
||
print(f"\n可选{label}:")
|
||
for index, item in enumerate(options, start=1):
|
||
print(f" {index}. {display(item)}")
|
||
while True:
|
||
raw = input(f"请选择{label}序号 [1-{len(options)}]: ").strip()
|
||
if raw.isdigit() and 1 <= int(raw) <= len(options):
|
||
return options[int(raw) - 1]
|
||
print("请输入列表中的有效序号。")
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="和平精英点券和角色选择（纯 CK 查询）")
|
||
parser.add_argument("--session", type=Path, default=ROOT / "config/mall-session.json")
|
||
parser.add_argument("--output", type=Path, default=ROOT / "config/peace-elite-selection.json")
|
||
parser.add_argument("--platform", choices=tuple(PLATFORMS), default=None,
|
||
help="预选 Android/iOS；不传则显示平台菜单")
|
||
parser.add_argument("--points", type=int, default=None, help="预选点券数；不传则显示菜单")
|
||
parser.add_argument("--list-products", action="store_true", help="仅列出当前所有点券档位")
|
||
args = parser.parse_args()
|
||
|
||
cookies = load_cookies(args.session)
|
||
if args.platform is None:
|
||
selected_platform = choose(
|
||
"平台", [{"id": key, **value} for key, value in PLATFORMS.items()],
|
||
lambda item: item["label"],
|
||
)
|
||
platform = str(selected_platform["id"])
|
||
else:
|
||
platform = args.platform
|
||
print(f"已选择平台: {PLATFORMS[platform]['label']}")
|
||
products = product_options(cookies, platform)
|
||
if args.list_products:
|
||
for product in products:
|
||
print(f"{product['points']}点券\t{product['price_fen'] / 100:g}元\t{product['product_id']}")
|
||
return 0
|
||
if args.points is None:
|
||
product = choose("点券档位", products, lambda item: f"{item['points']}点券（{item['price_fen'] / 100:g}元）")
|
||
else:
|
||
product = next((item for item in products if item["points"] == args.points), None)
|
||
if product is None:
|
||
available = ", ".join(str(item["points"]) for item in products)
|
||
raise ValueError(f"不支持 {args.points} 点券；当前可选: {available}")
|
||
print(f"已选择: {product['points']}点券（{product['price_fen'] / 100:g}元）")
|
||
|
||
cmall = Cmall(cookies, str(product["offer_id"]), platform)
|
||
zone = choose("区服", cmall.zones(), lambda item: f"{item['name']}（ID {item['zone_id']}）")
|
||
roles = cmall.roles(zone["zone_id"])
|
||
role = choose("角色", roles, lambda item: f"{item['name']}（{'禁用' if item['ban_status'] == '1' else '可用'}）")
|
||
if role["ban_status"] == "1":
|
||
raise RuntimeError("所选角色已被封禁，不能充值")
|
||
|
||
selection = {"platform": platform, "order_pf": PLATFORMS[platform]["order_pf"],
|
||
"points": product["points"], "product_id": product["product_id"],
|
||
"offer_id": product["offer_id"], "price_fen": product["price_fen"],
|
||
"zone_id": zone["zone_id"], "zone_name": zone["name"],
|
||
"role_id": role["role_id"], "role_name": role["name"]}
|
||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||
args.output.write_text(json.dumps(selection, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
print(f"\n已选择 {selection['points']}点券 / {selection['zone_name']} / {selection['role_name']}")
|
||
print(f"选择已保存: {args.output}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
raise SystemExit(main())
|
||
except (OSError, ValueError, RuntimeError, requests.RequestsError) as error:
|
||
print(f"选择失败: {error}")
|
||
raise SystemExit(1) from error
|