""" Models.py Contiene los modelos de sqlalchemy para representar las entidades, campos y relaciones en la base de datos. """ from sqlalchemy import Integer, String, DateTime, ForeignKey, Boolean, BIGINT, ForeignKeyConstraint from sqlalchemy.orm import Mapped, mapped_column, relationship from datetime import datetime from typing import Optional, List from database import Base class Group(Base): """ Entidad Grupos: * Se identifica por un ID de Telegram * Contiene de uno a muchos mensajes. """ __tablename__ = 'groups' #id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) id_telegram: Mapped[int] = mapped_column(BIGINT, primary_key=True, unique=True) name: Mapped[str] = mapped_column(String(255), unique=True) description: Mapped[Optional[str]] = mapped_column(String(255)) type: Mapped[str] = mapped_column(String(30)) message_position: Mapped[int] = mapped_column(BIGINT) messages: Mapped[List["Message"]] = relationship(back_populates="group") class Message(Base): """ Entidad Mensajes: * Se identifica por ID de Mensaje en un ID de grupo específico. * Esta relacionada con un Remitente * Puede estar relacionado con un adjunto * Puede estar relacionado con una o muchas alertas. """ __tablename__ = 'messages' #id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) id_mess_g: Mapped[int] = mapped_column(BIGINT, primary_key=True) # ID en el grupo group_id: Mapped[int] = mapped_column(BIGINT, ForeignKey('groups.id_telegram'), primary_key=True) content: Mapped[str] = mapped_column(String(4096)) date: Mapped[datetime] = mapped_column(DateTime) # Foreign keys sender_id: Mapped[int] = mapped_column(ForeignKey('senders.id_telegram')) # Relationships sender: Mapped["Sender"] = relationship(back_populates="messages") group: Mapped["Group"] = relationship(back_populates="messages") attachments: Mapped[List["Attachment"]] = relationship(back_populates="message") alerts: Mapped[List["Alert"]] = relationship(back_populates="message") class Attachment(Base): """ Clase Adjuntos: * Se identifica según su ID. * Esta relacionado a un mensaje en un grupo específico. """ __tablename__ = 'attachments' id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) type: Mapped[str] = mapped_column(String(50)) description: Mapped[Optional[str]] = mapped_column(String(512)) isDownloaded: Mapped[bool] = mapped_column(Boolean) message_id: Mapped[int] = mapped_column(BIGINT) group_id: Mapped[int] = mapped_column(BIGINT) __table_args__ = ( ForeignKeyConstraint( ['message_id', 'group_id'], ['messages.id_mess_g', 'messages.group_id'] ), ) message: Mapped["Message"] = relationship( back_populates="attachments", foreign_keys="[Attachment.message_id, Attachment.group_id]" ) class Sender(Base): """ Entidad "Sender" o Remitente: * Se identifica según el ID proporcionado por Telegram. * Esta relacionado con un o muchos mensajes. """ __tablename__ = 'senders' #id: Mapped[int] = mapped_column(Integer, index=True) id_telegram: Mapped[int] = mapped_column(BIGINT, primary_key=True, unique=True) type: Mapped[str] = mapped_column(String(10)) # user/channel/chat username: Mapped[Optional[str]] = mapped_column(String(100)) first_name: Mapped[Optional[str]] = mapped_column(String(100)) last_name: Mapped[Optional[str]] = mapped_column(String(100)) phone: Mapped[Optional[str]] = mapped_column(String(20)) messages: Mapped[List["Message"]] = relationship(back_populates="sender") # Entidad Reglas: # class Rule(Base): """ Entidad Regla: * Se identifica según su ID. * Una regla puede relacionarse con ninguna, una o múltiples alertas. """ __tablename__ = 'rules' id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) description: Mapped[str] = mapped_column(String(255)) regex: Mapped[str] = mapped_column(String(1024)) severity: Mapped[str] = mapped_column(String(20)) # baja/media/alta is_active: Mapped[bool] = mapped_column(Boolean, default=True) alerts: Mapped[List["Alert"]] = relationship(back_populates="rule") class Alert(Base): """Entidad Alertas:+ * Una alerta se identifica mediante un mensaje en un grupo en específico * Una o muchas Alertas se relacionan con una regla. * Una alerta esta relacionada con ninguna, una o múltiples notas. """ __tablename__ = 'alerts' id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) message_id: Mapped[int] = mapped_column(BIGINT) group_id: Mapped[int] = mapped_column(BIGINT) rule_id: Mapped[int] = mapped_column(ForeignKey('rules.id')) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) status: Mapped[str] = mapped_column(String(20), default='open') notes: Mapped[Optional[str]] = mapped_column(String(500)) __table_args__ = ( ForeignKeyConstraint( ['message_id', 'group_id'], ['messages.id_mess_g', 'messages.group_id'] ), ) message: Mapped["Message"] = relationship(back_populates="alerts") rule: Mapped["Rule"] = relationship(back_populates="alerts") notes_list: Mapped[List["Note"]] = relationship(back_populates="alert") class Note(Base): """ Entidad Nota: * Se identifica según ID de mensaje y ID de grupo específico. * Una o muchas notas estan relaciónada a una Alerta. """ __tablename__ = 'notes' id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) alert_id: Mapped[int] = mapped_column(ForeignKey('alerts.id'), index=True) user_id: Mapped[int] = mapped_column(Integer) content: Mapped[str] = mapped_column(String(1000)) creation_date: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) alert: Mapped["Alert"] = relationship(back_populates="notes_list") # Entidad para almacenar logs de auditoría class AuditLog(Base): __tablename__ = 'audit_logs' id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) entity_type: Mapped[str] = mapped_column(String(50)) # rule, alert, group, message, sender entity_id: Mapped[str] = mapped_column(String(100)) # string para cubrir PKs compuestas action: Mapped[str] = mapped_column(String(20)) # create, update, delete, status_change user_id: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) before_value: Mapped[Optional[str]] = mapped_column(String(4096), nullable=True) # JSON serializado after_value: Mapped[Optional[str]] = mapped_column(String(4096), nullable=True) # JSON serializado timestamp: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) ip_address: Mapped[Optional[str]] = mapped_column(String(45), nullable=True)