优化调整auto_excute_tasks代码功能。

app/core/data_flow/dataflows.py

@@ -1,7 +1,9 @@
 import contextlib
 import json
 import logging
+import uuid
 from datetime import datetime
+from pathlib import Path
 from typing import Any, Dict, List, Optional, Union
 
 from sqlalchemy import text
@@ -18,6 +20,9 @@ from app.core.meta_data import get_formatted_time, translate_and_parse
 
 logger = logging.getLogger(__name__)
 
+# 项目根目录
+PROJECT_ROOT = Path(__file__).parent.parent.parent.parent
+
 
 class DataFlowService:
     """数据流服务类，处理数据流相关的业务逻辑"""
@@ -636,6 +641,19 @@ class DataFlowService:
 
                 logger.info(f"成功将任务信息写入task_list表: task_name={script_name}")
 
+                # 自动生成 n8n 工作流 JSON 文件
+                try:
+                    DataFlowService._generate_n8n_workflow(
+                        script_name=script_name,
+                        code_name=code_name,
+                        code_path=code_path,
+                        update_mode=update_mode,
+                        is_import_task=bool(data_source_info),
+                    )
+                except Exception as wf_error:
+                    logger.warning(f"生成n8n工作流文件失败: {str(wf_error)}")
+                    # 不影响主流程
+
             except Exception as task_error:
                 db.session.rollback()
                 logger.error(f"写入task_list表失败: {str(task_error)}")
@@ -646,6 +664,256 @@ class DataFlowService:
             logger.error(f"保存到PG数据库失败: {str(e)}")
             raise e
 
+    @staticmethod
+    def _generate_n8n_workflow(
+        script_name: str,
+        code_name: str,
+        code_path: str,
+        update_mode: str = "append",
+        is_import_task: bool = False,
+    ) -> Optional[str]:
+        """
+        自动生成 n8n 工作流 JSON 文件
+
+        Args:
+            script_name: 脚本/任务名称
+            code_name: 代码文件名
+            code_path: 代码路径
+            update_mode: 更新模式
+            is_import_task: 是否为数据导入任务
+
+        Returns:
+            生成的工作流文件路径，失败返回 None
+        """
+        try:
+            # 确保工作流目录存在
+            workflows_dir = PROJECT_ROOT / "datafactory" / "workflows"
+            workflows_dir.mkdir(parents=True, exist_ok=True)
+
+            # 生成工作流文件名
+            workflow_filename = f"{script_name}_workflow.json"
+            workflow_path = workflows_dir / workflow_filename
+
+            # 生成唯一ID
+            def gen_id():
+                return str(uuid.uuid4())
+
+            # 构建完整的 SSH 命令，包含激活 venv
+            # 注意：由于 n8n 服务器与应用服务器分离，必须使用 SSH 节点
+            ssh_command = (
+                f"cd /opt/dataops-platform && source venv/bin/activate && "
+                f"python {code_path}/{code_name}"
+            )
+
+            workflow_json = {
+                "name": f"{script_name}_工作流",
+                "nodes": [
+                    {
+                        "parameters": {},
+                        "id": gen_id(),
+                        "name": "Manual Trigger",
+                        "type": "n8n-nodes-base.manualTrigger",
+                        "typeVersion": 1,
+                        "position": [250, 300],
+                    },
+                    {
+                        "parameters": {
+                            "resource": "command",
+                            "operation": "execute",
+                            "command": ssh_command,
+                            "cwd": "/opt/dataops-platform",
+                        },
+                        "id": gen_id(),
+                        "name": "Execute Script",
+                        "type": "n8n-nodes-base.ssh",
+                        "typeVersion": 1,
+                        "position": [450, 300],
+                        "credentials": {
+                            "sshPassword": {
+                                "id": "pYTwwuyC15caQe6y",
+                                "name": "SSH Password account",
+                            }
+                        },
+                    },
+                    {
+                        "parameters": {
+                            "conditions": {
+                                "options": {
+                                    "caseSensitive": True,
+                                    "leftValue": "",
+                                    "typeValidation": "strict",
+                                },
+                                "conditions": [
+                                    {
+                                        "id": "condition-success",
+                                        "leftValue": "={{ $json.code }}",
+                                        "rightValue": 0,
+                                        "operator": {
+                                            "type": "number",
+                                            "operation": "equals",
+                                        },
+                                    }
+                                ],
+                                "combinator": "and",
+                            }
+                        },
+                        "id": gen_id(),
+                        "name": "Check Result",
+                        "type": "n8n-nodes-base.if",
+                        "typeVersion": 2,
+                        "position": [650, 300],
+                    },
+                    {
+                        "parameters": {
+                            "assignments": {
+                                "assignments": [
+                                    {
+                                        "id": "result-success",
+                                        "name": "status",
+                                        "value": "success",
+                                        "type": "string",
+                                    },
+                                    {
+                                        "id": "result-message",
+                                        "name": "message",
+                                        "value": f"{script_name} 执行成功",
+                                        "type": "string",
+                                    },
+                                    {
+                                        "id": "result-output",
+                                        "name": "output",
+                                        "value": "={{ $json.stdout }}",
+                                        "type": "string",
+                                    },
+                                    {
+                                        "id": "result-time",
+                                        "name": "executionTime",
+                                        "value": "={{ $now.toISO() }}",
+                                        "type": "string",
+                                    },
+                                ]
+                            }
+                        },
+                        "id": gen_id(),
+                        "name": "Success Response",
+                        "type": "n8n-nodes-base.set",
+                        "typeVersion": 3.4,
+                        "position": [850, 200],
+                    },
+                    {
+                        "parameters": {
+                            "assignments": {
+                                "assignments": [
+                                    {
+                                        "id": "error-status",
+                                        "name": "status",
+                                        "value": "error",
+                                        "type": "string",
+                                    },
+                                    {
+                                        "id": "error-message",
+                                        "name": "message",
+                                        "value": f"{script_name} 执行失败",
+                                        "type": "string",
+                                    },
+                                    {
+                                        "id": "error-output",
+                                        "name": "error",
+                                        "value": "={{ $json.stderr }}",
+                                        "type": "string",
+                                    },
+                                    {
+                                        "id": "error-code",
+                                        "name": "exitCode",
+                                        "value": "={{ $json.code }}",
+                                        "type": "number",
+                                    },
+                                    {
+                                        "id": "error-time",
+                                        "name": "executionTime",
+                                        "value": "={{ $now.toISO() }}",
+                                        "type": "string",
+                                    },
+                                ]
+                            }
+                        },
+                        "id": gen_id(),
+                        "name": "Error Response",
+                        "type": "n8n-nodes-base.set",
+                        "typeVersion": 3.4,
+                        "position": [850, 400],
+                    },
+                ],
+                "connections": {
+                    "Manual Trigger": {
+                        "main": [
+                            [
+                                {
+                                    "node": "Execute Script",
+                                    "type": "main",
+                                    "index": 0,
+                                }
+                            ]
+                        ]
+                    },
+                    "Execute Script": {
+                        "main": [
+                            [
+                                {
+                                    "node": "Check Result",
+                                    "type": "main",
+                                    "index": 0,
+                                }
+                            ]
+                        ]
+                    },
+                    "Check Result": {
+                        "main": [
+                            [
+                                {
+                                    "node": "Success Response",
+                                    "type": "main",
+                                    "index": 0,
+                                }
+                            ],
+                            [
+                                {
+                                    "node": "Error Response",
+                                    "type": "main",
+                                    "index": 0,
+                                }
+                            ],
+                        ]
+                    },
+                },
+                "active": False,
+                "settings": {"executionOrder": "v1"},
+                "versionId": "1",
+                "meta": {
+                    "templateCredsSetupCompleted": False,
+                    "instanceId": "dataops-platform",
+                },
+                "tags": [
+                    {
+                        "createdAt": datetime.now().isoformat() + "Z",
+                        "updatedAt": datetime.now().isoformat() + "Z",
+                        "id": "1",
+                        "name": "数据导入" if is_import_task else "数据流程",
+                    }
+                ],
+            }
+
+            # 写入文件
+            with open(workflow_path, "w", encoding="utf-8") as f:
+                json.dump(workflow_json, f, ensure_ascii=False, indent=2)
+
+            logger.info(f"成功生成n8n工作流文件: {workflow_path}")
+            return str(workflow_path)
+
+        except Exception as e:
+            logger.error(f"生成n8n工作流失败: {str(e)}")
+            return None
+
     @staticmethod
     def _handle_children_relationships(dataflow_node, children_ids):
         """处理子节点关系"""

