Spaces:

JuloEco
/

Opsiom

Running

Opsiom / models.py

Upload 60 files

51bd271 verified 2 days ago

22.1 kB

"""
Modèles de données — comptes Opsiom, forfaits et progression.
Cette base locale à Opsiom ne stocke JAMAIS de mot de passe (voir auth.py,
authentification déléguée à Octix). Elle sert à trois choses :
  1. Le quota de tokens/jour, désormais dépendant du FORFAIT du compte
     (voir plans.py) et appliqué ici — Octix reste interrogé en best-effort
     pour rester synchronisé avec le CLI, mais l'application de la limite
     par forfait vit ici, car Octix ne connaît pas la notion de forfait.
  2. Le forfait courant (free / plus / pro) et le statut Lab.
  3. Le cache de progression (missions LearnCode/Classroom/Omnia Mind),
     recalculé toutes les PROGRESSION_CACHE_TTL (voir progression.py) pour
     éviter d'interroger trois bases externes à chaque page vue.
  4. Les codes de promo (tokens en plus + accès temporaire à des modèles
     supplémentaires, créés depuis l'espace admin — voir promo.py) et leur
     utilisation par compte.
Administrateur : un pseudo listé dans plans.ADMIN_USERNAMES (par défaut
"Jules") n'a aucune limite de tokens ni de modèles.
Quotas par personne et bêta-testeurs (voir QuotaOverride, BetaGrant et
/admin/acces) : l'admin peut attribuer un quota à un pseudo ; les bêta-testeurs
(tirés au sort chaque mois, ou permanents via la colonne `lab`) n'ont aucune
limite et accèdent aux modèles en avant-première.
"""
import json
from datetime import date, datetime

from flask_sqlalchemy import SQLAlchemy

from plans import (
    ADMIN_USERNAMES, DEFAULT_PLAN, DEFAULT_PREVIEW_MODELS, MODEL_DISPLAY, UNLIMITED_REMAINING, plan_config,
)

from dbprefix import t as T

db = SQLAlchemy()


