Backend: per-scan power/level/gold time-series for >=1 Qnt accounts

Collect a precise (second-granular) time-series point every time an
account with >= 1 Qnt power (1e18) is scanned via the normal lookup /
leaderboard ingest, so the player-lookup tab can later draw power/level/
gold over time. Stored in a SEPARATE persistent store (/data/power_series
.json, own lock + own 120s save interval) so the 600 MB leaderboard
write path is untouched.

- Point format: [t_sec, level, power, gold, sword_type, shield_type].
Active sword/shield captured as the small game type-code (translatable
back via SWORD_TYPE_NAMES / SHIELD_TYPE_NAMES); -1 when unknown.
- Add sword_type/shield_type to the live lookup result.
- Throttle to <=1 point/min/account (SM_POWER_SERIES_MIN_GAP) and cap
per account (SM_POWER_SERIES_MAX_POINTS=20000); env-tunable.
- GET /power-series?username=|player_id= returns the series (or
under_1qnt=true) for the frontend.

Accounts under 1 Qnt store nothing (frontend shows "unter 1qnt").
Frontend graph to follow.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app.py

@@ -141,6 +141,28 @@ SM_LEADERBOARD_QUERY_LIMIT = max(25, int(_os.getenv("SM_LEADERBOARD_QUERY_LIMIT"
 # und die Anzahl blieb bei 500k haengen. Jetzt 2 Mio (Headroom) + Eviction nach Power (unten).
 SM_LEADERBOARD_MAX_ROWS = max(1000, int(_os.getenv("SM_LEADERBOARD_MAX_ROWS", "2000000")))
 
+# ── Power/Level/Gold Zeitreihe (pro Scan) ────────────────────────────────────────────────────
+# Eigener, vom 600-MB-Leaderboard-Cache komplett GETRENNTER Store (eigene Datei, eigener Lock,
+# eigenes kurzes Save-Intervall) -> beruehrt den kritischen Leaderboard-Schreibpfad NICHT.
+# Nur Accounts mit >= 1 Qnt Power (1e18) bekommen Punkte; jeder Lookup/Ingest haengt (gedrosselt
+# auf <=1 Punkt/Min/Account) einen SEKUNDENgenauen Punkt an. Punktformat (kompakt):
+#   [t_sec, level, power_float, gold_float, sword_type, shield_type]
+# sword/shield = Spiel-Typ-Code (kleine Zahl) der AKTIVEN Waffe, -1 wenn unbekannt.
+def _sm_resolve_power_series_path() -> Path:
+    configured = (_os.getenv("SM_POWER_SERIES_FILE") or "").strip()
+    if configured:
+        path = Path(configured)
+        return path if path.is_absolute() else APP_DIR / path
+    if _os.getenv("SPACE_ID"):
+        return Path("/data") / "power_series.json"
+    return APP_DIR / "power_series.json"
+
+SM_POWER_SERIES_PATH = _sm_resolve_power_series_path()
+SM_POWER_SERIES_MIN_POWER = float(_os.getenv("SM_POWER_SERIES_MIN_POWER", "1e18"))   # 1 Qnt
+SM_POWER_SERIES_MIN_GAP = max(0.0, float(_os.getenv("SM_POWER_SERIES_MIN_GAP", "60.0")))       # <=1 Punkt/Min/Account
+SM_POWER_SERIES_MAX_POINTS = max(100, int(_os.getenv("SM_POWER_SERIES_MAX_POINTS", "20000")))  # Sicherheitsdeckel/Account
+SM_POWER_SERIES_SAVE_INTERVAL = max(10.0, float(_os.getenv("SM_POWER_SERIES_SAVE_INTERVAL", "120.0")))
+
 def _env_truthy(name: str, default: bool = False) -> bool:
     value = _os.getenv(name)
     if value is None:
@@ -8206,6 +8228,151 @@ def _sm_lb_publish_rows(rows):
                 pass
 
 
+# ── Power/Level/Gold Zeitreihen-Store (separat, eigener Lock/Datei) ───────────────────────────
+_SM_PSERIES_LOCK = threading.RLock()
+_SM_PSERIES_STATE = None          # {"v":1,"updated_at":ms,"series":{key:{"u":name,"p":[[...]]}}}
+_SM_PSERIES_DIRTY = False
+_SM_PSERIES_LAST_SAVE_MONOTONIC = 0.0
+_SM_PSERIES_FIELDS = ["t", "level", "power", "gold", "sword", "shield"]
+
+
+def _sm_pseries_read_file(path: Path):
+    if not path.exists():
+        return None
+    blob = path.read_bytes()
+    if blob[:2] == b"\x1f\x8b":          # gzip-Magic
+        blob = gzip.decompress(blob)
+    data = json.loads(blob.decode("utf-8"))
+    if isinstance(data, dict) and isinstance(data.get("series"), dict):
+        return data
+    return None
+
+
+def _sm_pseries_load_unlocked():
+    global _SM_PSERIES_STATE
+    if _SM_PSERIES_STATE is not None:
+        return _SM_PSERIES_STATE
+    state = {"v": 1, "updated_at": 0, "series": {}}
+    try:
+        data = _sm_pseries_read_file(SM_POWER_SERIES_PATH)
+        if data:
+            series = {}
+            for key, entry in (data.get("series") or {}).items():
+                if not isinstance(entry, dict) or not isinstance(entry.get("p"), list):
+                    continue
+                series[str(key)] = {"u": str(entry.get("u") or ""), "p": entry.get("p")}
+            state["series"] = series
+            state["updated_at"] = int(data.get("updated_at") or 0)
+    except Exception as exc:  # noqa: BLE001
+        print(f"[power-series load failed] {type(exc).__name__}: {str(exc)[:200]}")
+        state = {"v": 1, "updated_at": 0, "series": {}}
+    _SM_PSERIES_STATE = state
+    return state
+
+
+def _sm_pseries_write_unlocked():
+    state = _sm_pseries_load_unlocked()
+    state["updated_at"] = int(time.time() * 1000)
+    payload = json.dumps(state, ensure_ascii=True, separators=(",", ":")).encode("utf-8")
+    SM_POWER_SERIES_PATH.parent.mkdir(parents=True, exist_ok=True)
+    tmp = SM_POWER_SERIES_PATH.with_suffix(SM_POWER_SERIES_PATH.suffix + ".tmp")
+    tmp.write_bytes(gzip.compress(payload, 4))
+    tmp.replace(SM_POWER_SERIES_PATH)
+
+
+def _sm_pseries_save_unlocked(force: bool = False):
+    """Throttled: schreibt nur alle SM_POWER_SERIES_SAVE_INTERVAL Sekunden (ausser force)."""
+    global _SM_PSERIES_DIRTY, _SM_PSERIES_LAST_SAVE_MONOTONIC
+    if not _SM_PSERIES_DIRTY and not force:
+        return False
+    now = time.monotonic()
+    if not force and _SM_PSERIES_LAST_SAVE_MONOTONIC:
+        if now - _SM_PSERIES_LAST_SAVE_MONOTONIC < SM_POWER_SERIES_SAVE_INTERVAL:
+            return False
+    try:
+        _sm_pseries_write_unlocked()
+        _SM_PSERIES_LAST_SAVE_MONOTONIC = now
+        _SM_PSERIES_DIRTY = False
+        return True
+    except Exception as exc:  # noqa: BLE001
+        print(f"[power-series save failed] {type(exc).__name__}: {str(exc)[:200]}")
+        return False
+
+
+def _sm_pseries_gear_type(value):
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return -1
+
+
+def _sm_power_series_record(result: dict):
+    """Haengt einen Zeitreihen-Punkt an, wenn der Account >= 1 Qnt Power hat. Billig: nur
+    in-memory + throttled Datei-Write. Wird NACH dem Leaderboard-Lock pro Ergebnis aufgerufen
+    (kein verschachtelter Lock)."""
+    global _SM_PSERIES_DIRTY
+    if not isinstance(result, dict) or not result.get("ok"):
+        return
+    power = _sm_float_sort_value(result.get("power_raw"))
+    if power < SM_POWER_SERIES_MIN_POWER:
+        return   # unter 1 Qnt -> gar nicht speichern (Frontend zeigt "unter 1qnt")
+    try:
+        level = int(result.get("level"))
+    except (TypeError, ValueError):
+        return
+    key = _sm_leaderboard_storage_key(result.get("username") or "", result.get("player_id") or "")
+    if not key:
+        return
+    gold = _sm_float_sort_value(result.get("gold_raw"))
+    sword = _sm_pseries_gear_type(result.get("sword_type"))
+    shield = _sm_pseries_gear_type(result.get("shield_type"))
+    now = int(time.time())
+    with _SM_PSERIES_LOCK:
+        state = _sm_pseries_load_unlocked()
+        series = state["series"]
+        entry = series.get(key)
+        if entry is None:
+            entry = {"u": str(result.get("username") or ""), "p": []}
+            series[key] = entry
+        else:
+            uname = str(result.get("username") or "")
+            if uname:
+                entry["u"] = uname
+        pts = entry["p"]
+        if pts:
+            try:
+                if now - int(pts[-1][0]) < SM_POWER_SERIES_MIN_GAP:
+                    return   # zu frisch -> Spam-Schutz (<=1 Punkt/Min/Account)
+            except (TypeError, ValueError, IndexError):
+                pass
+        pts.append([now, level, power, gold, sword, shield])
+        if len(pts) > SM_POWER_SERIES_MAX_POINTS:
+            del pts[: len(pts) - SM_POWER_SERIES_MAX_POINTS]
+        _SM_PSERIES_DIRTY = True
+        _sm_pseries_save_unlocked(force=False)
+
+
+def _sm_power_series_record_many(results):
+    for result in results or []:
+        try:
+            _sm_power_series_record(result)
+        except Exception:  # noqa: BLE001
+            pass
+
+
+def _sm_power_series_flusher():
+    while True:
+        time.sleep(SM_POWER_SERIES_SAVE_INTERVAL)
+        try:
+            with _SM_PSERIES_LOCK:
+                _sm_pseries_save_unlocked(force=False)
+        except Exception:  # noqa: BLE001
+            pass
+
+
+threading.Thread(target=_sm_power_series_flusher, name="sm-power-series-flush", daemon=True).start()
+
+
 def _sm_leaderboard_upsert_lookup_results(results, source: str = "lookup", save: bool = True):
     updated = 0
     changed = []
@@ -8236,6 +8403,8 @@ def _sm_leaderboard_upsert_lookup_results(results, source: str = "lookup", save:
             state["updated_at"] = int(time.time() * 1000)
             if save:
                 _sm_leaderboard_save_unlocked(min_interval=SM_LEADERBOARD_SAVE_INTERVAL)
+    # Zeitreihe (>= 1 Qnt) ausserhalb des Leaderboard-Locks mitschreiben.
+    _sm_power_series_record_many(results)
     # Push changed rows to live (SSE) listeners, but only if someone is watching.
     if changed and _sm_lb_subscriber_count():
         try:
@@ -8273,6 +8442,8 @@ def _sm_leaderboard_ingest_rows(results, save: bool = True):
             state["updated_at"] = int(time.time() * 1000)
             if save:
                 _sm_leaderboard_save_unlocked(min_interval=SM_LEADERBOARD_SAVE_INTERVAL)
+    # Zeitreihe (>= 1 Qnt) ausserhalb des Leaderboard-Locks mitschreiben.
+    _sm_power_series_record_many(results)
     if updated:
         _SM_LB_DIRTY.set()   # neue Daten -> Hintergrund-Warmer rechnet die Views neu
     return updated
@@ -10274,6 +10445,11 @@ def _sm_inventory_result(username: str, inventory: dict):
     if isinstance(sword, dict):
         sword_name = _sm_gear_type_name(SWORD_TYPE_NAMES, SCHWERTER, sword.get("type"))
         result["sword"] = sword_name or ""
+        # Roher Spiel-Typ-Code der AKTIVEN Waffe (kleine Zahl) fuer die Zeitreihe/Translation.
+        try:
+            result["sword_type"] = int(sword.get("type"))
+        except (TypeError, ValueError):
+            pass
         try:
             result["sword_level"] = max(1, int(sword.get("level") or 1))
         except (TypeError, ValueError):
@@ -10286,6 +10462,10 @@ def _sm_inventory_result(username: str, inventory: dict):
     if isinstance(shield, dict):
         shield_name = _sm_gear_type_name(SHIELD_TYPE_NAMES, SCHILDE, shield.get("type"))
         result["shield"] = shield_name or ""
+        try:
+            result["shield_type"] = int(shield.get("type"))
+        except (TypeError, ValueError):
+            pass
         try:
             result["shield_level"] = max(1, int(shield.get("level") or 1))
         except (TypeError, ValueError):
@@ -11963,6 +12143,43 @@ def discord_lookup(username: str, include_rank: bool = True):
     return result
 
 
+@app.get("/power-series")
+def power_series(username: str = "", player_id: str = ""):
+    """Power/Level/Gold-Zeitreihe eines Accounts (nur Accounts mit >= 1 Qnt Power).
+    Bevorzugt player_id (exakter Match); username als Fallback. points = Liste von
+    [t_sec, level, power, gold, sword_type, shield_type]. Fuer das Frontend-Diagramm."""
+    key = _sm_leaderboard_storage_key(username or "", player_id or "")
+    if not key:
+        raise HTTPException(400, "username oder player_id erforderlich.")
+    with _SM_PSERIES_LOCK:
+        state = _sm_pseries_load_unlocked()
+        series = state.get("series") or {}
+        entry = series.get(key)
+        # Fallback: nur Name angegeben, gespeichert ist aber unter id:<player_id>.
+        if entry is None and username and not player_id:
+            want = _discord_lookup_key(username)
+            for cand_key, cand in series.items():
+                if _discord_lookup_key(cand.get("u")) == want:
+                    entry, key = cand, cand_key
+                    break
+        if entry is None:
+            # Kein Datensatz -> entweder noch nie >= 1 Qnt gescannt, oder unter der Schwelle.
+            return {"ok": True, "key": key, "under_1qnt": True, "count": 0,
+                    "fields": _SM_PSERIES_FIELDS, "min_power": SM_POWER_SERIES_MIN_POWER, "points": []}
+        pts = list(entry.get("p") or [])
+        uname = entry.get("u") or username
+    return {
+        "ok": True,
+        "key": key,
+        "username": uname,
+        "under_1qnt": False,
+        "min_power": SM_POWER_SERIES_MIN_POWER,
+        "fields": _SM_PSERIES_FIELDS,
+        "count": len(pts),
+        "points": pts,
+    }
+
+
 @app.get("/live-rank")
 def live_rank(username: str):
     clean = str(username or "").strip()