"""BAN YARO — Admin / Moderator Backend""" import asyncio import csv import io import logging import os import sys import time import platform from datetime import datetime from zoneinfo import ZoneInfo from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import Response from pydantic import BaseModel, Field from typing import Optional, List from database import db, DB_PATH from auth import get_current_user logger = logging.getLogger(__name__) router = APIRouter() _TZ = ZoneInfo("Europe/Berlin") _start_time = time.time() # Umami token cache _umami_token: dict = {"token": None, "expires": 0.0} async def _umami_auth() -> str: global _umami_token if _umami_token["token"] and time.time() < _umami_token["expires"]: return _umami_token["token"] import httpx url = os.getenv("UMAMI_URL", "").rstrip("/") resp = await httpx.AsyncClient().post( f"{url}/api/auth/login", json={"username": os.getenv("UMAMI_USERNAME"), "password": os.getenv("UMAMI_PASSWORD")}, timeout=10, ) resp.raise_for_status() tok = resp.json()["token"] _umami_token = {"token": tok, "expires": time.time() + 23 * 3600} return tok # Audit-Tabelle anlegen (einmalig beim Import) with db() as _conn: _conn.executescript(""" CREATE TABLE IF NOT EXISTS admin_audit ( id INTEGER PRIMARY KEY AUTOINCREMENT, admin_id INTEGER NOT NULL, admin_name TEXT, action TEXT NOT NULL, target TEXT, detail TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')) ); """) # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ def _audit(conn, admin, action: str, target: str = None, detail: str = None): """Schreibt einen Audit-Eintrag in die admin_audit-Tabelle.""" conn.execute( "INSERT INTO admin_audit (admin_id, admin_name, action, target, detail) VALUES (?,?,?,?,?)", (admin["id"], admin.get("name"), action, target, detail), ) # ------------------------------------------------------------------ # Dependency: Moderator oder Admin # ------------------------------------------------------------------ def require_mod(user=Depends(get_current_user)): if user["rolle"] not in ("admin", "moderator") and not user.get("is_moderator"): raise HTTPException(403, "Kein Zugriff.") return user def require_admin(user=Depends(get_current_user)): if user["rolle"] != "admin": raise HTTPException(403, "Nur Admins.") return user # ------------------------------------------------------------------ # Schemas # ------------------------------------------------------------------ _VALID_TIERS = {"standard", "pro", "breeder", "standard_test", "pro_test", "breeder_test"} class QuarterlyReportBody(BaseModel): year: int quarter: int email: str = Field(..., max_length=254) class UserPatch(BaseModel): rolle: Optional[str] = Field(None, max_length=30) # user | moderator | admin is_moderator: Optional[int] = None is_banned: Optional[int] = None ban_reason: Optional[str] = Field(None, max_length=1000) is_social_media: Optional[int] = None subscription_tier: Optional[str] = Field(None, max_length=50) class WikiEnrichBody(BaseModel): limit: int = 10 class ThreadAdminPatch(BaseModel): is_pinned: Optional[int] = None is_locked: Optional[int] = None is_deleted: Optional[int] = None # ------------------------------------------------------------------ # GET /api/admin/action-items # ------------------------------------------------------------------ @router.get("/action-items") async def action_items(user=Depends(require_mod)): with db() as conn: jobs = conn.execute( "SELECT COUNT(*) FROM job_applications WHERE status IN ('pending','reviewing')" ).fetchone()[0] breeders = conn.execute( "SELECT COUNT(*) FROM users WHERE breeder_status='pending'" ).fetchone()[0] reports = conn.execute( "SELECT COUNT(*) FROM forum_reports WHERE resolved=0" ).fetchone()[0] fotos = conn.execute( "SELECT COUNT(*) FROM wiki_foto_submissions WHERE status='pending'" ).fetchone()[0] poi_edits = conn.execute( "SELECT COUNT(*) FROM osm_poi_edits WHERE status='pending'" ).fetchone()[0] users_today = conn.execute( "SELECT COUNT(*) FROM users WHERE DATE(created_at)=DATE('now')" ).fetchone()[0] try: # JOIN mit users, damit verwaiste Anfragen von gelöschten Usern # nicht mehr im „Zu Erledigen"-Counter auftauchen (Liste filtert # das via JOIN bereits, der Counter tat es früher nicht). upgrades_pending = conn.execute( "SELECT COUNT(*) FROM upgrade_requests r " "JOIN users u ON u.id = r.user_id " "WHERE r.fulfilled_at IS NULL" ).fetchone()[0] except Exception: upgrades_pending = 0 try: invoices_unpaid = conn.execute( "SELECT COUNT(*) FROM invoices WHERE status='sent'" ).fetchone()[0] except Exception: invoices_unpaid = 0 return { "jobs_pending": jobs, "breeder_pending": breeders, "reports_open": reports, "fotos_pending": fotos, "poi_edits_pending": poi_edits, "users_today": users_today, "upgrades_pending": upgrades_pending, "invoices_unpaid": invoices_unpaid, } # ------------------------------------------------------------------ # GET /api/admin/stats # ------------------------------------------------------------------ @router.get("/stats") async def stats(user=Depends(require_mod)): with db() as conn: users_total = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0] users_today = conn.execute( "SELECT COUNT(*) FROM users WHERE DATE(created_at)=DATE('now')" ).fetchone()[0] threads = conn.execute( "SELECT COUNT(*) FROM forum_threads WHERE is_deleted=0" ).fetchone()[0] posts = conn.execute( "SELECT COUNT(*) FROM forum_posts WHERE is_deleted=0" ).fetchone()[0] open_reports = conn.execute( "SELECT COUNT(*) FROM forum_reports WHERE resolved=0" ).fetchone()[0] banned = conn.execute( "SELECT COUNT(*) FROM users WHERE is_banned=1" ).fetchone()[0] dogs_total = conn.execute("SELECT COUNT(*) FROM dogs").fetchone()[0] poison_total = conn.execute( "SELECT COUNT(*) FROM poison WHERE geloest=0 AND expires_at > datetime('now')" ).fetchone()[0] # Erweiterte Metriken push_subscriptions = conn.execute( "SELECT COUNT(*) FROM push_subscriptions" ).fetchone()[0] active_users_7d = conn.execute( "SELECT COUNT(*) FROM users WHERE last_seen > datetime('now', '-7 days')" ).fetchone()[0] # Media aus diary (media_url) + health (datei_url) media_diary = conn.execute( "SELECT COUNT(*) FROM diary WHERE media_url IS NOT NULL AND media_url != ''" ).fetchone()[0] media_health = conn.execute( "SELECT COUNT(*) FROM health WHERE datei_url IS NOT NULL AND datei_url != ''" ).fetchone()[0] media_count = media_diary + media_health routes_total = conn.execute("SELECT COUNT(*) FROM routes").fetchone()[0] events_total = conn.execute("SELECT COUNT(*) FROM events").fetchone()[0] osm_total = conn.execute("SELECT COUNT(*) FROM osm_pois").fetchone()[0] osm_tiles = conn.execute("SELECT COUNT(*) FROM osm_tiles").fetchone()[0] osm_by_type = { row[0]: row[1] for row in conn.execute( "SELECT type, COUNT(*) FROM osm_pois GROUP BY type ORDER BY 2 DESC" ).fetchall() } # User-erstellte POIs nach Typ (type-Spalte kann komma-separiert sein wie 'freilauf,treffpunkt') user_poi_total = conn.execute("SELECT COUNT(*) FROM user_map_pois").fetchone()[0] user_poi_by_type = {} for row in conn.execute("SELECT type FROM user_map_pois").fetchall(): for t in (row[0] or "").split(","): t = t.strip() if t: user_poi_by_type[t] = user_poi_by_type.get(t, 0) + 1 # absteigend sortieren user_poi_by_type = dict(sorted(user_poi_by_type.items(), key=lambda x: x[1], reverse=True)) # KI-Nutzung try: from ki import CLOUD_WEEKLY_LIMIT ki_today = conn.execute( "SELECT COALESCE(SUM(count),0) FROM ki_daily_calls WHERE date=DATE('now')" ).fetchone()[0] ki_week = conn.execute( "SELECT COALESCE(SUM(count),0) FROM ki_daily_calls WHERE date>=DATE('now','-6 days')" ).fetchone()[0] ki_month = conn.execute( "SELECT COALESCE(SUM(count),0) FROM ki_daily_calls WHERE date>=DATE('now','start of month')" ).fetchone()[0] ki_users_today = conn.execute( "SELECT COUNT(DISTINCT user_id) FROM ki_daily_calls WHERE date=DATE('now')" ).fetchone()[0] # Aufschlüsselung nach Quelle (diese Woche) _src_week = { r[0]: r[1] for r in conn.execute( "SELECT source, COALESCE(SUM(count),0) FROM ki_daily_calls " "WHERE date>=DATE('now','-6 days') GROUP BY source" ).fetchall() } ki_cloud_week = _src_week.get("cloud", 0) ki_local_week = _src_week.get("local", 0) ki_luna_week = _src_week.get("luna", 0) # Top-User Cloud diese Woche ki_top_users = [ {"user_id": r[0], "name": r[1], "cloud_calls": r[2]} for r in conn.execute( """SELECT k.user_id, u.name, SUM(k.count) as n FROM ki_daily_calls k JOIN users u ON u.id=k.user_id WHERE k.source='cloud' AND k.date>=DATE('now','-6 days') GROUP BY k.user_id ORDER BY n DESC LIMIT 10""" ).fetchall() ] except Exception: from ki import CLOUD_WEEKLY_LIMIT ki_today = ki_week = ki_month = ki_users_today = 0 ki_cloud_week = ki_local_week = ki_luna_week = 0 ki_top_users = [] # Ausstehende Wiki-Foto-Einreichungen try: pending_fotos = conn.execute( "SELECT COUNT(*) FROM wiki_foto_submissions WHERE status='pending'" ).fetchone()[0] except Exception: pending_fotos = 0 # Social Media Tracking try: social_total = conn.execute("SELECT COUNT(*) FROM social_content").fetchone()[0] social_published = conn.execute( "SELECT COUNT(*) FROM social_content WHERE status='published'" ).fetchone()[0] social_scheduled = conn.execute( "SELECT COUNT(*) FROM social_content WHERE status='scheduled'" ).fetchone()[0] social_ideas = conn.execute( "SELECT COUNT(*) FROM social_content WHERE status='idea'" ).fetchone()[0] social_this_week = conn.execute( "SELECT COUNT(*) FROM social_content WHERE status='published' " "AND published_at >= datetime('now', '-7 days')" ).fetchone()[0] social_by_cat = { row[0]: row[1] for row in conn.execute( "SELECT category, COUNT(*) FROM social_content " "WHERE category IS NOT NULL GROUP BY category ORDER BY 2 DESC" ).fetchall() } social_recent = [dict(r) for r in conn.execute( """SELECT topic, status, platform, format, created_at, published_at, category, ai_score FROM social_content ORDER BY created_at DESC LIMIT 10""" ).fetchall()] except Exception: social_total = social_published = social_scheduled = 0 social_ideas = social_this_week = 0 social_by_cat = {} social_recent = [] return { "users_total": users_total, "users_today": users_today, "threads": threads, "posts": posts, "open_reports": open_reports, "pending_fotos": pending_fotos, "banned": banned, "dogs_total": dogs_total, "poison_active": poison_total, "push_subscriptions": push_subscriptions, "active_users_7d": active_users_7d, "media_count": media_count, "routes_total": routes_total, "events_total": events_total, "osm_total": osm_total, "osm_tiles": osm_tiles, "osm_by_type": osm_by_type, "user_poi_total": user_poi_total, "user_poi_by_type": user_poi_by_type, "ki_today": ki_today, "ki_week": ki_week, "ki_month": ki_month, "ki_users_today": ki_users_today, "ki_cloud_week": ki_cloud_week, "ki_local_week": ki_local_week, "ki_luna_week": ki_luna_week, "ki_cloud_weekly_limit": CLOUD_WEEKLY_LIMIT, "ki_top_users": ki_top_users, "social_total": social_total, "social_published": social_published, "social_scheduled": social_scheduled, "social_ideas": social_ideas, "social_this_week": social_this_week, "social_by_cat": social_by_cat, "social_recent": social_recent, } # ------------------------------------------------------------------ # GET /api/admin/users # ------------------------------------------------------------------ @router.get("/users") async def list_users( q: str = "", rolle: str = "", limit: int = 50, offset: int = 0, user=Depends(require_mod), ): with db() as conn: where = "WHERE 1=1" params = [] if q.strip(): where += " AND (u.name LIKE ? OR u.email LIKE ?)" params.extend([f"%{q.strip()}%", f"%{q.strip()}%"]) if rolle: where += " AND u.rolle = ?" params.append(rolle) # E-Mail nur für Admins; Moderatoren sehen maskierte Version _email_col = "u.email" if user["rolle"] == "admin" else \ "SUBSTR(u.email,1,2)||'***@'||SUBSTR(u.email,INSTR(u.email,'@')+1) AS email" rows = conn.execute(f""" SELECT u.id, u.name, {_email_col}, u.rolle, u.is_premium, u.is_moderator, u.is_banned, u.ban_reason, u.is_founder, u.is_partner, u.founder_number, u.created_at, u.last_login, u.last_seen, u.subscription_tier, (SELECT COUNT(*) FROM dogs d WHERE d.user_id=u.id) AS dog_count, (SELECT COUNT(*) FROM forum_threads t WHERE t.user_id=u.id AND t.is_deleted=0) AS thread_count, ROUND(COALESCE((SELECT SUM(r.distanz_km) FROM routes r WHERE r.user_id=u.id), 0), 1) AS total_km, (SELECT COUNT(*) FROM routes r WHERE r.user_id=u.id) AS route_count, (SELECT COUNT(*) FROM user_map_pois p WHERE p.user_id=u.id) AS poi_count, (SELECT MAX(r.created_at) FROM routes r WHERE r.user_id=u.id) AS last_route FROM users u {where} ORDER BY u.created_at DESC LIMIT ? OFFSET ? """, [*params, limit, offset]).fetchall() total = conn.execute(f""" SELECT COUNT(*) FROM users u {where} """, params).fetchone()[0] return {"users": [dict(r) for r in rows], "total": total} # ------------------------------------------------------------------ # PATCH /api/admin/users/{id} — Rolle, Sperre # ------------------------------------------------------------------ @router.patch("/users/{uid}") async def patch_user(uid: int, data: UserPatch, user=Depends(require_mod)): # Rollenwechsel + Privileg-Flags nur für Admins if data.rolle is not None and user["rolle"] != "admin": raise HTTPException(403, "Rollenwechsel nur für Admins.") if data.rolle and data.rolle not in ("user", "moderator", "admin"): raise HTTPException(400, "Ungültige Rolle.") if data.is_moderator is not None and user["rolle"] != "admin": raise HTTPException(403, "is_moderator darf nur von Admins geändert werden.") if data.is_social_media is not None and user["rolle"] != "admin": raise HTTPException(403, "is_social_media darf nur von Admins geändert werden.") if data.subscription_tier is not None and user["rolle"] != "admin": raise HTTPException(403, "subscription_tier darf nur von Admins geändert werden.") if data.subscription_tier is not None and data.subscription_tier not in _VALID_TIERS: raise HTTPException(400, f"Ungültiger Tier. Erlaubt: {', '.join(sorted(_VALID_TIERS))}") with db() as conn: target = conn.execute("SELECT id, rolle, name FROM users WHERE id=?", (uid,)).fetchone() if not target: raise HTTPException(404, "User nicht gefunden.") # Mods dürfen keine Admins sperren if target["rolle"] == "admin" and user["rolle"] != "admin": raise HTTPException(403, "Admins können nur von Admins verwaltet werden.") updates = data.model_dump(exclude_none=True) if not updates: raise HTTPException(400, "Keine Änderungen.") # is_moderator aus rolle ableiten wenn rolle gesetzt wird if "rolle" in updates: updates["is_moderator"] = 1 if updates["rolle"] in ("moderator", "admin") else 0 cols = ", ".join(f"{k}=?" for k in updates) conn.execute(f"UPDATE users SET {cols} WHERE id=?", [*updates.values(), uid]) row = conn.execute( "SELECT id, name, email, rolle, is_moderator, is_banned, ban_reason, subscription_tier FROM users WHERE id=?", (uid,) ).fetchone() # Audit detail_parts = [] if "is_banned" in updates: detail_parts.append("gesperrt" if updates["is_banned"] else "entsperrt") if "rolle" in updates: detail_parts.append(f"Rolle→{updates['rolle']}") if "subscription_tier" in updates: detail_parts.append(f"Tier→{updates['subscription_tier']}") _audit(conn, user, "user_patch", f"user:{uid} ({target['name']})", ", ".join(detail_parts) or None) return dict(row) # ------------------------------------------------------------------ # DELETE /api/admin/users/{id} — Account löschen (Admin only) # ------------------------------------------------------------------ @router.delete("/users/{uid}", status_code=204) async def delete_user(uid: int, user=Depends(require_admin)): with db() as conn: target = conn.execute("SELECT id, rolle, name FROM users WHERE id=?", (uid,)).fetchone() if not target: raise HTTPException(404, "User nicht gefunden.") if target["id"] == user["id"]: raise HTTPException(400, "Du kannst deinen eigenen Account nicht löschen.") # Hund-zentrierte Daten zuerst löschen, sonst hängt der FK an der Hunde-ID dog_ids = [r["id"] for r in conn.execute( "SELECT id FROM dogs WHERE user_id=?", (uid,)).fetchall()] for did in dog_ids: conn.execute("DELETE FROM diary WHERE dog_id=?", (did,)) conn.execute("DELETE FROM health WHERE dog_id=?", (did,)) conn.execute("DELETE FROM training_sessions WHERE dog_id=?", (did,)) conn.execute("DELETE FROM training_streaks WHERE dog_id=?", (did,)) conn.execute("DELETE FROM expenses WHERE dog_id=?", (did,)) conn.execute("DELETE FROM dogs WHERE user_id=?", (uid,)) conn.execute("DELETE FROM upgrade_requests WHERE user_id=?", (uid,)) conn.execute("DELETE FROM push_subscriptions WHERE user_id=?", (uid,)) conn.execute("DELETE FROM notifications WHERE user_id=?", (uid,)) conn.execute("DELETE FROM forum_posts WHERE user_id=?", (uid,)) conn.execute("DELETE FROM users WHERE id=?", (uid,)) _audit(conn, user, "user_delete", f"user:{uid} ({target['name']})") # ------------------------------------------------------------------ # GET /api/admin/forum/threads — alle Threads inkl. gelöschte # ------------------------------------------------------------------ @router.get("/forum/threads") async def admin_threads( q: str = "", deleted: int = 0, limit: int = 50, offset: int = 0, user=Depends(require_mod), ): with db() as conn: where = "WHERE 1=1" params = [] if not deleted: where += " AND t.is_deleted=0" if q.strip(): where += " AND (t.titel LIKE ? OR t.text LIKE ?)" params.extend([f"%{q.strip()}%", f"%{q.strip()}%"]) rows = conn.execute(f""" SELECT t.id, t.kategorie, t.titel, SUBSTR(t.text,1,100) AS text_preview, t.antworten, t.likes, t.views, t.is_pinned, t.is_locked, t.is_deleted, t.created_at, u.id AS user_id, u.name AS autor_name FROM forum_threads t LEFT JOIN users u ON u.id=t.user_id {where} ORDER BY t.created_at DESC LIMIT ? OFFSET ? """, [*params, limit, offset]).fetchall() total = conn.execute(f""" SELECT COUNT(*) FROM forum_threads t {where} """, params).fetchone()[0] return {"threads": [dict(r) for r in rows], "total": total} # ------------------------------------------------------------------ # PATCH /api/admin/forum/threads/{id} # ------------------------------------------------------------------ @router.patch("/forum/threads/{tid}") async def admin_patch_thread(tid: int, data: ThreadAdminPatch, user=Depends(require_mod)): with db() as conn: thread = conn.execute("SELECT id, titel FROM forum_threads WHERE id=?", (tid,)).fetchone() if not thread: raise HTTPException(404, "Thread nicht gefunden.") updates = data.model_dump(exclude_none=True) if not updates: raise HTTPException(400, "Keine Änderungen.") cols = ", ".join(f"{k}=?" for k in updates) conn.execute(f"UPDATE forum_threads SET {cols} WHERE id=?", [*updates.values(), tid]) _audit(conn, user, "thread_patch", f"thread:{tid}", str(updates)) return {"ok": True} # ------------------------------------------------------------------ # DELETE /api/admin/forum/threads/{id} # ------------------------------------------------------------------ @router.delete("/forum/threads/{tid}", status_code=204) async def admin_delete_thread(tid: int, user=Depends(require_mod)): with db() as conn: thread = conn.execute("SELECT id, titel FROM forum_threads WHERE id=?", (tid,)).fetchone() if not thread: raise HTTPException(404, "Thread nicht gefunden.") conn.execute("UPDATE forum_threads SET is_deleted=1 WHERE id=?", (tid,)) _audit(conn, user, "thread_delete", f"thread:{tid} ({thread['titel'][:60]})") # ------------------------------------------------------------------ # DELETE /api/admin/forum/posts/{id} # ------------------------------------------------------------------ @router.delete("/forum/posts/{pid}", status_code=204) async def admin_delete_post(pid: int, user=Depends(require_mod)): with db() as conn: post = conn.execute("SELECT * FROM forum_posts WHERE id=?", (pid,)).fetchone() if not post: raise HTTPException(404, "Beitrag nicht gefunden.") conn.execute("UPDATE forum_posts SET is_deleted=1 WHERE id=?", (pid,)) conn.execute( "UPDATE forum_threads SET antworten=MAX(0,antworten-1) WHERE id=?", (post["thread_id"],) ) _audit(conn, user, "post_delete", f"post:{pid} thread:{post['thread_id']}") # ------------------------------------------------------------------ # GET /api/admin/reports — offene Meldungen # ------------------------------------------------------------------ @router.get("/reports") async def admin_reports(user=Depends(require_mod)): with db() as conn: rows = conn.execute(""" SELECT r.id, r.target_type, r.target_id, r.grund, r.resolved, r.created_at, u.name AS melder_name, CASE r.target_type WHEN 'thread' THEN (SELECT t.titel FROM forum_threads t WHERE t.id=r.target_id) WHEN 'post' THEN (SELECT SUBSTR(p.text,1,80) FROM forum_posts p WHERE p.id=r.target_id) END AS content_preview FROM forum_reports r LEFT JOIN users u ON u.id=r.user_id ORDER BY r.resolved ASC, r.created_at DESC LIMIT 100 """).fetchall() return [dict(r) for r in rows] # ------------------------------------------------------------------ # PATCH /api/admin/reports/{id} — erledigen / wiedereröffnen # ------------------------------------------------------------------ @router.patch("/reports/{rid}") async def admin_resolve_report(rid: int, user=Depends(require_mod)): with db() as conn: r = conn.execute("SELECT resolved FROM forum_reports WHERE id=?", (rid,)).fetchone() if not r: raise HTTPException(404, "Meldung nicht gefunden.") new_state = 0 if r["resolved"] else 1 conn.execute( "UPDATE forum_reports SET resolved=? WHERE id=?", (new_state, rid) ) _audit(conn, user, "report_resolve" if new_state else "report_reopen", f"report:{rid}") return {"ok": True} # ------------------------------------------------------------------ # GET /api/admin/scheduler/jobs # ------------------------------------------------------------------ @router.get("/scheduler/jobs") async def scheduler_jobs(user=Depends(require_admin)): from scheduler import _scheduler jobs = [] for job in _scheduler.get_jobs(): next_run = job.next_run_time jobs.append({ "id": job.id, "name": job.name, "next_run_time": next_run.isoformat() if next_run else None, "trigger": str(job.trigger), }) return jobs # ------------------------------------------------------------------ # POST /api/admin/scheduler/trigger/{job_id} # ------------------------------------------------------------------ @router.post("/scheduler/trigger/{job_id}") async def scheduler_trigger(job_id: str, user=Depends(require_admin)): from scheduler import _scheduler, _TZ as SCHED_TZ job = _scheduler.get_job(job_id) if not job: raise HTTPException(404, f"Job '{job_id}' nicht gefunden.") job.modify(next_run_time=datetime.now(tz=SCHED_TZ)) with db() as conn: _audit(conn, user, "scheduler_trigger", f"job:{job_id}") return {"ok": True, "job_id": job_id} # ------------------------------------------------------------------ # GET /api/admin/referrals — User-wirbt-User Top 100 # ------------------------------------------------------------------ @router.get("/referrals") async def referral_stats(user=Depends(require_mod)): with db() as conn: # Top-Werber mit Anzahl top = conn.execute(""" SELECT r.id, r.name, r.email, COUNT(u.id) AS invited_count, r.created_at AS member_since FROM users u JOIN users r ON r.id = u.referred_by GROUP BY r.id ORDER BY invited_count DESC LIMIT 100 """).fetchall() # Alle Einladungen (für Detail-Ansicht) invites = conn.execute(""" SELECT u.id, u.name, u.email, u.created_at, r.id AS referrer_id, r.name AS referrer_name FROM users u JOIN users r ON r.id = u.referred_by ORDER BY u.created_at DESC LIMIT 500 """).fetchall() total_users = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0] total_referred = conn.execute( "SELECT COUNT(*) FROM users WHERE referred_by IS NOT NULL" ).fetchone()[0] return { "top_referrers": [dict(r) for r in top], "recent_invites": [dict(r) for r in invites], "total_users": total_users, "total_referred": total_referred, "viral_factor": round(total_referred / max(total_users - total_referred, 1), 2), } # ------------------------------------------------------------------ # GET /api/admin/ki/history — 30-Tage-Verlauf + Top-User (all-time) # ------------------------------------------------------------------ @router.get("/ki/history") async def ki_history(user=Depends(require_mod)): with db() as conn: daily = conn.execute(""" SELECT date, SUM(count) AS total, SUM(CASE WHEN source='cloud' THEN count ELSE 0 END) AS cloud, SUM(CASE WHEN source='local' THEN count ELSE 0 END) AS local, SUM(CASE WHEN source='luna' THEN count ELSE 0 END) AS luna FROM ki_daily_calls WHERE date >= DATE('now', '-29 days') GROUP BY date ORDER BY date ASC """).fetchall() top_users = conn.execute(""" SELECT u.name, u.email, SUM(k.count) AS total, SUM(CASE WHEN k.source='cloud' THEN k.count ELSE 0 END) AS cloud, MAX(k.date) AS last_date FROM ki_daily_calls k JOIN users u ON u.id = k.user_id GROUP BY k.user_id ORDER BY total DESC LIMIT 15 """).fetchall() return { "daily_history": [{"date": r["date"], "total": r["total"], "cloud": r["cloud"], "local": r["local"], "luna": r["luna"]} for r in daily], "top_users": [{"name": r["name"], "email": r["email"], "total": r["total"], "cloud": r["cloud"], "last_date": r["last_date"]} for r in top_users], } # ------------------------------------------------------------------ # GET /api/admin/ki/status — lokale LLM-Erreichbarkeit prüfen # ------------------------------------------------------------------ @router.get("/ki/status") async def ki_status(user=Depends(require_mod)): import httpx from ki import KI_MODE, LOCAL_BASE_URL, LOCAL_MODEL, CLOUD_MODEL, ANTHROPIC_KEY result = { "mode": KI_MODE, "local_url": LOCAL_BASE_URL if KI_MODE != "off" else None, "local_model_config": LOCAL_MODEL, "local_reachable": False, "local_model_loaded": None, "cloud_model": CLOUD_MODEL, "cloud_key_set": bool(ANTHROPIC_KEY), } if KI_MODE != "off": try: async with httpx.AsyncClient(timeout=3.0) as client: resp = await client.get(f"{LOCAL_BASE_URL}/models") if resp.status_code == 200: data = resp.json() models = data.get("data", []) result["local_reachable"] = True if models: result["local_model_loaded"] = models[0].get("id") except Exception: pass return result # ------------------------------------------------------------------ # GET /api/admin/system # ------------------------------------------------------------------ @router.get("/system") async def system_info(user=Depends(require_admin)): # DB-Größe try: db_size_mb = os.path.getsize(DB_PATH) / 1024 ** 2 except OSError: db_size_mb = 0.0 # Media-Größe (rekursiv) media_dir = os.getenv("MEDIA_DIR", "/data/media") media_size_bytes = 0 try: for dirpath, _dirs, files in os.walk(media_dir): for fname in files: try: media_size_bytes += os.path.getsize(os.path.join(dirpath, fname)) except OSError: pass except OSError: pass media_size_mb = media_size_bytes / 1024 ** 2 # Disk-Info disk_total_gb = 0.0 disk_free_gb = 0.0 try: st = os.statvfs(DB_PATH) disk_total_gb = st.f_blocks * st.f_frsize / 1024 ** 3 disk_free_gb = st.f_bavail * st.f_frsize / 1024 ** 3 except (OSError, AttributeError): pass # SW-Cache-Version aus sw.js lesen sw_version = "?" try: import re as _re static_dir = os.getenv("STATIC_DIR", "/app/static") sw_content = open(os.path.join(static_dir, "sw.js")).readline() m = _re.search(r"'(by-v\d+)'", sw_content) if m: sw_version = m.group(1) except Exception: pass return { "db_size_mb": round(db_size_mb, 2), "media_size_mb": round(media_size_mb, 2), "uptime_seconds": int(time.time() - _start_time), "python_version": sys.version.split()[0], "disk_total_gb": round(disk_total_gb, 2), "disk_free_gb": round(disk_free_gb, 2), "sw_version": sw_version, } # ------------------------------------------------------------------ # GET /api/admin/logs # ------------------------------------------------------------------ @router.get("/logs") async def get_logs(lines: int = 200, level: str = "", user=Depends(require_admin)): from main import log_buffer entries = list(log_buffer) if level: entries = [e for e in entries if e['l'] == level.upper()] return entries[-lines:] # ------------------------------------------------------------------ # GET /api/admin/audit # ------------------------------------------------------------------ @router.get("/audit") async def audit_log(limit: int = 50, user=Depends(require_admin)): with db() as conn: rows = conn.execute( """ SELECT id, admin_id, admin_name, action, target, detail, created_at FROM admin_audit ORDER BY id DESC LIMIT ? """, (min(limit, 200),), ).fetchall() return [dict(r) for r in rows] # ------------------------------------------------------------------ # GET /api/admin/analytics — Umami-Proxy # ------------------------------------------------------------------ @router.get("/analytics") async def get_analytics(user=Depends(require_mod)): import httpx from datetime import timedelta url = os.getenv("UMAMI_URL", "").rstrip("/") site_id = os.getenv("UMAMI_SITE_ID", "") if not url or not site_id: raise HTTPException(503, "Umami nicht konfiguriert.") try: token = await _umami_auth() except Exception as e: raise HTTPException(503, f"Umami-Login fehlgeschlagen: Bitte UMAMI_USERNAME und UMAMI_PASSWORD in .env prüfen. ({e})") headers = {"Authorization": f"Bearer {token}"} now = datetime.now(_TZ) today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) week_start = today_start - timedelta(days=6) month_start = today_start - timedelta(days=29) year_start = today_start - timedelta(days=364) now_ms = int(now.timestamp() * 1000) today_ms = int(today_start.timestamp() * 1000) week_ms = int(week_start.timestamp() * 1000) month_ms = int(month_start.timestamp() * 1000) year_ms = int(year_start.timestamp() * 1000) async with httpx.AsyncClient(timeout=15) as c: r_today, r_week, r_month, r_pv_month, r_pv_year, r_pages, r_refs = await asyncio.gather( c.get(f"{url}/api/websites/{site_id}/stats", params={"startAt": today_ms, "endAt": now_ms}, headers=headers), c.get(f"{url}/api/websites/{site_id}/stats", params={"startAt": week_ms, "endAt": now_ms}, headers=headers), c.get(f"{url}/api/websites/{site_id}/stats", params={"startAt": month_ms, "endAt": now_ms}, headers=headers), c.get(f"{url}/api/websites/{site_id}/pageviews", params={"startAt": month_ms, "endAt": now_ms, "unit": "day", "timezone": "Europe/Berlin"}, headers=headers), c.get(f"{url}/api/websites/{site_id}/pageviews", params={"startAt": year_ms, "endAt": now_ms, "unit": "month", "timezone": "Europe/Berlin"}, headers=headers), c.get(f"{url}/api/websites/{site_id}/metrics", params={"startAt": month_ms, "endAt": now_ms, "type": "url", "limit": 10}, headers=headers), c.get(f"{url}/api/websites/{site_id}/metrics", params={"startAt": month_ms, "endAt": now_ms, "type": "referrer", "limit": 8}, headers=headers), ) # Monatliche Neuanmeldungen aus lokaler DB (letzte 12 Monate) with db() as conn: reg_rows = conn.execute(""" SELECT strftime('%Y-%m', created_at) AS month, COUNT(*) AS count FROM users WHERE created_at >= date('now', '-12 months') GROUP BY month ORDER BY month ASC """).fetchall() monthly_registrations = [{"month": r["month"], "count": r["count"]} for r in reg_rows] def _to_list(r): j = r.json() if isinstance(j, list): return j if isinstance(j, dict): return j.get("data", j.get("metrics", [])) return [] return { "today": r_today.json(), "week": r_week.json(), "month": r_month.json(), "pageviews": r_pv_month.json(), "pageviews_year": r_pv_year.json(), "monthly_registrations": monthly_registrations, "top_pages": _to_list(r_pages), "referrers": _to_list(r_refs), } # ------------------------------------------------------------------ # POST /api/admin/wiki/enrich — KI-Rassen-Anreicherung anstoßen # ------------------------------------------------------------------ @router.post("/wiki/enrich") async def wiki_enrich(data: WikiEnrichBody, user=Depends(require_mod)): from scraper.breed_enricher import enrich_breeds limit = max(1, min(data.limit, 100)) enriched = await enrich_breeds(limit) with db() as conn: remaining = conn.execute( "SELECT COUNT(*) FROM wiki_rassen WHERE ki_enriched=0" ).fetchone()[0] return {"enriched": enriched, "remaining": remaining} # ------------------------------------------------------------------ # GET /api/admin/wiki/evaluate — LLM-as-Judge Qualitätsbewertung # ------------------------------------------------------------------ @router.get("/wiki/evaluate") async def wiki_evaluate(sample: int = 20, user=Depends(require_mod)): from scraper.breed_evaluator import evaluate_enrichment sample = max(5, min(sample, 50)) return await evaluate_enrichment(sample_size=sample) # ------------------------------------------------------------------ # GET /api/admin/wiki/enrichment-status — Enrichment-Statistik # ------------------------------------------------------------------ @router.get("/wiki/enrichment-status") async def wiki_enrichment_status(user=Depends(require_mod)): with db() as conn: total = conn.execute("SELECT COUNT(*) FROM wiki_rassen").fetchone()[0] enriched = conn.execute("SELECT COUNT(*) FROM wiki_rassen WHERE ki_enriched=1").fetchone()[0] no_wiki = conn.execute("SELECT COUNT(*) FROM wiki_rassen WHERE ki_enriched=2").fetchone()[0] pending = conn.execute("SELECT COUNT(*) FROM wiki_rassen WHERE ki_enriched=0").fetchone()[0] by_model = { row[0] or "unbekannt": row[1] for row in conn.execute( "SELECT ki_model, COUNT(*) FROM wiki_rassen " "WHERE ki_enriched=1 GROUP BY ki_model ORDER BY 2 DESC" ).fetchall() } with_photo = conn.execute( "SELECT COUNT(*) FROM wiki_rassen WHERE foto_url IS NOT NULL AND foto_url != ''" ).fetchone()[0] return { "total": total, "enriched": enriched, "no_wiki": no_wiki, "pending": pending, "with_photo": with_photo, "by_model": by_model, } # ------------------------------------------------------------------ # POST /api/admin/wiki/fetch-photos — Wiki-Fotos laden # ------------------------------------------------------------------ @router.post("/wiki/fetch-photos") async def wiki_fetch_photos(limit: int = 50, user=Depends(require_mod)): import asyncio, subprocess proc = await asyncio.create_subprocess_exec( "python3", "/app/scraper/fetch_wiki_images.py", "--limit", str(limit), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await proc.communicate() lines = (stdout + stderr).decode() found = lines.count("Foto gespeichert") return {"launched": True, "found": found, "log": lines[-2000:]} # ------------------------------------------------------------------ # DELETE /api/admin/wiki/zuchter/{id} — Züchter-Eintrag löschen (Admin/Mod) # ------------------------------------------------------------------ @router.get("/social") async def admin_social_stats(user=Depends(require_mod)): """Social-Media-Übersicht für Admins — alle Manager, alle Plattformen.""" with db() as conn: # Pro Manager: Name + Counts managers = conn.execute(""" SELECT u.id, u.name, COUNT(sc.id) AS total, COUNT(CASE WHEN sc.status='published' THEN 1 END) AS published, COUNT(CASE WHEN sc.status='idea' THEN 1 END) AS ideas, COUNT(CASE WHEN sc.status='scheduled' THEN 1 END) AS scheduled, COUNT(CASE WHEN sc.post_url IS NOT NULL AND sc.post_url != '' AND sc.status='published' THEN 1 END) AS with_link FROM users u JOIN social_content sc ON sc.created_by = u.id GROUP BY u.id ORDER BY published DESC """).fetchall() # Veröffentlichte Posts nach Plattform by_platform = conn.execute(""" SELECT platform, COUNT(*) AS n FROM social_content WHERE status='published' GROUP BY platform ORDER BY n DESC """).fetchall() # Veröffentlichte Posts nach Monat (letzte 6 Monate) by_month = conn.execute(""" SELECT strftime('%Y-%m', published_at) AS monat, COUNT(*) AS n FROM social_content WHERE status='published' AND published_at IS NOT NULL GROUP BY monat ORDER BY monat DESC LIMIT 6 """).fetchall() # Letzte veröffentlichte Posts mit Link (für Abrechnung/Nachweis) recent_published = conn.execute(""" SELECT sc.id, sc.topic, sc.platform, sc.category, sc.published_at, sc.post_url, sc.ai_score, u.name AS manager FROM social_content sc LEFT JOIN users u ON u.id = sc.created_by WHERE sc.status = 'published' ORDER BY sc.published_at DESC LIMIT 50 """).fetchall() return { "managers": [dict(r) for r in managers], "by_platform": [dict(r) for r in by_platform], "by_month": [dict(r) for r in by_month], "recent_published": [dict(r) for r in recent_published], } @router.delete("/wiki/zuchter/{zuchter_id}", status_code=204) async def admin_delete_zuchter(zuchter_id: int, user=Depends(require_mod)): with db() as conn: row = conn.execute( "SELECT id FROM wiki_zuchter WHERE id=?", (zuchter_id,) ).fetchone() if not row: raise HTTPException(404, "Züchter nicht gefunden.") conn.execute("DELETE FROM wiki_zuchter WHERE id=?", (zuchter_id,)) _audit(conn, user, "wiki_zuchter_delete", f"zuchter:{zuchter_id}") # ------------------------------------------------------------------ # POST /api/admin/media/generate-previews — Previews für Bestandsmedien # ------------------------------------------------------------------ @router.get("/ors/stats") async def ors_stats(user=Depends(require_mod)): """ORS-Routenvorschlag Statistiken für Admin-Panel.""" with db() as conn: # Heute today = __import__('datetime').date.today().isoformat() today_row = conn.execute( "SELECT COALESCE(count,0) as count FROM ors_daily_total WHERE date=?", (today,) ).fetchone() today_count = today_row["count"] if today_row else 0 # Letzte 30 Tage (Verlauf) daily_history = conn.execute(""" SELECT date, count FROM ors_daily_total WHERE date >= DATE('now', '-29 days') ORDER BY date ASC """).fetchall() # Letzte 8 Wochen (Wochensummen aus route_suggest_usage) weekly_totals = conn.execute(""" SELECT week, SUM(count) as count FROM route_suggest_usage WHERE week >= DATE('now', '-56 days') GROUP BY week ORDER BY week ASC """).fetchall() # Top-Nutzer (alle Zeiten) top_users = conn.execute(""" SELECT u.name, u.email, SUM(r.count) as total, MAX(r.week) as last_week FROM route_suggest_usage r JOIN users u ON u.id = r.user_id GROUP BY r.user_id ORDER BY total DESC LIMIT 15 """).fetchall() return { "today_count": today_count, "today_limit": 2000, "daily_history": [{"date": r["date"], "count": r["count"]} for r in daily_history], "weekly_totals": [{"week": r["week"], "count": r["count"]} for r in weekly_totals], "top_users": [{"name": r["name"], "email": r["email"], "total": r["total"], "last_week": r["last_week"]} for r in top_users], } @router.post("/media/generate-previews") async def generate_media_previews(user=Depends(require_admin)): """Generiert fehlende _preview.webp für alle Bilder in /data/media.""" import logging as _log from media_utils import generate_preview, _PREVIEW_EXTS _logger = _log.getLogger(__name__) MEDIA_DIR = os.getenv("MEDIA_DIR", "/data/media") generated = 0 skipped = 0 errors = 0 dirs_info = {} for subdir in ("diary", "forum", "breeds", "breeds/gallery", "breeds/submissions"): folder = os.path.join(MEDIA_DIR, subdir) if not os.path.isdir(folder): dirs_info[subdir] = "not found" continue files = os.listdir(folder) dirs_info[subdir] = f"{len(files)} files" for fname in files: low = fname.lower() if "_preview" in low or "_thumb" in low: continue base, ext = os.path.splitext(fname) if ext.lower() not in _PREVIEW_EXTS: continue preview_path = os.path.join(folder, base + "_preview.webp") if os.path.exists(preview_path): skipped += 1 continue try: data = open(os.path.join(folder, fname), "rb").read() preview = generate_preview(data, ext) if preview: open(preview_path, "wb").write(preview) generated += 1 else: skipped += 1 _logger.warning(f"Preview None für {subdir}/{fname}") except Exception as exc: errors += 1 _logger.error(f"Preview-Fehler {subdir}/{fname}: {exc}") _logger.info(f"generate-previews: {generated} neu, {skipped} vorhanden, {errors} Fehler | dirs: {dirs_info}") return {"generated": generated, "skipped": skipped, "errors": errors, "dirs": dirs_info} # ------------------------------------------------------------------ # GET /api/admin/upgrade-requests — offene Upgrade-Anfragen # POST /api/admin/upgrade-requests/{id}/fulfill — Tier setzen + Mail # ------------------------------------------------------------------ @router.get("/upgrade-requests") async def list_upgrade_requests(user=Depends(require_admin)): with db() as conn: rows = conn.execute(""" SELECT r.id, r.user_id, r.tier, r.message, r.created_at, r.fulfilled_at, u.name, u.email, u.billing_address, u.is_founder, u.is_founder_pending, u.referred_by, COALESCE((SELECT COUNT(*) FROM users WHERE referred_by=u.id), 0) AS referral_count, (SELECT id FROM invoices i WHERE i.user_id=r.user_id AND i.status IN ('draft','sent') ORDER BY i.created_at DESC LIMIT 1) AS existing_invoice_id, (SELECT invoice_number FROM invoices i WHERE i.user_id=r.user_id AND i.status IN ('draft','sent') ORDER BY i.created_at DESC LIMIT 1) AS existing_invoice_number, (SELECT status FROM invoices i WHERE i.user_id=r.user_id AND i.status IN ('draft','sent') ORDER BY i.created_at DESC LIMIT 1) AS existing_invoice_status FROM upgrade_requests r JOIN users u ON u.id = r.user_id ORDER BY r.fulfilled_at IS NOT NULL, r.created_at DESC LIMIT 100 """).fetchall() result = [] for r in rows: d = dict(r) d_info = _get_discount_info(conn, r["user_id"]) d["discount_pct"] = d_info["discount_pct"] d["discount_reason"] = d_info["reason"] result.append(d) return result @router.get("/users/{user_id}/discount") def get_user_discount(user_id: int, admin=Depends(require_admin)): with db() as conn: return _get_discount_info(conn, user_id) @router.post("/upgrade-requests/{req_id}/fulfill") async def fulfill_upgrade_request(req_id: int, user=Depends(require_admin)): with db() as conn: req = conn.execute( "SELECT r.*, u.name, u.email, u.subscription_tier AS old_tier FROM upgrade_requests r JOIN users u ON u.id=r.user_id WHERE r.id=?", (req_id,) ).fetchone() if not req: raise HTTPException(404, "Anfrage nicht gefunden.") if req["fulfilled_at"]: raise HTTPException(400, "Bereits erledigt.") if req["tier"] not in _VALID_TIERS: raise HTTPException(400, "Ungültiger Tier.") from datetime import timedelta expires_at = (datetime.now(_TZ) + timedelta(days=365)).strftime('%Y-%m-%dT%H:%M:%SZ') # is_premium-Flag synchron halten: 1 für pro/breeder, sonst 0. # Sonst zeigt das Frontend „Kostenlos"-Badge obwohl Tier upgraded # ist (Settings-Header las is_premium statt subscription_tier). _is_prem = 1 if req["tier"] in ("pro", "breeder") else 0 conn.execute( """UPDATE users SET subscription_tier=?, subscription_expires_at=?, subscription_cancelled_at=NULL, is_premium=? WHERE id=?""", (req["tier"], expires_at, _is_prem, req["user_id"]) ) conn.execute( "UPDATE upgrade_requests SET fulfilled_at=strftime('%Y-%m-%dT%H:%M:%fZ','now'), fulfilled_by=? WHERE id=?", (user["id"], req_id) ) _audit(conn, user, "fulfill_upgrade", f"user:{req['user_id']}", f"tier={req['tier']}") tier_labels = {"pro": "Ban Yaro Pro", "breeder": "Züchter"} tier_label = tier_labels.get(req["tier"], req["tier"]) tier_price = {"pro": "29 €/Jahr", "breeder": "49 €/Jahr"}.get(req["tier"], "") _features_pro = [ ("users", "Mehrere Hunde", "Bis zu 10 Hunde mit getrennten Trainings-, Gesundheits- und Ernährungsdaten"), ("fork-knife", "Ernährung", "Kalorienbedarf-Rechner, BARF-Guide, vollständige Giftliste, KI-Ernährungsberater"), ("paw-print", "Gassi-Treffen", "Hundefotos & Rasse der Teilnehmer sichtbar, Fotos hochladen und teilen"), ("chat-circle-dots", "Direktnachrichten & Chat", "Schreibe direkt mit anderen Hundebesitzern"), ("handshake", "Playdate", "Spielkameraden in der Nähe finden und verabreden"), ("airplane", "Reise-Checkliste", "Editierbare Checkliste + EU-Länder-Einreiseregeln"), ("note-pencil", "Notizblock mit KI", "KI erkennt Muster in deinen Notizen und macht Vorschläge"), ("map-trifold", "Erweiterte Karten-Layer", "Regenradar, Temperatur-Layer und weitere Kartenmodi"), ] _features_breeder = [ ("check-circle", "Alle Pro-Features inklusive", "Mehrere Hunde, Ernährung, Gassi-Community, Chat, Playdate, Reise, Karten"), ("tree-structure", "Zuchtkartei", "Gesundheitstests (HD, ED, OCD, Augen, Herz, Patella, ZTP), Gentests (MDR1, PRA, DM, vWD), Titel"), ("notebook", "Wurfverwaltung", "Würfe und Welpen verwalten, Gewichtsverlauf, Fotos, Kaufvertrag automatisch ausfüllen"), ("list-bullets", "Warteliste", "Interessenten mit Präferenzen pro Zuchthündin verwalten"), ("thermometer", "Läufigkeit & Trächtigkeit", "Zykluskalender, Progesterontests, Deckdaten, automatische Meilensteinberechnung"), ("graph", "Stammbaum & IK-Rechner", "Stammbaum bis 4 Generationen, Inzuchtkoeffizient nach Wright, Probeverpaarung"), ("sparkle", "KI-Züchter-Assistent", "Wurfankündigungen schreiben, Genetik-Erklärung, Paarungsanalyse, Jahresbericht"), ("globe", "Öffentliches Züchter-Profil", "Visitenkarte unter banyaro.app/breeder/{zwingername} mit Hunden, Fotos und Gesundheitsstatistik"), ("download-simple", "Datenexport", "Vollständiger Export als HTML-Dossier und ODS-Tabelle (LibreOffice/Excel)"), ] features = _features_breeder if req["tier"] == "breeder" else _features_pro def _feature_row(icon, title, desc): return f"""
✓
{title}
{desc}
""" feature_rows = "".join(_feature_row(i, t, d) for i, t, d in features) try: from mailer import send_email, email_html import html as _html name_esc = _html.escape(req["name"]) body_html = f"""

