Astral/upsert.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
upsert.py —— 主脚本：扫码登录 → 上传成绩 → 登出
================================================================================

整条流程只有四步，而且**一行日志都不用自己写** ——
每个 api 函数被调用时会自动打印：

    === UserLogin 开始运行 ===
      [接口说明] 用户正式登录（开启本次登录会话）。
      [位置参数] 12345678, 'iUfgRmjr...'
      [关键字参数] region_id=1, place_id=1000
    === UserLogin 运行结束 ===

日志模式改 config.py 一行：

    LOG_LEVEL = "info"     只展示 api 脚本的运行情况和结果
    LOG_LEVEL = "debug"    额外展示向 api 发送的数据与 api 返回的数据

日志文件（两种同时写）：

    logs/latest.log        只含**当次**主脚本运行（下次覆盖）
    logs/YYYY-MM-DD.log    存放**当日**产生的所有日志（追加）

--------------------------------------------------------------------------------
用法
--------------------------------------------------------------------------------
    python upsert.py <二维码>                       # 预检：登录 + 抓快照 + 组包，不发包
    python upsert.py <二维码> --send                # 真实上传成绩（★ 有副作用，不可撤销）
    python upsert.py <二维码> --send --debug        # 同上，日志里带完整收发报文
    python upsert.py <二维码> --send --cooldown 0   # 跳过发包前的 60 秒冷却
    python upsert.py <二维码> --send --no-verify    # 跳过上传后的落库回查

    # ★ 额外塞节点（--extra-data，可重复，后给的覆盖先给的）
    python upsert.py <二维码> --send --extra-data extra.json
    python upsert.py <二维码> --send --extra-data '{"isNewMapList":"1"}'
    python upsert.py <二维码> --send --extra-data a.json --extra-data b.json

--------------------------------------------------------------------------------
★ UpsertUserAll 一个函数就把整条结算链路跑完了
--------------------------------------------------------------------------------
    1. 抓基线快照（没传 snapshot 时自动抓，用 token）
    2. 从快照推本局 5 只角色（游玩前等级）
    3. 构造本局成绩明细 userPlaylogList
    4. 调前置接口 GetUserNewItemListApi
    5. 组装 upsertUserAll 的 36 个节点（快照基线 + 本局演算 + 差分）
    6. 合并 extra_data 里的外部 JSON
    7. 发包前自检
    8. 冷却 -> 真实发包
    9. 落库回查（重新拉服务器数据，确认真的写进去了）

--------------------------------------------------------------------------------
★ --extra-data：额外塞节点
--------------------------------------------------------------------------------
    想手工覆盖 / 补上某几个节点（比如本局拿到的新 KaleidxScope 门）时用：

        # ① 从文件读（★ 推荐，不用跟 shell 引号搏斗）
        python upsert.py <二维码> --send --extra-data extra.json

        # ② 直接写 JSON 字符串
        python upsert.py <二维码> --send --extra-data '{"isNewMapList":"1"}'

        # ③ 多个来源，后给的覆盖先给的
        python upsert.py <二维码> --send --extra-data base.json --extra-data patch.json

    extra.json 长这样：

        {
          "userKaleidxScopeList": [
            {"gateId": 1, "isGateFound": true, "isKeyFound": true, "isClear": false,
             "totalRestLife": 0, "totalAchievement": 0, "totalDeluxscore": 0,
             "bestAchievement": 0, "bestDeluxscore": 0,
             "bestAchievementDate": "", "bestDeluxscoreDate": "",
             "playCount": 0, "clearDate": "", "lastPlayDate": "", "isInfoWatched": false}
          ],
          "isNewKaleidxScopeList": "1"
        }

    ★ 键必须是 upsertUserAll 的 36 个节点之一（含 11 条 isNew* 位串），
      写错会明确报错并列出可用节点名。
    ★ 没给对应位串时会按数组长度自动补 "0" 并提示 ——
      新增条目请显式传 "1"，"0" 表示存量更新。
    ★ JSON 在**扫码之前**就校验，写错了直接退出，不会浪费二维码。

  ── 在代码里直接调 api 的等价写法 ──

    UpsertUserAll(
        user_id, login.loginId,
        login_date_time=login.loginDateTime,
        auth_date_time=login.dateTime,
        token=token,
        extra_data={"userKaleidxScopeList": [...], "isNewKaleidxScopeList": "1"},
        confirm=True,
    )
