kdb-bot/cpl-workspace.json

@@ -17,6 +17,7 @@
       "permission": "src/modules/permission/permission.json",
       "technician": "src/modules/technician/technician.json",
       "short-role-name": "src/modules/short_role_name/short-role-name.json",
+      "steam-special-offers": "src/modules/steam_special_offers/steam-special-offers.json",
       "checks": "tools/checks/checks.json",
       "get-version": "tools/get_version/get-version.json",
       "post-build": "tools/post_build/post-build.json",

kdb-bot/src/bot/bot.json

@@ -69,6 +69,7 @@
       "../modules/level/level.json",
       "../modules/permission/permission.json",
       "../modules/short_role_name/short-role-name.json",
+      "../modules/steam_special_offers/steam-special-offers.json",
       "../modules/technician/technician.json"
     ]
   }

kdb-bot/src/bot/module_list.py

@@ -14,6 +14,7 @@ from modules.database.database_module import DatabaseModule
 from modules.level.level_module import LevelModule
 from modules.permission.permission_module import PermissionModule
 from modules.short_role_name.short_role_name_module import ShortRoleNameModule
+from modules.steam_special_offers.special_offers_module import SteamSpecialOffersModule
 from modules.technician.technician_module import TechnicianModule
 
 
@@ -37,6 +38,7 @@ class ModuleList:
                 TechnicianModule,
                 AchievementsModule,
                 ShortRoleNameModule,
+                SteamSpecialOffersModule,
                 # has to be last!
                 BootLogModule,
                 CoreExtensionModule,

kdb-bot/src/bot/startup.py

@@ -16,6 +16,7 @@ from bot_core.configuration.feature_flags_settings import FeatureFlagsSettings
 from bot_core.logging.command_logger import CommandLogger
 from bot_core.logging.database_logger import DatabaseLogger
 from bot_core.logging.message_logger import MessageLogger
+from bot_core.logging.task_logger import TaskLogger
 from bot_data.db_context import DBContext
 
 
@@ -43,6 +44,7 @@ class Startup(StartupABC):
             services.add_singleton(CustomFileLoggerABC, CommandLogger)
             services.add_singleton(CustomFileLoggerABC, DatabaseLogger)
             services.add_singleton(CustomFileLoggerABC, MessageLogger)
+            services.add_singleton(CustomFileLoggerABC, TaskLogger)
 
         if self._feature_flags.get_flag(FeatureFlagsEnum.api_module):
             services.add_singleton(CustomFileLoggerABC, ApiLogger)

kdb-bot/src/bot_core/configuration/feature_flags_enum.py

@@ -17,6 +17,7 @@ class FeatureFlagsEnum(Enum):
     moderator_module = "ModeratorModule"
     permission_module = "PermissionModule"
     short_role_name_module = "ShortRoleNameModule"
+    steam_special_offers_module = "SteamSpecialOffersModule"
     # features
     api_only = "ApiOnly"
     presence = "Presence"
@@ -25,3 +26,4 @@ class FeatureFlagsEnum(Enum):
     sync_xp = "SyncXp"
     short_role_name = "ShortRoleName"
     technician_full_access = "TechnicianFullAccess"
+    steam_special_offers = "SteamSpecialOffers"

kdb-bot/src/bot_core/configuration/feature_flags_settings.py

@@ -19,6 +19,7 @@ class FeatureFlagsSettings(ConfigurationModelABC):
         FeatureFlagsEnum.permission_module.value: True,  # 02.10.2022 #48
         FeatureFlagsEnum.config_module.value: True,  # 19.07.2023 #127
         FeatureFlagsEnum.short_role_name_module.value: True,  # 28.09.2023 #378
+        FeatureFlagsEnum.steam_special_offers_module.value: True,  # 11.10.2023 #188
         # features
         FeatureFlagsEnum.api_only.value: False,  # 13.10.2022 #70
         FeatureFlagsEnum.presence.value: True,  # 03.10.2022 #56
@@ -27,6 +28,7 @@ class FeatureFlagsSettings(ConfigurationModelABC):
         FeatureFlagsEnum.sync_xp.value: False,  # 25.09.2023 #366
         FeatureFlagsEnum.short_role_name.value: False,  # 28.09.2023 #378
         FeatureFlagsEnum.technician_full_access.value: False,  # 03.10.2023 #393
