353 lines
15 KiB
Python
353 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""使用 jsdom DeviceFP 完成一笔已创建 YYB 订单的 web_save。
|
|
|
|
本脚本不创建 mall 订单；先由 ``main.py mall auto`` 创建订单，再执行本脚本。
|
|
付款码只在服务端 ``web_save`` 返回 ``ret=0`` 后渲染。付款后通过商城官方订单
|
|
列表确认本次新出现的完成订单；该检查不触发付款或确认操作。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import random
|
|
import re
|
|
import string
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from pyvm.algorithm import derive_key1_from_key16, generate_encrypt_msg_offline # noqa: E402
|
|
from pyvm.login_profile import midas_login_params # noqa: E402
|
|
from pyvm.order_status import ( # noqa: E402
|
|
completion_summary,
|
|
find_completed_order,
|
|
get_official_orders,
|
|
order_completion_states,
|
|
order_ids,
|
|
)
|
|
|
|
APPID = "1450243039"
|
|
GOODS_URL = "https://pay.qq.com/midas/minipay_v2/views/cpay/goods.shtml"
|
|
SAVE_URL = f"https://api.unipay.qq.com/v1/r/{APPID}/web_save"
|
|
FP_URL = "https://api.unipay.qq.com/cgi-bin/fp-behv.fcg"
|
|
ORDER_FIELDS = [
|
|
"token_id", "openid", "openkey", "session_id", "session_type", "zoneid",
|
|
"pay_method", "buy_quantity", "mb_pwd", "pay_id", "auth_key", "card_value",
|
|
"accounttype", "provide_uin", "extend", "from_h5", "webversion",
|
|
]
|
|
|
|
|
|
def load_json(path: Path) -> dict:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def parse_url_params(mall_response: dict) -> dict[str, str]:
|
|
try:
|
|
call_reply = json.loads(mall_response["data"]["call_reply"])
|
|
url_params = call_reply["data"]["url_params"]
|
|
except (KeyError, TypeError, json.JSONDecodeError) as exc:
|
|
raise ValueError("mall 响应缺少 data.call_reply.data.url_params") from exc
|
|
return {key: values[-1] for key, values in urllib.parse.parse_qs(
|
|
urllib.parse.urlparse(url_params).query, keep_blank_values=True).items()}
|
|
|
|
|
|
def cookie_header(cookies: dict[str, str]) -> str:
|
|
pairs = dict(cookies)
|
|
if not pairs.get("midas_openid") and pairs.get("openid"):
|
|
pairs["midas_openid"] = pairs["openid"]
|
|
if not pairs.get("midas_openkey") and pairs.get("accesstoken"):
|
|
pairs["midas_openkey"] = pairs["accesstoken"]
|
|
return "; ".join(f"{name}={value}" for name, value in pairs.items() if value)
|
|
|
|
|
|
def request_bytes(url: str, cookies: dict[str, str], body: bytes | None = None) -> str:
|
|
headers = {
|
|
"User-Agent": "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",
|
|
"Cookie": cookie_header(cookies),
|
|
"Referer": "https://pay.qq.com/",
|
|
}
|
|
if body is not None:
|
|
headers["Content-Type"] = "application/x-www-form-urlencoded"
|
|
headers["Origin"] = "https://pay.qq.com"
|
|
request = urllib.request.Request(url, data=body, headers=headers,
|
|
method="POST" if body is not None else "GET")
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
return response.read().decode("utf-8", "replace")
|
|
except urllib.error.HTTPError as exc:
|
|
return exc.read().decode("utf-8", "replace")
|
|
|
|
|
|
def goods_page_url(cookies: dict[str, str], order: dict[str, str], zone_id: str = "1", pf: str = "") -> str:
|
|
openid = cookies.get("openid") or cookies.get("midas_openid")
|
|
openkey = cookies.get("accesstoken") or cookies.get("midas_openkey")
|
|
if not openid or not openkey:
|
|
raise ValueError("会话缺少 openid/accesstoken，请先完成授权扫码登录")
|
|
login = midas_login_params(cookies)
|
|
params = {
|
|
"appid": APPID,
|
|
"openid": openid,
|
|
"openkey": openkey,
|
|
"session_id": login["session_id"],
|
|
"session_type": login["session_type"],
|
|
"sandbox": "",
|
|
"wxappid": login["wx_appid"],
|
|
"qqAppid": login["qq_appid"],
|
|
"pf": pf or "mds_myappjp-__mds_myappjp_PC_aW9zd2hpdGVsaX0-android",
|
|
"buy_quantity": "1",
|
|
"goodstokenurl": order["url_params"],
|
|
"zoneid": zone_id,
|
|
"supportCloseConfirm": "1",
|
|
"t": str(int(time.time() * 1000)),
|
|
}
|
|
return GOODS_URL + "?" + urllib.parse.urlencode(params)
|
|
|
|
|
|
def extract_goods_state(html: str) -> tuple[list[int], str, str]:
|
|
ops = re.search(r"var\s+xMidasOps\s*=\s*\[([^]]+)]", html)
|
|
token = re.search(r'id="xMidasToken"\s+value="([0-9A-Fa-f]+)"', html)
|
|
anti = re.search(r'id="antiAutoScriptToken"\s+value="([0-9A-Fa-f]+)"', html)
|
|
if not ops or not token or not anti:
|
|
raise ValueError("goods 页面缺少 xMidasOps/xMidasToken/antiAutoScriptToken，登录态或订单已失效")
|
|
xmidas = [int(value) for value in ops.group(1).split(",") if value]
|
|
if len(xmidas) != 59640:
|
|
raise ValueError(f"goods xMidasOps 长度异常: {len(xmidas)}")
|
|
return xmidas, token.group(1).upper(), anti.group(1).upper()
|
|
|
|
|
|
def load_template_args() -> tuple[str, list]:
|
|
runtime_template = ROOT / "replay/live/default"
|
|
runtime_args = runtime_template / "args-template.json"
|
|
runtime_body = runtime_template / "body.txt"
|
|
if runtime_args.exists() and runtime_body.exists():
|
|
return runtime_body.read_text(encoding="utf-8"), load_json(runtime_args)
|
|
candidates = sorted((ROOT / "replay/live").glob("*/args-template.json"), reverse=True)
|
|
for path in candidates:
|
|
if path.exists():
|
|
body = (path.parent / "body.txt").read_text(encoding="utf-8")
|
|
return body, load_json(path)
|
|
raise FileNotFoundError("缺少归档 goods body.txt/args-template.json")
|
|
|
|
|
|
def replace_form_value(body: str, name: str, value: str) -> str:
|
|
return re.sub(rf"({re.escape(name)}=)[^&]*", rf"\g<1>{value}", body, count=1)
|
|
|
|
|
|
def build_save_body(template: str, order: dict[str, str], cookies: dict[str, str], web_token: str,
|
|
anti_token: str, encrypt_msg: str) -> str:
|
|
login = midas_login_params(cookies)
|
|
values = {
|
|
"token_id": order.get("token_id", ""),
|
|
"transaction_id": order.get("transaction_id", ""),
|
|
"out_trade_no": order.get("out_trade_no", ""),
|
|
"offer_type": order.get("offer_type", "0"),
|
|
"openid": cookies.get("openid", ""),
|
|
"openkey": cookies.get("accesstoken", ""),
|
|
"sck": hashlib.md5((APPID + cookies.get("accesstoken", "")).encode()).hexdigest().upper(),
|
|
"web_token": web_token,
|
|
"anti_auto_script_token_id": anti_token,
|
|
"pc_st": str(uuid.uuid4()).upper() + str(int(time.time() * 1000)),
|
|
"r": str(random.random()),
|
|
"t": str(int(time.time() * 1000)),
|
|
"encrypt_msg": encrypt_msg,
|
|
"session_id": login["session_id"],
|
|
"session_type": login["session_type"],
|
|
"wx_appid": login["wx_appid"],
|
|
"qq_appid": login["qq_appid"],
|
|
}
|
|
body = template
|
|
for name, value in values.items():
|
|
body = replace_form_value(body, name, value)
|
|
return body
|
|
|
|
|
|
def make_qr(sign: str, output: Path) -> None:
|
|
try:
|
|
import segno
|
|
except ImportError as exc:
|
|
raise RuntimeError("缺少 segno；请安装后重新执行: python3 -m pip install segno") from exc
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
segno.make(sign).save(str(output), scale=6, border=2)
|
|
|
|
|
|
def node_environment() -> dict[str, str]:
|
|
"""Prevent a developer's Node inspector setting from affecting the jsdom worker."""
|
|
environment = os.environ.copy()
|
|
environment.pop("NODE_OPTIONS", None)
|
|
return environment
|
|
|
|
|
|
def save_payment_status(path: Path, document: dict, baseline: dict[str, bool], matched: dict | None = None) -> None:
|
|
"""Persist a small, non-payment-side-effect status record for this run."""
|
|
record = {
|
|
"checked_at": int(time.time()),
|
|
"baseline_order_states": baseline,
|
|
"listed_order_ids": sorted(order_ids(document)),
|
|
"matched_completion": completion_summary(matched) if matched else None,
|
|
}
|
|
path.write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="YYB: jsdom DeviceFP + web_save -> 微信付款码")
|
|
parser.add_argument("--session", default=str(ROOT / "config/mall-session.json"))
|
|
parser.add_argument("--mall-response", default=str(ROOT / "config/mall-order-response.json"))
|
|
parser.add_argument("--out-dir", default=None, help="运行证据目录；默认 config/jsdom-order-<timestamp>")
|
|
parser.add_argument("--qr", default=None, help="付款二维码 PNG 路径")
|
|
parser.add_argument("--wait", type=int, default=12, help="jsdom 等待 DeviceFP 的秒数")
|
|
parser.add_argument("--zone-id", default="1", help="所选游戏区服 ID")
|
|
parser.add_argument("--pf", default="", help="所选 Android/iOS 支付平台标识")
|
|
parser.add_argument("--dry-run", action="store_true", help="仅拉取页面并生成 DeviceFP，不上报或创建付款码")
|
|
parser.add_argument("--payment-timeout", type=float, default=300,
|
|
help="付款码生成后等待订单完成的最长秒数")
|
|
parser.add_argument("--payment-interval", type=float, default=3,
|
|
help="订单完成状态检查间隔秒数")
|
|
parser.add_argument("--skip-payment-check", action="store_true", help="仅生成付款码，不等待订单完成")
|
|
args = parser.parse_args()
|
|
if args.payment_timeout <= 0 or args.payment_interval <= 0:
|
|
raise ValueError("--payment-timeout 和 --payment-interval 必须为正数")
|
|
|
|
session_path = Path(args.session).resolve()
|
|
response_path = Path(args.mall_response).resolve()
|
|
session = load_json(session_path)
|
|
cookies = dict(session.get("cookies", {}))
|
|
if not cookies:
|
|
raise ValueError("会话 cookies 为空，请先完成授权扫码登录")
|
|
response = load_json(response_path)
|
|
order = parse_url_params(response)
|
|
if not order.get("token_id"):
|
|
raise ValueError("mall 响应未包含 token_id")
|
|
try:
|
|
call_reply = json.loads(response["data"]["call_reply"])
|
|
order["url_params"] = call_reply["data"]["url_params"]
|
|
except (KeyError, TypeError, json.JSONDecodeError) as exc:
|
|
raise ValueError("mall 响应无法解析 url_params") from exc
|
|
|
|
stamp = time.strftime("%Y%m%d-%H%M%S")
|
|
out_dir = Path(args.out_dir) if args.out_dir else ROOT / "config" / f"jsdom-order-{stamp}"
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
url = goods_page_url(cookies, order, args.zone_id, args.pf)
|
|
print(f"[jsdom-pay] 订单: {order['token_id'][:20]}...")
|
|
print("[jsdom-pay] 拉取同订单 goods 页面...")
|
|
html = request_bytes(url, cookies)
|
|
html_path = out_dir / "goods.html"
|
|
html_path.write_text(html, encoding="utf-8")
|
|
xmidas, web_token, anti_token = extract_goods_state(html)
|
|
print(f"[jsdom-pay] goods: xMidasOps={len(xmidas)} web_token={web_token[:12]}...")
|
|
|
|
fp_path = out_dir / "device-fp.json"
|
|
command = [
|
|
"node", str(ROOT / "scripts/generate-devicefp-jsdom.mjs"),
|
|
"--html", str(html_path), "--goods-url", url, "--cookies", str(session_path),
|
|
"--output", str(fp_path), "--wait", str(args.wait * 1000),
|
|
]
|
|
print("[jsdom-pay] 运行 jsdom DeviceFP...")
|
|
subprocess.run(command, check=True, cwd=ROOT, env=node_environment())
|
|
fp = load_json(fp_path)
|
|
if args.dry_run:
|
|
print(f"[jsdom-pay] dry-run 完成，证据目录: {out_dir}")
|
|
return 0
|
|
|
|
print("[jsdom-pay] 上报 fp-behv...")
|
|
fp_response = request_bytes(fp.get("fp_url", FP_URL), cookies, fp["fp_body"].encode())
|
|
(out_dir / "fp-response.json").write_text(fp_response, encoding="utf-8")
|
|
|
|
template, web_args = load_template_args()
|
|
key16 = [random.randrange(256) for _ in range(16)]
|
|
tables = [web_args[index][0] for index in (1, 2, 3, 4)]
|
|
key1 = derive_key1_from_key16(key16, te_tables=tables, sbox=web_args[5][0])
|
|
random_suffix = "".join(random.choices(string.ascii_letters + string.digits, k=8)) + "\x01"
|
|
params = {
|
|
"token_id": order.get("token_id", ""),
|
|
"openid": cookies.get("openid", ""),
|
|
"openkey": cookies.get("accesstoken", ""),
|
|
"session_id": "hy_gameid",
|
|
"session_type": "wc_actoken",
|
|
"zoneid": "1",
|
|
"pay_method": "wechat",
|
|
"buy_quantity": "1",
|
|
"mb_pwd": "",
|
|
"pay_id": "",
|
|
"auth_key": "",
|
|
"card_value": "",
|
|
"accounttype": "",
|
|
"provide_uin": "",
|
|
"extend": "",
|
|
"from_h5": "1",
|
|
"webversion": "web_1.0.6",
|
|
}
|
|
# The archived page body contains server-selected payment fields. Preserve
|
|
# them, replacing only values that are tied to the fresh order/session.
|
|
for field in ORDER_FIELDS:
|
|
match = re.search(rf"(?:^|&){field}=([^&]*)", template)
|
|
if match:
|
|
params[field] = match.group(1)
|
|
params.update(midas_login_params(cookies))
|
|
now_seconds = str(int(time.time()))
|
|
encrypt_msg = generate_encrypt_msg_offline(
|
|
params, "tdrc_session%3D" + fp["session_id"], now_seconds, random_suffix,
|
|
key16=key16, key1=key1, args_template=web_args, xmidas=xmidas, xmidas_token=web_token,
|
|
)
|
|
body = build_save_body(template, order, cookies, web_token, anti_token, encrypt_msg)
|
|
print("[jsdom-pay] 提交 web_save...")
|
|
raw = request_bytes(SAVE_URL, cookies, body.encode())
|
|
(out_dir / "web-save-response.json").write_text(raw, encoding="utf-8")
|
|
try:
|
|
response_json = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
response_json = {}
|
|
if response_json.get("ret") != 0:
|
|
print(f"[jsdom-pay] web_save 失败: ret={response_json.get('ret')} {response_json.get('msg', '')}")
|
|
print(f"[jsdom-pay] 证据目录: {out_dir}")
|
|
return 1
|
|
sign = response_json.get("info", {}).get("channel_info", {}).get("sign", "")
|
|
if not sign.startswith("weixin://"):
|
|
print("[jsdom-pay] web_save 成功，但响应没有微信付款链接")
|
|
return 1
|
|
qr_path = Path(args.qr) if args.qr else out_dir / "wechat-pay.png"
|
|
make_qr(sign, qr_path)
|
|
print("[jsdom-pay] 微信付款码已生成")
|
|
print(f"[jsdom-pay] PNG: {qr_path}")
|
|
print(f"[jsdom-pay] 付款链接: {sign}")
|
|
if args.skip_payment_check:
|
|
print("[jsdom-pay] 已跳过付款结果检查。")
|
|
return 0
|
|
|
|
print("[jsdom-pay] 记录付款前订单列表，等待微信付款完成...")
|
|
baseline_document = get_official_orders(cookies)
|
|
baseline_states = order_completion_states(baseline_document)
|
|
status_path = out_dir / "payment-status.json"
|
|
save_payment_status(status_path, baseline_document, baseline_states)
|
|
deadline = time.monotonic() + args.payment_timeout
|
|
while time.monotonic() < deadline:
|
|
time.sleep(args.payment_interval)
|
|
document = get_official_orders(cookies)
|
|
completed = find_completed_order(document, baseline_states)
|
|
save_payment_status(status_path, document, baseline_states, completed)
|
|
if completed:
|
|
print("[jsdom-pay] 微信付款成功，商城订单已完成。")
|
|
print(f"[jsdom-pay] 状态记录: {status_path}")
|
|
return 0
|
|
print("[jsdom-pay] 在等待期限内未确认到本次订单完成。")
|
|
print(f"[jsdom-pay] 状态记录: {status_path}")
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except (FileNotFoundError, ValueError, RuntimeError, subprocess.CalledProcessError) as exc:
|
|
print(f"错误: {exc}", file=sys.stderr)
|
|
raise SystemExit(2) from exc
|