Merge branch 'dev'

dags/dag_data_model_daily.py

@@ -1,6 +1,7 @@
 # dag_data_model_daily.py
 from airflow import DAG
 from airflow.operators.python import PythonOperator
+from airflow.operators.empty import EmptyOperator
 from airflow.sensors.external_task import ExternalTaskSensor
 from datetime import datetime
 from utils import get_enabled_tables, is_data_model_table, run_model_script, get_model_dependency_graph
@@ -25,61 +26,105 @@ with DAG("dag_data_model_daily", start_date=datetime(2024, 1, 1), schedule_inter
     )
     logger.info("创建资源表等待任务 - wait_for_data_resource")
 
+    # 创建一个完成标记任务，确保即使没有处理任务也能标记DAG完成
+    daily_completed = EmptyOperator(
+        task_id="daily_processing_completed",
+        dag=dag
+    )
+    logger.info("创建任务完成标记 - daily_processing_completed")
+
     # 获取启用的 daily 模型表
     try:
         enabled_tables = get_enabled_tables("daily")
         model_tables = [t for t in enabled_tables if is_data_model_table(t['table_name'])]
         logger.info(f"获取到 {len(model_tables)} 个启用的 daily 模型表")
-    except Exception as e:
-        logger.error(f"获取 daily 模型表时出错: {str(e)}")
-        raise
+        
+        if not model_tables:
+            # 如果没有模型表需要处理，直接将等待任务与完成标记相连接
+            logger.info("没有找到需要处理的模型表，DAG将直接标记为完成")
+            wait_for_resource >> daily_completed
+        else:
+            # 获取依赖图
+            try:
+                table_names = [t['table_name'] for t in model_tables]
+                dependency_graph = get_model_dependency_graph(table_names)
+                logger.info(f"构建了 {len(dependency_graph)} 个表的依赖关系图")
+            except Exception as e:
+                logger.error(f"构建依赖关系图时出错: {str(e)}")
+                # 出错时也要确保完成标记被触发
+                wait_for_resource >> daily_completed
+                raise
 
-    # 获取依赖图
-    try:
-        table_names = [t['table_name'] for t in model_tables]
-        dependency_graph = get_model_dependency_graph(table_names)
-        logger.info(f"构建了 {len(dependency_graph)} 个表的依赖关系图")
-    except Exception as e:
-        logger.error(f"构建依赖关系图时出错: {str(e)}")
-        raise
+            # 构建 task 对象
+            task_dict = {}
+            for item in model_tables:
+                try:
+                    task = PythonOperator(
+                        task_id=f"process_model_{item['table_name']}",
+                        python_callable=run_model_script,
+                        op_kwargs={"table_name": item['table_name'], "execution_mode": item['execution_mode']},
+                    )
+                    task_dict[item['table_name']] = task
+                    logger.info(f"创建模型处理任务: process_model_{item['table_name']}")
+                except Exception as e:
+                    logger.error(f"创建任务 process_model_{item['table_name']} 时出错: {str(e)}")
+                    # 出错时也要确保完成标记被触发
+                    wait_for_resource >> daily_completed
+                    raise
 
-    # 构建 task 对象
-    task_dict = {}
-    for item in model_tables:
-        try:
-            task = PythonOperator(
-                task_id=f"process_model_{item['table_name']}",
-                python_callable=run_model_script,
-                op_kwargs={"table_name": item['table_name'], "execution_mode": item['execution_mode']},
-            )
-            task_dict[item['table_name']] = task
-            logger.info(f"创建模型处理任务: process_model_{item['table_name']}")
-        except Exception as e:
-            logger.error(f"创建任务 process_model_{item['table_name']} 时出错: {str(e)}")
-            raise
+            # 建立任务依赖（基于 DERIVED_FROM 图）
+            dependency_count = 0
+            for target, upstream_list in dependency_graph.items():
+                for upstream in upstream_list:
+                    if upstream in task_dict and target in task_dict:
+                        task_dict[upstream] >> task_dict[target]
+                        dependency_count += 1
+                        logger.debug(f"建立依赖关系: {upstream} >> {target}")
+                    else:
+                        logger.warning(f"无法建立依赖关系，缺少任务: {upstream} 或 {target}")
 
