mimic_multi_rank_cli.py

gistfile1.txt

#!/usr/bin/env python3
"""
多赛道排行榜 CLI（基于 rich 优雅显示），支持查看每个题目的解出情况。

赛道接口：
- 云网互联赛道: /api/ct/web/bwm_race/master/rank/cloudnetwork_detail/
- CTF 赛道:     /api/ct/web/bwm_race/master/rank/jeo/
- 车联网/低空/物联等赛道: /api/ct/web/bwm_race/master/rank/iov_detail/
- 人工智能赛道: /api/ct/web/bwm_race/master/rank/ai_detail/
- 韧性安全赛道: /api/ct/web/bwm_race/master/rank/resilience_detail/

特点：
- 使用 mimic_cli 的真实登录（DES-ECB 加密 + /api/mp/auth/login）
- 通过 SOCKS5 代理持续请求多赛道接口
- rich 表格展示：每条赛道一个表格，每题一列（显示得分或是否 solve）
- 以 JSON 行方式按赛道写入日志文件
"""

"""
./venv/bin/python mimic_multi_rank_cli.py \
  --login-user 123 \
  --login-pass 456 \
  --tracks cloud,ctf,iov,ai,resilience \
  --top 0 \
  --interval 15
"""

import argparse
import datetime as dt
import json
import os
import sys
import time
from typing import Any, Dict, List, Optional, Tuple

import requests
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich import box

from mimic_cli import login_via_proxy


CT_HOST_DEFAULT = ""
PROXY_DEFAULT = ""
PROXY_USER_DEFAULT = ""
PROXY_PASS_DEFAULT = ""


TRACK_CONFIG = {
    "cloud": {
        "label": "云网互联赛道",
        "path": "/api/ct/web/bwm_race/master/rank/cloudnetwork_detail/",
        "type": "detail",  # questions + rank_data + question_detail
    },
    "ctf": {
        "label": "CTF 赛道",
        "path": "/api/ct/web/bwm_race/master/rank/jeo/",
        "type": "jeo",  # device_types + rank_detail + device_type_score
    },
    "iov": {
        "label": "车联网/低空/物联赛道",
        "path": "/api/ct/web/bwm_race/master/rank/iov_detail/",
        "type": "detail",
    },
    "ai": {
        "label": "人工智能赛道",
        "path": "/api/ct/web/bwm_race/master/rank/ai_detail/",
        "type": "detail",
    },
    "resilience": {
        "label": "韧性安全赛道",
        "path": "/api/ct/web/bwm_race/master/rank/resilience_detail/",
        "type": "detail",
    },
}


def build_session(
    *,
    host: str,
    proxy: str,
    proxy_user: str,
    proxy_pass: str,
    login_user: str,
    login_pass: str,
) -> Tuple[requests.Session, str]:
    """
    登录并返回 (Session, jwt_token)。
    """

    login_res = login_via_proxy(
        host=host,
        proxy=proxy,
        proxy_user=proxy_user,
        proxy_pass=proxy_pass,
        login_user=login_user,
        login_pass=login_pass,
    )

    if not login_res.token:
        raise SystemExit(
            f"登录成功但没有拿到 token，原始响应：{json.dumps(login_res.raw, ensure_ascii=False)}"
        )

    proxy_url = f"socks5h://{proxy_user}:{proxy_pass}@{proxy}"
    sess = requests.Session()
    sess.proxies.update({"http": proxy_url, "https": proxy_url})
    sess.headers.update({"Authorization": f"JWT {login_res.token}"})
    return sess, login_res.token


def fetch_track_data(sess: requests.Session, host: str, track: str) -> Dict[str, Any]:
    """
    请求指定赛道接口，返回 data 段。
    """

    cfg = TRACK_CONFIG[track]
    url = f"http://{host}{cfg['path']}"
    resp = sess.get(url, timeout=15)
    resp.raise_for_status()
    payload = resp.json()
    if payload.get("code") != "AD-000000":
        raise RuntimeError(
            f"{track} 接口返回异常: {json.dumps(payload, ensure_ascii=False)}"
        )
    return payload["data"]


