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

from flask import request, jsonify
from app.api.data_flow import bp
from app.core.data_flow.dataflows import DataFlowService
import logging
from datetime import datetime

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)
        return jsonify({
            'code': 200,
            'message': 'success',
            'data': result
        })
    except Exception as e:
        logger.error(f"获取数据流列表失败: {str(e)}")
        return jsonify({
            'code': 500,
            'message': f'获取数据流列表失败: {str(e)}',
            'data': None
        }), 500

@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:
            return jsonify({
                'code': 200,
                'message': 'success',
                'data': result
            })
        else:
            return jsonify({
                'code': 404,
                'message': '数据流不存在',
                'data': None
            }), 404
    except Exception as e:
        logger.error(f"获取数据流详情失败: {str(e)}")
        return jsonify({
            'code': 500,
            'message': f'获取数据流详情失败: {str(e)}',
            'data': None
        }), 500

@bp.route('/add-dataflow', methods=['POST'])
def create_dataflow():
    """创建新的数据流"""
    try:
        data = request.get_json()
        if not data:
            return jsonify({
                'code': 400,
                'message': '请求数据不能为空',
                'data': None
            }), 400
            
        result = DataFlowService.create_dataflow(data)
        return jsonify({
            'code': 200,
            'message': '数据流创建成功',
            'data': result
        })
    except Exception as e:
        logger.error(f"创建数据流失败: {str(e)}")
        return jsonify({
            'code': 500,
            'message': f'创建数据流失败: {str(e)}',
            'data': None
        }), 500

@bp.route('/update-dataflow/<int:dataflow_id>', methods=['PUT'])
def update_dataflow(dataflow_id):
    """更新数据流"""
    try:
        data = request.get_json()
        if not data:
            return jsonify({
                'code': 400,
                'message': '请求数据不能为空',
                'data': None
            }), 400
            
        result = DataFlowService.update_dataflow(dataflow_id, data)
        if result:
            return jsonify({
                'code': 200,
                'message': '数据流更新成功',
                'data': result
            })
        else:
            return jsonify({
                'code': 404,
                'message': '数据流不存在',
                'data': None
            }), 404
    except Exception as e:
        logger.error(f"更新数据流失败: {str(e)}")
        return jsonify({
            'code': 500,
            'message': f'更新数据流失败: {str(e)}',
            'data': None
        }), 500

@bp.route('/delete-dataflow/<int:dataflow_id>', methods=['DELETE'])
def delete_dataflow(dataflow_id):
    """删除数据流"""
    try:
        result = DataFlowService.delete_dataflow(dataflow_id)
        if result:
            return jsonify({
                'code': 200,
                'message': '数据流删除成功',
                'data': None
            })
        else:
            return jsonify({
                'code': 404,
                'message': '数据流不存在',
                'data': None
            }), 404
    except Exception as e:
        logger.error(f"删除数据流失败: {str(e)}")
        return jsonify({
            'code': 500,
            'message': f'删除数据流失败: {str(e)}',
            'data': None
        }), 500

@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)
        return jsonify({
            'code': 200,
            'message': '数据流执行成功',
            'data': result
        })
    except Exception as e:
        logger.error(f"执行数据流失败: {str(e)}")
        return jsonify({
            'code': 500,
            'message': f'执行数据流失败: {str(e)}',
            'data': None
        }), 500

@bp.route('/get-dataflow-status/<int:dataflow_id>', methods=['GET'])
def get_dataflow_status(dataflow_id):
    """获取数据流执行状态"""
    try:
        result = DataFlowService.get_dataflow_status(dataflow_id)
        return jsonify({
            'code': 200,
            'message': 'success',
            'data': result
        })
    except Exception as e:
        logger.error(f"获取数据流状态失败: {str(e)}")
        return jsonify({
            'code': 500,
            'message': f'获取数据流状态失败: {str(e)}',
            'data': None
        }), 500

@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)
        return jsonify({
            'code': 200,
            'message': 'success',
            'data': result
        })
    except Exception as e:
        logger.error(f"获取数据流日志失败: {str(e)}")
        return jsonify({
            'code': 500,
            'message': f'获取数据流日志失败: {str(e)}',
            'data': None
        }), 500

@bp.route('/create-script', methods=['POST'])
def create_script():
    """使用Deepseek模型生成脚本"""
    try:
        json_data = request.get_json()
        if not json_data:
            return jsonify({
                'code': 400,
                'message': '请求数据不能为空',
                'data': None
            }), 400
        
        # 提取文本描述
        request_data = json_data.get('request_data')
        
        if not request_data or not isinstance(request_data, str):
            return jsonify({
                'code': 400,
                'message': '请求数据必须是文本描述',
                'data': None
            }), 400
        
        # 调用DataFlowService的create_script方法
        script_content = DataFlowService.create_script(request_data)
        
        return jsonify({
            'code': 200,
            'message': '脚本生成成功',
            'data': {
                'script_content': script_content,
                'format': 'txt',
                'generated_at': datetime.now().isoformat()
            }
        })
    except ValueError as ve:
        logger.error(f"脚本生成参数错误: {str(ve)}")
        return jsonify({
            'code': 400,
            'message': f'参数错误: {str(ve)}',
            'data': None
        }), 400
    except Exception as e:
        logger.error(f"脚本生成失败: {str(e)}")
        return jsonify({
            'code': 500,
            'message': f'脚本生成失败: {str(e)}',
            'data': None
        }), 500