Add internationalization support and enhance language handling
All checks were successful
CI / lint-and-test (push) Successful in 14s

- Introduced a new i18n module for managing translations and language normalization, supporting both Russian and English.
- Updated various handlers and services to utilize the new translation functions for user-facing messages, improving user experience based on language preferences.
- Enhanced error handling and response messages to be language-aware, ensuring appropriate feedback is provided to users in their preferred language.
- Added tests for the i18n module to validate language detection and translation functionality.
- Updated the example environment file to include a default language configuration.
This commit is contained in:
2026-02-18 13:56:49 +03:00
parent be57555d4f
commit 263c2fefbd
21 changed files with 594 additions and 92 deletions

View File

@@ -15,3 +15,6 @@ ADMIN_USERNAMES=admin1,admin2
# Timezone for the pinned duty message in groups (e.g. Europe/Moscow).
# DUTY_DISPLAY_TZ=Europe/Moscow
# Default UI language when user language is unknown: en or ru (default: en).
# DEFAULT_LANGUAGE=en

View File

@@ -0,0 +1,143 @@
Metadata-Version: 2.4
Name: duty-teller
Version: 0.1.0
Summary: Telegram bot for team duty shift calendar and group reminder
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: python-telegram-bot[job-queue]<23.0,>=22.0
Requires-Dist: python-dotenv<2.0,>=1.0
Requires-Dist: fastapi<1.0,>=0.115
Requires-Dist: uvicorn[standard]<1.0,>=0.32
Requires-Dist: sqlalchemy<3.0,>=2.0
Requires-Dist: alembic<2.0,>=1.14
Requires-Dist: pydantic<3.0,>=2.0
Requires-Dist: icalendar<6.0,>=5.0
Provides-Extra: dev
Requires-Dist: pytest<9.0,>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio<1.0,>=0.24; extra == "dev"
Requires-Dist: httpx<1.0,>=0.27; extra == "dev"
# Duty Teller (Telegram Bot)
A minimal Telegram bot boilerplate using [python-telegram-bot](https://github.com/python-telegram-bot/python-telegram-bot) v22 with the `Application` API.
## Get a bot token
1. Open Telegram and search for [@BotFather](https://t.me/BotFather).
2. Send `/newbot` and follow the prompts to create a bot.
3. Copy the token BotFather gives you.
## Setup
1. **Clone and enter the project**
```bash
cd duty-teller
```
2. **Create a virtual environment (recommended)**
```bash
python -m venv venv
source venv/bin/activate # Linux/macOS
# or: venv\Scripts\activate # Windows
```
3. **Install dependencies**
```bash
pip install -r requirements.txt
```
4. **Configure the bot**
```bash
cp .env.example .env
```
Edit `.env` and set `BOT_TOKEN` to the token from BotFather.
5. **Miniapp access (calendar)**
To allow access to the calendar miniapp, set `ALLOWED_USERNAMES` to a comma-separated list of Telegram usernames (without `@`). Users in `ADMIN_USERNAMES` also have access; the admin role is reserved for future bot commands and API features. If both are empty, no one can open the calendar.
**Mini App URL:** When configuring the bot's menu button or Web App URL (e.g. in @BotFather or via `setChatMenuButton`), use the URL **with a trailing slash**, e.g. `https://your-domain.com/app/`. A redirect from `/app` to `/app/` can cause the browser to drop the fragment that Telegram sends, which breaks authorization.
**How to open:** Users must open the calendar **via the bot's menu button** (⋮ → «Календарь» or the configured label) or a **Web App inline button**. If they use «Open in browser» or a direct link, Telegram may not send user data (`tgWebAppData`), and access will be denied.
**BOT_TOKEN:** The server that serves `/api/duties` (e.g. your production host) must have in `.env` the **same** bot token as the bot from which users open the Mini App. If the token differs (e.g. test vs production bot), validation returns "hash_mismatch" and access is denied.
6. **Optional env**
- `DATABASE_URL` DB connection (default: `sqlite:///data/duty_teller.db`).
- `MINI_APP_BASE_URL` Base URL of the miniapp (for documentation / CORS).
- `MINI_APP_SKIP_AUTH` Set to `1` to allow `/api/duties` without Telegram initData (dev only; insecure).
- `INIT_DATA_MAX_AGE_SECONDS` Reject Telegram initData older than this (e.g. `86400` = 24h). `0` = disabled (default).
- `CORS_ORIGINS` Comma-separated allowed origins for CORS, or leave unset for `*`.
- `EXTERNAL_CALENDAR_ICS_URL` URL of a public ICS calendar (e.g. holidays). If set, those days are highlighted on the duty grid; users can tap «i» on a cell to see the event summary. Empty = no external calendar.
## Run
```bash
python main.py
```
Or after `pip install -e .`:
```bash
duty-teller
```
The bot runs in polling mode. Send `/start` or `/help` to your bot in Telegram to test.
## Run with Docker
Ensure `.env` exists (e.g. `cp .env.example .env`) and contains `BOT_TOKEN`.
- **Dev** (volume mount; code changes apply without rebuild):
```bash
docker compose -f docker-compose.dev.yml up --build
```
Stop with `Ctrl+C` or `docker compose -f docker-compose.dev.yml down`.
- **Prod** (no volume; runs the built image; restarts on failure):
```bash
docker compose -f docker-compose.prod.yml up -d --build
```
For production deployments you may use Docker secrets or your orchestrators env instead of a `.env` file.
**Production behind a reverse proxy:** When the app is behind nginx/Caddy etc., `request.client.host` is usually the proxy (e.g. 127.0.0.1). The "private IP" bypass (allowing requests without initData from localhost) then applies to the proxy, not the real client. Either ensure the Mini App always sends initData, or forward the real client IP (e.g. `X-Forwarded-For`) and use it for that check. See `api/app.py` `_is_private_client` for details.
## Project layout
- `main.py` Entry point: builds the `Application`, registers handlers, runs polling and FastAPI in a thread. Calls `duty_teller.config.require_bot_token()` so the app exits with a clear message if `BOT_TOKEN` is missing.
- `duty_teller/` Main package (install with `pip install -e .`). Contains:
- `config.py` Loads `BOT_TOKEN`, `DATABASE_URL`, `ALLOWED_USERNAMES`, etc. from env; no exit on import; use `require_bot_token()` in `main` when running the bot. Optional `Settings` dataclass for tests. `PROJECT_ROOT` for webapp path.
- `api/` FastAPI app (`/api/duties`, `/api/calendar-events`), `dependencies.py` (DB session, auth, date validation), static webapp mounted from `PROJECT_ROOT/webapp`.
- `db/` SQLAlchemy models, session (`session_scope`), repository, schemas.
- `handlers/` Telegram command and chat handlers; register via `register_handlers(app)`.
- `services/` Business logic (group duty pin, import); accept session from caller.
- `utils/` Shared date, user, and handover helpers.
- `importers/` Duty-schedule JSON parser.
- `alembic/` Migrations; config in `pyproject.toml` under `[tool.alembic]`; URL and metadata from `duty_teller.config` and `duty_teller.db.models.Base`. Run: `alembic -c pyproject.toml upgrade head`.
- `webapp/` Miniapp UI (calendar, duty list); served at `/app`.
- `tests/` Tests; `helpers.py` provides `make_init_data` for auth tests.
- `pyproject.toml` Installable package (`pip install -e .`).
To add commands, define async handlers in `duty_teller/handlers/commands.py` (or a new module) and register them in `duty_teller/handlers/__init__.py`.
## Импорт расписания дежурств (duty-schedule)
Команда **`/import_duty_schedule`** доступна только пользователям из `ADMIN_USERNAMES`. Импорт выполняется в два шага:
1. **Время пересменки** — бот просит указать время и при необходимости часовой пояс (например `09:00 Europe/Moscow` или `06:00 UTC`). Время приводится к UTC и используется для границ смен при создании записей.
2. **Файл JSON** — отправьте файл в формате duty-schedule (см. ниже).
Формат **duty-schedule**:
- **meta**: обязательное поле `start_date` (YYYY-MM-DD), опционально `weeks`; количество дней определяется по длине строки `duty`.
- **schedule**: массив объектов с полями:
- `name` — ФИО (строка);
- `duty` — строка с разделителем `;`: каждый элемент соответствует дню с `start_date` по порядку. Пусто или пробелы — нет события; **в**, **В**, **б**, **Б** — дежурство; **Н** — недоступен; **О** — отпуск.
При повторном импорте дежурства в том же диапазоне дат для каждого пользователя заменяются новыми.
## Tests
Install dev dependencies and run pytest:
```bash
pip install -r requirements-dev.txt
pytest
```
Tests cover `api/telegram_auth` (validate_init_data, auth_date expiry), `config` (is_admin, can_access_miniapp), and the API (date validation, 403/200 with mocked auth, plus an E2E auth test without auth mocks).

View File

@@ -0,0 +1,46 @@
README.md
pyproject.toml
duty_teller/__init__.py
duty_teller/config.py
duty_teller/run.py
duty_teller.egg-info/PKG-INFO
duty_teller.egg-info/SOURCES.txt
duty_teller.egg-info/dependency_links.txt
duty_teller.egg-info/entry_points.txt
duty_teller.egg-info/requires.txt
duty_teller.egg-info/top_level.txt
duty_teller/api/__init__.py
duty_teller/api/app.py
duty_teller/api/calendar_ics.py
duty_teller/api/dependencies.py
duty_teller/api/telegram_auth.py
duty_teller/db/__init__.py
duty_teller/db/models.py
duty_teller/db/repository.py
duty_teller/db/schemas.py
duty_teller/db/session.py
duty_teller/handlers/__init__.py
duty_teller/handlers/commands.py
duty_teller/handlers/errors.py
duty_teller/handlers/group_duty_pin.py
duty_teller/handlers/import_duty_schedule.py
duty_teller/i18n/__init__.py
duty_teller/i18n/core.py
duty_teller/i18n/messages.py
duty_teller/importers/__init__.py
duty_teller/importers/duty_schedule.py
duty_teller/services/__init__.py
duty_teller/services/group_duty_pin_service.py
duty_teller/services/import_service.py
duty_teller/utils/__init__.py
duty_teller/utils/dates.py
duty_teller/utils/handover.py
duty_teller/utils/user.py
tests/test_app.py
tests/test_config.py
tests/test_duty_schedule_parser.py
tests/test_i18n.py
tests/test_import_duty_schedule_integration.py
tests/test_repository_duty_range.py
tests/test_telegram_auth.py
tests/test_utils.py

View File

@@ -0,0 +1 @@

View File

@@ -0,0 +1,2 @@
[console_scripts]
duty-teller = duty_teller.run:main

View File

@@ -0,0 +1,13 @@
python-telegram-bot[job-queue]<23.0,>=22.0
python-dotenv<2.0,>=1.0
fastapi<1.0,>=0.115
uvicorn[standard]<1.0,>=0.32
sqlalchemy<3.0,>=2.0
alembic<2.0,>=1.14
pydantic<3.0,>=2.0
icalendar<6.0,>=5.0
[dev]
pytest<9.0,>=8.0
pytest-asyncio<1.0,>=0.24
httpx<1.0,>=0.27

View File

@@ -0,0 +1 @@
duty_teller

View File

@@ -1,6 +1,7 @@
"""FastAPI dependencies: DB session, auth, date validation."""
import logging
import re
from typing import Annotated, Generator
from fastapi import Header, HTTPException, Query, Request
@@ -11,23 +12,55 @@ from duty_teller.api.telegram_auth import validate_init_data_with_reason
from duty_teller.db.repository import get_duties
from duty_teller.db.schemas import DutyWithUser
from duty_teller.db.session import session_scope
from duty_teller.i18n import t
from duty_teller.utils.dates import validate_date_range
log = logging.getLogger(__name__)
# First language tag from Accept-Language (e.g. "ru-RU,ru;q=0.9,en;q=0.8" -> "ru")
_ACCEPT_LANG_TAG_RE = re.compile(r"^([a-zA-Z]{2,3})(?:-[a-zA-Z0-9]+)?\s*(?:;|,|$)")
def _validate_duty_dates(from_date: str, to_date: str) -> None:
def _lang_from_accept_language(header: str | None) -> str:
"""Normalize Accept-Language to 'ru' or 'en'; fallback to config.DEFAULT_LANGUAGE."""
if not header or not header.strip():
return config.DEFAULT_LANGUAGE
first = header.strip().split(",")[0].strip()
m = _ACCEPT_LANG_TAG_RE.match(first)
if not m:
return "en"
code = m.group(1).lower()
return "ru" if code.startswith("ru") else "en"
def _auth_error_detail(auth_reason: str, lang: str) -> str:
"""Return translated auth error message."""
if auth_reason == "hash_mismatch":
return t(lang, "api.auth_bad_signature")
return t(lang, "api.auth_invalid")
def _validate_duty_dates(from_date: str, to_date: str, lang: str) -> None:
"""Validate date range; raise HTTPException with translated detail."""
try:
validate_date_range(from_date, to_date)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
msg = str(e)
if "YYYY-MM-DD" in msg or "формате" in msg:
detail = t(lang, "dates.bad_format")
else:
detail = t(lang, "dates.from_after_to")
raise HTTPException(status_code=400, detail=detail) from e
def get_validated_dates(
request: Request,
from_date: str = Query(..., description="ISO date YYYY-MM-DD", alias="from"),
to_date: str = Query(..., description="ISO date YYYY-MM-DD", alias="to"),
) -> tuple[str, str]:
_validate_duty_dates(from_date, to_date)
"""Validate from/to dates; lang from Accept-Language for error messages."""
lang = _lang_from_accept_language(request.headers.get("Accept-Language"))
_validate_duty_dates(from_date, to_date, lang)
return (from_date, to_date)
@@ -45,15 +78,6 @@ def require_miniapp_username(
return get_authenticated_username(request, x_telegram_init_data)
def _auth_error_detail(auth_reason: str) -> str:
if auth_reason == "hash_mismatch":
return (
"Неверная подпись. Убедитесь, что BOT_TOKEN на сервере совпадает с токеном бота, "
"из которого открыт календарь (тот же бот, что в меню)."
)
return "Неверные данные авторизации"
def _is_private_client(client_host: str | None) -> bool:
if not client_host:
return False
@@ -81,17 +105,20 @@ def get_authenticated_username(
log.warning("allowing without initData (MINI_APP_SKIP_AUTH is set)")
return ""
log.warning("no X-Telegram-Init-Data header (client=%s)", client_host)
raise HTTPException(status_code=403, detail="Откройте календарь из Telegram")
lang = _lang_from_accept_language(request.headers.get("Accept-Language"))
raise HTTPException(status_code=403, detail=t(lang, "api.open_from_telegram"))
max_age = config.INIT_DATA_MAX_AGE_SECONDS or None
username, auth_reason = validate_init_data_with_reason(
username, auth_reason, lang = validate_init_data_with_reason(
init_data, config.BOT_TOKEN, max_age_seconds=max_age
)
if username is None:
log.warning("initData validation failed: %s", auth_reason)
raise HTTPException(status_code=403, detail=_auth_error_detail(auth_reason))
raise HTTPException(
status_code=403, detail=_auth_error_detail(auth_reason, lang)
)
if not config.can_access_miniapp(username):
log.warning("username not in allowlist: %s", username)
raise HTTPException(status_code=403, detail="Доступ запрещён")
raise HTTPException(status_code=403, detail=t(lang, "api.access_denied"))
return username

View File

@@ -16,21 +16,32 @@ def validate_init_data(
max_age_seconds: int | None = None,
) -> str | None:
"""Validate initData and return username; see validate_init_data_with_reason for failure reason."""
username, _ = validate_init_data_with_reason(init_data, bot_token, max_age_seconds)
username, _, _ = validate_init_data_with_reason(
init_data, bot_token, max_age_seconds
)
return username
def _normalize_lang(language_code: str | None) -> str:
"""Normalize to 'ru' or 'en' for i18n."""
if not language_code or not isinstance(language_code, str):
return "en"
code = language_code.strip().lower()
return "ru" if code.startswith("ru") else "en"
def validate_init_data_with_reason(
init_data: str,
bot_token: str,
max_age_seconds: int | None = None,
) -> tuple[str | None, str]:
) -> tuple[str | None, str, str]:
"""
Validate initData signature and return (username, None) or (None, reason).
Validate initData signature and return (username, reason, lang) or (None, reason, lang).
reason is one of: "ok", "empty", "no_hash", "hash_mismatch", "auth_date_expired", "no_user", "user_invalid", "no_username".
lang is from user.language_code normalized to 'ru' or 'en'; 'en' when no user.
"""
if not init_data or not bot_token:
return (None, "empty")
return (None, "empty", "en")
init_data = init_data.strip()
params = {}
for part in init_data.split("&"):
@@ -42,7 +53,7 @@ def validate_init_data_with_reason(
params[key] = value
hash_val = params.pop("hash", None)
if not hash_val:
return (None, "no_hash")
return (None, "no_hash", "en")
data_pairs = sorted(params.items())
# Data-check string: key=value with URL-decoded values (per Telegram example)
data_string = "\n".join(f"{k}={unquote(v)}" for k, v in data_pairs)
@@ -58,27 +69,28 @@ def validate_init_data_with_reason(
digestmod=hashlib.sha256,
).hexdigest()
if not hmac.compare_digest(computed.lower(), hash_val.lower()):
return (None, "hash_mismatch")
return (None, "hash_mismatch", "en")
if max_age_seconds is not None and max_age_seconds > 0:
auth_date_raw = params.get("auth_date")
if not auth_date_raw:
return (None, "auth_date_expired")
return (None, "auth_date_expired", "en")
try:
auth_date = int(float(auth_date_raw))
except (ValueError, TypeError):
return (None, "auth_date_expired")
return (None, "auth_date_expired", "en")
if time.time() - auth_date > max_age_seconds:
return (None, "auth_date_expired")
return (None, "auth_date_expired", "en")
user_raw = params.get("user")
if not user_raw:
return (None, "no_user")
return (None, "no_user", "en")
try:
user = json.loads(unquote(user_raw))
except (json.JSONDecodeError, TypeError):
return (None, "user_invalid")
return (None, "user_invalid", "en")
if not isinstance(user, dict):
return (None, "user_invalid")
return (None, "user_invalid", "en")
lang = _normalize_lang(user.get("language_code"))
username = user.get("username")
if not username or not isinstance(username, str):
return (None, "no_username")
return (username.strip().lstrip("@").lower(), "ok")
return (None, "no_username", lang)
return (username.strip().lstrip("@").lower(), "ok", lang)

View File

@@ -12,6 +12,14 @@ load_dotenv()
PROJECT_ROOT = Path(__file__).resolve().parent.parent
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"
@dataclass(frozen=True)
class Settings:
"""Optional injectable settings built from env. Tests can override or build from env."""
@@ -27,6 +35,7 @@ class Settings:
cors_origins: list[str]
external_calendar_ics_url: str
duty_display_tz: str
default_language: str
@classmethod
def from_env(cls) -> "Settings":
@@ -62,6 +71,9 @@ class Settings:
).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()
),
)
@@ -95,6 +107,9 @@ 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:

View File

@@ -8,6 +8,7 @@ from telegram.ext import CommandHandler, ContextTypes
from duty_teller.db.session import session_scope
from duty_teller.db.repository import get_or_create_user, set_user_phone
from duty_teller.i18n import get_lang, t
from duty_teller.utils.user import build_full_name
@@ -36,21 +37,23 @@ async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
await asyncio.get_running_loop().run_in_executor(None, do_get_or_create)
text = "Привет! Я бот календаря дежурств. Используй /help для списка команд."
lang = get_lang(user)
text = t(lang, "start.greeting")
await update.message.reply_text(text)
async def set_phone(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message or not update.effective_user:
return
lang = get_lang(update.effective_user)
if update.effective_chat and update.effective_chat.type != "private":
await update.message.reply_text("Команда /set_phone доступна только в личке.")
await update.message.reply_text(t(lang, "set_phone.private_only"))
return
args = context.args or []
phone = " ".join(args).strip() if args else None
telegram_user_id = update.effective_user.id
def do_set_phone() -> str:
def do_set_phone() -> str | None:
with session_scope(config.DATABASE_URL) as session:
full_name = build_full_name(
update.effective_user.first_name, update.effective_user.last_name
@@ -65,27 +68,33 @@ async def set_phone(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
)
user = set_user_phone(session, telegram_user_id, phone or None)
if user is None:
return "Ошибка сохранения."
return "error"
if phone:
return f"Телефон сохранён: {phone}"
return "Телефон очищен."
return "saved"
return "cleared"
result = await asyncio.get_running_loop().run_in_executor(None, do_set_phone)
await update.message.reply_text(result)
if result == "error":
await update.message.reply_text(t(lang, "set_phone.error"))
elif result == "saved":
await update.message.reply_text(t(lang, "set_phone.saved", phone=phone or ""))
else:
await update.message.reply_text(t(lang, "set_phone.cleared"))
async def help_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message or not update.effective_user:
return
lang = get_lang(update.effective_user)
lines = [
"Доступные команды:",
"/start — Начать",
"/help — Показать эту справку",
"/set_phone — Указать или очистить телефон для отображения в дежурстве",
"/pin_dutyВ группе: закрепить сообщение о дежурстве (нужны права админа у бота)",
t(lang, "help.title"),
t(lang, "help.start"),
t(lang, "help.help"),
t(lang, "help.set_phone"),
t(lang, "help.pin_duty"),
]
if config.is_admin(update.effective_user.username or ""):
lines.append("/import_duty_schedule — Импорт расписания дежурств (JSON)")
lines.append(t(lang, "help.import_schedule"))
await update.message.reply_text("\n".join(lines))

View File

@@ -5,6 +5,9 @@ import logging
from telegram import Update
from telegram.ext import ContextTypes
import duty_teller.config as config
from duty_teller.i18n import get_lang, t
logger = logging.getLogger(__name__)
@@ -13,4 +16,6 @@ async def error_handler(
) -> None:
logger.exception("Exception while handling an update")
if isinstance(update, Update) and update.effective_message:
await update.effective_message.reply_text("Произошла ошибка. Попробуйте позже.")
user = getattr(update, "effective_user", None)
lang = get_lang(user) if user else config.DEFAULT_LANGUAGE
await update.effective_message.reply_text(t(lang, "errors.generic"))

View File

@@ -11,6 +11,7 @@ from telegram.error import BadRequest, Forbidden
from telegram.ext import ChatMemberHandler, CommandHandler, ContextTypes
from duty_teller.db.session import session_scope
from duty_teller.i18n import get_lang, t
from duty_teller.services.group_duty_pin_service import (
get_duty_message_text,
get_next_shift_end_utc,
@@ -26,9 +27,9 @@ JOB_NAME_PREFIX = "duty_pin_"
RETRY_WHEN_NO_DUTY_MINUTES = 15
def _get_duty_message_text_sync() -> str:
def _get_duty_message_text_sync(lang: str = "en") -> str:
with session_scope(config.DATABASE_URL) as session:
return get_duty_message_text(session, config.DUTY_DISPLAY_TZ)
return get_duty_message_text(session, config.DUTY_DISPLAY_TZ, lang)
def _get_next_shift_end_sync():
@@ -98,7 +99,9 @@ async def update_group_pin(context: ContextTypes.DEFAULT_TYPE) -> None:
if message_id is None:
logger.info("No pin record for chat_id=%s, skipping update", chat_id)
return
text = await loop.run_in_executor(None, _get_duty_message_text_sync)
text = await loop.run_in_executor(
None, lambda: _get_duty_message_text_sync(config.DEFAULT_LANGUAGE)
)
try:
await context.bot.edit_message_text(
chat_id=chat_id,
@@ -133,7 +136,10 @@ async def my_chat_member_handler(
ChatMemberStatus.BANNED,
):
loop = asyncio.get_running_loop()
text = await loop.run_in_executor(None, _get_duty_message_text_sync)
lang = get_lang(update.effective_user)
text = await loop.run_in_executor(
None, lambda: _get_duty_message_text_sync(lang)
)
try:
msg = await context.bot.send_message(chat_id=chat_id, text=text)
except (BadRequest, Forbidden) as e:
@@ -154,9 +160,7 @@ async def my_chat_member_handler(
try:
await context.bot.send_message(
chat_id=chat_id,
text="Сообщение о дежурстве отправлено, но закрепить его не удалось. "
"Сделайте бота администратором с правом «Закреплять сообщения» (Pin messages), "
"затем отправьте в чат команду /pin_duty — текущее сообщение будет закреплено.",
text=t(lang, "pin_duty.could_not_pin_make_admin"),
)
except (BadRequest, Forbidden):
pass
@@ -190,19 +194,18 @@ async def restore_group_pin_jobs(application) -> None:
async def pin_duty_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message or not update.effective_chat:
if not update.message or not update.effective_chat or not update.effective_user:
return
chat = update.effective_chat
lang = get_lang(update.effective_user)
if chat.type not in ("group", "supergroup"):
await update.message.reply_text("Команда /pin_duty работает только в группах.")
await update.message.reply_text(t(lang, "pin_duty.group_only"))
return
chat_id = chat.id
loop = asyncio.get_running_loop()
message_id = await loop.run_in_executor(None, _sync_get_message_id, chat_id)
if message_id is None:
await update.message.reply_text(
"В этом чате ещё нет сообщения о дежурстве. Добавьте бота в группу — оно создастся автоматически."
)
await update.message.reply_text(t(lang, "pin_duty.no_message"))
return
try:
await context.bot.pin_chat_message(
@@ -210,12 +213,10 @@ async def pin_duty_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
message_id=message_id,
disable_notification=True,
)
await update.message.reply_text("Сообщение о дежурстве закреплено.")
await update.message.reply_text(t(lang, "pin_duty.pinned"))
except (BadRequest, Forbidden) as e:
logger.warning("pin_duty failed chat_id=%s: %s", chat_id, e)
await update.message.reply_text(
"Не удалось закрепить. Убедитесь, что бот — администратор с правом «Закреплять сообщения»."
)
await update.message.reply_text(t(lang, "pin_duty.failed"))
group_duty_pin_handler = ChatMemberHandler(

View File

@@ -7,6 +7,7 @@ from telegram import Update
from telegram.ext import CommandHandler, ContextTypes, MessageHandler, filters
from duty_teller.db.session import session_scope
from duty_teller.i18n import get_lang, t
from duty_teller.importers.duty_schedule import (
DutyScheduleParseError,
parse_duty_schedule,
@@ -20,14 +21,12 @@ async def import_duty_schedule_cmd(
) -> None:
if not update.message or not update.effective_user:
return
lang = get_lang(update.effective_user)
if not config.is_admin(update.effective_user.username or ""):
await update.message.reply_text("Доступ только для администраторов.")
await update.message.reply_text(t(lang, "import.admin_only"))
return
context.user_data["awaiting_handover_time"] = True
await update.message.reply_text(
"Укажите время пересменки в формате ЧЧ:ММ и часовой пояс, "
"например 09:00 Europe/Moscow или 06:00 UTC."
)
await update.message.reply_text(t(lang, "import.handover_format"))
async def handle_handover_time_text(
@@ -39,18 +38,17 @@ async def handle_handover_time_text(
return
if not config.is_admin(update.effective_user.username or ""):
return
lang = get_lang(update.effective_user)
text = update.message.text.strip()
parsed = parse_handover_time(text)
if parsed is None:
await update.message.reply_text(
"Не удалось разобрать время. Укажите, например: 09:00 Europe/Moscow"
)
await update.message.reply_text(t(lang, "import.parse_time_error"))
return
hour_utc, minute_utc = parsed
context.user_data["handover_utc_time"] = (hour_utc, minute_utc)
context.user_data["awaiting_handover_time"] = False
context.user_data["awaiting_duty_schedule_file"] = True
await update.message.reply_text("Отправьте файл в формате duty-schedule (JSON).")
await update.message.reply_text(t(lang, "import.send_json"))
async def handle_duty_schedule_document(
@@ -60,11 +58,12 @@ async def handle_duty_schedule_document(
return
if not context.user_data.get("awaiting_duty_schedule_file"):
return
lang = get_lang(update.effective_user)
handover = context.user_data.get("handover_utc_time")
if not handover or not config.is_admin(update.effective_user.username or ""):
return
if not (update.message.document.file_name or "").lower().endswith(".json"):
await update.message.reply_text("Нужен файл с расширением .json")
await update.message.reply_text(t(lang, "import.need_json"))
return
hour_utc, minute_utc = handover
@@ -77,7 +76,7 @@ async def handle_duty_schedule_document(
except DutyScheduleParseError as e:
context.user_data.pop("awaiting_duty_schedule_file", None)
context.user_data.pop("handover_utc_time", None)
await update.message.reply_text(f"Ошибка разбора файла: {e}")
await update.message.reply_text(t(lang, "import.parse_error", error=str(e)))
return
def run_import_with_scope():
@@ -90,16 +89,29 @@ async def handle_duty_schedule_document(
None, run_import_with_scope
)
except Exception as e:
await update.message.reply_text(f"Ошибка импорта: {e}")
await update.message.reply_text(t(lang, "import.import_error", error=str(e)))
else:
total = num_duty + num_unavailable + num_vacation
parts = [f"{num_users} пользователей", f"{num_duty} дежурств"]
if num_unavailable:
parts.append(f"{num_unavailable} недоступностей")
if num_vacation:
parts.append(f"{num_vacation} отпусков")
unavailable_suffix = (
t(lang, "import.done_unavailable", count=str(num_unavailable))
if num_unavailable
else ""
)
vacation_suffix = (
t(lang, "import.done_vacation", count=str(num_vacation))
if num_vacation
else ""
)
await update.message.reply_text(
"Импорт выполнен: " + ", ".join(parts) + f" (всего {total} событий)."
t(
lang,
"import.done",
users=str(num_users),
duties=str(num_duty),
unavailable=unavailable_suffix,
vacation=vacation_suffix,
total=str(total),
)
)
finally:
context.user_data.pop("awaiting_duty_schedule_file", None)

View File

@@ -0,0 +1,6 @@
"""Internationalization: RU/EN by Telegram language_code. Normalize to 'ru' or 'en'."""
from duty_teller.i18n.messages import MESSAGES
from duty_teller.i18n.core import get_lang, t
__all__ = ["MESSAGES", "get_lang", "t"]

37
duty_teller/i18n/core.py Normal file
View File

@@ -0,0 +1,37 @@
"""get_lang and t(): language from Telegram user, translate by key with fallback to en."""
from typing import TYPE_CHECKING
import duty_teller.config as config
from duty_teller.i18n.messages import MESSAGES
if TYPE_CHECKING:
from telegram import User
def get_lang(user: "User | None") -> str:
"""
Normalize Telegram user language to 'ru' or 'en'.
If user has language_code starting with 'ru' (e.g. ru, ru-RU) return 'ru', else 'en'.
When user is None or has no language_code, return config.DEFAULT_LANGUAGE.
"""
if user is None or not getattr(user, "language_code", None):
return config.DEFAULT_LANGUAGE
code = (user.language_code or "").strip().lower()
return "ru" if code.startswith("ru") else "en"
def t(lang: str, key: str, **kwargs: str) -> str:
"""
Return translated string for lang and key; substitute kwargs into placeholders like {phone}.
Fallback to 'en' if key missing for lang.
"""
lang = "ru" if lang == "ru" else "en"
messages = MESSAGES.get(lang) or MESSAGES["en"]
template = messages.get(key)
if template is None:
template = MESSAGES["en"].get(key, key)
try:
return template.format(**kwargs)
except KeyError:
return template

View File

@@ -0,0 +1,82 @@
"""Translation dictionaries: MESSAGES[lang][key]. Keys: dotted, e.g. start.greeting."""
MESSAGES: dict[str, dict[str, str]] = {
"en": {
"start.greeting": "Hi! I'm the duty calendar bot. Use /help for the command list.",
"set_phone.private_only": "The /set_phone command is only available in private chat.",
"set_phone.saved": "Phone saved: {phone}",
"set_phone.cleared": "Phone cleared.",
"set_phone.error": "Error saving.",
"help.title": "Available commands:",
"help.start": "/start — Start",
"help.help": "/help — Show this help",
"help.set_phone": "/set_phone — Set or clear phone for duty display",
"help.pin_duty": "/pin_duty — In a group: pin the duty message (bot needs admin with Pin messages)",
"help.import_schedule": "/import_duty_schedule — Import duty schedule (JSON)",
"errors.generic": "An error occurred. Please try again later.",
"pin_duty.group_only": "The /pin_duty command works only in groups.",
"pin_duty.no_message": "There is no duty message in this chat yet. Add the bot to the group — it will create one automatically.",
"pin_duty.pinned": "Duty message pinned.",
"pin_duty.failed": "Could not pin. Make sure the bot is an administrator with «Pin messages» permission.",
"pin_duty.could_not_pin_make_admin": "Duty message was sent but could not be pinned. Make the bot an administrator with «Pin messages» permission, then send /pin_duty in the chat — the current message will be pinned.",
"duty.no_duty": "No duty at the moment.",
"duty.label": "Duty:",
"import.admin_only": "Access for administrators only.",
"import.handover_format": "Enter handover time as HH:MM and timezone, e.g. 09:00 Europe/Moscow or 06:00 UTC.",
"import.parse_time_error": "Could not parse time. Enter e.g.: 09:00 Europe/Moscow",
"import.send_json": "Send the duty-schedule file (JSON).",
"import.need_json": "File must have .json extension.",
"import.parse_error": "File parse error: {error}",
"import.import_error": "Import error: {error}",
"import.done": "Import done: {users} users, {duties} duties{unavailable}{vacation} ({total} events total).",
"import.done_unavailable": ", {count} unavailable",
"import.done_vacation": ", {count} vacation",
"api.open_from_telegram": "Open the calendar from Telegram",
"api.auth_bad_signature": "Invalid signature. Ensure BOT_TOKEN on the server matches the bot from which the calendar was opened (same bot as in the menu).",
"api.auth_invalid": "Invalid auth data",
"api.access_denied": "Access denied",
"dates.bad_format": "Parameters from and to must be in YYYY-MM-DD format",
"dates.from_after_to": "from date must not be after to",
},
"ru": {
"start.greeting": "Привет! Я бот календаря дежурств. Используй /help для списка команд.",
"set_phone.private_only": "Команда /set_phone доступна только в личке.",
"set_phone.saved": "Телефон сохранён: {phone}",
"set_phone.cleared": "Телефон очищен.",
"set_phone.error": "Ошибка сохранения.",
"help.title": "Доступные команды:",
"help.start": "/start — Начать",
"help.help": "/help — Показать эту справку",
"help.set_phone": "/set_phone — Указать или очистить телефон для отображения в дежурстве",
"help.pin_duty": "/pin_duty — В группе: закрепить сообщение о дежурстве (нужны права админа у бота)",
"help.import_schedule": "/import_duty_schedule — Импорт расписания дежурств (JSON)",
"errors.generic": "Произошла ошибка. Попробуйте позже.",
"pin_duty.group_only": "Команда /pin_duty работает только в группах.",
"pin_duty.no_message": "В этом чате ещё нет сообщения о дежурстве. Добавьте бота в группу — оно создастся автоматически.",
"pin_duty.pinned": "Сообщение о дежурстве закреплено.",
"pin_duty.failed": "Не удалось закрепить. Убедитесь, что бот — администратор с правом «Закреплять сообщения».",
"pin_duty.could_not_pin_make_admin": "Сообщение о дежурстве отправлено, но закрепить его не удалось. "
"Сделайте бота администратором с правом «Закреплять сообщения» (Pin messages), "
"затем отправьте в чат команду /pin_duty — текущее сообщение будет закреплено.",
"duty.no_duty": "Сейчас дежурства нет.",
"duty.label": "Дежурство:",
"import.admin_only": "Доступ только для администраторов.",
"import.handover_format": "Укажите время пересменки в формате ЧЧ:ММ и часовой пояс, "
"например 09:00 Europe/Moscow или 06:00 UTC.",
"import.parse_time_error": "Не удалось разобрать время. Укажите, например: 09:00 Europe/Moscow",
"import.send_json": "Отправьте файл в формате duty-schedule (JSON).",
"import.need_json": "Нужен файл с расширением .json",
"import.parse_error": "Ошибка разбора файла: {error}",
"import.import_error": "Ошибка импорта: {error}",
"import.done": "Импорт выполнен: {users} пользователей, {duties} дежурств{unavailable}{vacation} (всего {total} событий).",
"import.done_unavailable": ", {count} недоступностей",
"import.done_vacation": ", {count} отпусков",
"api.open_from_telegram": "Откройте календарь из Telegram",
"api.auth_bad_signature": "Неверная подпись. Убедитесь, что BOT_TOKEN на сервере совпадает с токеном бота, "
"из которого открыт календарь (тот же бот, что в меню).",
"api.auth_invalid": "Неверные данные авторизации",
"api.access_denied": "Доступ запрещён",
"dates.bad_format": "Параметры from и to должны быть в формате YYYY-MM-DD",
"dates.from_after_to": "Дата from не должна быть позже to",
},
}

View File

@@ -28,7 +28,7 @@ def _set_default_menu_button_webapp() -> None:
payload = {
"menu_button": {
"type": "web_app",
"text": "Календарь",
"text": "Calendar",
"web_app": {"url": menu_url},
}
}

View File

@@ -13,12 +13,13 @@ from duty_teller.db.repository import (
delete_group_duty_pin,
get_all_group_duty_pin_chat_ids,
)
from duty_teller.i18n import t
def format_duty_message(duty, user, tz_name: str) -> str:
def format_duty_message(duty, user, tz_name: str, lang: str = "en") -> str:
"""Build the text for the pinned message. duty, user may be None."""
if duty is None or user is None:
return "Сейчас дежурства нет."
return t(lang, "duty.no_duty")
try:
tz = ZoneInfo(tz_name)
except Exception:
@@ -39,8 +40,9 @@ def format_duty_message(duty, user, tz_name: str) -> str:
f"{start_local.strftime('%d.%m.%Y %H:%M')}"
f"{end_local.strftime('%d.%m.%Y %H:%M')} ({tz_hint})"
)
label = t(lang, "duty.label")
lines = [
f"🕐 Дежурство: {time_range}",
f"🕐 {label} {time_range}",
f"👤 {user.full_name}",
]
if user.phone:
@@ -50,14 +52,14 @@ def format_duty_message(duty, user, tz_name: str) -> str:
return "\n".join(lines)
def get_duty_message_text(session: Session, tz_name: str) -> str:
def get_duty_message_text(session: Session, tz_name: str, lang: str = "en") -> str:
"""Get current duty from DB and return formatted message."""
now = datetime.now(timezone.utc)
result = get_current_duty(session, now)
if result is None:
return "Сейчас дежурства нет."
return t(lang, "duty.no_duty")
duty, user = result
return format_duty_message(duty, user, tz_name)
return format_duty_message(duty, user, tz_name, lang)
def get_next_shift_end_utc(session: Session) -> datetime | None:

View File

@@ -19,13 +19,15 @@ def client():
def test_duties_invalid_date_format(client):
r = client.get("/api/duties", params={"from": "2025-01-01", "to": "invalid"})
assert r.status_code == 400
assert "YYYY-MM-DD" in r.json()["detail"]
detail = r.json()["detail"]
assert "from" in detail.lower() and "to" in detail.lower()
def test_duties_from_after_to(client):
r = client.get("/api/duties", params={"from": "2025-02-01", "to": "2025-01-01"})
assert r.status_code == 400
assert "from" in r.json()["detail"].lower() or "позже" in r.json()["detail"]
detail = r.json()["detail"].lower()
assert "from" in detail or "to" in detail or "after" in detail or "позже" in detail
@patch("duty_teller.api.dependencies._is_private_client")
@@ -54,7 +56,7 @@ def test_duties_200_when_skip_auth(mock_fetch, client):
@patch("duty_teller.api.dependencies.validate_init_data_with_reason")
def test_duties_403_when_init_data_invalid(mock_validate, client):
mock_validate.return_value = (None, "hash_mismatch")
mock_validate.return_value = (None, "hash_mismatch", "en")
r = client.get(
"/api/duties",
params={"from": "2025-01-01", "to": "2025-01-31"},
@@ -62,13 +64,18 @@ def test_duties_403_when_init_data_invalid(mock_validate, client):
)
assert r.status_code == 403
detail = r.json()["detail"]
assert "авторизации" in detail or "Неверные" in detail or "Неверная" in detail
assert (
"signature" in detail.lower()
or "авторизации" in detail
or "Неверные" in detail
or "Неверная" in detail
)
@patch("duty_teller.api.dependencies.validate_init_data_with_reason")
@patch("duty_teller.api.dependencies.config.can_access_miniapp")
def test_duties_403_when_username_not_allowed(mock_can_access, mock_validate, client):
mock_validate.return_value = ("someuser", "ok")
mock_validate.return_value = ("someuser", "ok", "en")
mock_can_access.return_value = False
with patch("duty_teller.api.app.fetch_duties_response") as mock_fetch:
r = client.get(
@@ -77,14 +84,16 @@ def test_duties_403_when_username_not_allowed(mock_can_access, mock_validate, cl
headers={"X-Telegram-Init-Data": "user=xxx&hash=yyy"},
)
assert r.status_code == 403
assert "Доступ запрещён" in r.json()["detail"]
assert (
"Access denied" in r.json()["detail"] or "Доступ запрещён" in r.json()["detail"]
)
mock_fetch.assert_not_called()
@patch("duty_teller.api.dependencies.validate_init_data_with_reason")
@patch("duty_teller.api.dependencies.config.can_access_miniapp")
def test_duties_200_with_allowed_user(mock_can_access, mock_validate, client):
mock_validate.return_value = ("alloweduser", "ok")
mock_validate.return_value = ("alloweduser", "ok", "en")
mock_can_access.return_value = True
with patch("duty_teller.api.app.fetch_duties_response") as mock_fetch:
mock_fetch.return_value = [

76
tests/test_i18n.py Normal file
View File

@@ -0,0 +1,76 @@
"""Unit tests for duty_teller.i18n: get_lang, t, fallback to en."""
from unittest.mock import MagicMock
from duty_teller.i18n import get_lang, t
def test_get_lang_none_returns_en():
assert get_lang(None) == "en"
def test_get_lang_ru_returns_ru():
user = MagicMock()
user.language_code = "ru"
assert get_lang(user) == "ru"
def test_get_lang_ru_ru_returns_ru():
user = MagicMock()
user.language_code = "ru-RU"
assert get_lang(user) == "ru"
def test_get_lang_en_returns_en():
user = MagicMock()
user.language_code = "en"
assert get_lang(user) == "en"
def test_get_lang_uk_returns_en():
user = MagicMock()
user.language_code = "uk"
assert get_lang(user) == "en"
def test_get_lang_empty_returns_en():
user = MagicMock()
user.language_code = ""
assert get_lang(user) == "en"
def test_get_lang_missing_attr_returns_en():
user = MagicMock(spec=[]) # no language_code
assert get_lang(user) == "en"
def test_t_en_start_greeting():
result = t("en", "start.greeting")
assert "help" in result.lower()
assert "bot" in result.lower()
def test_t_ru_start_greeting():
result = t("ru", "start.greeting")
assert "Привет" in result or "бот" in result or "календар" in result
def test_t_fallback_to_en_when_key_missing_for_ru():
"""When key exists in en but not in ru, fallback to en."""
result = t("ru", "start.greeting")
assert len(result) > 0
# start.greeting exists in both; use a key that might be en-only for fallback
result_any = t("ru", "errors.generic")
assert "error" in result_any.lower() or "ошибк" in result_any.lower()
def test_t_placeholder_substitution():
assert t("en", "set_phone.saved", phone="+7 999") == "Phone saved: +7 999"
assert "999" in t("ru", "set_phone.saved", phone="+7 999")
def test_t_unknown_lang_normalized_to_en():
assert "help" in t("de", "start.greeting").lower() or "Hi" in t(
"de", "start.greeting"
)