+        FeatureFlagsEnum.steam_special_offers.value: False,  # 11.10.2023 #188
     }
 
     def __init__(self, **kwargs: dict):

kdb-bot/src/bot_core/logging/task_logger.py

@@ -0,0 +1,15 @@
+from cpl_core.configuration import ConfigurationABC
+from cpl_core.environment import ApplicationEnvironmentABC
+from cpl_core.time import TimeFormatSettings
+
+from bot_core.abc.custom_file_logger_abc import CustomFileLoggerABC
+
+
+class TaskLogger(CustomFileLoggerABC):
+    def __init__(
+        self,
+        config: ConfigurationABC,
+        time_format: TimeFormatSettings,
+        env: ApplicationEnvironmentABC,
+    ):
+        CustomFileLoggerABC.__init__(self, "Task", config, time_format, env)

kdb-bot/src/modules/steam_special_offers/__init__.py

@@ -0,0 +1 @@
+# imports

kdb-bot/src/modules/steam_special_offers/base/__init__.py

@@ -0,0 +1 @@
+# imports

kdb-bot/src/modules/steam_special_offers/base/special_offer_watcher_abc.py

@@ -0,0 +1,13 @@
+from abc import ABC, abstractmethod
+
+from discord.ext import commands
+
+
+class SpecialOfferWatcherABC(commands.Cog):
+    @abstractmethod
+    def __init__(self):
+        commands.Cog.__init__(self)
+
+    @abstractmethod
+    def start(self):
+        pass

kdb-bot/src/modules/steam_special_offers/events/__init__.py

@@ -0,0 +1 @@
+# imports

kdb-bot/src/modules/steam_special_offers/events/special_offer_on_ready_event.py

@@ -0,0 +1,25 @@
+from cpl_core.logging import LoggerABC
+from cpl_discord.events import OnReadyABC
+from cpl_discord.service import DiscordBotServiceABC
+
+from bot_core.logging.task_logger import TaskLogger
+from modules.steam_special_offers.base.special_offer_watcher_abc import SpecialOfferWatcherABC
+
+
+class SpecialOfferOnReadyEvent(OnReadyABC):
+    def __init__(
+        self,
+        logger: TaskLogger,
+        bot: DiscordBotServiceABC,
+        watchers: list[SpecialOfferWatcherABC],
+    ):
+        OnReadyABC.__init__(self)
+
+        self._logger = logger
+        self._bot = bot
+        self._watchers = watchers
+
+    async def on_ready(self):
+        for watcher in self._watchers:
+            self._logger.info(__name__, f"Starting watcher {type(watcher).__name__}")
+            watcher.start()

kdb-bot/src/modules/steam_special_offers/model/__init__.py

@@ -0,0 +1 @@
+# imports

kdb-bot/src/modules/steam_special_offers/model/game_offer.py

@@ -0,0 +1,6 @@
+class GameOffer:
+    def __init__(self, name: str, original_price: float, discount_price: float, discount_pct: int):
+        self.name = name
+        self.original_price = original_price
+        self.discount_price = discount_price
+        self.discount_pct = discount_pct

kdb-bot/src/modules/steam_special_offers/special-offers.json

@@ -0,0 +1,46 @@
+{
+  "ProjectSettings": {
+    "Name": "steam-special-offers",
+    "Version": {
+      "Major": "0",
+      "Minor": "0",
+      "Micro": "0"
+    },
+    "Author": "",
+    "AuthorEmail": "",
+    "Description": "",
+    "LongDescription": "",
+    "URL": "",
+    "CopyrightDate": "",
+    "CopyrightName": "",
+    "LicenseName": "",
+    "LicenseDescription": "",
+    "Dependencies": [
+      "cpl-core>=2023.4.0.post5"
+    ],
+    "DevDependencies": [
+      "cpl-cli>=2023.4.0.post3"
+    ],
+    "PythonVersion": ">=3.10.4",
+    "PythonPath": {
+      "linux": ""
+    },
+    "Classifiers": []
+  },
+  "BuildSettings": {
+    "ProjectType": "library",
+    "SourcePath": "",
+    "OutputPath": "../../dist",
+    "Main": "steam_special_offers.main",
+    "EntryPoint": "steam-special-offers",
+    "IncludePackageData": false,
+    "Included": [],
+    "Excluded": [
+      "*/__pycache__",
+      "*/logs",
+      "*/tests"
+    ],
+    "PackageData": {},
+    "ProjectReferences": []
+  }
+}

