from sqlalchemy import ( Column, String, DateTime, Boolean, Integer, Text, Enum as SQLEnum, ForeignKey, Index, ) from sqlalchemy.orm import relationship from datetime import datetime from enum import Enum from database.models.generic_model import Base class UniversignTransactionStatus(str, Enum): DRAFT = "draft" READY = "ready" STARTED = "started" COMPLETED = "completed" CLOSED = "closed" REFUSED = "refused" EXPIRED = "expired" CANCELED = "canceled" FAILED = "failed" class UniversignSignerStatus(str, Enum): WAITING = "waiting" OPEN = "open" VIEWED = "viewed" SIGNED = "signed" COMPLETED = "completed" REFUSED = "refused" EXPIRED = "expired" STALLED = "stalled" UNKNOWN = "unknown" class LocalDocumentStatus(str, Enum): PENDING = "EN_ATTENTE" IN_PROGRESS = "EN_COURS" SIGNED = "SIGNE" REJECTED = "REFUSE" EXPIRED = "EXPIRE" ERROR = "ERREUR" class SageDocumentType(int, Enum): DEVIS = 0 BON_COMMANDE = 10 PREPARATION = 20 BON_LIVRAISON = 30 BON_RETOUR = 40 BON_AVOIR = 50 FACTURE = 60 class UniversignTransaction(Base): __tablename__ = "universign_transactions" id = Column(String(36), primary_key=True) transaction_id = Column( String(255), unique=True, nullable=False, index=True, comment="ID Universign (ex: tr_abc123)", ) sage_document_id = Column( String(50), nullable=False, index=True, comment="Numéro du document Sage (ex: DE00123)", ) sage_document_type = Column( SQLEnum(SageDocumentType), nullable=False, comment="Type de document Sage" ) universign_status = Column( SQLEnum(UniversignTransactionStatus), nullable=False, default=UniversignTransactionStatus.DRAFT, index=True, comment="Statut brut Universign", ) universign_status_updated_at = Column( DateTime, nullable=True, comment="Dernière MAJ du statut Universign" ) local_status = Column( SQLEnum(LocalDocumentStatus), nullable=False, default=LocalDocumentStatus.PENDING, index=True, comment="Statut métier simplifié pour l'UI", ) signer_url = Column(Text, nullable=True, comment="URL de signature") document_url = Column(Text, nullable=True, comment="URL du document signé") signed_document_path = Column( Text, nullable=True, comment="Chemin local du PDF signé" ) signed_document_downloaded_at = Column( DateTime, nullable=True, comment="Date de téléchargement du document" ) signed_document_size_bytes = Column( Integer, nullable=True, comment="Taille du fichier en octets" ) download_attempts = Column( Integer, default=0, comment="Nombre de tentatives de téléchargement" ) download_error = Column( Text, nullable=True, comment="Dernière erreur de téléchargement" ) certificate_url = Column(Text, nullable=True, comment="URL du certificat") signers_data = Column( Text, nullable=True, comment="JSON des signataires (snapshot)" ) requester_email = Column(String(255), nullable=True) requester_name = Column(String(255), nullable=True) document_name = Column(String(500), nullable=True) created_at = Column( DateTime, default=datetime.now, nullable=False, comment="Date de création locale", ) sent_at = Column( DateTime, nullable=True, comment="Date d'envoi Universign (started)" ) signed_at = Column(DateTime, nullable=True, comment="Date de signature complète") refused_at = Column(DateTime, nullable=True) expired_at = Column(DateTime, nullable=True) canceled_at = Column(DateTime, nullable=True) last_synced_at = Column( DateTime, nullable=True, comment="Dernière sync réussie avec Universign" ) sync_attempts = Column(Integer, default=0, comment="Nombre de tentatives de sync") sync_error = Column(Text, nullable=True) is_test = Column( Boolean, default=False, comment="Transaction en environnement .alpha" ) needs_sync = Column( Boolean, default=True, index=True, comment="À synchroniser avec Universign" ) webhook_received = Column(Boolean, default=False, comment="Webhook Universign reçu") signers = relationship( "UniversignSigner", back_populates="transaction", cascade="all, delete-orphan" ) sync_logs = relationship( "UniversignSyncLog", back_populates="transaction", cascade="all, delete-orphan" ) __table_args__ = ( Index("idx_sage_doc", "sage_document_id", "sage_document_type"), Index("idx_sync_status", "needs_sync", "universign_status"), Index("idx_dates", "created_at", "signed_at"), ) def __repr__(self): return ( f"" ) class UniversignSigner(Base): __tablename__ = "universign_signers" id = Column(String(36), primary_key=True) transaction_id = Column( String(36), ForeignKey("universign_transactions.id", ondelete="CASCADE"), nullable=False, index=True, ) email = Column(String(255), nullable=False, index=True) name = Column(String(255), nullable=True) phone = Column(String(50), nullable=True) status = Column( SQLEnum(UniversignSignerStatus), default=UniversignSignerStatus.WAITING, nullable=False, ) viewed_at = Column(DateTime, nullable=True) signed_at = Column(DateTime, nullable=True) refused_at = Column(DateTime, nullable=True) refusal_reason = Column(Text, nullable=True) ip_address = Column(String(45), nullable=True) user_agent = Column(Text, nullable=True) signature_method = Column(String(50), nullable=True) order_index = Column(Integer, default=0) transaction = relationship("UniversignTransaction", back_populates="signers") def __repr__(self): return f"" class UniversignSyncLog(Base): __tablename__ = "universign_sync_logs" id = Column(Integer, primary_key=True, autoincrement=True) transaction_id = Column( String(36), ForeignKey("universign_transactions.id", ondelete="CASCADE"), nullable=False, index=True, ) sync_type = Column(String(50), nullable=False, comment="webhook, polling, manual") sync_timestamp = Column(DateTime, default=datetime.now, nullable=False, index=True) previous_status = Column(String(50), nullable=True) new_status = Column(String(50), nullable=True) changes_detected = Column(Text, nullable=True, comment="JSON des changements") success = Column(Boolean, default=True) error_message = Column(Text, nullable=True) http_status_code = Column(Integer, nullable=True) response_time_ms = Column(Integer, nullable=True) transaction = relationship("UniversignTransaction", back_populates="sync_logs") def __repr__(self): return f"" class UniversignConfig(Base): __tablename__ = "universign_configs" id = Column(String(36), primary_key=True) user_id = Column(String(36), nullable=True, index=True) environment = Column( String(50), nullable=False, default="alpha", comment="alpha, prod" ) api_url = Column(String(500), nullable=False) api_key = Column(String(500), nullable=False, comment="À chiffrer") webhook_url = Column(String(500), nullable=True) webhook_secret = Column(String(255), nullable=True) auto_sync_enabled = Column(Boolean, default=True) sync_interval_minutes = Column(Integer, default=5) signature_expiry_days = Column(Integer, default=30) is_active = Column(Boolean, default=True) created_at = Column(DateTime, default=datetime.now) def __repr__(self): return f""