src/settings.py - master - yakobsonsa/bi_agent

src/settings.py
214 строк8 KB
Yakobsonsa
Enhance chart creation logic to support monthly data with multiple metrics and improve auto-detection of columns
16 фев 2026, 21:28
16 фев 2026, 21:28e3c99b2
"""
Centralized configuration settings using Pydantic v2 BaseSettings.
Supports environment variables and .env file loading.
"""

import os
from pathlib import Path
from typing import Optional, List, Union
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings
from functools import lru_cache
from dotenv import load_dotenv

# Load .env file explicitly (project root)
project_root = Path(__file__).parent.parent
env_path = project_root / ".env"
if env_path.exists():
    load_dotenv(env_path)
    # Debug: print loaded status
    # print(f"✅ Loaded .env from {env_path}")


class DatabaseSettings(BaseSettings):
    """Database connection and pooling configuration."""
    
    type: str = Field(default="sqlite", description="Database type: postgresql, mysql, or sqlite")
    host: str = Field(default="localhost")
    port: int = Field(default=5432)
    user: str = Field(default="admin")
    password: str = Field(default="")
    database: str = Field(default="bi_agent")
    sqlite_path: str = Field(default="./data/revenue.db")
    pool_min_size: int = Field(default=5)
    pool_max_size: int = Field(default=20)
    connection_timeout: int = Field(default=30, description="seconds")
    query_timeout: int = Field(default=30, description="seconds")
    
    class Config:
        env_prefix = "DB_"


class OpenAISettings(BaseSettings):
    """OpenAI/Qwen LLM API configuration."""
    
    api_url: str = Field(default="http://localhost:8000/v1", description="Qwen2.5 or OpenAI-compatible API URL")
    api_key: str = Field(default="", description="API key for Qwen/OpenAI")
    model: str = Field(default="qwen2.5", description="Model name (qwen2.5, gpt-4, etc)")
    max_tokens: int = Field(default=4000)
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    timeout: int = Field(default=120, description="API request timeout in seconds")
    max_retries: int = Field(default=3)
    retry_backoff_factor: float = Field(default=2.0)
    circuit_breaker_threshold: int = Field(default=5, description="Failed requests before circuit break")
    circuit_breaker_timeout: int = Field(default=60, description="seconds to wait before retry")
    
    class Config:
        env_prefix = "OPENAI_"


class TelegramSettings(BaseSettings):
    """Telegram bot configuration."""
    
    token: str = Field(default="", description="Bot token from BotFather")
    allowed_users: Union[List[int], str] = Field(default_factory=list, description="Comma-separated user IDs allowed to use bot")
    message_max_size: int = Field(default=4000, description="Max message length before splitting")
    rate_limit_requests: int = Field(default=10, description="Max requests per window")
    rate_limit_window: int = Field(default=60, description="Rate limit window in seconds")
    session_ttl_seconds: int = Field(default=3600, description="Session lifetime")
    
    @field_validator('allowed_users', mode='before')
    @classmethod
    def parse_allowed_users(cls, v):
        """Parse comma-separated string into list of integers."""
        if isinstance(v, str):
            if not v:
                return []
            return [int(x.strip()) for x in v.split(',') if x.strip()]
        return v
    
    class Config:
        env_prefix = "TELEGRAM_"


class AgentSettings(BaseSettings):
    """BI Agent behavior configuration."""
    
    max_rows: int = Field(default=1000, description="Max rows to return from queries")
    query_timeout: int = Field(default=30, description="Query execution timeout in seconds")
    generate_pdf: bool = Field(default=False)
    web_search_enabled: bool = Field(default=False)
    web_search_max_results: int = Field(default=3)
    web_search_timeout: int = Field(default=10, description="seconds")
    deep_analysis_enabled: bool = Field(default=True)
    hypothesis_count: int = Field(default=3)
    
    class Config:
        env_prefix = "AGENT_"


