diff --git a/shared/audit_relations.py b/shared/audit_relations.py index 3ae9430..d17d29b 100644 --- a/shared/audit_relations.py +++ b/shared/audit_relations.py @@ -1,13 +1,14 @@ """Audit qualitative relations.json against OpenDota + STRATZ. Sources: - - data/matchups.json OpenDota vs winrates - - data/stratz_matchups.json STRATZ vs advantage (fetch_stratz.py --mode matchups) - - data/synergies.json STRATZ teammate synergies + - shared/data/matchups.json OpenDota vs winrates + - shared/data/stratz_matchups.json STRATZ vs advantage (fetch_stratz.py --mode matchups) + - shared/data/synergies.json STRATZ teammate synergies + - web/data/stratz_matchup_tops.json Web 对位 Top (optional cross-check section) Usage: - python audit_relations.py - python audit_relations.py --min-games 80 --out data/relations_audit.json + python shared/audit_relations.py + python shared/audit_relations.py --min-games 80 --out shared/data/relations_audit.json """ from __future__ import annotations @@ -22,12 +23,23 @@ import json from datetime import datetime, timezone from shared.grid import hero_table -from shared.paths import SHARED_DATA +from shared.matchup_cross import audit_matchup_tops +from shared.paths import DATA, ROOT, SHARED_DATA from shared.relations import DEFAULT_RELATIONS, load_relations MATCHUPS = SHARED_DATA / "matchups.json" STRATZ_MATCHUPS = SHARED_DATA / "stratz_matchups.json" SYNERGIES = SHARED_DATA / "synergies.json" +STRATZ_MATCHUP_TOPS = DATA / "stratz_matchup_tops.json" +# Legacy monorepo-move leftovers (repo-root data/). +_LEGACY_DATA = ROOT / "data" + + +def _resolve_cache(*candidates: Path) -> Path: + for p in candidates: + if p.is_file(): + return p + return candidates[0] def _load_json_safe(path: Path) -> dict: @@ -109,12 +121,18 @@ def main() -> None: ap.add_argument("--out", type=Path, default=SHARED_DATA / "relations_audit.json") args = ap.parse_args() + matchups_path = _resolve_cache(MATCHUPS, _LEGACY_DATA / "matchups.json") + stratz_path = _resolve_cache(STRATZ_MATCHUPS, _LEGACY_DATA / "stratz_matchups.json") + syn_path = _resolve_cache(SYNERGIES, _LEGACY_DATA / "synergies.json") + tops_path = _resolve_cache(STRATZ_MATCHUP_TOPS, _LEGACY_DATA / "stratz_matchup_tops.json") + id_to_key, key_to_id, names = load_id_maps() all_keys = sorted(key_to_id.keys(), key=lambda k: key_to_id[k]) rel = load_relations() - odota = _load_json_safe(MATCHUPS) - stratz = _load_json_safe(STRATZ_MATCHUPS) - syn_file = _load_json_safe(SYNERGIES) + odota = _load_json_safe(matchups_path) + stratz = _load_json_safe(stratz_path) + syn_file = _load_json_safe(syn_path) + tops = _load_json_safe(tops_path) by_o = odota.get("by_hero") or {} by_s = stratz.get("by_hero") or {} by_syn = syn_file.get("by_hero") or {} @@ -123,7 +141,6 @@ def main() -> None: counters = [(e["a"], e["b"], e.get("reason") or "") for e in rel.get("counters") or []] syns = [(e["a"], e["b"], e.get("reason") or "") for e in rel.get("synergies") or []] known_counters = {(a, b) for a, b, _ in counters} - known_syn = {_pair_key(a, b) for a, b, _ in syns} counter_ok, counter_conflict, counter_weak, counter_no_data = [], [], [], [] for a, b, reason in counters: @@ -154,7 +171,7 @@ def main() -> None: s_bad = s_adv is not None and s_games >= args.min_games and s_adv <= args.stratz_disagree if o_adv is None and s_adv is None: counter_no_data.append({**row, "why": "missing_both"}) - elif (o_games < args.min_games and (s_games < args.min_games or s_adv is None)): + elif o_games < args.min_games and (s_games < args.min_games or s_adv is None): counter_no_data.append({**row, "why": "low_games"}) elif o_ok or s_ok: sources = [] @@ -182,13 +199,26 @@ def main() -> None: continue cell = (by_syn.get(str(aid)) or {}).get(str(bid)) or (by_syn.get(str(bid)) or {}).get(str(aid)) if not cell: - syn_missing.append({"a": a, "b": b, "a_loc": names.get(a), "b_loc": names.get(b), "why": "no_stratz"}) + syn_missing.append( + { + "a": a, + "b": b, + "a_loc": names.get(a), + "b_loc": names.get(b), + "why": "no_stratz", + } + ) continue score = float(cell.get("synergy") or 0) games = int(cell.get("games") or 0) row = { - "a": a, "a_loc": names.get(a, a), "b": b, "b_loc": names.get(b, b), - "reason": reason, "stratz_synergy": round(score, 3), "stratz_games": games, + "a": a, + "a_loc": names.get(a, a), + "b": b, + "b_loc": names.get(b, b), + "reason": reason, + "stratz_synergy": round(score, 3), + "stratz_games": games, } if games < args.min_games: syn_missing.append({**row, "why": "low_games"}) @@ -199,7 +229,6 @@ def main() -> None: else: syn_missing.append({**row, "why": "weak_synergy"}) - # Cross-source suggestions: both OpenDota and STRATZ agree A counters B, missing from relations suggestions: list[dict] = [] for akey in all_keys: aid = key_to_id[akey] @@ -210,19 +239,28 @@ def main() -> None: o_adv, o_games = odota_adv(by_o, baseline, aid, bid) s_adv, s_games = stratz_adv(by_s, aid, bid) if ( - o_adv is not None and o_games >= args.min_games and o_adv >= 0.03 - and s_adv is not None and s_games >= args.min_games and s_adv >= 2.0 + o_adv is not None + and o_games >= args.min_games + and o_adv >= 0.03 + and s_adv is not None + and s_games >= args.min_games + and s_adv >= 2.0 ): - suggestions.append({ - "a": akey, "a_loc": names[akey], - "b": bkey, "b_loc": names[bkey], - "opendota_adv": round(o_adv, 4), "opendota_games": o_games, - "stratz_adv": round(s_adv, 3), "stratz_games": s_games, - }) + suggestions.append( + { + "a": akey, + "a_loc": names[akey], + "b": bkey, + "b_loc": names[bkey], + "opendota_adv": round(o_adv, 4), + "opendota_games": o_games, + "stratz_adv": round(s_adv, 3), + "stratz_games": s_games, + } + ) suggestions.sort(key=lambda r: (-r["stratz_adv"], -r["opendota_adv"])) suggestions = suggestions[:100] - # Per-hero coverage in qualitative relations counters_from: dict[str, int] = {k: 0 for k in all_keys} counters_to: dict[str, int] = {k: 0 for k in all_keys} syn_count: dict[str, int] = {k: 0 for k in all_keys} @@ -249,20 +287,18 @@ def main() -> None: ], } - # Qualitative gaps: no counter-out, no counter-in, no synergy no_counter_out = [k for k in all_keys if counters_from[k] == 0] no_counter_in = [k for k in all_keys if counters_to[k] == 0] no_synergy = [k for k in all_keys if syn_count[k] == 0] thin = [ - k for k in all_keys - if counters_from[k] + counters_to[k] + syn_count[k] <= 1 + k for k in all_keys if counters_from[k] + counters_to[k] + syn_count[k] <= 1 ] empty = [ - k for k in all_keys + k + for k in all_keys if counters_from[k] == 0 and counters_to[k] == 0 and syn_count[k] == 0 ] - # Heroes with strong cross-source suggestions but empty/thin qualitative coverage suggest_by_hero: dict[str, int] = {} for row in suggestions: suggest_by_hero[row["a"]] = suggest_by_hero.get(row["a"], 0) + 1 @@ -278,31 +314,60 @@ def main() -> None: } for k in empty + [x for x in thin if x not in empty] ], - key=lambda r: (r["counters_out"] + r["counters_in"] + r["synergies"], -r["cross_source_suggestions"], r["key"]), + key=lambda r: ( + r["counters_out"] + r["counters_in"] + r["synergies"], + -r["cross_source_suggestions"], + r["key"], + ), ) + web_tops_audit = None + if tops.get("by_hero"): + web_tops_audit = audit_matchup_tops( + tops, + odota, + id_to_key=id_to_key, + key_to_id=key_to_id, + names=names, + min_games=args.min_games, + ) + + def _rel(path: Path) -> str | None: + if not path.is_file(): + return None + try: + return str(path.relative_to(ROOT)).replace("\\", "/") + except ValueError: + return str(path).replace("\\", "/") + report = { "generated_at": datetime.now(timezone.utc).isoformat(), "sources": { "relations": str(DEFAULT_RELATIONS.relative_to(ROOT)).replace("\\", "/"), "opendota_matchups": { - "path": str(MATCHUPS.relative_to(ROOT)).replace("\\", "/") if MATCHUPS.is_file() else None, + "path": _rel(matchups_path), "source": odota.get("source"), "fetched_at": odota.get("fetched_at"), "heroes": len(by_o), }, "stratz_matchups": { - "path": str(STRATZ_MATCHUPS.relative_to(ROOT)).replace("\\", "/") if STRATZ_MATCHUPS.is_file() else None, + "path": _rel(stratz_path), "source": stratz.get("source"), "fetched_at": stratz.get("fetched_at"), "heroes": len(by_s), }, "stratz_synergies": { - "path": str(SYNERGIES.relative_to(ROOT)).replace("\\", "/") if SYNERGIES.is_file() else None, + "path": _rel(syn_path), "source": syn_file.get("source"), "fetched_at": syn_file.get("fetched_at"), "heroes": len(by_syn), }, + "stratz_matchup_tops": { + "path": _rel(tops_path), + "source": tops.get("source"), + "fetched_at": tops.get("fetched_at"), + "heroes": len(tops.get("by_hero") or {}), + }, }, "thresholds": { "min_games": args.min_games, @@ -338,13 +403,20 @@ def main() -> None: counter_conflict, key=lambda r: ((r.get("stratz_adv") or 0), (r.get("opendota_adv") or 0)), ), - "counter_agree_both": [r for r in counter_ok if set(r.get("agree_sources") or []) == {"opendota", "stratz"}], + "counter_agree_both": [ + r + for r in counter_ok + if set(r.get("agree_sources") or []) == {"opendota", "stratz"} + ], "suggest_add_counters_cross_source": suggestions, "synergy_conflicts": syn_conflict, + "web_matchup_tops_cross": web_tops_audit, } args.out.parent.mkdir(parents=True, exist_ok=True) - args.out.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + args.out.write_text( + json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" + ) s = report["summary"] print("=== relations audit ===") @@ -361,6 +433,14 @@ def main() -> None: f"qualitative gaps: empty={s['heroes_empty_relations']} thin(<={1} edge)={s['heroes_thin_relations']} " f"/ heroes={s['heroes_total']}" ) + if web_tops_audit: + ws = web_tops_audit["summary"] + print( + "web matchup tops × OpenDota: " + f"agree={ws['agree']} conflict={ws['conflict']} weak={ws['weak']} " + f"/ rows={ws['counter_rows']} heroes={ws['heroes']}" + ) + print(f" note: {web_tops_audit['manual_review_note']}") print("cache gaps:") for k, v in missing_cache.items(): print(f" {k}: {len(v)}") diff --git a/shared/matchup_cross.py b/shared/matchup_cross.py new file mode 100644 index 0000000..d5f356f --- /dev/null +++ b/shared/matchup_cross.py @@ -0,0 +1,281 @@ +"""Cross-check STRATZ web matchup tops against OpenDota hero matchups. + +Web-only observation evidence — never merge into relations.json or recommend. + +Statuses: + agree — OpenDota baseline-adjusted advantage agrees with STRATZ direction + conflict — OpenDota disagrees with STRATZ direction (enough games) + weak — OpenDota sample too small or advantage near zero +""" + +from __future__ import annotations + +from typing import Any + +# Defaults aligned with shared/audit_relations.py thresholds. +DEFAULT_MIN_GAMES = 80 +DEFAULT_ODOTA_AGREE = 0.015 +DEFAULT_ODOTA_DISAGREE = -0.015 + + +def matchup_wr(by_hero: dict, aid: int, bid: int) -> tuple[float | None, int]: + cell = (by_hero.get(str(aid)) or {}).get(str(bid)) + if not cell: + return None, 0 + games = int(cell.get("games") or 0) + wins = int(cell.get("wins") or 0) + if games <= 0: + return None, 0 + return wins / games, games + + +def build_baseline(by_hero: dict) -> dict[int, float]: + out: dict[int, float] = {} + for hid_s, opps in by_hero.items(): + tw = tg = 0 + for cell in opps.values(): + g = int(cell.get("games") or 0) + w = int(cell.get("wins") or 0) + tw += w + tg += g + if tg > 0: + out[int(hid_s)] = tw / tg + return out + + +def odota_adv( + by_hero: dict, baseline: dict[int, float], aid: int, bid: int +) -> tuple[float | None, int]: + wr, games = matchup_wr(by_hero, aid, bid) + if wr is None: + return None, 0 + base = baseline.get(aid) + if base is None: + return None, games + return wr - base, games + + +def classify_cross_source( + *, + stratz_signed: float, + odota_adv_val: float | None, + odota_games: int, + min_games: int = DEFAULT_MIN_GAMES, + odota_agree: float = DEFAULT_ODOTA_AGREE, + odota_disagree: float = DEFAULT_ODOTA_DISAGREE, +) -> str: + """Classify whether OpenDota agrees with a STRATZ signed advantage. + + ``stratz_signed`` > 0 means STRATZ says A is favored vs B (counters). + For countered rows the caller should pass the original vs advantage + (positive = A still favored), not the negated display value. + """ + if odota_adv_val is None or odota_games < min_games: + return "weak" + if stratz_signed >= 0: + if odota_adv_val >= odota_agree: + return "agree" + if odota_adv_val <= odota_disagree: + return "conflict" + return "weak" + if odota_adv_val <= -odota_agree: + return "agree" + if odota_adv_val >= -odota_disagree: + return "conflict" + return "weak" + + +def enrich_entry_cross( + entry: dict, + *, + hero_id: int, + kind: str, + by_odota: dict, + baseline: dict[int, float], + min_games: int = DEFAULT_MIN_GAMES, +) -> dict: + """Attach ``cross`` quality blob to one counters/countered/synergies row. + + For ``countered`` rows, STRATZ stores negated advantage for display; we + restore the original vs-sign for classification (``-advantage``). + Synergies skip OpenDota (no teammate WR in matchups.json). + """ + out = dict(entry) + peer = int(entry.get("hero_id") or 0) + if kind == "synergies" or peer <= 0 or hero_id <= 0: + out["cross"] = {"status": "weak", "reason": "no_odota_synergy"} + return out + + o_adv, o_games = odota_adv(by_odota, baseline, hero_id, peer) + raw_adv = float(entry.get("advantage") or 0.0) + # countered display advantage is already negated; restore original vs sign. + stratz_signed = -raw_adv if kind == "countered" else raw_adv + status = classify_cross_source( + stratz_signed=stratz_signed, + odota_adv_val=o_adv, + odota_games=o_games, + min_games=min_games, + ) + out["cross"] = { + "status": status, + "opendota_adv": None if o_adv is None else round(o_adv, 4), + "opendota_games": o_games, + "min_games": min_games, + } + return out + + +def enrich_hero_matchups_cross( + cell: dict, + *, + hero_id: int, + by_odota: dict, + baseline: dict[int, float], + min_games: int = DEFAULT_MIN_GAMES, +) -> dict: + """Return a shallow-copied hero matchup cell with per-row ``cross`` fields.""" + out = { + "counters": [ + enrich_entry_cross( + e, + hero_id=hero_id, + kind="counters", + by_odota=by_odota, + baseline=baseline, + min_games=min_games, + ) + for e in (cell.get("counters") or []) + if isinstance(e, dict) + ], + "countered": [ + enrich_entry_cross( + e, + hero_id=hero_id, + kind="countered", + by_odota=by_odota, + baseline=baseline, + min_games=min_games, + ) + for e in (cell.get("countered") or []) + if isinstance(e, dict) + ], + "synergies": [ + enrich_entry_cross( + e, + hero_id=hero_id, + kind="synergies", + by_odota=by_odota, + baseline=baseline, + min_games=min_games, + ) + for e in (cell.get("synergies") or []) + if isinstance(e, dict) + ], + } + for k in ("fetched_at", "stale"): + if k in cell: + out[k] = cell[k] + return out + + +def summarize_cross_rows(rows: list[dict]) -> dict[str, int]: + counts = {"agree": 0, "conflict": 0, "weak": 0} + for e in rows: + status = ((e.get("cross") or {}).get("status")) or "weak" + if status not in counts: + status = "weak" + counts[status] += 1 + return counts + + +def audit_matchup_tops( + tops: dict, + odota: dict, + *, + id_to_key: dict[int, str], + key_to_id: dict[str, int], + names: dict[str, str], + min_games: int = DEFAULT_MIN_GAMES, +) -> dict[str, Any]: + """Audit STRATZ web tops vs OpenDota; return report section (no file IO).""" + by_odota = odota.get("by_hero") or {} + baseline = build_baseline(by_odota) + by_hero = tops.get("by_hero") or {} + + pairs: list[dict] = [] + summary = { + "heroes": 0, + "counter_rows": 0, + "agree": 0, + "conflict": 0, + "weak": 0, + } + + for key, cell in sorted(by_hero.items()): + if not isinstance(cell, dict): + continue + hid = key_to_id.get(key) + if hid is None: + # Prefer embedded id when key map lags new heroes. + hid = int(cell.get("id") or 0) or None + if hid is None: + continue + summary["heroes"] += 1 + enriched = enrich_hero_matchups_cross( + cell, + hero_id=hid, + by_odota=by_odota, + baseline=baseline, + min_games=min_games, + ) + for kind in ("counters", "countered"): + for e in enriched.get(kind) or []: + summary["counter_rows"] += 1 + cross = e.get("cross") or {} + status = cross.get("status") or "weak" + summary[status] = summary.get(status, 0) + 1 + peer_id = int(e.get("hero_id") or 0) + peer_key = id_to_key.get(peer_id, str(peer_id)) + pairs.append( + { + "hero": key, + "hero_loc": names.get(key, key), + "peer": peer_key, + "peer_loc": names.get(peer_key, peer_key), + "kind": kind, + "stratz_advantage": e.get("advantage"), + "stratz_wr": e.get("wr"), + "stratz_games": e.get("games"), + "status": status, + "opendota_adv": cross.get("opendota_adv"), + "opendota_games": cross.get("opendota_games"), + } + ) + + conflicts = [p for p in pairs if p["status"] == "conflict"] + conflicts.sort( + key=lambda r: ( + abs(float(r.get("stratz_advantage") or 0)), + -(int(r.get("opendota_games") or 0)), + ), + reverse=True, + ) + agrees = [p for p in pairs if p["status"] == "agree"] + agrees.sort( + key=lambda r: ( + abs(float(r.get("stratz_advantage") or 0)), + -(int(r.get("opendota_games") or 0)), + ), + reverse=True, + ) + + return { + "summary": summary, + "thresholds": {"min_games": min_games}, + "manual_review_note": ( + "Dota2ProTracker (7k+ MMR / pro) is a manual high-MMR reference for " + "conflict rows; not automated (login-gated)." + ), + "conflicts": conflicts[:100], + "agrees_sample": agrees[:40], + } diff --git a/shared/tests/__init__.py b/shared/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/shared/tests/test_matchup_cross.py b/shared/tests/test_matchup_cross.py new file mode 100644 index 0000000..c34663c --- /dev/null +++ b/shared/tests/test_matchup_cross.py @@ -0,0 +1,204 @@ +"""Unit tests for STRATZ matchup ranking + OpenDota cross classification. + +Run from repo root: + python -m unittest shared.tests.test_matchup_cross -v +""" + +from __future__ import annotations + +import sys +import tempfile +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from shared.matchup_cross import ( # noqa: E402 + classify_cross_source, + enrich_entry_cross, + enrich_hero_matchups_cross, +) +from web.fetch_stratz_meta import ( # noqa: E402 + annotate_matchup_cell, + build_matchup_file_payload, + load_previous_matchups, + rank_matchup_lists, + write_json_atomic, +) + + +class RankMatchupListsTests(unittest.TestCase): + def test_counters_sorted_by_advantage_not_raw_wr(self) -> None: + # Wraith King: wr < 0.5 but positive relative advantage — still a counter. + vs = [ + { + "hero_id": 42, + "games": 3286, + "wins": 1597, + "advantage": 4.444, + "wr": 0.486, + }, + { + "hero_id": 94, + "games": 1070, + "wins": 649, + "advantage": 12.994, + "wr": 0.607, + }, + { + "hero_id": 34, + "games": 1171, + "wins": 723, + "advantage": 7.318, + "wr": 0.617, + }, + ] + ranked = rank_matchup_lists(vs, [], take=3) + self.assertEqual( + [e["hero_id"] for e in ranked["counters"]], + [94, 34, 42], + ) + wk = ranked["counters"][2] + self.assertLess(wk["wr"], 0.5) + self.assertGreater(wk["advantage"], 0) + + def test_countered_mirrors_lowest_advantage(self) -> None: + vs = [ + {"hero_id": 1, "games": 100, "wins": 60, "advantage": 5.0, "wr": 0.6}, + {"hero_id": 2, "games": 100, "wins": 40, "advantage": -8.0, "wr": 0.4}, + {"hero_id": 3, "games": 50, "wins": 20, "advantage": -3.0, "wr": 0.4}, + ] + ranked = rank_matchup_lists(vs, [], take=2) + fear = ranked["countered"] + self.assertEqual([e["hero_id"] for e in fear], [2, 3]) + self.assertAlmostEqual(fear[0]["advantage"], 8.0) + self.assertAlmostEqual(fear[0]["wr"], 0.6) + + def test_synergies_sorted_desc(self) -> None: + with_rows = [ + {"hero_id": 10, "games": 10, "wins": 5, "synergy": 1.0, "wr": 0.5}, + {"hero_id": 11, "games": 10, "wins": 6, "synergy": 4.5, "wr": 0.6}, + ] + ranked = rank_matchup_lists([], with_rows, take=2) + self.assertEqual([e["hero_id"] for e in ranked["synergies"]], [11, 10]) + + +class MatchupCacheHelpersTests(unittest.TestCase): + def test_annotate_and_payload_metadata(self) -> None: + cell = annotate_matchup_cell( + {"counters": [{"hero_id": 1}], "countered": [], "synergies": []}, + fetched_at="2026-07-28T00:00:00+00:00", + stale=True, + ) + self.assertTrue(cell["stale"]) + self.assertEqual(cell["fetched_at"], "2026-07-28T00:00:00+00:00") + payload = build_matchup_file_payload( + {"antimage": cell}, + take=12, + match_limit=50, + started_at="2026-07-28T00:00:00+00:00", + finished_at="2026-07-28T01:00:00+00:00", + stats={"heroes": 1, "ok": 0, "failed": 1, "stale_kept": 1}, + ) + self.assertEqual(payload["scope"]["kind"], "global_aggregate") + self.assertIn("global", payload["scope"]["kind"]) + self.assertTrue(payload["scope"]["label_zh"]) + self.assertEqual(payload["stats"]["stale_kept"], 1) + + def test_atomic_write_and_resume_load(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "stratz_matchup_tops.json" + payload = build_matchup_file_payload( + { + "antimage": annotate_matchup_cell( + {"counters": [], "countered": [], "synergies": []}, + fetched_at="t0", + ) + }, + take=12, + match_limit=50, + started_at="t0", + finished_at="t0", + ) + write_json_atomic(path, payload) + loaded = load_previous_matchups(path) + self.assertIn("antimage", loaded) + + +class CrossClassifyTests(unittest.TestCase): + def test_agree_conflict_weak(self) -> None: + self.assertEqual( + classify_cross_source( + stratz_signed=5.0, odota_adv_val=0.04, odota_games=200 + ), + "agree", + ) + self.assertEqual( + classify_cross_source( + stratz_signed=5.0, odota_adv_val=-0.04, odota_games=200 + ), + "conflict", + ) + self.assertEqual( + classify_cross_source( + stratz_signed=5.0, odota_adv_val=0.04, odota_games=10 + ), + "weak", + ) + + def test_countered_uses_restored_sign(self) -> None: + # Display advantage for countered is +8 (negated); original vs was -8. + by_odota = { + "1": { + "2": {"games": 200, "wins": 80}, # AM wr 40% vs peer + } + } + # Baseline for hero 1 ≈ 0.4 from only this matchup. + from shared.matchup_cross import build_baseline + + baseline = build_baseline(by_odota) + entry = enrich_entry_cross( + { + "hero_id": 2, + "games": 100, + "wins": 40, + "advantage": 8.0, + "wr": 0.6, + }, + hero_id=1, + kind="countered", + by_odota=by_odota, + baseline=baseline, + min_games=80, + ) + # odota_adv = 0.4 - 0.4 = 0 → weak + self.assertEqual(entry["cross"]["status"], "weak") + + def test_enrich_hero_keeps_stale_flag(self) -> None: + cell = { + "counters": [ + { + "hero_id": 94, + "games": 100, + "wins": 60, + "advantage": 3.0, + "wr": 0.6, + } + ], + "countered": [], + "synergies": [], + "fetched_at": "t0", + "stale": True, + } + out = enrich_hero_matchups_cross( + cell, + hero_id=1, + by_odota={}, + baseline={}, + ) + self.assertTrue(out["stale"]) + self.assertEqual(out["cross"] if False else out["counters"][0]["cross"]["status"], "weak") + + +if __name__ == "__main__": + unittest.main() diff --git a/web/__init__.py b/web/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/web/assets/streamer_avatars/leize.jpg b/web/assets/streamer_avatars/leize.jpg new file mode 100644 index 0000000..34cb8d5 Binary files /dev/null and b/web/assets/streamer_avatars/leize.jpg differ diff --git a/web/data/streamers.json b/web/data/streamers.json index 448ed34..bb88ae3 100644 --- a/web/data/streamers.json +++ b/web/data/streamers.json @@ -215,6 +215,24 @@ "is_live": false, "live_probed_at": "2026-07-29T05:14:20.319003+00:00" }, + { + "id": "leize", + "platform": "douyin", + "profile_url": "https://v.douyin.com/bxYZijBybSE/", + "heroes": [ + "razor" + ], + "nickname": "卡奥兹Khs", + "unique_id": "Khaos995", + "signature": "万分核心｜战斗C位｜顶级心态 上车/打号：Khaos…", + "following_count": 90, + "follower_count": 2921, + "total_favorited": 46000, + "avatar": "streamer_avatars/leize.jpg", + "video": "streamer_videos/leize.mp4", + "video_title": "POS 1 雷泽全新分享 · 雷霆爆轰征服天梯", + "profile_fetched_at": "2026-07-29T07:20:00+00:00" + }, { "id": "shawang", "platform": "bilibili", diff --git a/web/fetch_stratz_meta.py b/web/fetch_stratz_meta.py index c7ff8e0..c8c1c58 100644 --- a/web/fetch_stratz_meta.py +++ b/web/fetch_stratz_meta.py @@ -9,11 +9,14 @@ Data sources (heroStats GraphQL): - winWeek(take=1, bracketIds, positionIds): per-medal per-position latest week → `positions` (same time window as headline cards; exact medal, not basic merge) - matchUp: counter / countered / synergy tops → stratz_matchup_tops.json + (global aggregate — no bracket / position / week filter) Usage: - python fetch_stratz_meta.py - python fetch_stratz_meta.py --weeks 8 --delay 0.25 - python fetch_stratz_meta.py --skip-matchups + python web/fetch_stratz_meta.py + python web/fetch_stratz_meta.py --weeks 8 --delay 0.25 + python web/fetch_stratz_meta.py --skip-matchups + python web/fetch_stratz_meta.py --matchups-only + python web/fetch_stratz_meta.py --resume-matchups # interrupt resume only """ from __future__ import annotations @@ -26,6 +29,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import argparse import json import os +import tempfile import time import urllib.error import urllib.request @@ -82,6 +86,27 @@ POSITION_ORDER = ( ATTRIBUTION = "https://stratz.com" +MATCHUP_SCOPE = { + "kind": "global_aggregate", + "bracket": None, + "position": None, + "week": None, + "label_zh": "全局聚合（未按段位 / 分路 / 周过滤）", + "note": ( + "STRATZ heroStats.matchUp without bracketIds/positionIds/week. " + "advantage is upstream synergy (relative), not raw win-rate pp. " + "Web-only; do not merge into relations.json." + ), +} + +MATCHUP_NOTE = ( + "counters = positive vs advantage (STRATZ synergy); " + "countered = heroes this hero loses to (negated advantage, mirrored wr); " + "synergies = with synergy. " + "advantage ≠ win-rate percentage points. " + "Web-only; do not merge into relations.json." +) + def load_token() -> str: for key in ( @@ -136,6 +161,32 @@ def _pw(pick: int, win: int) -> dict: return {"pick": pick, "win": win, "wr": wr} +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def write_json_atomic(path: Path, payload: dict) -> None: + """Write JSON via temp file then replace, so partial writes never corrupt cache.""" + path.parent.mkdir(parents=True, exist_ok=True) + text = json.dumps(payload, ensure_ascii=False, indent=2) + "\n" + fd, tmp_name = tempfile.mkstemp( + prefix=f".{path.name}.", + suffix=".tmp", + dir=str(path.parent), + ) + tmp_path = Path(tmp_name) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(text) + os.replace(tmp_path, path) + except Exception: + try: + tmp_path.unlink(missing_ok=True) + except OSError: + pass + raise + + def fetch_weeks_for_bracket( token: str, hero_ids: list[int], bracket: str, take: int ) -> list[dict]: @@ -216,18 +267,10 @@ query($id: Short!, $take: Int!, $limit: Int!) { """ -def fetch_matchup_tops( - token: str, hero_id: int, *, take: int, match_limit: int -) -> dict: - data = gql( - token, - MATCHUP_QUERY, - {"id": hero_id, "take": take, "limit": match_limit}, - ) - rows = (((data or {}).get("heroStats") or {}).get("matchUp")) or [] - row = rows[0] if rows else {} - vs_out = [] - with_out = [] +def parse_matchup_pairs(row: dict) -> tuple[list[dict], list[dict]]: + """Parse raw STRATZ matchUp row into vs / with pair lists.""" + vs_out: list[dict] = [] + with_out: list[dict] = [] for pair in row.get("vs") or []: other = pair.get("heroId2") games = int(pair.get("matchCount") or 0) @@ -258,8 +301,21 @@ def fetch_matchup_tops( "wr": float(pair.get("winsAverage") or (wins / games)), } ) + return vs_out, with_out + + +def rank_matchup_lists( + vs_out: list[dict], with_out: list[dict], take: int +) -> dict: + """Sort counters / countered / synergies from parsed vs / with pairs. + + Positive ``advantage`` means this hero's STRATZ relative score vs the + peer is favorable — not raw win-rate percentage points. A hero can appear + under counters with wr < 0.5 when advantage is still positive vs baseline. + """ + take = max(0, int(take)) # vs advantage: positive = hero wins more vs other → counters other. - # Also derive "disadvantage" as others with most negative advantage for hero. + # countered is the same vs list mirrored (negated advantage, 1-wr). disadvantage = [ { "hero_id": e["hero_id"], @@ -279,6 +335,75 @@ def fetch_matchup_tops( } +def fetch_matchup_tops( + token: str, hero_id: int, *, take: int, match_limit: int +) -> dict: + data = gql( + token, + MATCHUP_QUERY, + {"id": hero_id, "take": take, "limit": match_limit}, + ) + rows = (((data or {}).get("heroStats") or {}).get("matchUp")) or [] + row = rows[0] if rows else {} + vs_out, with_out = parse_matchup_pairs(row) + return rank_matchup_lists(vs_out, with_out, take) + + +def annotate_matchup_cell( + cell: dict, *, fetched_at: str, stale: bool = False +) -> dict: + """Attach per-hero fetch metadata; preserve list payloads.""" + out = { + "counters": list(cell.get("counters") or []), + "countered": list(cell.get("countered") or []), + "synergies": list(cell.get("synergies") or []), + "fetched_at": fetched_at, + "stale": bool(stale), + } + return out + + +def build_matchup_file_payload( + by_hero: dict[str, dict], + *, + take: int, + match_limit: int, + started_at: str, + finished_at: str | None = None, + stats: dict | None = None, +) -> dict: + return { + "fetched_at": finished_at or started_at, + "started_at": started_at, + "finished_at": finished_at, + "source": "stratz", + "attribution": ATTRIBUTION, + "take": take, + "match_limit": match_limit, + "scope": dict(MATCHUP_SCOPE), + "note": MATCHUP_NOTE, + "stats": stats + or { + "heroes": len(by_hero), + "ok": 0, + "failed": 0, + "stale_kept": 0, + }, + "by_hero": by_hero, + } + + +def load_previous_matchups(path: Path) -> dict[str, dict]: + if not path.is_file(): + return {} + try: + prev = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + by_hero = prev.get("by_hero") or {} + return {k: v for k, v in by_hero.items() if isinstance(v, dict)} + + def build_meta( weeks_rows_by_bracket: dict[str, list[dict]], position_rows_by_bracket: dict[str, list[dict]], @@ -371,7 +496,7 @@ def build_meta( meta_board[bracket] = rows_board return { - "fetched_at": datetime.now(timezone.utc).isoformat(), + "fetched_at": _now_iso(), "source": "stratz", "attribution": ATTRIBUTION, "weeks_take": weeks_take, @@ -386,6 +511,139 @@ def build_meta( } +def refresh_matchup_tops( + token: str, + id_to_key: dict[int, str], + *, + take: int, + match_limit: int, + delay: float, + out_path: Path, + resume: bool, +) -> dict: + """Full-refresh (default) or resume-only matchup tops. + + On per-hero failure, keep the previous cell and mark ``stale=True``. + """ + started_at = _now_iso() + prev = load_previous_matchups(out_path) + hero_ids = sorted(id_to_key.keys()) + + if resume: + pending = [hid for hid in hero_ids if id_to_key[hid] not in prev] + by_hero_mu: dict[str, dict] = dict(prev) + print( + f"resuming matchups: {len(pending)} pending / {len(hero_ids)} " + f"(cached={len(prev)})", + flush=True, + ) + else: + pending = list(hero_ids) + by_hero_mu = {} + print( + f"full matchup refresh: {len(pending)} heroes " + f"(prev cached={len(prev)} as failure fallback)", + flush=True, + ) + + ok = failed = stale_kept = 0 + for n, hid in enumerate(pending, start=1): + key = id_to_key[hid] + try: + cell = fetch_matchup_tops( + token, + hid, + take=take, + match_limit=match_limit, + ) + by_hero_mu[key] = annotate_matchup_cell( + cell, fetched_at=_now_iso(), stale=False + ) + ok += 1 + print( + f" [{n}/{len(pending)}] {key}: " + f"vs={len(cell['counters'])} fear={len(cell['countered'])} " + f"with={len(cell['synergies'])}", + flush=True, + ) + except ( + urllib.error.URLError, + TimeoutError, + RuntimeError, + json.JSONDecodeError, + ) as e: + failed += 1 + old = prev.get(key) + if old: + kept = annotate_matchup_cell( + old, + fetched_at=str(old.get("fetched_at") or started_at), + stale=True, + ) + by_hero_mu[key] = kept + stale_kept += 1 + print( + f" [{n}/{len(pending)}] {key} failed (kept stale): {e}", + flush=True, + ) + else: + print(f" [{n}/{len(pending)}] {key} failed: {e}", flush=True) + time.sleep(delay * 2) + continue + + finished_partial = _now_iso() + payload = build_matchup_file_payload( + by_hero_mu, + take=take, + match_limit=match_limit, + started_at=started_at, + finished_at=finished_partial, + stats={ + "heroes": len(by_hero_mu), + "ok": ok, + "failed": failed, + "stale_kept": stale_kept, + "pending_left": len(pending) - n, + }, + ) + write_json_atomic(out_path, payload) + time.sleep(delay) + + # Resume mode: ensure heroes already present stay; full mode already has all ok/stale. + if resume: + for hid in hero_ids: + key = id_to_key[hid] + if key not in by_hero_mu and key in prev: + by_hero_mu[key] = annotate_matchup_cell( + prev[key], + fetched_at=str(prev[key].get("fetched_at") or started_at), + stale=bool(prev[key].get("stale")), + ) + + finished_at = _now_iso() + payload = build_matchup_file_payload( + by_hero_mu, + take=take, + match_limit=match_limit, + started_at=started_at, + finished_at=finished_at, + stats={ + "heroes": len(by_hero_mu), + "ok": ok, + "failed": failed, + "stale_kept": stale_kept, + "pending_left": 0, + }, + ) + write_json_atomic(out_path, payload) + print( + f"done: matchups heroes={len(by_hero_mu)} ok={ok} failed={failed} " + f"stale_kept={stale_kept} → {out_path}", + flush=True, + ) + return payload + + def main() -> None: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--weeks", type=int, default=8, help="weekly buckets to keep") @@ -393,6 +651,16 @@ def main() -> None: ap.add_argument("--out-meta", type=Path, default=OUT_META) ap.add_argument("--out-matchups", type=Path, default=OUT_MATCHUPS) ap.add_argument("--skip-matchups", action="store_true") + ap.add_argument( + "--matchups-only", + action="store_true", + help="skip winWeek meta; only refresh matchup tops", + ) + ap.add_argument( + "--resume-matchups", + action="store_true", + help="only fetch heroes missing from existing matchup cache (interrupt resume)", + ) ap.add_argument("--matchup-take", type=int, default=12) ap.add_argument("--matchup-min-games", type=int, default=50) args = ap.parse_args() @@ -401,103 +669,68 @@ def main() -> None: heroes = hero_table() id_to_key = {int(h["id"]): h["key"] for h in heroes} hero_ids = sorted(id_to_key.keys()) - print( - f"fetching STRATZ meta for {len(hero_ids)} heroes, " - f"{args.weeks} weeks × {len(BRACKET_ORDER)} brackets", - flush=True, - ) - weeks_by_bracket: dict[str, list[dict]] = {} - position_rows_by_bracket: dict[str, list[dict]] = {} - for i, bracket in enumerate(BRACKET_ORDER, start=1): - try: - rows = fetch_weeks_for_bracket(token, hero_ids, bracket, args.weeks) - except (urllib.error.URLError, TimeoutError, RuntimeError, json.JSONDecodeError) as e: - print(f" [{i}/{len(BRACKET_ORDER)}] {bracket} failed: {e}", flush=True) - rows = [] - weeks_by_bracket[bracket] = rows - try: - pos_rows = fetch_latest_positions_for_bracket(token, hero_ids, bracket) - except (urllib.error.URLError, TimeoutError, RuntimeError, json.JSONDecodeError) as e: - print(f" [{i}/{len(BRACKET_ORDER)}] {bracket} positions failed: {e}", flush=True) - pos_rows = [] - position_rows_by_bracket[bracket] = pos_rows + if not args.matchups_only: print( - f" [{i}/{len(BRACKET_ORDER)}] {bracket}: {len(rows)} week rows, " - f"{len(pos_rows)} position rows", + f"fetching STRATZ meta for {len(hero_ids)} heroes, " + f"{args.weeks} weeks × {len(BRACKET_ORDER)} brackets", flush=True, ) - time.sleep(args.delay) - meta = build_meta( - weeks_by_bracket, position_rows_by_bracket, id_to_key, args.weeks - ) - args.out_meta.parent.mkdir(parents=True, exist_ok=True) - args.out_meta.write_text( - json.dumps(meta, ensure_ascii=False, indent=2) + "\n", - encoding="utf-8", - ) - print(f"wrote {args.out_meta}", flush=True) + weeks_by_bracket: dict[str, list[dict]] = {} + position_rows_by_bracket: dict[str, list[dict]] = {} + for i, bracket in enumerate(BRACKET_ORDER, start=1): + try: + rows = fetch_weeks_for_bracket(token, hero_ids, bracket, args.weeks) + except ( + urllib.error.URLError, + TimeoutError, + RuntimeError, + json.JSONDecodeError, + ) as e: + print(f" [{i}/{len(BRACKET_ORDER)}] {bracket} failed: {e}", flush=True) + rows = [] + weeks_by_bracket[bracket] = rows + try: + pos_rows = fetch_latest_positions_for_bracket(token, hero_ids, bracket) + except ( + urllib.error.URLError, + TimeoutError, + RuntimeError, + json.JSONDecodeError, + ) as e: + print( + f" [{i}/{len(BRACKET_ORDER)}] {bracket} positions failed: {e}", + flush=True, + ) + pos_rows = [] + position_rows_by_bracket[bracket] = pos_rows + print( + f" [{i}/{len(BRACKET_ORDER)}] {bracket}: {len(rows)} week rows, " + f"{len(pos_rows)} position rows", + flush=True, + ) + time.sleep(args.delay) + + meta = build_meta( + weeks_by_bracket, position_rows_by_bracket, id_to_key, args.weeks + ) + write_json_atomic(args.out_meta, meta) + print(f"wrote {args.out_meta}", flush=True) if args.skip_matchups: print("skip matchup tops", flush=True) return - by_hero_mu: dict[str, dict] = {} - if args.out_matchups.is_file(): - try: - prev = json.loads(args.out_matchups.read_text(encoding="utf-8")) - by_hero_mu = dict(prev.get("by_hero") or {}) - print(f"resuming matchups with {len(by_hero_mu)} heroes", flush=True) - except (OSError, json.JSONDecodeError): - pass - - pending = [hid for hid in hero_ids if id_to_key[hid] not in by_hero_mu] - print( - f"fetching matchup tops {len(pending)}/{len(hero_ids)} " - f"(take={args.matchup_take}, min_games={args.matchup_min_games})", - flush=True, + refresh_matchup_tops( + token, + id_to_key, + take=args.matchup_take, + match_limit=args.matchup_min_games, + delay=args.delay, + out_path=args.out_matchups, + resume=bool(args.resume_matchups), ) - for n, hid in enumerate(pending, start=1): - key = id_to_key[hid] - try: - cell = fetch_matchup_tops( - token, - hid, - take=args.matchup_take, - match_limit=args.matchup_min_games, - ) - except (urllib.error.URLError, TimeoutError, RuntimeError, json.JSONDecodeError) as e: - print(f" [{n}/{len(pending)}] {key} failed: {e}", flush=True) - time.sleep(args.delay * 2) - continue - by_hero_mu[key] = cell - print( - f" [{n}/{len(pending)}] {key}: " - f"vs={len(cell['counters'])} fear={len(cell['countered'])} " - f"with={len(cell['synergies'])}", - flush=True, - ) - payload = { - "fetched_at": datetime.now(timezone.utc).isoformat(), - "source": "stratz", - "attribution": ATTRIBUTION, - "take": args.matchup_take, - "match_limit": args.matchup_min_games, - "note": ( - "counters = positive vs advantage; countered = heroes this hero " - "loses to (negated advantage); synergies = with synergy. " - "Web-only; do not merge into relations.json." - ), - "by_hero": by_hero_mu, - } - args.out_matchups.write_text( - json.dumps(payload, ensure_ascii=False, indent=2) + "\n", - encoding="utf-8", - ) - time.sleep(args.delay) - - print(f"done: matchups {len(by_hero_mu)} heroes → {args.out_matchups}", flush=True) if __name__ == "__main__": diff --git a/web/refresh_web.py b/web/refresh_web.py index 112163c..06b0172 100644 --- a/web/refresh_web.py +++ b/web/refresh_web.py @@ -103,12 +103,21 @@ def diff_snapshots(before: dict[str, str | None], after: dict[str, str | None]) return data_changed, assets_changed -def run_script(script: str, *args: str, dry_run: bool = False) -> None: +def run_script(script: str, *args: str, dry_run: bool = False, soft_fail: bool = False) -> bool: cmd = [sys.executable, str(ROOT / script), *args] print(f"+ {' '.join(cmd)}", flush=True) if dry_run: - return - subprocess.run(cmd, cwd=str(ROOT), check=True) + return True + proc = subprocess.run(cmd, cwd=str(ROOT), check=False) + if proc.returncode == 0: + return True + if soft_fail: + print( + f"soft-fail: {script} exited {proc.returncode}; continuing", + flush=True, + ) + return False + raise subprocess.CalledProcessError(proc.returncode, cmd) def patch_check() -> dict: @@ -137,6 +146,13 @@ def run_patch_linked(*, dry_run: bool = False) -> None: run_script("fetch_item_shop.py", dry_run=dry_run) run_script("fetch_items_meta.py", dry_run=dry_run) run_script("item_fears.py", dry_run=dry_run) + # Soft: patch workflow may lack STRATZ token; weekly still does the hard refresh. + run_script( + "fetch_stratz_meta.py", + "--matchups-only", + dry_run=dry_run, + soft_fail=True, + ) def run_daily(*, dry_run: bool = False) -> bool: @@ -162,6 +178,7 @@ def run_daily(*, dry_run: bool = False) -> bool: def run_weekly(*, dry_run: bool = False) -> None: + # Default full matchup refresh (not --resume-matchups); stale cells kept on failure. run_script("fetch_stratz_meta.py", dry_run=dry_run) run_script("fetch_hero_items.py", dry_run=dry_run) run_script("fetch_items_meta.py", dry_run=dry_run) diff --git a/web/serve_relations.py b/web/serve_relations.py index a61559e..19f6c70 100644 --- a/web/serve_relations.py +++ b/web/serve_relations.py @@ -27,6 +27,7 @@ import urllib.error from shared.grid import ATTR_ORDER, hero_table from shared.hero_tags import TAG_ORDER, tags_for_hero from shared.http_utils import http_bytes +from shared.matchup_cross import build_baseline, enrich_hero_matchups_cross from shared.paths import ( ABILITY_ICONS, ABILITY_VIDEOS, @@ -37,6 +38,7 @@ from shared.paths import ( ITEM_ICONS, RANK_ICONS, ROOT, + SHARED_DATA, STREAMER_AVATARS, STREAMER_VIDEOS, TEMPLATES_CDN, @@ -62,6 +64,15 @@ PRO_MATCHES_PATH = DATA / "pro_matches.json" STREAMERS_PATH = DATA / "streamers.json" STRATZ_HERO_META_PATH = DATA / "stratz_hero_meta.json" STRATZ_MATCHUP_TOPS_PATH = DATA / "stratz_matchup_tops.json" +OPENDOTA_MATCHUPS_PATH = SHARED_DATA / "matchups.json" +_LEGACY_DATA = ROOT / "data" + + +def _first_existing(*candidates: Path) -> Path | None: + for p in candidates: + if p.is_file(): + return p + return None ATTR_COLS = {"str": 6, "agi": 6, "int": 6, "all": 4} ATTR_LABELS = {"str": "力量", "agi": "敏捷", "int": "智力", "all": "全才"} ABILITY_ICON_URL = ( @@ -364,10 +375,11 @@ def load_stratz_hero_meta() -> dict: "by_hero": {}, "meta_board": {}, } - if not STRATZ_HERO_META_PATH.is_file(): + path = _first_existing(STRATZ_HERO_META_PATH, _LEGACY_DATA / "stratz_hero_meta.json") + if path is None: return empty try: - raw = json.loads(STRATZ_HERO_META_PATH.read_text(encoding="utf-8")) + raw = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return empty if not isinstance(raw, dict): @@ -387,35 +399,109 @@ def load_stratz_hero_meta() -> dict: } +def _load_opendota_matchups() -> dict: + path = _first_existing(OPENDOTA_MATCHUPS_PATH, _LEGACY_DATA / "matchups.json") + if path is None: + return {} + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + return raw if isinstance(raw, dict) else {} + + def load_stratz_matchup_tops() -> dict: - """STRATZ vs/with top lists per hero (see fetch_stratz_meta.py). Web only.""" + """STRATZ vs/with top lists per hero (see fetch_stratz_meta.py). Web only. + + When OpenDota matchups.json is present, each counter/countered row gets a + ``cross`` quality blob (agree / conflict / weak). Observation only. + """ empty: dict = { "fetched_at": None, + "started_at": None, + "finished_at": None, "source": "stratz", "attribution": "https://stratz.com", "take": 0, "match_limit": 0, + "scope": { + "kind": "global_aggregate", + "label_zh": "全局聚合（未按段位 / 分路 / 周过滤）", + }, "note": None, + "stats": {}, + "cross_source": None, "by_hero": {}, } - if not STRATZ_MATCHUP_TOPS_PATH.is_file(): + path = _first_existing( + STRATZ_MATCHUP_TOPS_PATH, _LEGACY_DATA / "stratz_matchup_tops.json" + ) + if path is None: return empty try: - raw = json.loads(STRATZ_MATCHUP_TOPS_PATH.read_text(encoding="utf-8")) + raw = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return empty if not isinstance(raw, dict): return empty - return { + + by_hero_raw = dict(raw.get("by_hero") or {}) + scope = raw.get("scope") if isinstance(raw.get("scope"), dict) else empty["scope"] + out: dict = { "fetched_at": raw.get("fetched_at"), + "started_at": raw.get("started_at"), + "finished_at": raw.get("finished_at"), "source": raw.get("source") or "stratz", "attribution": raw.get("attribution") or empty["attribution"], "take": raw.get("take") or 0, "match_limit": raw.get("match_limit") or 0, + "scope": scope, "note": raw.get("note"), - "by_hero": dict(raw.get("by_hero") or {}), + "stats": dict(raw.get("stats") or {}), + "cross_source": None, + "by_hero": by_hero_raw, } + odota = _load_opendota_matchups() + by_odota = odota.get("by_hero") or {} + if not by_odota: + out["cross_source"] = { + "available": False, + "label_zh": "暂无可靠交叉结论", + "reason": "missing_opendota_matchups", + } + return out + + key_to_id = {h["key"]: int(h["id"]) for h in hero_table()} + baseline = build_baseline(by_odota) + enriched: dict[str, dict] = {} + for key, cell in by_hero_raw.items(): + if not isinstance(cell, dict): + continue + hid = key_to_id.get(key) + if hid is None: + enriched[key] = cell + continue + enriched[key] = enrich_hero_matchups_cross( + cell, + hero_id=hid, + by_odota=by_odota, + baseline=baseline, + ) + out["by_hero"] = enriched + out["cross_source"] = { + "available": True, + "source": "opendota", + "attribution": odota.get("attribution") or "https://opendota.com", + "fetched_at": odota.get("fetched_at"), + "label_zh": "OpenDota 独立交叉（基线校正）", + "manual_review_note": ( + "Dota2ProTracker（7k+ / 职业）仅作人工复核「来源分歧」条目的高分段参考，" + "不进入自动流水线。" + ), + } + return out + def load_hero_abilities() -> dict: """Slim per-hero abilities / Aghs upgrades / talents for the web hero pane.""" @@ -851,7 +937,14 @@ class Handler(BaseHTTPRequestHandler): if "/" in key or "\\" in key or ".." in key: self._json(400, {"error": "bad path"}) return - if not (key.endswith(".mp4") or key.endswith(".webm")): + if not ( + key.endswith(".mp4") + or key.endswith(".webm") + or key.endswith(".jpg") + or key.endswith(".jpeg") + or key.endswith(".webp") + or key.endswith(".png") + ): self._json(400, {"error": "bad path"}) return if key.startswith("_"): @@ -861,10 +954,16 @@ class Handler(BaseHTTPRequestHandler): if not fpath.is_file(): self.send_error(404) return - ctype = ( - "video/webm" if key.endswith(".webm") else "video/mp4" - ) - self._send_file(fpath, ctype) + if key.endswith(".webm"): + ctype = "video/webm" + elif key.endswith(".mp4"): + ctype = "video/mp4" + else: + ctype = mimetypes.guess_type(key)[0] or "image/jpeg" + if ctype.startswith("video/"): + self._send_file(fpath, ctype) + else: + self._send(200, fpath.read_bytes(), ctype) return if path.startswith("/portrait/") or path.startswith("/cdn/"): prefix = "/portrait/" if path.startswith("/portrait/") else "/cdn/"