Initial commit — Pedrito Hechakuaa v0
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>
This commit is contained in:
113
config.py
Normal file
113
config.py
Normal file
@@ -0,0 +1,113 @@
|
||||
"""
|
||||
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
|
||||
Reference in New Issue
Block a user