duty-teller/duty_teller/services/group_duty_pin_service.py

"""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.cache import duty_pin_cache
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,
    is_trusted_group,
    add_trusted_group,
    remove_trusted_group,
)
from duty_teller.i18n import t
from duty_teller.utils.dates import parse_utc_iso


def get_pin_refresh_data(
    session: Session, chat_id: int, tz_name: str, lang: str = "en"
) -> tuple[int | None, str, datetime | None]:
    """Get all data needed for pin refresh in a single DB session.

    Args:
        session: DB session.
        chat_id: Telegram chat id.
        tz_name: Timezone name for display.
        lang: Language code for i18n.

    Returns:
        (message_id, duty_message_text, next_shift_end_utc).
        message_id is None if no pin record. next_shift_end_utc is naive UTC or None.
    """
    pin = get_group_duty_pin(session, chat_id)
    message_id = pin.message_id if pin else None
    if message_id is None:
        return (None, t(lang, "duty.no_duty"), None)
    now = datetime.now(timezone.utc)
    text = get_duty_message_text(session, tz_name, lang)
    next_end = get_next_shift_end(session, now)
    return (message_id, text, next_end)


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}",
    ]
    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. Cached 90s."""
    cache_key = ("duty_message_text", tz_name, lang)
    text, found = duty_pin_cache.get(cache_key)
    if found:
        return text
    now = datetime.now(timezone.utc)
    result = get_current_duty(session, now)
    if result is None:
        text = t(lang, "duty.no_duty")
    else:
        duty, user = result
        text = format_duty_message(duty, user, tz_name, lang)
    duty_pin_cache.set(cache_key, text)
    return text


def get_next_shift_end_utc(session: Session) -> datetime | None:
    """Return next shift end as naive UTC datetime for job scheduling. Cached 90s."""
    cache_key = ("next_shift_end",)
    value, found = duty_pin_cache.get(cache_key)
    if found:
        return value
    result = get_next_shift_end(session, datetime.now(timezone.utc))
    duty_pin_cache.set(cache_key, result)
    return result


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)


def is_group_trusted(session: Session, chat_id: int) -> bool:
    """Check if the group is in the trusted list.

    Args:
        session: DB session.
        chat_id: Telegram chat id.

    Returns:
        True if the group is trusted.
    """
    return is_trusted_group(session, chat_id)


def trust_group(
    session: Session, chat_id: int, added_by_user_id: int | None = None
) -> None:
    """Add the group to the trusted list.

    Args:
        session: DB session.
        chat_id: Telegram chat id.
        added_by_user_id: Telegram user id of the admin who added the group (optional).
    """
    add_trusted_group(session, chat_id, added_by_user_id)


def untrust_group(session: Session, chat_id: int) -> None:
    """Remove the group from the trusted list.

    Args:
        session: DB session.
        chat_id: Telegram chat id.
    """
    remove_trusted_group(session, chat_id)