Files
climperor/web/fetch_hero_abilities.py
T

855 lines
30 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Fetch hero ability notes into data/hero_abilities.json.
Sources:
- Valve herodata?language=schinese (Chinese desc + dispellable)
- OpenDota abilities.json (bkbpierce, English dispellable cross-check)
Usage:
python fetch_hero_abilities.py
python fetch_hero_abilities.py --force
python fetch_hero_abilities.py --delay 0.2
python fetch_hero_abilities.py --icons # also cache ability PNGs
python fetch_hero_abilities.py --icons-only # icons from existing JSON
python fetch_hero_abilities.py --tags-only # recompute mechanic tags only
"""
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 re
import time
import urllib.error
from datetime import datetime, timezone
from shared.grid import hero_table
from shared.hero_tags import ILLUSION_KEYS
from shared.http_utils import download_icons, http_json, write_json_atomic
from shared.paths import ABILITY_ICONS, DATA
from loc_format import format_loc, strip_html
from mechanic_tags import (
QUERY_MECHANIC_ORDER,
apply_mechanic_tags,
merge_tag_overrides,
)
HERODATA_URL = (
"https://www.dota2.com/datafeed/herodata?language=schinese&hero_id={hero_id}"
)
ABILITIES_URL = (
"https://raw.githubusercontent.com/odota/dotaconstants/master/build/abilities.json"
)
ABILITY_ICON_URL = (
"https://cdn.cloudflare.steamstatic.com/apps/dota2/images/dota_react/abilities/{key}.png"
)
OUT = DATA / "hero_abilities.json"
ABILITY_TAG_OVERRIDES = DATA / "ability_tag_overrides.json"
DESC_MAX = 600
UPGRADE_MAX = 500
TALENT_LEVELS = (10, 10, 15, 15, 20, 20, 25, 25)
# Bundled badges — never fetched from CDN (many innate keys 404 there).
SKIP_ICON_KEYS = frozenset({"innate", "talent_tree"})
def _load_ability_tag_overrides() -> dict[str, dict]:
if not ABILITY_TAG_OVERRIDES.is_file():
return {}
try:
raw = json.loads(ABILITY_TAG_OVERRIDES.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return {}
items = raw.get("abilities") or {}
return {str(k): v for k, v in items.items() if isinstance(v, dict)}
def ability_mechanic_tags(ab: dict, overrides: dict[str, dict] | None = None) -> list[str]:
"""Compute applies-* mechanic tags for one ability row."""
key = str(ab.get("key") or "")
blob = " ".join(
[
str(ab.get("name_loc") or ""),
str(ab.get("desc_loc") or ""),
str(ab.get("shard_loc") or ""),
str(ab.get("scepter_loc") or ""),
]
)
auto = apply_mechanic_tags(blob, key=key)
ov = (overrides or {}).get(key)
return merge_tag_overrides(auto, ov, QUERY_MECHANIC_ORDER)
def retag_abilities(by_hero: dict, overrides: dict[str, dict] | None = None) -> int:
"""Write ``tags`` onto every ability; return number of abilities with tags."""
ov = overrides if overrides is not None else _load_ability_tag_overrides()
tagged = 0
for cell in by_hero.values():
if not isinstance(cell, dict):
continue
for ab in cell.get("abilities") or []:
if not isinstance(ab, dict) or not ab.get("key"):
continue
tags = ability_mechanic_tags(ab, ov)
ab["tags"] = tags
if tags:
tagged += 1
return tagged
def _load_talent_overrides() -> dict[str, str]:
"""Load manual name_loc overrides for talents Valve omits bonus data for."""
p = DATA / "talent_overrides.json"
if not p.is_file():
return {}
try:
raw = json.loads(p.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return {}
out: dict[str, str] = {}
items = (raw.get("talents") if isinstance(raw, dict) else None) or {}
if isinstance(items, dict):
for k, v in items.items():
if isinstance(k, str) and isinstance(v, str):
out[k] = v
return out
TALENT_OVERRIDES = _load_talent_overrides()
ABILITY_TAG_OVERRIDES_MAP = _load_ability_tag_overrides()
# Valve dispellable int → label
DISPEL_MAP = {
0: "none",
1: "strong_only",
2: "yes",
3: "no",
}
# DOTA_ABILITY_BEHAVIOR bits (Valve enum)
B_PASSIVE = 1 << 1
B_NO_TARGET = 1 << 2
B_UNIT_TARGET = 1 << 3
B_POINT = 1 << 4
B_CHANNELLED = 1 << 7
B_TOGGLE = 1 << 9
B_AUTOCAST = 1 << 12
B_AURA = 1 << 16
B_VECTOR = 1 << 30
# Valve damage int → 伤害类型 label
DAMAGE_MAP = {1: "物理", 2: "魔法", 4: "纯粹", 8: "纯粹"}
# Valve spell-immunity int → 无视技能免疫 label
IMMUNITY_MAP = {1: "是", 2: "否", 3: "是", 4: "否", 5: "友军是 / 敌军否"}
# Valve target_team int → prefix
TARGET_TEAM_MAP = {1: "友方", 2: "敌方", 3: "双方"}
OD_DISPEL_MAP = {
"yes": "yes",
"no": "no",
"strong dispels only": "strong_only",
}
EVASION_RE = re.compile(r"闪避|evasion|miss chance|落空", re.I)
INVIS_RE = re.compile(r"隐身|invisible|invisibility|渐隐|潜行|shadow walk|fade", re.I)
# Only "creates illusions" style — bare 「幻象不会」mentions are noise.
ILLUSION_CREATE_RE = re.compile(
r"创造幻象|制造幻象|召唤幻象|产生幻象|creates?\s+illusions?|summons?\s+illusions?",
re.I,
)
BREAKABLE_PASSIVE_RE = re.compile(
r"破坏会|Break|被动.*失效|禁用被动|disabled by break",
re.I,
)
MAGIC_NUKE_RE = re.compile(r"魔法伤害|magical damage|魔法伤害", re.I)
DISABLE_RE = re.compile(
r"眩晕|沉默|妖术|缠绕|击飞|定身|stun|silence|hex|root|cyclone",
re.I,
)
def ability_icon_keys(by_hero: dict) -> set[str]:
"""Non-innate ability keys shown in the relations skill row (CDN PNGs).
Innates (and Aghs chips of innates) use bundled ``innate.png`` in the UI —
many of those keys 404 on Steam CDN, so they are skipped here.
"""
keys: set[str] = set()
for cell in by_hero.values():
if not isinstance(cell, dict):
continue
for ab in cell.get("abilities") or []:
if not isinstance(ab, dict):
continue
key = ab.get("key")
if not isinstance(key, str) or not key or key in SKIP_ICON_KEYS:
continue
if ab.get("is_innate"):
continue
keys.add(key)
return keys
def download_ability_icons(
keys: set[str], *, force: bool = False, delay: float = 0.05
) -> tuple[int, int, int]:
"""Cache Steam ability icons. Returns (saved, skipped_existing, fail)."""
return download_icons(
keys, ABILITY_ICON_URL, ABILITY_ICONS,
force=force, delay=delay, skip_keys=SKIP_ICON_KEYS,
)
def truncate(text: str, n: int = DESC_MAX) -> str:
t = re.sub(r"\s+", " ", strip_html(text))
if len(t) <= n:
return t
return t[: n - 1] + "…"
def load_odota_abilities() -> dict[str, dict]:
raw = http_json(ABILITIES_URL)
out: dict[str, dict] = {}
if not isinstance(raw, dict):
return out
for key, row in raw.items():
if isinstance(row, dict):
out[str(key)] = row
return out
def map_dispellable(valve_val: object, od_val: object) -> str:
try:
iv = int(valve_val) if valve_val is not None else None
except (TypeError, ValueError):
iv = None
if iv is not None and iv in DISPEL_MAP:
label = DISPEL_MAP[iv]
if label != "none":
return label
if isinstance(od_val, str):
return OD_DISPEL_MAP.get(od_val.strip().lower(), "none")
return "none"
def _int_of(val: object) -> int:
try:
return int(val) # behavior/damage/immunity arrive as str or int
except (TypeError, ValueError):
return 0
def fmt_num_list(vals: object, *, is_pct: bool = False) -> str:
"""[100, 200, 300] → '100 / 200 / 300'; integral floats shown as ints."""
out: list[str] = []
if not isinstance(vals, list):
return ""
for v in vals:
try:
f = float(v)
except (TypeError, ValueError):
continue
out.append(str(int(f)) if f == int(f) else f"{f:g}")
txt = " / ".join(out)
if txt and is_pct:
txt += "%"
return txt
def fmt_time_value(vals: object) -> str:
"""Cast point / channel time rarely vary per level; collapse repeats.
[0.3, 0.3, 0.3, 0.3] -> '0.3'; [1, 2, 3] -> '1 / 2 / 3'.
"""
txt = fmt_num_list(vals)
parts = [p for p in txt.split(" / ") if p]
if parts and len(set(parts)) == 1:
return parts[0]
return txt
def target_label(behavior: int, is_innate: bool) -> str:
"""技能 targeting label matching dota2.com (点目标 / 单位目标 / 无目标 / 被动…)."""
if behavior & B_VECTOR:
return "矢量目标"
if behavior & B_UNIT_TARGET and behavior & B_POINT:
return "单位或点目标"
if behavior & B_UNIT_TARGET:
return "单位目标"
if behavior & B_POINT:
return "点目标"
if behavior & B_NO_TARGET:
return "无目标"
if behavior & B_TOGGLE:
return "开关"
if behavior & B_AUTOCAST:
return "自动施法"
if behavior & B_CHANNELLED:
return "持续施法"
if behavior & B_AURA:
return "光环"
if behavior & B_PASSIVE or is_innate:
return "被动"
return ""
def affects_label(behavior: int, team: int, type_bits: int) -> str:
"""影响 label (e.g. 敌方单位) for unit-target abilities."""
if not (behavior & B_UNIT_TARGET):
return ""
prefix = TARGET_TEAM_MAP.get(team, "")
if not prefix:
return ""
hero = bool(type_bits & 1)
creep = bool(type_bits & 2)
building = bool(type_bits & 4)
if hero and creep:
noun = "单位"
elif hero:
noun = "英雄"
elif creep:
noun = "普通单位"
else:
noun = "单位"
if building:
noun += " / 建筑"
return prefix + noun
def special_rows(sv: list) -> list[dict]:
"""special_values → [{label, value}] rows with a non-empty Chinese heading."""
out: list[dict] = []
for row in sv:
if not isinstance(row, dict):
continue
# Valve paints effect words with <font color=…>; strip for plain UI text.
heading = strip_html(str(row.get("heading_loc") or "")).rstrip(":").strip()
if not heading:
continue
value = fmt_num_list(
row.get("values_float"), is_pct=bool(row.get("is_percentage"))
)
if not value:
continue
out.append({"label": heading, "value": value})
return out
def _any_nonzero(vals: object) -> bool:
if not isinstance(vals, list):
return False
for v in vals:
try:
if float(v) != 0:
return True
except (TypeError, ValueError):
continue
return False
def fetch_herodata(hero_id: int) -> dict | None:
try:
raw = http_json(HERODATA_URL.format(hero_id=hero_id), timeout=90)
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError):
return None
heroes = (((raw or {}).get("result") or {}).get("data") or {}).get("heroes") or []
if not heroes or not isinstance(heroes[0], dict):
return None
return heroes[0]
def summarize_hero(
hero_key: str,
abilities: list[dict],
hero_tags: list[str],
) -> dict:
disp_yes = 0
strong_only = 0
has_illusion = hero_key in ILLUSION_KEYS or "幻象" in (hero_tags or [])
has_evasion = False
has_invis = False
has_passive_breakable = False
magic_nuke = False
disable_heavy = 0
for ab in abilities:
desc = (ab.get("desc_loc") or "") + " " + (ab.get("name_loc") or "")
d = ab.get("dispellable") or "none"
if d == "yes":
disp_yes += 1
elif d == "strong_only":
strong_only += 1
if ILLUSION_CREATE_RE.search(desc):
has_illusion = True
if EVASION_RE.search(desc):
has_evasion = True
if INVIS_RE.search(desc):
has_invis = True
if BREAKABLE_PASSIVE_RE.search(desc):
has_passive_breakable = True
name_key = (ab.get("key") or "").lower()
# Known breakable passives by ability key fragments
if any(
x in name_key
for x in (
"blur",
"juxtapose",
"dispersion",
"brilliance_aura",
"dragon_blood",
"bash",
"crippling_fear",
"hunter_in_the_night",
"essence_aura",
"mana_shield",
"backtrack",
"reactive_armor",
"feral_impulse",
"natural_order",
"gravekeepers_cloak",
)
):
has_passive_breakable = True
if MAGIC_NUKE_RE.search(desc):
magic_nuke = True
if DISABLE_RE.search(desc):
disable_heavy += 1
return {
"dispellable_buff_count": disp_yes,
"has_strong_only_buff": strong_only > 0,
"has_illusion": has_illusion,
"has_evasion": has_evasion,
"has_invis": has_invis,
"has_passive_breakable": has_passive_breakable,
"magic_nuke": magic_nuke,
"disable_heavy": disable_heavy >= 2,
"mana_dependent": False, # filled lightly below
}
def ability_row(raw: dict, odota: dict[str, dict]) -> dict | None:
name = (raw.get("name") or "").strip()
if not name:
return None
# Valve uses ability name without npc prefix; OD keys match
key = name
if key.startswith("item_"):
return None
od = odota.get(key) or {}
sv = raw.get("special_values") or []
if not isinstance(sv, list):
sv = []
desc_loc = format_loc(raw.get("desc_loc") or "", sv)
shard_loc = format_loc(raw.get("shard_loc") or "", sv, prefer="shard")
scepter_loc = format_loc(raw.get("scepter_loc") or "", sv, prefer="scepter")
disp = map_dispellable(raw.get("dispellable"), od.get("dispellable"))
bkb = od.get("bkbpierce")
if isinstance(bkb, str):
bkb_s = bkb
else:
bkb_s = None
# Trust Valve's explicit upgrade flags. Valve keeps residual
# scepter_loc/shard_loc text even after the upgrade is removed
# (ability_has_scepter False); the old `or bool(scepter_loc)` fallback
# turned that stale text into a phantom upgrade. Clear it instead.
has_scepter = bool(raw.get("ability_has_scepter"))
has_shard = bool(raw.get("ability_has_shard"))
if not has_scepter:
scepter_loc = ""
if not has_shard:
shard_loc = ""
behavior = _int_of(raw.get("behavior"))
is_innate = bool(raw.get("ability_is_innate"))
cast_range = fmt_num_list(raw.get("cast_ranges"))
if not _any_nonzero(raw.get("cast_ranges")):
cast_range = ""
cooldown = fmt_num_list(raw.get("cooldowns"))
mana_cost = fmt_num_list(raw.get("mana_costs"))
cast_point = (
fmt_time_value(raw.get("cast_points"))
if _any_nonzero(raw.get("cast_points"))
else ""
)
channel_time = (
fmt_time_value(raw.get("channel_times"))
if _any_nonzero(raw.get("channel_times"))
else ""
)
return {
"key": key,
"name_loc": (raw.get("name_loc") or key).strip(),
"desc_loc": truncate(desc_loc),
"shard_loc": truncate(shard_loc, UPGRADE_MAX),
"scepter_loc": truncate(scepter_loc, UPGRADE_MAX),
"has_shard": has_shard,
"has_scepter": has_scepter,
"granted_by_shard": bool(raw.get("ability_is_granted_by_shard")),
"granted_by_scepter": bool(raw.get("ability_is_granted_by_scepter")),
"dispellable": disp,
"immunity": raw.get("immunity"),
"bkbpierce": bkb_s,
"is_innate": is_innate,
# Detail pane fields (dota2.com ability panel)
"target_label": target_label(behavior, is_innate),
"affects_label": affects_label(
behavior, _int_of(raw.get("target_team")), _int_of(raw.get("target_type"))
),
"damage_label": DAMAGE_MAP.get(_int_of(raw.get("damage")), ""),
"immunity_label": IMMUNITY_MAP.get(_int_of(raw.get("immunity")), ""),
"cast_range": cast_range,
"cast_point": cast_point,
"channel_time": channel_time,
"cooldown": cooldown,
"mana_cost": mana_cost,
"specials": special_rows(sv),
"lore_loc": strip_html(raw.get("lore_loc") or "").strip(),
"tags": [],
}
def talent_rows(
raw_list: object,
odota: dict[str, dict],
abilities_raw: list | None = None,
) -> list[dict]:
"""8 talents → level 10/15/20/25 left/right (Valve order).
Talent name_loc templates use {s:bonus_<sv_name>} tokens, but the bonus
values are NOT on the talent (its special_values is empty). They live on
each ability's special_values[].bonuses. Valve's data is inconsistent:
- Normal case: bonus.name == talent key, sv_name == token's sv name.
- tinker_5: bonus.name != talent key (uses '..._rearm_channel_time'),
but token sv name == actual sv name → resolve by sv name fallback.
- invoker forged_spirit: bonus found by talent key, but sv_name
('armor_per_attack') != token sv name ('armor_removed') → pair the
single bonus to the single token regardless of name.
The +/- sign and unit (秒/%) are already in name_loc, so we fill the raw
bonus.value regardless of operation (ADD=0, SUBTRACT=2 dominate).
"""
if not isinstance(raw_list, list):
return []
# talent_key -> [(sv_name, bonus_value), ...]
bonus_by_talent: dict[str, list[tuple[str, float]]] = {}
# sv_name -> [bonus_value, ...] (fallback when bonus.name != talent key)
sv_bonus_by_name: dict[str, list[float]] = {}
for ab in abilities_raw or []:
if not isinstance(ab, dict):
continue
for sv in ab.get("special_values") or []:
if not isinstance(sv, dict):
continue
sv_name = str(sv.get("name") or "").strip()
if not sv_name:
continue
for b in sv.get("bonuses") or []:
if not isinstance(b, dict):
continue
tkey = str(b.get("name") or "").strip()
try:
bv = float(b.get("value"))
except (TypeError, ValueError):
continue
if tkey:
bonus_by_talent.setdefault(tkey, []).append((sv_name, bv))
sv_bonus_by_name.setdefault(sv_name, []).append(bv)
out: list[dict] = []
for i, raw in enumerate(raw_list[:8]):
if not isinstance(raw, dict):
continue
key = (raw.get("name") or "").strip()
if not key:
continue
name_loc_raw = raw.get("name_loc") or ""
# sv names the template expects (from {s:bonus_<X>} tokens).
expected = re.findall(r"\{s:bonus_([A-Za-z0-9_]+)\}", name_loc_raw)
keyed = bonus_by_talent.get(key, [])
# Map each expected sv name to a bonus value, with two fallbacks.
value_by_expected: dict[str, float] = {}
for exp in expected:
# 1. exact sv_name match among this talent's keyed bonuses
val = next((v for sn, v in keyed if sn == exp), None)
# 2. single token + single keyed bonus → pair regardless of name
if val is None and len(keyed) == 1 and len(expected) == 1:
val = keyed[0][1]
# 3. fallback: any bonus on a sv with this name across abilities
if val is None:
cands = sv_bonus_by_name.get(exp, [])
if cands:
val = cands[0]
if val is not None:
value_by_expected[exp] = val
pseudo_sv: list[dict] = [
{"name": f"bonus_{exp}", "values_float": [val]}
for exp, val in value_by_expected.items()
]
own_sv = raw.get("special_values") or []
if isinstance(own_sv, list):
pseudo_sv.extend(s for s in own_sv if isinstance(s, dict))
name_loc = format_loc(name_loc_raw, pseudo_sv)
if not name_loc:
od = odota.get(key) or {}
name_loc = format_loc(str(od.get("dname") or key), [])
# Drop unresolved empty placeholder leftovers like "+ %"
name_loc = re.sub(r"\+\s*%", "+", name_loc).strip()
# Manual override for talents Valve omits bonus data for (still '?').
if key in TALENT_OVERRIDES and "?" in name_loc:
name_loc = TALENT_OVERRIDES[key]
level = TALENT_LEVELS[i] if i < len(TALENT_LEVELS) else 10
side = "left" if i % 2 == 0 else "right"
out.append(
{
"key": key,
"name_loc": truncate(name_loc, 160),
"level": level,
"side": side,
}
)
return out
def main() -> None:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--delay", type=float, default=0.2)
ap.add_argument("--out", type=Path, default=OUT)
ap.add_argument("--force", action="store_true")
ap.add_argument(
"--resummarize",
action="store_true",
help="Recompute summary from cached abilities without refetching herodata",
)
ap.add_argument(
"--icons",
action="store_true",
help="Also cache Steam CDN ability icons into assets/ability_icons/",
)
ap.add_argument(
"--icons-only",
action="store_true",
help="Only download ability icons from existing hero_abilities.json",
)
ap.add_argument(
"--icons-force",
action="store_true",
help="Re-download ability icons even when a local PNG already exists",
)
ap.add_argument(
"--tags-only",
action="store_true",
help="Recompute ability mechanic tags from cached JSON without refetching",
)
args = ap.parse_args()
heroes = hero_table()
heroes_by_key = {h["key"]: h for h in heroes}
if args.tags_only:
if not args.out.is_file():
raise SystemExit(f"missing {args.out}; run without --tags-only first")
prev = json.loads(args.out.read_text(encoding="utf-8"))
by_hero = dict(prev.get("by_hero") or {})
overrides = _load_ability_tag_overrides()
tagged = retag_abilities(by_hero, overrides)
meta = dict(prev.get("meta") or {})
meta["fetched_at"] = datetime.now(timezone.utc).isoformat()
meta["tags_only"] = True
meta["mechanic_tag_order"] = list(QUERY_MECHANIC_ORDER)
payload = {"meta": meta, "by_hero": by_hero}
write_json_atomic(args.out, payload)
print(
f"retag done: {len(by_hero)} heroes, {tagged} abilities with tags -> {args.out}",
flush=True,
)
return
if args.icons_only:
if not args.out.is_file():
raise SystemExit(f"missing {args.out}; run without --icons-only first")
prev = json.loads(args.out.read_text(encoding="utf-8"))
by_hero = dict(prev.get("by_hero") or {})
keys = ability_icon_keys(by_hero)
print(
f"downloading {len(keys)} ability icons from Steam CDN -> {ABILITY_ICONS}",
flush=True,
)
saved, skipped, fail = download_ability_icons(
keys, force=args.icons_force, delay=min(args.delay, 0.1)
)
print(
f"icons done: saved={saved} skipped={skipped} fail={fail} -> {ABILITY_ICONS}",
flush=True,
)
return
if args.resummarize:
if not args.out.is_file():
raise SystemExit(f"missing {args.out}; run without --resummarize first")
prev = json.loads(args.out.read_text(encoding="utf-8"))
by_hero = dict(prev.get("by_hero") or {})
for key, cell in by_hero.items():
if not isinstance(cell, dict):
continue
h = heroes_by_key.get(key) or {}
tags = list(h.get("tags") or [])
abilities = list(cell.get("abilities") or [])
summary = summarize_hero(key, abilities, tags)
manaish = sum(
1
for ab in abilities
if re.search(r"魔法|mana", (ab.get("desc_loc") or ""), re.I)
)
summary["mana_dependent"] = manaish >= 2 or key in {
"medusa",
"obsidian_destroyer",
"storm_spirit",
"leshrac",
}
cell["summary"] = summary
payload = {
"meta": {
"source": "valve+opendota",
"attribution": "https://www.dota2.com ; https://www.opendota.com",
"fetched_at": datetime.now(timezone.utc).isoformat(),
"heroes": len(by_hero),
"resummarized": True,
},
"by_hero": by_hero,
}
write_json_atomic(args.out, payload)
illus = sum(1 for c in by_hero.values() if (c.get("summary") or {}).get("has_illusion"))
print(f"resummarized {len(by_hero)} heroes (illusion={illus}) -> {args.out}", flush=True)
return
print("loading OpenDota abilities.json...", flush=True)
odota = load_odota_abilities()
print(f" {len(odota)} abilities", flush=True)
by_hero: dict[str, dict] = {}
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():
# Require talents + detail-pane fields (post-upgrade schema).
abs_prev = cell.get("abilities") if isinstance(cell, dict) else None
abs_list = [a for a in (abs_prev or []) if isinstance(a, dict)]
# Require every schema generation's fields; missing any of them
# invalidates the cache so the hero is refetched with new data.
if (
isinstance(cell, dict)
and abs_prev is not None
and cell.get("talents") is not None
and any("specials" in a for a in abs_list)
and any("cast_point" in a for a in abs_list)
):
by_hero[str(k)] = cell
print(f"resuming with {len(by_hero)} heroes cached", flush=True)
except (OSError, json.JSONDecodeError):
pass
pending = [h for h in heroes if h["key"] not in by_hero]
print(f"fetching herodata for {len(pending)} / {len(heroes)} heroes...", flush=True)
for n, h in enumerate(pending, start=1):
key = h["key"]
hid = int(h["id"])
tags = list(h.get("tags") or [])
data = fetch_herodata(hid)
if data is None:
print(f" [{n}/{len(pending)}] {key} failed", flush=True)
time.sleep(max(args.delay, 0.1) * 2)
continue
abs_raw = data.get("abilities") or []
abilities: list[dict] = []
if isinstance(abs_raw, list):
for row in abs_raw:
if not isinstance(row, dict):
continue
ab = ability_row(row, odota)
if ab:
ab["tags"] = ability_mechanic_tags(ab, ABILITY_TAG_OVERRIDES_MAP)
abilities.append(ab)
talents = talent_rows(data.get("talents"), odota, abs_raw)
summary = summarize_hero(key, abilities, tags)
manaish = sum(
1
for ab in abilities
if re.search(r"魔法|mana", (ab.get("desc_loc") or ""), re.I)
)
summary["mana_dependent"] = manaish >= 2 or key in {
"medusa",
"obsidian_destroyer",
"storm_spirit",
"leshrac",
}
by_hero[key] = {
"id": hid,
"abilities": abilities,
"talents": talents,
"summary": summary,
}
n_up = sum(1 for a in abilities if a.get("shard_loc") or a.get("scepter_loc"))
print(
f" [{n}/{len(pending)}] {key}: {len(abilities)} abilities "
f"upgrades={n_up} talents={len(talents)} "
f"dispel={summary['dispellable_buff_count']}",
flush=True,
)
# Checkpoint every 10 heroes (or on the last one) so a crash only
# loses a small slice, without writing the full file on every hero.
if n % 10 == 0 or n == len(pending):
args.out.parent.mkdir(parents=True, exist_ok=True)
payload = {
"meta": {
"source": "valve+opendota",
"attribution": "https://www.dota2.com ; https://www.opendota.com",
"fetched_at": datetime.now(timezone.utc).isoformat(),
},
"by_hero": by_hero,
}
write_json_atomic(args.out, payload)
time.sleep(args.delay)
tagged = retag_abilities(by_hero, ABILITY_TAG_OVERRIDES_MAP)
payload = {
"meta": {
"source": "valve+opendota",
"attribution": "https://www.dota2.com ; https://www.opendota.com",
"fetched_at": datetime.now(timezone.utc).isoformat(),
"heroes": len(by_hero),
"mechanic_tag_order": list(QUERY_MECHANIC_ORDER),
"abilities_with_tags": tagged,
},
"by_hero": by_hero,
}
write_json_atomic(args.out, payload)
print(f"done: {len(by_hero)} heroes -> {args.out}", flush=True)
if args.icons:
keys = ability_icon_keys(by_hero)
print(
f"downloading {len(keys)} ability icons from Steam CDN -> {ABILITY_ICONS}",
flush=True,
)
saved, skipped, fail = download_ability_icons(
keys, force=args.icons_force, delay=min(args.delay, 0.1)
)
print(
f"icons done: saved={saved} skipped={skipped} fail={fail} -> {ABILITY_ICONS}",
flush=True,
)
if __name__ == "__main__":
main()