"""

from __future__ import annotations

import sys

from api import (
    setup_logging,
    teardown_logging,
    log_info,
    log_warning,
    log_error,
    ParseQRCode,
    GetUserPreview,
    UserLogin,
    UserLogout,
    LogoutType,
    UpsertUserAll,
    ConfirmRequiredError,
    load_extra_data,
    PatchError,
    NODE_ORDER,
    NODE_TYPES,
)
from config import PLACE_ID, REGION_ID


def parse_args(argv):
    """极简命令行解析：第一个非 -- 开头的参数是二维码。"""
    opts = {"qr": "", "send": False, "debug": False, "no_verify": False,
            "cooldown": None, "extra_data": []}
    i = 0
    while i < len(argv):
        a = argv[i]
        if a == "--send":
            opts["send"] = True
        elif a == "--debug":
            opts["debug"] = True
        elif a == "--no-verify":
            opts["no_verify"] = True
        elif a == "--extra-data" and i + 1 < len(argv):
            # 可重复；值是 JSON 字符串或 JSON 文件路径
            opts["extra_data"].append(argv[i + 1])
            i += 1
        elif a == "--cooldown" and i + 1 < len(argv):
            opts["cooldown"] = float(argv[i + 1])
            i += 1
        elif not a.startswith("--") and not opts["qr"]:
            opts["qr"] = a
        i += 1
    return opts


def main() -> int:
    opts = parse_args(sys.argv[1:])
    if not opts["qr"]:
        print(__doc__)
        return 1

    # --debug 只影响本次运行，不改 config.py
    if opts["debug"]:
        import config as _cfg
        _cfg.LOG_LEVEL = "debug"

    # 日志会话：开头一句。之后每次 api 调用都会自动记录，主脚本不用再写日志
    setup_logging("upsert")

    # ---- ★ 先解析 --extra-data（放在扫码之前：JSON 写错就别浪费二维码了）----
    extra = None
    if opts["extra_data"]:
        merged = {}
        try:
            for item in opts["extra_data"]:
                part = load_extra_data(item)
                merged.update(part)
                log_info("--extra-data 读入 " + str(len(part)) + " 个节点：" + ", ".join(part))
        except PatchError as exc:
            log_error("--extra-data 解析失败：" + str(exc))
            return 6

        # 节点名也在这里查（不然要等到登录后组包时才报错，白跑一趟）
        unknown = [k for k in merged if k not in NODE_TYPES]
        if unknown:
            log_error("--extra-data 里有未知节点名：" + ", ".join(unknown))
            log_error("  可用节点（" + str(len(NODE_ORDER)) + " 个）：" + ", ".join(NODE_ORDER))
            return 6

        if not merged:
            log_warning("--extra-data 解析出来是空的，忽略")
        else:
            extra = merged
            log_info("--extra-data 合计 " + str(len(extra)) + " 个节点：" + ", ".join(extra))

    # 会话状态：保证异常时也能登出（否则账号会卡在 isLogin=True，下次登录 returnCode=100）
    session = {"user_id": 0, "token": "", "login_dt": 0, "logged_in": False}
    try:
        # ---- 扫码：二维码 -> userId + token ----
        user_id, token = ParseQRCode(opts["qr"])
        session.update(user_id=user_id, token=token)

        # ---- 预览：顺便挡住「账号已在游戏中」的情况，避免顶号 ----
        preview = GetUserPreview(user_id, token)
        if preview.isLogin:
            log_error("账号处于已登录状态，请先在机台登出或等待会话超时")
            return 2

        # ---- 登录：拿到 playlogId 与 loginDateTime（结算上报要用）----
        login = UserLogin(user_id, token, region_id=REGION_ID, place_id=PLACE_ID)
        session.update(login_dt=login.loginDateTime, logged_in=True)

        # ---- 结算：★ 一个函数搞定构造 + 发包 + 落库回查 ----
        UpsertUserAll(
            user_id, login.loginId,
            login_date_time=login.loginDateTime,
            auth_date_time=login.dateTime,
            token=token,
            extra_data=extra,
            dry_run=not opts["send"],
            confirm=opts["send"],
            cooldown_sec=opts["cooldown"],
            verify_landing=not opts["no_verify"],
            place_id=PLACE_ID, region_id=REGION_ID,
        )
        return 0

    except ConfirmRequiredError:
        return 5
    except Exception as exc:
        log_error(type(exc).__name__ + ": " + str(exc))
        return 1
    finally:
        # 兜底登出：登录过且还没登出就释放会话，否则账号卡在 isLogin=True
        if session["logged_in"]:
            try:
                UserLogout(session["user_id"], session["token"], session["login_dt"],
                           logout_type=LogoutType.TestIn,
                           region_id=REGION_ID, place_id=PLACE_ID)
            except Exception as exc:
                log_error("[收尾] 强制登出失败（会话可能需等待服务端超时）：" + str(exc))
        teardown_logging()


if __name__ == "__main__":
    sys.exit(main())