class VisualizerSettings(BaseSettings):
    """Data visualization configuration."""
    
    max_points_on_chart: int = Field(default=200)
    figure_width: int = Field(default=12)
    figure_height: int = Field(default=6)
    label_fontsize: int = Field(default=12)
    title_fontsize: int = Field(default=14)
    pdf_a4_width_mm: int = Field(default=210)
    pdf_a4_height_mm: int = Field(default=297)
    pdf_margin_mm: int = Field(default=10)
    dpi: int = Field(default=100)
    table_max_rows: int = Field(default=10)
    chart_colors: List[str] = Field(default_factory=lambda: ["#1f77b4", "#ff7f0e", "#2ca02c"])
    
    class Config:
        env_prefix = "VIZ_"


class LoggingSettings(BaseSettings):
    """Logging configuration."""
    
    level: str = Field(default="INFO", description="Log level: DEBUG, INFO, WARNING, ERROR, CRITICAL")
    format: str = Field(default="json", description="Log format: json or text")
    file_path: str = Field(default="./logs/agent.log")
    max_file_size: int = Field(default=10485760, description="Max log file size in bytes (10MB)")
    backup_count: int = Field(default=5)
    mask_pii: bool = Field(default=True, description="Mask PII in logs")
    
    class Config:
        env_prefix = "LOG_"


class RedisSettings(BaseSettings):
    """Redis cache configuration."""

    enabled: bool = Field(default=False)
    url: str = Field(default="redis://localhost:6379/0")
    ttl_seconds: int = Field(default=600, description="Cache TTL in seconds")
    prefix: str = Field(default="agent_bi")

    class Config:
        env_prefix = "REDIS_"


class MonitoringSettings(BaseSettings):
    """Monitoring and observability configuration."""
    
    health_check_enabled: bool = Field(default=True)
    health_check_interval: int = Field(default=60, description="seconds")
    metrics_enabled: bool = Field(default=True)
    sentry_dsn: str = Field(default="", description="Sentry error tracking DSN")
    telegram_alerts_enabled: bool = Field(default=False)
    alert_telegram_chat_id: int = Field(default=0)
    
    class Config:
        env_prefix = "MONITOR_"


class SecuritySettings(BaseSettings):
    """Security and data protection configuration."""
    
    ssl_verify: bool = Field(default=True, description="Verify SSL certificates")
    database_read_only: bool = Field(default=False, description="Force read-only database access")
    allowed_tables: List[str] = Field(default_factory=list, description="Comma-separated list of allowed tables")
    sql_validation_enabled: bool = Field(default=True)
    use_sqlparse: bool = Field(default=True, description="Use sqlparse for SQL validation")
    
    class Config:
        env_prefix = "SEC_"


class Settings(BaseSettings):
    """Main settings container combining all subsystems."""
    
    environment: str = Field(default="development", description="Environment: development, staging, production")
    debug: bool = Field(default=False)
    
    database: DatabaseSettings = Field(default_factory=DatabaseSettings)
    openai: OpenAISettings = Field(default_factory=OpenAISettings)
    telegram: TelegramSettings = Field(default_factory=TelegramSettings)
    agent: AgentSettings = Field(default_factory=AgentSettings)
    visualizer: VisualizerSettings = Field(default_factory=VisualizerSettings)
    logging: LoggingSettings = Field(default_factory=LoggingSettings)
    redis: RedisSettings = Field(default_factory=RedisSettings)
    monitoring: MonitoringSettings = Field(default_factory=MonitoringSettings)
    security: SecuritySettings = Field(default_factory=SecuritySettings)
    
    class Config:
        # Get path to .env file (project root, not src/ folder)
        project_root = Path(__file__).parent.parent  # agent_bi/
        env_file = str(project_root / ".env")
        env_file_encoding = "utf-8"
        case_sensitive = False
        extra = "ignore"  # Ignore extra fields from .env that don't match model



# Global settings instance
_settings: Optional[Settings] = None


def get_settings() -> Settings:
    """Get or create the global settings instance (lazy singleton)."""
    global _settings
    if _settings is None:
        _settings = Settings()
    return _settings


def reload_settings() -> Settings:
    """Reload settings from environment (useful for testing)."""
    global _settings
    _settings = Settings()
    return _settings