application.py

@@ -1,6 +0,0 @@
-from app import create_app
-
-app = create_app()
-
-if __name__ == "__main__":
-    app.run(host="0.0.0.0", port=app.config["PORT"])

datafactory/scripts/DF_DO202601130001.py

@@ -0,0 +1,483 @@
+"""
+数据流程脚本：DF_DO202601130001
+仓库库存汇总表 数据流程
+
+功能：
+- 从产品库存表(test_product_inventory)中读取数据
+- 按仓库(warehouse)汇总库存数量(current_stock)
+- 输出仓库编号和总库存数量到目标表(warehouse_inventory_summary)
+- 更新模式：Full Refresh (全量更新)
+
+任务信息：
+- DataFlow ID: 2220
+- DataFlow Name: 仓库库存汇总表_数据流程
+- Order ID: 17
+- Order No: DO202601130001
+
+作者：cursor (自动生成)
+创建时间：2026-01-13
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import os
+import sys
+from datetime import datetime
+from typing import Any
+
+# 添加项目根目录到路径
+sys.path.insert(
+    0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+)
+
+try:
+    from sqlalchemy import create_engine, text
+    from sqlalchemy.orm import sessionmaker
+except ImportError:
+    print("错误：请安装 sqlalchemy 库")
+    sys.exit(1)
+
+# 配置日志
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+class WarehouseInventorySummaryFlow:
+    """仓库库存汇总表数据流程处理器"""
+
+    # 配置常量
+    SOURCE_TABLE = "test_product_inventory"
+    TARGET_TABLE = "warehouse_inventory_summary"
+    SOURCE_SCHEMA = "public"
+    TARGET_SCHEMA = "public"
+    UPDATE_MODE = "full"  # full = 全量更新
+
+    def __init__(self, db_uri: str | None = None):
+        """
+        初始化数据流程处理器
+
+        Args:
+            db_uri: 数据库连接 URI，如果不提供则从配置中获取
+        """
+        self.db_uri = db_uri or self._get_db_uri()
+        self.engine = None
+        self.session = None
+        self.processed_count = 0
+        self.error_count = 0
+
+    def _get_db_uri(self) -> str:
+        """获取数据库连接 URI"""
+        # 优先从环境变量获取
+        db_uri = os.environ.get("DATABASE_URL")
+        if db_uri:
+            return db_uri
+
+        # 尝试从 Flask 配置获取
+        try:
+            from app.config.config import config, get_environment
+
+            env = get_environment()
+            cfg = config.get(env, config["default"])
+            return cfg.SQLALCHEMY_DATABASE_URI
+        except ImportError:
+            pass
+
+        # 默认使用开发环境配置
+        return "postgresql://postgres:postgres@localhost:5432/dataops"
+
+    def connect(self) -> bool:
+        """
+        连接数据库
+
+        Returns:
+            连接是否成功
+        """
+        try:
+            self.engine = create_engine(self.db_uri)
+            Session = sessionmaker(bind=self.engine)
+            self.session = Session()
+
+            # 测试连接
+            with self.engine.connect() as conn:
+                conn.execute(text("SELECT 1"))
+
+            # 隐藏密码显示连接信息
+            safe_uri = self.db_uri.split("@")[-1] if "@" in self.db_uri else self.db_uri
+            logger.info(f"成功连接数据库: {safe_uri}")
+            return True
+
+        except Exception as e:
+            logger.error(f"连接数据库失败: {str(e)}")
+            return False
+
+    def ensure_target_table(self) -> bool:
+        """
+        确保目标表存在，如果不存在则创建
+
+        Returns:
+            操作是否成功
+        """
+        try:
+            if not self.session:
+                logger.error("数据库会话未初始化")
+                return False
+
+            # 检查目标表是否存在
+            check_sql = text("""
+                SELECT EXISTS (
+                    SELECT FROM information_schema.tables
+                    WHERE table_schema = :schema
+                    AND table_name = :table_name
+                )
+            """)
+
+            result = self.session.execute(
+                check_sql,
+                {"schema": self.TARGET_SCHEMA, "table_name": self.TARGET_TABLE},
+            )
+            exists = result.scalar()
+
+            if exists:
+                logger.info(f"目标表 {self.TARGET_SCHEMA}.{self.TARGET_TABLE} 已存在")
+                return True
+
+            # 创建目标表
+            create_sql = text(f"""
+                CREATE TABLE {self.TARGET_SCHEMA}.{self.TARGET_TABLE} (
+                    id SERIAL PRIMARY KEY,
+                    warehouse VARCHAR(100) NOT NULL COMMENT '仓库编号',
+                    total_stock BIGINT NOT NULL DEFAULT 0 COMMENT '总库存数量',
+                    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '数据创建时间',
+                    update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '数据更新时间'
+                );
+                COMMENT ON TABLE {self.TARGET_SCHEMA}.{self.TARGET_TABLE}
+                    IS '仓库库存汇总表';
+                COMMENT ON COLUMN {self.TARGET_SCHEMA}.{self.TARGET_TABLE}.warehouse
+                    IS '仓库编号';
+                COMMENT ON COLUMN {self.TARGET_SCHEMA}.{self.TARGET_TABLE}.total_stock
+                    IS '总库存数量';
+                COMMENT ON COLUMN {self.TARGET_SCHEMA}.{self.TARGET_TABLE}.create_time
+                    IS '数据创建时间';
+                COMMENT ON COLUMN {self.TARGET_SCHEMA}.{self.TARGET_TABLE}.update_time
+                    IS '数据更新时间';
+            """)
+
+            self.session.execute(create_sql)
+            self.session.commit()
+
+            logger.info(f"成功创建目标表: {self.TARGET_SCHEMA}.{self.TARGET_TABLE}")
+            return True
+
+        except Exception as e:
+            if self.session:
+                self.session.rollback()
+            logger.error(f"创建目标表失败: {str(e)}")
+            # 尝试使用简化的 DDL（PostgreSQL 不支持 COMMENT 在列定义中）
+            return self._create_table_simple()
+
+    def _create_table_simple(self) -> bool:
+        """使用简化的 DDL 创建目标表（PostgreSQL 兼容）"""
+        try:
+            if not self.session:
+                return False
+
+            # PostgreSQL 简化的建表语句
+            create_sql = text(f"""
+                CREATE TABLE IF NOT EXISTS {self.TARGET_SCHEMA}.{self.TARGET_TABLE} (
+                    id SERIAL PRIMARY KEY,
+                    warehouse VARCHAR(100) NOT NULL,
+                    total_stock BIGINT NOT NULL DEFAULT 0,
+                    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                )
+            """)
+
+            self.session.execute(create_sql)
+
+            # 添加表注释
+            comment_table_sql = text(f"""
+                COMMENT ON TABLE {self.TARGET_SCHEMA}.{self.TARGET_TABLE}
+                IS '仓库库存汇总表'
+            """)
+            self.session.execute(comment_table_sql)
+
+            # 添加列注释
+            comments = [
+                ("warehouse", "仓库编号"),
+                ("total_stock", "总库存数量"),
+                ("create_time", "数据创建时间"),
+                ("update_time", "数据更新时间"),
+            ]
+            for col_name, col_comment in comments:
+                comment_col_sql = text(f"""
+                    COMMENT ON COLUMN {self.TARGET_SCHEMA}.{self.TARGET_TABLE}.{col_name}
+                    IS '{col_comment}'
+                """)
+                self.session.execute(comment_col_sql)
+
+            self.session.commit()
+            logger.info(
+                f"成功创建目标表(简化模式): {self.TARGET_SCHEMA}.{self.TARGET_TABLE}"
+            )
+            return True
+
+        except Exception as e:
+            if self.session:
+                self.session.rollback()
+            logger.error(f"创建目标表(简化模式)失败: {str(e)}")
+            return False
+
+    def extract_and_transform(self) -> list[dict[str, Any]]:
+        """
+        从源表提取数据并进行转换（按仓库汇总）
+
+        Returns:
+            转换后的数据列表
+        """
+        try:
+            if not self.session:
+                logger.error("数据库会话未初始化")
+                return []
+
+            # 执行汇总查询
+            # 1. 从产品库存表中提取字段：仓库编号、产品编号、库存数量
+            # 2. 对库存数量进行按仓库编号进行求和计算
+            # 3. 无特殊过滤条件
+            # 4. 最终输出数据格式包含字段：仓库编号、总库存数量
+            query_sql = text(f"""
+                SELECT
+                    warehouse,
+                    SUM(current_stock) as total_stock
+                FROM {self.SOURCE_SCHEMA}.{self.SOURCE_TABLE}
+                GROUP BY warehouse
+                ORDER BY warehouse
+            """)
+
+            result = self.session.execute(query_sql)
+            rows = result.fetchall()
+
+            data_list = []
+            for row in rows:
+                data_list.append(
+                    {
+                        "warehouse": row.warehouse,
+                        "total_stock": int(row.total_stock) if row.total_stock else 0,
+                    }
+                )
+
+            logger.info(f"从源表提取并汇总了 {len(data_list)} 条仓库库存记录")
+            return data_list
+
+        except Exception as e:
+            logger.error(f"提取和转换数据失败: {str(e)}")
+            return []
+
+    def load_to_target(self, data_list: list[dict[str, Any]]) -> bool:
+        """
+        将转换后的数据加载到目标表
+
+        Args:
+            data_list: 转换后的数据列表
+
+        Returns:
+            加载是否成功
+        """
+        try:
+            if not data_list:
+                logger.warning("没有数据需要加载")
+                return True
+
+            if not self.session:
+                logger.error("数据库会话未初始化")
+                return False
+
+            # 全量更新模式：先清空目标表
+            if self.UPDATE_MODE == "full":
+                delete_sql = text(
+                    f"DELETE FROM {self.TARGET_SCHEMA}.{self.TARGET_TABLE}"
+                )
+                self.session.execute(delete_sql)
+                logger.info(f"目标表 {self.TARGET_TABLE} 已清空（全量更新模式）")
+
+            # 插入新数据
+            current_time = datetime.now()
+            insert_sql = text(f"""
+                INSERT INTO {self.TARGET_SCHEMA}.{self.TARGET_TABLE}
+                    (warehouse, total_stock, create_time, update_time)
+                VALUES
+                    (:warehouse, :total_stock, :create_time, :update_time)
+            """)
+
+            for data in data_list:
+                try:
+                    self.session.execute(
+                        insert_sql,
+                        {
+                            "warehouse": data["warehouse"],
+                            "total_stock": data["total_stock"],
+                            "create_time": current_time,
+                            "update_time": current_time,
+                        },
+                    )
+                    self.processed_count += 1
+                except Exception as e:
+                    self.error_count += 1
+                    logger.error(f"插入数据失败: {str(e)}, 数据: {data}")
+
+            self.session.commit()
+            logger.info(
+                f"数据加载完成: 成功 {self.processed_count} 条, 失败 {self.error_count} 条"
+            )
+            return True
+
+        except Exception as e:
+            if self.session:
+                self.session.rollback()
+            logger.error(f"加载数据到目标表失败: {str(e)}")
+            return False
+
+    def close(self) -> None:
+        """关闭数据库连接"""
+        if self.session:
+            try:
+                self.session.close()
+                logger.info("数据库会话已关闭")
+            except Exception as e:
+                logger.error(f"关闭数据库会话失败: {str(e)}")
+
+        if self.engine:
+            try:
+                self.engine.dispose()
+                logger.info("数据库引擎已释放")
+            except Exception as e:
+                logger.error(f"释放数据库引擎失败: {str(e)}")
+
+    def run(self) -> dict[str, Any]:
+        """
+        执行完整的 ETL 流程
+
+        Returns:
+            执行结果字典
+        """
+        result = {
+            "success": False,
+            "processed_count": 0,
+            "error_count": 0,
+            "update_mode": self.UPDATE_MODE,
+            "source_table": f"{self.SOURCE_SCHEMA}.{self.SOURCE_TABLE}",
+            "target_table": f"{self.TARGET_SCHEMA}.{self.TARGET_TABLE}",
+            "message": "",
+        }
+
+        try:
+            logger.info("=" * 60)
+            logger.info("开始执行数据流程: DF_DO202601130001")
+            logger.info(f"源表: {self.SOURCE_SCHEMA}.{self.SOURCE_TABLE}")
+            logger.info(f"目标表: {self.TARGET_SCHEMA}.{self.TARGET_TABLE}")
+            logger.info(f"更新模式: {self.UPDATE_MODE}")
+            logger.info("=" * 60)
+
+            # 1. 连接数据库
+            if not self.connect():
+                result["message"] = "连接数据库失败"
+                return result
+
+            # 2. 确保目标表存在
+            if not self.ensure_target_table():
+                result["message"] = "创建目标表失败"
+                return result
+
+            # 3. 提取和转换数据
+            data_list = self.extract_and_transform()
+
+            if not data_list:
+                result["message"] = "未提取到数据"
+                result["success"] = True  # 没有数据不算失败
+                return result
+
+            # 4. 加载到目标表
+            if self.load_to_target(data_list):
+                result["success"] = True
+                result["processed_count"] = self.processed_count
+                result["error_count"] = self.error_count
+                result["message"] = (
+                    f"数据流程执行成功: "
+                    f"处理 {self.processed_count} 条, 失败 {self.error_count} 条"
+                )
+            else:
+                result["message"] = "加载数据到目标表失败"
+
+        except Exception as e:
+            logger.error(f"数据流程执行异常: {str(e)}")
+            result["message"] = f"数据流程执行异常: {str(e)}"
+
+        finally:
+            self.close()
+
+        logger.info("=" * 60)
+        logger.info(f"执行结果: {result['message']}")
+        logger.info("=" * 60)
+
+        return result
+
+
+def main():
+    """主函数"""
+    parser = argparse.ArgumentParser(
+        description="DF_DO202601130001 - 仓库库存汇总表数据流程"
+    )
+    parser.add_argument(
+        "--db-uri",
+        type=str,
+        default=None,
+        help="数据库连接 URI (可选，默认从配置获取)",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="仅测试连接和查询，不执行写入",
+    )
+
+    args = parser.parse_args()
+
+    # 创建并执行数据流程
+    flow = WarehouseInventorySummaryFlow(db_uri=args.db_uri)
+
+    if args.dry_run:
+        logger.info("Dry-run 模式: 仅测试连接和查询")
+        if flow.connect():
+            data_list = flow.extract_and_transform()
+            logger.info(f"预览数据 ({len(data_list)} 条):")
+            for data in data_list:
+                logger.info(f"  {data}")
+            flow.close()
+            print("\nDry-run 完成，未执行写入操作")
+            sys.exit(0)
+        else:
+            print("\n连接失败")
+            sys.exit(1)
+
+    result = flow.run()
+
+    # 输出结果
+    print("\n" + "=" * 60)
+    print(f"数据流程执行结果: {'成功' if result['success'] else '失败'}")
+    print(f"消息: {result['message']}")
+    print(f"处理记录数: {result['processed_count']}")
+    print(f"失败记录数: {result['error_count']}")
+    print(f"更新模式: {result['update_mode']}")
+    print(f"源表: {result['source_table']}")
+    print(f"目标表: {result['target_table']}")
+    print("=" * 60)
+
+    # 设置退出代码
+    sys.exit(0 if result["success"] else 1)
+
+
+if __name__ == "__main__":
+    main()

datafactory/scripts/import_resource_data.py

@@ -29,10 +29,14 @@ sys.path.insert(
 )
 
 try:
-    from app.config.config import Config  # type: ignore
+    from app.config.config import config, get_environment  # type: ignore
+
+    # 获取当前环境的配置类
+    _current_env = get_environment()
+    Config = config.get(_current_env, config["default"])
 except ImportError:
     # 如果无法导入，使用环境变量
-    class Config:
+    class Config:  # type: ignore
         SQLALCHEMY_DATABASE_URI = os.environ.get(
             "DATABASE_URI", "postgresql://user:password@localhost:5432/database"
         )
@@ -56,6 +60,9 @@ logger = logging.getLogger(__name__)
 class ResourceDataImporter:
     """数据资源导入器"""
 
+    # 目标表所在的 schema
+    TARGET_SCHEMA = "dags"
+
     def __init__(
         self,
         source_config: Dict[str, Any],
@@ -98,7 +105,7 @@ class ResourceDataImporter:
             )
 
         logger.info(
-            f"初始化数据导入器: 目标表={target_table_name}, 更新模式={update_mode}"
+            f"初始化数据导入器: 目标表={self.TARGET_SCHEMA}.{target_table_name}, 更新模式={update_mode}"
         )
 
     def connect_target_database(self) -> bool:
@@ -179,6 +186,15 @@ class ResourceDataImporter:
             logger.error(f"连接源数据库失败: {str(e)}")
             return False
 
+    def get_full_table_name(self) -> str:
+        """
+        获取带 schema 的完整表名
+
+        Returns:
+            完整表名 (schema.table_name)
+        """
+        return f"{self.TARGET_SCHEMA}.{self.target_table_name}"
+
     def get_target_table_columns(self) -> List[str]:
         """
         获取目标表的列名
