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

流程通了

Browse Source
This commit is contained in:
yml2213
2026-08-12 17:30:00 +08:00
parent 03f5b8a6c4
commit cebbd94ae8
7 changed files with 124 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+2
View File
@@ -71,6 +71,8 @@ RUN mkdir -p /app/data /app/logs
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
ENV PATH="/app/.venv/bin:$PATH"
# 生产镜像默认关闭文件监控重载，避免 data/logs 写入触发重启循环。
ENV UVICORN_RELOAD=false
EXPOSE 8800
deploy.sh
+2 -1
View File
@@ -69,7 +69,8 @@ cmd_deploy() {
echo ""
echo "正在启动服务..."
$COMPOSE up -d
# 重新创建应用容器，确保生产环境不会沿用旧的 reload 配置。
$COMPOSE up -d --force-recreate
echo ""
echo "等待服务启动..."
docker-compose.yml
+2 -2
View File
@@ -42,8 +42,8 @@ services:
- YYB_WORKER_KEY=${YYB_WORKER_KEY:-}
# Roundcube 邮件验证码读取服务地址（不填则使用代码默认值）
- MAIL_ROUNDCUBE_URL=${MAIL_ROUNDCUBE_URL:-}
# Uvicorn reload（生产环境保持 false）
- UVICORN_RELOAD=${UVICORN_RELOAD:-false}
# 生产容器禁止 reload；开发环境由 dev.sh 单独启动 reload。
- UVICORN_RELOAD=false
networks:
- default
- order-site-net
services/yyb-worker/runtime/replay/live/default/body.txt
-1
View File
@@ -1 +0,0 @@
token_id=&transaction_id=&out_trade_no=&offer_type=&type=&pf=&pfkey=&from_h5=&pc_st=&r=&openid=&openkey=&session_id=&session_type=&sck=&anti_auto_script_token_id=&zoneid=&buy_quantity=&provide_uin=&wx_appid=&extend=&jump_url=&wcp=&biz_appid=&wx_direct_pay=&wx_publice_pay=&uuid=&pay_method=&pushtype=&wx_order_interface=&encrypt_msg=&base_key_version=&encrypt_way=&web_token=&webversion=&from_https=&t=&__refer=
services/yyb-worker/runtime/scripts/jsdom-pay.py
+65 -51
View File
@@ -28,25 +28,11 @@ 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"))
@@ -126,50 +112,75 @@ def extract_goods_state(html: str) -> tuple[list[int], str, str]:
return xmidas, token.group(1).upper(), anti.group(1).upper()
def load_template_args() -> tuple[str, list]:
def load_template_args() -> 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)
if runtime_args.exists():
return 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")
return load_json(path)
raise FileNotFoundError("缺少归档 goods 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:
def build_save_fields(order: dict[str, str], cookies: dict[str, str], web_token: str,
anti_token: str, encrypt_msg: str, zone_id: str, pf: str,
amount_fen: int) -> dict[str, str]:
"""按当前订单构造完整 web_save 表单，不能使用脱敏空模板。"""
if amount_fen <= 0:
raise ValueError("充值金额必须为正数")
login = midas_login_params(cookies)
values = {
return {
"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,
"type": "bg",
"pf": pf,
"pfkey": "pfkey",
"from_h5": "1",
"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,
"openid": cookies.get("openid", ""),
"openkey": cookies.get("accesstoken", ""),
"session_id": login["session_id"],
"session_type": login["session_type"],
"sck": hashlib.md5((APPID + cookies.get("accesstoken", "")).encode()).hexdigest().upper(),
"anti_auto_script_token_id": anti_token,
"zoneid": zone_id,
"buy_quantity": "1",
"provide_uin": "",
"wx_appid": login["wx_appid"],
"extend": "",
"jump_url": "https://pay.qq.com/midas/minipay_v2/views/public/mb.shtml",
"wcp": f"type=CNY&amt={amount_fen}",
"biz_appid": "wx951bdcac522929b6",
"wx_direct_pay": "1",
"wx_publice_pay": "1",
"uuid": str(uuid.uuid4()).upper(),
"pay_method": "wechat",
"pushtype": "NodeJS",
"wx_order_interface": "1",
"encrypt_msg": encrypt_msg,
"qq_appid": login["qq_appid"],
"base_key_version": "web_1.0.6",
"encrypt_way": "web_new_encrypt",
"web_token": web_token,
"webversion": "minipayv2",
"from_https": "1",
"t": str(int(time.time() * 1000)),
"__refer": "https://z.iwan.yyb.qq.com/",
}
body = template
for name, value in values.items():
body = replace_form_value(body, name, value)
return body
def build_save_body(order: dict[str, str], cookies: dict[str, str], web_token: str,
anti_token: str, encrypt_msg: str, zone_id: str, pf: str,
amount_fen: int) -> str:
"""编码完整支付表单，确保加密明文和实际请求使用同一上下文。"""
return urllib.parse.urlencode(build_save_fields(
order, cookies, web_token, anti_token, encrypt_msg, zone_id, pf, amount_fen,
))
def make_qr(sign: str, output: Path) -> None:
@@ -190,6 +201,8 @@ def node_environment() -> dict[str, str]:
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."""
from pyvm.order_status import completion_summary, order_ids
record = {
"checked_at": int(time.time()),
"baseline_order_states": baseline,
@@ -208,6 +221,7 @@ def main() -> int:
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("--amount-fen", type=int, required=True, help="所选点券的价格，单位分")
parser.add_argument("--dry-run", action="store_true", help="仅拉取页面并生成 DeviceFP，不上报或创建付款码")
parser.add_argument("--payment-timeout", type=float, default=300,
help="付款码生成后等待订单完成的最长秒数")
@@ -263,7 +277,8 @@ def main() -> int:
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()
web_args = load_template_args()
payment_pf = args.pf or "mds_myappjp-__mds_myappjp_PC_aW9zd2hpdGVsaX0-android"
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])
@@ -272,9 +287,9 @@ def main() -> int:
"token_id": order.get("token_id", ""),
"openid": cookies.get("openid", ""),
"openkey": cookies.get("accesstoken", ""),
"session_id": "hy_gameid",
"session_type": "wc_actoken",
"zoneid": "1",
"session_id": midas_login_params(cookies)["session_id"],
"session_type": midas_login_params(cookies)["session_type"],
"zoneid": args.zone_id,
"pay_method": "wechat",
"buy_quantity": "1",
"mb_pwd": "",
@@ -285,21 +300,18 @@ def main() -> int:
"provide_uin": "",
"extend": "",
"from_h5": "1",
"webversion": "web_1.0.6",
"webversion": "minipayv2",
}
# 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)
body = build_save_body(
order, cookies, web_token, anti_token, encrypt_msg,
args.zone_id, payment_pf, args.amount_fen,
)
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")
@@ -324,6 +336,8 @@ def main() -> int:
print("[jsdom-pay] 已跳过付款结果检查。")
return 0
from pyvm.order_status import find_completed_order, get_official_orders, order_completion_states
print("[jsdom-pay] 记录付款前订单列表，等待微信付款完成...")
baseline_document = get_official_orders(cookies)
baseline_states = order_completion_states(baseline_document)
services/yyb-worker/runtime/scripts/yyb-worker.py
+3 -1
View File
@@ -152,7 +152,8 @@ def _start_payment(job_id: str, selection: dict) -> None:
pay_command = ["python3", "scripts/jsdom-pay.py", "--session", str(session),
"--mall-response", str(response), "--out-dir", str(output),
"--zone-id", str(selection["zone_id"]),
"--pf", str(selection.get("order_pf", ""))]
"--pf", str(selection.get("order_pf", "")),
"--amount-fen", str(selection["price_fen"])]
_run_process(job_id, pay_command, "payment")
threading.Thread(target=run, daemon=True).start()
@@ -280,6 +281,7 @@ class Handler(BaseHTTPRequestHandler):
return self._json(400, {"detail": "角色不可充值，请重新选择"})
_jobs[job_id]["selection"] = {
"platform": body["platform"], "points": product["points"],
"price_fen": product["price_fen"],
"product_id": product["product_id"], "offer_id": product["offer_id"],
"zone_id": zone["zone_id"], "zone_name": zone["name"],
"role_id": role["role_id"], "role_name": role["name"],
services/yyb-worker/runtime/tests/test_jsdom_pay.py
+50
View File
@@ -0,0 +1,50 @@
"""YYB 付款表单的离线回归测试。"""
from __future__ import annotations
import importlib.util
import sys
import unittest
from pathlib import Path
from urllib.parse import parse_qs
_SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "jsdom-pay.py"
_SPEC = importlib.util.spec_from_file_location("yyb_jsdom_pay", _SCRIPT)
assert _SPEC and _SPEC.loader
_MODULE = importlib.util.module_from_spec(_SPEC)
sys.modules[_SPEC.name] = _MODULE
_SPEC.loader.exec_module(_MODULE)
class TestJsdomPay(unittest.TestCase):
def test_build_save_body_has_current_payment_context(self):
body = _MODULE.build_save_body(
{"token_id": "token", "transaction_id": "transaction", "out_trade_no": "trade"},
{"openid": "openid", "accesstoken": "access-token", "logintype": "WX"},
"web-token", "anti-token", "cipher", "2",
"mds_myappjp-__mds_myappjp_PC_aW9zd2hpdGVsaX0-iap", 600,
)
fields = parse_qs(body, keep_blank_values=True)
self.assertEqual(fields["type"], ["bg"])
self.assertEqual(fields["pf"], ["mds_myappjp-__mds_myappjp_PC_aW9zd2hpdGVsaX0-iap"])
self.assertEqual(fields["zoneid"], ["2"])
self.assertEqual(fields["pay_method"], ["wechat"])
self.assertEqual(fields["wcp"], ["type=CNY&amt=600"])
self.assertEqual(fields["encrypt_way"], ["web_new_encrypt"])
self.assertEqual(fields["webversion"], ["minipayv2"])
def test_qq_payment_uses_qq_oauth_session_fields(self):
body = _MODULE.build_save_body(
{"token_id": "token", "transaction_id": "transaction", "out_trade_no": "trade"},
{"openid": "openid", "accesstoken": "access-token", "logintype": "QC", "appid": "102033112"},
"web-token", "anti-token", "cipher", "1", "pf", 100,
)
fields = parse_qs(body, keep_blank_values=True)
self.assertEqual(fields["session_id"], ["openid"])
self.assertEqual(fields["session_type"], ["kp_accesstoken"])
self.assertEqual(fields["qq_appid"], ["102033112"])
self.assertEqual(fields["wx_appid"], [""])
if __name__ == "__main__":
unittest.main()