class User(db.Model):
    """Ombre locale d'un compte Octix : pseudo + forfait + quota + progression.
    Aucun mot de passe n'est stocké ici : l'authentification est déléguée à
    Octix (voir auth.py). Cette table associe un pseudo Octix à son forfait
    Opsiom et à son compteur de tokens/jour.
    """
    __tablename__ = T("users")

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(32), unique=True, nullable=False, index=True)
    email = db.Column(db.String(255), nullable=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)

    # --- Forfait -----------------------------------------------------
    plan = db.Column(db.String(16), default=DEFAULT_PLAN, nullable=False)
    plan_updated_at = db.Column(db.DateTime, nullable=True)
    # Statut Lab : séparé des forfaits classiques (voir plans.py). Pour
    # l'instant attribué uniquement à la main (pas de critère automatique).
    lab = db.Column(db.Boolean, default=False, nullable=False)

    # --- Quota de tokens, dépendant du forfait ------------------------
    tokens_used_today = db.Column(db.Integer, default=0, nullable=False)
    quota_date = db.Column(db.Date, default=date.today, nullable=False)

    # --- Suivi "jours actifs" / "conversations" pour les missions Opsiom
    # (voir plans.py: opsiom.active_days / opsiom.conversations).
    # active_days_json = liste JSON des dates ISO (YYYY-MM-DD) où le compte
    # a envoyé au moins un message. Suffisant pour compter les jours
    # distincts sans repasser par une table séparée.
    active_days_json = db.Column(db.Text, default="[]", nullable=False)
    conversations_total = db.Column(db.Integer, default=0, nullable=False)

    # --- Cache de progression (missions LearnCode/Classroom/Omnia Mind) --
    progression_json = db.Column(db.Text, nullable=True)
    progression_updated_at = db.Column(db.DateTime, nullable=True)

    # ------------------------------------------------------------------
    # Quota
    # ------------------------------------------------------------------
    def _reset_quota_if_needed(self):
        """Remet le compteur à zéro si on a changé de jour depuis le dernier appel."""
        if self.quota_date != date.today():
            self.quota_date = date.today()
            self.tokens_used_today = 0

    # ------------------------------------------------------------------
    # Administrateur & codes de promo
    # ------------------------------------------------------------------
    @property
    def is_admin(self) -> bool:
        return self.username in ADMIN_USERNAMES

    def active_grants(self) -> list["PromoRedemption"]:
        """Codes de promo utilisés par ce compte et encore en cours de validité
        (les avantages sont TEMPORAIRES : ils disparaissent à `expires_at`)."""
        if self.id is None:
            return []
        return (
            PromoRedemption.query
            .filter(PromoRedemption.user_id == self.id, PromoRedemption.expires_at > datetime.utcnow())
            .order_by(PromoRedemption.expires_at)
            .all()
        )

    @property
    def bonus_tokens(self) -> int:
        """Tokens/jour supplémentaires apportés par les codes actifs."""
        return sum(g.promo.bonus_tokens for g in self.active_grants())

    @property
    def bonus_models(self) -> list[str]:
        extra: list[str] = []
        for g in self.active_grants():
            for model_id in g.promo.models:
                if model_id not in extra:
                    extra.append(model_id)
        return extra

    # ------------------------------------------------------------------
    # Quota attribué à la personne (admin) & statut bêta-testeur
    # ------------------------------------------------------------------
    @property
    def override(self) -> "QuotaOverride | None":
        """Quota attribué à la main à CE compte (voir /admin/acces), s'il est actif."""
        ov = QuotaOverride.query.filter_by(username=self.username).first()
        return ov if ov is not None and ov.is_active() else None

    @property
    def is_beta(self) -> bool:
        """Bêta-testeur : statut permanent (colonne `lab`, attribué à la main) OU
        tiré au sort / ajouté pour le mois en cours (table BetaGrant)."""
        if self.lab:
            return True
        return BetaGrant.query.filter_by(username=self.username, month=current_month()).first() is not None

    @property
    def unlimited_reason(self) -> str | None:
        """Pourquoi ce compte n'a aucune limite (None = il en a une)."""
        if self.is_admin:
            return "admin"
        if self.is_beta:
            return "beta"
        ov = self.override
        if ov is not None and ov.unlimited:
            return "custom"
        return None

    @property
    def is_unlimited(self) -> bool:
        return self.unlimited_reason is not None

    @property
    def base_daily_tokens(self) -> int:
        """Quota de base : celui attribué à la personne s'il existe, sinon celui du forfait."""
        ov = self.override
        if ov is not None and ov.daily_tokens is not None:
            return ov.daily_tokens
        return plan_config(self.plan)["daily_tokens"]

    @property
    def daily_limit(self) -> int:
        return self.base_daily_tokens + self.bonus_tokens

    @property
    def allowed_models(self) -> list[str]:
        # Admin et bêta-testeurs : tous les modèles, y compris ceux en avant-première.
        if self.is_admin or self.is_beta:
            return list(MODEL_DISPLAY)
        preview = set(get_preview_models())
        allowed = [m for m in plan_config(self.plan)["models"] if m not in preview]
        ov = self.override
        if ov is not None:
            for model_id in ov.models:
                if model_id not in allowed and model_id not in preview:
                    allowed.append(model_id)
        # Un code de promo, créé par l'admin, peut en revanche ouvrir un modèle en avant-première.
        for model_id in self.bonus_models:
            if model_id not in allowed:
                allowed.append(model_id)
        return allowed

    def quota_status(self) -> dict:
        self._reset_quota_if_needed()
        reason = self.unlimited_reason
        if reason:
            return {
                "used": self.tokens_used_today,
                "limit": 0,
                "remaining": UNLIMITED_REMAINING,
                "plan": self.plan,
                "unlimited": True,
                "unlimited_reason": reason,
                "bonus": 0,
            }
        bonus = self.bonus_tokens
        limit = self.base_daily_tokens + bonus
        return {
            "used": self.tokens_used_today,
            "limit": limit,
            "remaining": max(0, limit - self.tokens_used_today),
            "plan": self.plan,
            "unlimited": False,
            "bonus": bonus,
            "custom": self.override is not None and self.override.daily_tokens is not None,
        }

    def can_send_message(self) -> bool:
        if self.is_unlimited:
            return True
        self._reset_quota_if_needed()
        return self.tokens_used_today < self.daily_limit

    def register_message_sent(self, tokens: int = 1):
        """À appeler une fois la réponse obtenue avec succès (pas avant),
        pour ne jamais décompter un message qui a échoué côté serveur.
        Met aussi à jour les compteurs utilisés par les missions Opsiom
        (jours actifs, nombre de conversations)."""
        self._reset_quota_if_needed()
        self.tokens_used_today += max(0, int(tokens))

        today_iso = date.today().isoformat()
        try:
            days = set(json.loads(self.active_days_json or "[]"))
        except (ValueError, TypeError):
            days = set()
        days.add(today_iso)
        self.active_days_json = json.dumps(sorted(days)[-90:])  # 90 derniers jours suffisent
        self.conversations_total += 1

        db.session.commit()

    # ------------------------------------------------------------------
    # Stats Opsiom utilisées par progression.py pour les missions
    # ------------------------------------------------------------------
    def opsiom_mission_stats(self) -> dict:
        try:
            days = json.loads(self.active_days_json or "[]")
        except (ValueError, TypeError):
            days = []
        return {
            "active_days": len(days),
            "conversations": self.conversations_total,
        }

    # ------------------------------------------------------------------
    # Progression (cache)
    # ------------------------------------------------------------------
    def cached_progression(self):
        """Renvoie (raw_stats, is_fresh) à partir du cache local, ou
        (None, False) si rien n'est encore en cache."""
        if not self.progression_json or not self.progression_updated_at:
            return None, False
        from progression import PROGRESSION_CACHE_TTL
        fresh = (datetime.utcnow() - self.progression_updated_at) < PROGRESSION_CACHE_TTL
        try:
            return json.loads(self.progression_json), fresh
        except (ValueError, TypeError):
            return None, False

    def store_progression(self, raw_stats: dict):
        self.progression_json = json.dumps(raw_stats)
        self.progression_updated_at = datetime.utcnow()
        db.session.commit()

    def __repr__(self):
        return f"<User {self.username} plan={self.plan}>"


