290 lines
No EOL
9.5 KiB
Python
290 lines
No EOL
9.5 KiB
Python
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"<EmailLog {self.id} to={self.destinataire} status={self.statut.value}>"
|
|
|
|
|
|
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)
|
|
|
|
# Métadonnées
|
|
raison_refus = Column(Text, nullable=True)
|
|
ip_signature = Column(String(45), nullable=True)
|
|
|
|
def __repr__(self):
|
|
return f"<SignatureLog {self.id} doc={self.document_id} status={self.statut.value}>"
|
|
|
|
|
|
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"<WorkflowLog {self.document_source} → {self.document_cible}>"
|
|
|
|
|
|
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"<CacheMetadata type={self.cache_type} items={self.item_count}>"
|
|
|
|
|
|
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"<AuditLog {self.action} on {self.ressource_type}/{self.ressource_id}>"
|
|
|
|
# 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"<User {self.email} verified={self.is_verified}>"
|
|
|
|
|
|
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"<RefreshToken user={self.user_id} revoked={self.is_revoked}>"
|
|
|
|
|
|
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"<LoginAttempt {self.email} success={self.success}>" |