settings.py

import boto3
import os
import re
import json
import yaml
from pathlib import Path
from pydantic_settings import BaseSettings

from src.modules.logger import get_logger

LOGGER = get_logger(__name__, level=os.getenv("LOG_LEVEL", "info"))
CWF = Path(__file__)
ROOT = CWF.parent.parent.parent.absolute().__str__()
PARAMS = yaml.safe_load(
    Path(os.path.join(ROOT, "config", "params.yaml")).read_text(encoding="utf-8")
)
PROMPTS = yaml.safe_load(
    Path(os.path.join(ROOT, "config", "prompts.yaml")).read_text(encoding="utf-8")
)
CHANGELOG_PATH = os.path.join(ROOT, "CHANGELOG.md")
AWS_SESSION = boto3.Session()
AWS_SSM_CLIENT = AWS_SESSION.client("ssm")


def extract_latest_version(filepath: str | None = None) -> str | None:
    """Extracts the first version number found under a '##' heading.
    Args:
        filepath (str | None): Path to the changelog file. If None, defaults to CHANGELOG.md in the root directory.
    Returns:
        str | None: The latest version number as a string, or None if not found.
    """

    filepath = filepath if filepath else CHANGELOG_PATH

    version_pattern = r"^##\s+(\d+\.\d+\.\d+)"
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            for line in f:
                match = re.search(version_pattern, line)
                if match:
                    return match.group(1)
    except FileNotFoundError:
        LOGGER.error(f"Error: {filepath} not found.")
        return None
    return None


def get_ssm_parameter(name: str | None, default: str | None = None) -> str | None:
    """
    Retrieves a specific value from AWS Systems Manager's Parameter Store.
    Args:
        name (str | None): The name of the parameter to retrieve.
        default (str | None): The default value to return if the parameter is not found.
    Returns:
        str | None: The value of the parameter, or the default value if not found.
    """

    if name is None:
        name = "none-params-in-ssm"
    try:
        response = AWS_SSM_CLIENT.get_parameter(Name=name, WithDecryption=True)
        value = response["Parameter"]["Value"]
    except AWS_SSM_CLIENT.exceptions.ParameterNotFound:
        LOGGER.warning(
            f"Parameter {name} not found in SSM, returning default: {default}"
        )
        return default

    return value


GOOGLE_SERVICE_ACCOUNT = get_ssm_parameter(
    os.getenv("CHB_AWS_SSM_GOOGLE_SERVICE_ACCOUNT")
)
if GOOGLE_SERVICE_ACCOUNT is None:
    with open(os.path.join(ROOT, ".google_service_account.json"), "r") as file:
        GOOGLE_JSON_ACCOUNT_INFO = json.load(file)
else:
    GOOGLE_JSON_ACCOUNT_INFO = json.loads(GOOGLE_SERVICE_ACCOUNT)


def mock_user_pool_id() -> str:
    client_cognito = AWS_SESSION.client("cognito-idp")
    user_pool_response = client_cognito.create_user_pool(PoolName="test_pool")
    user_pool_id = user_pool_response["UserPool"]["Id"]
    return user_pool_id


class ChatbotSettings(BaseSettings):
    """Settings for the chatbot application."""

    # api
    environment: str = os.getenv("ENVIRONMENT", os.getenv("environment", "local"))
    aws_endpoint_url: str | None = os.getenv("AWS_ENDPOINT_URL")
    aws_cognito_region: str = os.getenv("CHB_AWS_COGNITO_REGION") or os.getenv(
        "AWS_REGION"
    )
    auth_cognito_userpool_id: str = (
        mock_user_pool_id()
        if os.getenv("ENVIRONMENT", "local") in ["test", "local"]
        else os.getenv("AUTH_COGNITO_USERPOOL_ID")
    )
    google_api_key: str = get_ssm_parameter(
        name=os.getenv("CHB_AWS_SSM_GOOGLE_API_KEY"),
        default=os.getenv("CHB_AWS_GOOGLE_API_KEY"),
    )
    google_service_account: dict = GOOGLE_JSON_ACCOUNT_INFO
    cors_domains: str = os.getenv("CORS_DOMAINS", '["*"]')
    log_level: str = os.getenv("LOG_LEVEL", "info")
    max_daily_evaluations: int = int(os.getenv("CHB_MAX_DAILY_EVALUATIONS", "200"))
    expire_days: int = int(os.getenv("EXPIRE_DAYS", "90"))
    session_max_duration_days: float = float(
        os.getenv("CHB_SESSION_MAX_DURATION_DAYS", "1")
    )

    # RAG settings
    chatbot_release: str = extract_latest_version() or "---"
    embed_batch_size: int = int(os.getenv("CHB_EMBED_BATCH_SIZE", "100"))
    embed_dim: int = int(os.getenv("CHB_EMBEDDING_DIM", "768"))
    embed_model_id: str = os.getenv("CHB_EMBED_MODEL_ID", "gemini-embedding-001")
    embed_retries: int = int(os.getenv("CHB_EMBED_RETRIES", "3"))
    embed_retry_min_seconds: float = float(
        os.getenv("CHB_EMBED_RETRY_MIN_SECONDS", "1")
    )
    embed_task: str = "RETRIEVAL_QUERY"
    max_tokens: int = int(os.getenv("CHB_MODEL_MAXTOKENS", "2048"))
    model_id: str = os.getenv("CHB_MODEL_ID", "gemini-3.1-flash-lite-preview")
    provider: str = os.getenv("CHB_PROVIDER", "google")
    reranker_id: str = os.getenv("CHB_RERANKER_ID", "semantic-ranker-default-004")
    similarity_topk: int = int(os.getenv("CHB_ENGINE_SIMILARITY_TOPK", "5"))
    temperature_agent: float = 0.5
    temperature_rag: float = float(os.getenv("CHB_MODEL_TEMPERATURE", "0.3"))
    use_async: bool = os.getenv("CHB_ENGINE_USE_ASYNC", "True").lower() == "true"

    # vector index and docs params
    chunk_overlap: int = PARAMS["vector_index"]["chunk_overlap"]
    chunk_size: int = PARAMS["vector_index"]["chunk_size"]
    devportal_index_id: str = os.getenv("CHB_DEVP_INDEX_ID", "devportal-index")
    cittadino_index_id: str = os.getenv("CHB_CITTADINO_INDEX_ID", "cittadino-index")
    bucket_static_content: str = os.getenv(
        "CHB_AWS_S3_BUCKET_NAME_STATIC_CONTENT", "devportal-d-website-static-content"
    )

    # prompts
    qa_prompt_str: str = PROMPTS["qa_prompt_str"]
    react_system_str: str = PROMPTS["react_system_header_str"]
    refine_prompt_str: str = PROMPTS["refine_prompt_str"]

    # urls
    redis_url: str = os.getenv("CHB_REDIS_URL")
    website_url: str = os.getenv("CHB_WEBSITE_URL")

    # API
    query_table_prefix: str = os.getenv("CHB_QUERY_TABLE_PREFIX", "chatbot")

    # sqs
    aws_sqs_queue_monitor_name: str = os.getenv(
        "CHB_AWS_SQS_QUEUE_MONITOR_NAME", "chatbot-monitor"
    )
    aws_sqs_queue_evaluate_name: str = os.getenv(
        "CHB_AWS_SQS_QUEUE_EVALUATE_NAME", "chatbot-evaluate"
    )

    # other
    language_code: str = os.getenv("CHB_LANGUAGE_CODE_STATIC_FILES", "it")


SETTINGS = ChatbotSettings()