kdb-bot/src/modules/steam_special_offers/special_offers_module.py

@@ -0,0 +1,25 @@
+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.steam_special_offers.base.special_offer_watcher_abc import SpecialOfferWatcherABC
+from modules.steam_special_offers.events.special_offer_on_ready_event import SpecialOfferOnReadyEvent
+from modules.steam_special_offers.steam_offer_watcher import SteamOfferWatcher
+
+
+class SteamSpecialOffersModule(ModuleABC):
+    def __init__(self, dc: DiscordCollectionABC):
+        ModuleABC.__init__(self, dc, FeatureFlagsEnum.steam_special_offers_module)
+
+    def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
+        pass
+
+    def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC):
+        services.add_singleton(SpecialOfferWatcherABC, SteamOfferWatcher)
+        # commands
+        # events
+        self._dc.add_event(DiscordEventTypesEnum.on_ready.value, SpecialOfferOnReadyEvent)

kdb-bot/src/modules/steam_special_offers/steam_offer_watcher.py

@@ -0,0 +1,88 @@
+import bs4
+import requests
+from cpl_query.extension import List
+from discord.ext import tasks
+
+from bot_core.logging.task_logger import TaskLogger
+from modules.steam_special_offers.base.special_offer_watcher_abc import SpecialOfferWatcherABC
+from modules.steam_special_offers.model.game_offer import GameOffer
+
+
+class SteamOfferWatcher(SpecialOfferWatcherABC):
+    def __init__(self, logger: TaskLogger):
+        SpecialOfferWatcherABC.__init__(self)
+
+        self._logger = logger
+
+    def start(self):
+        self.watch.start()
+
+    def _get_max_count(self) -> int:
+        count = 0
+        result = requests.get(f"https://store.steampowered.com/search/results?specials=1")
+        soup = bs4.BeautifulSoup(result.text, "lxml")
+        element = soup.find_all("div", {"class": "search_results_count"})
+        if len(element) < 1:
+            return count
+
+        count = int(element[0].contents[0].split(" ")[0].replace(",", ""))
+
+        return count
+
+    def _get_games_from_page(self, start: int, count: int) -> List[GameOffer]:
+        games = List(GameOffer)
+        result = requests.get(
+            f"https://store.steampowered.com/search/results?query&start={start}&count={count}&force_infinite=1&specials=1"
+        )
+        soup = bs4.BeautifulSoup(result.text, "lxml")
+        elements = soup.find_all("a", {"class": "search_result_row"})
+        if len(elements) < 1:
+            return games
+
+        for element in elements:
+            name_element = element.find("span", {"class": "title"})
+            original_price_element = element.find("div", {"class": "discount_original_price"})
+            discount_element = element.find("div", {"class": "discount_pct"})
+            discount_price_element = element.find("div", {"class": "discount_final_price"})
+
+            if (
+                name_element is None
+                or len(name_element.contents) < 1
+                or original_price_element is None
+                or len(original_price_element.contents) < 1
+                or discount_element is None
+                or len(discount_element.contents) < 1
+                or discount_price_element is None
+                or len(discount_price_element.contents) < 1
+            ):
+                continue
+
+            name = name_element.contents[0]
+            original_price = float(
+                original_price_element.contents[0].replace(" ", "").replace("€", "").replace(",", ".")
+            )
+            discount = int(discount_element.contents[0].replace("%", ""))
+            discount_price = float(
+                discount_price_element.contents[0].replace(" ", "").replace("€", "").replace(",", ".")
+            )
+            games.add(GameOffer(name, original_price, discount_price, discount))
+
+        return games
+
+    async def _watch(self):
+        self._logger.warn(__name__, "Watching")
+        new_offers = List(GameOffer)
+
+        max = self._get_max_count()
+        for i in range(0, max + 100, 100):
+            self._logger.debug(__name__, f"Get special offers from {i}")
+            new_offers.extend(self._get_games_from_page(i, 100))
+
+        self._logger.trace(__name__, "Finished watching")
+
+    @tasks.loop(seconds=5)
+    async def watch(self):
+        try:
+            await self._watch()
+        except Exception as e:
+            self._logger.error(__name__, f"Steam offer watcher failed", e)