yml/live-hub-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

style: 统一 Ruff 代码格式

Browse Source
This commit is contained in:
yml2213
2026-08-30 21:04:52 +08:00
parent c891ac982e
commit 47e19ed7b2
90 changed files with 5574 additions and 2350 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/yyb-worker/runtime/scripts/select-peace-elite.py
+135 -51
View File
@@ -1,5 +1,6 @@
#!/usr/bin/env python3
"""用已登录的 YYB CK 选择和平精英点券档位、区服和角色。"""
from __future__ import annotations
import argparse
@@ -45,26 +46,43 @@ UA = (
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}
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]]:
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,
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', '')}")
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:
@@ -72,15 +90,19 @@ def product_options(cookies: dict[str, str], platform: str) -> list[dict[str, st
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.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):
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
@@ -95,30 +117,48 @@ class Cmall:
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,
"from_h5": "1",
"pf": PLATFORMS[self.platform]["cmall_pf"],
"r": str(random.random()),
"cmd": cmd,
"session_token": self.session_token,
"pfkey": "pfkey", "webversion": "", **extra,
"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,
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', '')}")
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")]
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)
@@ -126,14 +166,18 @@ class Cmall:
# 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")]
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:
@@ -151,18 +195,31 @@ def choose(label: str, options: list[dict], display) -> dict:
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="仅列出当前所有点券档位")
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()],
"平台",
[{"id": key, **value} for key, value in PLATFORMS.items()],
lambda item: item["label"],
)
platform = str(selected_platform["id"])
@@ -172,32 +229,59 @@ def main() -> int:
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']}")
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}元）")
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)
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']}）")
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 '可用'}）")
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"]}
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']}")
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