7865008618
- utils/rate_limiter.py: ventana deslizante 20 req/60s por chat_id, aplicado en todos los handlers de telegram_bot.py - database.py: tabla invite_codes (un solo uso por código), funciones crear_invite_code() y verificar_y_consumir_invite_code() - onboarding.py: verifica primero en DB (un solo uso), fallback al INVITE_CODE del .env (ilimitado, para uso interno) - scripts/generar_codigo.py: genera 1 o N códigos de invitación - scripts/rotar_fernet_key.py: re-cifra todos los tenants y el .env con nueva FERNET_KEY, backup automático de .env antes de escribir Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
117 lines
3.6 KiB
Python
117 lines
3.6 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")
|
|
|
|
# Código de invitación para registro (vacío = registro abierto)
|
|
invite_code: str = ""
|
|
|
|
# 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
|