@@ -192,12 +208,15 @@ class ResourceDataImporter:
                 return []
 
             inspector = inspect(self.target_engine)
-            columns = inspector.get_columns(self.target_table_name)
+            # 指定 schema 来获取表的列名
+            columns = inspector.get_columns(
+                self.target_table_name, schema=self.TARGET_SCHEMA
+            )
             column_names = [
                 col["name"] for col in columns if col["name"] != "create_time"
             ]
 
-            logger.info(f"目标表 {self.target_table_name} 的列: {column_names}")
+            logger.info(f"目标表 {self.get_full_table_name()} 的列: {column_names}")
             return column_names
 
         except Exception as e:
@@ -276,11 +295,12 @@ class ResourceDataImporter:
                 logger.error("目标数据库会话未初始化")
                 return False
 
-            delete_sql = text(f"DELETE FROM {self.target_table_name}")
+            full_table_name = self.get_full_table_name()
+            delete_sql = text(f"DELETE FROM {full_table_name}")
             self.target_session.execute(delete_sql)
             self.target_session.commit()
 
-            logger.info(f"目标表 {self.target_table_name} 已清空")
+            logger.info(f"目标表 {full_table_name} 已清空")
             return True
 
         except Exception as e:
@@ -343,18 +363,18 @@ class ResourceDataImporter:
                 return False
 
             # 全量更新模式：先清空目标表
