feat(crawler): implement Celery Beat periodic scheduler for active crawl tasks

backend/config/settings.py

@@ -46,6 +46,7 @@ INSTALLED_APPS = [
     # Third party apps
     'rest_framework',
     'corsheaders',
+    'django_celery_beat',
 
     # Local apps
     'core',
@@ -178,5 +179,11 @@ CELERY_TASK_SERIALIZER = 'json'
 CELERY_RESULT_SERIALIZER = 'json'
 CELERY_TIMEZONE = 'UTC'
 
-# Eager mode option for local testing/unit tests (defaults to True for easy standalone manual testing)
-CELERY_TASK_ALWAYS_EAGER = os.getenv('CELERY_TASK_ALWAYS_EAGER', 'True').lower() in ('true', '1', 't')
+# Celery Beat Schedule
+CELERY_BEAT_SCHEDULE = {
+    'check-and-schedule-crawl-tasks-every-minute': {
+        'task': 'crawler.tasks.check_and_schedule_crawl_tasks',
+        'schedule': 60.0,
+    },
+}
+

backend/crawler/tasks.py

@@ -209,3 +209,43 @@ def run_crawl_pipeline(run_id, force=False):
         run.ads_evaluated_count = len(new_eval_ids)
         run.save()
 
+
+@shared_task
+def check_and_schedule_crawl_tasks():
+    """
+    Periodic master task executed every minute by Celery Beat.
+    Evaluates all active CrawlTasks and dispatches run_crawl_pipeline if:
+    - Current time falls within start_hour and end_hour window
+    - Time elapsed since the last run is >= interval_minutes (or has never run)
+    """
+    now_dt = timezone.now()
+    now_time = timezone.localtime(now_dt).time()
+
+    active_tasks = CrawlTask.objects.filter(is_active=True)
+    for task in active_tasks:
+        # Check start_hour and end_hour window (handles overnight windows like 22:00 to 06:00)
+        in_window = False
+        if task.start_hour <= task.end_hour:
+            in_window = task.start_hour <= now_time <= task.end_hour
+        else:
+            in_window = now_time >= task.start_hour or now_time <= task.end_hour
+
+        if not in_window:
+            continue
+
+        # Check last run
+        last_run = CrawlRun.objects.filter(crawl_task=task).order_by('-started_at').first()
+        if last_run:
+            # If a run is currently in progress, skip scheduling another
+            if last_run.status == 'RUNNING':
+                continue
+
+            elapsed_minutes = (now_dt - last_run.started_at).total_seconds() / 60.0
+            if elapsed_minutes < task.interval_minutes:
+                continue
+
+        # Create new run and trigger background pipeline
+        run = CrawlRun.objects.create(crawl_task=task, status='RUNNING')
+        run_crawl_pipeline.delay(str(run.id))
+
+

docker-compose.yml

@@ -70,6 +70,26 @@ services:
       redis:
         condition: service_healthy
 
+  celery_beat:
+    build:
+      context: .
+      dockerfile: Dockerfile.backend
+    container_name: divar_celery_beat
+    command: celery -A config beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler
+    environment:
+      - REDIS_URL=redis://redis:6379/0
+      - POSTGRES_HOST=db
+      - CELERY_TASK_ALWAYS_EAGER=False
+    env_file:
+      - .env
+    volumes:
+      - ./backend:/app/backend
+    depends_on:
+      db:
+        condition: service_healthy
+      redis:
+        condition: service_healthy
+
   frontend:
     build:
       context: .