airflow_scheduler/dags/dataops_productline_prepare_dag.py

# dataops_productline_prepare_dag.py
from airflow import DAG
from airflow.operators.python import PythonOperator, ShortCircuitOperator
from airflow.operators.empty import EmptyOperator
from datetime import datetime, timedelta
import logging
import networkx as nx
import json
import os
import re
import glob
from pathlib import Path
import hashlib
import pendulum
from utils import (
    get_pg_conn, 
    get_neo4j_driver,
    get_today_date
)
from config import PG_CONFIG, NEO4J_CONFIG

# 创建日志记录器
logger = logging.getLogger(__name__)

def get_enabled_tables():
    """获取所有启用的表"""
    try:
        conn = get_pg_conn()
        cursor = conn.cursor()
        try:
            cursor.execute("""
                SELECT owner_id, table_name 
                FROM schedule_status 
                WHERE schedule_is_enabled = TRUE
            """)
            result = cursor.fetchall()
            return [row[1] for row in result]  # 只返回表名
        except Exception as e:
            logger.error(f"获取启用表失败: {str(e)}")
            raise Exception(f"PostgreSQL数据库查询失败: {str(e)}")
        finally:
            cursor.close()
            conn.close()
    except Exception as e:
        logger.error(f"连接PostgreSQL数据库失败: {str(e)}")
        raise Exception(f"无法连接PostgreSQL数据库: {str(e)}")

def check_table_directly_subscribed(table_name):
    """检查表是否在schedule_status表中直接调度"""
    try:
        conn = get_pg_conn()
        cursor = conn.cursor()
        try:
            cursor.execute("""
                SELECT schedule_is_enabled
                FROM schedule_status 
                WHERE table_name = %s
            """, (table_name,))
            result = cursor.fetchone()
            return result and result[0] is True
        except Exception as e:
            logger.error(f"检查表订阅状态失败: {str(e)}")
            raise Exception(f"PostgreSQL查询表订阅状态失败: {str(e)}")
        finally:
            cursor.close()
            conn.close()
    except Exception as e:
        logger.error(f"连接PostgreSQL数据库失败: {str(e)}")
        raise Exception(f"无法连接PostgreSQL数据库: {str(e)}")

def should_execute_today(table_name, frequency, exec_date):
    """
    判断指定频率的表在给定执行日期是否应该执行
    
    参数:
        table_name (str): 表名，用于日志记录
        frequency (str): 调度频率，如'daily'、'weekly'、'monthly'、'yearly'，为None时默认为'daily'
        exec_date (str): 执行日期，格式为'YYYY-MM-DD'
    
    返回:
        bool: 如果该表应该在执行日期执行，则返回True，否则返回False
    """
    # 将执行日期字符串转换为pendulum日期对象
    try:
        exec_date_obj = pendulum.parse(exec_date)
    except Exception as e:
        logger.error(f"解析执行日期 {exec_date} 出错: {str(e)}，使用当前日期")
        exec_date_obj = pendulum.today()
    
    # 计算下一个日期，用于判断是否是月初、周初等
    next_date = exec_date_obj.add(days=1)
    
    # 如果频率为None或空字符串，默认为daily
    if not frequency:
        logger.info(f"表 {table_name} 未指定调度频率，默认为daily")
        return True
    
    frequency = frequency.lower() if isinstance(frequency, str) else 'daily'
    
    if frequency == 'daily':
        # 日任务每天都执行
        return True
    elif frequency == 'weekly':
        # 周任务只在周日执行（因为exec_date+1是周一时才执行）
        is_sunday = next_date.day_of_week == 1  # 1表示周一
        logger.info(f"表 {table_name} 是weekly任务，exec_date={exec_date}，next_date={next_date.to_date_string()}，是否周日: {is_sunday}")
        return is_sunday
    elif frequency == 'monthly':
        # 月任务只在每月最后一天执行（因为exec_date+1是月初时才执行）
        is_month_end = next_date.day == 1
        logger.info(f"表 {table_name} 是monthly任务，exec_date={exec_date}，next_date={next_date.to_date_string()}，是否月末: {is_month_end}")
        return is_month_end
    elif frequency == 'quarterly':
        # 季度任务只在每季度最后一天执行（因为exec_date+1是季度初时才执行）
        is_quarter_end = next_date.day == 1 and next_date.month in [1, 4, 7, 10]
        logger.info(f"表 {table_name} 是quarterly任务，exec_date={exec_date}，next_date={next_date.to_date_string()}，是否季末: {is_quarter_end}")
        return is_quarter_end
    elif frequency == 'yearly':
        # 年任务只在每年最后一天执行（因为exec_date+1是年初时才执行）
        is_year_end = next_date.day == 1 and next_date.month == 1
        logger.info(f"表 {table_name} 是yearly任务，exec_date={exec_date}，next_date={next_date.to_date_string()}，是否年末: {is_year_end}")
        return is_year_end
    else:
        # 未知频率，默认执行
        logger.warning(f"表 {table_name} 使用未知的调度频率: {frequency}，默认执行")
        return True

