All checks were successful
CI / lint-and-test (push) Successful in 23s
- Introduced a new configuration option `DUTY_PIN_NOTIFY` to control whether the bot re-pins the duty message when updated, providing notifications to group members. - Updated the architecture documentation to reflect the new functionality of re-pinning duty messages. - Enhanced the `.env.example` file to include the new configuration option with a description. - Added tests to verify the behavior of the new refresh pin command and its integration with the existing group duty pin functionality. - Updated internationalization messages to include help text for the new `/refresh_pin` command.
203 lines
6.5 KiB
Python
203 lines
6.5 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
|
|
duty_pin_notify: bool
|
|
|
|
@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")),
|
|
duty_pin_notify=os.getenv("DUTY_PIN_NOTIFY", "1").strip().lower()
|
|
not in ("0", "false", "no"),
|
|
)
|
|
|
|
|
|
# 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
|
|
DUTY_PIN_NOTIFY = _settings.duty_pin_notify
|
|
|
|
|
|
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."
|
|
)
|