duty-teller/duty_teller/config.py

"""Load configuration from environment. BOT_TOKEN is not validated on import; check in main/entry point."""

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

from dotenv import load_dotenv

load_dotenv()

# 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). Empty string if None/empty."""
    if not phone or not isinstance(phone, str):
        return ""
    return _PHONE_DIGITS_RE.sub("", phone.strip())


def _normalize_default_language(value: str) -> str:
    """Normalize DEFAULT_LANGUAGE from env to 'ru' or 'en'."""
    if not value:
        return "en"
    v = value.strip().lower()
    return "ru" if v.startswith("ru") else "en"


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


@dataclass(frozen=True)
class Settings:
    """Optional injectable settings built from env. Tests can override or build from env."""

    bot_token: str
    database_url: str
    mini_app_base_url: 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

    @classmethod
    def from_env(cls) -> "Settings":
        """Build Settings from current environment (same logic as module-level vars)."""
        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 ["*"]
        )
        return cls(
            bot_token=bot_token,
            database_url=os.getenv("DATABASE_URL", "sqlite:///data/duty_teller.db"),
            mini_app_base_url=os.getenv("MINI_APP_BASE_URL", "").rstrip("/"),
            http_port=int(os.getenv("HTTP_PORT", "8080")),
            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=int(os.getenv("INIT_DATA_MAX_AGE_SECONDS", "0")),
            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_default_language(
                os.getenv("DEFAULT_LANGUAGE", "en").strip()
            ),
        )


# Module-level vars: no validation on import; entry point must check BOT_TOKEN when needed.
BOT_TOKEN = os.getenv("BOT_TOKEN") or ""
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///data/duty_teller.db")
MINI_APP_BASE_URL = os.getenv("MINI_APP_BASE_URL", "").rstrip("/")
HTTP_PORT = int(os.getenv("HTTP_PORT", "8080"))

_raw_allowed = os.getenv("ALLOWED_USERNAMES", "").strip()
ALLOWED_USERNAMES = {
    s.strip().lstrip("@").lower() for s in _raw_allowed.split(",") if s.strip()
}

_raw_admin = os.getenv("ADMIN_USERNAMES", "").strip()
ADMIN_USERNAMES = {
    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", ""))

MINI_APP_SKIP_AUTH = os.getenv("MINI_APP_SKIP_AUTH", "").strip() in ("1", "true", "yes")
INIT_DATA_MAX_AGE_SECONDS = int(os.getenv("INIT_DATA_MAX_AGE_SECONDS", "0"))

_raw_cors = os.getenv("CORS_ORIGINS", "").strip()
CORS_ORIGINS = (
    [_o.strip() for _o in _raw_cors.split(",") if _o.strip()]
    if _raw_cors and _raw_cors != "*"
    else ["*"]
)

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_default_language(
    os.getenv("DEFAULT_LANGUAGE", "en").strip()
)


def is_admin(username: str) -> bool:
    """True if the given Telegram username (no @, any case) is in ADMIN_USERNAMES."""
    return (username or "").strip().lower() in ADMIN_USERNAMES


def can_access_miniapp(username: str) -> bool:
    """True if username is 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:
    """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:
    """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 entry point."""
    if not BOT_TOKEN:
        raise SystemExit(
            "BOT_TOKEN is not set. Copy .env.example to .env and set your token from @BotFather."
        )