def get_table_info_from_neo4j(table_name):
    """从Neo4j获取表的详细信息，保留完整的scripts_info并确保正确获取源表依赖"""
    try:
        driver = get_neo4j_driver()
    except Exception as e:
        logger.error(f"连接Neo4j数据库失败: {str(e)}")
        raise Exception(f"无法连接Neo4j数据库: {str(e)}")
        
    # 检查表是否直接订阅
    is_directly_schedule = check_table_directly_subscribed(table_name)

    table_info = {
        'target_table': table_name,
        'is_directly_schedule': is_directly_schedule,  # 初始值设为True，从schedule_status表获取
    }
    
    try:
        with driver.session() as session:
            # 尝试执行一个简单查询来验证连接
            try:
                test_result = session.run("RETURN 1 as test")
                test_record = test_result.single()
                if not test_record or test_record.get("test") != 1:
                    logger.error("Neo4j连接测试失败")
                    raise Exception("Neo4j连接测试失败")
            except Exception as e:
                logger.error(f"Neo4j连接测试失败: {str(e)}")
                raise Exception(f"Neo4j连接测试失败: {str(e)}")
            
            # 查询表标签和状态
            query_table = """
                MATCH (t {en_name: $table_name})
                RETURN labels(t) AS labels, t.status AS status, t.frequency AS frequency,
                       t.type AS type, t.storage_location AS storage_location
            """
            try:
                result = session.run(query_table, table_name=table_name)
                record = result.single()
            except Exception as e:
                logger.error(f"Neo4j查询表信息失败: {str(e)}")
                raise Exception(f"Neo4j查询表信息失败: {str(e)}")
            
            if record:
                labels = record.get("labels", [])
                table_info['target_table_label'] = [label for label in labels if label in ["DataResource", "DataModel", "DataSource"]][0] if labels else None
                table_info['target_table_status'] = record.get("status", True)  # 默认为True
                table_info['frequency'] = record.get("frequency")
                table_info['target_type'] = record.get("type")  # 获取type属性
                table_info['storage_location'] = record.get("storage_location")  # 获取storage_location属性
                
                # 根据标签类型查询关系和脚本信息
                if "DataResource" in labels:
                    # 检查是否为structure类型
                    if table_info.get('target_type') == "structure":
                        # 对于structure类型，设置默认值，不查询关系
                        table_info['source_tables'] = []  # 使用空数组表示无源表
                        table_info['script_name'] = "load_file.py"
                        table_info['script_type'] = "python"
                        
                        # csv类型的DataResource没有上游，使用默认的append模式
                        table_info['script_exec_mode'] = "append"
                        logger.info(f"表 {table_name} 为structure类型，使用默认执行模式: append")

                        # 添加脚本信息
                        table_info['scripts_info'] = {
                            "load_file.py": {
                                "sources": [],
                                "script_type": "python",
                                "script_exec_mode": "append"
                            }
                        }

                        return table_info
                    else:
                        # 查询源表关系和脚本信息
                        query_rel = """
                            MATCH (target {en_name: $table_name})-[rel:ORIGINATES_FROM]->(source)
                            WITH source, rel, 
                                 CASE WHEN rel.script_name IS NULL THEN target.en_name + '_script.py' ELSE rel.script_name END AS script_name,
                                 CASE WHEN rel.script_type IS NULL THEN 'python' ELSE rel.script_type END AS script_type
                            RETURN source.en_name AS source_table, script_name AS script_name,
                                   script_type AS script_type, 'append' AS script_exec_mode
                        """
                elif "DataModel" in labels:
                    # 查询源表关系和脚本信息
                    query_rel = """
                        MATCH (target {en_name: $table_name})-[rel:DERIVED_FROM]->(source)
                        WITH source, rel, 
                             CASE WHEN rel.script_name IS NULL THEN target.en_name + '_script.py' ELSE rel.script_name END AS script_name,
                             CASE WHEN rel.script_type IS NULL THEN 'python' ELSE rel.script_type END AS script_type
                        RETURN source.en_name AS source_table, script_name AS script_name,
                               script_type AS script_type, 'append' AS script_exec_mode
                    """
                else:
                    logger.warning(f"表 {table_name} 不是DataResource或DataModel类型")
                    # 即使不是这两种类型，也尝试查询其源表依赖关系
                    query_rel = """
                        MATCH (target {en_name: $table_name})-[rel]->(source)
                        WITH source, rel, 
                             CASE WHEN rel.script_name IS NULL THEN target.en_name + '_script.py' ELSE rel.script_name END AS script_name,
                             CASE WHEN rel.script_type IS NULL THEN 'python' ELSE rel.script_type END AS script_type
                        RETURN source.en_name AS source_table, script_name AS script_name,
                               script_type AS script_type, 'append' AS script_exec_mode
                    """
                
                # 收集所有关系记录
                result = session.run(query_rel, table_name=table_name)
                # 检查result对象是否有collect方法，否则使用data方法或list直接转换
                try:
                    if hasattr(result, 'collect'):
                        records = result.collect()  # 使用collect()获取所有记录
                    else:
                        # 尝试使用其他方法获取记录
                        logger.info(f"表 {table_name} 的查询结果不支持collect方法，尝试使用其他方法")
                        try:
                            records = list(result)  # 直接转换为列表
                        except Exception as e1:
                            logger.warning(f"尝试列表转换失败: {str(e1)}，尝试使用data方法")
                            try:
                                records = result.data()  # 使用data()方法
                            except Exception as e2:
                                logger.warning(f"所有方法都失败，使用空列表: {str(e2)}")
                                records = []
                except Exception as e:
                    logger.warning(f"获取查询结果时出错: {str(e)}，使用空列表")
                    records = []
                
                # 记录查询到的原始记录
                logger.info(f"表 {table_name} 查询到 {len(records)} 条关系记录")
                for idx, rec in enumerate(records):
                    logger.info(f"关系记录[{idx}]: source_table={rec.get('source_table')}, script_name={rec.get('script_name')}, " 
                                f"script_type={rec.get('script_type')}, script_exec_mode={rec.get('script_exec_mode')}")
                
                if records:
                    # 按脚本名称分组源表
                    scripts_info = {}
                    for record in records:
                        script_name = record.get("script_name")
                        source_table = record.get("source_table")
                        script_type = record.get("script_type", "python")
                        script_exec_mode = record.get("script_exec_mode", "append")
                        
                        logger.info(f"处理记录: source_table={source_table}, script_name={script_name}")
                        
                        # 如果script_name为空，生成默认的脚本名
                        if not script_name:
                            script_name = f"{table_name}_script.py"
                            logger.warning(f"表 {table_name} 的关系中没有script_name属性，使用默认值: {script_name}")
                            
                        if script_name not in scripts_info:
                            scripts_info[script_name] = {
                                "sources": [],
                                "script_type": script_type,
                                "script_exec_mode": script_exec_mode
                            }
                        
                        # 确保source_table有值且不为None才添加到sources列表中
                        if source_table and source_table not in scripts_info[script_name]["sources"]:
                            scripts_info[script_name]["sources"].append(source_table)
                            logger.debug(f"为表 {table_name} 的脚本 {script_name} 添加源表: {source_table}")
                    
                    # 处理分组信息
                    if scripts_info:
                        # 存储完整的脚本信息
                        table_info['scripts_info'] = scripts_info
                        
                        # 如果只有一个脚本，直接使用它
                        if len(scripts_info) == 1:
                            script_name = list(scripts_info.keys())[0]
                            script_info = scripts_info[script_name]
                            
                            table_info['source_tables'] = script_info["sources"]  # 使用数组
                            table_info['script_name'] = script_name
                            table_info['script_type'] = script_info["script_type"]
                            table_info['script_exec_mode'] = script_info["script_exec_mode"]
                            logger.info(f"表 {table_name} 有单个脚本 {script_name}，源表: {script_info['sources']}")
                        else:
                            # 如果有多个不同脚本，记录多脚本信息
                            logger.info(f"表 {table_name} 有多个不同脚本: {list(scripts_info.keys())}")
                            # 暂时使用第一个脚本的信息作为默认值
                            first_script = list(scripts_info.keys())[0]
                            table_info['source_tables'] = scripts_info[first_script]["sources"]
                            table_info['script_name'] = first_script
                            table_info['script_type'] = scripts_info[first_script]["script_type"]
                            table_info['script_exec_mode'] = scripts_info[first_script]["script_exec_mode"]
                    else:
                        logger.warning(f"表 {table_name} 未找到有效的脚本信息")
                        table_info['source_tables'] = []  # 使用空数组
                else:
                    # 如果没有找到关系记录，则尝试直接查询表的上游依赖
                    logger.info(f"表 {table_name} 未通过关系查询找到源表，尝试直接查询依赖...")
                    
                    # 查询任何类型的上游依赖关系
                    query_deps = """
                        MATCH (target {en_name: $table_name})-[rel]->(source)
                        RETURN source.en_name AS source_table
                    """
                    
                    try:
                        deps_result = session.run(query_deps, table_name=table_name)
                        deps_records = list(deps_result)
                        
                        source_tables = []
                        for rec in deps_records:
                            src_table = rec.get("source_table")
                            if src_table and src_table not in source_tables:
                                source_tables.append(src_table)
                                logger.info(f"直接依赖查询: 表 {table_name} 依赖于 {src_table}")
                        
                        # 设置默认的脚本名和源表
                        script_name = f"{table_name}_script.py"
                        table_info['source_tables'] = source_tables
                        table_info['script_name'] = script_name
                        table_info['script_type'] = "python"
                        table_info['script_exec_mode'] = "append"
                        
                        # 创建scripts_info
                        table_info['scripts_info'] = {
                            script_name: {
                                "sources": source_tables,
                                "script_type": "python",
                                "script_exec_mode": "append"
                            }
                        }
                        
                        logger.info(f"为表 {table_name} 设置默认脚本 {script_name}，源表: {source_tables}")
                    except Exception as e:
                        logger.warning(f"直接查询表 {table_name} 的依赖关系时出错: {str(e)}")
                        table_info['source_tables'] = []  # 使用空数组
                        table_info['script_name'] = f"{table_name}_script.py"
                        table_info['script_type'] = "python"
                        table_info['script_exec_mode'] = "append"
                        
                        # 创建空的scripts_info
                        table_info['scripts_info'] = {
                            table_info['script_name']: {
                                "sources": [],
                                "script_type": "python",
                                "script_exec_mode": "append"
                            }
                        }
            else:
                logger.warning(f"在Neo4j中找不到表 {table_name} 的信息，设置默认值")
                table_info['source_tables'] = []
                table_info['script_name'] = f"{table_name}_script.py"
                table_info['script_type'] = "python"
                table_info['script_exec_mode'] = "append"
                
                # 创建空的scripts_info
                table_info['scripts_info'] = {
                    table_info['script_name']: {
                        "sources": [],
                        "script_type": "python",
                        "script_exec_mode": "append"
                    }
                }
    except Exception as e:
        if "Neo4j连接" in str(e) or "Neo4j查询" in str(e):
            # 这是我们已经处理过的错误，直接抛出
            raise
        else:
            logger.error(f"处理表 {table_name} 的信息时出错: {str(e)}")
            raise Exception(f"Neo4j数据处理失败: {str(e)}")
    finally:
        driver.close()
    
    return table_info

