All checks were successful
CI / lint-and-test (push) Successful in 21s
- Introduced a new `normalize_lang` function to standardize language codes across the application, ensuring consistent handling of user language preferences. - Refactored date handling utilities by adding `parse_utc_iso` and `parse_utc_iso_naive` functions for better parsing of ISO 8601 date strings, enhancing timezone awareness. - Updated various modules to utilize the new language normalization and date parsing functions, improving code clarity and maintainability. - Enhanced error handling in date validation to raise specific `DateRangeValidationError` exceptions, providing clearer feedback on validation issues. - Improved test coverage for date range validation and language normalization functionalities, ensuring robustness and reliability.
199 lines
6.3 KiB
Python
199 lines
6.3 KiB
Python
"""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.
|
|
"""
|
|
|
|
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()
|
|
|
|
# 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
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Settings:
|
|
"""Injectable settings built from environment. Used in tests or when env is overridden."""
|
|
|
|
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 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 ["*"]
|
|
)
|
|
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_lang(os.getenv("DEFAULT_LANGUAGE", "en")),
|
|
)
|
|
|
|
|
|
# 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
|
|
MINI_APP_BASE_URL = _settings.mini_app_base_url
|
|
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
|
|
|
|
|
|
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.
|
|
|
|
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.
|
|
|
|
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."
|
|
)
|