231 lines
7.4 KiB
Python
231 lines
7.4 KiB
Python
from __future__ import annotations
|
|
|
|
from abc import abstractmethod
|
|
from enum import Enum
|
|
from typing import TYPE_CHECKING, Self
|
|
|
|
from app.utils import truncate
|
|
|
|
from .achievement import Achievement
|
|
from .score import GameMode
|
|
|
|
from pydantic import BaseModel, PrivateAttr
|
|
from sqlmodel import select
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
|
|
CONTENT_TRUNCATE = 36
|
|
|
|
if TYPE_CHECKING:
|
|
from app.database import ChannelType, ChatMessage, User
|
|
|
|
|
|
# https://github.com/ppy/osu-web/blob/master/app/Models/Notification.php
|
|
class NotificationName(str, Enum):
|
|
BEATMAP_OWNER_CHANGE = "beatmap_owner_change"
|
|
BEATMAPSET_DISCUSSION_LOCK = "beatmapset_discussion_lock"
|
|
BEATMAPSET_DISCUSSION_POST_NEW = "beatmapset_discussion_post_new"
|
|
BEATMAPSET_DISCUSSION_QUALIFIED_PROBLEM = "beatmapset_discussion_qualified_problem"
|
|
BEATMAPSET_DISCUSSION_REVIEW_NEW = "beatmapset_discussion_review_new"
|
|
BEATMAPSET_DISCUSSION_UNLOCK = "beatmapset_discussion_unlock"
|
|
BEATMAPSET_DISQUALIFY = "beatmapset_disqualify"
|
|
BEATMAPSET_LOVE = "beatmapset_love"
|
|
BEATMAPSET_NOMINATE = "beatmapset_nominate"
|
|
BEATMAPSET_QUALIFY = "beatmapset_qualify"
|
|
BEATMAPSET_RANK = "beatmapset_rank"
|
|
BEATMAPSET_REMOVE_FROM_LOVED = "beatmapset_remove_from_loved"
|
|
BEATMAPSET_RESET_NOMINATIONS = "beatmapset_reset_nominations"
|
|
CHANNEL_ANNOUNCEMENT = "channel_announcement"
|
|
CHANNEL_MESSAGE = "channel_message"
|
|
CHANNEL_TEAM = "channel_team"
|
|
COMMENT_NEW = "comment_new"
|
|
FORUM_TOPIC_REPLY = "forum_topic_reply"
|
|
TEAM_APPLICATION_ACCEPT = "team_application_accept"
|
|
TEAM_APPLICATION_REJECT = "team_application_reject"
|
|
TEAM_APPLICATION_STORE = "team_application_store"
|
|
USER_ACHIEVEMENT_UNLOCK = "user_achievement_unlock"
|
|
USER_BEATMAPSET_NEW = "user_beatmapset_new"
|
|
USER_BEATMAPSET_REVIVE = "user_beatmapset_revive"
|
|
|
|
# NAME_TO_CATEGORY
|
|
@property
|
|
def category(self) -> str:
|
|
return {
|
|
NotificationName.BEATMAP_OWNER_CHANGE: "beatmap_owner_change",
|
|
NotificationName.BEATMAPSET_DISCUSSION_LOCK: "beatmapset_discussion",
|
|
NotificationName.BEATMAPSET_DISCUSSION_POST_NEW: "beatmapset_discussion",
|
|
NotificationName.BEATMAPSET_DISCUSSION_QUALIFIED_PROBLEM: "beatmapset_problem", # noqa: E501
|
|
NotificationName.BEATMAPSET_DISCUSSION_REVIEW_NEW: "beatmapset_discussion",
|
|
NotificationName.BEATMAPSET_DISCUSSION_UNLOCK: "beatmapset_discussion",
|
|
NotificationName.BEATMAPSET_DISQUALIFY: "beatmapset_state",
|
|
NotificationName.BEATMAPSET_LOVE: "beatmapset_state",
|
|
NotificationName.BEATMAPSET_NOMINATE: "beatmapset_state",
|
|
NotificationName.BEATMAPSET_QUALIFY: "beatmapset_state",
|
|
NotificationName.BEATMAPSET_RANK: "beatmapset_state",
|
|
NotificationName.BEATMAPSET_REMOVE_FROM_LOVED: "beatmapset_state",
|
|
NotificationName.BEATMAPSET_RESET_NOMINATIONS: "beatmapset_state",
|
|
NotificationName.CHANNEL_ANNOUNCEMENT: "announcement",
|
|
NotificationName.CHANNEL_MESSAGE: "channel",
|
|
NotificationName.CHANNEL_TEAM: "channel_team",
|
|
NotificationName.COMMENT_NEW: "comment",
|
|
NotificationName.FORUM_TOPIC_REPLY: "forum_topic_reply",
|
|
NotificationName.TEAM_APPLICATION_ACCEPT: "team_application",
|
|
NotificationName.TEAM_APPLICATION_REJECT: "team_application",
|
|
NotificationName.TEAM_APPLICATION_STORE: "team_application",
|
|
NotificationName.USER_ACHIEVEMENT_UNLOCK: "user_achievement_unlock",
|
|
NotificationName.USER_BEATMAPSET_NEW: "user_beatmapset_new",
|
|
NotificationName.USER_BEATMAPSET_REVIVE: "user_beatmapset_new",
|
|
}[self]
|
|
|
|
|
|
class NotificationDetail(BaseModel):
|
|
@property
|
|
@abstractmethod
|
|
def name(self) -> NotificationName:
|
|
raise NotImplementedError
|
|
|
|
@property
|
|
@abstractmethod
|
|
def object_type(self) -> str:
|
|
raise NotImplementedError
|
|
|
|
@property
|
|
@abstractmethod
|
|
def object_id(self) -> int:
|
|
raise NotImplementedError
|
|
|
|
@property
|
|
@abstractmethod
|
|
def source_user_id(self) -> int:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def get_receivers(self, session: AsyncSession) -> list[int]:
|
|
raise NotImplementedError
|
|
|
|
|
|
class ChannelMessageBase(NotificationDetail):
|
|
title: str = ""
|
|
type: str = ""
|
|
cover_url: str = ""
|
|
|
|
_message: "ChatMessage" = PrivateAttr()
|
|
_user: "User" = PrivateAttr()
|
|
_receiver: list[int] = PrivateAttr()
|
|
|
|
@classmethod
|
|
def init(
|
|
cls,
|
|
message: "ChatMessage",
|
|
user: "User",
|
|
receiver: list[int],
|
|
channel_type: "ChannelType",
|
|
) -> Self:
|
|
instance = cls(
|
|
title=truncate(message.content, CONTENT_TRUNCATE),
|
|
type=channel_type.value.lower(),
|
|
cover_url=user.avatar_url,
|
|
)
|
|
instance._message = message
|
|
instance._user = user
|
|
instance._receiver = receiver
|
|
return instance
|
|
|
|
async def get_receivers(self, session: AsyncSession) -> list[int]:
|
|
return self._receiver
|
|
|
|
@property
|
|
def source_user_id(self) -> int:
|
|
return self._user.id
|
|
|
|
@property
|
|
def object_type(self) -> str:
|
|
return "channel"
|
|
|
|
@property
|
|
def object_id(self) -> int:
|
|
return self._message.channel_id
|
|
|
|
|
|
class ChannelMessage(ChannelMessageBase):
|
|
@property
|
|
def name(self) -> NotificationName:
|
|
return NotificationName.CHANNEL_MESSAGE
|
|
|
|
|
|
class ChannelMessageTeam(ChannelMessageBase):
|
|
@classmethod
|
|
def init(
|
|
cls,
|
|
message: "ChatMessage",
|
|
user: "User",
|
|
) -> ChannelMessageTeam:
|
|
from app.database import ChannelType
|
|
|
|
return super().init(message, user, [], ChannelType.TEAM)
|
|
|
|
@property
|
|
def name(self) -> NotificationName:
|
|
return NotificationName.CHANNEL_TEAM
|
|
|
|
async def get_receivers(self, session: AsyncSession) -> list[int]:
|
|
from app.database import TeamMember
|
|
|
|
user_team_id = (
|
|
await session.exec(
|
|
select(TeamMember.team_id).where(TeamMember.user_id == self._user.id)
|
|
)
|
|
).first()
|
|
if not user_team_id:
|
|
return []
|
|
user_ids = (
|
|
await session.exec(
|
|
select(TeamMember.user_id).where(TeamMember.team_id == user_team_id)
|
|
)
|
|
).all()
|
|
return list(user_ids)
|
|
|
|
|
|
class UserAchievementUnlock(NotificationDetail):
|
|
achievement_id: int
|
|
achievement_mode: str
|
|
cover_url: str
|
|
slug: str
|
|
title: str
|
|
description: str
|
|
user_id: int
|
|
|
|
@classmethod
|
|
def init(cls, achievement: Achievement, user_id: int, mode: "GameMode") -> Self:
|
|
instance = cls(
|
|
title=achievement.name,
|
|
cover_url=achievement.url,
|
|
slug=achievement.assets_id,
|
|
achievement_id=achievement.id,
|
|
achievement_mode=mode.value.lower(),
|
|
description=achievement.desc,
|
|
user_id=user_id,
|
|
)
|
|
return instance
|
|
|
|
async def get_receivers(self, session: AsyncSession) -> list[int]:
|
|
return [self.user_id]
|
|
|
|
@property
|
|
def name(self) -> NotificationName:
|
|
return NotificationName.USER_ACHIEVEMENT_UNLOCK
|
|
|
|
@property
|
|
def object_id(self) -> int:
|
|
return self.achievement_id
|
|
|
|
@property
|
|
def source_user_id(self) -> int:
|
|
return self.user_id
|
|
|
|
@property
|
|
def object_type(self) -> str:
|
|
return "achievement"
|
|
|
|
|
|
NotificationDetails = ChannelMessage | ChannelMessageTeam | UserAchievementUnlock
|