def process_dependencies(tables_info):
    """处理表间依赖关系，添加被动调度的表"""
    # 存储所有表信息的字典
    all_tables = {t['target_table']: t for t in tables_info}
    try:
        driver = get_neo4j_driver()
    except Exception as e:
        logger.error(f"连接Neo4j数据库失败: {str(e)}")
        raise Exception(f"无法连接Neo4j数据库: {str(e)}")
    
    try:
        with driver.session() as session:
            # 验证连接
            try:
                test_result = session.run("RETURN 1 as test")
                test_record = test_result.single()
                if not test_record or test_record.get("test") != 1:
                    logger.error("Neo4j连接测试失败")
                    raise Exception("Neo4j连接测试失败")
            except Exception as e:
                logger.error(f"Neo4j连接测试失败: {str(e)}")
                raise Exception(f"Neo4j连接测试失败: {str(e)}")
                
            for table_name, table_info in list(all_tables.items()):
                if table_info.get('target_table_label') == 'DataModel':
                    # 查询其依赖表
                    query = """
                        MATCH (dm {en_name: $table_name})-[:DERIVED_FROM]->(dep)
                        RETURN dep.en_name AS dep_name, labels(dep) AS dep_labels, 
                               dep.status AS dep_status, dep.frequency AS dep_frequency
                    """
                    try:
                        result = session.run(query, table_name=table_name)
                    except Exception as e:
                        logger.error(f"Neo4j查询依赖关系失败: {str(e)}")
                        raise Exception(f"Neo4j查询依赖关系失败: {str(e)}")
                    
                    for record in result:
                        dep_name = record.get("dep_name")
                        dep_labels = record.get("dep_labels", [])
                        dep_status = record.get("dep_status", True)
                        dep_frequency = record.get("dep_frequency")
                        
                        # 处理未被直接调度的依赖表
                        if dep_name and dep_name not in all_tables:
                            logger.info(f"发现被动依赖表: {dep_name}, 标签: {dep_labels}")
                            
                            # 获取依赖表详细信息
                            dep_info = get_table_info_from_neo4j(dep_name)
                            dep_info['is_directly_schedule'] = False
                            
                            # 处理调度频率继承
                            if not dep_info.get('frequency'):
                                dep_info['frequency'] = table_info.get('frequency')
                            
                            all_tables[dep_name] = dep_info
    except Exception as e:
        if "Neo4j" in str(e):
            # 已经处理过的错误，直接抛出
            raise
        else:
            logger.error(f"处理依赖关系时出错: {str(e)}")
            raise Exception(f"处理依赖关系时出错: {str(e)}")
    finally:
        driver.close()
    
    return list(all_tables.values())

