climperor/web/fetch_pro_matches.py

"""Fetch recent league/tournament (+ optional ranked) matches for OpenDota pros.

Default source: web/data/pro_player_watchlist.json (hand-curated star IDs).
Falls back to /proPlayers auto-pick only with --all-pros.

Per player: /players/{id}/matches per lobby_type (practice/tournament, and
ranked with --include-pubs), up to --limit newest each; then /matches/{id}
for final items + skill builds. League rows do not crowd out ranked pubs.

Output: web/data/pro_matches.json (Climperor web only; not used by recommend).

Usage:
  python web/fetch_pro_matches.py
  python web/fetch_pro_matches.py --include-pubs --limit 8
  python web/fetch_pro_matches.py --players 898754153,Ame
  python web/fetch_pro_matches.py --all-pros --limit-pros 20 --with-team
"""

from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import argparse
import json
import time
import urllib.error
from datetime import datetime, timedelta, timezone

from shared.grid import hero_table
from shared.http_utils import http_json, write_json_atomic
from shared.paths import DATA

from fetch_hero_matches import (
    collect_item_ids,
    extract_player_row,
    fetch_match,
    load_ability_id_map,
)
from fetch_hero_items import load_item_catalog
from fetch_pro_builds import fetch_pro_index

OPENDOTA = "https://api.opendota.com/api"
OUT = DATA / "pro_matches.json"
WATCHLIST = DATA / "pro_player_watchlist.json"
DEFAULT_LIMIT = 8
DEFAULT_LIMIT_PROS = 40
# OpenDota lobby_type: 1=practice, 2=tournament (pro/league biased).
LOBBY_LEAGUE = (1, 2)


def _log(msg: str) -> None:
    try:
        print(msg, flush=True)
    except UnicodeEncodeError:
        print(msg.encode("ascii", "backslashreplace").decode("ascii"), flush=True)


def load_watchlist(path: Path) -> list[dict]:
    """Load hand-curated star players from pro_player_watchlist.json."""
    if not path.is_file():
        return []
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as e:
        raise SystemExit(f"watchlist read failed: {path}: {e}") from e
    players = raw.get("players") if isinstance(raw, dict) else None
    if not isinstance(players, list):
        raise SystemExit(f"watchlist missing players[]: {path}")
    out: list[dict] = []
    seen: set[int] = set()
    for row in players:
        if not isinstance(row, dict):
            continue
        try:
            aid = int(row.get("account_id") or 0)
        except (TypeError, ValueError):
            continue
        if aid <= 0 or aid in seen:
            continue
        seen.add(aid)
        name = row.get("name")
        if isinstance(name, str):
            name = name.strip() or None
        else:
            name = None
        team_tag = row.get("team_tag")
        if isinstance(team_tag, str):
            team_tag = team_tag.strip() or None
        else:
            team_tag = None
        team_name = row.get("team_name")
        if isinstance(team_name, str):
            team_name = team_name.strip() or None
        else:
            team_name = None
        region = row.get("region")
        if isinstance(region, str):
            region = region.strip().lower() or None
        else:
            region = None
        out.append(
            {
                "account_id": aid,
                "name": name,
                "team_tag": team_tag,
                "team_name": team_name,
                "region": region,
            }
        )
    return out


def parse_pro_filter(raw: str, pro_index: dict[int, dict]) -> list[int]:
    """Comma-separated account ids or registered pro names."""
    if not raw.strip():
        return []
    name_to_id: dict[str, int] = {}
    for aid, prof in pro_index.items():
        val = prof.get("name")
        if isinstance(val, str) and val.strip():
            name_to_id[val.strip().lower()] = aid
    out: list[int] = []
    for part in raw.split(","):
        token = part.strip()
        if not token:
            continue
        if token.isdigit():
            out.append(int(token))
            continue
        aid = name_to_id.get(token.lower())
        if aid:
            out.append(aid)
        else:
            _log(f"  warn: unknown pro {token!r}")
    return out


