197 lines
7.6 KiB
Python
197 lines
7.6 KiB
Python
"""Configuración centralizada de la aplicación.
|
|
|
|
Usa Pydantic BaseSettings para cargar valores por defecto y variables de entorno.
|
|
Se lee automáticamente un archivo `.env` en la raíz del proyecto si existe.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional, Tuple
|
|
import os
|
|
|
|
# Cargar variables desde .env en el entorno (para que otros módulos que usan os.environ las vean)
|
|
try:
|
|
from dotenv import load_dotenv # type: ignore
|
|
# En desarrollo, dar precedencia al archivo .env sobre variables previas del entorno
|
|
load_dotenv(override=True)
|
|
except Exception:
|
|
pass
|
|
|
|
from pydantic import Field
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
from pydantic import AliasChoices
|
|
|
|
|
|
class ApiSettings(BaseSettings):
|
|
"""Backends LLM compatibles con la API de OpenAI.
|
|
|
|
Extraccion y normalizacion tienen credenciales independientes para poder
|
|
mezclar proveedores, por ejemplo Grok para vision y OpenAI para normalizar.
|
|
"""
|
|
extractor_provider: str = Field("custom", validation_alias=AliasChoices("EXTRACTOR_PROVIDER"))
|
|
extractor_base_url: str = Field("", validation_alias=AliasChoices("EXTRACTOR_BASE_URL"))
|
|
extractor_api_key: str = Field("", validation_alias=AliasChoices("EXTRACTOR_API_KEY"))
|
|
extractor_model: str = Field("gpt-5.6-luna", validation_alias=AliasChoices("EXTRACTOR_MODEL"))
|
|
|
|
normalizer_provider: str = Field("custom", validation_alias=AliasChoices("NORMALIZER_PROVIDER"))
|
|
normalizer_base_url: str = Field("", validation_alias=AliasChoices("NORMALIZER_BASE_URL"))
|
|
normalizer_api_key: str = Field("", validation_alias=AliasChoices("NORMALIZER_API_KEY"))
|
|
normalizer_model: str = Field("gpt-5.6-luna", validation_alias=AliasChoices("NORMALIZER_MODEL"))
|
|
|
|
request_timeout_s: int = Field(120, validation_alias=AliasChoices("EXTRACTOR_TIMEOUT_S", "LLM_TIMEOUT_S"))
|
|
normalizer_timeout_s: int = Field(0, validation_alias=AliasChoices("NORMALIZER_TIMEOUT_S"))
|
|
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
|
|
|
|
@property
|
|
def base_url(self) -> str:
|
|
"""Alias de compatibilidad para codigo anterior: backend de extraccion."""
|
|
return self.extractor_base_url
|
|
|
|
@property
|
|
def api_key(self) -> str:
|
|
"""Alias de compatibilidad para codigo anterior: API key de extraccion."""
|
|
return self.extractor_api_key
|
|
|
|
@property
|
|
def model(self) -> str:
|
|
"""Alias de compatibilidad para codigo anterior: modelo de extraccion."""
|
|
return self.extractor_model
|
|
|
|
@property
|
|
def effective_normalizer_base_url(self) -> str:
|
|
return self.normalizer_base_url or self.extractor_base_url
|
|
|
|
@property
|
|
def effective_normalizer_api_key(self) -> str:
|
|
return self.normalizer_api_key or self.extractor_api_key
|
|
|
|
@property
|
|
def effective_normalizer_timeout_s(self) -> int:
|
|
return self.normalizer_timeout_s or self.request_timeout_s
|
|
|
|
|
|
class ImagingSettings(BaseSettings):
|
|
"""Parámetros de conversión de documentos a imágenes."""
|
|
poppler_path: Optional[str] = Field(None, validation_alias=AliasChoices("POPPLER_PATH"))
|
|
pdf_dpi: int = Field(600, validation_alias=AliasChoices("PDF_DPI"))
|
|
pdf_thread_count: int = Field(10, validation_alias=AliasChoices("PDF_THREAD_COUNT"))
|
|
tiff_pages: Optional[Tuple[int, int]] = Field((2, 3), validation_alias=AliasChoices("TIFF_PAGES"))
|
|
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
|
|
|
|
|
|
class RetrySettings(BaseSettings):
|
|
"""Estrategia de reintentos para llamadas al backend."""
|
|
max_attempts: int = Field(3, validation_alias=AliasChoices("RETRY_MAX_ATTEMPTS"))
|
|
wait_min_s: float = Field(1.0, validation_alias=AliasChoices("RETRY_WAIT_MIN_S"))
|
|
wait_max_s: float = Field(8.0, validation_alias=AliasChoices("RETRY_WAIT_MAX_S"))
|
|
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
|
|
|
|
|
|
class ProcessingSettings(BaseSettings):
|
|
"""Configuración de procesamiento y concurrencia."""
|
|
concurrency_limit: int = Field(50, validation_alias=AliasChoices("CONCURRENCY_LIMIT"))
|
|
redis_url: str = Field("redis://localhost:6379/0", validation_alias=AliasChoices("REDIS_URL"))
|
|
worker_concurrency: int = Field(25, validation_alias=AliasChoices("WORKER_CONCURRENCY"))
|
|
# Límites OpenAI para el rate limiter distribuido
|
|
openai_rpm_limit: int = Field(500, validation_alias=AliasChoices("OPENAI_RPM_LIMIT"))
|
|
openai_tpm_limit: int = Field(800_000, validation_alias=AliasChoices("OPENAI_TPM_LIMIT"))
|
|
# Directorio compartido entre API y workers para encolar archivos
|
|
queue_dir: str = Field("/tmp/ips_queue", validation_alias=AliasChoices("QUEUE_DIR"))
|
|
# False (default) = procesamiento sincrono sin Redis/workers.
|
|
# True = cola ARQ distribuida (requiere Redis corriendo y workers activos).
|
|
async_queue: bool = Field(False, validation_alias=AliasChoices("ASYNC_QUEUE"))
|
|
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
|
|
|
|
|
|
class AppSettings(BaseSettings):
|
|
"""Configuración raíz agrupada."""
|
|
api: ApiSettings = ApiSettings()
|
|
imaging: ImagingSettings = ImagingSettings()
|
|
retry: RetrySettings = RetrySettings()
|
|
processing: ProcessingSettings = ProcessingSettings()
|
|
csv_delimiter: str = Field("|", validation_alias=AliasChoices("CSV_DELIMITER"))
|
|
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
|
|
|
|
# Accesos directos para conveniencia
|
|
@property
|
|
def redis_url(self) -> str:
|
|
return self.processing.redis_url
|
|
|
|
@property
|
|
def openai_rpm_limit(self) -> int:
|
|
return self.processing.openai_rpm_limit
|
|
|
|
@property
|
|
def openai_tpm_limit(self) -> int:
|
|
return self.processing.openai_tpm_limit
|
|
|
|
@property
|
|
def queue_dir(self) -> str:
|
|
return self.processing.queue_dir
|
|
|
|
|
|
settings = AppSettings()
|
|
|
|
|
|
def extractor_llm_config() -> dict[str, str | int]:
|
|
"""Configuracion efectiva para llamadas de extraccion."""
|
|
api = settings.api
|
|
return {
|
|
"provider": api.extractor_provider or "custom",
|
|
"api_key": api.extractor_api_key,
|
|
"base_url": api.extractor_base_url,
|
|
"model": api.extractor_model,
|
|
"timeout_s": api.request_timeout_s,
|
|
}
|
|
|
|
|
|
def normalizer_llm_config() -> dict[str, str | int]:
|
|
"""Configuracion efectiva para llamadas de normalizacion."""
|
|
api = settings.api
|
|
return {
|
|
"provider": api.normalizer_provider or "custom",
|
|
"api_key": api.effective_normalizer_api_key,
|
|
"base_url": api.effective_normalizer_base_url,
|
|
"model": api.normalizer_model,
|
|
"timeout_s": api.effective_normalizer_timeout_s,
|
|
}
|
|
|
|
|
|
# Tipos documentales válidos por defecto para el campo especial `documento_forzate`.
|
|
# NIT se excluye explícitamente porque no debe tratarse como tipo de documento personal.
|
|
DEFAULT_ALLOWED_DOCUMENT_TYPES: Tuple[str, ...] = (
|
|
"CC",
|
|
"TI",
|
|
"CE",
|
|
"RC",
|
|
"PA",
|
|
"PPT",
|
|
"PEP",
|
|
"EPT",
|
|
"NUIP",
|
|
)
|
|
|
|
# Alias de compatibilidad.
|
|
ALLOWED_DOCUMENT_TYPES: Tuple[str, ...] = DEFAULT_ALLOWED_DOCUMENT_TYPES
|
|
|
|
|
|
def normalize_document_types(values: object) -> tuple[str, ...]:
|
|
"""Normaliza tipos documentales configurables para API/servicios."""
|
|
raw_values = values if isinstance(values, (list, tuple, set)) else []
|
|
normalized: list[str] = []
|
|
seen: set[str] = set()
|
|
for item in raw_values:
|
|
text = str(item or "").strip().upper()
|
|
if not text or text == "NIT" or text in seen:
|
|
continue
|
|
seen.add(text)
|
|
normalized.append(text)
|
|
return tuple(normalized or DEFAULT_ALLOWED_DOCUMENT_TYPES)
|
|
|
|
|
|
def get_allowed_document_types() -> tuple[str, ...]:
|
|
"""Devuelve tipos documentales válidos desde `ALLOWED_DOCUMENT_TYPES` o defaults."""
|
|
raw = os.getenv("ALLOWED_DOCUMENT_TYPES", "")
|
|
if not raw.strip():
|
|
return DEFAULT_ALLOWED_DOCUMENT_TYPES
|
|
return normalize_document_types([part.strip() for part in raw.split(",")])
|