feat: add trusted groups functionality for duty information

- Introduced a new `trusted_groups` table to store groups authorized to receive duty information.
- Implemented functions to add, remove, and check trusted groups in the database.
- Enhanced command handlers to manage trusted groups, including `/trust_group` and `/untrust_group` commands for admin users.
- Updated internationalization messages to support new commands and group status notifications.
- Added unit tests for trusted groups repository functions to ensure correct behavior and data integrity.
This commit is contained in:
2026-03-02 13:07:13 +03:00
parent 322b553b80
commit f8aceabab5
10 changed files with 837 additions and 112 deletions

View File

@@ -11,6 +11,7 @@ from duty_teller.db.models import (
User,
Duty,
GroupDutyPin,
TrustedGroup,
CalendarSubscriptionToken,
Role,
)
@@ -593,6 +594,71 @@ def get_all_group_duty_pin_chat_ids(session: Session) -> list[int]:
return [r[0] for r in rows]
def is_trusted_group(session: Session, chat_id: int) -> bool:
"""Check if the chat is in the trusted groups list.
Args:
session: DB session.
chat_id: Telegram chat id.
Returns:
True if the group is trusted.
"""
return (
session.query(TrustedGroup).filter(TrustedGroup.chat_id == chat_id).first()
is not None
)
def add_trusted_group(
session: Session, chat_id: int, added_by_user_id: int | None = None
) -> TrustedGroup:
"""Add a 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).
Returns:
Created TrustedGroup instance.
"""
now_iso = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
record = TrustedGroup(
chat_id=chat_id,
added_by_user_id=added_by_user_id,
added_at=now_iso,
)
session.add(record)
session.commit()
session.refresh(record)
return record
def remove_trusted_group(session: Session, chat_id: int) -> None:
"""Remove a group from the trusted list.
Args:
session: DB session.
chat_id: Telegram chat id.
"""
session.query(TrustedGroup).filter(TrustedGroup.chat_id == chat_id).delete()
session.commit()
def get_all_trusted_group_ids(session: Session) -> list[int]:
"""Return all chat_ids that are trusted.
Args:
session: DB session.
Returns:
List of trusted chat ids.
"""
rows = session.query(TrustedGroup.chat_id).all()
return [r[0] for r in rows]
def set_user_phone(
session: Session, telegram_user_id: int, phone: str | None
) -> User | None: