Improved achievement logic #268_achievements
This commit is contained in:
78
kdb-bot/src/modules/achievements/achievement_service.py
Normal file
78
kdb-bot/src/modules/achievements/achievement_service.py
Normal file
@@ -0,0 +1,78 @@
|
||||
from cpl_core.configuration import ConfigurationABC
|
||||
from cpl_core.database.context import DatabaseContextABC
|
||||
from cpl_core.logging import LoggerABC
|
||||
from cpl_discord.service import DiscordBotServiceABC
|
||||
from cpl_translation import TranslatePipe
|
||||
|
||||
from bot_core.configuration.server_settings import ServerSettings
|
||||
from bot_core.service.message_service import MessageService
|
||||
from bot_data.abc.achievement_repository_abc import AchievementRepositoryABC
|
||||
from bot_data.model.achievement import Achievement
|
||||
from bot_data.model.user import User
|
||||
from bot_data.model.user_got_achievement import UserGotAchievement
|
||||
|
||||
|
||||
class AchievementService:
|
||||
def __init__(
|
||||
self,
|
||||
config: ConfigurationABC,
|
||||
logger: LoggerABC,
|
||||
bot: DiscordBotServiceABC,
|
||||
achievements: AchievementRepositoryABC,
|
||||
db: DatabaseContextABC,
|
||||
message_service: MessageService,
|
||||
t: TranslatePipe,
|
||||
):
|
||||
self._config = config
|
||||
self._logger = logger
|
||||
self._bot = bot
|
||||
self._achievements = achievements
|
||||
self._db = db
|
||||
self._message_service = message_service
|
||||
self._t = t
|
||||
|
||||
def _match(self, value: str, operator: str, expected_value: str) -> bool:
|
||||
match operator:
|
||||
case "==":
|
||||
return value == expected_value
|
||||
case "!=":
|
||||
return value != expected_value
|
||||
case "<=":
|
||||
return value <= expected_value
|
||||
case ">=":
|
||||
return value >= expected_value
|
||||
case "<":
|
||||
return value < expected_value
|
||||
case ">":
|
||||
return value > expected_value
|
||||
case _:
|
||||
raise ValueError(f"Invalid operator: ${operator}")
|
||||
|
||||
def has_user_achievement_already(self, user: User, achievement: Achievement) -> bool:
|
||||
user_achievements = self._achievements.get_achievements_by_user_id(user.id)
|
||||
return user_achievements.where(lambda x: x.name == achievement.name).count() > 0
|
||||
|
||||
def has_user_achievement(self, user: User, achievement: Achievement) -> bool:
|
||||
return self._match(str(getattr(user, achievement.attribute)), achievement.operator, achievement.value)
|
||||
|
||||
async def validate_achievements_for_user(self, user: User):
|
||||
achievements = self._achievements.get_achievements_by_server_id(user.server.id)
|
||||
for achievement in achievements:
|
||||
if self.has_user_achievement_already(user, achievement) and not self.has_user_achievement(
|
||||
user, achievement
|
||||
):
|
||||
continue
|
||||
|
||||
self._achievements.add_user_got_achievement(UserGotAchievement(user, achievement, user.server))
|
||||
self._db.save_changes()
|
||||
await self._send_achievement_notification(user.server.discord_id, user.discord_id, achievement.name)
|
||||
|
||||
async def _send_achievement_notification(self, guild_id: int, member_id: int, achievement_name: str):
|
||||
member = self._bot.get_guild(guild_id).get_member(member_id)
|
||||
|
||||
settings: ServerSettings = self._config.get_configuration(f"ServerSettings_{guild_id}")
|
||||
await self._message_service.send_channel_message(
|
||||
self._bot.get_channel(settings.notification_chat_id),
|
||||
self._t.transform("modules.achievements.got_new_achievement").format(member.mention, achievement_name),
|
||||
is_persistent=True,
|
||||
)
|
||||
@@ -1,10 +1,13 @@
|
||||
from cpl_core.configuration import ConfigurationABC
|
||||
from cpl_core.dependency_injection import ServiceCollectionABC
|
||||
from cpl_core.environment import ApplicationEnvironmentABC
|
||||
from cpl_discord.discord_event_types_enum import DiscordEventTypesEnum
|
||||
from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
|
||||
|
||||
from bot_core.abc.module_abc import ModuleABC
|
||||
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
|
||||
from modules.achievements.achievement_service import AchievementService
|
||||
from modules.achievements.events.achievement_on_message_event import AchievementOnMessageEvent
|
||||
|
||||
|
||||
class AchievementsModule(ModuleABC):
|
||||
@@ -15,4 +18,5 @@ class AchievementsModule(ModuleABC):
|
||||
pass
|
||||
|
||||
def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC):
|
||||
pass
|
||||
services.add_transient(AchievementService)
|
||||
self._dc.add_event(DiscordEventTypesEnum.on_message.value, AchievementOnMessageEvent)
|
||||
|
||||
1
kdb-bot/src/modules/achievements/events/__init__.py
Normal file
1
kdb-bot/src/modules/achievements/events/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
# imports
|
||||
@@ -0,0 +1,43 @@
|
||||
import discord
|
||||
from cpl_core.database.context import DatabaseContextABC
|
||||
from cpl_core.logging import LoggerABC
|
||||
from cpl_discord.events import OnMessageABC
|
||||
from cpl_discord.service import DiscordBotServiceABC
|
||||
|
||||
from bot_core.helper.event_checks import EventChecks
|
||||
from bot_data.abc.server_repository_abc import ServerRepositoryABC
|
||||
from bot_data.abc.user_repository_abc import UserRepositoryABC
|
||||
from modules.achievements.achievement_service import AchievementService
|
||||
|
||||
|
||||
class AchievementOnMessageEvent(OnMessageABC):
|
||||
def __init__(
|
||||
self,
|
||||
logger: LoggerABC,
|
||||
bot: DiscordBotServiceABC,
|
||||
achievements: AchievementService,
|
||||
db: DatabaseContextABC,
|
||||
servers: ServerRepositoryABC,
|
||||
users: UserRepositoryABC,
|
||||
):
|
||||
OnMessageABC.__init__(self)
|
||||
|
||||
self._logger = logger
|
||||
self._bot = bot
|
||||
self._achievements = achievements
|
||||
self._db = db
|
||||
self._servers = servers
|
||||
self._users = users
|
||||
|
||||
@EventChecks.check_is_ready()
|
||||
async def on_message(self, message: discord.Message):
|
||||
if message.author.bot:
|
||||
return
|
||||
|
||||
server = self._servers.get_server_by_discord_id(message.guild.id)
|
||||
user = self._users.get_user_by_discord_id_and_server_id(message.author.id, server.id)
|
||||
|
||||
user.message_count += 1
|
||||
self._db.save_changes()
|
||||
|
||||
await self._achievements.validate_achievements_for_user(user)
|
||||
@@ -1,28 +0,0 @@
|
||||
import traceback
|
||||
|
||||
from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
|
||||
from cpl_core.console import Console
|
||||
|
||||
|
||||
class LevelServerSettings(ConfigurationModelABC):
|
||||
def __init__(self):
|
||||
ConfigurationModelABC.__init__(self)
|
||||
|
||||
self._id: int = 0
|
||||
self._changed_level_notification_channel = 0
|
||||
|
||||
@property
|
||||
def id(self) -> int:
|
||||
return self._id
|
||||
|
||||
@property
|
||||
def changed_level_notification_channel(self) -> int:
|
||||
return self._changed_level_notification_channel
|
||||
|
||||
def from_dict(self, settings: dict):
|
||||
try:
|
||||
self._id = int(settings["Id"])
|
||||
self._changed_level_notification_channel = int(settings["ChangedLevelNotificationChannelId"])
|
||||
except Exception as e:
|
||||
Console.error(f"[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings")
|
||||
Console.error(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")
|
||||
@@ -1,31 +0,0 @@
|
||||
import traceback
|
||||
|
||||
from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
|
||||
from cpl_core.console import Console
|
||||
from cpl_query.extension import List
|
||||
|
||||
from modules.level.configuration.level_server_settings import LevelServerSettings
|
||||
|
||||
|
||||
class LevelSettings(ConfigurationModelABC):
|
||||
def __init__(self):
|
||||
ConfigurationModelABC.__init__(self)
|
||||
|
||||
self._servers: List[LevelServerSettings] = List()
|
||||
|
||||
@property
|
||||
def servers(self) -> List[LevelServerSettings]:
|
||||
return self._servers
|
||||
|
||||
def from_dict(self, settings: dict):
|
||||
try:
|
||||
servers = List(LevelServerSettings)
|
||||
for s in settings:
|
||||
st = LevelServerSettings()
|
||||
settings[s]["Id"] = s
|
||||
st.from_dict(settings[s])
|
||||
servers.append(st)
|
||||
self._servers = servers
|
||||
except Exception as e:
|
||||
Console.error(f"[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings")
|
||||
Console.error(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")
|
||||
@@ -6,13 +6,13 @@ from cpl_discord.container import Guild, Role, Member
|
||||
from cpl_discord.service import DiscordBotServiceABC
|
||||
from cpl_translation import TranslatePipe
|
||||
|
||||
from bot_core.configuration.server_settings import ServerSettings
|
||||
from bot_core.service.message_service import MessageService
|
||||
from bot_data.model.level import Level
|
||||
from bot_data.model.user import User
|
||||
from bot_data.service.level_repository_service import LevelRepositoryService
|
||||
from bot_data.service.server_repository_service import ServerRepositoryService
|
||||
from bot_data.service.user_repository_service import UserRepositoryService
|
||||
from modules.level.configuration.level_server_settings import LevelServerSettings
|
||||
|
||||
|
||||
class LevelService:
|
||||
@@ -75,9 +75,9 @@ class LevelService:
|
||||
self._logger.error(__name__, f"Adding role {level_role.name} to {member.name} failed!", e)
|
||||
|
||||
if notification_needed:
|
||||
level_settings: LevelServerSettings = self._config.get_configuration(f"LevelServerSettings_{guild.id}")
|
||||
settings: ServerSettings = self._config.get_configuration(f"ServerSettings_{guild.id}")
|
||||
await self._message_service.send_channel_message(
|
||||
self._bot.get_channel(level_settings.changed_level_notification_channel),
|
||||
self._bot.get_channel(settings.notification_chat_id),
|
||||
self._t.transform("modules.level.new_level_message").format(member.mention, level.name),
|
||||
is_persistent=True,
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user