from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.dummy import DummyOperator
# 导入 HiveOperator，用于执行 Hive SQL，目前暂时不使用
# from airflow.providers.apache.hive.operators.hive import HiveOperator

# 导入外部任务传感器：等待其他 DAG 中的任务完成
from airflow.sensors.external_task import ExternalTaskSensor

# 导入文件传感器：等待指定文件出现
from airflow.sensors.filesystem import FileSensor

from datetime import datetime, timedelta


# 定义一个 DAG，即一组具有依赖关系的任务
with DAG(
    dag_id='file_sensor_example',       # DAG 的唯一标识
    schedule_interval=None,              # 不设置定时调度，需要手动触发 DAG
    start_date=datetime(2024, 10, 1),   # DAG 的开始日期
    catchup=False,                       # 不补执行过去未运行的调度任务
    tags=["demo"]                        # 给 DAG 添加标签，方便在 Web 页面筛选
) as dag:

    # 1. 开始任务
    # DummyOperator 是一个空操作任务，本身不执行实际业务逻辑。
    # 通常用于标记工作流的开始、结束，或者组织任务依赖关系。
    start = DummyOperator(
        task_id="start"                 # 当前任务的唯一标识
    )

    # 2. 文件传感器
    # FileSensor 本质上也是一个 Operator。
    # 它会反复检查指定文件是否存在，满足条件后才允许下游任务执行。
    wait_for_file = FileSensor(
        task_id='wait_for_file',                    # 任务 ID
        filepath='/home/hadoop/app/airflow/success',    # 等待出现的文件路径
        fs_conn_id='fs_default',                    # 文件系统连接 ID
        poke_interval=5,                            # 每隔 5 秒检查一次
        timeout=30,                                  # 最多等待 30 秒
        mode='poke'                                 # 使用 poke 模式
        # poke 模式下，传感器每次检查后继续占用 Worker 工作槽等待。
        # 超过 timeout 仍未检测到文件，则任务超时失败。
    )

    # 3. 外部任务传感器
    # ExternalTaskSensor 用来等待另一个 DAG 中的指定任务。
    # 只有满足指定的外部任务状态条件后，当前传感器才会成功。
    wait_for_other_task = ExternalTaskSensor(
        task_id='wait_for_other_task',       # 当前传感器的任务 ID

        external_dag_id='ods',               # 要等待的外部 DAG ID
        external_task_id='user_ods',         # 要等待的外部任务 ID

        # 当前 DAG 运行时，查找逻辑日期早 1 天的外部任务运行记录。
        # 例如，当前逻辑日期为 10 月 12 日，
        # 则查找逻辑日期为 10 月 11 日的 ods.user_ods。
        execution_delta=timedelta(days=1),

        allowed_states=["success"],          # 外部任务成功后，传感器才成功

        poke_interval=5,                     # 每隔 5 秒检查一次外部任务状态
        timeout=3600,                         # 最多等待 3600 秒，也就是 1 小时

        # 未指定 mode 时，默认通常为 poke。
        # 传感器会持续轮询外部任务状态。
    )

    # 4. Hive 任务示例
    # HiveOperator 用于提交并执行 Hive SQL。
    # 目前整段代码被注释，因此不会创建这个任务。
    #
    # user_dwd_task = HiveOperator(
    #     task_id='all_finished',              # Hive 任务 ID
    #     hql="""
    #         insert overwrite xxx
    #         select ....
    #     """                                  # 需要执行的 Hive SQL
    # )

    # 5. Bash 任务
    # BashOperator 用于执行 Linux Shell 命令。
    some_task = BashOperator(
        task_id='all_finished',               # 当前任务的唯一标识
        # {{ ts }} 是 Airflow 模板变量，运行时会被替换为当前 DAG Run 的逻辑时间戳。
        bash_command='echo "Execution time is {{ ts }}"'
    )

    # 6. 定义任务之间的依赖关系
    # >> 表示设置执行顺序：左侧任务成功后，右侧任务才能执行。
    #
    # 执行顺序：
    # start → wait_for_file → some_task
    #
    # 即：
    # ① 执行 start；
    # ② 等待指定文件出现；
    # ③ 文件检测成功后，执行 Bash 命令。
    start >> wait_for_file >> some_task

    # 7. 外部任务依赖示例（当前已注释，不会生效）
    # 如果需要等待 ods DAG 中的 user_ods 任务成功，
    # 然后再执行 user_dwd_task，可以配置下面的依赖关系。
    #
    # start >> wait_for_other_task >> user_dwd_task
    #
    # 注意：
    # 当前 user_dwd_task 的定义也被注释了，因此不能直接启用上面的依赖。
    # 此外，ExternalTaskSensor 虽然已经定义，但当前没有连接到任何任务，
    # 所以它不会按照这条工作流依赖链执行。