def append_log(log_dir: str, track: str, data: Dict[str, Any]) -> None:
    """
    以 JSON 行写入对应赛道日志。
    """

    os.makedirs(log_dir, exist_ok=True)
    path = os.path.join(log_dir, f"mimic_rank_{track}.log")
    rec = {"ts": dt.datetime.utcnow().isoformat() + "Z", "data": data}
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")


def build_detail_table(
    data: Dict[str, Any],
    track_label: str,
    top_n: Optional[int],
) -> Table:
    """
    通用 detail 型接口（cloud/iov/ai/resilience）渲染。
    data: {questions: [...], rank_data: [...]}
    """

    questions: List[Dict[str, Any]] = data.get("questions") or []
    rank_data: List[Dict[str, Any]] = data.get("rank_data") or []

    # 按 rank 排序
    rank_data = sorted(rank_data, key=lambda x: x.get("rank", 10**9))
    if top_n is not None:
        rank_data = rank_data[:top_n]

    table = Table(
        title=track_label,
        box=box.MINIMAL_DOUBLE_HEAD,
        show_lines=False,
        highlight=True,
    )
    table.add_column("Rank", justify="right", style="bold cyan", no_wrap=True)
    table.add_column("Team", style="bold", overflow="fold")
    table.add_column("Score", justify="right", style="bold magenta", no_wrap=True)
    table.add_column("Latest", style="dim", no_wrap=True)

    # 每个题目一列：使用 cn_name 或 name
    for q in questions:
        name = q.get("cn_name") or q.get("name") or "Q"
        table.add_column(name, justify="center", no_wrap=True)

    for item in rank_data:
        team_name = str(item.get("name", ""))
        rank = str(item.get("rank", ""))
        score = str(item.get("score", ""))
        latest = str(item.get("latest", ""))

        detail = item.get("question_detail") or []
        detail_by_res = {d.get("resource_id"): d for d in detail}

        cells = []
        for q in questions:
            qid = q.get("resource_id")
            d = detail_by_res.get(qid) or {}
            s = d.get("score", 0)
            if s and isinstance(s, (int, float)) and s > 0:
                cells.append(f"[green]{s}[/]")
            else:
                cells.append("[dim]-[/]")

        table.add_row(rank, team_name, score, latest, *cells)

    if not rank_data:
        table.caption = "[dim]暂无数据[/]"

    return table


def build_ctf_table(
    data: Dict[str, Any],
    track_label: str,
    top_n: Optional[int],
) -> Table:
    """
    CTF jeo 接口渲染。
    data: {device_types: [...], rank_detail: [...]}
    """

    device_types: List[Dict[str, Any]] = data.get("device_types") or []
    rank_detail: List[Dict[str, Any]] = data.get("rank_detail") or []

    rank_detail = sorted(rank_detail, key=lambda x: x.get("rank", 10**9))
    if top_n is not None:
        rank_detail = rank_detail[:top_n]

    table = Table(
        title=track_label,
        box=box.MINIMAL_DOUBLE_HEAD,
        show_lines=False,
        highlight=True,
    )
    table.add_column("Rank", justify="right", style="bold cyan", no_wrap=True)
    table.add_column("Team", style="bold", overflow="fold")
    table.add_column("Score", justify="right", style="bold magenta", no_wrap=True)
    table.add_column("Latest", style="dim", no_wrap=True)

    # 每个 device_type 一列（题目）
    for dt in device_types:
        name = dt.get("name") or dt.get("direction_name") or "Q"
        table.add_column(name, justify="center", no_wrap=True)

    for item in rank_detail:
        team_name = str(item.get("name", ""))
        rank = str(item.get("rank", ""))
        score = str(item.get("score", ""))
        latest = str(item.get("latest", ""))

        dts = item.get("device_type_score") or []
        cells = []
        for idx, dt in enumerate(device_types):
            # device_type_score 按顺序对应 device_types
            cell = "-"
            if idx < len(dts):
                info = dts[idx] or {}
                solved = info.get("is_solved")
                r = info.get("rank", 0)
                if solved:
                    # 已解：显示 ✔，附加 rank（如有）
                    if r:
                        cell = f"[green]✔({r})[/]"
                    else:
                        cell = "[green]✔[/]"
                else:
                    cell = "[dim]·[/]"
            cells.append(cell)

        table.add_row(rank, team_name, score, latest, *cells)

    if not rank_detail:
        table.caption = "[dim]暂无数据[/]"

    return table


