Nikolay Tatarinov
ae21883e1e
All checks were successful
CI / lint-and-test (push) Successful in 20s
- Refactored the configuration loading in `config.py` to utilize a single source of truth through the `Settings` class, improving maintainability and clarity. - Introduced the `is_admin_for_telegram_user` function in `repository.py` to centralize admin checks based on usernames and phone numbers. - Updated command handlers to use the new admin check function, ensuring consistent access control across the application. - Enhanced error handling in the `error_handler` to log exceptions when sending error replies to users, improving debugging capabilities. - Improved the handling of user phone updates in `repository.py` to ensure proper normalization and validation of phone numbers.
143 lines
4.1 KiB
Python
143 lines
4.1 KiB
Python
"""Group duty pin: current duty message text, next shift end, pin CRUD. All accept session."""
|
|
|
|
from datetime import datetime, timezone
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from duty_teller.db.repository import (
|
|
get_current_duty,
|
|
get_next_shift_end,
|
|
get_group_duty_pin,
|
|
save_group_duty_pin,
|
|
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, lang: str = "en") -> str:
|
|
"""Build the text for the pinned duty message.
|
|
|
|
Args:
|
|
duty: Duty instance or None.
|
|
user: User instance or None.
|
|
tz_name: Timezone name for display (e.g. Europe/Moscow).
|
|
lang: Language code for i18n ('ru' or 'en').
|
|
|
|
Returns:
|
|
Formatted message string; "No duty" if duty or user is None.
|
|
"""
|
|
if duty is None or user is None:
|
|
return t(lang, "duty.no_duty")
|
|
try:
|
|
tz = ZoneInfo(tz_name)
|
|
except ZoneInfoNotFoundError:
|
|
tz = ZoneInfo("Europe/Moscow")
|
|
tz_name = "Europe/Moscow"
|
|
start_dt = datetime.fromisoformat(duty.start_at.replace("Z", "+00:00"))
|
|
end_dt = datetime.fromisoformat(duty.end_at.replace("Z", "+00:00"))
|
|
start_local = start_dt.astimezone(tz)
|
|
end_local = end_dt.astimezone(tz)
|
|
offset_sec = (
|
|
start_local.utcoffset().total_seconds() if start_local.utcoffset() else 0
|
|
)
|
|
sign = "+" if offset_sec >= 0 else "-"
|
|
h, r = divmod(abs(int(offset_sec)), 3600)
|
|
m = r // 60
|
|
tz_hint = f"UTC{sign}{h:d}:{m:02d}, {tz_name}"
|
|
time_range = (
|
|
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"🕐 {label} {time_range}",
|
|
f"👤 {user.full_name}",
|
|
]
|
|
if user.phone:
|
|
lines.append(f"📞 {user.phone}")
|
|
if user.username:
|
|
lines.append(f"@{user.username}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def get_duty_message_text(session: Session, tz_name: str, lang: str = "en") -> str:
|
|
"""Get current duty from DB and return formatted message text.
|
|
|
|
Args:
|
|
session: DB session.
|
|
tz_name: Timezone name for display.
|
|
lang: Language code for i18n.
|
|
|
|
Returns:
|
|
Formatted duty message or "No duty" if none.
|
|
"""
|
|
now = datetime.now(timezone.utc)
|
|
result = get_current_duty(session, now)
|
|
if result is None:
|
|
return t(lang, "duty.no_duty")
|
|
duty, user = result
|
|
return format_duty_message(duty, user, tz_name, lang)
|
|
|
|
|
|
def get_next_shift_end_utc(session: Session) -> datetime | None:
|
|
"""Return next shift end as naive UTC datetime for job scheduling.
|
|
|
|
Args:
|
|
session: DB session.
|
|
|
|
Returns:
|
|
Next shift end (naive UTC) or None.
|
|
"""
|
|
return get_next_shift_end(session, datetime.now(timezone.utc))
|
|
|
|
|
|
def save_pin(session: Session, chat_id: int, message_id: int) -> None:
|
|
"""Save or update the pinned duty message record for a chat.
|
|
|
|
Args:
|
|
session: DB session.
|
|
chat_id: Telegram chat id.
|
|
message_id: Message id to store.
|
|
"""
|
|
save_group_duty_pin(session, chat_id, message_id)
|
|
|
|
|
|
def delete_pin(session: Session, chat_id: int) -> None:
|
|
"""Remove the pinned message record for the chat (e.g. when bot leaves).
|
|
|
|
Args:
|
|
session: DB session.
|
|
chat_id: Telegram chat id.
|
|
"""
|
|
delete_group_duty_pin(session, chat_id)
|
|
|
|
|
|
def get_message_id(session: Session, chat_id: int) -> int | None:
|
|
"""Return message_id for the pinned duty message in this chat.
|
|
|
|
Args:
|
|
session: DB session.
|
|
chat_id: Telegram chat id.
|
|
|
|
Returns:
|
|
Message id or None if no pin record.
|
|
"""
|
|
pin = get_group_duty_pin(session, chat_id)
|
|
return pin.message_id if pin else None
|
|
|
|
|
|
def get_all_pin_chat_ids(session: Session) -> list[int]:
|
|
"""Return all chat_ids that have a pinned duty message.
|
|
|
|
Used to restore update jobs on bot startup.
|
|
|
|
Args:
|
|
session: DB session.
|
|
|
|
Returns:
|
|
List of chat ids.
|
|
"""
|
|
return get_all_group_duty_pin_chat_ids(session)
|