Files
ForbiddenStarsApp/backend/app/routers/groups.py
T
NotBigGhostandClaude Opus 4.8 f4286ffa6c Уведомления: persistent + live (push-to-invalidate → pull)
Новая система уведомлений поверх готовой SSE-шины: события пишутся в БД,
живут 72 часа и чистятся (при чтении списка + фоновой задачей), всплывают
сверху экрана в момент прихода, доступны через колокольчик в правом верхнем
углу (бейдж непрочитанных + панель).

Типы: приглашение в группу (→ /group), старт/финиш партии участникам кроме
инициатора (→ /match/{id}). Титулы — готовый хелпер-задел (не подключён, т.к.
выдача титулов игрокам ещё не реализована).

Бэкенд: модель Notification + миграция 0008 (идемпотентная), notification_service,
notify.notifications_changed, роутер /api/notifications (GET + /read), триггеры
в groups/matches, фоновая чистка в lifespan, защита hub.publish от закрытого loop.
Фронт: useNotifications/useMarkNotificationsRead, NotificationBell/Panel/Toaster,
перекомпоновка top-bar, стили; useServerEvents знает тип notifications.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-26 15:16:40 +03:00

226 lines
8.5 KiB
Python

"""Группы, членство, доступные фракции, список партий и статистика группы."""
from __future__ import annotations
from fastapi import APIRouter, Depends, Query, Request
from sqlmodel import Session
from app.auth.deps import get_current_user
from app.core.timeutil import iso_utc
from app.db.session import get_session
from app.models import User
from app.schemas import api as s
from app.services import (
audit_service,
group_service,
invitation_service,
membership_service,
notification_service,
notify,
stats_service,
)
router = APIRouter(prefix="/groups", tags=["groups"])
def _detail(session: Session, group_id: int, user_id: int) -> s.GroupDetail:
group = group_service.get_group(session, group_id)
member = group_service.assert_member(session, group_id, user_id)
return s.GroupDetail(
id=group.id, # type: ignore[arg-type]
name=group.name,
owner_id=group.owner_id,
my_role=member.role,
expansion_ids=group_service.group_expansion_ids(session, group_id),
)
@router.get("", response_model=list[s.GroupBrief])
def my_groups(
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> list[s.GroupBrief]:
return [
s.GroupBrief(id=g.id, name=g.name, role=role)
for g, role in group_service.list_user_groups(session, user.id) # type: ignore[arg-type]
]
@router.post("", response_model=s.GroupDetail)
def create_group(
body: s.GroupCreate,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.GroupDetail:
group = group_service.create_group(session, user, body.name, body.expansion_ids)
audit_service.record(
session,
actor_id=user.id,
action="create",
entity_type="group",
entity_id=group.id,
payload={"name": group.name},
ip=request.client.host if request.client else None,
user_agent=request.headers.get("user-agent"),
)
session.commit()
return _detail(session, group.id, user.id) # type: ignore[arg-type]
@router.get("/{group_id}", response_model=s.GroupDetail)
def get_group(
group_id: int,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.GroupDetail:
return _detail(session, group_id, user.id) # type: ignore[arg-type]
@router.patch("/{group_id}", response_model=s.GroupDetail)
def rename_group(
group_id: int,
body: s.GroupRename,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.GroupDetail:
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
group = group_service.get_group(session, group_id)
group_service.rename_group(session, group, body.name)
return _detail(session, group_id, user.id) # type: ignore[arg-type]
@router.put("/{group_id}/expansions", response_model=s.GroupDetail)
def set_expansions(
group_id: int,
body: s.GroupExpansionsUpdate,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.GroupDetail:
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
group = group_service.get_group(session, group_id)
group_service.set_expansions(session, group, body.expansion_ids)
return _detail(session, group_id, user.id) # type: ignore[arg-type]
@router.get("/{group_id}/factions", response_model=list[s.FactionRead])
def group_factions(
group_id: int,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> list[s.FactionRead]:
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
return [
s.FactionRead(
id=f.id, code=f.code, name_ru=f.name_ru, expansion_id=f.expansion_id # type: ignore[arg-type]
)
for f in group_service.available_factions(session, group_id)
]
# ─── Членство ────────────────────────────────────────────────────────────────
@router.get("/{group_id}/members", response_model=list[s.MemberRead])
def list_members(
group_id: int,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> list[s.MemberRead]:
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
return [
s.MemberRead(user_id=u.id, nickname=u.nickname, role=m.role)
for m, u in membership_service.list_members(session, group_id)
]
@router.post("/{group_id}/invitations", response_model=s.InvitationRead)
def invite_member(
group_id: int,
body: s.MemberAdd,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.InvitationRead:
# Приглашать вправе ЛЮБОЙ участник группы (не только владелец) — assert_member.
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
group = group_service.get_group(session, group_id)
inv, invited = invitation_service.create_invitation(session, group, user, body.nickname)
audit_service.record(
session,
actor_id=user.id,
action="create",
entity_type="group_invitation",
entity_id=group_id,
payload={"invited_user_id": invited.id, "nickname": invited.nickname},
ip=request.client.host if request.client else None,
user_agent=request.headers.get("user-agent"),
)
session.commit()
notify.invitations_changed(invited.id) # type: ignore[arg-type] # живое появление у получателя
notification_service.invited_to_group(
session, invited.id, group.name, user.nickname # type: ignore[arg-type]
)
return s.InvitationRead(
id=inv.id, # type: ignore[arg-type]
group_id=group_id,
group_name=group.name,
invited_by_nickname=user.nickname,
created_at=iso_utc(inv.created_at),
)
@router.delete("/{group_id}/members/{user_id}", response_model=s.OkResponse)
def remove_member(
group_id: int,
user_id: int,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.OkResponse:
# Любой участник управляет составом; защита создателя — в membership_service
# (нельзя удалить последнего владельца).
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
group = group_service.get_group(session, group_id)
membership_service.remove_member(session, group, user_id)
notify.group_changed(session, group_id, extra_user_ids=[user_id]) # + удалённому
return s.OkResponse()
@router.patch("/{group_id}/members/{user_id}", response_model=s.MemberRead)
def change_member_role(
group_id: int,
user_id: int,
body: s.MemberRoleUpdate,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.MemberRead:
group_service.assert_owner(session, group_id, user.id) # type: ignore[arg-type]
group = group_service.get_group(session, group_id)
member = membership_service.change_role(session, group, user_id, body.role)
from app.services.user_service import get_user
u = get_user(session, user_id)
return s.MemberRead(user_id=u.id, nickname=u.nickname, role=member.role)
# ─── Партии и статистика группы ──────────────────────────────────────────────
@router.get("/{group_id}/matches", response_model=s.MatchList)
def group_matches(
group_id: int,
limit: int = Query(20, ge=1, le=100),
offset: int = Query(0, ge=0),
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> dict:
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
return stats_service.group_match_list(session, group_id, limit=limit, offset=offset)
@router.get("/{group_id}/stats", response_model=s.GroupStats)
def group_stats(
group_id: int,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> dict:
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
return stats_service.group_stats(session, group_id)