def filter_invalid_tables(tables_info):
    """过滤无效表及其依赖，使用NetworkX构建依赖图"""
    # 构建表名到索引的映射
    table_dict = {t['target_table']: i for i, t in enumerate(tables_info)}
    
    # 找出无效表
    invalid_tables = set()
    for table in tables_info:
        if table.get('target_table_status') is False:
            invalid_tables.add(table['target_table'])
            logger.info(f"表 {table['target_table']} 的状态为无效")
    
    # 构建依赖图
    G = nx.DiGraph()
    
    # 添加所有节点
    for table in tables_info:
        G.add_node(table['target_table'])
    
    # 查询并添加依赖边
    try:
        driver = get_neo4j_driver()
    except Exception as e:
        logger.error(f"连接Neo4j数据库失败: {str(e)}")
        raise Exception(f"无法连接Neo4j数据库: {str(e)}")
    
    try:
        with driver.session() as session:
            # 验证连接
            try:
                test_result = session.run("RETURN 1 as test")
                test_record = test_result.single()
                if not test_record or test_record.get("test") != 1:
                    logger.error("Neo4j连接测试失败")
                    raise Exception("Neo4j连接测试失败")
            except Exception as e:
                logger.error(f"Neo4j连接测试失败: {str(e)}")
                raise Exception(f"Neo4j连接测试失败: {str(e)}")
                
            for table in tables_info:
                if table.get('target_table_label') == 'DataModel':
                    query = """
                        MATCH (source {en_name: $table_name})-[:DERIVED_FROM]->(target)
                        RETURN target.en_name AS target_name
                    """
                    try:
                        result = session.run(query, table_name=table['target_table'])
                    except Exception as e:
                        logger.error(f"Neo4j查询表依赖关系失败: {str(e)}")
                        raise Exception(f"Neo4j查询表依赖关系失败: {str(e)}")
                    
                    for record in result:
                        target_name = record.get("target_name")
                        if target_name and target_name in table_dict:
                            # 添加从目标到源的边，表示目标依赖于源
                            G.add_edge(table['target_table'], target_name)
                            logger.debug(f"添加依赖边: {table['target_table']} -> {target_name}")
    except Exception as e:
        if "Neo4j" in str(e):
            # 已经处理过的错误，直接抛出
            raise
        else:
            logger.error(f"构建依赖图时出错: {str(e)}")
            raise Exception(f"构建依赖图时出错: {str(e)}")
    finally:
        driver.close()
    
    # 找出依赖于无效表的所有表
    downstream_invalid = set()
    for invalid_table in invalid_tables:
        # 获取可从无效表到达的所有节点
        try:
            descendants = nx.descendants(G, invalid_table)
            downstream_invalid.update(descendants)
            logger.info(f"表 {invalid_table} 的下游无效表: {descendants}")
        except Exception as e:
            logger.error(f"处理表 {invalid_table} 的下游依赖时出错: {str(e)}")
            raise Exception(f"处理下游依赖失败: {str(e)}")
    
    # 合并所有无效表
    all_invalid = invalid_tables.union(downstream_invalid)
    logger.info(f"总共 {len(all_invalid)} 个表被标记为无效: {all_invalid}")
    
    # 过滤出有效表
    valid_tables = [t for t in tables_info if t['target_table'] not in all_invalid]
    logger.info(f"过滤后保留 {len(valid_tables)} 个有效表")
    
    return valid_tables

def touch_product_scheduler_file():
    """
    更新产品线调度器DAG文件的修改时间，触发重新解析
    
    返回:
        bool: 是否成功更新
    """
    data_scheduler_path = os.path.join(os.path.dirname(__file__), 'dataops_productline_execute_dag.py')
    
    success = False
    try:
        if os.path.exists(data_scheduler_path):
            # 更新文件修改时间，触发Airflow重新解析
            os.utime(data_scheduler_path, None)
            logger.info(f"已触发产品线执行DAG重新解析: {data_scheduler_path}")
            success = True
        else:
            logger.warning(f"产品线执行DAG文件不存在: {data_scheduler_path}")
                
        return success
    except Exception as e:
        logger.error(f"触发DAG重新解析时出错: {str(e)}")
        return False

def get_subscription_state_hash():
    """获取订阅表状态的哈希值"""
    try:
        conn = get_pg_conn()
        cursor = conn.cursor()
        try:
            cursor.execute("""
                SELECT table_name, schedule_is_enabled
                FROM schedule_status
                ORDER BY table_name
            """)
            rows = cursor.fetchall()
            # 将所有行拼接成一个字符串，然后计算哈希值
            data_str = '|'.join(f"{row[0]}:{row[1]}" for row in rows)
            return hashlib.md5(data_str.encode()).hexdigest()
        except Exception as e:
            logger.error(f"计算订阅表状态哈希值时出错: {str(e)}")
            raise Exception(f"PostgreSQL查询订阅表状态失败: {str(e)}")
        finally:
            cursor.close()
            conn.close()
    except Exception as e:
        logger.error(f"连接PostgreSQL数据库失败: {str(e)}")
        raise Exception(f"无法连接PostgreSQL数据库: {str(e)}")

def check_execution_plan_in_db(**kwargs):
    """
    检查当天的执行计划是否存在于数据库中
    返回False将阻止所有下游任务执行
    """
    # 获取执行日期
    dag_run = kwargs.get('dag_run')
    logical_date = dag_run.logical_date
    local_logical_date = pendulum.instance(logical_date).in_timezone('Asia/Shanghai')
    exec_date = local_logical_date.strftime('%Y-%m-%d')
    logger.info(f"logical_date： {logical_date} ")
    logger.info(f"local_logical_date {local_logical_date} ")
    logger.info(f"检查执行日期 exec_date {exec_date} 的执行计划是否存在于数据库中")
   
    # 检查数据库中是否存在执行计划
    try:
        conn = get_pg_conn()
        cursor = conn.cursor()
        try:
            cursor.execute("""
                SELECT plan
                FROM airflow_exec_plans
                WHERE exec_date = %s AND dag_id = 'dataops_productline_prepare_dag'
                ORDER BY logical_date DESC
                LIMIT 1
            """, (exec_date,))
            
            result = cursor.fetchone()
            if not result:
                logger.error(f"数据库中不存在执行日期 {exec_date} 的执行计划")
                return False
            
            # 检查执行计划内容是否有效
            try:
                # PostgreSQL的jsonb类型会被psycopg2自动转换为Python字典，无需再使用json.loads
                plan_data = result[0]            
                # 检查必要字段
                if "exec_date" not in plan_data:
                    logger.error("执行计划缺少exec_date字段")
                    return False
                    
                if not isinstance(plan_data.get("scripts", []), list):
                    logger.error("执行计划的scripts字段无效")
                    return False
                    
                if not isinstance(plan_data.get("resource_scripts", []), list):
                    logger.error("执行计划的resource_scripts字段无效")
                    return False

                if not isinstance(plan_data.get("model_scripts", []), list):
                    logger.error("执行计划的model_scripts字段无效")
                    return False
                
                # 检查是否有脚本数据
                scripts = plan_data.get("scripts", [])
                resource_scripts = plan_data.get("resource_scripts", [])
                model_scripts = plan_data.get("model_scripts", [])
                
                logger.info(f"执行计划验证成功: 包含 {len(scripts)} 个脚本，{len(resource_scripts)} 个资源脚本和 {len(model_scripts)} 个模型脚本")
                return True
                
            except Exception as je:
                logger.error(f"处理执行计划数据时出错: {str(je)}")
                return False
            
        except Exception as e:
            logger.error(f"检查数据库中执行计划时出错: {str(e)}")
            raise Exception(f"PostgreSQL查询执行计划失败: {str(e)}")
        finally:
            cursor.close()
            conn.close()
    except Exception as e:
        logger.error(f"连接PostgreSQL数据库失败: {str(e)}")
        raise Exception(f"无法连接PostgreSQL数据库: {str(e)}")

