Repository: langgenius/dify
File: api/extensions/ext_celery.py (Python)

Code before:
        imports.append("schedule.check_upgradable_plugin_task")
        imports.append("tasks.process_tenant_plugin_autoupgrade_check_task")
        beat_schedule["check_upgradable_plugin_task"] = {
            "task": "schedule.check_upgradable_plugin_task.check_upgradable_plugin_task",
            "schedule": crontab(minute="*/15"),
        }
    if dify_config.WORKFLOW_LOG_CLEANUP_ENABLED:
        # 2:00 AM every day
        imports.append("schedule.clean_workflow_runlogs_precise")
        beat_schedule["clean_workflow_runlogs_precise"] = {
            "task": "schedule.clean_workflow_runlogs_precise.clean_workflow_runlogs_precise",
            "schedule": crontab(minute="0", hour="2"),
        }
    if dify_config.SPECIFIC_WORKFLOW_LOG_CLEANUP_ENABLED:
        # 3:00 AM every day - clean logs for specific workflows
        imports.append("schedule.clean_specific_workflow_logs")
        beat_schedule["clean_specific_workflow_logs"] = {
            "task": "schedule.clean_specific_workflow_logs.clean_specific_workflow_logs",
            "schedule": crontab(minute="0", hour="3"),
        }
    if dify_config.ENABLE_WORKFLOW_RUN_CLEANUP_TASK:
        # for saas only
        imports.append("schedule.clean_workflow_runs_task")
        beat_schedule["clean_workflow_runs_task"] = {
            "task": "schedule.clean_workflow_runs_task.clean_workflow_runs_task",
            "schedule": crontab(minute="0", hour="0"),
        }
    if dify_config.ENABLE_WORKFLOW_SCHEDULE_POLLER_TASK:
        imports.append("schedule.workflow_schedule_task")
        beat_schedule["workflow_schedule_task"] = {
            "task": "schedule.workflow_schedule_task.poll_workflow_schedules",
            "schedule": timedelta(minutes=dify_config.WORKFLOW_SCHEDULE_POLLER_INTERVAL),
        }
    if dify_config.ENABLE_TRIGGER_PROVIDER_REFRESH_TASK:
        imports.append("schedule.trigger_provider_refresh_task")
        beat_schedule["trigger_provider_refresh"] = {
            "task": "schedule.trigger_provider_refresh_task.trigger_provider_refresh",
            "schedule": timedelta(minutes=dify_config.TRIGGER_PROVIDER_REFRESH_INTERVAL),
        }
    celery_app.conf.update(beat_schedule=beat_schedule, imports=imports)

    return celery_app


Review comment (refactor):
We already have `schedule.clean_workflow_runs_task` and `schedule.clean_messages` to cover the clean up for general workflow_run and messages clean scenario, not sure if `schedule.clean_specific_workflow_logs` is still necessary.

If it's duplicated, you can consider enhance the existed tasks instead of creating a new one.

Code after:
        imports.append("schedule.queue_monitor_task")
        beat_schedule["datasets-queue-monitor"] = {
            "task": "schedule.queue_monitor_task.queue_monitor_task",
            "schedule": timedelta(minutes=dify_config.QUEUE_MONITOR_INTERVAL or 30),
        }
    if dify_config.ENABLE_HUMAN_INPUT_TIMEOUT_TASK:
        imports.append("tasks.human_input_timeout_tasks")
        beat_schedule["human_input_form_timeout"] = {
            "task": "human_input_form_timeout.check_and_resume",
            "schedule": timedelta(minutes=dify_config.HUMAN_INPUT_TIMEOUT_TASK_INTERVAL),
        }
    if dify_config.ENABLE_CHECK_UPGRADABLE_PLUGIN_TASK and dify_config.MARKETPLACE_ENABLED:
        imports.append("schedule.check_upgradable_plugin_task")
        imports.append("tasks.process_tenant_plugin_autoupgrade_check_task")
        beat_schedule["check_upgradable_plugin_task"] = {
            "task": "schedule.check_upgradable_plugin_task.check_upgradable_plugin_task",
            "schedule": crontab(minute="*/15"),
        }
    if dify_config.WORKFLOW_LOG_CLEANUP_ENABLED:
        # 2:00 AM every day
        imports.append("schedule.clean_workflow_runlogs_precise")
        beat_schedule["clean_workflow_runlogs_precise"] = {
            "task": "schedule.clean_workflow_runlogs_precise.clean_workflow_runlogs_precise",
            "schedule": crontab(minute="0", hour="2"),
        }
    if dify_config.ENABLE_WORKFLOW_RUN_CLEANUP_TASK:
        # for saas only
        imports.append("schedule.clean_workflow_runs_task")
        beat_schedule["clean_workflow_runs_task"] = {
            "task": "schedule.clean_workflow_runs_task.clean_workflow_runs_task",
            "schedule": crontab(minute="0", hour="0"),
        }
    if dify_config.ENABLE_WORKFLOW_SCHEDULE_POLLER_TASK:
        imports.append("schedule.workflow_schedule_task")
        beat_schedule["workflow_schedule_task"] = {
            "task": "schedule.workflow_schedule_task.poll_workflow_schedules",
            "schedule": timedelta(minutes=dify_config.WORKFLOW_SCHEDULE_POLLER_INTERVAL),
        }
    if dify_config.ENABLE_TRIGGER_PROVIDER_REFRESH_TASK:
        imports.append("schedule.trigger_provider_refresh_task")
        beat_schedule["trigger_provider_refresh"] = {
            "task": "schedule.trigger_provider_refresh_task.trigger_provider_refresh",
            "schedule": timedelta(minutes=dify_config.TRIGGER_PROVIDER_REFRESH_INTERVAL),
        }

    if dify_config.ENABLE_API_TOKEN_LAST_USED_UPDATE_TASK:
        imports.append("schedule.update_api_token_last_used_task")
        beat_schedule["batch_update_api_token_last_used"] = {
            "task": "schedule.update_api_token_last_used_task.batch_update_api_token_last_used",
            "schedule": timedelta(minutes=dify_config.API_TOKEN_LAST_USED_UPDATE_INTERVAL),
        }