219 lines
5.2 KiB
Python
219 lines
5.2 KiB
Python
from pydantic import BaseModel, field_validator
|
|
from typing import Optional, List
|
|
from datetime import datetime
|
|
|
|
# Group Schemas
|
|
class GroupBase(BaseModel):
|
|
id_telegram: int
|
|
name: str
|
|
description: Optional[str] = None
|
|
type: str
|
|
message_position: int
|
|
|
|
class GroupCreate(GroupBase):
|
|
pass
|
|
|
|
class GroupResponse(GroupBase):
|
|
# Eliminar la lista de messages para evitar recursión
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
class GroupWithMessages(GroupBase):
|
|
# Solo incluir información básica de los mensajes, no completa
|
|
messages: List["MessageSimple"] = []
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
class GroupUpdatePosition(BaseModel):
|
|
message_position: int
|
|
|
|
class GroupUpdate(BaseModel):
|
|
name: str
|
|
description: Optional[str] = None
|
|
type: str
|
|
|
|
# Sender Schemas
|
|
class SenderBase(BaseModel):
|
|
id_telegram: int
|
|
type: str
|
|
username: Optional[str] = None
|
|
first_name: Optional[str] = None
|
|
last_name: Optional[str] = None
|
|
phone: Optional[str] = None
|
|
|
|
class SenderCreate(SenderBase):
|
|
pass
|
|
|
|
class SenderResponse(SenderBase):
|
|
# Eliminar la lista de messages para evitar recursión
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
class SenderWithMessages(SenderBase):
|
|
# Solo incluir información básica de los mensajes
|
|
messages: List["MessageSimple"] = []
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
# Message Schemas - Versiones simples para evitar recursión
|
|
class MessageSimple(BaseModel):
|
|
id_mess_g: int
|
|
group_id: int
|
|
content: str
|
|
date: datetime
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
class MessageBase(BaseModel):
|
|
id_mess_g: int
|
|
content: str
|
|
date: datetime
|
|
sender_id: int
|
|
group_id: int
|
|
|
|
class MessageCreate(MessageBase):
|
|
pass
|
|
|
|
class MessageResponse(MessageBase):
|
|
sender: Optional[SenderResponse] = None
|
|
group: Optional[GroupResponse] = None
|
|
attachments: List["AttachmentResponse"] = []
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
class MessageWithRelations(MessageBase):
|
|
# Versión con relaciones completas (usar con cuidado)
|
|
sender: Optional[SenderResponse] = None
|
|
group: Optional[GroupResponse] = None
|
|
attachments: List["AttachmentResponse"] = []
|
|
alerts: List["AlertResponse"] = []
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
class AttachmentBase(BaseModel):
|
|
type: str
|
|
description: Optional[str] = None
|
|
isDownloaded: bool
|
|
|
|
class AttachmentCreate(AttachmentBase):
|
|
message_id: int
|
|
group_id: int # necesario para la FK compuesta
|
|
|
|
class AttachmentResponse(AttachmentBase):
|
|
id: int
|
|
message_id: int
|
|
group_id: int
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
# Rule Schemas
|
|
class RuleBase(BaseModel):
|
|
description: str
|
|
regex: str
|
|
severity: str = "media"
|
|
is_active: bool = True
|
|
|
|
@field_validator("description")
|
|
@classmethod
|
|
def description_not_empty(cls, v: str) -> str:
|
|
if not v or not v.strip():
|
|
raise ValueError("La descripción no puede estar vacía")
|
|
return v.strip()
|
|
|
|
@field_validator("regex")
|
|
@classmethod
|
|
def regex_not_empty(cls, v: str) -> str:
|
|
if not v or not v.strip():
|
|
raise ValueError("La regex no puede estar vacía")
|
|
import re
|
|
try:
|
|
re.compile(v)
|
|
except re.error as e:
|
|
raise ValueError(f"La regex no es válida: {e}")
|
|
return v.strip()
|
|
|
|
@field_validator("severity")
|
|
@classmethod
|
|
def severity_valid(cls, v: str) -> str:
|
|
allowed = {"baja", "media", "alta"}
|
|
if v.lower() not in allowed:
|
|
raise ValueError(f"Severidad debe ser una de: {', '.join(allowed)}")
|
|
return v.lower()
|
|
|
|
class RuleCreate(RuleBase):
|
|
apply_to_history: bool = False # nuevo campo
|
|
pass
|
|
|
|
class RuleResponse(RuleBase):
|
|
id: int
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
# Alert Schemas
|
|
class AlertBase(BaseModel):
|
|
message_id: int
|
|
group_id: int
|
|
rule_id: int
|
|
status: str = "open"
|
|
notes: Optional[str] = None
|
|
|
|
class AlertCreate(AlertBase):
|
|
pass
|
|
|
|
class AlertResponse(AlertBase):
|
|
id: int
|
|
created_at: datetime
|
|
message: Optional[MessageSimple] = None
|
|
rule: Optional[RuleResponse] = None
|
|
notes_list: List["NoteResponse"] = []
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
# Note Schemas
|
|
class NoteBase(BaseModel):
|
|
alert_id: int
|
|
user_id: int
|
|
content: str
|
|
creation_date: datetime
|
|
|
|
class NoteCreate(BaseModel):
|
|
alert_id: int
|
|
user_id: int
|
|
content: str
|
|
|
|
class NoteResponse(NoteBase):
|
|
id: int
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
class AuditLogResponse(BaseModel):
|
|
id: int
|
|
entity_type: str
|
|
entity_id: str
|
|
action: str
|
|
user_id: Optional[int] = None
|
|
before_value: Optional[str] = None
|
|
after_value: Optional[str] = None
|
|
timestamp: datetime
|
|
ip_address: Optional[str] = None
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
# Resolver referencias circulares
|
|
MessageSimple.model_rebuild()
|
|
MessageResponse.model_rebuild()
|
|
MessageWithRelations.model_rebuild()
|
|
AttachmentResponse.model_rebuild()
|
|
AlertResponse.model_rebuild()
|
|
GroupWithMessages.model_rebuild()
|
|
SenderWithMessages.model_rebuild()
|
|
NoteResponse.model_rebuild() |