7743919695
Bot de Telegram para monitoreo de notificaciones judiciales del PJ Paraguay. Multi-tenant con onboarding conversacional, audit log, y polling automático. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
114 lines
3.5 KiB
Python
114 lines
3.5 KiB
Python
"""
|
|
config.py — Carga y valida variables de entorno.
|
|
|
|
Todas las partes de la app importan Settings desde acá.
|
|
Las credenciales del PJ se descifran aquí y se exponen como propiedades.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from functools import cached_property
|
|
from pathlib import Path
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
from pydantic import field_validator, model_validator
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
model_config = SettingsConfigDict(
|
|
env_file=".env",
|
|
env_file_encoding="utf-8",
|
|
case_sensitive=False,
|
|
extra="ignore",
|
|
)
|
|
|
|
# Cifrado
|
|
fernet_key: str
|
|
|
|
# Credenciales PJ cifradas (valores Fernet base64)
|
|
pj_usuario_enc: str
|
|
pj_password_enc: str
|
|
|
|
# Telegram
|
|
telegram_bot_token: str
|
|
telegram_chat_id: int # chat ID principal (recibe notificaciones automáticas)
|
|
telegram_extra_ids: str = "" # IDs adicionales separados por coma, ej: "111,222"
|
|
|
|
@property
|
|
def telegram_authorized_ids(self) -> set[int]:
|
|
"""Todos los chat IDs habilitados para usar el bot."""
|
|
ids = {self.telegram_chat_id}
|
|
for raw in self.telegram_extra_ids.split(","):
|
|
raw = raw.strip()
|
|
if raw:
|
|
ids.add(int(raw))
|
|
return ids
|
|
|
|
# Configuración del polling
|
|
poll_interval_minutes: int = 60
|
|
poll_hora_inicio: int = 7
|
|
poll_hora_fin: int = 18
|
|
pj_rol_id: int = 16
|
|
|
|
# Directorios
|
|
data_dir: Path = Path("./data")
|
|
db_path: Path = Path("./data/pasante.db")
|
|
|
|
# Logging
|
|
log_level: str = "INFO"
|
|
|
|
# URLs del PJ (no configurables, solo para tests)
|
|
pj_base_url: str = "https://apps.csj.gov.py"
|
|
|
|
@field_validator("poll_hora_inicio", "poll_hora_fin")
|
|
@classmethod
|
|
def validar_hora(cls, v: int) -> int:
|
|
if not (0 <= v <= 23):
|
|
raise ValueError(f"Hora debe ser entre 0 y 23, recibido: {v}")
|
|
return v
|
|
|
|
@model_validator(mode="after")
|
|
def validar_horario(self) -> Settings:
|
|
if self.poll_hora_inicio >= self.poll_hora_fin:
|
|
raise ValueError(
|
|
f"poll_hora_inicio ({self.poll_hora_inicio}) "
|
|
f"debe ser menor que poll_hora_fin ({self.poll_hora_fin})"
|
|
)
|
|
return self
|
|
|
|
@cached_property
|
|
def _fernet(self) -> Fernet:
|
|
try:
|
|
return Fernet(self.fernet_key.encode())
|
|
except Exception as e:
|
|
raise ValueError(f"FERNET_KEY inválida: {e}") from e
|
|
|
|
def descifrar_usuario(self) -> str:
|
|
"""Descifra y retorna la cédula del abogado. Usar solo al momento de login."""
|
|
try:
|
|
return self._fernet.decrypt(self.pj_usuario_enc.encode()).decode()
|
|
except InvalidToken as e:
|
|
raise ValueError("PJ_USUARIO_ENC no se pudo descifrar. ¿Usaste la misma FERNET_KEY?") from e
|
|
|
|
def descifrar_password(self) -> str:
|
|
"""Descifra y retorna el password del PJ. Usar solo al momento de login."""
|
|
try:
|
|
return self._fernet.decrypt(self.pj_password_enc.encode()).decode()
|
|
except InvalidToken as e:
|
|
raise ValueError("PJ_PASSWORD_ENC no se pudo descifrar. ¿Usaste la misma FERNET_KEY?") from e
|
|
|
|
def asegurar_data_dir(self) -> None:
|
|
self.data_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
# Singleton — importar desde cualquier lado
|
|
_settings: Settings | None = None
|
|
|
|
|
|
def get_settings() -> Settings:
|
|
global _settings
|
|
if _settings is None:
|
|
_settings = Settings()
|
|
_settings.asegurar_data_dir()
|
|
return _settings
|