def save_execution_plan_to_db(execution_plan, dag_id, run_id, logical_date, ds):
    """
    将执行计划保存到airflow_exec_plans表
    
    参数:
        execution_plan (dict): 执行计划字典
        dag_id (str): DAG的ID
        run_id (str): DAG运行的ID
        logical_date (datetime): 逻辑日期
        ds (str): 日期字符串，格式为YYYY-MM-DD
    
    返回:
        bool: 操作是否成功
    """
    try:
        conn = get_pg_conn()
        cursor = conn.cursor()
        
        try:
            # 将执行计划转换为JSON字符串
            plan_json = json.dumps(execution_plan)
            
            # 获取本地时间
            local_logical_date = pendulum.instance(logical_date).in_timezone('Asia/Shanghai')
            
            # 插入记录
            cursor.execute("""
                INSERT INTO airflow_exec_plans
                (dag_id, run_id, logical_date, local_logical_date, exec_date, plan)
                VALUES (%s, %s, %s, %s, %s, %s)
            """, (dag_id, run_id, logical_date, local_logical_date, ds, plan_json))
            
            conn.commit()
            logger.info(f"成功将执行计划保存到airflow_exec_plans表，dag_id={dag_id}, run_id={run_id}, exec_date={ds}")
            return True
        except Exception as e:
            logger.error(f"保存执行计划到数据库时出错: {str(e)}")
            conn.rollback()
            raise Exception(f"PostgreSQL保存执行计划失败: {str(e)}")
        finally:
            cursor.close()
            conn.close()
    except Exception as e:
        logger.error(f"连接PostgreSQL数据库失败: {str(e)}")
        raise Exception(f"无法连接PostgreSQL数据库: {str(e)}")

def generate_task_id(script_name, source_tables, target_table):
    """
    根据脚本名和表名生成唯一任务ID
    
    参数:
        script_name (str): 脚本文件名
        source_tables (list): 源表列表
        target_table (str): 目标表名
        
    返回:
        str: 唯一的任务ID
    """
    # 移除脚本名的文件扩展名
    script_base = os.path.splitext(script_name)[0]
    
    # 对于特殊脚本如load_file.py，直接使用目标表名
    if script_name.lower() in ['load_file.py']:
        return f"{script_base}_{target_table}"
    
    # 处理源表部分
    if source_tables:
        # 将所有源表按字母顺序排序并连接
        source_part = "_".join(sorted(source_tables))
        # 生成任务ID: 脚本名_源表_to_目标表
        return f"{script_base}_{source_part}_to_{target_table}"
    else:
        # 没有源表时，只使用脚本名和目标表
        return f"{script_base}_{target_table}"

def prepare_scripts_from_tables(tables_info):
    """
    将表信息转换为脚本信息
    
    参数:
        tables_info (list): 表信息列表
        
    返回:
        list: 脚本信息列表
    """
    scripts = []
    
    for table in tables_info:
        target_table = table['target_table']
        target_table_label = table.get('target_table_label')
        frequency = table.get('frequency')
        
        # 处理表的脚本信息
        if 'scripts_info' in table and table['scripts_info']:
            # 表有多个脚本
            for script_name, script_info in table['scripts_info'].items():
                source_tables = script_info.get('sources', [])
                script_type = script_info.get('script_type', 'python')
                script_exec_mode = script_info.get('script_exec_mode', 'append')
                
                # 生成任务ID
                task_id = generate_task_id(script_name, source_tables, target_table)
                
                # 创建脚本信息
                script = {
                    "script_id": task_id,
                    "script_name": script_name,
                    "source_tables": source_tables,
                    "target_table": target_table,
                    "target_table_label": target_table_label,
                    "script_type": script_type,
                    "script_exec_mode": script_exec_mode,
                    "frequency": frequency,
                    "task_id": task_id
                }
                
                # 为structure类型添加特殊属性
                if table.get('target_type') == "structure":
                    script["target_type"] = "structure"
                    script["storage_location"] = table.get('storage_location')
                
                scripts.append(script)
                logger.info(f"为表 {target_table} 创建脚本 {script_name}，任务ID: {task_id}")
        else:
            # 表只有单个脚本或没有明确指定脚本信息
            script_name = table.get('script_name')
            
            # 如果没有script_name，使用默认值
            if not script_name:
                script_name = f"{target_table}_script.py"
                logger.warning(f"表 {target_table} 没有指定脚本名，使用默认值: {script_name}")
            
            source_tables = table.get('source_tables', [])
            script_type = table.get('script_type', 'python')
            script_exec_mode = table.get('script_exec_mode', 'append')
            
            # 生成任务ID
            task_id = generate_task_id(script_name, source_tables, target_table)
            
            # 创建脚本信息
            script = {
                "script_id": task_id,
                "script_name": script_name,
                "source_tables": source_tables,
                "target_table": target_table,
                "target_table_label": target_table_label,
                "script_type": script_type,
                "script_exec_mode": script_exec_mode,
                "frequency": frequency,
                "task_id": task_id
            }
            
            # 为structure类型添加特殊属性
            if table.get('target_type') == "structure":
                script["target_type"] = "structure"
                script["storage_location"] = table.get('storage_location')
            
            scripts.append(script)
            logger.info(f"为表 {target_table} 创建脚本 {script_name}，任务ID: {task_id}")
    
    return scripts