-    # 建立任务依赖（基于 DERIVED_FROM 图）
-    dependency_count = 0
-    for target, upstream_list in dependency_graph.items():
-        for upstream in upstream_list:
-            if upstream in task_dict and target in task_dict:
-                task_dict[upstream] >> task_dict[target]
-                dependency_count += 1
-                logger.debug(f"建立依赖关系: {upstream} >> {target}")
-            else:
-                logger.warning(f"无法建立依赖关系，缺少任务: {upstream} 或 {target}")
+            logger.info(f"总共建立了 {dependency_count} 个任务依赖关系")
 
-    logger.info(f"总共建立了 {dependency_count} 个任务依赖关系")
-
-    # 最顶层的 task（没有任何上游）需要依赖资源任务完成
-    all_upstreams = set()
-    for upstreams in dependency_graph.values():
-        all_upstreams.update(upstreams)
-    top_level_tasks = [t for t in table_names if t not in all_upstreams]
-    
-    if top_level_tasks:
-        logger.info(f"发现 {len(top_level_tasks)} 个顶层任务: {', '.join(top_level_tasks)}")
-        for name in top_level_tasks:
-            wait_for_resource >> task_dict[name]
-    else:
-        logger.warning("没有找到顶层任务，请检查依赖关系图是否正确")
+            # 最顶层的 task（没有任何上游）需要依赖资源任务完成
+            all_upstreams = set()
+            for upstreams in dependency_graph.values():
+                all_upstreams.update(upstreams)
+            top_level_tasks = [t for t in table_names if t not in all_upstreams]
+            
+            if top_level_tasks:
+                logger.info(f"发现 {len(top_level_tasks)} 个顶层任务: {', '.join(top_level_tasks)}")
+                for name in top_level_tasks:
+                    wait_for_resource >> task_dict[name]
+            else:
+                logger.warning("没有找到顶层任务，请检查依赖关系图是否正确")
+                # 如果没有顶层任务，直接将等待任务与完成标记相连接
+                wait_for_resource >> daily_completed
+            
+            # 连接所有末端任务（没有下游任务的）到完成标记
+            # 找出所有没有下游任务的任务（即终端任务）
+            terminal_tasks = []
+            for table_name, task in task_dict.items():
+                is_terminal = True
+                for upstream_list in dependency_graph.values():
+                    if table_name in upstream_list:
+                        is_terminal = False
+                        break
+                if is_terminal:
+                    terminal_tasks.append(task)
+                    logger.debug(f"发现终端任务: {table_name}")
+            
+            # 如果有终端任务，将它们连接到完成标记
+            if terminal_tasks:
+                logger.info(f"连接 {len(terminal_tasks)} 个终端任务到完成标记")
+                for task in terminal_tasks:
+                    task >> daily_completed
+            else:
+                # 如果没有终端任务（可能是因为存在循环依赖），直接将等待任务与完成标记相连接
+                logger.warning("没有找到终端任务，直接将等待任务与完成标记相连接")
+                wait_for_resource >> daily_completed
+                
+    except Exception as e:
+        logger.error(f"获取 daily 模型表时出错: {str(e)}")
+        # 出错时也要确保完成标记被触发
+        wait_for_resource >> daily_completed
+        raise

dags/dag_data_model_monthly.py

@@ -1,33 +1,140 @@
 from airflow import DAG
 from airflow.operators.python import PythonOperator
+from airflow.operators.empty import EmptyOperator
 from airflow.sensors.external_task import ExternalTaskSensor
 from datetime import datetime
-from utils import get_subscribed_tables, get_neo4j_dependencies
+from utils import get_enabled_tables, is_data_model_table, run_model_script, get_model_dependency_graph
+from config import NEO4J_CONFIG
 import pendulum
+import logging
 
-def process_monthly_model(table_name, execution_mode):
-    deps = get_neo4j_dependencies(table_name)
-    print(f"Processing monthly model for {table_name} with dependencies: {deps}, mode: {execution_mode}")
+# 创建日志记录器
+logger = logging.getLogger(__name__)
 
 def is_first_day():
-    return pendulum.now().day == 1
+    return True
+    # 生产环境中应使用实际判断
+    # return pendulum.now().day == 1
 
 with DAG("dag_data_model_monthly", start_date=datetime(2024, 1, 1), schedule_interval="@daily", catchup=False) as dag:
+    logger.info("初始化 dag_data_model_monthly DAG")
+    
+    # 等待周模型 DAG 完成
     wait_for_weekly = ExternalTaskSensor(
         task_id="wait_for_weekly_model",
         external_dag_id="dag_data_model_weekly",
-        external_task_id=None,
+        external_task_id="weekly_processing_completed",  # 指定完成标记任务
         mode="poke",
         timeout=3600,
         poke_interval=30
     )
