forge_api.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
forge_api.py — Forge-/Aktions-Statistik-Endpoints fuer den SwordMasters-Optimizer.

Liest die aggregierte SQLite-DB (privates HF-Dataset oettaM123/swordmasters-forge-stats).

  GET /forge/meta                          globale Totals + Aktionsliste + Update-Zeit
  GET /forge/player?name=                  Pro-Aktion-Zusammenfassung + Luck (nur dwarf)
  GET /forge/player-detail?name=&action=   Drill-Down, gruppiert nach dim (item/ore/enchant/level/drop/...)
  GET /forge/leaderboard?action=&sort=&q=  Leaderboard (bei Suche mit Rang der Treffer)
  GET /forge/items?action=dwarf            Raritaet x Level-Raster (bucket_totals, exakt)
  GET /forge/item-cell?rarity=&level=      Items einer Dwarf-Zelle (Erfolge) + Zellen-Total
  GET /forge/item-leaderboard?item=&level= Spieler, die dieses Item auf dem Level am meisten forgen
  GET /forge/breakdown?action=&dim=        Globaler Breakdown (Ore/Enchant/Drop/...) aus item_totals

Alle luck/value nur fuer dwarf (nur dort exakte erwartete Rate). Nur lesend.
DB-Quelle: env FORGE_DB_PATH -> ./forge_stats.db -> HF-Dataset (Token env HF_TOKEN).
"""
import json
import math
import os
import sqlite3
import threading
import time
from pathlib import Path

from fastapi import APIRouter, HTTPException

router = APIRouter(prefix="/forge", tags=["forge"])

FORGE_REPO = os.getenv("FORGE_DATASET_REPO", "oettaM123/swordmasters-forge-stats")
FORGE_DB_FILE = "forge_stats.db"

ACTION_LABEL = {
    "dwarf": "Dwarf Forge", "enchant": "Enchant", "smelt": "Smelt",
    "downgrade": "Downgrade", "elf": "Elf", "arena": "Arena", "pet_enhance": "Pet Enhance",
    "nightmare": "Nightmare Arena",
}
ACTION_ORDER = ["dwarf", "enchant", "smelt", "downgrade", "elf", "pet_enhance", "arena", "nightmare"]
# Aktionen ohne sinnvolle Fail-/Rate-Anzeige (Fails werden nicht geloggt). Arena + Nightmare
# Arena loggen nur Drops (keine Fails) -> beide drop-basiert, Luck = Legendary-Ratio.
NO_RATE = {"arena", "nightmare"}
# Drop-basierte Aktionen: Leaderboard zeigt Legendary statt Luck-z, Luck = Legendary-Drops.
DROP_ACTIONS = {"arena", "nightmare"}
# Aktionen mit exakter erwarteter Rate -> Luck-Sortierung/Bewertung moeglich
HAS_LUCK = {"dwarf"}
HAS_VALUE = {"dwarf"}
# Mindest-Versuche fuer die Luck-Sortierung je Aktion (kleinvolumige Aktionen brauchen
# einen niedrigeren Floor, sonst ist das Luckiest-Board leer). Bei Namenssuche ignoriert.
LUCK_MIN = {"dwarf": 200, "elf": 40, "pet_enhance": 40, "smelt": 100,
            "enchant": 80, "downgrade": 50, "arena": 30, "nightmare": 20}
# Aktionen, die in die kombinierte "Lucky/Unlucky"-Gesamtwertung eines Spielers eingehen.
OVERALL_LUCK_ACTIONS = ("dwarf", "elf", "pet_enhance", "arena")

_DB = {"conn": None, "err": None, "path": None, "checked": 0.0}
_LOCK = threading.Lock()
# Forge-DB periodisch nachladen: der taegliche Upload erzeugt eine neue Dataset-Revision ->
# hf_hub_download liefert dann einen neuen Pfad. Ohne das zeigte der Space die Forge-Stats bis zum
# naechsten Space-Neustart veraltet (die Connection war prozesslang gecacht). 0 = aus.
_DB_RELOAD_TTL = float(os.getenv("FORGE_DB_RELOAD_TTL", "1800"))   # 30 min
_RELOADING = {"busy": False}
_META_CACHE = {"data": None}


def _luck_z(successes, expected, attempts):
    if not expected or not attempts:
        return 0.0
    var = expected * (1.0 - expected / attempts)
    return (successes - expected) / math.sqrt(var) if var > 0 else 0.0


def _norm_cdf(z):
    return 0.5 * (1.0 + math.erf(z / math.sqrt(2.0)))


# ── Live-Luck pro Spieler: Prozent-Luck je Aktion + Overall-Value (dwarf/elf/pet) ──
# Alles zur Laufzeit aus player_detail + gebuendeltem Preis-Snapshot -> kein DB-Rebuild.
ELF_OUTPUT = {"dragon's poison": "Dragon's Devil", "solbrand": "Divinity Edge",
              "soulkeeper's blade": "Last Horizon", "dragon's anger": "Dragon's Soul",
              "sunward bulwark": "Asgardian Aegis", "sealguard": "Final Bastion"}
PET_OUTPUT = {"peppy": "XPeppy", "blue axolotl": "XAxolotl", "pengoo": "XPengoo",
              "leo": "XLeo", "phoenix": "XPhoenix", "octoo": "XOctoo", "xen": "XXen",
              "inferno": "XInferno", "runix": "XRunix"}
_VALUE_ACTIONS = ("dwarf", "elf", "pet_enhance")   # Overall-Wert nur aus diesen
_STD_DIM = {"smelt": "ore", "enchant": "enchant", "downgrade": "level", "pet_enhance": "pet"}
_VAL_DIM = {"dwarf": "item", "elf": "input", "pet_enhance": "pet"}

_FORGE_PRICES = {"data": None}
_BUCKET_RATES = {"data": None}


def _forge_prices():
    if _FORGE_PRICES["data"] is None:
        try:
            _FORGE_PRICES["data"] = json.loads(
                (Path(__file__).with_name("forge_prices.json")).read_text(encoding="utf-8"))
        except Exception:  # noqa: BLE001
            _FORGE_PRICES["data"] = {"items": {}, "pets": {}}
    return _FORGE_PRICES["data"]


def _item_price(name, level):
    it = _forge_prices().get("items", {})
    nl = str(name or "").lower()
    for cls in ("sword", "shield"):   # Namen sind pro Klasse eindeutig
        v = it.get(f"{cls}|{nl}|{int(level)}")
        if v:
            return v
    return None


def _pet_price(name):
    return _forge_prices().get("pets", {}).get(str(name or "").lower())


def _bucket_rates(c):
    """Globale Erfolgsraten je Bucket (dwarf: rarity×level; smelt/enchant/downgrade/pet:
    je Ore/Typ/Level/Pet; arena: Legendary-Rate) aus item_totals/bucket_totals, gecacht."""
    if _BUCKET_RATES["data"] is None:
        R = {"dwarf": {}, "smelt": {}, "enchant": {}, "downgrade": {}, "pet_enhance": {}, "arena_leg": 0.0}
        for r in c.execute("SELECT rarity, level, SUM(attempts) a, SUM(successes) s "
                           "FROM bucket_totals GROUP BY rarity, level"):
            if r["a"]:
                R["dwarf"][(r["rarity"], r["level"])] = r["s"] / r["a"]
        for act, dim in (("smelt", "ore"), ("downgrade", "level"), ("elf", "input"),
                         ("pet_enhance", "pet"), ("enchant", "enchant")):
            agg = {}
            for r in c.execute("SELECT items.name k, SUM(attempts) a, SUM(successes) s "
                               "FROM item_totals JOIN items ON items.id = item_totals.key_id "
                               "WHERE action=? AND dim=? GROUP BY items.name", (act, dim)):
                if r["a"]:
                    agg[str(r["k"] or "").lower()] = r["s"] / r["a"]
            R[act] = agg
        r = c.execute("SELECT SUM(CASE WHEN rarity='Legendary' THEN successes ELSE 0 END) leg, "
                      "SUM(attempts) a FROM item_totals WHERE action='arena' AND dim='drop'").fetchone()
        R["arena_leg"] = (r["leg"] / r["a"]) if r and r["a"] and r["leg"] else 0.0
        _BUCKET_RATES["data"] = R
    return _BUCKET_RATES["data"]


def _compute_player_luck(c, pids, actions):
    """pids = alle lokalen player_ids des Accounts (ueber Namenswechsel aggregiert).
    Setzt je Aktion `luck_pct` (Prozent ueber/unter Erwartung) + `luck_percentile` (Farbe),
    mit exakten Einzel-Wahrscheinlichkeiten (dwarf/elf: Σ angezeigtes luck% inkl. Fortune;
    smelt/enchant/downgrade/pet: je Bucket-Rate; arena: erwartete Legendarys). Berechnet das
    Overall aus dwarf+elf+pet als WERT-Luck (realisierter vs. erwarteter Wert; Wert = Input
    immer verloren, Output nur bei Erfolg: dwarf->Item L+1, elf->Legendary L1, pet->XPet)."""
    R = _bucket_rates(c)
    pids = list(pids) if not isinstance(pids, (list, tuple)) else list(pids)
    if not pids:
        return None
    qmarks = ",".join("?" * len(pids))
    buckets = {}   # (action, dim) -> [(key_lower, level, rarity, att, succ), ...]
    for r in c.execute("SELECT action, dim, items.name k, level, rarity, attempts, successes "
                       "FROM player_detail JOIN items ON items.id = player_detail.key_id "
                       f"WHERE player_id IN ({qmarks})", pids):
        buckets.setdefault((r["action"], r["dim"]), []).append(
            (str(r["k"] or "").lower(), r["level"] or 0, r["rarity"] or "", r["attempts"] or 0, r["successes"] or 0))
    by_action = {a["action"]: a for a in actions}

    def _std_expected(act):
        rate = R.get(act, {})
        return sum(att * rate.get(k, 0.0) for k, lv, rr, att, su in buckets.get((act, _STD_DIM[act]), []))

    # --- per-Aktion Prozent-Luck NUR fuer dwarf/elf/pet ---
    # (smelt/enchant/downgrade: Odds haengen an free/paid bzw. Spieler-Level -> nicht im Log,
    #  also kein sauberes Luck; arena nur als Legendary-Count. Bewusst weggelassen.)
    for a in actions:
        act = a["action"]
        if act not in _VALUE_ACTIONS:
            continue
        att = a.get("attempts") or 0; su = a.get("successes") or 0
        exp = (a.get("expected") or 0.0) if act in ("dwarf", "elf") else _std_expected("pet_enhance")
        if exp and exp > 0:
            a["luck_pct"] = (su - exp) / exp * 100.0
            a["luck_percentile"] = _norm_cdf(_luck_z(su, exp, att))

    # --- Overall = WERT-Luck aus dwarf/elf/pet ---
    num = var = 0.0
    parts = []
    for act in _VALUE_ACTIONS:
        rows = buckets.get((act, _VAL_DIM[act]), [])
        if not rows:
            continue
        prate = R["dwarf"] if act == "dwarf" else R.get(act, {})
        def _p(k, lv, rr):
            return prate.get((rr, lv), 0.0) if act == "dwarf" else prate.get(k, 0.0)
        # Fortune-Skalierung: erwartete Erfolge auf das exakte Σluck% des Spielers ziehen
        factor = 1.0
        if act in ("dwarf", "elf"):
            base = sum(att * _p(k, lv, rr) for k, lv, rr, att, su in rows)
            aexp = by_action.get(act, {}).get("expected") or 0.0
            if base > 0 and aexp > 0:
                factor = aexp / base
        anum = avar = 0.0
        for k, lv, rr, att, su in rows:
            p = _p(k, lv, rr) * factor
            if p <= 0.0 or p >= 1.0:
                continue
            if act == "dwarf":
                swing = _item_price(k, lv + 1)
            elif act == "elf":
                out = ELF_OUTPUT.get(k); swing = _item_price(out, 1) if out else None
            else:
                out = PET_OUTPUT.get(k); swing = _pet_price(out) if out else None
            if not swing:
                continue
            anum += swing * (su - att * p)
            avar += swing * swing * att * p * (1.0 - p)
        if avar > 0:
            num += anum; var += avar; parts.append(act)
            av = by_action.get(act)   # Value pro Aktion (dwarf/elf/pet) fuer die Anzeige
            if av is not None:
                av["value_luck_coins"] = anum
                av["value_luck_percentile"] = _norm_cdf(anum / math.sqrt(avar))
    if var <= 0:
        return None
    z = num / math.sqrt(var)
    return {"luck_z": z, "luck_percentile": _norm_cdf(z), "value_luck_coins": num,
            "parts": parts, "weighting": "value"}


_PA_HAS_VLUCK = {"v": None}


def _has_vluck(c):
    """Hat der aktuelle DB-Stand die Wert-Luck-Spalten? (Abwaertskompatibel zu alten DBs.)"""
    if _PA_HAS_VLUCK["v"] is None:
        cols = {r[1] for r in c.execute("PRAGMA table_info(player_action)")}
        _PA_HAS_VLUCK["v"] = "vluck_num" in cols
    return _PA_HAS_VLUCK["v"]


_LEG_RATE = {}   # action -> globale Legendary-Droprate (Legendary-Drops / alle Drops)


def _leg_rate(c, action="arena"):
    """Globale Legendary-Droprate einer drop-basierten Aktion (arena/nightmare)."""
    if action not in _LEG_RATE:
        leg = c.execute("SELECT SUM(successes) s FROM item_totals "
                        "WHERE action=? AND rarity='Legendary'", (action,)).fetchone()["s"] or 0
        att = c.execute("SELECT SUM(attempts) a FROM item_totals "
                        "WHERE action=?", (action,)).fetchone()["a"] or 0
        _LEG_RATE[action] = (leg / att) if att else 0.0
    return _LEG_RATE[action]


def _arena_leg_rate(c):
    """Rueckwaerts-kompatibler Alias (Arena-Legendary-Rate)."""
    return _leg_rate(c, "arena")


def _combined_overall(c, actions):
    """Kombinierte Lucky/Unlucky-Wertung — WERT-gewichtet ueber dwarf+elf+pet_enhance+arena.
    Z = Σ vluck_num / √(Σ vluck_var), mit Einsatz = Wert-at-Risk pro Event (dwarf: Item@Level,
    elf: 10×Input@L5-Epic, pet: 10×Normal-Pet, arena: Ø Legendary-Wert). So zaehlt ein teurer
    Elf-/L9-Dwarf-Fail schwer und billiger Kram ist Rauschen — statt statistisch gleicher Pools.
    Faellt auf die alte count-basierte Pool-Wertung zurueck, solange die DB keine vluck-Spalten hat."""
    num = 0.0
    var = 0.0
    ev = 0
    parts = []
    for a in actions:
        act = a["action"]
        if act not in OVERALL_LUCK_ACTIONS:
            continue
        vv = a.get("vluck_var") or 0.0
        if vv <= 0:
            continue
        num += a.get("vluck_num") or 0.0
        var += vv
        ev += a.get("attempts") or 0
        parts.append(act)
    if var > 0:
        z = num / math.sqrt(var)
        return {"luck_z": z, "luck_percentile": _norm_cdf(z), "events": ev, "parts": parts,
                "value_luck_coins": num, "weighting": "value"}
    return _combined_overall_count(c, actions)


def _combined_overall_count(c, actions):
    """Alt/Fallback: statistisch gleich gewichteter Pool-z-Score (kein Wert)."""
    arate = _arena_leg_rate(c)
    num = 0.0
    var = 0.0
    ev = 0
    parts = []
    for a in actions:
        act = a["action"]
        att = a.get("attempts") or 0
        if not att or act not in OVERALL_LUCK_ACTIONS:
            continue
        if act == "arena":
            succ = a.get("legendary") or 0
            exp = att * arate
            v = exp * (1.0 - arate)
        else:
            exp = a.get("expected") or 0.0
            succ = a.get("successes") or 0
            v = exp * (1.0 - exp / att) if att else 0.0
        if exp <= 0 or v <= 0:
            continue
        num += (succ - exp)
        var += v
        ev += att
        parts.append(act)
    if var <= 0:
        return None
    z = num / math.sqrt(var)
    return {"luck_z": z, "luck_percentile": _norm_cdf(z), "events": ev, "parts": parts,
            "weighting": "count"}


_RATE_MAP = {"data": None}


def _rate_map(c):
    """(rarity, level) -> exakte Dwarf-Erfolgsrate (aus bucket_totals, beide Klassen)."""
    if _RATE_MAP["data"] is None:
        m = {}
        for r in c.execute("SELECT rarity, level, SUM(attempts) a, SUM(successes) s "
                           "FROM bucket_totals GROUP BY rarity, level"):
            if r["a"]:
                m[(r["rarity"], r["level"])] = r["s"] / r["a"]
        _RATE_MAP["data"] = m
    return _RATE_MAP["data"]


def _est_attempts(rm, rarity, level, successes):
    """Geschaetzte Forge-VERSUCHE eines Items = Erfolge / (fixe Rate je rarity,level).
    Dwarf-Burns sind namentlich korrupt, aber die Rate ist je (rarity,level) exakt."""
    rate = rm.get((rarity, level))
    return int(round(successes / rate)) if rate and rate > 0 else successes


def _resolve_db_path():
    p = os.getenv("FORGE_DB_PATH")
    if p and Path(p).exists():
        return p
    local = Path(__file__).with_name(FORGE_DB_FILE)
    if local.exists():
        return str(local)
    from huggingface_hub import hf_hub_download
    token = (os.getenv("HF_TOKEN") or os.getenv("HUGGING_FACE_HUB_TOKEN")
             or os.getenv("HUGGINGFACE_TOKEN"))
    return hf_hub_download(repo_id=FORGE_REPO, repo_type="dataset",
                          filename=FORGE_DB_FILE, token=token)


def _open_db(path):
    c = sqlite3.connect(f"file:{path}?mode=ro", uri=True, check_same_thread=False)
    c.row_factory = sqlite3.Row
    c.create_function("luck_z", 3, _luck_z, deterministic=True)
    return c


def _invalidate_db_caches():
    """Alle aus der Forge-DB abgeleiteten Modul-Caches verwerfen (nach einem DB-Tausch).
    Lazy: der naechste Zugriff fuellt sie aus der neuen DB neu. (forge_prices ist DATEI-basiert
    -> bleibt.)"""
    _META_CACHE["data"] = None
    _BUCKET_RATES["data"] = None
    _RATE_MAP["data"] = None
    _PA_HAS_VLUCK["v"] = None
    _LEG_RATE.clear()


def _reload_worker():
    """Im HINTERGRUND: neue DB-Version holen (falls vorhanden) und Connection ATOMAR tauschen.
    Kein Request blockiert -- bis zum Tausch wird die alte DB weiter bedient. Die alte Connection
    schliesst Python per Refcount, sobald kein laufender Request sie mehr haelt (nie mitten im
    Query -> kein close-Race). Fehler sind nie fatal (alte DB laeuft weiter)."""
    try:
        newpath = _resolve_db_path()
        if newpath and newpath != _DB["path"]:
            newc = _open_db(newpath)
            _DB["conn"] = newc          # atomarer Pointer-Tausch (GIL)
            _DB["path"] = newpath
            _invalidate_db_caches()
    except Exception:  # noqa: BLE001
        pass
    finally:
        _RELOADING["busy"] = False


def _conn():
    c = _DB["conn"]
    if c is not None:
        now = time.monotonic()
        # periodisch pruefen, ob eine neue DB-Version vorliegt; den Download macht EIN
        # Hintergrund-Thread, Requests servieren derweil die alte DB (kein Blockieren).
        if (_DB_RELOAD_TTL > 0 and not _RELOADING["busy"]
                and (now - _DB["checked"]) >= _DB_RELOAD_TTL and _LOCK.acquire(blocking=False)):
            try:
                if not _RELOADING["busy"] and (time.monotonic() - _DB["checked"]) >= _DB_RELOAD_TTL:
                    _RELOADING["busy"] = True
                    _DB["checked"] = time.monotonic()
                    threading.Thread(target=_reload_worker, daemon=True).start()
            finally:
                _LOCK.release()
        return _DB["conn"]
    with _LOCK:
        if _DB["conn"] is None and _DB["err"] is None:
            try:
                path = _resolve_db_path()
                _DB["conn"] = _open_db(path)
                _DB["path"] = path
                _DB["checked"] = time.monotonic()
            except Exception as exc:  # noqa: BLE001
                _DB["err"] = f"{type(exc).__name__}: {exc}"
    if _DB["conn"] is None:
        raise HTTPException(503, f"Forge stats DB unavailable ({_DB['err']}).")
    return _DB["conn"]


def warm():
    def _run():
        try:
            _conn()
        except Exception:  # noqa: BLE001
            pass
    threading.Thread(target=_run, daemon=True).start()


def _find_player(c, name):
    row = c.execute("SELECT id, name FROM players WHERE name = ?", (name,)).fetchone()
    if row:
        return row["id"], row["name"]
    row = c.execute("SELECT id, name FROM players WHERE name = ? COLLATE NOCASE", (name,)).fetchone()
    return (row["id"], row["name"]) if row else (None, None)


# ── id-gebundene Aggregation ueber Namenswechsel ────────────────────────────────────────────────
# Die Forge-DB ist nach NAME gekeyt (player_id = kompakter Namens-Index). Um Stats wie die
# Power-Series an den ACCOUNT zu binden, loest app.py (das das Leaderboard + aliases hat) einen
# Namen -> alle Namen des Accounts (aktuell + fruehere/aliases) auf; wir aggregieren dann ueber
# alle lokalen player_ids dieser Namen. So bleiben die Forge-Stats nach einer Umbenennung beim
# selben Account. Kein Resolver gesetzt / Name unbekannt -> Fallback auf den einen Namen.
_ACCOUNT_NAMES_RESOLVER = None


def set_account_names_resolver(fn):
    global _ACCOUNT_NAMES_RESOLVER
    _ACCOUNT_NAMES_RESOLVER = fn


def _account_names(name):
    names = []
    if _ACCOUNT_NAMES_RESOLVER is not None:
        try:
            names = list(_ACCOUNT_NAMES_RESOLVER(name) or [])
        except Exception:  # noqa: BLE001
            names = []
    if name:
        names.insert(0, name)
    seen, out = set(), []
    for n in names:
        n = (n or "").strip()
        if n and n.lower() not in seen:
            seen.add(n.lower())
            out.append(n)
    return out


def _find_players(c, name):
    """Alle lokalen player_ids des Accounts (ueber alle Namen/aliases) + Anzeigename."""
    pids, canonical = [], None
    for nm in _account_names(name):
        pid, canon = _find_player(c, nm)
        if pid is not None and pid not in pids:
            pids.append(pid)
            if canonical is None:
                canonical = canon
    return pids, canonical


def _split_names(names):
    """'a, b; c' -> ['a','b','c'] (Komma/Semikolon-getrennt, getrimmt, leere raus)."""
    return [n.strip() for n in str(names or "").replace(";", ",").split(",") if n.strip()]


def _find_players_multi(c, names_list):
    """Union der lokalen player_ids ueber MEHRERE Accounts (jeder inkl. seiner Aliases) —
    fuer die kombinierte Mehr-Account-Ansicht. Gibt (pids, [canonical display names])."""
    pids, labels, seen = [], [], set()
    for raw in names_list[:20]:          # Deckel gegen Missbrauch
        sub, canon = _find_players(c, raw)
        for pid in sub:
            if pid not in seen:
                seen.add(pid)
                pids.append(pid)
        if canon and canon not in labels:
            labels.append(canon)
    return pids, labels


def _summary_row(r):
    """player_action-Zeile -> Anzeige-dict (rate, luck, value je nach Aktion).
    Luck fuer ALLE Aktionen: dwarf/elf via per-Event-luck, smelt/enchant/downgrade/
    pet via globaler Rate (expected=attempts*rate), arena via Legendary-Ratio
    (expected-Slot enthaelt die Legendary-Drop-Anzahl)."""
    a = dict(r)
    act = a["action"]
    att = a.get("attempts") or 0
    su = a.get("successes") or 0
    exp = a.get("expected") if "expected" in a.keys() else a.get("expected_successes")
    exp = exp or 0.0
    a["label"] = ACTION_LABEL.get(act, act)
    a["has_rate"] = act not in NO_RATE
    a["rate"] = (su / att) if (att and a["has_rate"]) else None
    a["fails"] = (att - su) if a["has_rate"] else 0
    if act in DROP_ACTIONS:
        a["legendary"] = int(round(exp))
        a["legendary_ratio"] = (exp / att) if att else None
    elif exp and att:
        z = _luck_z(su, exp, att)
        a["expected_rate"] = exp / att
        a["luck_delta"] = su - exp
        a["luck_z"] = z
        a["luck_percentile"] = _norm_cdf(z)
    if act in HAS_VALUE:
        a["value_net"] = a.get("value_net")
    return a


@router.get("/meta")
def forge_meta():
    if _META_CACHE["data"] is not None:
        return _META_CACHE["data"]
    c = _conn()
    meta = {r["key"]: r["value"] for r in c.execute("SELECT key, value FROM meta")}
    seen, totals = set(), []
    at = {}
    if meta.get("action_totals"):
        try:
            at = json.loads(meta["action_totals"])
        except Exception:  # noqa: BLE001
            at = {}
    for action, pair in at.items():
        seen.add(action)
        totals.append({"action": action, "label": ACTION_LABEL.get(action, action),
                       "attempts": pair[0], "successes": pair[1],
                       "has_rate": action not in NO_RATE})
    totals.sort(key=lambda t: (ACTION_ORDER.index(t["action"])
                               if t["action"] in ACTION_ORDER else 99))
    players = int(meta.get("compact_players") or 0) or \
        c.execute("SELECT COUNT(*) n FROM players").fetchone()["n"]
    result = {"meta": meta, "totals": totals, "player_count": players,
              "actions": [a for a in ACTION_ORDER if a in seen],
              "built_at_unix": int(meta.get("built_at_unix") or 0),
              "data_through": meta.get("data_through", "")}
    _META_CACHE["data"] = result
    return result


@router.get("/player")
def forge_player(name: str = "", names: str = ""):
    """Ein Account (name=) ODER mehrere kombiniert (names=a,b,c -> Dwarf-Multi: Stats ueber
    alle genannten Accounts + deren Aliases zusammengerechnet)."""
    c = _conn()
    nlist = _split_names(names)
    if nlist:
        pids, labels = _find_players_multi(c, nlist)
        canonical = ", ".join(labels) if labels else ", ".join(nlist)
        if not pids:
            raise HTTPException(404, "No forge stats logged for these accounts.")
    else:
        name = (name or "").strip()
        if not name:
            raise HTTPException(400, "Missing name.")
        pids, canonical = _find_players(c, name)
        if not pids:
            raise HTTPException(404, f"No forge stats logged for '{name}'.")
    # player_action hat 1 Zeile je (player_id, action) -> ueber alle Account-pids je Aktion summieren.
    _SUM = ("attempts", "successes", "expected", "value_gained", "value_lost", "value_known", "value_net")
    qmarks = ",".join("?" * len(pids))
    agg = {}
    for r in c.execute(
        "SELECT action, attempts, successes, expected, value_gained, value_lost, "
        f"value_known, value_net FROM player_action WHERE player_id IN ({qmarks})", pids):
        act = r["action"]
        cur = agg.get(act)
        if cur is None:
            agg[act] = {k: (r[k] if k in r.keys() else None) for k in r.keys()}
        else:
            for k in _SUM:
                if k in r.keys():
                    cur[k] = (cur.get(k) or 0) + (r[k] or 0)
    actions = [_summary_row(a) for a in agg.values()]
    actions.sort(key=lambda a: (ACTION_ORDER.index(a["action"])
                                if a["action"] in ACTION_ORDER else 99))
    try:
        overall = _compute_player_luck(c, pids, actions)  # Prozent-Luck je Aktion + Overall-Wert
    except Exception:  # noqa: BLE001  -> Anzeige darf nie am Luck haengen
        overall = None
    return {"name": canonical, "actions": actions, "overall": overall}


@router.get("/player-detail")
def forge_player_detail(name: str = "", action: str = "", limit: int = 400, names: str = ""):
    c = _conn()
    nlist = _split_names(names)
    if nlist:
        pids, labels = _find_players_multi(c, nlist)
        canonical = ", ".join(labels) if labels else ", ".join(nlist)
        if not pids:
            raise HTTPException(404, "No forge stats logged for these accounts.")
    else:
        name = (name or "").strip()
        pids, canonical = _find_players(c, name)
        if not pids:
            raise HTTPException(404, f"No forge stats logged for '{name}'.")
    limit = max(1, min(int(limit), 2000))
    qmarks = ",".join("?" * len(pids))
    # Ueber alle Account-pids ziehen und gleiche (dim,key,level,rarity)-Buckets summieren; das
    # SQL-LIMIT erst NACH dem Mergen je dim anwenden (sonst wuerde es vor der Aggregation schneiden).
    merged = {}
    for r in c.execute(
            "SELECT dim, items.name AS key, level, rarity, attempts, successes, extra "
            "FROM player_detail JOIN items ON items.id = player_detail.key_id "
            f"WHERE player_id IN ({qmarks}) AND action = ? AND items.name <> 'Empty' COLLATE NOCASE",
            pids + [action]):
        d = dict(r)
        k = (d["dim"], d["key"], d["level"], d["rarity"])
        m = merged.get(k)
        if m is None:
            merged[k] = d
        else:
            for col in ("attempts", "successes", "extra"):
                m[col] = (m.get(col) or 0) + (d.get(col) or 0)
    dims = {}
    for d in merged.values():
        dims.setdefault(d.pop("dim"), []).append(d)
    for dim in dims:
        dims[dim].sort(key=lambda x: (-(x.get("successes") or 0), -(x.get("attempts") or 0)))
        dims[dim] = dims[dim][:limit]
    return {"name": canonical, "action": action,
            "label": ACTION_LABEL.get(action, action), "dims": dims}


def _leaderboard_rank(c, action, sort, expr, direction, min_attempts, row):
    """1-basierter Rang der Zeile in der ungefilterten sortierten Liste."""
    att = row["attempts"] or 0
    exp = row["expected"] or 0
    if sort in ("luckiest", "unluckiest"):
        if action in DROP_ACTIONS:                   # Ratio Legendary/Versuch (exp = Legendary-Anzahl)
            luckval = (exp / att) if att else 0.0
        else:
            luckval = _luck_z(row["successes"], exp, att)
    else:
        luckval = None
    val = {"attempts": att, "gain": row["value_net"], "loss": row["value_net"],
           "luckiest": luckval, "unluckiest": luckval, "legendary": exp}.get(sort)
    if val is None:
        return None
    # Rang = 1 + Anzahl der "besseren" Zeilen. DESC (hoechster oben) -> besser = groesser.
    cmp = ">" if direction == "DESC" else "<"
    where = ["action = ?"]
    params = [action]
    if min_attempts > 0:
        where.append("attempts >= ?"); params.append(min_attempts)
    where.append(f"{expr} {cmp} ?"); params.append(val)
    n = c.execute(f"SELECT COUNT(*) n FROM player_action WHERE {' AND '.join(where)}",
                  params).fetchone()["n"]
    return n + 1


@router.get("/leaderboard")
def forge_leaderboard(action: str = "dwarf", sort: str = "attempts",
                      q: str = "", offset: int = 0, limit: int = 25,
                      min_attempts: int = 0):
    c = _conn()
    action = (action or "dwarf").strip()
    limit = max(1, min(int(limit), 200))
    offset = max(0, int(offset))
    luck_sorts = {"luckiest": "DESC", "unluckiest": "ASC"}
    if sort in ("gain", "loss") and action not in HAS_VALUE:
        sort = "attempts"
    if sort == "legendary" and action not in DROP_ACTIONS:   # "Most legendarys" nur fuer arena/nightmare
        sort = "attempts"
    qq = q.strip()
    if sort in luck_sorts:
        direction = luck_sorts[sort]
        if action in DROP_ACTIONS:
            # Luckiest = hoechste Legendary/Versuch-Ratio (expected = Legendary-Anzahl). Damit steht
            # JEDER mit >=1 Legendary ueber jedem mit 0, und "mehr Legendarys in weniger Versuchen"
            # gewinnt. KEIN Mindest-Versuche-Floor -> alle Finder sichtbar, 0-Legendary landen unten.
            expr = "(expected*1.0/attempts)"
        else:                                       # z-Score gegen erwartete Rate
            expr = "luck_z(successes, expected, attempts)"
            if min_attempts <= 0:
                min_attempts = LUCK_MIN.get(action, 100)
    elif sort == "legendary":
        # expected traegt bei drop-Aktionen die Legendary-Anzahl -> nach den meisten Legendarys.
        expr, direction = "expected", "DESC"
    elif sort in ("gain", "loss"):
        expr = "value_net"
        direction = "ASC" if sort == "loss" else "DESC"
    else:
        expr, direction, sort = "attempts", "DESC", "attempts"
    # Bei Namenssuche den Luck-Floor ignorieren, damit der gesuchte Name IMMER auftaucht
    # (der Rang wird trotzdem gegen ALLE Spieler dieser Aktion berechnet).
    if qq:
        min_attempts = 0

    where = ["pa.action = ?"]
    params = [action]
    if min_attempts > 0:
        where.append("pa.attempts >= ?"); params.append(int(min_attempts))
    if qq:
        where.append("p.name LIKE ? ESCAPE '\\'")
        params.append("%" + qq.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + "%")
    # Tiebreak bei gleicher Ratio (drop-Aktionen): luckiest -> mehr Legendarys zuerst,
    # unluckiest -> mehr Versuche zuerst (viele Versuche, 0 Legendary = am ungluecklichsten).
    order_extra = ""
    if action in DROP_ACTIONS and sort == "luckiest":
        order_extra = ", pa.expected DESC"
    elif action in DROP_ACTIONS and sort == "unluckiest":
        order_extra = ", pa.attempts DESC"
    sql = (f"SELECT p.name, pa.action, pa.attempts, pa.successes, pa.expected, "
           f"pa.value_gained, pa.value_lost, pa.value_known, pa.value_net "
           f"FROM player_action pa JOIN players p ON p.id = pa.player_id "
           f"WHERE {' AND '.join(where)} ORDER BY {expr} {direction}{order_extra} LIMIT ? OFFSET ?")
    rows = []
    for r in c.execute(sql, params + [limit + 1, offset]):
        d = _summary_row(r)
        if qq:  # bei Suche: echten Rang in der Gesamtliste zeigen
            d["rank"] = _leaderboard_rank(c, action, sort, expr, direction, min_attempts, r)
        rows.append(d)
    has_more = len(rows) > limit
    rows = rows[:limit]
    return {"action": action, "sort": sort, "offset": offset, "has_more": has_more,
            "min_attempts": min_attempts, "has_rate": action not in NO_RATE,
            "has_luck": True, "is_arena": action in DROP_ACTIONS,
            "has_value": action in HAS_VALUE, "rows": rows}


@router.get("/items")
def forge_items(action: str = "dwarf"):
    c = _conn()
    grid = [dict(r) for r in c.execute(
        "SELECT item_class, level, rarity, attempts, successes FROM bucket_totals ORDER BY level")]
    return {"action": "dwarf", "grid": grid}


@router.get("/item-dist")
def forge_item_dist(rarity: str = "", level: int = -1, offset: int = 0, limit: int = 60):
    """Dwarf-Items nach echten VERSUCHEN (attempts=forged+burned, exakt). Filter optional
    nach rarity und/oder level: keine=Overview, nur rarity=Zeile, nur level=Spalte, beide=Zelle."""
    c = _conn()
    limit = max(1, min(int(limit), 1000)); offset = max(0, int(offset))
    # 'Empty' ist kein echtes Katalog-Item (Log-Artefakt) -> ausblenden.
    where = ["action='dwarf'", "dim='item'", "items.name <> 'Empty' COLLATE NOCASE"]
    params = []
    if rarity:
        where.append("rarity = ?"); params.append(rarity)
    if level >= 0:
        where.append("level = ?"); params.append(int(level))
    agg = {}
    for r in c.execute(
            "SELECT items.name AS item, rarity, attempts, successes FROM item_totals "
            "JOIN items ON items.id = item_totals.key_id "
            f"WHERE {' AND '.join(where)}", params):
        e = agg.setdefault(r["item"], {"item": r["item"], "successes": 0, "attempts": 0, "_rar": {}})
        e["successes"] += r["successes"]; e["attempts"] += r["attempts"]
        if r["rarity"]:                       # DOMINANTE Raritaet (nach Versuchen), nicht die erste —
            e["_rar"][r["rarity"]] = e["_rar"].get(r["rarity"], 0) + r["attempts"]  # sonst filtert der
    for e in agg.values():                    # Drill auf eine 1x-fehlklassifizierte Minderheit (Darkblade-Bug)
        e["rarity"] = max(e["_rar"], key=e["_rar"].get) if e["_rar"] else ""
        e.pop("_rar", None)
    items = sorted(agg.values(), key=lambda x: -x["attempts"])
    total = len(items)
    page = items[offset:offset + limit]
    bw, bp = [], []
    if rarity:
        bw.append("rarity = ?"); bp.append(rarity)
    if level >= 0:
        bw.append("level = ?"); bp.append(int(level))
    bt = c.execute("SELECT SUM(attempts) a, SUM(successes) s FROM bucket_totals"
                   + (" WHERE " + " AND ".join(bw) if bw else ""), bp).fetchone()
    return {"rarity": rarity, "level": level, "offset": offset, "total": total,
            "has_more": offset + limit < total,
            "tot_attempts": (bt["a"] or 0) if bt else 0,
            "tot_successes": (bt["s"] or 0) if bt else 0, "items": page}


@router.get("/item-levels")
def forge_item_levels(item: str, rarity: str = ""):
    """Pro Level fuer ein Dwarf-Item: echte Versuche/Fails/Erfolge (exakt)."""
    c = _conn()
    it = c.execute("SELECT id FROM items WHERE name = ?", (item,)).fetchone()
    if not it:
        return {"item": item, "levels": []}
    where = ["action='dwarf'", "dim='item'", "key_id = ?"]
    params = [it["id"]]
    if rarity:
        where.append("rarity = ?"); params.append(rarity)
    out = []
    for r in c.execute(f"SELECT level, rarity, attempts, successes FROM item_totals "
                       f"WHERE {' AND '.join(where)} ORDER BY level", params):
        out.append({"level": r["level"], "rarity": r["rarity"], "successes": r["successes"],
                    "attempts": r["attempts"], "fails": max(0, r["attempts"] - r["successes"])})
    return {"item": item, "levels": out}


@router.get("/item-leaderboard")
def forge_item_leaderboard(item: str, level: int, rarity: str = "", action: str = "dwarf",
                           offset: int = 0, limit: int = 50):
    """Spieler, die dieses (Item, Level[, rarity]) am meisten forgen — echte Versuche.
    Raritaets-Filter verhindert, dass 1x fehlklassifizierte Items die falschen Forger zeigen."""
    c = _conn()
    limit = max(1, min(int(limit), 500)); offset = max(0, int(offset))
    it = c.execute("SELECT id FROM items WHERE name = ?", (item,)).fetchone()
    if not it:
        return {"item": item, "level": level, "rows": [], "has_more": False,
                "tot": {"attempts": 0, "successes": 0}}
    where = ["pd.action = ?", "pd.dim = 'item'", "pd.key_id = ?", "pd.level = ?"]
    params = [action, it["id"], level]
    if rarity:
        where.append("pd.rarity = ?"); params.append(rarity)
    rows = []
    for r in c.execute(
            "SELECT p.name, pd.attempts, pd.successes FROM player_detail pd "
            "JOIN players p ON p.id = pd.player_id "
            f"WHERE {' AND '.join(where)} ORDER BY pd.attempts DESC LIMIT ? OFFSET ?",
            params + [limit + 1, offset]):
        rows.append({"name": r["name"], "successes": r["successes"], "attempts": r["attempts"]})
    has_more = len(rows) > limit
    # Exaktes Gesamt-Total (auch fuer Klein-Accounts, die nicht einzeln gespeichert sind)
    tw = ["action = ?", "dim = 'item'", "key_id = ?", "level = ?"]
    tp = [action, it["id"], level]
    if rarity:
        tw.append("rarity = ?"); tp.append(rarity)
    tr = c.execute(f"SELECT SUM(attempts) a, SUM(successes) s FROM item_totals "
                   f"WHERE {' AND '.join(tw)}", tp).fetchone()
    tot = {"attempts": (tr["a"] or 0) if tr else 0, "successes": (tr["s"] or 0) if tr else 0}
    return {"item": item, "level": level, "rarity": rarity, "action": action,
            "offset": offset, "has_more": has_more, "rows": rows[:limit], "tot": tot}


@router.get("/breakdown")
def forge_breakdown(action: str, dim: str, limit: int = 100):
    """Globaler Breakdown aus item_totals (z.B. alle Ores, alle Enchant-Typen, alle Drops)."""
    c = _conn()
    limit = max(1, min(int(limit), 500))
    rows = [dict(r) for r in c.execute(
        "SELECT items.name AS key, level, rarity, item_class, attempts, successes "
        "FROM item_totals JOIN items ON items.id = item_totals.key_id "
        "WHERE action = ? AND dim = ? ORDER BY successes DESC, attempts DESC LIMIT ?",
        (action, dim, limit))]
    return {"action": action, "dim": dim, "rows": rows}