def merge_prof(aid: int, base: dict | None, override: dict | None = None) -> dict:
    """Prefer OpenDota live fields, keep watchlist name/team as fallback."""
    out: dict = {
        "account_id": aid,
        "name": None,
        "team_tag": None,
        "team_name": None,
        "country_code": None,
        "last_match_time": None,
    }
    if isinstance(base, dict):
        for key in out:
            if key == "account_id":
                continue
            val = base.get(key)
            if val is not None and val != "":
                out[key] = val
    if isinstance(override, dict):
        for key in ("name", "team_tag", "team_name"):
            if out.get(key):
                continue
            val = override.get(key)
            if isinstance(val, str) and val.strip():
                out[key] = val.strip()
            elif val:
                out[key] = val
    return out


def filter_pros(
    pro_index: dict[int, dict],
    *,
    with_team: bool,
    active_days: int | None,
    limit_pros: int,
    player_ids: list[int],
    watchlist: list[dict] | None = None,
) -> list[tuple[int, dict]]:
    if player_ids:
        rows: list[tuple[int, dict]] = []
        for aid in player_ids:
            rows.append((aid, merge_prof(aid, pro_index.get(aid))))
        return rows

    if watchlist:
        rows = []
        for entry in watchlist:
            aid = int(entry["account_id"])
            rows.append((aid, merge_prof(aid, pro_index.get(aid), entry)))
        return rows

    cutoff = None
    if active_days is not None and active_days > 0:
        cutoff = datetime.now(timezone.utc) - timedelta(days=active_days)

    candidates: list[tuple[int, dict, float]] = []
    for aid, prof in pro_index.items():
        if with_team and not prof.get("team_tag") and not prof.get("team_name"):
            continue
        last = prof.get("last_match_time")
        score = 0.0
        if isinstance(last, str) and last.strip():
            try:
                ts = datetime.fromisoformat(last.replace("Z", "+00:00"))
                if cutoff and ts < cutoff:
                    continue
                score = ts.timestamp()
            except ValueError:
                if cutoff:
                    continue
        candidates.append((aid, prof, score))

    candidates.sort(key=lambda t: (-t[2], str(t[1].get("name") or ""), t[0]))
    picked = candidates[: max(1, limit_pros)]
    return [(aid, merge_prof(aid, prof)) for aid, prof, _ in picked]


def player_match_metas(
    account_id: int,
    limit: int,
    *,
    lobby_types: tuple[int, ...] = LOBBY_LEAGUE,
) -> list[dict]:
    """Recent match list rows for a pro (deduped, newest first).

    Fetches up to ``limit`` newest matches **per lobby_type**, then merges.
    With league + ranked pubs this can return up to ``limit * len(lobby_types)``
    rows — important so recent league games do not crowd out ranked pubs.
    """
    by_id: dict[int, dict] = {}
    for lt in lobby_types:
        url = f"{OPENDOTA}/players/{account_id}/matches?limit={limit}&lobby_type={lt}"
        try:
            raw = http_json(url)
        except (
            urllib.error.HTTPError,
            urllib.error.URLError,
            TimeoutError,
            json.JSONDecodeError,
            OSError,
        ):
            continue
        if not isinstance(raw, list):
            continue
        for row in raw:
            if not isinstance(row, dict):
                continue
            try:
                mid = int(row.get("match_id") or 0)
            except (TypeError, ValueError):
                continue
            if mid <= 0:
                continue
            prev = by_id.get(mid)
            if prev is None:
                by_id[mid] = row
                continue
            try:
                st_new = int(row.get("start_time") or 0)
                st_old = int(prev.get("start_time") or 0)
            except (TypeError, ValueError):
                st_new = st_old = 0
            if st_new >= st_old:
                by_id[mid] = row

    return sorted(
        by_id.values(),
        key=lambda r: (-int(r.get("start_time") or 0), -int(r.get("match_id") or 0)),
    )


