#!/usr/bin/env python3 from celery.schedules import crontab # Ensure tasks are loaded from app import tasks # noqa: F401 - Imports app/tasks.py so Celery can register tasks # Import the shared Celery instance from app.celery_app import celery from app.config import settings from app.tasks.backup_tasks import cleanup_old_backups, create_backup # noqa: F401 from app.tasks.check_credentials import check_credentials from app.tasks.compute_embedding import backfill_missing_embeddings, compute_document_embedding # noqa: F401 from app.tasks.convert_to_pdf import convert_to_pdf # noqa: F401 from app.tasks.convert_to_pdfa import convert_to_pdfa # noqa: F401 from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf # noqa: F401 from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt # noqa: F401 from app.tasks.finalize_document_storage import finalize_document_storage # noqa: F401 from app.tasks.imap_tasks import pull_all_inboxes # noqa: F401 from app.tasks.monitor_stalled_steps import monitor_stalled_steps # noqa: F401 # **Ensure all tasks are imported before Celery starts** from app.tasks.process_document import process_document # noqa: F401 from app.tasks.process_with_azure_document_intelligence import process_with_azure_document_intelligence # noqa: F401 from app.tasks.process_with_ocr import process_with_ocr # noqa: F401 from app.tasks.refine_text_with_gpt import refine_text_with_gpt # noqa: F401 from app.tasks.rotate_pdf_pages import rotate_pdf_pages # noqa: F401 from app.tasks.send_to_all import send_to_all_destinations # noqa: F401 from app.tasks.subscription_tasks import apply_pending_subscription_changes_all # noqa: F401 # Import new send tasks from app.tasks.upload_to_dropbox import upload_to_dropbox # noqa: F401 from app.tasks.upload_to_email import upload_to_email # noqa: F401 from app.tasks.upload_to_ftp import upload_to_ftp # noqa: F401 from app.tasks.upload_to_google_drive import upload_to_google_drive # noqa: F401 from app.tasks.upload_to_nextcloud import upload_to_nextcloud # noqa: F401 from app.tasks.upload_to_onedrive import upload_to_onedrive # noqa: F401 from app.tasks.upload_to_paperless import upload_to_paperless # noqa: F401 from app.tasks.upload_to_s3 import upload_to_s3 # noqa: F401 from app.tasks.upload_to_sftp import upload_to_sftp # noqa: F401 from app.tasks.upload_to_webdav import upload_to_webdav # noqa: F401 from app.tasks.upload_with_rclone import send_to_all_rclone_destinations, upload_with_rclone # noqa: F401 from app.tasks.uptime_kuma_tasks import ping_uptime_kuma # noqa: F401 from app.tasks.watch_folder_tasks import scan_all_watch_folders # noqa: F401 from app.tasks.webhook_tasks import deliver_webhook_task # noqa: F401 # Register the settings reload signal handler so workers pick up config changes from app.utils.settings_sync import register_settings_reload_signal register_settings_reload_signal() celery.conf.task_routes = { "app.tasks.*": {"queue": "default"}, } @celery.task def test_task(): return "Celery is working!" # Run the check_credentials task at startup check_credentials.apply_async(countdown=10) # Run 10 seconds after worker starts celery.conf.beat_schedule = { "poll-inboxes-every-minute": ( { "task": "app.tasks.imap_tasks.pull_all_inboxes", "schedule": crontab(minute="*/1"), # every 1 minute "options": {"expires": 55}, # Ensure tasks don't pile up } if (settings.imap1_host or settings.imap2_host) else None ), # Add Uptime Kuma ping task if configured "ping-uptime-kuma": ( { "task": "app.tasks.uptime_kuma_tasks.ping_uptime_kuma", "schedule": crontab(minute=f"*/{settings.uptime_kuma_ping_interval}"), "options": {"expires": 55}, # Ensure tasks don't pile up } if settings.uptime_kuma_url else None ), # Check credentials every 5 minutes "check-credentials-regularly": { "task": "app.tasks.check_credentials.check_credentials", "schedule": crontab(minute="*/5"), # Every 5 minutes "options": {"expires": 240}, # 4 minutes expiry }, # Also keep daily check for logs and statistics purposes "check-credentials-daily": { "task": "app.tasks.check_credentials.check_credentials", "schedule": crontab(hour="0", minute="0"), # Midnight "options": {"expires": 3600}, # 1 hour expiry }, # Monitor for stalled processing steps every minute "monitor-stalled-steps": { "task": "app.tasks.monitor_stalled_steps.monitor_stalled_steps", "schedule": crontab(minute="*/1"), # Every minute "options": {"expires": 55}, # Must complete within 55 seconds }, # Watch folder scanning — polls local paths, FTP, SFTP, and cloud ingest folders. # Schedule is controlled by WATCH_FOLDER_POLL_INTERVAL (default: 1 minute). "scan-watch-folders": ( { "task": "app.tasks.watch_folder_tasks.scan_all_watch_folders", "schedule": crontab(minute=f"*/{max(1, settings.watch_folder_poll_interval)}"), "options": {"expires": 55}, } if ( settings.watch_folders or settings.ftp_ingest_enabled or settings.sftp_ingest_enabled or settings.dropbox_ingest_enabled or settings.google_drive_ingest_enabled or settings.onedrive_ingest_enabled or settings.nextcloud_ingest_enabled or settings.s3_ingest_enabled or settings.webdav_ingest_enabled ) else None ), # Backfill embeddings for files that were processed before the # embedding pipeline was enabled, or where the embedding task failed. "backfill-missing-embeddings": { "task": "backfill_missing_embeddings", "schedule": crontab(minute="*/5"), # Every 5 minutes "options": {"expires": 240}, # 4 minutes expiry }, # Apply scheduled subscription downgrades daily at 00:05 UTC "apply-pending-subscription-changes": { "task": "app.tasks.subscription_tasks.apply_pending_subscription_changes_all", "schedule": crontab(hour="0", minute="5"), # 00:05 UTC daily "options": {"expires": 3600}, }, # ── Database backup tasks ────────────────────────────────────────────── # Hourly backup (kept for 4 days) "backup-hourly": ( { "task": "app.tasks.backup_tasks.create_backup", "schedule": crontab(minute="0"), # top of every hour "kwargs": {"backup_type": "hourly"}, "options": {"expires": 3300}, } if settings.backup_enabled else None ), # Daily backup (kept for 3 weeks) – runs at 02:30 UTC "backup-daily": ( { "task": "app.tasks.backup_tasks.create_backup", "schedule": crontab(hour="2", minute="30"), "kwargs": {"backup_type": "daily"}, "options": {"expires": 3600}, } if settings.backup_enabled else None ), # Weekly backup (kept for 13 weeks) – runs every Sunday at 03:00 UTC "backup-weekly": ( { "task": "app.tasks.backup_tasks.create_backup", "schedule": crontab(hour="3", minute="0", day_of_week="0"), "kwargs": {"backup_type": "weekly"}, "options": {"expires": 3600}, } if settings.backup_enabled else None ), } # Remove None entries from beat_schedule celery.conf.beat_schedule = {k: v for k, v in celery.conf.beat_schedule.items() if v is not None}