"""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 from duty_teller.utils.dates import parse_utc_iso 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 = parse_utc_iso(duty.start_at) end_dt = parse_utc_iso(duty.end_at) 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)