duty-teller/duty_teller/config.py

"""Load configuration from environment (e.g. .env via python-dotenv).

BOT_TOKEN is not validated on import; call require_bot_token() in the entry point
when running the bot. Numeric env vars (HTTP_PORT, INIT_DATA_MAX_AGE_SECONDS) use
safe parsing with defaults on invalid values.
"""

import logging
import os
import re
from dataclasses import dataclass
from pathlib import Path

from dotenv import load_dotenv

from duty_teller.i18n.lang import normalize_lang

load_dotenv()

logger = logging.getLogger(__name__)

# Valid port range for HTTP_PORT.
HTTP_PORT_MIN, HTTP_PORT_MAX = 1, 65535

# Project root (parent of duty_teller package). Used for webapp path, etc.
PROJECT_ROOT = Path(__file__).resolve().parent.parent

# Only digits for phone comparison (same when saving and when checking allowlist).
_PHONE_DIGITS_RE = re.compile(r"\D")


def normalize_phone(phone: str | None) -> str:
    """Return phone as digits only (spaces, +, parentheses, dashes removed).

    Args:
        phone: Raw phone string or None.

    Returns:
        Digits-only string, or empty string if None or empty.
    """
    if not phone or not isinstance(phone, str):
        return ""
    return _PHONE_DIGITS_RE.sub("", phone.strip())


def _parse_phone_list(raw: str) -> set[str]:
    """Parse comma-separated phones into set of normalized (digits-only) strings."""
    result = set()
    for s in (raw or "").strip().split(","):
        normalized = normalize_phone(s)
        if normalized:
            result.add(normalized)
    return result


def _normalize_log_level(raw: str) -> str:
    """Return a valid log level name (DEBUG, INFO, WARNING, ERROR); default INFO."""
    level = (raw or "").strip().upper()
    if level in ("DEBUG", "INFO", "WARNING", "ERROR"):
        return level
    return "INFO"


def _parse_int_env(
    name: str, default: int, min_val: int | None = None, max_val: int | None = None
) -> int:
    """Parse an integer from os.environ; use default on invalid or out-of-range. Log on fallback."""
    raw = os.getenv(name)
    if raw is None or raw == "":
        return default
    try:
        value = int(raw.strip())
    except ValueError:
        logger.warning(
            "Invalid %s=%r (expected integer); using default %s",
            name,
            raw,
            default,
        )
        return default
    if min_val is not None and value < min_val:
        logger.warning(
            "%s=%s is below minimum %s; using %s", name, value, min_val, min_val
        )
        return min_val
    if max_val is not None and value > max_val:
        logger.warning(
            "%s=%s is above maximum %s; using %s", name, value, max_val, max_val
        )
        return max_val
    return value


def _validate_database_url(url: str) -> bool:
    """Return True if URL looks like a supported SQLAlchemy URL (sqlite or postgres)."""
    if not url or not isinstance(url, str):
        return False
    u = url.strip().split("?", 1)[0].lower()
    return (
        u.startswith("sqlite://")
        or u.startswith("postgresql://")
        or u.startswith("postgres://")
    )


@dataclass(frozen=True)
class Settings:
    """Injectable settings built from environment. Used in tests or when env is overridden."""

    bot_token: str
    database_url: str
    bot_username: str
    mini_app_base_url: str
    mini_app_short_name: str
    http_host: str
    http_port: int
    allowed_usernames: set[str]
    admin_usernames: set[str]
    allowed_phones: set[str]
    admin_phones: set[str]
    mini_app_skip_auth: bool
    init_data_max_age_seconds: int
    cors_origins: list[str]
    external_calendar_ics_url: str
    duty_display_tz: str
    default_language: str
    duty_pin_notify: bool
    log_level: str

    @classmethod
    def from_env(cls) -> "Settings":
        """Build Settings from current environment (same logic as module-level variables).

        Returns:
            Settings instance with all fields populated from env.
        """
        bot_token = os.getenv("BOT_TOKEN") or ""
        raw_allowed = os.getenv("ALLOWED_USERNAMES", "").strip()
        allowed = {
            s.strip().lstrip("@").lower() for s in raw_allowed.split(",") if s.strip()
        }
        raw_admin = os.getenv("ADMIN_USERNAMES", "").strip()
        admin = {
            s.strip().lstrip("@").lower() for s in raw_admin.split(",") if s.strip()
        }
        allowed_phones = _parse_phone_list(os.getenv("ALLOWED_PHONES", ""))
        admin_phones = _parse_phone_list(os.getenv("ADMIN_PHONES", ""))
        raw_cors = os.getenv("CORS_ORIGINS", "").strip()
        cors = (
            [_o.strip() for _o in raw_cors.split(",") if _o.strip()]
            if raw_cors and raw_cors != "*"
            else ["*"]
        )
        raw_host = (os.getenv("HTTP_HOST") or "127.0.0.1").strip()
        http_host = raw_host if raw_host else "127.0.0.1"
        bot_username = (os.getenv("BOT_USERNAME", "") or "").strip().lstrip("@").lower()
        database_url = os.getenv("DATABASE_URL", "sqlite:///data/duty_teller.db")
        if not _validate_database_url(database_url):
            logger.warning(
                "DATABASE_URL does not look like a supported URL (sqlite:// or postgresql://); "
                "DB connection may fail."
            )
        http_port = _parse_int_env(
            "HTTP_PORT", 8080, min_val=HTTP_PORT_MIN, max_val=HTTP_PORT_MAX
        )
        init_data_max_age = _parse_int_env("INIT_DATA_MAX_AGE_SECONDS", 0, min_val=0)
        return cls(
            bot_token=bot_token,
            database_url=database_url,
            bot_username=bot_username,
            mini_app_base_url=os.getenv("MINI_APP_BASE_URL", "").rstrip("/"),
            mini_app_short_name=(os.getenv("MINI_APP_SHORT_NAME", "") or "").strip().strip("/"),
            http_host=http_host,
            http_port=http_port,
            allowed_usernames=allowed,
            admin_usernames=admin,
            allowed_phones=allowed_phones,
            admin_phones=admin_phones,
            mini_app_skip_auth=os.getenv("MINI_APP_SKIP_AUTH", "").strip()
            in ("1", "true", "yes"),
            init_data_max_age_seconds=init_data_max_age,
            cors_origins=cors,
            external_calendar_ics_url=os.getenv(
                "EXTERNAL_CALENDAR_ICS_URL", ""
            ).strip(),
            duty_display_tz=os.getenv("DUTY_DISPLAY_TZ", "Europe/Moscow").strip()
            or "Europe/Moscow",
            default_language=normalize_lang(os.getenv("DEFAULT_LANGUAGE", "en")),
            duty_pin_notify=os.getenv("DUTY_PIN_NOTIFY", "1").strip().lower()
            not in ("0", "false", "no"),
            log_level=_normalize_log_level(os.getenv("LOG_LEVEL", "INFO")),
        )


# Single source of truth: load once at import; entry point must check BOT_TOKEN when needed.
_settings = Settings.from_env()

BOT_TOKEN = _settings.bot_token
DATABASE_URL = _settings.database_url
BOT_USERNAME = _settings.bot_username
MINI_APP_BASE_URL = _settings.mini_app_base_url
MINI_APP_SHORT_NAME = _settings.mini_app_short_name
HTTP_HOST = _settings.http_host
HTTP_PORT = _settings.http_port
ALLOWED_USERNAMES = _settings.allowed_usernames
ADMIN_USERNAMES = _settings.admin_usernames
ALLOWED_PHONES = _settings.allowed_phones
ADMIN_PHONES = _settings.admin_phones
MINI_APP_SKIP_AUTH = _settings.mini_app_skip_auth
INIT_DATA_MAX_AGE_SECONDS = _settings.init_data_max_age_seconds
CORS_ORIGINS = _settings.cors_origins
EXTERNAL_CALENDAR_ICS_URL = _settings.external_calendar_ics_url
DUTY_DISPLAY_TZ = _settings.duty_display_tz
DEFAULT_LANGUAGE = _settings.default_language
DUTY_PIN_NOTIFY = _settings.duty_pin_notify
LOG_LEVEL = getattr(logging, _settings.log_level.upper(), logging.INFO)
LOG_LEVEL_STR = _settings.log_level


def is_admin(username: str) -> bool:
    """Check if Telegram username is in ADMIN_USERNAMES.

    Args:
        username: Telegram username (with or without @; case-insensitive).

    Returns:
        True if in ADMIN_USERNAMES.
    """
    return (username or "").strip().lower() in ADMIN_USERNAMES


def can_access_miniapp(username: str) -> bool:
    """Check if username is allowed to open the calendar Miniapp (env allowlist only).

    Legacy: Miniapp access in production is determined by repository.can_access_miniapp_for_telegram_user
    (DB roles + fallback to ADMIN_*). This function is kept for tests and backward compatibility.

    Args:
        username: Telegram username (with or without @; case-insensitive).

    Returns:
        True if in ALLOWED_USERNAMES or ADMIN_USERNAMES.
    """
    u = (username or "").strip().lower()
    return u in ALLOWED_USERNAMES or u in ADMIN_USERNAMES


def can_access_miniapp_by_phone(phone: str | None) -> bool:
    """Check if phone (set via /set_phone) is allowed to open the Miniapp (env allowlist only).

    Legacy: Miniapp access in production is determined by repository.can_access_miniapp_for_telegram_user
    (DB roles + fallback to ADMIN_*). This function is kept for tests and backward compatibility.

    Args:
        phone: Raw phone string or None.

    Returns:
        True if normalized phone is in ALLOWED_PHONES or ADMIN_PHONES.
    """
    normalized = normalize_phone(phone)
    if not normalized:
        return False
    return normalized in ALLOWED_PHONES or normalized in ADMIN_PHONES


def is_admin_by_phone(phone: str | None) -> bool:
    """Check if phone is in ADMIN_PHONES.

    Args:
        phone: Raw phone string or None.

    Returns:
        True if normalized phone is in ADMIN_PHONES.
    """
    normalized = normalize_phone(phone)
    return bool(normalized and normalized in ADMIN_PHONES)


def require_bot_token() -> None:
    """Raise SystemExit with a clear message if BOT_TOKEN is not set.

    Call from the application entry point (e.g. main.py or duty_teller.run) so the
    process exits with a helpful message instead of failing later.

    Raises:
        SystemExit: If BOT_TOKEN is empty.
    """
    if not BOT_TOKEN:
        raise SystemExit(
            "BOT_TOKEN is not set. Copy .env.example to .env and set your token from @BotFather."
        )