DataOps-platform/app/api/data_flow/routes.py

import json
import logging

from flask import request

from app.api.data_flow import bp
from app.core.data_flow.dataflows import DataFlowService
from app.core.graph.graph_operations import MyEncoder
from app.models.result import failed, success

logger = logging.getLogger(__name__)


@bp.route("/get-dataflows-list", methods=["GET"])
def get_dataflows():
    """获取数据流列表"""
    try:
        page = request.args.get("page", 1, type=int)
        page_size = request.args.get("page_size", 10, type=int)
        search = request.args.get("search", "")

        result = DataFlowService.get_dataflows(
            page=page,
            page_size=page_size,
            search=search,
        )
        res = success(result, "success")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        logger.error(f"获取数据流列表失败: {str(e)}")
        res = failed(f"获取数据流列表失败: {str(e)}")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/get-dataflow/<int:dataflow_id>", methods=["GET"])
def get_dataflow(dataflow_id):
    """根据ID获取数据流详情"""
    try:
        result = DataFlowService.get_dataflow_by_id(dataflow_id)
        if result:
            res = success(result, "success")
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
        else:
            res = failed("数据流不存在", code=404)
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        logger.error(f"获取数据流详情失败: {str(e)}")
        res = failed(f"获取数据流详情失败: {str(e)}")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/add-dataflow", methods=["POST"])
def create_dataflow():
    """创建新的数据流"""
    try:
        data = request.get_json()
        if not data:
            res = failed("请求数据不能为空", code=400)
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)

        result = DataFlowService.create_dataflow(data)
        res = success(result, "数据流创建成功")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except ValueError as ve:
        logger.error(f"创建数据流参数错误: {str(ve)}")
        res = failed(f"参数错误: {str(ve)}", code=400)
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        logger.error(f"创建数据流失败: {str(e)}")
        res = failed(f"创建数据流失败: {str(e)}")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/update-dataflow/<int:dataflow_id>", methods=["PUT"])
def update_dataflow(dataflow_id):
    """更新数据流"""
    try:
        data = request.get_json()
        if not data:
            res = failed("请求数据不能为空", code=400)
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)

        result = DataFlowService.update_dataflow(dataflow_id, data)
        if result:
            res = success(result, "数据流更新成功")
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
        else:
            res = failed("数据流不存在", code=404)
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        logger.error(f"更新数据流失败: {str(e)}")
        res = failed(f"更新数据流失败: {str(e)}")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/delete-dataflow/<int:dataflow_id>", methods=["DELETE"])
def delete_dataflow(dataflow_id):
    """删除数据流"""
    try:
        result = DataFlowService.delete_dataflow(dataflow_id)
        if result:
            res = success({}, "数据流删除成功")
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
        else:
            res = failed("数据流不存在", code=404)
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        logger.error(f"删除数据流失败: {str(e)}")
        res = failed(f"删除数据流失败: {str(e)}")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/execute-dataflow/<int:dataflow_id>", methods=["POST"])
def execute_dataflow(dataflow_id):
    """执行数据流"""
    try:
        data = request.get_json() or {}
        result = DataFlowService.execute_dataflow(dataflow_id, data)
        res = success(result, "数据流执行成功")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        logger.error(f"执行数据流失败: {str(e)}")
        res = failed(f"执行数据流失败: {str(e)}")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/get-dataflow-status/<int:dataflow_id>", methods=["GET"])
def get_dataflow_status(dataflow_id):
    """获取数据流执行状态"""
    try:
        result = DataFlowService.get_dataflow_status(dataflow_id)
        res = success(result, "success")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        logger.error(f"获取数据流状态失败: {str(e)}")
        res = failed(f"获取数据流状态失败: {str(e)}")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/get-dataflow-logs/<int:dataflow_id>", methods=["GET"])
def get_dataflow_logs(dataflow_id):
    """获取数据流执行日志"""
    try:
        page = request.args.get("page", 1, type=int)
        page_size = request.args.get("page_size", 50, type=int)

        result = DataFlowService.get_dataflow_logs(
            dataflow_id,
            page=page,
            page_size=page_size,
        )
        res = success(result, "success")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        logger.error(f"获取数据流日志失败: {str(e)}")
        res = failed(f"获取数据流日志失败: {str(e)}")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/get-BD-list", methods=["GET"])
def get_business_domain_list():
    """获取BusinessDomain节点列表"""
    try:
        logger.info("接收到获取BusinessDomain列表请求")

        # 调用服务层函数获取BusinessDomain列表
        bd_list = DataFlowService.get_business_domain_list()

        res = success(bd_list, "操作成功")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        logger.error(f"获取BusinessDomain列表失败: {str(e)}")
        res = failed(f"获取BusinessDomain列表失败: {str(e)}", 500, {})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/get-script/<int:dataflow_id>", methods=["GET"])
def get_script(dataflow_id):
    """
    获取 DataFlow 关联的脚本内容

    Args:
        dataflow_id: DataFlow 节点的 ID

    Returns:
        包含脚本内容和元信息的 JSON 响应:
        - script_path: 脚本路径
        - script_content: 脚本内容
        - script_type: 脚本类型（python/javascript/sql等）
        - dataflow_id: DataFlow ID
        - dataflow_name: DataFlow 中文名称
        - dataflow_name_en: DataFlow 英文名称
    """
    try:
        logger.info(f"接收到获取脚本请求, DataFlow ID: {dataflow_id}")

        result = DataFlowService.get_script_content(dataflow_id)

        res = success(result, "获取脚本成功")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except ValueError as ve:
        logger.warning(f"获取脚本参数错误: {str(ve)}")
        res = failed(f"{str(ve)}", code=400)
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except FileNotFoundError as fe:
        logger.warning(f"脚本文件不存在: {str(fe)}")
        res = failed(f"脚本文件不存在: {str(fe)}", code=404)
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        logger.error(f"获取脚本失败: {str(e)}")
        res = failed(f"获取脚本失败: {str(e)}")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)