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

"""
数据指标API接口模块

该模块提供了数据指标的各种API接口，包括：
- 指标新增、更新和查询
- 指标列表展示
- 指标图谱生成
- 指标代码生成
- 指标关系管理
"""

from flask import request, jsonify
import json
from app.api.data_metric import bp
from app.core.data_metric.metric_interface import (
    metric_list, handle_metric_relation, handle_data_metric, 
    handle_meta_data_metric, handle_id_metric, metric_kinship_graph, 
    metric_impact_graph, metric_all_graph, data_metric_edit, metric_check,
    metric_delete
)
from app.core.llm import code_generate_metric
from app.core.meta_data import translate_and_parse
from app.models.result import success, failed
from app.core.graph.graph_operations import MyEncoder, connect_graph


@bp.route('/relation', methods=['POST'])
def data_metric_relation():
    """
    处理数据指标血缘关系
    
    请求参数:
    - id: 数据模型ID
    
    返回:
    - 处理结果，包含血缘关系数据
    """
    try:
        # 传入请求参数
        receiver = request.get_json()
        model_ids = receiver['id']  # 给定一个数据模型的id
        response_data = handle_metric_relation(model_ids)
        res = success(response_data, "success")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        res = failed({}, {"error": f"{e}"})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route('/add', methods=['POST'])
def data_metric_add():
    """
    新增数据指标
    
    请求参数:
    - name: 指标名称
    - 其他指标相关属性
    
    返回:
    - 处理结果，成功或失败
    """
    # 传入请求参数
    receiver = request.get_json()
    metric_name_zh = receiver['name_zh']  # 数据指标的name
    try:
        result_list = translate_and_parse(metric_name_zh)
        id, id_list = handle_data_metric(metric_name_zh, result_list, receiver)
        handle_meta_data_metric(id, id_list)

        res = success({}, "success")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)

    except Exception as e:
        res = failed({}, {"error": f"{e}"})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route('/code', methods=['POST'])
def data_metric_code():
    """
    生成指标计算的代码
    
    请求参数:
    - content: 指标规则描述
    - relation: 映射关系
    
    返回:
    - 生成的代码
    """
    # 传入请求参数
    receiver = request.get_json()
    content = receiver['content']  # 指标规则描述
    relation = receiver['relation']  # 映射关系

    try:
        id_list = [item["id"] for item in relation]
        cql = """
        MATCH (n:DataMetric) WHERE id(n) IN $Id_list
        WITH n.name_en AS name_en, n.name_zh AS name_zh
        WITH collect({name_en: name_en, name_zh: name_zh}) AS res
        WITH reduce(acc = {}, item IN res | apoc.map.setKey(acc, item.name_en, item.name_zh)) AS result
        RETURN result
        """
        driver = None
        try:
            driver = connect_graph()
            with driver.session() as session:
                query_result = session.run(cql, Id_list=id_list)
                id_relation = query_result.single()[0]
        except (ConnectionError, ValueError) as e:
            return json.dumps(failed({}, f"无法连接到数据库: {str(e)}"), ensure_ascii=False, cls=MyEncoder)
        finally:
            if driver:
                driver.close()
            
        result = code_generate_metric(content, id_relation)
        res = success(result, "success")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)

    except Exception as e:
        res = failed({}, {"error": f"{e}"})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route('/detail', methods=['POST'])
def data_metric_detail():
    """
    获取数据指标详情
    
    请求参数:
    - id: 指标ID
    
    返回:
    - 指标详情数据
    """
    try:
        # 传入请求参数
        receiver = request.get_json()
        id = int(receiver.get('id'))
        response_data = handle_id_metric(id)
        res = success(response_data, "success")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)

    except Exception as e:
        res = failed({}, {"error": f"{e}"})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route('/list', methods=['POST'])
def data_metric_list():
    """
    获取数据指标列表
    
    请求参数:
    - current: 当前页码，默认1
    - size: 每页大小，默认10
    - name_en: 英文名称过滤
    - name_zh: 名称过滤
    - category: 类别过滤
    - time: 时间过滤
    - tag: 标签过滤
    
    返回:
    - 指标列表数据和分页信息
    """
    try:
        # 传入请求参数
        receiver = request.get_json()
        page = int(receiver.get('current', 1))
        page_size = int(receiver.get('size', 10))
        name_en_filter = receiver.get('name_en', None)
        name_zh_filter = receiver.get('name_zh', None)
        category = receiver.get('category', None)
        create_time = receiver.get('create_time', None)
        tag = receiver.get('tag', None)

        # 计算跳过的记录的数量
        skip_count = (page - 1) * page_size

        data, total = metric_list(skip_count, page_size, name_en_filter,
                                  name_zh_filter, category, create_time, tag)

        response_data = {'records': data, 'total': total, 'size': page_size, 'current': page}
        res = success(response_data, "success")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        res = failed({}, {"error": f"{e}"})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route('/graph/all', methods=['POST'])