def build_script_dependency_graph(scripts):
    """
    处理脚本间的依赖关系
    
    参数:
        scripts (list): 脚本信息列表
        
    返回:
        tuple: (依赖关系字典, 图对象)
    """
    # 打印所有脚本的源表信息，用于调试
    logger.info("构建脚本依赖图，当前脚本信息:")
    for script in scripts:
        script_id = script['script_id']
        script_name = script['script_name']
        target_table = script['target_table']
        source_tables = script['source_tables']
        logger.info(f"脚本: {script_id} ({script_name}), 目标表: {target_table}, 源表: {source_tables}")
    
    # 创建目标表到脚本ID的映射
    table_to_scripts = {}
    for script in scripts:
        target_table = script['target_table']
        if target_table not in table_to_scripts:
            table_to_scripts[target_table] = []
        table_to_scripts[target_table].append(script['script_id'])
    
    # 记录表到脚本的映射关系
    logger.info("表到脚本的映射关系:")
    for table, script_ids in table_to_scripts.items():
        logger.info(f"表 {table} 由脚本 {script_ids} 生成")
    
    # 创建脚本依赖关系
    script_dependencies = {}
    for script in scripts:
        script_id = script['script_id']
        source_tables = script['source_tables']
        target_table = script['target_table']
        
        # 初始化依赖列表
        script_dependencies[script_id] = []
        
        # 查找源表对应的脚本
        if source_tables:
            logger.info(f"处理脚本 {script_id} 的依赖关系，源表: {source_tables}")
            for source_table in source_tables:
                if source_table in table_to_scripts:
                    # 添加所有生成源表的脚本作为依赖
                    for source_script_id in table_to_scripts[source_table]:
                        if source_script_id != script_id:  # 避免自我依赖
                            script_dependencies[script_id].append(source_script_id)
                            logger.info(f"添加依赖: {script_id} 依赖于 {source_script_id} (表 {target_table} 依赖于表 {source_table})")
                else:
                    logger.warning(f"源表 {source_table} 没有对应的脚本，无法为脚本 {script_id} 创建依赖")
        else:
            logger.info(f"脚本 {script_id} 没有源表依赖")
    
    # 尝试从Neo4j额外查询依赖关系（如果脚本没有显式的source_tables）
    try:
        driver = get_neo4j_driver()
    except Exception as e:
        logger.error(f"连接Neo4j数据库失败: {str(e)}")
        raise Exception(f"无法连接Neo4j数据库: {str(e)}")
    
    try:
        with driver.session() as session:
            # 验证连接
            try:
                test_result = session.run("RETURN 1 as test")
                test_record = test_result.single()
                if not test_record or test_record.get("test") != 1:
                    logger.error("Neo4j连接测试失败")
                    raise Exception("Neo4j连接测试失败")
            except Exception as e:
                logger.error(f"Neo4j连接测试失败: {str(e)}")
                raise Exception(f"Neo4j连接测试失败: {str(e)}")
                
            for script in scripts:
                script_id = script['script_id']
                target_table = script['target_table']
                
                # 只处理没有源表的脚本
                if not script['source_tables'] and not script_dependencies[script_id]:
                    logger.info(f"脚本 {script_id} 没有源表，尝试从Neo4j直接查询表 {target_table} 的依赖")
                    
                    # 查询表的直接依赖
                    query = """
                        MATCH (target {en_name: $table_name})-[rel]->(dep)
                        RETURN dep.en_name AS dep_name
                    """
                    
                    try:
                        result = session.run(query, table_name=target_table)
                        records = list(result)
                        
                        for record in records:
                            dep_name = record.get("dep_name")
                            if dep_name and dep_name in table_to_scripts:
                                for dep_script_id in table_to_scripts[dep_name]:
                                    if dep_script_id != script_id:  # 避免自我依赖
                                        script_dependencies[script_id].append(dep_script_id)
                                        logger.info(f"从Neo4j添加额外依赖: {script_id} 依赖于 {dep_script_id} (表 {target_table} 依赖于表 {dep_name})")
                    except Exception as e:
                        logger.warning(f"从Neo4j查询表 {target_table} 依赖时出错: {str(e)}")
                        raise Exception(f"Neo4j查询表依赖失败: {str(e)}")
    except Exception as e:
        if "Neo4j" in str(e):
            # 已经处理过的错误，直接抛出
            raise
        else:
            logger.error(f"访问Neo4j获取额外依赖时出错: {str(e)}")
            raise Exception(f"Neo4j依赖查询失败: {str(e)}")
    finally:
        driver.close()
    
    # 构建依赖图
    G = nx.DiGraph()
    
    # 添加所有脚本作为节点
    for script in scripts:
        G.add_node(script['script_id'])
    
    # 添加依赖边
    for script_id, dependencies in script_dependencies.items():
        if dependencies:
            for dep_id in dependencies:
                # 添加从script_id到dep_id的边，表示script_id依赖于dep_id
                G.add_edge(script_id, dep_id)
                logger.debug(f"添加依赖边: {script_id} -> {dep_id}")
        else:
            logger.info(f"脚本 {script_id} 没有依赖的上游脚本")
    
    # 确保所有脚本ID都在依赖关系字典中
    for script in scripts:
        script_id = script['script_id']
        if script_id not in script_dependencies:
            script_dependencies[script_id] = []
    
    # 记录每个脚本的依赖数量
    for script_id, deps in script_dependencies.items():
        logger.info(f"脚本 {script_id} 有 {len(deps)} 个依赖: {deps}")
    
    return script_dependencies, G

def optimize_script_execution_order(scripts, script_dependencies, G):
    """
    使用NetworkX优化脚本执行顺序
    
    参数:
        scripts (list): 脚本信息列表
        script_dependencies (dict): 脚本依赖关系字典
        G (nx.DiGraph): 依赖图对象
        
    返回:
        list: 优化后的脚本执行顺序（脚本ID列表）
    """
    # 检查是否有循环依赖
    try:
        cycles = list(nx.simple_cycles(G))
        if cycles:
            logger.warning(f"检测到循环依赖: {cycles}")
            # 处理循环依赖，可以通过删除一些边来打破循环
            for cycle in cycles:
                # 选择一条边删除，这里简单地选择第一条边
                if len(cycle) > 1:
                    G.remove_edge(cycle[0], cycle[1])
                    logger.warning(f"删除边 {cycle[0]} -> {cycle[1]} 以打破循环")
    except Exception as e:
        logger.error(f"检测循环依赖时出错: {str(e)}")
    
    # 使用拓扑排序获取执行顺序
    try:
        # 反转图，因为我们的边表示"依赖于"关系，而拓扑排序需要"优先于"关系
        reverse_G = G.reverse()
        execution_order = list(nx.topological_sort(reverse_G))
        
        # 反转结果，使上游任务先执行
        execution_order.reverse()
        
        logger.info(f"生成优化的脚本执行顺序: {execution_order}")
        return execution_order
    except Exception as e:
        logger.error(f"生成脚本执行顺序时出错: {str(e)}")
        # 出错时返回原始脚本ID列表，不进行优化
        return [script['script_id'] for script in scripts] 

