v0.2.0: relations preview, item shop, abilities, overlay recommend, GSI enhancements
- Add relations/item/abilities preview (serve_relations.py + web/relations/) - Add fetch scripts: hero_items, item_shop, items_meta, hero_abilities, ability_videos, patches, stratz, matchups, portraits - Add overlay.py (role tags + Top-3 cyan marks), recommend.py - Add http_utils.py, loc_format.py, hero_tags.py, item_fears.py - GSI: full payload JSONL dump, foreground window detection - Drop real template library; CDN-only matching - Update docs: CHANGELOG 0.2.0, DESIGN config table, AGENTS module table - .gitignore: exclude large regenerable assets (icons/portraits/videos)
This commit is contained in:
voson
@@ -0,0 +1,314 @@
|
||||
"""Fetch OpenDota hero item popularity into data/hero_items.json.
|
||||
|
||||
Stores a flat Top-N of core finished items per hero (no start/early/mid/late).
|
||||
Item names (Chinese): Valve dota2.com datafeed (schinese).
|
||||
Item icons: Steam CDN dota_react/items/{key}.png → assets/item_icons/.
|
||||
|
||||
Usage:
|
||||
python fetch_hero_items.py
|
||||
python fetch_hero_items.py --force
|
||||
python fetch_hero_items.py --delay 0.3 --skip-icons
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import time
|
||||
import urllib.error
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from common import DATA, ITEM_ICONS
|
||||
from grid import hero_table
|
||||
from http_utils import download_icons, http_json, load_itemlist_zh
|
||||
|
||||
OPENDOTA = "https://api.opendota.com/api"
|
||||
ITEMS_URL = (
|
||||
"https://raw.githubusercontent.com/odota/dotaconstants/master/build/items.json"
|
||||
)
|
||||
ICON_URL = (
|
||||
"https://cdn.cloudflare.steamstatic.com/apps/dota2/images/dota_react/items/{key}.png"
|
||||
)
|
||||
OUT = DATA / "hero_items.json"
|
||||
|
||||
PHASE_API = (
|
||||
"start_game_items",
|
||||
"early_game_items",
|
||||
"mid_game_items",
|
||||
"late_game_items",
|
||||
)
|
||||
# Crafted boots / mid-game cores (phase, treads, vanguard); skip bracer/wand.
|
||||
MIN_CREATED_COST = 1400
|
||||
# Non-crafted shop cores that are still finished pickups (blink / 跳刀).
|
||||
ALWAYS_CORE = frozenset({"blink", "aghanims_shard"})
|
||||
# Aghanim's Blessing (ultimate_scepter_2, id 271) is the activated form of
|
||||
# Aghanim's Scepter (ultimate_scepter, id 108). Merge blessing counts into
|
||||
# scepter so heroes show "阿哈利姆神杖" once, never "阿哈利姆福佑".
|
||||
BLESSING_TO_SCEPTER = {271: 108}
|
||||
TOP_N = 12
|
||||
# If any upgrade's popularity >= this fraction of the intermediate's, hide it
|
||||
# (e.g. AM yasha→manta). If upgrades are rare (Jugg yasha only), keep it.
|
||||
UPGRADE_RATIO = 0.35
|
||||
|
||||
|
||||
def load_item_catalog() -> tuple[dict[int, dict], dict[int, list[int]]]:
|
||||
"""Return (id -> meta, id -> upgrade item ids that list it as a component)."""
|
||||
raw = http_json(ITEMS_URL)
|
||||
zh = load_itemlist_zh()
|
||||
key_to_id: dict[str, int] = {}
|
||||
out: dict[int, dict] = {}
|
||||
for key, row in raw.items():
|
||||
if not isinstance(row, dict):
|
||||
continue
|
||||
iid = row.get("id")
|
||||
if iid is None:
|
||||
continue
|
||||
key_s = str(key)
|
||||
if key_s.startswith("recipe_"):
|
||||
continue
|
||||
dname = str(row.get("dname") or key_s)
|
||||
try:
|
||||
cost = int(row.get("cost") or 0)
|
||||
except (TypeError, ValueError):
|
||||
cost = 0
|
||||
comps = [str(c) for c in (row.get("components") or []) if c]
|
||||
out[int(iid)] = {
|
||||
"key": key_s,
|
||||
"dname": dname,
|
||||
"name_loc": zh.get(int(iid)) or dname,
|
||||
"created": bool(row.get("created")),
|
||||
"cost": cost,
|
||||
"tier": row.get("tier"),
|
||||
"qual": row.get("qual"),
|
||||
"components": comps,
|
||||
}
|
||||
key_to_id[key_s] = int(iid)
|
||||
|
||||
upgrades_of: dict[int, list[int]] = {iid: [] for iid in out}
|
||||
for iid, meta in out.items():
|
||||
for comp_key in meta["components"]:
|
||||
cid = key_to_id.get(comp_key)
|
||||
if cid is None or cid == iid:
|
||||
continue
|
||||
upgrades_of.setdefault(cid, []).append(iid)
|
||||
return out, upgrades_of
|
||||
|
||||
|
||||
def is_core_finished(meta: dict) -> bool:
|
||||
"""Core finished items — skip consumables, neutrals, secret-shop parts, cheap early."""
|
||||
key = meta.get("key") or ""
|
||||
if meta.get("tier") is not None:
|
||||
return False
|
||||
if meta.get("qual") == "consumable":
|
||||
return False
|
||||
cost = int(meta.get("cost") or 0)
|
||||
if cost <= 0:
|
||||
return False
|
||||
if key in ALWAYS_CORE:
|
||||
return True
|
||||
# Only recipe-assembled items (phase boots, bfury, bkb, …).
|
||||
if not meta.get("created"):
|
||||
return False
|
||||
return cost >= MIN_CREATED_COST
|
||||
|
||||
|
||||
def merge_phase_counts(raw: dict) -> dict[int, int]:
|
||||
"""Max count per item id across OpenDota phases."""
|
||||
merged: dict[int, int] = {}
|
||||
for api_key in PHASE_API:
|
||||
counts = raw.get(api_key) or {}
|
||||
if not isinstance(counts, dict):
|
||||
continue
|
||||
for sid, cnt in counts.items():
|
||||
try:
|
||||
iid = int(sid)
|
||||
c = int(cnt)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
if c <= 0:
|
||||
continue
|
||||
prev = merged.get(iid, 0)
|
||||
if c > prev:
|
||||
merged[iid] = c
|
||||
return merged
|
||||
|
||||
|
||||
def fold_blessing_into_scepter(counts: dict[int, int]) -> dict[int, int]:
|
||||
"""Merge Aghanim's Blessing (id 271) counts into Aghanim's Scepter (id 108).
|
||||
|
||||
Blessing is the activated/synthesized form of Scepter; OpenDota records
|
||||
them under separate ids. Take max (not sum): the same game can register
|
||||
both the purchase and the activation, which would double-count.
|
||||
"""
|
||||
out = dict(counts)
|
||||
for bid, sid in BLESSING_TO_SCEPTER.items():
|
||||
b = out.pop(bid, 0)
|
||||
s = out.get(sid, 0)
|
||||
if b or s:
|
||||
out[sid] = max(s, b)
|
||||
return out
|
||||
|
||||
|
||||
def is_terminal_for_hero(
|
||||
iid: int,
|
||||
count: int,
|
||||
all_counts: dict[int, int],
|
||||
upgrades_of: dict[int, list[int]],
|
||||
ratio: float = UPGRADE_RATIO,
|
||||
) -> bool:
|
||||
"""True if hero rarely upgrades this item further (keep yasha for Jugg, drop for AM)."""
|
||||
parents = upgrades_of.get(iid) or []
|
||||
if not parents:
|
||||
return True
|
||||
threshold = max(1, int(count * ratio))
|
||||
for pid in parents:
|
||||
if all_counts.get(pid, 0) >= threshold:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def core_from_popularity(
|
||||
raw: dict,
|
||||
catalog: dict[int, dict],
|
||||
upgrades_of: dict[int, list[int]],
|
||||
n: int = TOP_N,
|
||||
) -> list[dict]:
|
||||
"""Merge phases → core items → drop intermediates that this hero upgrades."""
|
||||
all_counts = fold_blessing_into_scepter(merge_phase_counts(raw))
|
||||
merged: dict[int, int] = {}
|
||||
for iid, c in all_counts.items():
|
||||
meta = catalog.get(iid)
|
||||
if meta is None or not is_core_finished(meta):
|
||||
continue
|
||||
if not is_terminal_for_hero(iid, c, all_counts, upgrades_of):
|
||||
continue
|
||||
merged[iid] = c
|
||||
ranked = sorted(merged.items(), key=lambda t: (-t[1], t[0]))
|
||||
return [{"id": iid, "count": c} for iid, c in ranked[:n]]
|
||||
|
||||
|
||||
def fetch_popularity(hero_id: int) -> dict:
|
||||
return http_json(f"{OPENDOTA}/heroes/{hero_id}/itemPopularity") # type: ignore[return-value]
|
||||
|
||||
|
||||
def build_payload(by_hero: dict, catalog: dict[int, dict], used_ids: set[int]) -> dict:
|
||||
items_out = {
|
||||
str(iid): {
|
||||
"key": catalog[iid]["key"],
|
||||
"dname": catalog[iid]["dname"],
|
||||
"name_loc": catalog[iid]["name_loc"],
|
||||
}
|
||||
for iid in sorted(used_ids)
|
||||
if iid in catalog
|
||||
}
|
||||
return {
|
||||
"meta": {
|
||||
"source": "opendota+valve",
|
||||
"attribution": "https://www.opendota.com ; https://www.dota2.com",
|
||||
"fetched_at": datetime.now(timezone.utc).isoformat(),
|
||||
"top_n": TOP_N,
|
||||
"mode": "core_finished_terminal",
|
||||
"upgrade_ratio": UPGRADE_RATIO,
|
||||
"icons": "steam_cdn",
|
||||
},
|
||||
"items": items_out,
|
||||
"by_hero": by_hero,
|
||||
}
|
||||
|
||||
|
||||
def collect_used_ids(by_hero: dict) -> set[int]:
|
||||
used: set[int] = set()
|
||||
for cell in by_hero.values():
|
||||
if not isinstance(cell, list):
|
||||
continue
|
||||
for row in cell:
|
||||
if isinstance(row, dict) and "id" in row:
|
||||
used.add(int(row["id"]))
|
||||
return used
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser(description=__doc__)
|
||||
ap.add_argument("--delay", type=float, default=0.3)
|
||||
ap.add_argument("--out", type=Path, default=OUT)
|
||||
ap.add_argument("--skip-icons", action="store_true")
|
||||
ap.add_argument("--force-icons", action="store_true")
|
||||
ap.add_argument(
|
||||
"--force",
|
||||
action="store_true",
|
||||
help="Refetch all heroes (ignore existing by_hero cache)",
|
||||
)
|
||||
args = ap.parse_args()
|
||||
|
||||
print("fetching item catalog (dotaconstants + Valve schinese)...", flush=True)
|
||||
catalog, upgrades_of = load_item_catalog()
|
||||
print(f" {len(catalog)} items", flush=True)
|
||||
|
||||
heroes = hero_table()
|
||||
by_key = {h["key"]: int(h["id"]) for h in heroes}
|
||||
|
||||
by_hero: dict[str, list] = {}
|
||||
if args.out.is_file() and not args.force:
|
||||
try:
|
||||
prev = json.loads(args.out.read_text(encoding="utf-8"))
|
||||
for k, cell in (prev.get("by_hero") or {}).items():
|
||||
if isinstance(cell, list):
|
||||
by_hero[str(k)] = cell
|
||||
print(f"resuming with {len(by_hero)} heroes already cached", flush=True)
|
||||
except (OSError, json.JSONDecodeError):
|
||||
pass
|
||||
elif args.force:
|
||||
print("force: refetching all heroes", flush=True)
|
||||
|
||||
pending = [k for k in sorted(by_key) if k not in by_hero]
|
||||
print(f"fetching {len(pending)} / {len(by_key)} heroes -> {args.out}", flush=True)
|
||||
|
||||
used_ids = collect_used_ids(by_hero)
|
||||
|
||||
for n, key in enumerate(pending, start=1):
|
||||
hid = by_key[key]
|
||||
try:
|
||||
raw = fetch_popularity(hid)
|
||||
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, TypeError) as e:
|
||||
print(f" [{n}/{len(pending)}] {key} ({hid}) failed: {e}", flush=True)
|
||||
time.sleep(max(args.delay, 0.1) * 2)
|
||||
continue
|
||||
|
||||
core = core_from_popularity(raw, catalog, upgrades_of)
|
||||
by_hero[key] = core
|
||||
used_ids.update(int(row["id"]) for row in core if isinstance(row, dict) and "id" in row)
|
||||
print(f" [{n}/{len(pending)}] {key}: {len(core)} core items", flush=True)
|
||||
|
||||
args.out.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.out.write_text(
|
||||
json.dumps(build_payload(by_hero, catalog, used_ids), ensure_ascii=False, indent=2)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
time.sleep(args.delay)
|
||||
|
||||
used_ids = collect_used_ids(by_hero)
|
||||
payload = build_payload(by_hero, catalog, used_ids)
|
||||
args.out.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.out.write_text(
|
||||
json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
if not args.skip_icons:
|
||||
keys = {catalog[iid]["key"] for iid in used_ids if iid in catalog}
|
||||
print(f"downloading {len(keys)} item icons from Steam CDN -> {ITEM_ICONS}", flush=True)
|
||||
saved, skipped, fail = download_icons(keys, ICON_URL, ITEM_ICONS, force=args.force_icons)
|
||||
print(f" icons saved={saved} skipped={skipped} fail={fail}", flush=True)
|
||||
|
||||
missing = [k for k in by_key if k not in by_hero]
|
||||
print(
|
||||
f"done: {len(by_hero)} heroes, {len(payload['items'])} items"
|
||||
+ (f", {len(missing)} still missing" if missing else ""),
|
||||
flush=True,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user