def main(argv: Optional[list[str]] = None) -> int:
    parser = argparse.ArgumentParser(
        description="多赛道 mimic 排行榜（cloud/ctf/iov/ai/resilience），rich 美化展示 + 写日志。"
    )

    parser.add_argument("--ct-host", default=CT_HOST_DEFAULT, help="CT 平台主机 (默认: 10.60.10.10)")
    parser.add_argument("--proxy", default=PROXY_DEFAULT, help="SOCKS5 代理 host:port")
    parser.add_argument("--proxy-user", default=PROXY_USER_DEFAULT, help="SOCKS5 用户名")
    parser.add_argument("--proxy-pass", default=PROXY_PASS_DEFAULT, help="SOCKS5 密码")

    parser.add_argument("--login-user", required=True, help="登录用户名（比赛账号）")
    parser.add_argument("--login-pass", required=True, help="登录密码")

    parser.add_argument(
        "--tracks",
        default="cloud,ctf,iov,ai,resilience",
        help="逗号分隔的赛道列表，例如: cloud,ctf,iov",
    )
    parser.add_argument(
        "--interval",
        type=int,
        default=15,
        help="刷新间隔秒数（默认 15）",
    )
    parser.add_argument(
        "--top",
        type=int,
        default=10,
        help="每个赛道只展示前 N 名（默认 10，<=0 表示全部）",
    )
    parser.add_argument(
        "--log-dir",
        default="logs",
        help="日志目录（每个赛道一个 JSON 行日志文件，默认 logs/）",
    )
    parser.add_argument(
        "--once",
        action="store_true",
        help="只抓取一次并展示，然后退出（默认持续轮询）。",
    )

    args = parser.parse_args(argv)

    track_names = [t.strip() for t in args.tracks.split(",") if t.strip()]
    for t in track_names:
        if t not in TRACK_CONFIG:
            raise SystemExit(f"未知赛道: {t!r}，可选: {', '.join(TRACK_CONFIG.keys())}")

    sess, token = build_session(
        host=args.ct_host,
        proxy=args.proxy,
        proxy_user=args.proxy_user,
        proxy_pass=args.proxy_pass,
        login_user=args.login_user,
        login_pass=args.login_pass,
    )

    console = Console()
    top_n = args.top if args.top > 0 else None

    while True:
        console.clear()
        now = dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        header = f"第八届强网拟态国际精英挑战赛 排行榜快照 @ {now}"

        console.rule(f"[bold blue]{header}[/bold blue]")
        console.print(
            f"JWT 前缀: [dim]{token[:32]}...[/dim]    赛道: {', '.join(track_names)}",
            style="dim",
        )
        console.print()

        for t in track_names:
            cfg = TRACK_CONFIG[t]
            label = cfg["label"]
            try:
                data = fetch_track_data(sess, args.ct_host, t)
                append_log(args.log_dir, t, data)

                if cfg["type"] == "detail":
                    table = build_detail_table(data, label, top_n)
                else:  # jeo / ctf
                    table = build_ctf_table(data, label, top_n)

                console.print(table)
                console.print()
            except Exception as exc:
                console.print(
                    Panel.fit(
                        f"{label} 抓取失败: {exc}",
                        title=f"{label}",
                        border_style="red",
                    )
                )
                console.print()

        console.rule("[dim]按 Ctrl+C 退出[/dim]")
        console.print()

        if args.once:
            break

        try:
            time.sleep(max(1, args.interval))
        except KeyboardInterrupt:
            console.print("\n收到中断信号，退出。", style="bold red")
            break

    return 0


if __name__ == "__main__":
    raise SystemExit(main())