def prepare_productline_dag_schedule(**kwargs):
    """准备产品线DAG调度任务的主函数"""
    # 添加更严格的异常处理
    try:
        # 检查是否是手动触发模式
        is_manual_trigger = False
        params = kwargs.get('params', {})
        if params and 'MANUAL_TRIGGER' in params:
            is_manual_trigger = params.get('MANUAL_TRIGGER', False)
            if is_manual_trigger:
                logger.info(f"接收到手动触发参数: MANUAL_TRIGGER={is_manual_trigger}")
        
        # 获取执行日期
        dag_run = kwargs.get('dag_run')
        logical_date = dag_run.logical_date
        local_logical_date = pendulum.instance(logical_date).in_timezone('Asia/Shanghai')
        exec_date = local_logical_date.strftime('%Y-%m-%d')
        logger.info(f"开始准备执行日期 {exec_date} 的产品线调度任务")
        
        # 检查是否需要创建新的执行计划
        need_create_plan = False
        
        # 条件1: 数据库中不存在当天的执行计划
        try:
            has_plan_in_db = check_execution_plan_in_db(**kwargs)
            if not has_plan_in_db:
                logger.info(f"数据库中不存在执行日期exec_date {exec_date} 的执行计划，需要创建新的执行计划")
                need_create_plan = True
        except Exception as e:
            # 如果执行计划查询失败，直接报错
            logger.error(f"检查执行计划失败: {str(e)}")
            raise Exception(f"检查执行计划失败，可能是数据库连接问题: {str(e)}")
        
        # 条件2: schedule_status表中的数据发生了变更
        if not need_create_plan:
            # 计算当前哈希值
            current_hash = get_subscription_state_hash()
            # 读取上次记录的哈希值
            hash_file = os.path.join(os.path.dirname(__file__), '.subscription_state')
            last_hash = None
            if os.path.exists(hash_file):
                try:
                    with open(hash_file, 'r') as f:
                        last_hash = f.read().strip()
                except Exception as e:
                    logger.warning(f"读取上次订阅状态哈希值失败: {str(e)}")
            
            # 如果哈希值不同，表示数据发生了变更
            if current_hash != last_hash:
                logger.info(f"检测到schedule_status表数据变更。旧哈希值: {last_hash}, 新哈希值: {current_hash}")
                need_create_plan = True
        
        # 手动触发模式覆盖以上判断
        if is_manual_trigger:
            logger.info("手动触发模式，将创建新的执行计划")
            need_create_plan = True
        
        # 如果不需要创建新的执行计划，直接返回
        if not need_create_plan:
            logger.info("无需创建新的执行计划")
            return 0
        
        # 继续处理，创建新的执行计划
        # 1. 获取启用的表
        enabled_tables = get_enabled_tables()
        logger.info(f"从schedule_status表获取到 {len(enabled_tables)} 个启用的表")
        
        if not enabled_tables:
            logger.warning("没有找到启用的表，准备工作结束")
            return 0
        
        # 2. 获取表的详细信息
        tables_info = []
        for table_name in enabled_tables:
            table_info = get_table_info_from_neo4j(table_name)
            if table_info:
                tables_info.append(table_info)
        
        logger.info(f"成功获取 {len(tables_info)} 个表的详细信息")
        
        # 2.1 根据调度频率过滤表
        filtered_tables_info = []
        for table_info in tables_info:
            table_name = table_info['target_table']
            frequency = table_info.get('frequency')
            
            if should_execute_today(table_name, frequency, exec_date):
                filtered_tables_info.append(table_info)
                logger.info(f"表 {table_name} (频率: {frequency}) 将在今天{exec_date}执行")
            else:
                logger.info(f"表 {table_name} (频率: {frequency}) 今天{exec_date}不执行，已过滤")
        
        logger.info(f"按调度频率过滤后，今天{exec_date}需要执行的表有 {len(filtered_tables_info)} 个")

        # 3. 处理依赖关系，添加被动调度的表
        enriched_tables = process_dependencies(filtered_tables_info)
        logger.info(f"处理依赖后，总共有 {len(enriched_tables)} 个表")
        
        # 4. 过滤无效表及其依赖
        valid_tables = filter_invalid_tables(enriched_tables)
        logger.info(f"过滤无效表后，最终有 {len(valid_tables)} 个有效表")
        
        # 5. 将表信息转换为脚本信息
        scripts = prepare_scripts_from_tables(valid_tables)
        logger.info(f"生成了 {len(scripts)} 个脚本信息")
        
        # 检查所有脚本的source_tables字段
        scripts_without_sources = [s['script_id'] for s in scripts if not s.get('source_tables')]
        if scripts_without_sources:
            logger.warning(f"有 {len(scripts_without_sources)} 个脚本没有源表信息: {scripts_without_sources}")
        
        # 6. 处理脚本依赖关系
        script_dependencies, dependency_graph = build_script_dependency_graph(scripts)
        logger.info(f"构建了脚本依赖关系图，包含 {len(script_dependencies)} 个节点")
        
        # 检查依赖关系是否为空
        empty_deps = {k: v for k, v in script_dependencies.items() if not v}
        if len(empty_deps) == len(script_dependencies):
            logger.warning(f"所有脚本的依赖关系为空，这可能表示Neo4j查询未能正确获取表之间的关系")
            
            # 尝试从Neo4j额外构建顶层表间依赖关系
            logger.info("尝试通过Neo4j直接查询表间依赖关系...")
            driver = get_neo4j_driver()
            try:
                with driver.session() as session:
                    # 查询所有表之间的依赖关系
                    query = """
                        MATCH (a)-[r]->(b)
                        WHERE a.en_name IS NOT NULL AND b.en_name IS NOT NULL
                        RETURN a.en_name AS source, b.en_name AS target
                    """
                    
                    try:
                        result = session.run(query)
                        records = list(result)
                        
                        # 创建表之间的依赖映射
                        table_dependencies = {}
                        for record in records:
                            source = record.get("source")
                            target = record.get("target")
                            
                            if source and target:
                                if source not in table_dependencies:
                                    table_dependencies[source] = []
                                
                                if target not in table_dependencies[source]:
                                    table_dependencies[source].append(target)
                                    logger.info(f"从Neo4j发现表依赖: {source} 依赖于 {target}")
                        
                        # 创建表到脚本的映射
                        table_to_script_map = {}
                        for script in scripts:
                            target_table = script['target_table']
                            script_id = script['script_id']
                            
                            if target_table not in table_to_script_map:
                                table_to_script_map[target_table] = []
                            
                            table_to_script_map[target_table].append(script_id)
                        
                        # 基于表依赖关系构建脚本依赖关系
                        updated_dependencies = False
                        for table, deps in table_dependencies.items():
                            if table in table_to_script_map:
                                for table_script_id in table_to_script_map[table]:
                                    for dep_table in deps:
                                        if dep_table in table_to_script_map:
                                            for dep_script_id in table_to_script_map[dep_table]:
                                                if table_script_id != dep_script_id and dep_script_id not in script_dependencies[table_script_id]:
                                                    script_dependencies[table_script_id].append(dep_script_id)
                                                    logger.info(f"添加额外脚本依赖: {table_script_id} 依赖于 {dep_script_id}")
                                                    updated_dependencies = True
                        
                        if updated_dependencies:
                            # 如果依赖关系有更新，重新构建依赖图并优化执行顺序
                            G = nx.DiGraph()
                            
                            # 添加所有脚本作为节点
                            for script in scripts:
                                G.add_node(script['script_id'])
                            
                            # 添加依赖边
                            for script_id, deps in script_dependencies.items():
                                for dep_id in deps:
                                    G.add_edge(script_id, dep_id)
                            
                            dependency_graph = G
                            logger.info("依赖图已基于表依赖关系重新构建")
                    except Exception as e:
                        logger.warning(f"查询Neo4j获取表级依赖时出错: {str(e)}")
            except Exception as e:
                logger.error(f"尝试直接查询表依赖关系时出错: {str(e)}")
            finally:
                driver.close()
        
        # 7. 优化脚本执行顺序
        execution_order = optimize_script_execution_order(scripts, script_dependencies, dependency_graph)
        logger.info(f"生成优化的执行顺序，包含 {len(execution_order)} 个脚本")
        
        # 8. 分类脚本
        resource_scripts = []
        model_scripts = []
        
        for script in scripts:
            script_id = script['script_id']
            if script['target_table_label'] == 'DataResource':
                resource_scripts.append(script_id)
            elif script['target_table_label'] == 'DataModel':
                model_scripts.append(script_id)
        
        logger.info(f"分类完成: {len(resource_scripts)} 个资源脚本, {len(model_scripts)} 个模型脚本")
        
        # 构建执行计划并保存到数据库
        try:
            # 9. 创建最终执行计划
            execution_plan = {
                "exec_date": exec_date,
                "version": "2.0",
                "scripts": scripts,
                "script_dependencies": script_dependencies,
                "resource_scripts": resource_scripts,
                "model_scripts": model_scripts,
                "execution_order": execution_order
            }
            
            # 10. 更新订阅表状态哈希值
            current_hash = get_subscription_state_hash()
            hash_file = os.path.join(os.path.dirname(__file__), '.subscription_state')
            with open(hash_file, 'w') as f:
                f.write(current_hash)
            logger.info(f"已更新订阅表状态哈希值: {current_hash}")
            
            # 11. 触发产品线执行DAG重新解析
            touch_product_scheduler_file()
            
            # 12. 保存执行计划到数据库表
            try:
                # 获取DAG运行信息
                dag_run = kwargs.get('dag_run')
                if dag_run:
                    dag_id = dag_run.dag_id
                    run_id = dag_run.run_id
                    logical_date = dag_run.logical_date
                    local_logical_date = pendulum.instance(logical_date).in_timezone('Asia/Shanghai')
                else:
                    # 如果无法获取dag_run，使用默认值
                    dag_id = kwargs.get('dag').dag_id if 'dag' in kwargs else "dataops_productline_prepare_dag"
                    run_id = f"manual_{datetime.now().strftime('%Y%m%d%H%M%S')}"
                    logical_date = datetime.now()
                
                # 保存到数据库
                save_result = save_execution_plan_to_db(
                    execution_plan=execution_plan,
                    dag_id=dag_id,
                    run_id=run_id,
                    logical_date=local_logical_date,
                    ds=exec_date
                )
                
                if save_result:
                    logger.info("执行计划已成功保存到数据库")
                else:
                    raise Exception("执行计划保存到数据库失败")
                
            except Exception as db_e:
                # 捕获数据库保存错误
                error_msg = f"保存执行计划到数据库时出错: {str(db_e)}"
                logger.error(error_msg)
                raise Exception(error_msg)
            
        except Exception as e:
            error_msg = f"创建或保存执行计划时出错: {str(e)}"
            logger.error(error_msg)
            # 强制抛出异常，确保任务失败，阻止下游DAG执行
            raise Exception(error_msg)
        
        return len(scripts)  # 返回脚本数量
    except Exception as e:
        error_msg = f"产品线DAG调度任务准备失败: {str(e)}"
        logger.error(error_msg)
        # 强制抛出异常，确保任务失败
        raise Exception(error_msg)

