debug line test

server/stats.py

@@ -0,0 +1,167 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from typing import Any
+from zoneinfo import ZoneInfo
+
+import aiosqlite
+
+from server.config_loader import load_meds_config, today_str
+from server.db import new_id, utc_now_iso
+
+MILESTONE_DEFS: dict[str, str] = {
+    "log_50": "Pharma-Intern: 50 Logs. Dein Arzt wäre stolz. Vielleicht.",
+    "log_100": "Century Club: 100 Logs. Du bist offiziell zuverlässiger als dein WLAN.",
+    "streak_7": "Wochenkrieger: 7 Tage Streak. ADHS.exe has stopped crashing.",
+    "streak_30": "Monatslegende: 30 Tage. Das ist fast schon verdächtig diszipliniert.",
+    "streak_100": "Unmöglichkeitsgrad: 100 Tage Streak. Cheater oder Heilung?",
+}
+
+
+async def count_logs(db: aiosqlite.Connection) -> int:
+    cur = await db.execute("SELECT COUNT(*) FROM intake_log WHERE status = 'taken'")
+    row = await cur.fetchone()
+    await cur.close()
+    return int(row[0]) if row else 0
+
+
+async def get_milestones(db: aiosqlite.Connection) -> list[dict[str, str]]:
+    cur = await db.execute("SELECT id, unlocked_at FROM milestones ORDER BY unlocked_at")
+    rows = await cur.fetchall()
+    await cur.close()
+    return [{"id": r[0], "title": MILESTONE_DEFS.get(r[0], r[0]), "unlocked_at": r[1]} for r in rows]
+
+
+async def unlock_milestone(db: aiosqlite.Connection, milestone_id: str) -> bool:
+    cur = await db.execute("SELECT 1 FROM milestones WHERE id = ?", (milestone_id,))
+    exists = await cur.fetchone()
+    await cur.close()
+    if exists:
+        return False
+    await db.execute(
+        "INSERT INTO milestones (id, unlocked_at) VALUES (?, ?)",
+        (milestone_id, utc_now_iso()),
+    )
+    await db.commit()
+    return True
+
+
+async def check_milestones(db: aiosqlite.Connection, streak: int) -> list[str]:
+    new: list[str] = []
+    total = await count_logs(db)
+    checks = []
+    if total >= 50:
+        checks.append("log_50")
+    if total >= 100:
+        checks.append("log_100")
+    if streak >= 7:
+        checks.append("streak_7")
+    if streak >= 30:
+        checks.append("streak_30")
+    if streak >= 100:
+        checks.append("streak_100")
+    for mid in checks:
+        if await unlock_milestone(db, mid):
+            new.append(mid)
+    return new
+
+
+async def day_compliance(db: aiosqlite.Connection, day: str, slot_ids: list[str]) -> bool:
+    if not slot_ids:
+        return True
+    placeholders = ",".join("?" * len(slot_ids))
+    cur = await db.execute(
+        f"SELECT COUNT(DISTINCT slot_id) FROM intake_log WHERE day = ? AND status = 'taken' AND slot_id IN ({placeholders})",
+        (day, *slot_ids),
+    )
+    row = await cur.fetchone()
+    await cur.close()
+    return int(row[0]) >= len(slot_ids) if row else False
+
+
+async def compute_streak(db: aiosqlite.Connection, tz: ZoneInfo) -> int:
+    config = load_meds_config()
+    slot_ids = [s.id for s in config.slots]
+    streak = 0
+    day = datetime.now(tz).date()
+    while True:
+        day_str = day.isoformat()
+        ok = await day_compliance(db, day_str, slot_ids)
+        if not ok:
+            break
+        streak += 1
+        day -= timedelta(days=1)
+    return streak
+
+
+async def compute_stats(db: aiosqlite.Connection) -> dict[str, Any]:
+    config = load_meds_config()
+    tz = config.timezone
+    slot_ids = [s.id for s in config.slots]
+    streak = await compute_streak(db, tz)
+
+    days = 90
+    taken = 0
+    total = 0
+    today = datetime.now(tz).date()
+    for i in range(days):
+        d = (today - timedelta(days=i)).isoformat()
+        for sid in slot_ids:
+            total += 1
+            cur = await db.execute(
+                "SELECT 1 FROM intake_log WHERE day = ? AND slot_id = ? AND status = 'taken' LIMIT 1",
+                (d, sid),
+            )
+            if await cur.fetchone():
+                taken += 1
+            await cur.close()
+
+    compliance = round((taken / total) * 100, 1) if total else 0.0
+    milestones = await get_milestones(db)
+
+    return {
+        "streak": streak,
+        "compliance_percent": compliance,
+        "total_taken": await count_logs(db),
+        "milestones": milestones,
+    }
+
+
+async def build_history(db: aiosqlite.Connection, days: int = 90) -> dict[str, Any]:
+    config = load_meds_config()
+    tz = config.timezone
+    slot_ids = [s.id for s in config.slots]
+    today = datetime.now(tz).date()
+    days_out: list[dict[str, Any]] = []
+
+    for i in range(days - 1, -1, -1):
+        d = (today - timedelta(days=i)).isoformat()
+        cur = await db.execute(
+            "SELECT slot_id, status FROM intake_log WHERE day = ?",
+            (d,),
+        )
+        rows = await cur.fetchall()
+        await cur.close()
+        by_slot = {sid: st for sid, st in rows}
+        slots_status = {}
+        all_taken = True
+        any_missed = False
+        for sid in slot_ids:
+            st = by_slot.get(sid)
+            slots_status[sid] = st or "none"
+            if st != "taken":
+                all_taken = False
+            if st == "missed":
+                any_missed = True
+        if all_taken and slot_ids:
+            level = "good"
+        elif any_missed:
+            level = "bad"
+        elif any(st != "none" for st in slots_status.values()):
+            level = "partial"
+        else:
+            level = "none"
+        days_out.append({"day": d, "level": level, "slots": slots_status})
+
+    stats = await compute_stats(db)
+    return {"days": days_out, "stats": stats}