S

backend/api/schedules.py

@@ -136,7 +136,7 @@ def create_schedule():
             try:
                 print("[INFO] Triggering immediate Celery Beat schedule update...")
                 # Execute the schedule loader task immediately to update Celery Beat
-                celery_result = load_schedules_task.delay()
+                celery_result = load_schedules_task.apply_async(queue='scheduler')
                 print(f"[INFO] Celery Beat update task queued: {celery_result.id}")
                 
                 # Add the task ID to the response for tracking
@@ -229,7 +229,7 @@ def delete_schedule(schedule_id):
             try:
                 print("[INFO] Triggering immediate Celery Beat schedule update after deletion...")
                 # Execute the schedule loader task immediately to update Celery Beat
-                celery_result = load_schedules_task.delay()
+                celery_result = load_schedules_task.apply_async(queue='scheduler')
                 print(f"[INFO] Celery Beat update task queued: {celery_result.id}")
                 
                 # Add the task ID to the response for tracking

backend/celery_tasks/__init__.py

@@ -1 +1,6 @@
-# Initialize the celery_tasks package
+# Initialize the celery_tasks package
+from . import content_tasks
+from . import schedule_loader
+
+# This ensures that the tasks are registered with Celery
+__all__ = ['content_tasks', 'schedule_loader']

backend/celery_tasks/schedule_loader.py

@@ -58,7 +58,8 @@ def parse_schedule_time(schedule_time):
             'day_of_week': '*'
         }
 
-def load_schedules_task():
+@celery_app.task(bind=True)
+def load_schedules_task(self):
     """
     Celery task to load schedules from the database and create periodic tasks.
     This task runs every 5 minutes to check for new or updated schedules.
@@ -69,6 +70,7 @@ def load_schedules_task():
         
         # Get Supabase configuration
         supabase_config = get_supabase_config()
+        print(f"[CELERY BEAT] Supabase config: URL={supabase_config['SUPABASE_URL'][:50]}...")
         
         # Initialize Supabase client
         supabase_client = init_supabase(
@@ -77,6 +79,7 @@ def load_schedules_task():
         )
         
         # Fetch all schedules from Supabase
+        print("[CELERY BEAT] Executing database query...")
         response = (
             supabase_client
             .table("Scheduling")
@@ -84,10 +87,22 @@ def load_schedules_task():
             .execute()
         )
         
+        print(f"[CELERY BEAT] Database query response: {type(response)}")
+        print(f"[CELERY BEAT] Response data: {response.data if response.data else 'None'}")
+        
         schedules = response.data if response.data else []
         print(f"[CELERY BEAT] Found {len(schedules)} schedules in database")
         logger.info(f"Found {len(schedules)} schedules")
         
+        # Log details of each schedule for debugging
+        for i, schedule in enumerate(schedules):
+            print(f"[CELERY BEAT] Schedule {i}: {schedule}")
+            schedule_id = schedule.get('id')
+            schedule_time = schedule.get('schedule_time')
+            adjusted_time = schedule.get('adjusted_time')
+            social_network = schedule.get('Social_network', {})
+            print(f"[CELERY BEAT] Schedule {schedule_id} - schedule_time: {schedule_time}, adjusted_time: {adjusted_time}, social_network: {social_network}")
+        
         # Get current beat schedule
         current_schedule = celery_app.conf.beat_schedule
         
@@ -103,18 +118,23 @@ def load_schedules_task():
                 schedule_time = schedule.get('schedule_time')
                 adjusted_time = schedule.get('adjusted_time')
                 
+                print(f"[CELERY BEAT] Processing schedule {schedule_id}: schedule_time={schedule_time}, adjusted_time={adjusted_time}")
+                
                 if not schedule_time or not adjusted_time:
                     logger.warning(f"Invalid schedule format for schedule {schedule_id}")
+                    print(f"[CELERY BEAT] WARNING: Invalid schedule format for schedule {schedule_id}")
                     continue
                 
                 # Parse schedule times
                 content_gen_time = parse_schedule_time(adjusted_time)
                 publish_time = parse_schedule_time(schedule_time)
                 
+                print(f"[CELERY BEAT] Parsed times - Content gen: {content_gen_time}, Publish: {publish_time}")
+                
                 # Create content generation job (5 minutes before publishing)
                 gen_job_id = f"gen_{schedule_id}"
                 new_schedule[gen_job_id] = {
-                    'task': 'celery_tasks.content_tasks.generate_content_task',
+                    'task': 'backend.celery_tasks.content_tasks.generate_content_task',
                     'schedule': crontab(
                         minute=content_gen_time['minute'],
                         hour=content_gen_time['hour'],
@@ -127,11 +147,12 @@ def load_schedules_task():
                     )
                 }
                 logger.info(f"Created content generation job: {gen_job_id}")
+                print(f"[CELERY BEAT] Created content generation job: {gen_job_id}")
                 
                 # Create publishing job
                 pub_job_id = f"pub_{schedule_id}"
                 new_schedule[pub_job_id] = {
-                    'task': 'celery_tasks.content_tasks.publish_post_task',
+                    'task': 'backend.celery_tasks.content_tasks.publish_post_task',
                     'schedule': crontab(
                         minute=publish_time['minute'],
                         hour=publish_time['hour'],
@@ -143,6 +164,7 @@ def load_schedules_task():
                     )
                 }
                 logger.info(f"Created publishing job: {pub_job_id}")
+                print(f"[CELERY BEAT] Created publishing job: {pub_job_id}")
                 
             except Exception as e:
                 logger.error(f"Error creating jobs for schedule {schedule.get('id')}: {str(e)}")

start_app.py

@@ -46,7 +46,8 @@ def start_celery_components():
         "--loglevel=debug",
         "--pool=solo",
         "--max-tasks-per-child=100",
-        "--events"  # Enable task events for monitoring
+        "--events",  # Enable task events for monitoring
+        "-Q", "content,publish,scheduler"  # Listen to all queues
     ]
     
     # Start Celery beat