-
+    logger.info("创建周模型等待任务 - wait_for_weekly_model")
+    
+    # 创建一个完成标记任务，确保即使没有处理任务也能标记DAG完成
+    monthly_completed = EmptyOperator(
+        task_id="monthly_processing_completed",
+        dag=dag
+    )
+    logger.info("创建任务完成标记 - monthly_processing_completed")
+    
+    # 检查今天是否是月初
     if is_first_day():
-        monthly_tables = get_subscribed_tables('monthly')
-        for item in monthly_tables:
-            t = PythonOperator(
-                task_id=f"process_monthly_{item['table_name']}",
-                python_callable=process_monthly_model,
-                op_kwargs={"table_name": item['table_name'], "execution_mode": item['execution_mode']},
-            )
-            wait_for_weekly >> t
+        logger.info("今天是月初，开始处理月模型")
+        # 获取启用的 monthly 模型表
+        try:
+            enabled_tables = get_enabled_tables("monthly")
+            model_tables = [t for t in enabled_tables if is_data_model_table(t['table_name'])]
+            logger.info(f"获取到 {len(model_tables)} 个启用的 monthly 模型表")
+            
+            if not model_tables:
+                # 如果没有模型表需要处理，直接将等待任务与完成标记相连接
+                logger.info("没有找到需要处理的月模型表，DAG将直接标记为完成")
+                wait_for_weekly >> monthly_completed
+            else:
+                # 获取依赖图
+                try:
+                    table_names = [t['table_name'] for t in model_tables]
+                    dependency_graph = get_model_dependency_graph(table_names)
+                    logger.info(f"构建了 {len(dependency_graph)} 个表的依赖关系图")
+                except Exception as e:
+                    logger.error(f"构建依赖关系图时出错: {str(e)}")
+                    # 出错时也要确保完成标记被触发
+                    wait_for_weekly >> monthly_completed
+                    raise
+
+                # 构建 task 对象
+                task_dict = {}
+                for item in model_tables:
+                    try:
+                        task = PythonOperator(
+                            task_id=f"process_monthly_{item['table_name']}",
+                            python_callable=run_model_script,
+                            op_kwargs={"table_name": item['table_name'], "execution_mode": item['execution_mode']},
+                        )
+                        task_dict[item['table_name']] = task
+                        logger.info(f"创建模型处理任务: process_monthly_{item['table_name']}")
+                    except Exception as e:
+                        logger.error(f"创建任务 process_monthly_{item['table_name']} 时出错: {str(e)}")
+                        # 出错时也要确保完成标记被触发
+                        wait_for_weekly >> monthly_completed
+                        raise
+
+                # 建立任务依赖（基于 DERIVED_FROM 图）
+                dependency_count = 0
+                for target, upstream_list in dependency_graph.items():
+                    for upstream in upstream_list:
+                        if upstream in task_dict and target in task_dict:
+                            task_dict[upstream] >> task_dict[target]
+                            dependency_count += 1
+                            logger.debug(f"建立依赖关系: {upstream} >> {target}")
+                        else:
+                            logger.warning(f"无法建立依赖关系，缺少任务: {upstream} 或 {target}")
+
+                logger.info(f"总共建立了 {dependency_count} 个任务依赖关系")
+
+                # 最顶层的 task（没有任何上游）需要依赖周模型任务完成
+                all_upstreams = set()
+                for upstreams in dependency_graph.values():
+                    all_upstreams.update(upstreams)
+                top_level_tasks = [t for t in table_names if t not in all_upstreams]
+                
+                if top_level_tasks:
+                    logger.info(f"发现 {len(top_level_tasks)} 个顶层任务: {', '.join(top_level_tasks)}")
+                    for name in top_level_tasks:
+                        wait_for_weekly >> task_dict[name]
+                else:
+                    logger.warning("没有找到顶层任务，请检查依赖关系图是否正确")
+                    # 如果没有顶层任务，直接将等待任务与完成标记相连接
+                    wait_for_weekly >> monthly_completed
+                
+                # 连接所有末端任务（没有下游任务的）到完成标记
+                # 找出所有没有下游任务的任务（即终端任务）
+                terminal_tasks = []
+                for table_name, task in task_dict.items():
+                    is_terminal = True
+                    for upstream_list in dependency_graph.values():
+                        if table_name in upstream_list:
+                            is_terminal = False
+                            break
+                    if is_terminal:
+                        terminal_tasks.append(task)
+                        logger.debug(f"发现终端任务: {table_name}")
+                
+                # 如果有终端任务，将它们连接到完成标记
+                if terminal_tasks:
+                    logger.info(f"连接 {len(terminal_tasks)} 个终端任务到完成标记")
+                    for task in terminal_tasks:
+                        task >> monthly_completed
+                else:
+                    # 如果没有终端任务（可能是因为存在循环依赖），直接将等待任务与完成标记相连接
+                    logger.warning("没有找到终端任务，直接将等待任务与完成标记相连接")
+                    wait_for_weekly >> monthly_completed
+        except Exception as e:
+            logger.error(f"获取 monthly 模型表时出错: {str(e)}")
+            # 出错时也要确保完成标记被触发
+            wait_for_weekly >> monthly_completed
+            raise
+    else:
+        # 如果不是月初，直接将等待任务与完成标记相连接，跳过处理
+        logger.info("今天不是月初，跳过月模型处理")
+        wait_for_weekly >> monthly_completed