class PromoCode(db.Model):
    """Code de promo créé par un administrateur (voir promo.py).
    Un code donne, à qui l'utilise, des tokens/jour EN PLUS et/ou l'accès à des
    modèles supplémentaires, pendant `duration_hours` (24 h par défaut) à
    partir de l'utilisation. Rien n'est définitif : à l'expiration, le compte
    retrouve exactement les limites de son forfait.
    """
    __tablename__ = T("promo_codes")

    id = db.Column(db.Integer, primary_key=True)
    code = db.Column(db.String(40), unique=True, nullable=False, index=True)
    note = db.Column(db.String(120), nullable=True)          # mémo libre, visible seulement de l'admin
    bonus_tokens = db.Column(db.Integer, default=0, nullable=False)
    models_json = db.Column(db.Text, default="[]", nullable=False)   # ex: ["large"]
    duration_hours = db.Column(db.Integer, default=24, nullable=False)
    max_uses = db.Column(db.Integer, nullable=True)           # None = illimité
    valid_until = db.Column(db.DateTime, nullable=True)       # date limite pour UTILISER le code
    active = db.Column(db.Boolean, default=True, nullable=False)
    created_by = db.Column(db.String(32), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)

    redemptions = db.relationship(
        "PromoRedemption", backref="promo", cascade="all, delete-orphan", lazy="dynamic",
    )

    @property
    def models(self) -> list[str]:
        try:
            value = json.loads(self.models_json or "[]")
        except (ValueError, TypeError):
            return []
        return [m for m in value if m in MODEL_DISPLAY] if isinstance(value, list) else []

    @property
    def uses_count(self) -> int:
        return self.redemptions.count()

    def redeemable_error(self) -> str | None:
        """None si le code peut encore être utilisé, sinon la raison (en français)."""
        if not self.active:
            return "Ce code n'est plus actif."
        if self.valid_until and datetime.utcnow() > self.valid_until:
            return "Ce code a expiré."
        if self.max_uses is not None and self.uses_count >= self.max_uses:
            return "Ce code a atteint son nombre maximal d'utilisations."
        return None


class PromoRedemption(db.Model):
    """Utilisation d'un code de promo par un compte (une fois par compte et par code)."""
    __tablename__ = T("promo_redemptions")
    __table_args__ = (db.UniqueConstraint("user_id", "promo_id", name="uq_promo_user"),)

    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey(T("users") + ".id"), nullable=False, index=True)
    promo_id = db.Column(db.Integer, db.ForeignKey(T("promo_codes") + ".id"), nullable=False, index=True)
    redeemed_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
    expires_at = db.Column(db.DateTime, nullable=False)

    user = db.relationship("User")


# ---------------------------------------------------------------------------
# Quotas par personne, bêta-testeurs du mois, réglages admin.
# Ces tables sont indexées par PSEUDO Octix (et non par User.id) : l'admin peut
# ainsi attribuer un quota — ou le tirage peut désigner quelqu'un — qui ne s'est
# pas encore connecté à Opsiom. Nouvelles tables : db.create_all() les crée sans
# toucher aux bases existantes.
# ---------------------------------------------------------------------------
def current_month() -> str:
    """Mois en cours, 'AAAA-MM' (heure serveur, comme le reste des quotas)."""
    return date.today().strftime("%Y-%m")


class QuotaOverride(db.Model):
    """Quota de tokens/jour attribué à une personne précise par l'admin.
    - daily_tokens : remplace le quota du forfait (None = on garde celui du forfait) ;
    - unlimited    : aucune limite de tokens ;
    - models_json  : modèles supplémentaires accessibles à cette personne ;
    - expires_at   : date de fin (None = permanent) ; ensuite le forfait reprend.
    """
    __tablename__ = T("quota_overrides")

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(32), unique=True, nullable=False, index=True)
    daily_tokens = db.Column(db.Integer, nullable=True)
    unlimited = db.Column(db.Boolean, default=False, nullable=False)
    models_json = db.Column(db.Text, default="[]", nullable=False)
    note = db.Column(db.String(120), nullable=True)
    expires_at = db.Column(db.DateTime, nullable=True)
    set_by = db.Column(db.String(32), nullable=False)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)

    @property
    def models(self) -> list[str]:
        try:
            value = json.loads(self.models_json or "[]")
        except (ValueError, TypeError):
            return []
        return [m for m in value if m in MODEL_DISPLAY] if isinstance(value, list) else []

    def is_active(self) -> bool:
        return self.expires_at is None or self.expires_at > datetime.utcnow()


class BetaGrant(db.Model):
    """Un bêta-testeur pour un mois donné : modèles sans aucune limite et accès
    en avant-première. `source` = 'random' (tirage mensuel) ou 'manual' (admin)."""
    __tablename__ = T("beta_grants")
    __table_args__ = (db.UniqueConstraint("username", "month", name="uq_beta_user_month"),)

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(32), nullable=False, index=True)
    month = db.Column(db.String(7), nullable=False, index=True)       # 'AAAA-MM'
    source = db.Column(db.String(12), default="random", nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)