-            if self.update_mode == "full":
-                if not self.clear_target_table():
-                    return False
+            if self.update_mode == "full" and not self.clear_target_table():
+                return False
 
-            # 构建插入 SQL
+            # 构建插入 SQL（使用带 schema 的完整表名）
+            full_table_name = self.get_full_table_name()
             columns_str = ", ".join(target_columns + ["create_time"])
             placeholders = ", ".join(
                 [f":{col}" for col in target_columns] + ["CURRENT_TIMESTAMP"]
             )
 
             insert_sql = text(f"""
-                INSERT INTO {self.target_table_name} ({columns_str})
+                INSERT INTO {full_table_name} ({columns_str})
                 VALUES ({placeholders})
             """)
 
@@ -442,7 +462,7 @@ class ResourceDataImporter:
             logger.info("=" * 60)
             logger.info("开始数据导入")
             logger.info(f"源表: {self.source_config.get('table_name')}")
-            logger.info(f"目标表: {self.target_table_name}")
+            logger.info(f"目标表: {self.get_full_table_name()}")
             logger.info(f"更新模式: {self.update_mode}")
             logger.info("=" * 60)
 
@@ -565,7 +585,7 @@ if __name__ == "__main__":
     except json.JSONDecodeError:
         # 尝试作为文件路径读取
         try:
-            with open(args.source_config, "r", encoding="utf-8") as f:
+            with open(args.source_config, encoding="utf-8") as f:
                 source_config = json.load(f)
         except Exception as e:
             logger.error(f"解析源数据库配置失败: {str(e)}")

datafactory/workflows/DF_DO202601130001_workflow.json

@@ -0,0 +1,119 @@
+{
+  "name": "DF_DO202601130001_仓库库存汇总表",
+  "nodes": [
+    {
+      "parameters": {},
+      "id": "trigger-001",
+      "name": "Manual Trigger",
+      "type": "n8n-nodes-base.manualTrigger",
+      "typeVersion": 1,
+      "position": [250, 300]
+    },
+    {
+      "parameters": {
+        "command": "=cd {{ $env.DATAOPS_PROJECT_ROOT || '/opt/dataops-platform' }} && python datafactory/scripts/DF_DO202601130001.py"
+      },
+      "id": "exec-001",
+      "name": "Execute DataFlow Script",
+      "type": "n8n-nodes-base.executeCommand",
+      "typeVersion": 1,
+      "position": [450, 300]
+    },
+    {
+      "parameters": {
+        "conditions": {
+          "boolean": [
+            {
+              "value1": "={{ $json.exitCode === 0 }}",
+              "value2": true
+            }
+          ]
+        }
+      },
+      "id": "check-001",
+      "name": "Check Result",
+      "type": "n8n-nodes-base.if",
+      "typeVersion": 1,
+      "position": [650, 300]
+    },
+    {
+      "parameters": {
+        "functionCode": "return [{ json: { success: true, message: '仓库库存汇总表数据流程执行成功', workflow: 'DF_DO202601130001' } }];"
+      },
+      "id": "success-001",
+      "name": "Success Response",
+      "type": "n8n-nodes-base.function",
+      "typeVersion": 1,
+      "position": [850, 200]
+    },
+    {
+      "parameters": {
+        "functionCode": "return [{ json: { success: false, message: '仓库库存汇总表数据流程执行失败', error: $json.stderr || $json.stdout, workflow: 'DF_DO202601130001' } }];"
+      },
+      "id": "error-001",
+      "name": "Error Response",
+      "type": "n8n-nodes-base.function",
+      "typeVersion": 1,
+      "position": [850, 400]
+    }
+  ],
+  "connections": {
+    "Manual Trigger": {
+      "main": [
+        [
+          {
+            "node": "Execute DataFlow Script",
+            "type": "main",
+            "index": 0
+          }
+        ]
+      ]
+    },
+    "Execute DataFlow Script": {
+      "main": [
+        [
+          {
+            "node": "Check Result",
+            "type": "main",
+            "index": 0
+          }
+        ]
+      ]
+    },
+    "Check Result": {
+      "main": [
+        [
+          {
+            "node": "Success Response",
+            "type": "main",
+            "index": 0
+          }
+        ],
+        [
+          {
+            "node": "Error Response",
+            "type": "main",
+            "index": 0
+          }
+        ]
+      ]
+    }
+  },
+  "active": false,
+  "settings": {
+    "executionOrder": "v1"
+  },
+  "versionId": "1",
+  "meta": {
+    "templateCredsSetupCompleted": false,
+    "instanceId": "dataops-platform"
+  },
+  "tags": [
+    {
+      "createdAt": "2026-01-13T19:30:00.000Z",
+      "updatedAt": "2026-01-13T19:30:00.000Z",
+      "id": "2",
+      "name": "数据流程"
+    }
+  ]
+}

datafactory/workflows/import_product_inventory_workflow.json