dags/dag_data_model_weekly.py

@@ -1,33 +1,140 @@
+# dag_data_model_weekly.py
 from airflow import DAG
 from airflow.operators.python import PythonOperator
+from airflow.operators.empty import EmptyOperator
 from airflow.sensors.external_task import ExternalTaskSensor
 from datetime import datetime
-from utils import get_subscribed_tables, get_neo4j_dependencies
+from utils import get_enabled_tables, is_data_model_table, run_model_script, get_model_dependency_graph
+from config import NEO4J_CONFIG
 import pendulum
+import logging
 
-def process_weekly_model(table_name, execution_mode):
-    deps = get_neo4j_dependencies(table_name)
-    print(f"Processing weekly model for {table_name} with dependencies: {deps}, mode: {execution_mode}")
+# 创建日志记录器
+logger = logging.getLogger(__name__)
 
 def is_monday():
-    return pendulum.now().day_of_week == 0
+    return True
+    #return pendulum.now().day_of_week == 0
 
 with DAG("dag_data_model_weekly", start_date=datetime(2024, 1, 1), schedule_interval="@daily", catchup=False) as dag:
+    logger.info("初始化 dag_data_model_weekly DAG")
+    
+    # 等待日模型 DAG 完成
     wait_for_daily = ExternalTaskSensor(
         task_id="wait_for_daily_model",
         external_dag_id="dag_data_model_daily",
-        external_task_id=None,
+        external_task_id="daily_processing_completed",  # 指定完成标记任务
         mode="poke",
         timeout=3600,
         poke_interval=30
     )
-
+    logger.info("创建日模型等待任务 - wait_for_daily_model")
+    
+    # 创建一个完成标记任务，确保即使没有处理任务也能标记DAG完成
+    weekly_completed = EmptyOperator(
+        task_id="weekly_processing_completed",
+        dag=dag
+    )
+    logger.info("创建任务完成标记 - weekly_processing_completed")
+    
+    # 检查今天是否是周一
     if is_monday():