class AppSetting(db.Model):
    """Réglages modifiables depuis l'espace admin (clé -> valeur texte/JSON)."""
    __tablename__ = T("app_settings")

    key = db.Column(db.String(40), primary_key=True)
    value = db.Column(db.Text, nullable=False, default="")


def get_setting(key: str, default=None):
    row = db.session.get(AppSetting, key)
    return row.value if row is not None else default


def set_setting(key: str, value: str) -> None:
    row = db.session.get(AppSetting, key)
    if row is None:
        db.session.add(AppSetting(key=key, value=value))
    else:
        row.value = value
    db.session.commit()


def get_preview_models() -> list[str]:
    """Modèles en avant-première (réservés aux bêta-testeurs), choisis par l'admin."""
    raw = get_setting("preview_models")
    if raw is None:
        return list(DEFAULT_PREVIEW_MODELS)
    try:
        value = json.loads(raw)
    except (ValueError, TypeError):
        return list(DEFAULT_PREVIEW_MODELS)
    return [m for m in value if m in MODEL_DISPLAY] if isinstance(value, list) else list(DEFAULT_PREVIEW_MODELS)


class UsageEvent(db.Model):
    """Une tentative d'envoi de message (réussie, en erreur ou refusée par le quota),
    pour les statistiques de /admin/stats. Conservée OPSIOM_STATS_RETENTION_DAYS jours
    (30 par défaut) et bornée en nombre de lignes : voir stats.py."""
    __tablename__ = T("usage_events")

    id = db.Column(db.Integer, primary_key=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False, index=True)
    username = db.Column(db.String(32), nullable=False, index=True)
    model_id = db.Column(db.String(32), nullable=True)
    status = db.Column(db.String(8), nullable=False, default="ok")     # ok | error | quota
    tokens = db.Column(db.Integer, default=0, nullable=False)
    duration_ms = db.Column(db.Integer, nullable=True)                 # requête -> fin de la réponse
    ttft_ms = db.Column(db.Integer, nullable=True)                     # requête -> premier token (flux)
    used_after = db.Column(db.Integer, default=0, nullable=False)      # tokens du jour après ce message
    quota_limit = db.Column(db.Integer, default=0, nullable=False)     # limite du jour (0 si illimité)
    unlimited = db.Column(db.Boolean, default=False, nullable=False)
    beta = db.Column(db.Boolean, default=False, nullable=False)


class BetaNotification(db.Model):
    """E-mail « tu es bêta-testeur » envoyé (ou non) à une personne pour un mois."""
    __tablename__ = T("beta_notifications")
    __table_args__ = (db.UniqueConstraint("username", "month", name="uq_betanotif_user_month"),)

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(32), nullable=False, index=True)
    month = db.Column(db.String(7), nullable=False, index=True)        # 'AAAA-MM' ou 'permanent'
    status = db.Column(db.String(12), nullable=False)                  # sent | failed | no_email
    detail = db.Column(db.String(200), nullable=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)


# Limites de l'historique (par compte) : évitent qu'une base ou un instantané
# de sauvegarde grossisse sans fin.
MAX_CONVERSATIONS_PER_USER = 50
MAX_MESSAGES_PER_CONVERSATION = 200


class Conversation(db.Model):
    """Une conversation sauvegardée dans l'historique d'un compte.
    Chaque compte ne voit que les siennes (toutes les routes filtrent sur
    user_id) et peut les supprimer à tout moment (voir app.py, /api/conversations).
    """
    __tablename__ = T("conversations")

    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey(T("users") + ".id"), nullable=False, index=True)
    title = db.Column(db.String(80), default="Nouvelle conversation", nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False, index=True)

    messages = db.relationship(
        "ChatMessage", backref="conversation", cascade="all, delete-orphan",
        order_by="ChatMessage.id", lazy="dynamic",
    )

    def summary(self) -> dict:
        return {
            "id": self.id,
            "title": self.title,
            "updated_at": self.updated_at.isoformat() + "Z",
        }


class ChatMessage(db.Model):
    """Un message (utilisateur ou assistant) d'une conversation sauvegardée."""
    __tablename__ = T("chat_messages")

    id = db.Column(db.Integer, primary_key=True)
    conversation_id = db.Column(db.Integer, db.ForeignKey(T("conversations") + ".id"), nullable=False, index=True)
    role = db.Column(db.String(12), nullable=False)          # "user" | "assistant"
    content = db.Column(db.Text, nullable=False)
    model_id = db.Column(db.String(32), nullable=True)       # modèle qui a répondu (assistant)
    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)


