from prefect import flow, task
from prefect_shell import shell_run_command
from prefect_dask.task_runners import DaskTaskRunner


def tele_caller():
    return shell_run_command(
        command="python3 /data-analytics/telecaller_leads.py -r 0 ",
        return_all=True,
    )


def teleconsultation():
    return shell_run_command(
        command="python3 /data-analytics/teleconsultation.py -r 0  ",
        return_all=True,
    )


def user_profiles():
    return shell_run_command(
        command=
        "python3 /data-analytics/user_profiles_created_last_24_hours.py -r 0  ",
        return_all=True,
    )


def error_pages():
    return shell_run_command(
        command="python3 /data-analytics/error_pages_data.py -r 0  ",
        return_all=True,
    )


def non_del():
    return shell_run_command(
        command="python3 /data-analytics/non_delivered_orders.py -r 0  ",
        return_all=True,
    )


def non_ship():
    return shell_run_command(
        command="python3 /data-analytics/non_shipped_orders.py -r 0  ",
        return_all=True)


def gmd_review():
    return shell_run_command(
        command="python3 /data-analytics/good_md_content_data.py -r 0  ",
        return_all=True,
    )


def gmd_content():
    return shell_run_command(
        command="python3 /data-analytics/good_md_review_data.py -r 0  ",
        return_all=True)


@flow
def final_docker_run(task_runner=DaskTaskRunner()):
    user_profiles.submit()
    tele_caller.submit()
    teleconsultation.submit()
    non_del.submit()
    non_ship.submit()
    error_pages.submit()
    gmd_content.submit()
    gmd_review.submit()


if __name__ == '__main__':
    final_docker_run()
# print(final_shell_run())

# print(example_shell_run_command_flow())