diff --git a/backend/config/settings.py b/backend/config/settings.py index 12e1db2..6e8a410 100644 --- a/backend/config/settings.py +++ b/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, + }, +} + diff --git a/backend/crawler/tasks.py b/backend/crawler/tasks.py index f3de09f..240156a 100644 --- a/backend/crawler/tasks.py +++ b/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)) + + diff --git a/docker-compose.yml b/docker-compose.yml index 4fad9f3..b8961d3 100644 --- a/docker-compose.yml +++ b/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: .