def data_metric_graph_all():
    """
    获取数据指标图谱
    
    请求参数:
    - id: 指标ID
    - type: 图谱类型，可选kinship/impact/all
    - meta: 是否返回元数据，true/false
    
    返回:
    - 图谱数据
    """
    try:
        # 传入请求参数
        receiver = request.get_json()
        nodeid = receiver['id']
        type = receiver['type']  # kinship/impact/all
        meta = receiver['meta']  # true/false 是否返回元数据

        if type == 'kinship':
            result = metric_kinship_graph(nodeid, meta)
        elif type == 'impact':
            result = metric_impact_graph(nodeid, meta)
        else:
            result = metric_all_graph(nodeid, meta)
        return json.dumps(success(result, "success"), ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        return json.dumps(failed({}, str(e)), ensure_ascii=False, cls=MyEncoder)


@bp.route('/list/graph', methods=['POST'])
def data_metric_list_graph():
    """
    获取数据指标列表图谱
    
    请求参数:
    - tag: 标签ID
    
    返回:
    - 图谱数据，包含节点和连线
    """
    try:
        # 传入请求参数
        receiver = request.get_json()

        if not receiver or 'tag' not in receiver:
            raise ValueError("Missing 'tag' parameter in request body")

        nodeid = receiver['tag']

        # 构建查询条件
        params = {}
        where_clause = ""

        if nodeid is not None:
            where_clause = "MATCH (n)-[:LABEL]->(la) WHERE id(la) = $nodeId"
            params['nodeId'] = nodeid
        query = f"""
        MATCH (n:DataMetric)
        OPTIONAL MATCH (n)-[:child]->(child)
        {where_clause}
        WITH 
            collect(DISTINCT {{id: toString(id(n)), text: n.name_zh, type: split(labels(n)[0], '_')[1]}}) AS nodes,
            collect(DISTINCT {{id: toString(id(child)), text: child.name_zh, type: split(labels(child)[0], '_')[1]}}) AS nodes2,
            collect(DISTINCT {{from: toString(id(n)), to: toString(id(child)), text: '下级'}}) AS lines
        RETURN nodes  + nodes2 AS nodes, lines  AS lines
        """
        driver = None
        try:
            driver = connect_graph()
            with driver.session() as session:
                result = session.run(query, **params)
                res = {}
                for item in result:
                    res = {
                        "nodes": [record for record in item['nodes'] if record['id']],
                        "lines": [record for record in item['lines'] if record['from'] and record['to']],
                    }
        except (ConnectionError, ValueError) as e:
            return json.dumps(failed({}, f"无法连接到数据库: {str(e)}"), ensure_ascii=False, cls=MyEncoder)
        finally:
            if driver:
                driver.close()
        return json.dumps(success(res, "success"), ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        return json.dumps(failed({}, str(e)), ensure_ascii=False, cls=MyEncoder)


@bp.route('/update', methods=['POST'])
def data_metric_update():
    """
    更新数据指标
    
    请求参数:
    - id: 指标ID
    - 其他需要更新的属性
    
    返回:
    - 处理结果，成功或失败
    """
    try:
        # 传入请求参数
        receiver = request.get_json()
        data_metric_edit(receiver)

        res = success({}, "success")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        res = failed({}, {"error": f"{e}"})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route('/check', methods=['POST'])
def data_metric_check():
    """
    检查指标计算公式中的变量
    
    请求参数:
    - formula: 指标计算公式文本，格式如：指标名称 = 变量1 + 变量2 * 数字
    
    返回:
    - 变量检查结果列表，包含每个变量的匹配信息
    """
    try:
        # 获取请求参数
        receiver = request.get_json()
        formula_text = receiver.get('formula', '')
        
        if not formula_text:
            res = failed({}, {"error": "公式文本不能为空"})
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
        
        # 调用核心业务逻辑
        result = metric_check(formula_text)
        
        res = success(result, "success")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        res = failed({}, {"error": f"{e}"})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route('/delete', methods=['POST'])
def data_metric_delete():
    """
    删除数据指标
    
    请求参数:
    - id: 指标节点ID
    
    返回:
    - 删除结果状态信息
    """
    try:
        # 获取请求参数
        receiver = request.get_json()
        metric_id = receiver.get('id')
        
        # 验证参数
        if not metric_id:
            res = failed({}, {"error": "指标ID不能为空"})
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
        
        # 转换为整数
        try:
            metric_id = int(metric_id)
        except (ValueError, TypeError):
            res = failed({}, {"error": "指标ID必须为整数"})
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
        
        # 调用核心业务逻辑执行删除
        delete_result = metric_delete(metric_id)
        
        # 根据删除结果返回响应
        if delete_result["success"]:
            res = success({
                "id": metric_id,
                "message": delete_result["message"]
            }, "删除成功")
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
        else:
            res = failed({
                "id": metric_id,
                "message": delete_result["message"]
            }, delete_result["message"])
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
            
    except Exception as e:
        res = failed({}, {"error": f"删除失败: {str(e)}"})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)