Feature: Partner-Codes + Gründer-Lizenz-System für Influencer-Kooperationen

- partner_codes Tabelle: Code, Label, max_uses, grants_founder, uses-Counter
- users: is_founder + is_partner Flags (DB-Migration + auth.py SELECT)
- Registrierung: Partner-Code löst Gründer-Lizenz aus (vor User-Referral geprüft)
- API: GET/POST/DELETE /api/admin/partner/codes, POST /api/admin/partner/users/{id}/grant
- API: GET /api/partner/codes/{code}/info (öffentlich, für Registrierungsvalidierung)
- API: GET /api/admin/users/search (Name-Suche für Admin-UI)
- Admin-Tab "Partner & Codes": Code anlegen, Stats, User-Status vergeben
- Registrierungsformular: Einladungscode-Feld mit Live-Validierung
- Settings: Gründer (lila) + Partner (blau) Badge neben Kostenlos/Plus
- SW by-v515, APP_VER 492
This commit is contained in:
rene 2026-04-29 21:20:16 +02:00
parent 810c1a79dc
commit e57c6db013
9 changed files with 469 additions and 20 deletions

136
backend/routes/partner.py Normal file
View file

@ -0,0 +1,136 @@
"""BAN YARO — Partner-Codes + Gründer-Lizenz"""
from typing import Optional
from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel
from database import db
from auth import require_admin, get_current_user
router = APIRouter()
class PartnerCodeCreate(BaseModel):
code: str
label: str
grants_founder: int = 1
max_uses: Optional[int] = None
class GrantRequest(BaseModel):
is_founder: Optional[int] = None
is_partner: Optional[int] = None
# ------------------------------------------------------------------
# Admin: Partner-Codes verwalten
# ------------------------------------------------------------------
@router.get("/admin/partner/codes")
def list_partner_codes(user=Depends(require_admin)):
"""Alle Partner-Codes mit Stats (admin only)."""
with db() as conn:
rows = conn.execute(
"""SELECT pc.id, pc.code, pc.label, pc.grants_founder,
pc.max_uses, pc.uses, pc.created_at,
u.name AS created_by_name
FROM partner_codes pc
LEFT JOIN users u ON u.id = pc.created_by
ORDER BY pc.created_at DESC"""
).fetchall()
return [dict(r) for r in rows]
@router.post("/admin/partner/codes", status_code=201)
def create_partner_code(data: PartnerCodeCreate, user=Depends(require_admin)):
"""Neuen Partner-Code erstellen (admin only)."""
code = data.code.strip().upper()
if not code:
raise HTTPException(400, "Code darf nicht leer sein.")
with db() as conn:
existing = conn.execute(
"SELECT id FROM partner_codes WHERE code=?", (code,)
).fetchone()
if existing:
raise HTTPException(400, "Dieser Code existiert bereits.")
conn.execute(
"""INSERT INTO partner_codes (code, label, grants_founder, max_uses, created_by)
VALUES (?, ?, ?, ?, ?)""",
(code, data.label.strip(), data.grants_founder, data.max_uses, user["id"])
)
row = conn.execute(
"SELECT * FROM partner_codes WHERE code=?", (code,)
).fetchone()
return dict(row)
@router.delete("/admin/partner/codes/{code_id}", status_code=204)
def delete_partner_code(code_id: int, user=Depends(require_admin)):
"""Partner-Code löschen (admin only)."""
with db() as conn:
existing = conn.execute(
"SELECT id FROM partner_codes WHERE id=?", (code_id,)
).fetchone()
if not existing:
raise HTTPException(404, "Partner-Code nicht gefunden.")
conn.execute("DELETE FROM partner_codes WHERE id=?", (code_id,))
return None
@router.post("/admin/partner/users/{user_id}/grant")
def grant_user_status(user_id: int, data: GrantRequest, user=Depends(require_admin)):
"""Founder- und/oder Partner-Status für einen User setzen (admin only)."""
updates = {}
if data.is_founder is not None:
updates["is_founder"] = data.is_founder
if data.is_partner is not None:
updates["is_partner"] = data.is_partner
if not updates:
raise HTTPException(400, "Mindestens is_founder oder is_partner muss angegeben werden.")
with db() as conn:
target = conn.execute("SELECT id FROM users WHERE id=?", (user_id,)).fetchone()
if not target:
raise HTTPException(404, "User nicht gefunden.")
set_clause = ", ".join(f"{k}=?" for k in updates)
conn.execute(
f"UPDATE users SET {set_clause} WHERE id=?",
(*updates.values(), user_id)
)
row = conn.execute(
"SELECT id, name, email, is_founder, is_partner FROM users WHERE id=?",
(user_id,)
).fetchone()
return dict(row)
@router.get("/admin/users/search")
def search_users(q: str, user=Depends(require_admin)):
"""User-Suche für Admin (Name-Präfix, max. 10 Ergebnisse)."""
with db() as conn:
rows = conn.execute(
"""SELECT id, name, email, is_founder, is_partner, rolle
FROM users WHERE name LIKE ? COLLATE NOCASE
ORDER BY name LIMIT 10""",
(f"{q}%",)
).fetchall()
return [dict(r) for r in rows]
# ------------------------------------------------------------------
# Öffentlich: Code-Info für Registrierungsseite
# ------------------------------------------------------------------
@router.get("/partner/codes/{code}/info")
def partner_code_info(code: str):
"""Gibt zurück ob ein Partner-Code existiert und dessen Label (öffentlich)."""
with db() as conn:
row = conn.execute(
"""SELECT code, label, grants_founder, max_uses, uses
FROM partner_codes WHERE code=?""",
(code.strip().upper(),)
).fetchone()
if not row:
raise HTTPException(404, "Partner-Code nicht gefunden.")
r = dict(row)
# Einlösbar?
r["redeemable"] = r["max_uses"] is None or r["uses"] < r["max_uses"]
return r