110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
import requests
|
|
import urllib.parse
|
|
import logging
|
|
import time
|
|
from datetime import datetime, timezone
|
|
|
|
BASE_URL = 'https://5-data.amae-koromo.com/api/v2/pl4'
|
|
OFFSET_2 = [1117113, 1358437]
|
|
XOR_CODE_2 = 86216345
|
|
REQUEST_HEADERS = {"Accept": "application/json"}
|
|
SESSION = requests.Session()
|
|
|
|
def acc2match(account_id: int) -> int:
|
|
"""
|
|
Convert an account_id into the match_id embedded at the end of a Mahjong Soul paipu URL.
|
|
"""
|
|
return ((7 * account_id + OFFSET_2[0]) ^ XOR_CODE_2) + OFFSET_2[1]
|
|
|
|
def search_player(nickname: str) -> int:
|
|
"""
|
|
Search for a player by nickname and return their account_id.
|
|
"""
|
|
url = f"{BASE_URL}/search_player/{urllib.parse.quote(nickname)}?limit=20&tag=all"
|
|
try:
|
|
res = SESSION.get(url, timeout=15, headers=REQUEST_HEADERS)
|
|
res.raise_for_status()
|
|
data = res.json()
|
|
except Exception as e:
|
|
raise RuntimeError(f"API request failed while searching for '{nickname}': {e}")
|
|
|
|
if not isinstance(data, list) or len(data) == 0:
|
|
raise ValueError(f"Player not found: '{nickname}'. The search returned an empty result.")
|
|
|
|
player = data[0]
|
|
if "id" not in player:
|
|
raise ValueError("Unexpected API response structure: missing 'id' field.")
|
|
|
|
logging.info(f"[API] Found player: '{player['nickname']}' (account_id={player['id']})")
|
|
return player["id"]
|
|
|
|
def get_player_nickname_by_id(account_id: int) -> str | None:
|
|
"""
|
|
Fetch a player's nickname using their account_id.
|
|
"""
|
|
end_ms = int(time.time() * 1000)
|
|
start_ms = 1262304000000
|
|
|
|
url = f"{BASE_URL}/player_stats/{account_id}/{start_ms}/{end_ms}?mode=16.12.9.15.11.8"
|
|
try:
|
|
res = SESSION.get(url, timeout=15, headers=REQUEST_HEADERS)
|
|
if res.status_code == 404:
|
|
return None
|
|
res.raise_for_status()
|
|
data = res.json()
|
|
return data.get("nickname")
|
|
except Exception as e:
|
|
logging.warning(f"[API] Failed to fetch nickname for account_id={account_id}: {e}")
|
|
return None
|
|
|
|
def get_player_records(account_id: int, limit: int, mode: int) -> list:
|
|
"""
|
|
Fetch a player's recent game records for the given mode.
|
|
"""
|
|
end_ms = int(time.time() * 1000)
|
|
start_ms = 1262304000000
|
|
|
|
url = f"{BASE_URL}/player_records/{account_id}/{end_ms}/{start_ms}?limit={limit}&mode={mode}&descending=true"
|
|
try:
|
|
res = SESSION.get(url, timeout=15, headers=REQUEST_HEADERS)
|
|
res.raise_for_status()
|
|
data = res.json()
|
|
except Exception as e:
|
|
raise RuntimeError(f"API request failed while fetching records (mode={mode}): {e}")
|
|
|
|
if not isinstance(data, list):
|
|
raise ValueError("Unexpected response format for player_records: not a list.")
|
|
|
|
logging.info(f"[API] Fetched {len(data)} records for mode={mode}")
|
|
return data
|
|
|
|
def format_timestamp(ts: int) -> str:
|
|
if not ts:
|
|
return ""
|
|
if ts > 1e11:
|
|
ts = ts / 1000.0
|
|
return datetime.fromtimestamp(ts, timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
def build_paipu_urls(records: list, account_id: int) -> list:
|
|
"""
|
|
Build a list of paipu URLs from game records.
|
|
"""
|
|
match_id = 'a' + str(acc2match(account_id))
|
|
results = []
|
|
for rec in records:
|
|
uuid_str = rec.get("uuid")
|
|
if not uuid_str:
|
|
continue
|
|
|
|
start_time = format_timestamp(rec.get("startTime", 0))
|
|
end_time = format_timestamp(rec.get("endTime", 0))
|
|
|
|
results.append({
|
|
"uuid": uuid_str,
|
|
"matchId": match_id,
|
|
"paipuUrl": f"https://game.maj-soul.com/1/?paipu={uuid_str}_{match_id}",
|
|
"startTime": start_time,
|
|
"endTime": end_time
|
|
})
|
|
return results
|