-        weekly_tables = get_subscribed_tables('weekly')
-        for item in weekly_tables:
-            t = PythonOperator(
-                task_id=f"process_weekly_{item['table_name']}",
-                python_callable=process_weekly_model,
-                op_kwargs={"table_name": item['table_name'], "execution_mode": item['execution_mode']},
-            )
-            wait_for_daily >> t
+        logger.info("今天是周一，开始处理周模型")
+        # 获取启用的 weekly 模型表
+        try:
+            enabled_tables = get_enabled_tables("weekly")
+            model_tables = [t for t in enabled_tables if is_data_model_table(t['table_name'])]
+            logger.info(f"获取到 {len(model_tables)} 个启用的 weekly 模型表")
+            
+            if not model_tables:
+                # 如果没有模型表需要处理，直接将等待任务与完成标记相连接
+                logger.info("没有找到需要处理的周模型表，DAG将直接标记为完成")
+                wait_for_daily >> weekly_completed
+            else:
+                # 获取依赖图
+                try:
+                    table_names = [t['table_name'] for t in model_tables]
+                    dependency_graph = get_model_dependency_graph(table_names)
+                    logger.info(f"构建了 {len(dependency_graph)} 个表的依赖关系图")
+                except Exception as e:
+                    logger.error(f"构建依赖关系图时出错: {str(e)}")
+                    # 出错时也要确保完成标记被触发
+                    wait_for_daily >> weekly_completed
+                    raise
+
+                # 构建 task 对象
+                task_dict = {}
+                for item in model_tables:
+                    try:
+                        task = PythonOperator(
+                            task_id=f"process_weekly_{item['table_name']}",
+                            python_callable=run_model_script,
+                            op_kwargs={"table_name": item['table_name'], "execution_mode": item['execution_mode']},
+                        )
+                        task_dict[item['table_name']] = task
+                        logger.info(f"创建模型处理任务: process_weekly_{item['table_name']}")
+                    except Exception as e:
+                        logger.error(f"创建任务 process_weekly_{item['table_name']} 时出错: {str(e)}")
+                        # 出错时也要确保完成标记被触发
+                        wait_for_daily >> weekly_completed
+                        raise
+
+                # 建立任务依赖（基于 DERIVED_FROM 图）
+                dependency_count = 0
+                for target, upstream_list in dependency_graph.items():
+                    for upstream in upstream_list:
+                        if upstream in task_dict and target in task_dict:
+                            task_dict[upstream] >> task_dict[target]
+                            dependency_count += 1
+                            logger.debug(f"建立依赖关系: {upstream} >> {target}")
+                        else:
+                            logger.warning(f"无法建立依赖关系，缺少任务: {upstream} 或 {target}")
+
+                logger.info(f"总共建立了 {dependency_count} 个任务依赖关系")
+
+                # 最顶层的 task（没有任何上游）需要依赖日模型任务完成
+                all_upstreams = set()
+                for upstreams in dependency_graph.values():
+                    all_upstreams.update(upstreams)
+                top_level_tasks = [t for t in table_names if t not in all_upstreams]
+                
+                if top_level_tasks:
+                    logger.info(f"发现 {len(top_level_tasks)} 个顶层任务: {', '.join(top_level_tasks)}")
+                    for name in top_level_tasks:
+                        wait_for_daily >> task_dict[name]
+                else:
+                    logger.warning("没有找到顶层任务，请检查依赖关系图是否正确")
+                    # 如果没有顶层任务，直接将等待任务与完成标记相连接
+                    wait_for_daily >> weekly_completed
+                
+                # 连接所有末端任务（没有下游任务的）到完成标记
+                # 找出所有没有下游任务的任务（即终端任务）
+                terminal_tasks = []
+                for table_name, task in task_dict.items():
+                    is_terminal = True
+                    for upstream_list in dependency_graph.values():
+                        if table_name in upstream_list:
+                            is_terminal = False
+                            break
+                    if is_terminal:
+                        terminal_tasks.append(task)
+                        logger.debug(f"发现终端任务: {table_name}")
+                
+                # 如果有终端任务，将它们连接到完成标记
+                if terminal_tasks:
+                    logger.info(f"连接 {len(terminal_tasks)} 个终端任务到完成标记")
+                    for task in terminal_tasks:
+                        task >> weekly_completed
+                else:
+                    # 如果没有终端任务（可能是因为存在循环依赖），直接将等待任务与完成标记相连接
+                    logger.warning("没有找到终端任务，直接将等待任务与完成标记相连接")
+                    wait_for_daily >> weekly_completed
+        except Exception as e:
+            logger.error(f"获取 weekly 模型表时出错: {str(e)}")
+            # 出错时也要确保完成标记被触发
+            wait_for_daily >> weekly_completed
+            raise
+    else:
+        # 如果不是周一，直接将等待任务与完成标记相连接，跳过处理
+        logger.info("今天不是周一，跳过周模型处理")
+        wait_for_daily >> weekly_completed

dags/dag_manual_trigger_chain.py

