Files
climperor/fetch_items_meta.py
T
voson a91789b72f 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)
2026-07-27 11:56:51 +08:00

481 lines
16 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 shop item descriptions and mechanism tags into data/items_meta.json.
Sources:
- OpenDota items.json (structure, EN ability text)
- Valve itemlist / itemdata (schinese names + descriptions)
- data/item_tag_overrides.json (manual add/remove)
Usage:
python fetch_items_meta.py
python fetch_items_meta.py --force
python fetch_items_meta.py --reformat-desc --skip-icons
python fetch_items_meta.py --skip-icons --delay 0.2
"""
from __future__ import annotations
import argparse
import json
import re
import time
import urllib.error
from datetime import datetime, timezone
from pathlib import Path
from common import DATA, ITEM_ICONS
from http_utils import download_icons, http_json, load_itemlist
from loc_format import HAS_PLACEHOLDER, format_loc
ITEMS_URL = (
"https://raw.githubusercontent.com/odota/dotaconstants/master/build/items.json"
)
ITEMDATA_URL = (
"https://www.dota2.com/datafeed/itemdata?language={lang}&item_id={item_id}"
)
ICON_URL = (
"https://cdn.cloudflare.steamstatic.com/apps/dota2/images/dota_react/items/{key}.png"
)
OUT = DATA / "items_meta.json"
OVERRIDES = DATA / "item_tag_overrides.json"
MIN_CREATED_COST = 1400
ALWAYS_CORE = frozenset({"blink", "aghanims_shard", "gem", "dust", "ghost"})
TAG_ORDER = [
"basic_dispel",
"strong_dispel",
"mana_burn",
"break",
"true_strike",
"illusion_clear",
"magic_immune",
"spell_block",
"spell_reflect",
"invis_detect",
"invis_break",
"heal_reduce",
"armor_reduce",
"silence",
"hex",
"root",
]
# key -> extra tags from curated whitelist (also covered by overrides)
ILLUSION_CLEAR_KEYS = frozenset({
"bfury",
"radiance",
"maelstrom",
"mjollnir",
"gungir",
})
H1_RE = re.compile(
r"<h1>\s*(主动|被动|使用|开关|升级|Active|Passive|Use|Toggle|Upgrade)\s*[:]?\s*([^<]*)</h1>",
re.I,
)
DISPEL_TYPE_RE = re.compile(
r"(?:驱散类型|Dispel\s*Type)\s*[:]\s*([^<\n]+)",
re.I,
)
def is_core_finished(meta: dict) -> bool:
key = meta.get("key") or ""
if meta.get("tier") is not None:
return False
if meta.get("qual") == "consumable" and key not in ALWAYS_CORE:
return False
cost = int(meta.get("cost") or 0)
if cost <= 0 and key not in ALWAYS_CORE:
return False
if key in ALWAYS_CORE:
return True
if not meta.get("created"):
return False
return cost >= MIN_CREATED_COST
def load_odota_catalog() -> dict[int, dict]:
raw = http_json(ITEMS_URL)
zh_index = load_itemlist()
out: dict[int, dict] = {}
for key, row in raw.items():
if not isinstance(row, dict):
continue
key_s = str(key)
if key_s.startswith("recipe_"):
continue
iid = row.get("id")
if iid is None:
continue
try:
cost = int(row.get("cost") or 0)
except (TypeError, ValueError):
cost = 0
zh = zh_index.get(int(iid)) or {}
abilities = row.get("abilities") or []
if not isinstance(abilities, list):
abilities = []
en_bits = []
for ab in abilities:
if isinstance(ab, dict):
en_bits.append(str(ab.get("description") or ""))
en_bits.append(str(ab.get("title") or ""))
out[int(iid)] = {
"key": key_s,
"dname": str(row.get("dname") or key_s),
"name_loc": zh.get("name_loc") or str(row.get("dname") or key_s),
"created": bool(row.get("created")),
"cost": cost,
"tier": row.get("tier"),
"qual": row.get("qual"),
"od_abilities": abilities,
"en_text": " ".join(en_bits),
"od_dispellable": row.get("dispellable"),
}
return out
def fetch_itemdata(item_id: int, lang: str = "schinese") -> dict | None:
try:
raw = http_json(ITEMDATA_URL.format(lang=lang, item_id=item_id))
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError):
return None
items = (((raw or {}).get("result") or {}).get("data") or {}).get("items") or []
if not items or not isinstance(items[0], dict):
return None
return items[0]
def parse_ability_kinds(desc_loc: str, od_abilities: list) -> list[str]:
kinds: set[str] = set()
for ab in od_abilities or []:
if not isinstance(ab, dict):
continue
t = str(ab.get("type") or "").lower()
if t in ("active", "passive", "use", "toggle", "upgrade"):
kinds.add(t)
for m in H1_RE.finditer(desc_loc or ""):
label = m.group(1).lower()
mapping = {
"主动": "active",
"被动": "passive",
"使用": "use",
"开关": "toggle",
"升级": "upgrade",
"active": "active",
"passive": "passive",
"use": "use",
"toggle": "toggle",
"upgrade": "upgrade",
}
if label in mapping:
kinds.add(mapping[label])
return sorted(kinds)
def extract_dispel_from_text(text: str) -> set[str]:
tags: set[str] = set()
if not text:
return tags
for m in DISPEL_TYPE_RE.finditer(text):
chunk = m.group(1).lower()
if "强" in chunk or "strong" in chunk:
tags.add("strong_dispel")
if "基础" in chunk or "basic" in chunk:
tags.add("basic_dispel")
if "强" not in chunk and "strong" not in chunk and ("驱散" in chunk or "dispel" in chunk):
tags.add("basic_dispel")
low = text.lower()
if "strong dispel" in low or "强驱散" in text:
tags.add("strong_dispel")
if "basic dispel" in low or "基础驱散" in text:
tags.add("basic_dispel")
# Explicit purge / 阻止 (diffusal) — avoid bare 「幻象不会」lines
if re.search(
r"(?:施加|应用|造成)?.{0,8}(?:基础)?驱散|applies?\s+(?:a\s+)?(?:basic\s+)?dispel|"
r"\bpurge\b|阻止敌|Inhibit",
text,
re.I,
):
if "strong" in low or "强驱散" in text:
tags.add("strong_dispel")
else:
tags.add("basic_dispel")
return tags
def auto_tags(key: str, desc_zh: str, desc_en: str, en_od: str, od_dispellable: object) -> list[str]:
blob = " ".join([desc_zh or "", desc_en or "", en_od or "", key])
low = blob.lower()
tags: set[str] = set()
tags |= extract_dispel_from_text(blob)
if od_dispellable and str(od_dispellable).lower() in ("yes", "both"):
# Item itself being dispellable is not "applies dispel"
pass
if re.search(
r"mana burn|burn(?:s|ed)?\s+\d*\s*mana|燃烧.{0,6}魔法|破法|mana.?burn|反馈",
blob,
re.I,
):
tags.add("mana_burn")
if re.search(r"(?:施加|造成|应用).{0,6}破坏|\bbreaks?\b(?:\s+passives?)?|Break\b", blob):
tags.add("break")
if key == "monkey_king_bar" or re.search(
r"true strike|无视闪避|必定命中|attacks? cannot miss",
blob,
re.I,
):
tags.add("true_strike")
if key in ILLUSION_CLEAR_KEYS or re.search(
r"(?:对|伤害).{0,8}幻象|bonus damage to illusions|cleave|分裂攻击|闪电链|chain lightning",
blob,
re.I,
):
tags.add("illusion_clear")
if key == "black_king_bar" or re.search(
r"magic immunity|spell immunity|魔法免疫",
blob,
re.I,
):
tags.add("magic_immune")
if key == "sphere" or re.search(r"spell block|法术格挡", blob, re.I):
tags.add("spell_block")
if key == "lotus_orb" or re.search(
r"re-?casts? most targeted|echo shell|反射|回到施法者|回施",
blob,
re.I,
):
tags.add("spell_reflect")
if key in ("gem", "dust") or re.search(r"true sight|真实视域", blob, re.I):
tags.add("invis_detect")
if key in ("silver_edge", "invis_sword") or re.search(
r"破隐|break(?:s)? invis",
blob,
re.I,
):
tags.add("invis_break")
if key == "spirit_vessel" or re.search(
r"heal(?:ing)? reduction|减少治疗|治疗降低|回复降低",
blob,
re.I,
):
tags.add("heal_reduce")
if re.search(r"armor reduction|reduce(?:s)? armor|减甲|降低护甲", blob, re.I):
tags.add("armor_reduce")
if re.search(r"\bsilence[sd]?\b|沉默", blob, re.I):
tags.add("silence")
if re.search(r"\bhex(?:es|ed)?\b|变羊|妖术", blob, re.I):
tags.add("hex")
if re.search(r"\broot(?:s|ed)?\b|缠绕|定身", blob, re.I):
tags.add("root")
return [t for t in TAG_ORDER if t in tags]
def load_overrides() -> dict[str, dict]:
if not OVERRIDES.is_file():
return {}
try:
raw = json.loads(OVERRIDES.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return {}
items = raw.get("items") or {}
return {str(k): v for k, v in items.items() if isinstance(v, dict)}
def apply_overrides(key: str, tags: list[str], overrides: dict[str, dict]) -> list[str]:
row = overrides.get(key) or {}
s = set(tags)
for t in row.get("add") or []:
if isinstance(t, str):
s.add(t)
for t in row.get("remove") or []:
if isinstance(t, str):
s.discard(t)
# Keep known order, then any extras
ordered = [t for t in TAG_ORDER if t in s]
extras = sorted(s - set(TAG_ORDER))
return ordered + extras
def main() -> None:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--delay", type=float, default=0.15)
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 itemdata even if cached in existing out file",
)
ap.add_argument(
"--reformat-desc",
action="store_true",
help="Refetch special_values and fill %%token%% in cached descriptions",
)
args = ap.parse_args()
print("loading OpenDota items + Valve itemlist...", flush=True)
catalog = load_odota_catalog()
candidates = {
iid: meta
for iid, meta in catalog.items()
if is_core_finished(meta) or meta["key"] in ALWAYS_CORE
}
print(f" {len(candidates)} finished shop items", flush=True)
cached: 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, row in (prev.get("items") or {}).items():
if isinstance(row, dict) and row.get("desc_loc") is not None:
cached[str(k)] = row
print(f"resuming with {len(cached)} cached itemdata rows", flush=True)
except (OSError, json.JSONDecodeError):
pass
overrides = load_overrides()
items_out: dict[str, dict] = {}
pending = []
for iid in sorted(candidates):
sid = str(iid)
if args.force or sid not in cached:
pending.append(iid)
continue
if args.reformat_desc:
prev = cached[sid]
if HAS_PLACEHOLDER.search(prev.get("desc_loc") or "") or HAS_PLACEHOLDER.search(
prev.get("desc_en") or ""
):
pending.append(iid)
print(f"fetching/reformatting itemdata for {len(pending)} items...", flush=True)
for n, iid in enumerate(sorted(candidates), start=1):
meta = candidates[iid]
key = meta["key"]
sid = str(iid)
need_fetch = args.force or sid not in cached
if not need_fetch and args.reformat_desc and sid in cached:
prev = cached[sid]
if HAS_PLACEHOLDER.search(prev.get("desc_loc") or "") or HAS_PLACEHOLDER.search(
prev.get("desc_en") or ""
):
need_fetch = True
if sid in cached and not need_fetch:
row = dict(cached[sid])
# refresh tags from stored text + overrides
tags = auto_tags(
key,
row.get("desc_loc") or "",
row.get("desc_en") or "",
meta.get("en_text") or "",
meta.get("od_dispellable"),
)
row["tags"] = apply_overrides(key, tags, overrides)
row["ability_kinds"] = parse_ability_kinds(
row.get("desc_loc") or "", meta.get("od_abilities") or []
)
row["name_loc"] = meta["name_loc"] or row.get("name_loc") or key
row["dname"] = meta["dname"]
row["key"] = key
row["cost"] = meta["cost"]
row["desc_loc"] = format_loc(row.get("desc_loc") or "", [])
row["desc_en"] = format_loc(row.get("desc_en") or "", [])
items_out[sid] = row
continue
zh = fetch_itemdata(iid, "schinese")
time.sleep(args.delay)
en = fetch_itemdata(iid, "english")
time.sleep(args.delay)
sv = (zh or {}).get("special_values") or []
if not isinstance(sv, list):
sv = []
desc_zh_raw = (zh or {}).get("desc_loc") or ""
desc_en_raw = (en or {}).get("desc_loc") or ""
desc_zh = format_loc(desc_zh_raw, sv)
desc_en = format_loc(desc_en_raw, sv)
notes = (zh or {}).get("notes_loc") or []
if not isinstance(notes, list):
notes = []
tags = auto_tags(
key,
desc_zh_raw,
desc_en_raw,
meta.get("en_text") or "",
meta.get("od_dispellable"),
)
tags = apply_overrides(key, tags, overrides)
items_out[sid] = {
"id": iid,
"key": key,
"dname": meta["dname"],
"name_loc": meta["name_loc"] or ((zh or {}).get("name_loc") or key),
"cost": meta["cost"],
"desc_loc": desc_zh,
"desc_en": desc_en,
"notes_loc": notes,
"dispellable": (zh or {}).get("dispellable"),
"immunity": (zh or {}).get("immunity"),
"ability_kinds": parse_ability_kinds(
desc_zh_raw, meta.get("od_abilities") or []
),
"tags": tags,
}
print(
f" [{n}/{len(candidates)}] {key}: tags={tags or '-'}",
flush=True,
)
# Incremental save
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(),
"tag_order": TAG_ORDER,
},
"items": items_out,
}
args.out.write_text(
json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
payload = {
"meta": {
"source": "valve+opendota",
"attribution": "https://www.dota2.com ; https://www.opendota.com",
"fetched_at": datetime.now(timezone.utc).isoformat(),
"tag_order": TAG_ORDER,
"count": len(items_out),
},
"items": items_out,
}
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 = {row["key"] for row in items_out.values()}
print(f"downloading {len(keys)} icons -> {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)
tagged = sum(1 for r in items_out.values() if r.get("tags"))
print(f"done: {len(items_out)} items, {tagged} with tags -> {args.out}", flush=True)
if __name__ == "__main__":
main()