131 lines
4.8 KiB
Python
131 lines
4.8 KiB
Python
"""Приглашения в группу: создание (любым участником), список, принятие, отклонение.
|
||
|
||
Хранятся только ожидающие приглашения; ответ удаляет строку (accept → создаётся
|
||
GroupMember). Лимит участников — общий с membership_service.MAX_GROUP_SIZE.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from sqlmodel import Session, select
|
||
|
||
from app.core.errors import ConflictError, NotFoundError, ValidationError
|
||
from app.core.timeutil import iso_utc
|
||
from app.models import Group, GroupInvitation, GroupMember, User
|
||
from app.services.membership_service import MAX_GROUP_SIZE
|
||
|
||
|
||
def _member_exists(session: Session, group_id: int, user_id: int) -> bool:
|
||
return (
|
||
session.exec(
|
||
select(GroupMember.id).where(
|
||
GroupMember.group_id == group_id, GroupMember.user_id == user_id
|
||
)
|
||
).first()
|
||
is not None
|
||
)
|
||
|
||
|
||
def _counts(session: Session, group_id: int) -> tuple[int, int]:
|
||
members = len(
|
||
session.exec(select(GroupMember.id).where(GroupMember.group_id == group_id)).all()
|
||
)
|
||
pending = len(
|
||
session.exec(
|
||
select(GroupInvitation.id).where(GroupInvitation.group_id == group_id)
|
||
).all()
|
||
)
|
||
return members, pending
|
||
|
||
|
||
def create_invitation(
|
||
session: Session, group: Group, inviter: User, nickname: str
|
||
) -> tuple[GroupInvitation, User]:
|
||
nickname = (nickname or "").strip()
|
||
if not nickname:
|
||
raise ValidationError("Укажите никнейм игрока.")
|
||
user = session.exec(select(User).where(User.nickname == nickname)).first()
|
||
if user is None:
|
||
raise NotFoundError("Игрок с таким ником не найден.")
|
||
if _member_exists(session, group.id, user.id): # type: ignore[arg-type]
|
||
raise ConflictError("Игрок уже в группе.")
|
||
|
||
existing = session.exec(
|
||
select(GroupInvitation).where(
|
||
GroupInvitation.group_id == group.id, GroupInvitation.user_id == user.id
|
||
)
|
||
).first()
|
||
if existing is not None:
|
||
raise ConflictError("Игрок уже приглашён в эту группу.")
|
||
|
||
members, pending = _counts(session, group.id) # type: ignore[arg-type]
|
||
if members + pending >= MAX_GROUP_SIZE:
|
||
raise ConflictError(f"В группе нет мест (максимум {MAX_GROUP_SIZE}).")
|
||
|
||
inv = GroupInvitation(
|
||
group_id=group.id, user_id=user.id, invited_by_id=inviter.id # type: ignore[arg-type]
|
||
)
|
||
session.add(inv)
|
||
session.commit()
|
||
session.refresh(inv)
|
||
return inv, user
|
||
|
||
|
||
def list_for_user(session: Session, user_id: int) -> list[dict]:
|
||
rows = session.exec(
|
||
select(GroupInvitation, Group.name).join(Group, Group.id == GroupInvitation.group_id)
|
||
.where(GroupInvitation.user_id == user_id)
|
||
.order_by(GroupInvitation.created_at.desc())
|
||
).all()
|
||
out = []
|
||
for inv, gname in rows:
|
||
inviter_nick = None
|
||
if inv.invited_by_id is not None:
|
||
inviter = session.get(User, inv.invited_by_id)
|
||
inviter_nick = inviter.nickname if inviter else None
|
||
out.append(
|
||
{
|
||
"id": inv.id,
|
||
"group_id": inv.group_id,
|
||
"group_name": gname,
|
||
"invited_by_nickname": inviter_nick,
|
||
"created_at": iso_utc(inv.created_at),
|
||
}
|
||
)
|
||
return out
|
||
|
||
|
||
def _own_invitation(session: Session, user: User, invitation_id: int) -> GroupInvitation:
|
||
inv = session.get(GroupInvitation, invitation_id)
|
||
if inv is None or inv.user_id != user.id:
|
||
raise NotFoundError("Приглашение не найдено.")
|
||
return inv
|
||
|
||
|
||
def accept(session: Session, user: User, invitation_id: int) -> Group:
|
||
inv = _own_invitation(session, user, invitation_id)
|
||
group = session.get(Group, inv.group_id)
|
||
if group is None:
|
||
session.delete(inv)
|
||
session.commit()
|
||
raise NotFoundError("Группа не найдена.")
|
||
|
||
if not _member_exists(session, group.id, user.id): # type: ignore[arg-type]
|
||
members, _ = _counts(session, group.id) # type: ignore[arg-type]
|
||
if members >= MAX_GROUP_SIZE:
|
||
raise ConflictError(f"В группе нет мест (максимум {MAX_GROUP_SIZE}).")
|
||
session.add(
|
||
GroupMember(group_id=group.id, user_id=user.id, role="member") # type: ignore[arg-type]
|
||
)
|
||
if user.active_group_id is None: # первая группа — делаем активной
|
||
user.active_group_id = group.id
|
||
session.add(user)
|
||
|
||
session.delete(inv)
|
||
session.commit()
|
||
return group
|
||
|
||
|
||
def decline(session: Session, user: User, invitation_id: int) -> None:
|
||
inv = _own_invitation(session, user, invitation_id)
|
||
session.delete(inv)
|
||
session.commit()
|