def fetch_player_matches(
    account_id: int,
    *,
    limit: int,
    id_map: dict[int, str],
    catalog: dict[int, dict],
    id_to_key: dict[int, str],
    delay: float,
    lobby_types: tuple[int, ...],
) -> list[dict]:
    metas = player_match_metas(account_id, limit, lobby_types=lobby_types)
    rows: list[dict] = []
    for meta in metas:
        try:
            mid = int(meta.get("match_id") or 0)
            hid = int(meta.get("hero_id") or 0)
        except (TypeError, ValueError):
            continue
        if mid <= 0 or hid <= 0:
            continue
        lt_raw = meta.get("lobby_type")
        try:
            lt_i = int(lt_raw) if lt_raw is not None else None
        except (TypeError, ValueError):
            lt_i = None
        row_origin = "public" if lt_i == 7 else "pro"
        detail = fetch_match(mid)
        if delay > 0:
            time.sleep(delay)
        if not detail:
            continue
        slim = extract_player_row(
            detail,
            hid,
            origin=row_origin,
            id_map=id_map,
            item_catalog=catalog,
            list_meta=meta,
        )
        if not slim:
            continue
        row_aid = slim.get("account_id")
        try:
            row_aid_i = int(row_aid) if row_aid is not None else 0
        except (TypeError, ValueError):
            row_aid_i = 0
        if row_aid_i and row_aid_i != account_id:
            continue
        slim["hero_id"] = hid
        slim["hero_key"] = id_to_key.get(hid)
        slim["lobby_type"] = lt_i
        rows.append(slim)
    return rows


def build_indexes(
    by_pro: dict[str, dict],
    id_to_key: dict[int, str],
) -> dict[str, dict]:
    by_hero: dict[str, list[dict]] = {}
    for cell in by_pro.values():
        for row in cell.get("matches") or []:
            if not isinstance(row, dict):
                continue
            key = row.get("hero_key")
            if not key:
                hid = row.get("hero_id")
                try:
                    key = id_to_key.get(int(hid)) if hid is not None else None
                except (TypeError, ValueError):
                    key = None
            if not key:
                continue
            by_hero.setdefault(str(key), []).append(row)
    out: dict[str, dict] = {}
    for key, rows in by_hero.items():
        seen: set[int] = set()
        deduped: list[dict] = []
        for row in sorted(
            rows,
            key=lambda r: (-int(r.get("start_time") or 0), -int(r.get("match_id") or 0)),
        ):
            try:
                mid = int(row.get("match_id") or 0)
            except (TypeError, ValueError):
                continue
            if mid in seen:
                continue
            seen.add(mid)
            deduped.append(row)
        out[key] = {"matches": deduped}
    return out


def write_out(
    path: Path,
    *,
    by_pro: dict[str, dict],
    by_hero: dict[str, dict],
    pros_meta: dict[str, dict],
    item_catalog: dict[int, dict],
    limit: int,
    limit_pros: int,
    lobby_types: tuple[int, ...],
    player_source: str,
) -> None:
    used = collect_item_ids(by_hero)
    items_out = {
        str(iid): {
            "key": item_catalog[iid]["key"],
            "dname": item_catalog[iid]["dname"],
            "name_loc": item_catalog[iid].get("name_loc") or item_catalog[iid]["dname"],
        }
        for iid in sorted(used)
        if iid in item_catalog
    }
    match_count = sum(
        len(cell.get("matches") or [])
        for cell in by_pro.values()
        if isinstance(cell, dict)
    )
    payload = {
        "meta": {
            "source": "opendota",
            "attribution": "https://www.opendota.com",
            "fetched_at": datetime.now(timezone.utc).isoformat(),
            "player_source": player_source,
            "limit_per_lobby": limit,
            "limit_pros": limit_pros,
            "lobby_types": list(lobby_types),
            "pro_count": len(by_pro),
            "match_count": match_count,
            "hero_count": len(by_hero),
            "note_zh": (
                "OpenDota 明星选手近期联赛/锦标赛对局（可选含天梯 lobby_type=7）；"
                "默认名单见 web/data/pro_player_watchlist.json；"
                "每种 lobby 各保留最近 limit 场，避免联赛挤掉天梯；"
                "lobby_type 1=训练/practice、2=tournament、7=ranked；"
                "含终局出装、加点与联赛名（若有）。"
            ),
        },
        "items": items_out,
        "pros": pros_meta,
        "by_pro": by_pro,
        "by_hero": by_hero,
    }
    write_json_atomic(path, payload)


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--out", type=Path, default=OUT)
    ap.add_argument(
        "--watchlist",
        type=Path,
        default=WATCHLIST,
        help=f"Star player list JSON (default: {WATCHLIST})",
    )
    ap.add_argument(
        "--limit",
        type=int,
        default=DEFAULT_LIMIT,
        help=f"Matches per pro (default: {DEFAULT_LIMIT})",
    )
    ap.add_argument(
        "--limit-pros",
        type=int,
        default=DEFAULT_LIMIT_PROS,
        help=f"Max pros when --all-pros (default: {DEFAULT_LIMIT_PROS})",
    )
    ap.add_argument(
        "--players",
        default="",
        help="Comma-separated account_id or registered pro name (overrides watchlist)",
    )
    ap.add_argument(
        "--all-pros",
        action="store_true",
        help="Ignore watchlist; auto-pick from /proPlayers (see --limit-pros/--with-team)",
    )
    ap.add_argument(
        "--with-team",
        action="store_true",
        help="With --all-pros: only pros with a team_tag/team_name",
    )
    ap.add_argument(
        "--active-days",
        type=int,
        default=60,
        help="With --all-pros: skip pros idle > N days (0=disable; default: 60)",
    )
    ap.add_argument("--delay", type=float, default=0.35)
    ap.add_argument(
        "--include-pubs",
        action="store_true",
        help="Also include ranked pub lobby_type=7 (high-MMR scrims)",
    )
    args = ap.parse_args()

    limit = max(1, int(args.limit))
    limit_pros = max(1, int(args.limit_pros))
    active_days = int(args.active_days) if args.active_days > 0 else None
    lobby_types: tuple[int, ...] = LOBBY_LEAGUE
    if args.include_pubs:
        lobby_types = LOBBY_LEAGUE + (7,)

    _log("loading pro players ...")
    pro_index = fetch_pro_index()
    _log(f"  {len(pro_index)} registered pros")

    player_ids = parse_pro_filter(args.players, pro_index)
    watchlist: list[dict] | None = None
    player_source = "cli"

    if player_ids:
        player_source = "cli"
    elif not args.all_pros:
        watchlist = load_watchlist(args.watchlist)
        if not watchlist:
            raise SystemExit(
                f"Empty/missing watchlist {args.watchlist}. "
                "Add players or pass --players / --all-pros."
            )
        player_source = f"watchlist:{args.watchlist.name}"
        _log(f"  watchlist {len(watchlist)} players ← {args.watchlist}")
    else:
        player_source = "proPlayers"

    picked = filter_pros(
        pro_index,
        with_team=args.with_team,
        active_days=active_days,
        limit_pros=limit_pros,
        player_ids=player_ids,
        watchlist=watchlist,
    )
    if not picked:
        raise SystemExit("No pros matched filters")

    heroes = hero_table()
    id_to_key = {int(h["id"]): h["key"] for h in heroes}
    catalog, _ = load_item_catalog()
    id_map = load_ability_id_map()

    by_pro: dict[str, dict] = {}
    pros_meta: dict[str, dict] = {}
    total_matches = 0

    for i, (aid, prof) in enumerate(picked, 1):
        label = prof.get("name") or prof.get("team_tag") or str(aid)
        _log(f"[{i}/{len(picked)}] {label} ({aid}) ...")
        matches = fetch_player_matches(
            aid,
            limit=limit,
            id_map=id_map,
            catalog=catalog,
            id_to_key=id_to_key,
            delay=args.delay,
            lobby_types=lobby_types,
        )
        sid = str(aid)
        by_pro[sid] = {
            "account_id": aid,
            "name": prof.get("name"),
            "team_tag": prof.get("team_tag"),
            "team_name": prof.get("team_name"),
            "country_code": prof.get("country_code"),
            "match_count": len(matches),
            "matches": matches,
        }
        pros_meta[sid] = {
            "account_id": aid,
            "name": prof.get("name"),
            "team_tag": prof.get("team_tag"),
            "team_name": prof.get("team_name"),
            "country_code": prof.get("country_code"),
        }
        total_matches += len(matches)
        _log(f"  {len(matches)} matches")

    by_hero = build_indexes(by_pro, id_to_key)
    write_out(
        args.out,
        by_pro=by_pro,
        by_hero=by_hero,
        pros_meta=pros_meta,
        item_catalog=catalog,
        limit=limit,
        limit_pros=len(picked),
        lobby_types=lobby_types,
        player_source=player_source,
    )
    _log(
        f"done pros={len(by_pro)} matches={total_matches} heroes={len(by_hero)} → {args.out}"
    )


if __name__ == "__main__":
    main()