Merge pull request #1 from ClashKingInc/feat/giveaways

feat: added giveaways task

.github/workflows/main.yml

@@ -1,6 +1,9 @@
 name: Image Builder
 
-on: [push]
+on:
+  push:
+    branches:
+      - main
 
 jobs:
   docker:

.gitignore

@@ -2,3 +2,5 @@
 .env
 *.iml
 .idea/
+.venv/
+*__pycache__*

bot/giveaway/track.py

@@ -0,0 +1,103 @@
+import asyncio
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from datetime import datetime
+
+from aiokafka import AIOKafkaProducer
+from loguru import logger
+
+from bot.giveaway.utils import produce_giveaway_event
+from local.main import config
+from utility.classes import MongoDatabase  # Your MongoDB utility
+
+# Schedule giveaways to start or end
+async def schedule_giveaways(db_client, producer, scheduler):
+    """
+    Check the database for giveaways to start or end and schedule them.
+    """
+    now = datetime.utcnow()
+    try:
+        # Fetch giveaways to start
+        giveaways_to_start = await db_client.giveaways.find({
+            "start_time": {"$lte": now},
+            "status": "scheduled"
+        }).to_list(length=None)
+
+        # Fetch giveaways to end
+        giveaways_to_end = await db_client.giveaways.find({
+            "end_time": {"$lte": now},
+            "status": "ongoing"
+        }).to_list(length=None)
+
+        # Schedule start events
+        for giveaway in giveaways_to_start:
+            logger.info(f"Scheduling giveaway start: {giveaway['_id']}")
+            scheduler.add_job(
+                produce_giveaway_event,
+                "date",
+                run_date=giveaway["start_time"],
+                args=[producer, "giveaway_start", giveaway],  # Call Kafka producer
+                id=f"start-{giveaway['_id']}",
+            )
+            # Update database status
+            await db_client.giveaways.update_one(
+                {"_id": giveaway["_id"]},
+                {"$set": {"status": "ongoing"}}
+            )
+
+        # Schedule end events
+        for giveaway in giveaways_to_end:
+            logger.info(f"Scheduling giveaway end: {giveaway['_id']}")
+            scheduler.add_job(
+                produce_giveaway_event,
+                "date",
+                run_date=giveaway["end_time"],
+                args=[producer, "giveaway_end", giveaway],  # Call Kafka producer
+                id=f"end-{giveaway['_id']}",
+            )
+            # Update database status
+            await db_client.giveaways.update_one(
+                {"_id": giveaway["_id"]},
+                {"$set": {"status": "ended"}}
+            )
+    except Exception as e:
+        logger.error(f"Error while scheduling giveaways: {e}")
+
+
+# Main function
+async def main():
+    """
+    Main entry point for giveaway tracking.
+    """
+    logger.info("Starting giveaway tracker...")
+    try:
+        # Initialize database and Kafka producer
+        db_client = MongoDatabase(
+            stats_db_connection=config.stats_mongodb,
+            static_db_connection=config.static_mongodb,
+        )
+        producer = AIOKafkaProducer(
+            bootstrap_servers=['85.10.200.219:9092'], api_version=(3, 6, 0),
+        )
+        await producer.start()
+
+        # Initialize the scheduler
+        scheduler = AsyncIOScheduler()
+        scheduler.start()
+
+        # Run the scheduling loop
+        while True:
+            await schedule_giveaways(db_client, producer, scheduler)
+            await asyncio.sleep(60)  # Check every minute
+    except Exception as e:
+        logger.error(f"Unexpected error in main loop: {e}")
+    finally:
+        logger.info("Shutting down Kafka producer...")
+        await producer.stop()
+
+
+# Run the script
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except Exception as e:
+        logger.critical(f"Critical error while running the tracker: {e}")

bot/giveaway/utils.py

@@ -0,0 +1,28 @@
+import orjson
+from loguru import logger
+
+# Kafka event producer
+async def produce_giveaway_event(producer, event_type, giveaway):
+    """
+    Send a Kafka event for the given giveaway.
+    """
+    # Build the base message
+    message = {
+        "type": event_type,
+        "channel_id": giveaway['channel_id'],
+        "prize": giveaway['prize'],
+        "mentions": giveaway.get('mentions', []),
+        "winner_count": giveaway.get('winners') if event_type == "giveaway_end" else None,
+    }
+
+    # Include participants if it's a giveaway end event
+    if event_type == "giveaway_end":
+        participants = giveaway.get('entries', [])
+        message["participants"] = participants  # Add the list of participants to the message
+
+    logger.info(f"Sending Kafka event: {message}")
+    # Send the message to Kafka
+    await producer.send_and_wait(
+        topic="giveaway",
+        value=orjson.dumps(message),
+    )

bot/websockets/server.py

@@ -42,7 +42,7 @@ async def send_ws(ws, json):
             logger.error(e)
             EVENT_CLIENTS.discard(ws)
 
-    topics = ['clan', 'player', 'war', 'capital', 'reminder', 'reddit']
+    topics = ['clan', 'player', 'war', 'capital', 'reminder', 'reddit', 'giveaway']
     consumer: AIOKafkaConsumer = AIOKafkaConsumer(
         *topics,
         bootstrap_servers='85.10.200.219:9092',

main.py

@@ -5,6 +5,7 @@
 from bot.player.track import main as bot_player_main
 from bot.reddit.track import main as bot_reddit_main
 from bot.websockets.server import main as websocket_server
+from bot.giveaway.track import main as bot_giveaway_main
 from gamewide.clan_verify.track import main as clan_verify_main
 from gamewide.players.track import broadcast as global_player_main
 from gamewide.scheduled.track import main as global_scheduled_main
@@ -27,6 +28,8 @@
         task = global_player_main
     elif config.tracking_type == 'BOTLEGENDS':
         task = bot_legend_main
+    elif config.tracking_type == 'GIVEAWAY':
+        task = bot_giveaway_main
     elif config.tracking_type == 'GLOBALSCHEDULED':
         task = global_scheduled_main
     elif config.tracking_type == 'GLOBALWAR':

utility/classes.py

@@ -44,6 +44,7 @@ def __init__(self, stats_db_connection, static_db_connection):
             self.looper.join_leave_history
         )
         self.global_players: collection_class = self.clashking.global_players
+        self.giveaways: collection_class = self.clashking.giveaways
         self.raid_weekends: collection_class = self.looper.raid_weekends
         self.war_timer: collection_class = self.looper.war_timer
         self.legend_history: collection_class = self.looper.legend_history