def get_or_create_user(username: str) -> User:
    user = User.query.filter_by(username=username).first()
    if user is None:
        user = User(username=username)
        db.session.add(user)
        db.session.commit()
    return user

	"""
	Modèles de données — comptes Opsiom, forfaits et progression.

	Cette base locale à Opsiom ne stocke JAMAIS de mot de passe (voir auth.py,
	authentification déléguée à Octix). Elle sert à trois choses :
	1. Le quota de tokens/jour, désormais dépendant du FORFAIT du compte
	(voir plans.py) et appliqué ici — Octix reste interrogé en best-effort
	pour rester synchronisé avec le CLI, mais l'application de la limite
	par forfait vit ici, car Octix ne connaît pas la notion de forfait.
	2. Le forfait courant (free / plus / pro) et le statut Lab.
	3. Le cache de progression (missions LearnCode/Classroom/Omnia Mind),
	recalculé toutes les PROGRESSION_CACHE_TTL (voir progression.py) pour
	éviter d'interroger trois bases externes à chaque page vue.
	4. Les codes de promo (tokens en plus + accès temporaire à des modèles
	supplémentaires, créés depuis l'espace admin — voir promo.py) et leur
	utilisation par compte.

	Administrateur : un pseudo listé dans plans.ADMIN_USERNAMES (par défaut
	"Jules") n'a aucune limite de tokens ni de modèles.

	Quotas par personne et bêta-testeurs (voir QuotaOverride, BetaGrant et
	/admin/acces) : l'admin peut attribuer un quota à un pseudo ; les bêta-testeurs
	(tirés au sort chaque mois, ou permanents via la colonne `lab`) n'ont aucune
	limite et accèdent aux modèles en avant-première.
	"""
	import json
	from datetime import date, datetime

	from flask_sqlalchemy import SQLAlchemy

	from plans import (
	ADMIN_USERNAMES, DEFAULT_PLAN, DEFAULT_PREVIEW_MODELS, MODEL_DISPLAY, UNLIMITED_REMAINING, plan_config,
	)

	from dbprefix import t as T

	db = SQLAlchemy()


	class User(db.Model):
	"""Ombre locale d'un compte Octix : pseudo + forfait + quota + progression.

	Aucun mot de passe n'est stocké ici : l'authentification est déléguée à
	Octix (voir auth.py). Cette table associe un pseudo Octix à son forfait
	Opsiom et à son compteur de tokens/jour.
	"""
	__tablename__ = T("users")

	id = db.Column(db.Integer, primary_key=True)
	username = db.Column(db.String(32), unique=True, nullable=False, index=True)
	email = db.Column(db.String(255), nullable=True)
	created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)

	# --- Forfait -----------------------------------------------------
	plan = db.Column(db.String(16), default=DEFAULT_PLAN, nullable=False)
	plan_updated_at = db.Column(db.DateTime, nullable=True)
	# Statut Lab : séparé des forfaits classiques (voir plans.py). Pour
	# l'instant attribué uniquement à la main (pas de critère automatique).
	lab = db.Column(db.Boolean, default=False, nullable=False)

	# --- Quota de tokens, dépendant du forfait ------------------------
	tokens_used_today = db.Column(db.Integer, default=0, nullable=False)
	quota_date = db.Column(db.Date, default=date.today, nullable=False)

	# --- Suivi "jours actifs" / "conversations" pour les missions Opsiom
	# (voir plans.py: opsiom.active_days / opsiom.conversations).
	# active_days_json = liste JSON des dates ISO (YYYY-MM-DD) où le compte
	# a envoyé au moins un message. Suffisant pour compter les jours
	# distincts sans repasser par une table séparée.
	active_days_json = db.Column(db.Text, default="[]", nullable=False)
	conversations_total = db.Column(db.Integer, default=0, nullable=False)

	# --- Cache de progression (missions LearnCode/Classroom/Omnia Mind) --
	progression_json = db.Column(db.Text, nullable=True)
	progression_updated_at = db.Column(db.DateTime, nullable=True)

	# ------------------------------------------------------------------
	# Quota
	# ------------------------------------------------------------------
	def _reset_quota_if_needed(self):
	"""Remet le compteur à zéro si on a changé de jour depuis le dernier appel."""
	if self.quota_date != date.today():
	self.quota_date = date.today()
	self.tokens_used_today = 0

	# ------------------------------------------------------------------
	# Administrateur & codes de promo
	# ------------------------------------------------------------------
	@property
	def is_admin(self) -> bool:
	return self.username in ADMIN_USERNAMES

	def active_grants(self) -> list["PromoRedemption"]:
	"""Codes de promo utilisés par ce compte et encore en cours de validité
	(les avantages sont TEMPORAIRES : ils disparaissent à `expires_at`)."""
	if self.id is None:
	return []
	return (
	PromoRedemption.query
	.filter(PromoRedemption.user_id == self.id, PromoRedemption.expires_at > datetime.utcnow())
	.order_by(PromoRedemption.expires_at)
	.all()
	)

	@property
	def bonus_tokens(self) -> int:
	"""Tokens/jour supplémentaires apportés par les codes actifs."""
	return sum(g.promo.bonus_tokens for g in self.active_grants())

	@property
	def bonus_models(self) -> list[str]:
	extra: list[str] = []
	for g in self.active_grants():
	for model_id in g.promo.models:
	if model_id not in extra:
	extra.append(model_id)
	return extra

	# ------------------------------------------------------------------
	# Quota attribué à la personne (admin) & statut bêta-testeur
	# ------------------------------------------------------------------
	@property
	def override(self) -> "QuotaOverride \| None":
	"""Quota attribué à la main à CE compte (voir /admin/acces), s'il est actif."""
	ov = QuotaOverride.query.filter_by(username=self.username).first()
	return ov if ov is not None and ov.is_active() else None

	@property
	def is_beta(self) -> bool:
	"""Bêta-testeur : statut permanent (colonne `lab`, attribué à la main) OU
	tiré au sort / ajouté pour le mois en cours (table BetaGrant)."""
	if self.lab:
	return True
	return BetaGrant.query.filter_by(username=self.username, month=current_month()).first() is not None

	@property
	def unlimited_reason(self) -> str \| None:
	"""Pourquoi ce compte n'a aucune limite (None = il en a une)."""
	if self.is_admin:
	return "admin"
	if self.is_beta:
	return "beta"
	ov = self.override
	if ov is not None and ov.unlimited:
	return "custom"
	return None

	@property
	def is_unlimited(self) -> bool:
	return self.unlimited_reason is not None

	@property
	def base_daily_tokens(self) -> int:
	"""Quota de base : celui attribué à la personne s'il existe, sinon celui du forfait."""
	ov = self.override
	if ov is not None and ov.daily_tokens is not None:
	return ov.daily_tokens
	return plan_config(self.plan)["daily_tokens"]

	@property
	def daily_limit(self) -> int:
	return self.base_daily_tokens + self.bonus_tokens

	@property
	def allowed_models(self) -> list[str]:
	# Admin et bêta-testeurs : tous les modèles, y compris ceux en avant-première.
	if self.is_admin or self.is_beta:
	return list(MODEL_DISPLAY)
	preview = set(get_preview_models())
	allowed = [m for m in plan_config(self.plan)["models"] if m not in preview]
	ov = self.override
	if ov is not None:
	for model_id in ov.models:
	if model_id not in allowed and model_id not in preview:
	allowed.append(model_id)
	# Un code de promo, créé par l'admin, peut en revanche ouvrir un modèle en avant-première.
	for model_id in self.bonus_models:
	if model_id not in allowed:
	allowed.append(model_id)
	return allowed

	def quota_status(self) -> dict:
	self._reset_quota_if_needed()
	reason = self.unlimited_reason
	if reason:
	return {
	"used": self.tokens_used_today,
	"limit": 0,
	"remaining": UNLIMITED_REMAINING,
	"plan": self.plan,
	"unlimited": True,
	"unlimited_reason": reason,
	"bonus": 0,
	}
	bonus = self.bonus_tokens
	limit = self.base_daily_tokens + bonus
	return {
	"used": self.tokens_used_today,
	"limit": limit,
	"remaining": max(0, limit - self.tokens_used_today),
	"plan": self.plan,
	"unlimited": False,
	"bonus": bonus,
	"custom": self.override is not None and self.override.daily_tokens is not None,
	}

	def can_send_message(self) -> bool:
	if self.is_unlimited:
	return True
	self._reset_quota_if_needed()
	return self.tokens_used_today < self.daily_limit

	def register_message_sent(self, tokens: int = 1):
	"""À appeler une fois la réponse obtenue avec succès (pas avant),
	pour ne jamais décompter un message qui a échoué côté serveur.
	Met aussi à jour les compteurs utilisés par les missions Opsiom
	(jours actifs, nombre de conversations)."""
	self._reset_quota_if_needed()
	self.tokens_used_today += max(0, int(tokens))

	today_iso = date.today().isoformat()
	try:
	days = set(json.loads(self.active_days_json or "[]"))
	except (ValueError, TypeError):
	days = set()
	days.add(today_iso)
	self.active_days_json = json.dumps(sorted(days)[-90:]) # 90 derniers jours suffisent
	self.conversations_total += 1

	db.session.commit()

	# ------------------------------------------------------------------
	# Stats Opsiom utilisées par progression.py pour les missions
	# ------------------------------------------------------------------
	def opsiom_mission_stats(self) -> dict:
	try:
	days = json.loads(self.active_days_json or "[]")
	except (ValueError, TypeError):
	days = []
	return {
	"active_days": len(days),
	"conversations": self.conversations_total,
	}

	# ------------------------------------------------------------------
	# Progression (cache)
	# ------------------------------------------------------------------
	def cached_progression(self):
	"""Renvoie (raw_stats, is_fresh) à partir du cache local, ou
	(None, False) si rien n'est encore en cache."""
	if not self.progression_json or not self.progression_updated_at:
	return None, False
	from progression import PROGRESSION_CACHE_TTL
	fresh = (datetime.utcnow() - self.progression_updated_at) < PROGRESSION_CACHE_TTL
	try:
	return json.loads(self.progression_json), fresh
	except (ValueError, TypeError):
	return None, False

	def store_progression(self, raw_stats: dict):
	self.progression_json = json.dumps(raw_stats)
	self.progression_updated_at = datetime.utcnow()
	db.session.commit()

	def __repr__(self):
	return f"<User {self.username} plan={self.plan}>"


	class PromoCode(db.Model):
	"""Code de promo créé par un administrateur (voir promo.py).

	Un code donne, à qui l'utilise, des tokens/jour EN PLUS et/ou l'accès à des
	modèles supplémentaires, pendant `duration_hours` (24 h par défaut) à
	partir de l'utilisation. Rien n'est définitif : à l'expiration, le compte
	retrouve exactement les limites de son forfait.
	"""
	__tablename__ = T("promo_codes")

	id = db.Column(db.Integer, primary_key=True)
	code = db.Column(db.String(40), unique=True, nullable=False, index=True)
	note = db.Column(db.String(120), nullable=True) # mémo libre, visible seulement de l'admin
	bonus_tokens = db.Column(db.Integer, default=0, nullable=False)
	models_json = db.Column(db.Text, default="[]", nullable=False) # ex: ["large"]
	duration_hours = db.Column(db.Integer, default=24, nullable=False)
	max_uses = db.Column(db.Integer, nullable=True) # None = illimité
	valid_until = db.Column(db.DateTime, nullable=True) # date limite pour UTILISER le code
	active = db.Column(db.Boolean, default=True, nullable=False)
	created_by = db.Column(db.String(32), nullable=False)
	created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)

	redemptions = db.relationship(
	"PromoRedemption", backref="promo", cascade="all, delete-orphan", lazy="dynamic",
	)

	@property
	def models(self) -> list[str]:
	try:
	value = json.loads(self.models_json or "[]")
	except (ValueError, TypeError):
	return []
	return [m for m in value if m in MODEL_DISPLAY] if isinstance(value, list) else []

	@property
	def uses_count(self) -> int:
	return self.redemptions.count()

	def redeemable_error(self) -> str \| None:
	"""None si le code peut encore être utilisé, sinon la raison (en français)."""
	if not self.active:
	return "Ce code n'est plus actif."
	if self.valid_until and datetime.utcnow() > self.valid_until:
	return "Ce code a expiré."
	if self.max_uses is not None and self.uses_count >= self.max_uses:
	return "Ce code a atteint son nombre maximal d'utilisations."
	return None


	class PromoRedemption(db.Model):
	"""Utilisation d'un code de promo par un compte (une fois par compte et par code)."""
	__tablename__ = T("promo_redemptions")
	__table_args__ = (db.UniqueConstraint("user_id", "promo_id", name="uq_promo_user"),)

	id = db.Column(db.Integer, primary_key=True)
	user_id = db.Column(db.Integer, db.ForeignKey(T("users") + ".id"), nullable=False, index=True)
	promo_id = db.Column(db.Integer, db.ForeignKey(T("promo_codes") + ".id"), nullable=False, index=True)
	redeemed_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
	expires_at = db.Column(db.DateTime, nullable=False)

	user = db.relationship("User")


	# ---------------------------------------------------------------------------
	# Quotas par personne, bêta-testeurs du mois, réglages admin.
	# Ces tables sont indexées par PSEUDO Octix (et non par User.id) : l'admin peut
	# ainsi attribuer un quota — ou le tirage peut désigner quelqu'un — qui ne s'est
	# pas encore connecté à Opsiom. Nouvelles tables : db.create_all() les crée sans
	# toucher aux bases existantes.
	# ---------------------------------------------------------------------------
	def current_month() -> str:
	"""Mois en cours, 'AAAA-MM' (heure serveur, comme le reste des quotas)."""
	return date.today().strftime("%Y-%m")


	class QuotaOverride(db.Model):
	"""Quota de tokens/jour attribué à une personne précise par l'admin.

	- daily_tokens : remplace le quota du forfait (None = on garde celui du forfait) ;
	- unlimited : aucune limite de tokens ;
	- models_json : modèles supplémentaires accessibles à cette personne ;
	- expires_at : date de fin (None = permanent) ; ensuite le forfait reprend.
	"""
	__tablename__ = T("quota_overrides")

	id = db.Column(db.Integer, primary_key=True)
	username = db.Column(db.String(32), unique=True, nullable=False, index=True)
	daily_tokens = db.Column(db.Integer, nullable=True)
	unlimited = db.Column(db.Boolean, default=False, nullable=False)
	models_json = db.Column(db.Text, default="[]", nullable=False)
	note = db.Column(db.String(120), nullable=True)
	expires_at = db.Column(db.DateTime, nullable=True)
	set_by = db.Column(db.String(32), nullable=False)
	updated_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)

	@property
	def models(self) -> list[str]:
	try:
	value = json.loads(self.models_json or "[]")
	except (ValueError, TypeError):
	return []
	return [m for m in value if m in MODEL_DISPLAY] if isinstance(value, list) else []

	def is_active(self) -> bool:
	return self.expires_at is None or self.expires_at > datetime.utcnow()


	class BetaGrant(db.Model):
	"""Un bêta-testeur pour un mois donné : modèles sans aucune limite et accès
	en avant-première. `source` = 'random' (tirage mensuel) ou 'manual' (admin)."""
	__tablename__ = T("beta_grants")
	__table_args__ = (db.UniqueConstraint("username", "month", name="uq_beta_user_month"),)

	id = db.Column(db.Integer, primary_key=True)
	username = db.Column(db.String(32), nullable=False, index=True)
	month = db.Column(db.String(7), nullable=False, index=True) # 'AAAA-MM'
	source = db.Column(db.String(12), default="random", nullable=False)
	created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)


	class AppSetting(db.Model):
	"""Réglages modifiables depuis l'espace admin (clé -> valeur texte/JSON)."""
	__tablename__ = T("app_settings")

	key = db.Column(db.String(40), primary_key=True)
	value = db.Column(db.Text, nullable=False, default="")


	def get_setting(key: str, default=None):
	row = db.session.get(AppSetting, key)
	return row.value if row is not None else default


	def set_setting(key: str, value: str) -> None:
	row = db.session.get(AppSetting, key)
	if row is None:
	db.session.add(AppSetting(key=key, value=value))
	else:
	row.value = value
	db.session.commit()


	def get_preview_models() -> list[str]:
	"""Modèles en avant-première (réservés aux bêta-testeurs), choisis par l'admin."""
	raw = get_setting("preview_models")
	if raw is None:
	return list(DEFAULT_PREVIEW_MODELS)
	try:
	value = json.loads(raw)
	except (ValueError, TypeError):
	return list(DEFAULT_PREVIEW_MODELS)
	return [m for m in value if m in MODEL_DISPLAY] if isinstance(value, list) else list(DEFAULT_PREVIEW_MODELS)


	class UsageEvent(db.Model):
	"""Une tentative d'envoi de message (réussie, en erreur ou refusée par le quota),
	pour les statistiques de /admin/stats. Conservée OPSIOM_STATS_RETENTION_DAYS jours
	(30 par défaut) et bornée en nombre de lignes : voir stats.py."""
	__tablename__ = T("usage_events")

	id = db.Column(db.Integer, primary_key=True)
	created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False, index=True)
	username = db.Column(db.String(32), nullable=False, index=True)
	model_id = db.Column(db.String(32), nullable=True)
	status = db.Column(db.String(8), nullable=False, default="ok") # ok \| error \| quota
	tokens = db.Column(db.Integer, default=0, nullable=False)
	duration_ms = db.Column(db.Integer, nullable=True) # requête -> fin de la réponse
	ttft_ms = db.Column(db.Integer, nullable=True) # requête -> premier token (flux)
	used_after = db.Column(db.Integer, default=0, nullable=False) # tokens du jour après ce message
	quota_limit = db.Column(db.Integer, default=0, nullable=False) # limite du jour (0 si illimité)
	unlimited = db.Column(db.Boolean, default=False, nullable=False)
	beta = db.Column(db.Boolean, default=False, nullable=False)


	class BetaNotification(db.Model):
	"""E-mail « tu es bêta-testeur » envoyé (ou non) à une personne pour un mois."""
	__tablename__ = T("beta_notifications")
	__table_args__ = (db.UniqueConstraint("username", "month", name="uq_betanotif_user_month"),)

	id = db.Column(db.Integer, primary_key=True)
	username = db.Column(db.String(32), nullable=False, index=True)
	month = db.Column(db.String(7), nullable=False, index=True) # 'AAAA-MM' ou 'permanent'
	status = db.Column(db.String(12), nullable=False) # sent \| failed \| no_email
	detail = db.Column(db.String(200), nullable=True)
	created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)


	# Limites de l'historique (par compte) : évitent qu'une base ou un instantané
	# de sauvegarde grossisse sans fin.
	MAX_CONVERSATIONS_PER_USER = 50
	MAX_MESSAGES_PER_CONVERSATION = 200


	class Conversation(db.Model):
	"""Une conversation sauvegardée dans l'historique d'un compte.

	Chaque compte ne voit que les siennes (toutes les routes filtrent sur
	user_id) et peut les supprimer à tout moment (voir app.py, /api/conversations).
	"""
	__tablename__ = T("conversations")

	id = db.Column(db.Integer, primary_key=True)
	user_id = db.Column(db.Integer, db.ForeignKey(T("users") + ".id"), nullable=False, index=True)
	title = db.Column(db.String(80), default="Nouvelle conversation", nullable=False)
	created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
	updated_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False, index=True)

	messages = db.relationship(
	"ChatMessage", backref="conversation", cascade="all, delete-orphan",
	order_by="ChatMessage.id", lazy="dynamic",
	)

	def summary(self) -> dict:
	return {
	"id": self.id,
	"title": self.title,
	"updated_at": self.updated_at.isoformat() + "Z",
	}


	class ChatMessage(db.Model):
	"""Un message (utilisateur ou assistant) d'une conversation sauvegardée."""
	__tablename__ = T("chat_messages")

	id = db.Column(db.Integer, primary_key=True)
	conversation_id = db.Column(db.Integer, db.ForeignKey(T("conversations") + ".id"), nullable=False, index=True)
	role = db.Column(db.String(12), nullable=False) # "user" \| "assistant"
	content = db.Column(db.Text, nullable=False)
	model_id = db.Column(db.String(32), nullable=True) # modèle qui a répondu (assistant)
	created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)


	def get_or_create_user(username: str) -> User:
	user = User.query.filter_by(username=username).first()
	if user is None:
	user = User(username=username)
	db.session.add(user)
	db.session.commit()
	return user