auth.py

import json
import os
import time
from urllib.parse import urlencode
import requests as req
from itsdangerous import URLSafeTimedSerializer, SignatureExpired, BadSignature
from errors import get_logger

log = get_logger("auth")

SECRET_KEY = os.environ.get("SECRET_KEY", "generai-fallback-secret-change-in-prod")
GOOGLE_CLIENT_ID = os.environ.get("GOOGLE_CLIENT_ID", "")
GOOGLE_CLIENT_SECRET = os.environ.get("GOOGLE_CLIENT_SECRET", "")
BASE_URL = "https://amogaddy-generai.hf.space"
REDIRECT_URI = f"{BASE_URL}/auth/callback"
COOKIE_NAME = "generai_session"
DEFAULT_CREDITS = 10_000
UNLIMITED_CREDITS = 999_999_999
USERS_FILE = "./database/users.json"

# Email del proprietario/admin → crediti illimitati. Configurabile via Secret ADMIN_EMAILS
# (lista separata da virgole). Default: account dell'owner del progetto.
ADMIN_EMAILS = {
    e.strip().lower()
    for e in os.environ.get("ADMIN_EMAILS", "btpfab1@gmail.com,amogaddyofficial@gmail.com").split(",")
    if e.strip()
}

def is_admin_email(email: str) -> bool:
    return (email or "").strip().lower() in ADMIN_EMAILS

_signer = URLSafeTimedSerializer(SECRET_KEY)


# ── UserDB ─────────────────────────────────────────────────────────────────────

class UserDB:
    def __init__(self):
        os.makedirs("./database", exist_ok=True)
        self._data: dict = {}
        self._load()

    def _load(self):
        if os.path.exists(USERS_FILE):
            try:
                with open(USERS_FILE) as f:
                    self._data = json.load(f)
            except Exception:
                self._data = {}

    def _save(self):
        try:
            with open(USERS_FILE, "w") as f:
                json.dump(self._data, f, indent=2)
        except Exception as e:
            log.warning("Salvataggio users.json fallito: %s", e)

    def get_or_create(self, gid: str, email: str, name: str, picture: str) -> dict:
        starting_credits = UNLIMITED_CREDITS if is_admin_email(email) else DEFAULT_CREDITS
        if gid not in self._data:
            self._data[gid] = {
                "id": gid,
                "email": email,
                "name": name,
                "picture": picture,
                "credits": starting_credits,
                "created_at": time.time(),
            }
            log.info("Nuovo utente: %s (%s) — %d crediti", name, email, starting_credits)
        else:
            self._data[gid].update({"name": name, "picture": picture, "email": email})
            if is_admin_email(email):
                self._data[gid]["credits"] = UNLIMITED_CREDITS
        self._save()
        return dict(self._data[gid])

    def get(self, gid: str) -> dict | None:
        u = self._data.get(gid)
        return dict(u) if u else None

    def use_credits(self, gid: str, n: int = 1) -> bool:
        u = self._data.get(gid)
        if not u:
            return False
        if is_admin_email(u.get("email", "")):
            return True
        if u.get("credits", 0) < n:
            return False
        u["credits"] -= n
        self._save()
        return True

    def remaining(self, gid: str) -> int:
        u = self._data.get(gid)
        if not u:
            return 0
        if is_admin_email(u.get("email", "")):
            return UNLIMITED_CREDITS
        return int(u.get("credits", 0))

    def add_credits(self, gid: str, n: int):
        u = self._data.get(gid)
        if u:
            u["credits"] = u.get("credits", 0) + n
            self._save()

    def count(self) -> int:
        return len(self._data)


_user_db: UserDB | None = None

def get_user_db() -> UserDB:
    global _user_db
    if _user_db is None:
        _user_db = UserDB()
    return _user_db


# ── Session ────────────────────────────────────────────────────────────────────

def create_session(gid: str) -> str:
    return _signer.dumps(gid, salt="session-v1")

def verify_session(token: str) -> str | None:
    try:
        return _signer.loads(token, salt="session-v1", max_age=86400 * 30)
    except (SignatureExpired, BadSignature):
        return None

def get_current_user(request) -> dict | None:
    token = request.cookies.get(COOKIE_NAME)
    if not token:
        return None
    gid = verify_session(token)
    if not gid:
        return None
    return get_user_db().get(gid)


# ── Google OAuth ───────────────────────────────────────────────────────────────

def google_login_url() -> str:
    if not GOOGLE_CLIENT_ID:
        return ""
    params = {
        "client_id": GOOGLE_CLIENT_ID,
        "redirect_uri": REDIRECT_URI,
        "response_type": "code",
        "scope": "openid email profile",
        "access_type": "online",
        "prompt": "select_account",
    }
    return "https://accounts.google.com/o/oauth2/v2/auth?" + urlencode(params)

def exchange_code(code: str) -> dict:
    try:
        r = req.post("https://oauth2.googleapis.com/token", data={
            "client_id": GOOGLE_CLIENT_ID,
            "client_secret": GOOGLE_CLIENT_SECRET,
            "code": code,
            "grant_type": "authorization_code",
            "redirect_uri": REDIRECT_URI,
        }, timeout=10)
        return r.json()
    except Exception as e:
        log.warning("Token exchange fallito: %s", e)
        return {}

def get_google_userinfo(access_token: str) -> dict:
    try:
        r = req.get("https://www.googleapis.com/oauth2/v3/userinfo",
                    headers={"Authorization": f"Bearer {access_token}"}, timeout=10)
        return r.json()
    except Exception as e:
        log.warning("Userinfo fallita: %s", e)
        return {}