BurningTimesAi/공유/기획/FirstRising/BM/tools/steam_reviews.py

# Steam 공개 API로 F2P·비교군 게임의 리뷰 집계(전체·최근 30일·월별 추이)·무료 여부를 실측해 CSV로 저장 (BT26-Plan 2026-09-25)
# 사용: python steam_reviews.py  → ../data/steam_reviews_2026-09-25.csv
#  - 이름 확인: ICommunityService/GetApps(일괄)  - 리뷰: store appreviews(전체) + appreviewhistogram(최근 30일·월별)  - 무료 여부: appdetails(제한 시 '?')
import csv, json, os, sys, time, urllib.parse, urllib.request

OUT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "steam_reviews_2026-09-25.csv")
APPS = {  # 이름(내 표기): appid — 실행 시 GetApps 로 실제 이름 대조
    "Dota 2": 570, "Counter-Strike 2": 730, "Team Fortress 2": 440, "Warframe": 230410, "Path of Exile": 238960, "Path of Exile 2": 2694490,
    "Apex Legends": 1172470, "Destiny 2": 1085660, "Lost Ark": 1599340, "Marvel Rivals": 2767030, "The Finals": 2073850, "Once Human": 2139460,
    "War Thunder": 236390, "World of Tanks": 1407200, "PUBG: BATTLEGROUNDS": 578080, "Halo Infinite": 1240440, "Overwatch 2": 2357570,
    "Brawlhalla": 291550, "Paladins": 444090, "Yu-Gi-Oh! Master Duel": 1449850, "Black Desert": 582660, "Throne and Liberty": 2429640,
    "MapleStory": 216150, "Albion Online": 761890, "EVE Online": 8500, "Delta Force": 2507950, "Rust": 252490,
    "Legends of IdleOn": 1476970, "NGU Idle": 1147690, "Clicker Heroes": 363970, "Idle Slayer": 1353300, "Melvor Idle": 1267910,
    "Banana": 2923300, "Tower of Fantasy": 2064650, "TBH: Task Bar Hero": 3678970, "Zenless Zone Zero": 4162040, "Wuthering Waves": 3513350,
    "GIRLS' FRONTLINE 2: EXILIUM": 3347400, "Reverse: 1999": 3092660, "Blue Archive": 3557620, "Farmer Against Potatoes Idle": 1535560,
    "Leaf Blower Revolution": 1468260, "Tap Ninja": 1891700, "Eternal Return": 1049590, "NARAKA: BLADEPOINT": 1203220, "Idle Wizard": 992070,
    "Predecessor": 961200, "SMITE 2": 2437170,
}
SEARCH = ["Idle Champions of the Forgotten Realms", "Deadlock"]
H = {"User-Agent": "Mozilla/5.0"}


def get(url):
    for i in range(2):
        try:
            return json.load(urllib.request.urlopen(urllib.request.Request(url, headers=H), timeout=30))
        except Exception as e:
            err = e; time.sleep(2)
    print("FAIL", url[:80], err, file=sys.stderr); return None


def search(term):
    items = (get("https://store.steampowered.com/api/storesearch/?cc=US&l=english&term=" + urllib.parse.quote(term)) or {}).get("items") or []
    return (items[0]["id"], items[0]["name"]) if items else (None, None)


def names(ids):
    out = {}
    for i in range(0, len(ids), 30):
        q = "&".join(f"appids[{k}]={a}" for k, a in enumerate(ids[i:i + 30]))
        for a in ((get("https://api.steampowered.com/ICommunityService/GetApps/v1/?" + q) or {}).get("response") or {}).get("apps", []):
            out[a["appid"]] = a.get("name")
        time.sleep(1)
    return out


def is_free(appid, name):
    """appdetails 가 success=false 를 돌려주는 앱이 많아 storesearch(동작 확인) 의 price 로 판정"""
    items = (get("https://store.steampowered.com/api/storesearch/?cc=US&l=english&term=" + urllib.parse.quote(name.split(" -> ")[-1])) or {}).get("items") or []
    for it in items:
        if it.get("id") == appid:
            pr = it.get("price")
            return ("무료" if not pr or not pr.get("final") else "유료"), (f"${pr['final'] / 100:.2f}" if pr and pr.get("final") else "")
    return "?", ""


def reviews(appid):
    s = ((get(f"https://store.steampowered.com/appreviews/{appid}?json=1&language=all&purchase_type=all&num_per_page=0&filter=all") or {}).get("query_summary")) or {}
    p, n = s.get("total_positive", 0), s.get("total_negative", 0)
    return s.get("review_score_desc", ""), p, n, (round(100 * p / (p + n)) if p + n else "")


def histogram(appid):
    r = (get(f"https://store.steampowered.com/appreviewhistogram/{appid}?l=english") or {}).get("results") or {}
    rec = r.get("recent") or []
    up = sum(x.get("recommendations_up", 0) for x in rec); dn = sum(x.get("recommendations_down", 0) for x in rec)
    roll = r.get("rollups") or []
    months = ";".join(f"{time.strftime('%Y-%m', time.gmtime(x['date']))}:{x.get('recommendations_up', 0)}/{x.get('recommendations_down', 0)}" for x in roll[-18:])
    return up, dn, (round(100 * up / (up + dn)) if up + dn else ""), months


targets = dict(APPS)
for term in SEARCH:
    a, nm = search(term); time.sleep(1)
    if a:
        targets[f"{term} -> {nm}"] = a
nm = names(list(targets.values()))
rows = []
for name, appid in targets.items():
    real = nm.get(appid, "?")
    free, price = is_free(appid, name); time.sleep(1.2)
    desc, p, n, pct = reviews(appid); time.sleep(1.2)
    p30, n30, pct30, months = histogram(appid); time.sleep(1.2)
    rows.append({"name": name, "steam_name": real, "appid": appid, "free": free, "price": price, "review_all": desc, "pos_all": p, "neg_all": n, "pct_all": pct,
                 "pos_30d": p30, "neg_30d": n30, "pct_30d": pct30, "months_up_down": months})
    flag = "" if real and real.lower().split(":")[0][:8] in name.lower() else "  <-- 이름 불일치 확인"
    print(f"{name[:34]:34} {appid:>8} {real[:28]:28} {free:>3} | all {pct}% ({p + n:,}) {desc} | 30d {pct30}% ({p30 + n30:,}){flag}", flush=True)
keys = ["name", "steam_name", "appid", "free", "price", "review_all", "pos_all", "neg_all", "pct_all", "pos_30d", "neg_30d", "pct_30d", "months_up_down"]
with open(OUT, "w", newline="", encoding="utf-8-sig") as f:
    w = csv.DictWriter(f, fieldnames=keys); w.writeheader(); [w.writerow(r) for r in rows]
print("saved", OUT, len(rows))