@@ -11,13 +11,85 @@
     },
     {
       "parameters": {
-        "command": "=python \"{{ $env.DATAOPS_PROJECT_ROOT || 'g:\\\\code-lab\\\\DataOps-platform-new' }}\\\\datafactory\\\\scripts\\\\import_resource_data.py\" --source-config '{\"type\":\"postgresql\",\"host\":\"192.168.3.143\",\"port\":5432,\"database\":\"dataops\",\"username\":\"postgres\",\"password\":\"YOUR_PASSWORD_HERE\",\"table_name\":\"test_product_inventory\"}' --target-table test_product_inventory --update-mode append"
+        "resource": "command",
+        "operation": "execute",
+        "command": "cd /opt/dataops-platform && source venv/bin/activate && python datafactory/scripts/import_resource_data.py --source-config '{\"type\":\"postgresql\",\"host\":\"192.168.3.143\",\"port\":5432,\"database\":\"dataops\",\"username\":\"postgres\",\"password\":\"dataOps\",\"table_name\":\"test_product_inventory\"}' --target-table test_product_inventory --update-mode append",
+        "cwd": "/opt/dataops-platform"
       },
       "id": "a1b2c3d4-5e6f-7g8h-9i0j-1k2l3m4n5o6p",
       "name": "Execute Import Script",
-      "type": "n8n-nodes-base.executeCommand",
+      "type": "n8n-nodes-base.ssh",
       "typeVersion": 1,
-      "position": [450, 300]
+      "position": [450, 300],
+      "credentials": {
+        "sshPassword": {
+          "id": "pYTwwuyC15caQe6y",
+          "name": "SSH Password account"
+        }
+      }
+    },
+    {
+      "parameters": {
+        "conditions": {
+          "options": {
+            "caseSensitive": true,
+            "leftValue": "",
+            "typeValidation": "strict"
+          },
+          "conditions": [
+            {
+              "id": "condition-success",
+              "leftValue": "={{ $json.code }}",
+              "rightValue": 0,
+              "operator": {
+                "type": "number",
+                "operation": "equals"
+              }
+            }
+          ],
+          "combinator": "and"
+        }
+      },
+      "id": "check-result-001",
+      "name": "Check Result",
+      "type": "n8n-nodes-base.if",
+      "typeVersion": 2,
+      "position": [650, 300]
+    },
+    {
+      "parameters": {
+        "assignments": {
+          "assignments": [
+            {"id": "result-success", "name": "status", "value": "success", "type": "string"},
+            {"id": "result-message", "name": "message", "value": "产品库存表数据导入成功", "type": "string"},
+            {"id": "result-output", "name": "output", "value": "={{ $json.stdout }}", "type": "string"},
+            {"id": "result-time", "name": "executionTime", "value": "={{ $now.toISO() }}", "type": "string"}
+          ]
+        }
+      },
+      "id": "success-response-001",
+      "name": "Success Response",
+      "type": "n8n-nodes-base.set",
+      "typeVersion": 3.4,
+      "position": [850, 200]
+    },
+    {
+      "parameters": {
+        "assignments": {
+          "assignments": [
+            {"id": "error-status", "name": "status", "value": "error", "type": "string"},
+            {"id": "error-message", "name": "message", "value": "产品库存表数据导入失败", "type": "string"},
+            {"id": "error-output", "name": "error", "value": "={{ $json.stderr }}", "type": "string"},
+            {"id": "error-code", "name": "exitCode", "value": "={{ $json.code }}", "type": "number"},
+            {"id": "error-time", "name": "executionTime", "value": "={{ $now.toISO() }}", "type": "string"}
+          ]
+        }
+      },
+      "id": "error-response-001",
+      "name": "Error Response",
+      "type": "n8n-nodes-base.set",
+      "typeVersion": 3.4,
+      "position": [850, 400]
     }
   ],
   "connections": {
@@ -31,21 +103,50 @@
           }
         ]
       ]
+    },
+    "Execute Import Script": {
+      "main": [
+        [
+          {
+            "node": "Check Result",
+            "type": "main",
+            "index": 0
+          }
+        ]
+      ]
+    },
+    "Check Result": {
+      "main": [
+        [
+          {
+            "node": "Success Response",
+            "type": "main",
+            "index": 0
+          }
+        ],
+        [
+          {
+            "node": "Error Response",
+            "type": "main",
+            "index": 0
+          }
+        ]
+      ]
     }
   },
   "active": false,
   "settings": {
     "executionOrder": "v1"
   },