Herzlichen Glückwunsch, {name_esc}! 🎉

Dein Account wurde soeben auf {tier_label} freigeschaltet. Vielen Dank für dein Vertrauen in Ban Yaro!

Dein Tarif
{tier_label} {tier_price}
Deine neuen Features
{feature_rows}
So aktivierst du deine Features:
Öffne Ban Yaro und lade die App einmal neu (Startseite antippen → herunterziehen oder App schließen und neu öffnen). Alle Features sind dann sofort verfügbar.
""" html = email_html(body_html, cta_url="https://banyaro.app", cta_label="Ban Yaro jetzt öffnen") plain = (f"Herzlichen Glückwunsch, {req['name']}!\n\n" f"Dein Account wurde auf {tier_label} ({tier_price}) freigeschaltet.\n\n" f"Öffne Ban Yaro und lade die App neu — alle Features sind dann aktiv.\n\n" f"Viele Grüße\nRené & das Ban Yaro Team") await send_email(req["email"], f"🎉 Dein {tier_label}-Zugang ist aktiv!", html, plain) except Exception as e: import logging logging.getLogger(__name__).warning(f"Bestätigungsmail fehlgeschlagen: {e}") # Offene Rechnungen (sent/draft) des alten Tiers stornieren + neuen Entwurf anlegen inv_number = None try: inv_number = await _handle_upgrade_invoices(req, tier_label) except Exception as e: logger.warning(f"Upgrade-Rechnungslogik fehlgeschlagen für {req['name']}: {e}") return {"ok": True, "tier": req["tier"], "user": req["name"], "invoice_number": inv_number} def _get_discount_info(conn, user_id: int) -> dict: """Berechnet Rabatt für einen User basierend auf Gründer-Status und Referrals.""" row = conn.execute( """SELECT u.is_founder, u.is_founder_pending, u.referred_by, COALESCE((SELECT COUNT(*) FROM users WHERE referred_by=u.id), 0) AS referral_count FROM users u WHERE u.id=?""", (user_id,) ).fetchone() if not row: return {"discount_pct": 0, "reason": None, "referral_count": 0} if row["is_founder"] or row["is_founder_pending"]: return {"discount_pct": 100, "reason": "founder", "referral_count": row["referral_count"]} referred_by = row["referred_by"] or 0 if referred_by > 0: referrer = conn.execute( "SELECT is_founder, is_founder_pending FROM users WHERE id=?", (referred_by,) ).fetchone() if referrer and (referrer["is_founder"] or referrer["is_founder_pending"]): return {"discount_pct": 100, "reason": "referred_by_founder", "referral_count": row["referral_count"]} count = row["referral_count"] for threshold, pct in [(50, 50), (20, 30), (10, 20)]: if count >= threshold: return {"discount_pct": pct, "reason": "referral", "referral_count": count} return {"discount_pct": 0, "reason": None, "referral_count": count} async def _handle_upgrade_invoices(req: dict, new_tier_label: str): """Storniert offene Rechnungen des alten Tiers und legt neuen Entwurf an.""" from routes.invoices import _next_invoice_number from datetime import timedelta with db() as conn: # Offene Rechnungen (draft + sent) dieses Users finden open_invoices = conn.execute( "SELECT * FROM invoices WHERE user_id=? AND status IN ('draft','sent')", (req["user_id"],) ).fetchall() for inv in open_invoices: cancel_num = _next_invoice_number(conn, "ST") conn.execute( """UPDATE invoices SET status='cancelled', cancelled_at=strftime('%Y-%m-%dT%H:%M:%SZ','now'), cancellation_reason=?, cancellation_number=? WHERE id=?""", (f"Tarif-Upgrade auf {new_tier_label}", cancel_num, inv["id"]) ) logger.info(f"Rechnung {inv['invoice_number']} storniert ({cancel_num}) — Upgrade auf {new_tier_label}") # Neuen Entwurf für den neuen Tier anlegen tier = req["tier"] price = {"pro": 29.00, "breeder": 49.00}.get(tier, 29.00) today = datetime.now(_TZ).date() end_date = today.replace(year=today.year + 1) - timedelta(days=1) period = f"{today.strftime('%d.%m.%Y')} - {end_date.strftime('%d.%m.%Y')}" description = f"{new_tier_label} Jahresabo" billing = conn.execute( "SELECT billing_address FROM users WHERE id=?", (req["user_id"],) ).fetchone() billing_address = billing["billing_address"] if billing else None disc_info = _get_discount_info(conn, req["user_id"]) discount_pct = disc_info["discount_pct"] discount_amt = round(price * discount_pct / 100, 2) after_disc = round(price - discount_amt, 2) _AGB = "Jahresbeitrag gem. AGB. Bei vorzeitiger Kündigung keine anteilige Rückerstattung; Zugang bleibt bis Laufzeitende bestehen." if disc_info["reason"] == "founder": note = f"Gründer-Sonderkonditionen: {new_tier_label} kostenfrei als Dankeschön für deine Unterstützung als Gründer! {_AGB}" elif disc_info["reason"] == "referred_by_founder": note = f"Willkommen in der Gründer-Community! Als persönlich von einem Gründer eingeladenes Mitglied erhältst du dauerhaft {discount_pct}% Rabatt. {_AGB}" elif disc_info["reason"] == "referral": note = f"Herzlichen Dank für deine Unterstützung! Für {disc_info['referral_count']} geworbene Freunde erhältst du {discount_pct}% Rabatt. {_AGB}" else: note = f"{_AGB} (Upgrade von {req.get('old_tier','Standard')} auf {new_tier_label})" inv_number = _next_invoice_number(conn) conn.execute(""" INSERT INTO invoices (invoice_number, user_id, recipient_name, recipient_email, recipient_address, description, service_period, amount_net, discount_pct, discount_amount, amount_after_discount, tax_rate, tax_amount, amount_gross, notes) VALUES (?,?,?,?,?,?,?,?,?,?,?,0,0,?,?) """, ( inv_number, req["user_id"], req["name"], req["email"], billing_address, description, period, price, discount_pct, discount_amt, after_disc, after_disc, note, )) invoice_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0] conn.execute( "INSERT INTO invoice_items (invoice_id, description, quantity, unit_price, total) VALUES (?,?,1,?,?)", (invoice_id, description, price, price) ) logger.info(f"Neuer Rechnungsentwurf {inv_number} für {req['email']} nach Upgrade auf {new_tier_label}") return inv_number # ------------------------------------------------------------------ # Helpers: Quartalsdaten # ------------------------------------------------------------------ def _quarter_bounds(year: int, q: int): """Gibt (start_date, end_date) als ISO-Strings zurück (YYYY-MM-DD).""" if q not in (1, 2, 3, 4): raise HTTPException(400, "Quartal muss 1–4 sein.") month_start = (q - 1) * 3 + 1 month_end = month_start + 2 # Letzter Tag des Endmonats import calendar last_day = calendar.monthrange(year, month_end)[1] return ( f"{year:04d}-{month_start:02d}-01", f"{year:04d}-{month_end:02d}-{last_day:02d}", ) def _fetch_quarter_invoices(conn, year: int, q: int): """Liest alle bezahlten/gesendeten Rechnungen des Quartals.""" start, end = _quarter_bounds(year, q) rows = conn.execute(""" SELECT invoice_number, created_at, recipient_name, recipient_email, amount_net, tax_amount, amount_gross, status, paid_at, paid_amount FROM invoices WHERE status IN ('paid', 'sent') AND DATE(created_at) BETWEEN ? AND ? ORDER BY created_at ASC """, (start, end)).fetchall() return rows, start, end def _build_csv(rows) -> bytes: """Erstellt CSV-Bytes aus den Rechnungszeilen.""" buf = io.StringIO() writer = csv.writer(buf, delimiter=";", quoting=csv.QUOTE_MINIMAL) writer.writerow([ "Rechnungsnummer", "Datum", "Empfänger", "E-Mail", "Nettobetrag", "Steuer", "Bruttobetrag", "Status", "Bezahlt-am", "Gezahlter-Betrag", ]) for r in rows: # Datum auf YYYY-MM-DD kürzen datum = (r["created_at"] or "")[:10] paid_at = (r["paid_at"] or "")[:10] if r["paid_at"] else "" writer.writerow([ r["invoice_number"], datum, r["recipient_name"], r["recipient_email"], f"{r['amount_net']:.2f}".replace(".", ","), f"{r['tax_amount']:.2f}".replace(".", ","), f"{r['amount_gross']:.2f}".replace(".", ","), r["status"], paid_at, f"{r['paid_amount']:.2f}".replace(".", ",") if r["paid_amount"] is not None else "", ]) return buf.getvalue().encode("utf-8-sig") # BOM für Excel-Kompatibilität # ------------------------------------------------------------------ # GET /api/admin/invoices/quarterly/{year}/{q}/csv # ------------------------------------------------------------------ @router.get("/invoices/quarterly/{year}/{q}/csv") async def invoice_quarterly_csv(year: int, q: int, user=Depends(require_admin)): """CSV-Download aller Rechnungen eines Quartals (paid + sent).""" with db() as conn: rows, start, end = _fetch_quarter_invoices(conn, year, q) csv_bytes = _build_csv(rows) filename = f"rechnungen_{year}_Q{q}.csv" logger.info(f"CSV-Download Q{q}/{year}: {len(rows)} Rechnungen → {filename}") return Response( content=csv_bytes, media_type="text/csv; charset=utf-8", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, ) # ------------------------------------------------------------------ # POST /api/admin/invoices/send-quarterly-report # ------------------------------------------------------------------ @router.post("/invoices/send-quarterly-report") async def send_quarterly_report(data: QuarterlyReportBody, user=Depends(require_admin)): """Sendet Quartalsbericht als CSV-Anhang an Steuerberater + Zusammenfassung an René.""" if data.quarter not in (1, 2, 3, 4): raise HTTPException(400, "Quartal muss 1–4 sein.") with db() as conn: rows, start, end = _fetch_quarter_invoices(conn, data.year, data.quarter) csv_bytes = _build_csv(rows) filename = f"rechnungen_{data.year}_Q{data.quarter}.csv" # Zusammenfassungs-Zahlen total_net = sum(r["amount_net"] for r in rows) total_tax = sum(r["tax_amount"] for r in rows) total_gross = sum(r["amount_gross"] for r in rows) count_paid = sum(1 for r in rows if r["status"] == "paid") count_sent = sum(1 for r in rows if r["status"] == "sent") subject_stb = ( f"Ban Yaro - Rechnungen Q{data.quarter}/{data.year} " f"({start} bis {end})" ) body_stb = ( f"Hallo,\n\n" f"anbei die Rechnungsübersicht für Q{data.quarter}/{data.year} " f"({start} bis {end}).\n\n" f"Anzahl Rechnungen: {len(rows)}\n" f" davon bezahlt: {count_paid}\n" f" davon ausstehend: {count_sent}\n\n" f"Summe Netto: {total_net:>10.2f} EUR\n" f"Summe Steuer: {total_tax:>10.2f} EUR\n" f"Summe Brutto: {total_gross:>10.2f} EUR\n\n" f"Die vollständige Liste finden Sie als CSV-Anhang.\n\n" f"Viele Grüße\nRené Nitzsche / Ban Yaro" ) from mailer import send_email, SMTP_FROM # Steuerberater-Mail (mit CSV-Anhang wenn unterstützt) try: await send_email( data.email, subject_stb, f"
{body_stb}
", body_stb, attachments=[{ "filename": filename, "content": csv_bytes, "content_type": "text/csv", }], ) logger.info(f"Quartalsbericht Q{data.quarter}/{data.year} → {data.email} (mit Anhang)") except TypeError: # send_email unterstützt noch kein attachments-Argument → ohne Anhang senden await send_email( data.email, subject_stb, f"
{body_stb}
", body_stb, ) logger.warning(f"Quartalsbericht Q{data.quarter}/{data.year} → {data.email} (OHNE Anhang, attachments nicht unterstützt)") # Zusammenfassung an René (SMTP_FROM-Adresse) # Reine E-Mail-Adresse aus "Name " extrahieren from_addr = SMTP_FROM if "<" in from_addr: from_addr = from_addr[from_addr.index("<") + 1 : from_addr.index(">")].strip() subject_rene = f"[Ban Yaro Admin] Quartalsbericht Q{data.quarter}/{data.year} versendet" body_rene = ( f"Der Quartalsbericht Q{data.quarter}/{data.year} wurde an {data.email} gesendet.\n\n" f"Zeitraum: {start} bis {end}\n" f"Rechnungen gesamt: {len(rows)} (bezahlt: {count_paid}, ausstehend: {count_sent})\n\n" f"Netto: {total_net:>10.2f} EUR\n" f"Steuer: {total_tax:>10.2f} EUR\n" f"Brutto: {total_gross:>10.2f} EUR\n" ) try: await send_email( from_addr, subject_rene, f"
{body_rene}
", body_rene, ) except Exception as e: logger.warning(f"Zusammenfassungs-Mail an René fehlgeschlagen: {e}") return { "ok": True, "sent_to": data.email, "year": data.year, "quarter": data.quarter, "period": f"{start} - {end}", "count": len(rows), "count_paid": count_paid, "count_sent": count_sent, "total_net": round(total_net, 2), "total_tax": round(total_tax, 2), "total_gross": round(total_gross, 2), }