vosonandCursor
9c5aa5b610
Separate the local recognition, web publishing, and shared data paths while preserving direct script execution and existing site content. Co-authored-by: Cursor <cursoragent@cursor.com>
442 lines
14 KiB
Python
442 lines
14 KiB
Python
"""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 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.http_utils import download_icons, http_json, load_itemlist
|
||
from shared.paths import DATA, ITEM_ICONS
|
||
|
||
from loc_format import HAS_PLACEHOLDER, format_loc
|
||
from mechanic_tags import apply_mechanic_tags, merge_tag_overrides
|
||
|
||
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",
|
||
"stun",
|
||
"disarm",
|
||
"mute",
|
||
"invis",
|
||
"sleep",
|
||
"fear",
|
||
"taunt",
|
||
"blind",
|
||
"leash",
|
||
"ethereal",
|
||
"cyclone",
|
||
]
|
||
|
||
# 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,
|
||
)
|
||
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 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])
|
||
tags: set[str] = set()
|
||
# Query-facing applies-* tags (dispel / CC); shared with ability tagging.
|
||
tags |= apply_mechanic_tags(blob, key=key)
|
||
|
||
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 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")
|
||
|
||
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]:
|
||
return merge_tag_overrides(tags, overrides.get(key), TAG_ORDER)
|
||
|
||
|
||
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()
|