# 创建DAG
with DAG(
    "dataops_productline_prepare_dag",
    start_date=datetime(2024, 1, 1),
    # 每小时执行一次
    schedule_interval="0 * * * *",
    catchup=False,
    default_args={
        'owner': 'airflow',
        'depends_on_past': False,
        'email_on_failure': False,
        'email_on_retry': False,
        'retries': 1,
        'retry_delay': timedelta(minutes=5)
    },
    params={
        'MANUAL_TRIGGER': False, 
    },
) as dag:
    
    # 任务开始标记
    start_preparation = EmptyOperator(
        task_id="start_preparation",
        dag=dag
    )
    
    # 准备调度任务
    prepare_task = PythonOperator(
        task_id="prepare_productline_dag_schedule",
        python_callable=prepare_productline_dag_schedule,
        provide_context=True,
        dag=dag
    )
    
    # 检查执行计划是否存在于数据库中
    check_plan_in_db = ShortCircuitOperator(
        task_id="check_execution_plan_in_db",
        python_callable=check_execution_plan_in_db,
        provide_context=True,
        dag=dag
    )
    
    # 准备完成标记
    preparation_completed = EmptyOperator(
        task_id="preparation_completed",
        dag=dag
    )
    
    # 设置任务依赖
    start_preparation >> prepare_task >> check_plan_in_db >> preparation_completed