-  "versionId": "1",
+  "versionId": "3",
   "meta": {
-    "templateCredsSetupCompleted": false,
+    "templateCredsSetupCompleted": true,
     "instanceId": "dataops-platform"
   },
   "tags": [
     {
       "createdAt": "2026-01-07T08:35:00.000Z",
-      "updatedAt": "2026-01-07T08:35:00.000Z",
+      "updatedAt": "2026-01-13T20:00:00.000Z",
       "id": "1",
       "name": "数据导入"
     }

datafactory/创建工作流程.txt

@@ -1,21 +0,0 @@
-接口：/api/dataflow/update-dataflow/258
-
-参数：{
-  "name_zh": "测试用销售数据",
-  "name_en": "try_sales_data",
-  "category": "应用类",
-  "leader": "system",
-  "organization": "citu",
-  "script_type": "sql",
-  "update_mode": "append",
-  "frequency": "月",
-  "tag": [],
-  "status": "active",
-  "script_requirement": {
-    "rule": null,
-    "source_table": [],
-    "target_table": [
-      241
-    ]
-  }
-}

run_project.bat

@@ -1,91 +0,0 @@
-@echo off
-echo ========================================
-echo DataOps-platform 项目启动器
-echo ========================================
-echo.
-
-REM 检查Python是否安装
-python --version >nul 2>&1
-if errorlevel 1 (
-    echo 错误: 未找到Python，请先安装Python 3.8+
-    pause
-    exit /b 1
-)
-
-echo Python版本检查通过
-echo.
-
-REM 检查项目文件完整性
-if not exist "pyproject.toml" (
-    echo 警告: 缺少pyproject.toml文件，Cursor可能无法正确识别项目
-)
-
-if not exist "setup.py" (
-    echo 警告: 缺少setup.py文件，可能影响项目识别
-)
-
-if not exist ".cursorrules" (
-    echo 警告: 缺少.cursorrules文件，Cursor编辑器配置不完整
-)
-
-echo.
-echo 项目文件检查完成
-echo.
-
-REM 检查虚拟环境
-if not exist "venv" (
-    echo 创建虚拟环境...
-    python -m venv venv
-    if errorlevel 1 (
-        echo 错误: 创建虚拟环境失败
-        pause
-        exit /b 1
-    )
-    echo 虚拟环境创建成功
-) else (
-    echo 虚拟环境已存在
-)
-
-REM 激活虚拟环境
-echo 激活虚拟环境...
-call venv\Scripts\activate.bat
-
-REM 升级pip
-echo 升级pip...
-python -m pip install --upgrade pip
-
-REM 安装依赖
-echo 安装项目依赖...
-pip install -r requirements.txt
-if errorlevel 1 (
-    echo 错误: 依赖安装失败
-    pause
-    exit /b 1
-)
-
-REM 设置环境变量
-set FLASK_ENV=development
-set FLASK_APP=application.py
-set FLASK_DEBUG=1
-
-echo.
-echo ========================================
-echo 环境设置完成！
-echo ========================================
-echo.
-echo 现在您可以：
-echo 1. 在Cursor中正常打开项目
-echo 2. 使用Python解释器: venv\Scripts\python.exe
-echo 3. 项目将在 http://localhost:5500 运行
-echo.
-echo 按任意键启动应用...
-pause >nul
-
-REM 启动应用
-echo 启动Flask应用...
-echo 访问地址: http://localhost:5500
-echo 按 Ctrl+C 停止应用
-echo.
-python application.py
-
-pause

run_project.py

@@ -1,119 +0,0 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""
-DataOps-platform 项目启动脚本
-"""
-
-import os
-import sys
-import subprocess
-import platform
-
-def check_python_version():
-    """检查Python版本"""
-    if sys.version_info < (3, 8):
-        print("错误: 需要Python 3.8或更高版本")
-        print(f"当前版本: {sys.version}")
-        return False
-    print(f"Python版本: {sys.version}")
-    return True
-
-def create_virtual_env():
-    """创建虚拟环境"""
-    if not os.path.exists("venv"):
-        print("创建虚拟环境...")
-        try:
-            subprocess.run([sys.executable, "-m", "venv", "venv"], check=True)
-            print("虚拟环境创建成功")
-        except subprocess.CalledProcessError:
-            print("错误: 创建虚拟环境失败")
-            return False
-    else:
-        print("虚拟环境已存在")
-    return True
-
-def activate_virtual_env():
-    """激活虚拟环境"""
-    if platform.system() == "Windows":
-        activate_script = os.path.join("venv", "Scripts", "activate.bat")
-        if os.path.exists(activate_script):
-            print("激活虚拟环境...")
-            # 在Windows上，我们需要使用cmd来运行activate.bat
-            os.system(f'cmd /c "{activate_script} && pip install -r requirements.txt"')
-        else:
-            print("错误: 虚拟环境激活脚本未找到")
-            return False
-    else:
-        activate_script = os.path.join("venv", "bin", "activate")
-        if os.path.exists(activate_script):
-            print("激活虚拟环境...")
-            os.system(f"source {activate_script} && pip install -r requirements.txt")
-        else:
-            print("错误: 虚拟环境激活脚本未找到")
-            return False
-    return True
-
-def install_dependencies():
-    """安装项目依赖"""
-    print("安装项目依赖...")
-    try:
-        if platform.system() == "Windows":
-            subprocess.run([os.path.join("venv", "Scripts", "python.exe"), "-m", "pip", "install", "-r", "requirements.txt"], check=True)
-        else:
-            subprocess.run([os.path.join("venv", "bin", "python"), "-m", "pip", "install", "-r", "requirements.txt"], check=True)
-        print("依赖安装成功")
-        return True
-    except subprocess.CalledProcessError:
-        print("错误: 依赖安装失败")
-        return False
-
-def set_environment_variables():
-    """设置环境变量"""
-    os.environ["FLASK_ENV"] = "development"
-    os.environ["FLASK_APP"] = "application.py"
-    os.environ["FLASK_DEBUG"] = "1"
-    print("环境变量设置完成")
-
-def run_application():
-    """运行应用"""
-    print("启动DataOps-platform...")
-    print("访问地址: http://localhost:5500")
-    print("按 Ctrl+C 停止应用")
-    print("-" * 50)
-    
-    try:
-        if platform.system() == "Windows":
-            subprocess.run([os.path.join("venv", "Scripts", "python.exe"), "application.py"])
-        else:
-            subprocess.run([os.path.join("venv", "bin", "python"), "application.py"])
-    except KeyboardInterrupt:
-        print("\n应用已停止")
-    except Exception as e:
-        print(f"启动应用时出错: {e}")
-
-def main():
-    """主函数"""
-    print("=" * 50)
-    print("DataOps-platform 项目启动器")
-    print("=" * 50)
-    
-    # 检查Python版本
-    if not check_python_version():
-        return
-    
-    # 创建虚拟环境
-    if not create_virtual_env():
-        return
-    
-    # 安装依赖
-    if not install_dependencies():
-        return
-    
-    # 设置环境变量
-    set_environment_variables()
-    
-    # 运行应用
-    run_application()
-
-if __name__ == "__main__":
-    main()

scripts/create_test_tables_direct.sql

@@ -1,65 +0,0 @@
--- 直接创建测试数据表的 SQL 脚本
--- 用于在生产环境数据库中创建测试表
-
--- 表1: 销售数据表
-DROP TABLE IF EXISTS test_sales_data CASCADE;
-
-CREATE TABLE test_sales_data (
-    id SERIAL PRIMARY KEY,
-    order_id VARCHAR(50) NOT NULL,
-    customer_name VARCHAR(100),
-    product_name VARCHAR(200),
-    quantity INTEGER,
-    unit_price DECIMAL(10, 2),
-    total_amount DECIMAL(10, 2),
-    order_date DATE,
-    region VARCHAR(50),
-    status VARCHAR(20),
-    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-);
-
-COMMENT ON TABLE test_sales_data IS '测试销售数据表';
-
--- 表2: 用户统计表
-DROP TABLE IF EXISTS test_user_statistics CASCADE;
-
-CREATE TABLE test_user_statistics (
-    id SERIAL PRIMARY KEY,
-    user_id VARCHAR(50) NOT NULL,
-    username VARCHAR(100),
-    email VARCHAR(200),
-    registration_date DATE,
-    last_login_date DATE,
-    total_orders INTEGER DEFAULT 0,
-    total_amount DECIMAL(10, 2) DEFAULT 0,
-    user_level VARCHAR(20),
-    is_active BOOLEAN DEFAULT TRUE,
-    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-);
-
-COMMENT ON TABLE test_user_statistics IS '测试用户统计表';
-
--- 表3: 产品库存表
-DROP TABLE IF EXISTS test_product_inventory CASCADE;
-
-CREATE TABLE test_product_inventory (
-    id SERIAL PRIMARY KEY,
-    product_code VARCHAR(50) UNIQUE NOT NULL,
-    product_name VARCHAR(200),
-    category VARCHAR(100),
-    current_stock INTEGER,
-    min_stock INTEGER,
-    max_stock INTEGER,
-    unit_price DECIMAL(10, 2),
-    supplier VARCHAR(200),
-    last_restock_date DATE,
-    status VARCHAR(20),
-    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-);
-
-COMMENT ON TABLE test_product_inventory IS '测试产品库存表';
-
-
-
-
-

scripts/start_task_scheduler.bat

@@ -1,10 +1,10 @@
 @echo off
 chcp 65001 >nul
 REM ============================================================
-REM 自动任务调度脚本启动器
+REM 自动任务调度脚本启动器 (Agent 模式)
 REM ============================================================
 REM 功能：启动核心任务调度脚本 auto_execute_tasks.py
-REM 支持前台运行、后台运行、Agent模式、自动部署等多种模式
+REM 支持 Agent 单次执行、循环模式、部署管理等功能
 REM ============================================================
 
 setlocal enabledelayedexpansion
@@ -14,7 +14,7 @@ cd /d %~dp0..
 
 echo.
 echo ========================================================
-echo           自动任务调度脚本启动器 v2.0
+echo           自动任务调度脚本启动器 (Agent 模式)
 echo ========================================================
 echo.
 
@@ -50,46 +50,32 @@ echo ========================================================
 echo                    请选择运行模式
 echo ========================================================
 echo.
-echo   【基础模式】
-echo    1. 前台运行 (实时日志)
-echo    2. 后台运行 (日志写入文件)
-echo    3. 单次执行 (执行一次后退出)
+echo   【Agent 模式】
+echo    1. Agent 单次执行 (执行一次任务后退出)
+echo    2. Agent 循环模式 (持续监听任务)
+echo    3. Agent 循环模式 + 禁用自动部署
 echo.
-echo   【Agent 自动化模式】(推荐)
-echo    4. Agent 循环模式 (自动启动/关闭 Agent)
-echo    5. Agent 单次执行 (执行一次任务)
-echo    6. Agent 循环模式 + 禁用自动部署
+echo   【部署管理】
+echo    4. 测试生产服务器连接
+echo    5. 立即部署指定任务
 echo.
-echo   【传统 Chat 模式】
-echo    7. Chat 循环模式 (定期发送消息)
-echo    8. 立即发送 Chat 消息
-echo.
-echo   【部署功能】
-echo    9. 测试生产服务器连接
-echo   10. 立即部署指定任务
-echo.
-echo   【管理功能】
-echo   11. 查看服务状态
-echo   12. 停止后台服务
+echo   【系统管理】
+echo    6. 查看服务状态
+echo    7. 停止后台服务
 echo.
 echo    0. 退出
 echo ========================================================
 echo.
 
-set /p choice="请输入选择 [0-12]: "
+set /p choice="请输入选择 [0-7]: "
 
-if "%choice%"=="1" goto :run_foreground
-if "%choice%"=="2" goto :run_background
-if "%choice%"=="3" goto :run_once
-if "%choice%"=="4" goto :run_agent_loop
-if "%choice%"=="5" goto :run_agent_once
-if "%choice%"=="6" goto :run_agent_no_deploy
-if "%choice%"=="7" goto :run_chat_loop
-if "%choice%"=="8" goto :send_chat_now
-if "%choice%"=="9" goto :test_connection
-if "%choice%"=="10" goto :deploy_now
-if "%choice%"=="11" goto :check_status
-if "%choice%"=="12" goto :stop_service
+if "%choice%"=="1" goto :run_agent_once
+if "%choice%"=="2" goto :run_agent_loop
+if "%choice%"=="3" goto :run_agent_no_deploy
+if "%choice%"=="4" goto :test_connection
+if "%choice%"=="5" goto :deploy_now
+if "%choice%"=="6" goto :check_status
+if "%choice%"=="7" goto :stop_service
 if "%choice%"=="0" goto :exit
 
 echo [错误] 无效的选择，请重新运行
@@ -97,78 +83,15 @@ pause
 exit /b 1
 
 REM ============================================================
-REM 基础模式
-REM ============================================================
-
-:run_foreground
-echo.
-echo ========================================================
-echo                   前台运行模式
-echo ========================================================
-echo [启动] 检查间隔: 5分钟
-echo [提示] 按 Ctrl+C 可停止服务
-echo ========================================================
-echo.
-python scripts\auto_execute_tasks.py --interval 300
-pause
-goto :exit
-
-:run_background
-echo.
-echo ========================================================
-echo                   后台运行模式
-echo ========================================================
-echo [启动] 检查间隔: 5分钟
-echo [信息] 日志输出到: logs\auto_execute.log
-echo ========================================================
-start /B "" python scripts\auto_execute_tasks.py --interval 300 > logs\auto_execute.log 2>&1
-echo.
-echo [成功] 服务已在后台启动！
-echo.
-echo [提示] 相关命令:
-echo   - 查看日志: type logs\auto_execute.log
-echo   - 停止服务: 再次运行此脚本选择 12
-echo.
-pause
-goto :exit
-
-:run_once
-echo.
-echo ========================================================
-echo                   单次执行模式
-echo ========================================================
-echo [执行] 检查一次 pending 任务后退出
-echo ========================================================
-echo.
-python scripts\auto_execute_tasks.py --once
-echo.
-pause
-goto :exit
-
-REM ============================================================
-REM Agent 自动化模式 (推荐)
+REM Agent 模式
 REM ============================================================
 
-:run_agent_loop
-echo.
-echo ========================================================
-echo              Agent 循环模式 (推荐)
-echo ========================================================
-echo [功能] 自动启动/关闭 Agent，自动部署到生产服务器
-echo [间隔] 任务检查: 60秒, Agent超时: 3600秒
-echo [提示] 按 Ctrl+C 可停止服务
-echo ========================================================
-echo.
-python scripts\auto_execute_tasks.py --chat-loop --use-agent
-pause
-goto :exit
-
 :run_agent_once
 echo.
 echo ========================================================
 echo              Agent 单次执行模式
 echo ========================================================
-echo [功能] 启动 Agent 执行一次任务后退出
+echo [功能] 执行一次任务后自动退出
 echo [超时] 3600秒 (1小时)
 echo ========================================================
 echo.
@@ -176,59 +99,38 @@ python scripts\auto_execute_tasks.py --agent-run
 pause
 goto :exit
 
-:run_agent_no_deploy
+:run_agent_loop
 echo.
 echo ========================================================
-echo         Agent 循环模式 (禁用自动部署)
+echo              Agent 循环模式
 echo ========================================================
-echo [功能] 自动启动/关闭 Agent，但不自动部署
-echo [间隔] 任务检查: 60秒, Agent超时: 3600秒
+echo [功能] 持续监听任务，自动启动/关闭 Agent
+echo [间隔] 任务检查: 300秒
+echo [部署] 自动部署已启用
 echo [提示] 按 Ctrl+C 可停止服务
 echo ========================================================
 echo.
-python scripts\auto_execute_tasks.py --chat-loop --use-agent --no-deploy
-pause
-goto :exit
-
-REM ============================================================
-REM 传统 Chat 模式
-REM ============================================================
-
-:run_chat_loop
-echo.
-echo ========================================================
-echo               Chat 循环模式 (传统)
-echo ========================================================
-echo [功能] 定期发送 Chat 消息提醒执行任务
-echo [间隔] 60秒
-echo [提示] 需要手动在 Cursor 中响应
-echo ========================================================
-echo.
-set /p chat_pos="请输入 Chat 输入框位置 (格式: x,y，直接回车使用默认): "
-echo.
-if "%chat_pos%"=="" (
-    python scripts\auto_execute_tasks.py --chat-loop --no-agent
-) else (
-    python scripts\auto_execute_tasks.py --chat-loop --no-agent --chat-input-pos "%chat_pos%"
-)
+python scripts\auto_execute_tasks.py --agent-loop
 pause
 goto :exit
 
-:send_chat_now
+:run_agent_no_deploy
 echo.
 echo ========================================================
-echo               立即发送 Chat 消息
+echo         Agent 循环模式 (禁用自动部署)
 echo ========================================================
-echo [功能] 立即发送一次 Chat 消息到 Cursor
+echo [功能] 持续监听任务，自动启动/关闭 Agent
+echo [间隔] 任务检查: 300秒
+echo [部署] 自动部署已禁用
+echo [提示] 按 Ctrl+C 可停止服务
 echo ========================================================
 echo.
-python scripts\auto_execute_tasks.py --send-chat-now
-echo.
+python scripts\auto_execute_tasks.py --agent-loop --no-deploy
 pause
 goto :exit
 
 REM ============================================================
-REM 部署功能
+REM 部署管理
 REM ============================================================
 
 :test_connection
@@ -266,7 +168,7 @@ pause
 goto :exit
 
 REM ============================================================
-REM 管理功能
+REM 系统管理
 REM ============================================================
 
 :check_status

setup.py

@@ -1,56 +0,0 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""
-DataOps Platform Setup
-"""
-
-from setuptools import setup, find_packages
-
-# 读取README文件
-with open("README.md", "r", encoding="utf-8") as fh:
-    long_description = fh.read()
-
-# 读取requirements.txt
-with open("requirements.txt", "r", encoding="utf-8") as fh:
-    requirements = [line.strip() for line in fh if line.strip() and not line.startswith("#")]
-
-setup(
-    name="dataops-platform",
-    version="1.0.0",
-    author="DataOps Team",
-    author_email="team@dataops.com",
-    description="DataOps Platform - 数据运营平台",
-    long_description=long_description,
-    long_description_content_type="text/markdown",
-    url="https://github.com/dataops/dataops-platform",
-    packages=find_packages(),
-    classifiers=[
-        "Development Status :: 4 - Beta",
-        "Intended Audience :: Developers",
-        "License :: OSI Approved :: MIT License",
-        "Operating System :: OS Independent",
-        "Programming Language :: Python :: 3",
-        "Programming Language :: Python :: 3.8",
-        "Programming Language :: Python :: 3.9",
-        "Programming Language :: Python :: 3.10",
-        "Programming Language :: Python :: 3.11",
-        "Programming Language :: Python :: 3.12",
-    ],
-    python_requires=">=3.8",
-    install_requires=requirements,
-    extras_require={
-        "dev": [
-            "pytest>=7.4.0",
-            "black>=23.11.0",
-            "flake8>=6.1.0",
-            "mypy>=1.7.0",
-        ],
-    },
-    entry_points={
-        "console_scripts": [
-            "dataops=application:main",
-        ],
-    },
-    include_package_data=True,
-    zip_safe=False,
-)

tasks/task_execute_instructions.md

@@ -1,38 +1,38 @@
-# 🤖 Cursor 自动任务执行指令
+# Cursor 自动任务执行指令
 
-**⚠️ 重要：请立即执行以下任务！**
+**重要：请立即执行以下任务！**
 
-**生成时间**: 2026-01-07 16:35:20
+**生成时间**: 2026-01-13 19:15:13
 
-**待执行任务数量**: 1
+**待执行任务数量**: 2
 
-## 📋 任务完成后的操作
+## 任务完成后的操作
 
 完成每个任务后，请更新 `tasks/pending_tasks.json` 中对应任务的 `status` 为 `completed`，
 并填写 `code_name`（代码文件名）和 `code_path`（代码路径）。
 
 调度脚本会自动将完成的任务同步到数据库。
 
-## ⚠️ 任务约束要求
+## 任务约束要求
 
 **重要约束**：完成脚本创建后，**不需要生成任务总结文件**。
 
-- ❌ 不要创建任何 summary、report、总结类的文档文件
-- ❌ 不要生成 task_summary.md、execution_report.md 等总结文件
-- ✅ 只需创建任务要求的功能脚本文件
-- ✅ 只需更新 `tasks/pending_tasks.json` 中的任务状态
+- 不要创建任何 summary、report、总结类的文档文件
+- 不要生成 task_summary.md、execution_report.md 等总结文件
+- 只需创建任务要求的功能脚本文件
+- 只需更新 `tasks/pending_tasks.json` 中的任务状态
 
 ---
 
-## 🔴 任务 1: 导入原始的产品库存表
+## 任务 1: 导入产品库存表原始数据
 
-- **任务ID**: `22`
-- **创建时间**: 2026-01-07 10:29:12
+- **任务ID**: `26`
+- **创建时间**: 2026-01-13 18:02:57
 - **创建者**: cursor
 
-### 📝 任务描述
+### 任务描述
 
-# Task: 导入原始的产品库存表
+# Task: 导入产品库存表原始数据
 
 ## DataFlow Configuration
 - **Schema**: dags
@@ -48,7 +48,7 @@
 CREATE TABLE test_product_inventory (
     updated_at timestamp COMMENT '更新时间',
     created_at timestamp COMMENT '创建时间',
-    is_active boolean COMMENT '是否有效',
+    is_active boolean COMMENT '是否启用',
     turnover_rate numeric(5, 2) COMMENT '周转率',
     outbound_quantity_30d integer COMMENT '30天出库数量',
     inbound_quantity_30d integer COMMENT '30天入库数量',
@@ -64,8 +64,9 @@ CREATE TABLE test_product_inventory (
     supplier varchar(200) COMMENT '供应商',
     brand varchar(100) COMMENT '品牌',
     category varchar(100) COMMENT '类别',
-    sku varchar(50) COMMENT '商品货号',
-    id serial COMMENT '编号',
+    product_name varchar(200) COMMENT '产品名称',
+    sku varchar(50) COMMENT 'SKU',
+    id serial COMMENT 'ID',
     create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '数据创建时间'
 );
 COMMENT ON TABLE test_product_inventory IS '产品库存表';
@@ -76,7 +77,7 @@ COMMENT ON TABLE test_product_inventory IS '产品库存表';
 - **Description**: 新数据将追加到目标表，不删除现有数据
 
 ## Request Content
-从数据源中导入原始数据到数据资源的产品库存表里。 
+从数据源的原始数据表导入数据资源中的产品库存表
 
 ## Implementation Steps
 1. Create an n8n workflow to execute the data import task
@@ -92,3 +93,40 @@ COMMENT ON TABLE test_product_inventory IS '产品库存表';
 
 ---
 
+## 任务 2: DF_DO202601130001
+
+- **任务ID**: `27`
+- **创建时间**: 2026-01-13 18:04:21
+- **创建者**: system
+
+### 任务描述
+
+# Task: DF_DO202601130001
+
+## DataFlow Configuration
+- **DataFlow ID**: 2220
+- **DataFlow Name**: 仓库库存汇总表_数据流程
+- **Order ID**: 17
+- **Order No**: DO202601130001
+
+## Source Tables
+- 产品库存表
+
+## Target Table
+- 仓库库存汇总表
+
+## Update Mode
+- **Mode**: Full Refresh (全量更新)
+- **Description**: 目标表将被清空后重新写入数据
+
+## Request Content
+1.从产品库存表中提取字段：仓库编号、产品编号、库存数量；2.对库存数量进行按仓库编号进行求和计算；3.无特殊过滤条件；4.最终输出数据格式包含字段：仓库编号、总库存数量
+
+## Implementation Steps
+1. 连接数据源，读取源数据表
+2. 根据处理逻辑执行数据转换
+3. 写入目标数据表
+4. 完成后回调更新订单状态为 onboard
+
+---
+

tasks/task_trigger.txt

@@ -1,11 +1,10 @@
 CURSOR_AUTO_EXECUTE_TASK_TRIGGER
-生成时间: 2026-01-12 10:52:01
-状态: 所有任务已完成
-待处理任务数: 0
-任务ID列表: []
+生成时间: 2026-01-13 19:15:13
+状态: 有待执行任务
+待处理任务数: 2
+任务ID列表: [26, 27]
 
 此文件用于触发Cursor自动执行任务。
-当此文件更新时间在5分钟内时，Cursor应该检查 tasks/task_execute_instructions.md 并执行其中的任务。
 
 【重要】Cursor AI 请立即执行以下操作：
 1. 读取 tasks/task_execute_instructions.md 文件

test_check_bd241.py

@@ -1,22 +0,0 @@
-#!/usr/bin/env python3
-"""检查 BusinessDomain ID=241 的名称"""
-from app import create_app
-from app.core.graph.graph_operations import connect_graph
-
-app = create_app()
-with app.app_context():
-    with connect_graph().session() as session:
-        query = """
-        MATCH (bd:BusinessDomain)
-        WHERE id(bd) = 241
-        RETURN bd.name_zh as name_zh, bd.name_en as name_en, bd.describe as describe
-        """
-        result = session.run(query).single()
-        
-        if result:
-            print(f'BusinessDomain ID=241:')
-            print(f'  name_zh: {result["name_zh"]}')
-            print(f'  name_en: {result["name_en"]}')
-            print(f'  describe: {result["describe"]}')
-        else:
-            print('未找到 BusinessDomain ID=241')

test_check_data_products.py

@@ -1,25 +0,0 @@
-#!/usr/bin/env python3
-"""检查数据产品表记录"""
-from app import create_app, db
-from sqlalchemy import text
-
-app = create_app()
-with app.app_context():
-    # 查询最新的数据产品记录
-    result = db.session.execute(text('''
-        SELECT id, product_name, product_name_en, target_table, source_dataflow_name, created_at 
-        FROM data_products 
-        ORDER BY created_at DESC 
-        LIMIT 5
-    ''')).fetchall()
-    
-    print('最新的数据产品记录:')
-    print('-' * 100)
-    for row in result:
-        print(f'ID: {row[0]}')
-        print(f'  产品名: {row[1]}')
-        print(f'  英文名: {row[2]}')
-        print(f'  目标表: {row[3]}')
-        print(f'  来源数据流: {row[4]}')
-        print(f'  创建时间: {row[5]}')
-        print('-' * 100)

test_create_dataflow.py

@@ -1,32 +0,0 @@
-#!/usr/bin/env python3
-"""测试创建数据流接口"""
-import requests
-import json
-import time
-
-data = {
-    "name_zh": f"测试数据产品注册_{int(time.time())}",
-    "describe": "测试数据产品注册功能",
-    "category": "应用类",
-    "leader": "system",
-    "organization": "citu",
-    "script_type": "sql",
-    "update_mode": "append",
-    "frequency": "月",
-    "tag": [],
-    "status": "active",
-    "script_requirement": {
-        "rule": "测试规则",
-        "source_table": [],
-        "target_table": [241]
-    }
-}
-
-print("创建数据流...")
-response = requests.post(
-    "http://localhost:5500/api/dataflow/add-dataflow",
-    json=data,
-    headers={"Content-Type": "application/json"}
-)
-print("Status Code:", response.status_code)
-print("Response:", json.dumps(response.json(), ensure_ascii=False, indent=2))

test_update_dataflow.py

@@ -1,30 +0,0 @@
-#!/usr/bin/env python3
-"""测试更新数据流接口"""
-import requests
-import json
-
-data = {
-    "name_zh": "测试用销售数据",
-    "name_en": "try_sales_data",
-    "category": "应用类",
-    "leader": "system",
-    "organization": "citu",
-    "script_type": "sql",
-    "update_mode": "append",
-    "frequency": "月",
-    "tag": [],
-    "status": "active",
-    "script_requirement": {
-        "rule": None,
-        "source_table": [],
-        "target_table": [241]
-    }
-}
-
-response = requests.put(
-    "http://localhost:5500/api/dataflow/update-dataflow/258",
-    json=data,
-    headers={"Content-Type": "application/json"}
-)
-print("Status Code:", response.status_code)
-print("Response:", response.text)

创建工作流程.txt

@@ -1,21 +0,0 @@
-接口：/api/dataflow/update-dataflow/258
-
-参数：{
-  "name_zh": "测试用销售数据",
-  "name_en": "try_sales_data",
-  "category": "应用类",
-  "leader": "system",
-  "organization": "citu",
-  "script_type": "sql",
-  "update_mode": "append",
-  "frequency": "月",
-  "tag": [],
-  "status": "active",
-  "script_requirement": {
-    "rule": null,
-    "source_table": [],
-    "target_table": [
-      241
-    ]
-  }
-}

待解决问题.md

@@ -1,33 +0,0 @@
-### 1、数据产品-数据可视化
-
-返回metadata标签
-
-/api/dataservice/products/7/preview?limit=200
-
-
-
-![8815cc0f772f07fba886e5c196e4d382](F:\wechat-storage\xwechat_files\wxid_5xec4jpl35ug21_89f6\temp\RWTemp\2025-12\01f404e86b61bc52ca649f3f60a19100\8815cc0f772f07fba886e5c196e4d382.png)
-
-
-
-### 2、数据流程
-
-提供查看代码接口
-
-
-
-![6396b199109c96d6e546e749387bb629](F:\wechat-storage\xwechat_files\wxid_5xec4jpl35ug21_89f6\temp\RWTemp\2025-12\01f404e86b61bc52ca649f3f60a19100\6396b199109c96d6e546e749387bb629.png)
-
-
-
-### 3、数据审核
-
-返回完整的标签信息
-
-/api/meta/review/list
-
-/api/meta/review/detail?id=44
-
-
-
-![bcad353f54825d075816fa8bbba77e4a](F:\wechat-storage\xwechat_files\wxid_5xec4jpl35ug21_89f6\temp\RWTemp\2025-12\01f404e86b61bc52ca649f3f60a19100\bcad353f54825d075816fa8bbba77e4a.png)