@@ -407,7 +407,7 @@ def process_models(**context):
 
 # 创建DAG
 with DAG(
-    'dag_manual_trigger_chain_two_level',
+    'dag_manual_trigger_chain',
     default_args=default_args,
     description='手动触发指定表的依赖链执行（两级任务）',
     schedule_interval=None,  # 设置为None表示只能手动触发

dags/utils.py

@@ -260,35 +260,106 @@ def is_data_model_table(table_name):
     finally:
         driver.close()
 
-# 动态导入并执行模型脚本的 run() 方法（基于 script_name + execution_mode）
-def run_model_script(table_name, execution_mode):
-    script_name = get_script_name_from_neo4j(table_name)
-    if not script_name:
-        print(f"[WARN] 未找到表 {table_name} 的 script_name，跳过")
-        return
-
-    # scripts_base_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "dataops", "scripts")
-    # script_path = os.path.join(scripts_base_path, script_name)
+# 检查脚本文件是否存在于指定路径
+def check_script_exists(script_name):
+    """
+    检查脚本文件是否存在于配置的脚本目录中
+    
+    参数:
+        script_name (str): 脚本文件名
+        
+    返回:
+        bool: 如果脚本存在返回True，否则返回False
+        str: 完整的脚本路径
+    """
     from pathlib import Path
+    import os
+    import logging
+    
+    logger = logging.getLogger("airflow.task")
+    
+    if not script_name:
+        logger.error("脚本名称为空，无法检查")
+        return False, None
+    
     script_path = Path(SCRIPTS_BASE_PATH) / script_name
+    script_path_str = str(script_path)
+    
+    logger.info(f"检查脚本路径: {script_path_str}")
+    
+    if os.path.exists(script_path_str):
+        logger.info(f"脚本文件已找到: {script_path_str}")
+        return True, script_path_str
+    else:
+        logger.error(f"脚本文件不存在: {script_path_str}")
+        
+        # 尝试列出目录中的文件
+        try:
+            base_dir = Path(SCRIPTS_BASE_PATH)
+            if base_dir.exists():
+                files = list(base_dir.glob("*.py"))
+                logger.info(f"目录 {SCRIPTS_BASE_PATH} 中的Python文件: {[f.name for f in files]}")
+            else:
+                logger.error(f"基础目录不存在: {SCRIPTS_BASE_PATH}")
+        except Exception as e:
+            logger.error(f"列出目录内容时出错: {str(e)}")
+            
+        return False, script_path_str
 
-    if not os.path.exists(script_path):
-        print(f"[ERROR] 脚本文件不存在: {script_path}")
-        return
-
-    print(f"[INFO] 执行模型脚本: {script_path}")
+# 更新run_model_script函数以使用上述检查
+def run_model_script(table_name, execution_mode):
+    """
+    根据表名查找并执行对应的模型脚本
+    
+    参数:
+        table_name (str): 要处理的表名
+        execution_mode (str): 执行模式 (append/full_refresh)
+    
+    返回:
+        bool: 执行成功返回True，否则返回False
+    """
+    import logging
+    logger = logging.getLogger("airflow.task")
+    
+    # 从Neo4j获取脚本名称
+    script_name = get_script_name_from_neo4j(table_name)
+    if not script_name:
+        logger.error(f"未找到表 {table_name} 的脚本名称，跳过处理")
+        return False
+    
+    logger.info(f"从Neo4j获取到表 {table_name} 的脚本名称: {script_name}")
+    
+    # 检查脚本文件是否存在
+    exists, script_path = check_script_exists(script_name)
+    if not exists:
+        logger.error(f"表 {table_name} 的脚本文件 {script_name} 不存在，跳过处理")
+        return False
+    
+    # 执行脚本
+    logger.info(f"开始执行脚本: {script_path}")
     try:
-        spec = importlib.util.spec_from_file_location("model_script", script_path)
+        # 动态导入模块
+        import importlib.util
+        import sys
+        
+        spec = importlib.util.spec_from_file_location("dynamic_module", script_path)
         module = importlib.util.module_from_spec(spec)
         spec.loader.exec_module(module)
-
+        
+        # 检查并调用标准入口函数run
         if hasattr(module, "run"):
+            logger.info(f"调用脚本 {script_name} 的标准入口函数 run()")
             module.run(table_name=table_name, execution_mode=execution_mode)
+            logger.info(f"脚本 {script_name} 执行成功")
+            return True
         else:
-            print(f"[WARN] 模型脚本 {script_name} 中未定义 run(...) 方法，跳过")
+            logger.error(f"脚本 {script_name} 中未定义标准入口函数 run()，无法执行")
+            return False
     except Exception as e:
-        print(f"[ERROR] 执行模型脚本 {script_name} 时出错: {str(e)}")
-
+        logger.error(f"执行脚本 {script_name} 时出错: {str(e)}")
+        import traceback
+        logger.error(traceback.format_exc())
+        return False
 
 # 从 Neo4j 获取指定 DataModel 表之间的依赖关系图
 # 返回值为 dict：{目标表: [上游依赖表1, 上游依赖表2, ...]}