Txiaochen
25ac522391
- 新增 unlockmusic.py：解锁（解禁）歌曲 + 解禁 Master / Re:Master 难度，整个流程复用 upsert.py 的 run() - 买票前置闸门 api/ChargeGuard.py：必须先跑过 GetUserCharge，且同进程只允许买一次 - buy_ticket.py 票源改为 GetUserCharge，不再调用 GetGameCharge - UpsertUserAll 与买票的冷却统一为 15 秒 - api/ 文件名统一 PascalCase（文件名 = 函数名） - 项目改名 Astral，README 加 Logo、鸣谢与开发者说明 - 自检 284 项
151 lines
5.1 KiB
Python
151 lines
5.1 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
ChargeGuard.py —— 买票前置闸门（GetUserCharge 必须已经跑过）
|
||
================================================================================
|
||
|
||
★ 这个模块解决什么问题：
|
||
|
||
````
|
||
错误写法（危险）：
|
||
UpsertUserChargelog(user_id, 2, login.loginDateTime, confirm=True)
|
||
# ↑ 没先查 GetUserCharge —— 万一账号已经持有一张功能票，
|
||
# 服务端看到的就是「重复购买」，有封号风险。
|
||
```
|
||
|
||
正确写法：
|
||
```
|
||
held = GetUserCharge(user_id) # ① 先查：有没有持票
|
||
...
|
||
UpsertUserChargelog(user_id, 2, ...) # ② 再买
|
||
```
|
||
|
||
本模块在 ① 里记一笔「这个 user_id 刚查过票券」，在 ② 里检查这笔记录，
|
||
没查过就抛 `ChargelogGuardError`（**未发包**）。
|
||
|
||
★ 还顺手挡第二件事：**同一个账号在同一个进程里只允许买一次票**。
|
||
真机买票是「买完立刻被这一局用掉」（stock--），正常状态 stock 恒为 0；
|
||
脚本连着买两张是明显的异常行为，这里直接拦掉。
|
||
|
||
★ 记忆只存在于**当前进程**（内存字典），脚本退出即清空 —— 不做持久化，
|
||
免得下次运行时被上一次的记录误拦。
|
||
|
||
依据：
|
||
source/Assembly-CSharp/Manager/TicketManager.cs:130-174 UseTicketPrepare
|
||
source/Assembly-CSharp/Manager/TicketManager.cs:234-253 RequestUserCharge
|
||
source/Assembly-CSharp/Net.Packet.Mai2/PacketGetUserCharge.cs:20
|
||
source/Assembly-CSharp/Net.Packet.Mai2/PacketUploadUserChargelog.cs:19
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
#: GetUserCharge 结果的「新鲜度」窗口（秒）。
|
||
#: 超过这个时间才买票，认为那次查询已经过期，需要重查。
|
||
CHARGE_READ_FRESH_SEC = 600.0
|
||
|
||
#: user_id -> 最近一次 GetUserCharge 的 monotonic 时刻
|
||
_LAST_READ: Dict[int, float] = {}
|
||
|
||
#: user_id -> 最近一次 GetUserCharge 返回的 userChargeList（原样）
|
||
_LAST_LIST: Dict[int, List[Dict[str, Any]]] = {}
|
||
|
||
#: user_id -> 本进程内已经买过的 chargeId 列表
|
||
_PURCHASES: Dict[int, List[int]] = {}
|
||
|
||
|
||
# =============================================================================
|
||
# 一、GetUserCharge 侧：记一笔
|
||
# =============================================================================
|
||
|
||
def note_user_charge_read(
|
||
user_id: int,
|
||
charge_list: Optional[List[Dict[str, Any]]] = None,
|
||
) -> None:
|
||
"""
|
||
记录「刚刚为这个 user_id 查过票券」。
|
||
|
||
★ 由 GetUserCharge() 在拿到响应后自动调用，调用方不用管。
|
||
"""
|
||
try:
|
||
uid = int(user_id)
|
||
except (TypeError, ValueError):
|
||
return
|
||
_LAST_READ[uid] = time.monotonic()
|
||
if isinstance(charge_list, list):
|
||
_LAST_LIST[uid] = [c for c in charge_list if isinstance(c, dict)]
|
||
|
||
|
||
def last_read_age(user_id: int) -> Optional[float]:
|
||
"""距上次 GetUserCharge 过了多少秒；从没查过返回 None。"""
|
||
try:
|
||
uid = int(user_id)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
ts = _LAST_READ.get(uid)
|
||
if ts is None:
|
||
return None
|
||
return max(0.0, time.monotonic() - ts)
|
||
|
||
|
||
def has_fresh_read(
|
||
user_id: int,
|
||
max_age: float = CHARGE_READ_FRESH_SEC,
|
||
) -> bool:
|
||
"""这个 user_id 最近（max_age 秒内）查过 GetUserCharge 吗？"""
|
||
age = last_read_age(user_id)
|
||
if age is None:
|
||
return False
|
||
return age <= float(max_age)
|
||
|
||
|
||
def last_charge_list(user_id: int) -> List[Dict[str, Any]]:
|
||
"""最近一次 GetUserCharge 返回的 userChargeList（没有则空表）。"""
|
||
try:
|
||
return list(_LAST_LIST.get(int(user_id)) or [])
|
||
except (TypeError, ValueError):
|
||
return []
|
||
|
||
|
||
def held_charges(user_id: int) -> List[Dict[str, Any]]:
|
||
"""最近一次查到的「持有中」（stock > 0）的票券。"""
|
||
return [c for c in last_charge_list(user_id) if int(c.get("stock", 0) or 0) > 0]
|
||
|
||
|
||
# =============================================================================
|
||
# 二、UpsertUserChargelog 侧：买之前查
|
||
# =============================================================================
|
||
|
||
def note_charge_purchase(user_id: int, charge_id: int) -> None:
|
||
"""记录「这个 user_id 刚刚买了一张 chargeId 的票」。"""
|
||
try:
|
||
uid = int(user_id)
|
||
except (TypeError, ValueError):
|
||
return
|
||
_PURCHASES.setdefault(uid, []).append(int(charge_id))
|
||
|
||
|
||
def purchased_charge_ids(user_id: int) -> List[int]:
|
||
"""本进程内这个 user_id 已经买过的 chargeId。"""
|
||
try:
|
||
return list(_PURCHASES.get(int(user_id)) or [])
|
||
except (TypeError, ValueError):
|
||
return []
|
||
|
||
|
||
def reset(user_id: Optional[int] = None) -> None:
|
||
"""清空记忆（测试用；不传 user_id 就全清）。"""
|
||
if user_id is None:
|
||
_LAST_READ.clear()
|
||
_LAST_LIST.clear()
|
||
_PURCHASES.clear()
|
||
return
|
||
try:
|
||
uid = int(user_id)
|
||
except (TypeError, ValueError):
|
||
return
|
||
_LAST_READ.pop(uid, None)
|
||
_LAST_LIST.pop(uid, None)
|
||
_PURCHASES.pop(uid, None)
|