BLOCCO 1 - Multi-user data model: - User: email, display_name, avatar_url, auth_provider, google_id - User: subscription_plan, subscription_expires_at, is_admin, post counters - SubscriptionCode table per redeem codes - user_id FK su Character, Post, AffiliateLink, EditorialPlan, SocialAccount, SystemSetting - Migrazione SQLite-safe (ALTER TABLE) + preserva dati esistenti BLOCCO 2 - Auth completo: - Registrazione email/password + login multi-user - Google OAuth 2.0 (httpx, no deps esterne) - Callback flow: Google -> /auth/callback?token=JWT -> frontend - Backward compat login admin con username BLOCCO 3 - Piani e abbonamenti: - Freemium: 1 character, 15 post/mese, FB+IG only, no auto-plans - Pro: illimitato, tutte le piattaforme, tutte le feature - Enforcement automatico in tutti i router - Redeem codes con durate 1/3/6/12 mesi - Admin panel: genera codici, lista utenti BLOCCO 4 - Frontend completo: - Login page design Leopost (split coral/cream, Google, social coming soon) - AuthCallback per OAuth redirect - PlanBanner, UpgradeModal con pricing - AdminSettings per generazione codici - CharacterForm con tab Account Social + guide setup Deploy: - Dockerfile con ARG VITE_BASE_PATH/VITE_API_BASE - docker-compose.prod.yml per leopost.it (no subpath) - docker-compose.yml aggiornato per lab Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
114 lines
3.5 KiB
Python
114 lines
3.5 KiB
Python
"""Admin router — user management and subscription code generation."""
|
|
|
|
import secrets
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..auth import get_current_user
|
|
from ..database import get_db
|
|
from ..models import SubscriptionCode, User
|
|
|
|
router = APIRouter(prefix="/api/admin", tags=["admin"])
|
|
|
|
|
|
def _require_admin(current_user: User = Depends(get_current_user)) -> User:
|
|
if not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Accesso riservato agli amministratori.")
|
|
return current_user
|
|
|
|
|
|
class GenerateCodeRequest(BaseModel):
|
|
duration_months: int # 1, 3, 6, 12
|
|
|
|
|
|
@router.get("/users")
|
|
def list_users(
|
|
db: Session = Depends(get_db),
|
|
admin: User = Depends(_require_admin),
|
|
):
|
|
"""List all users (admin only)."""
|
|
users = db.query(User).order_by(User.created_at.desc()).all()
|
|
return [
|
|
{
|
|
"id": u.id,
|
|
"email": u.email,
|
|
"username": u.username,
|
|
"display_name": u.display_name,
|
|
"subscription_plan": u.subscription_plan or "freemium",
|
|
"subscription_expires_at": u.subscription_expires_at.isoformat() if u.subscription_expires_at else None,
|
|
"is_admin": bool(u.is_admin),
|
|
"auth_provider": u.auth_provider or "local",
|
|
"created_at": u.created_at.isoformat() if u.created_at else None,
|
|
}
|
|
for u in users
|
|
]
|
|
|
|
|
|
@router.post("/codes/generate")
|
|
def generate_code(
|
|
request: GenerateCodeRequest,
|
|
db: Session = Depends(get_db),
|
|
admin: User = Depends(_require_admin),
|
|
):
|
|
"""Generate a new Pro subscription code (admin only)."""
|
|
if request.duration_months not in (1, 3, 6, 12):
|
|
raise HTTPException(status_code=400, detail="duration_months deve essere 1, 3, 6 o 12.")
|
|
|
|
raw = secrets.token_urlsafe(12).upper()[:12]
|
|
code_str = f"LP-{raw}"
|
|
|
|
# Ensure uniqueness
|
|
attempts = 0
|
|
while db.query(SubscriptionCode).filter(SubscriptionCode.code == code_str).first():
|
|
raw = secrets.token_urlsafe(12).upper()[:12]
|
|
code_str = f"LP-{raw}"
|
|
attempts += 1
|
|
if attempts > 10:
|
|
raise HTTPException(status_code=500, detail="Impossibile generare codice univoco.")
|
|
|
|
code = SubscriptionCode(
|
|
code=code_str,
|
|
duration_months=request.duration_months,
|
|
created_by_admin_id=admin.id,
|
|
)
|
|
db.add(code)
|
|
db.commit()
|
|
db.refresh(code)
|
|
|
|
return {
|
|
"code": code.code,
|
|
"duration_months": code.duration_months,
|
|
"created_at": code.created_at.isoformat(),
|
|
}
|
|
|
|
|
|
@router.get("/codes")
|
|
def list_codes(
|
|
db: Session = Depends(get_db),
|
|
admin: User = Depends(_require_admin),
|
|
):
|
|
"""List all subscription codes (admin only)."""
|
|
codes = db.query(SubscriptionCode).order_by(SubscriptionCode.created_at.desc()).all()
|
|
result = []
|
|
for c in codes:
|
|
used_by_email = None
|
|
if c.used_by_user_id:
|
|
used_user = db.query(User).filter(User.id == c.used_by_user_id).first()
|
|
if used_user:
|
|
used_by_email = used_user.email or used_user.username
|
|
|
|
result.append({
|
|
"id": c.id,
|
|
"code": c.code,
|
|
"duration_months": c.duration_months,
|
|
"status": "used" if c.used_by_user_id else "active",
|
|
"used_by": used_by_email,
|
|
"used_at": c.used_at.isoformat() if c.used_at else None,
|
|
"created_at": c.created_at.isoformat() if c.created_at else None,
|
|
})
|
|
return result
|