from sqlalchemy import ( Column, Integer, String, DateTime, Float, Text, Boolean, Enum as SQLEnum, ) from sqlalchemy.ext.declarative import declarative_base from datetime import datetime import enum Base = declarative_base() # ============================================================================ # Enums # ============================================================================ class StatutEmail(str, enum.Enum): """Statuts possibles d'un email""" EN_ATTENTE = "EN_ATTENTE" EN_COURS = "EN_COURS" ENVOYE = "ENVOYE" OUVERT = "OUVERT" ERREUR = "ERREUR" BOUNCE = "BOUNCE" class StatutSignature(str, enum.Enum): """Statuts possibles d'une signature électronique""" EN_ATTENTE = "EN_ATTENTE" ENVOYE = "ENVOYE" SIGNE = "SIGNE" REFUSE = "REFUSE" EXPIRE = "EXPIRE" # ============================================================================ # Tables # ============================================================================ class EmailLog(Base): """ Journal des emails envoyés via l'API Permet le suivi et le retry automatique """ __tablename__ = "email_logs" # Identifiant id = Column(String(36), primary_key=True) # Destinataires destinataire = Column(String(255), nullable=False, index=True) cc = Column(Text, nullable=True) # JSON stringifié cci = Column(Text, nullable=True) # JSON stringifié # Contenu sujet = Column(String(500), nullable=False) corps_html = Column(Text, nullable=False) # Documents attachés document_ids = Column(Text, nullable=True) # Séparés par virgules type_document = Column(Integer, nullable=True) # Statut statut = Column(SQLEnum(StatutEmail), default=StatutEmail.EN_ATTENTE, index=True) # Tracking temporel date_creation = Column(DateTime, default=datetime.now, nullable=False) date_envoi = Column(DateTime, nullable=True) date_ouverture = Column(DateTime, nullable=True) # Retry automatique nb_tentatives = Column(Integer, default=0) derniere_erreur = Column(Text, nullable=True) prochain_retry = Column(DateTime, nullable=True) # Métadonnées ip_envoi = Column(String(45), nullable=True) user_agent = Column(String(500), nullable=True) def __repr__(self): return f"" class SignatureLog(Base): """ Journal des demandes de signature Universign Permet le suivi du workflow de signature """ __tablename__ = "signature_logs" # Identifiant id = Column(String(36), primary_key=True) # Document Sage associé document_id = Column(String(100), nullable=False, index=True) type_document = Column(Integer, nullable=False) # Universign transaction_id = Column(String(100), unique=True, index=True, nullable=True) signer_url = Column(String(500), nullable=True) # Signataire email_signataire = Column(String(255), nullable=False, index=True) nom_signataire = Column(String(255), nullable=False) # Statut statut = Column( SQLEnum(StatutSignature), default=StatutSignature.EN_ATTENTE, index=True ) date_envoi = Column(DateTime, default=datetime.now) date_signature = Column(DateTime, nullable=True) date_refus = Column(DateTime, nullable=True) # Relances est_relance = Column(Boolean, default=False) nb_relances = Column(Integer, default=0) derniere_relance = Column(DateTime, nullable=True) # Métadonnées raison_refus = Column(Text, nullable=True) ip_signature = Column(String(45), nullable=True) def __repr__(self): return f"" class WorkflowLog(Base): """ Journal des transformations de documents (Devis → Commande → Facture) Permet la traçabilité du workflow commercial """ __tablename__ = "workflow_logs" # Identifiant id = Column(String(36), primary_key=True) # Documents document_source = Column(String(100), nullable=False, index=True) type_source = Column(Integer, nullable=False) # 0=Devis, 3=Commande, etc. document_cible = Column(String(100), nullable=False, index=True) type_cible = Column(Integer, nullable=False) # Métadonnées de transformation nb_lignes = Column(Integer, nullable=True) montant_ht = Column(Float, nullable=True) montant_ttc = Column(Float, nullable=True) # Tracking date_transformation = Column(DateTime, default=datetime.now, nullable=False) utilisateur = Column(String(100), nullable=True) # Résultat succes = Column(Boolean, default=True) erreur = Column(Text, nullable=True) duree_ms = Column(Integer, nullable=True) # Durée en millisecondes def __repr__(self): return f"" class CacheMetadata(Base): """ Métadonnées sur le cache Sage (clients, articles) Permet le monitoring du cache géré par la gateway Windows """ __tablename__ = "cache_metadata" id = Column(Integer, primary_key=True, autoincrement=True) # Type de cache cache_type = Column( String(50), unique=True, nullable=False ) # 'clients' ou 'articles' # Statistiques last_refresh = Column(DateTime, default=datetime.now) item_count = Column(Integer, default=0) refresh_duration_ms = Column(Float, nullable=True) # Santé last_error = Column(Text, nullable=True) error_count = Column(Integer, default=0) def __repr__(self): return f"" class AuditLog(Base): """ Journal d'audit pour la sécurité et la conformité Trace toutes les actions importantes dans l'API """ __tablename__ = "audit_logs" id = Column(Integer, primary_key=True, autoincrement=True) # Action action = Column( String(100), nullable=False, index=True ) # 'CREATE_DEVIS', 'SEND_EMAIL', etc. ressource_type = Column(String(50), nullable=True) # 'devis', 'facture', etc. ressource_id = Column(String(100), nullable=True, index=True) # Utilisateur (si authentification ajoutée plus tard) utilisateur = Column(String(100), nullable=True) ip_address = Column(String(45), nullable=True) # Résultat succes = Column(Boolean, default=True) details = Column(Text, nullable=True) # JSON stringifié erreur = Column(Text, nullable=True) # Timestamp date_action = Column(DateTime, default=datetime.now, nullable=False, index=True) def __repr__(self): return f"" # Ajouter ces modèles à la fin de database/models.py class User(Base): """ Utilisateurs de l'API avec validation email """ __tablename__ = "users" id = Column(String(36), primary_key=True) email = Column(String(255), unique=True, nullable=False, index=True) hashed_password = Column(String(255), nullable=False) # Profil nom = Column(String(100), nullable=False) prenom = Column(String(100), nullable=False) role = Column(String(50), default="user") # user, admin, commercial # Validation email is_verified = Column(Boolean, default=False) verification_token = Column(String(255), nullable=True, unique=True, index=True) verification_token_expires = Column(DateTime, nullable=True) # Sécurité is_active = Column(Boolean, default=True) failed_login_attempts = Column(Integer, default=0) locked_until = Column(DateTime, nullable=True) # Mot de passe oublié reset_token = Column(String(255), nullable=True, unique=True, index=True) reset_token_expires = Column(DateTime, nullable=True) # Timestamps created_at = Column(DateTime, default=datetime.now, nullable=False) updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now) last_login = Column(DateTime, nullable=True) def __repr__(self): return f"" class RefreshToken(Base): """ Tokens de rafraîchissement JWT """ __tablename__ = "refresh_tokens" id = Column(String(36), primary_key=True) user_id = Column(String(36), nullable=False, index=True) token_hash = Column(String(255), nullable=False, unique=True, index=True) # Métadonnées device_info = Column(String(500), nullable=True) ip_address = Column(String(45), nullable=True) # Expiration expires_at = Column(DateTime, nullable=False) created_at = Column(DateTime, default=datetime.now, nullable=False) # Révocation is_revoked = Column(Boolean, default=False) revoked_at = Column(DateTime, nullable=True) def __repr__(self): return f"" class LoginAttempt(Base): """ Journal des tentatives de connexion (détection bruteforce) """ __tablename__ = "login_attempts" id = Column(Integer, primary_key=True, autoincrement=True) email = Column(String(255), nullable=False, index=True) ip_address = Column(String(45), nullable=False, index=True) user_agent = Column(String(500), nullable=True) success = Column(Boolean, default=False) failure_reason = Column(String(255), nullable=True) timestamp = Column(DateTime, default=datetime.now, nullable=False, index=True) def __repr__(self): return f""