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

from flask import request
from app import app
from app.models.result import success, failed
from app.core.graph.graph_operations import connect_graph, MyEncoder, create_or_get_node
from app.core.data_interface import interface
from app.core.meta_data import translate_and_parse, get_formatted_time
from app.core.llm import code_generate_standard
import json


# 数据标准新增 data_standard
@app.route('/data/standard/add', methods=['POST'])
def data_standard_add():
    try:
        # 传入请求参数
        receiver = request.get_json()
        name = receiver['name']  # 名称
        en_name = translate_and_parse(name)  # 英文名
        receiver['en_name'] = en_name[0]
        receiver['time'] = get_formatted_time()
        receiver['tag'] = json.dumps(receiver['tag'], ensure_ascii=False)

        create_or_get_node('data_standard', **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)


# 数据标准详情 data_standard
@app.route('/data/standard/detail', methods=['POST'])
def data_standard_detail():
    try:
        # 传入请求参数
        receiver = request.get_json()
        nodeid = receiver['id']  # id

        cql = """MATCH (n:data_standard) where id(n) = $nodeId 
                  RETURN properties(n) as property"""
        # Create a session from the driver returned by connect_graph
        with connect_graph().session() as session:
            property = session.run(cql, nodeId=nodeid).evaluate()
            if "tag" not in property:
                property["tag"] = None
            else:
                property["tag"] = json.loads(property["tag"])
            if "describe" not in property:
                property["describe"] = None
            res = success(property, "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)


# 数据标准代码 data_standard
@app.route('/data/standard/code', methods=['POST'])
def data_standard_code():
    try:
        # 传入请求参数
        receiver = request.get_json()
        input = receiver['input']
        describe = receiver['describe']
        output = receiver['output']
        relation = {
            "输入参数": input,
            "输出参数": output
        }
        result = code_generate_standard(describe, 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)


# 数据标准更新 data_standard 未加到接口文档
@app.route('/data/standard/update', methods=['POST'])
def data_standard_update():
    try:
        # 传入请求参数
        receiver = request.get_json()
        name = receiver['name']  # 名称
        en_name = translate_and_parse(name)  # 英文名
        receiver['en_name'] = en_name[0]
        receiver['time'] = get_formatted_time()

        create_or_get_node('data_standard', **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)


# 数据标准列表展示
@app.route('/data/standard/list', methods=['POST'])
def data_standard_list():
    try:
        # 传入请求参数
        receiver = request.get_json()
        page = int(receiver.get('current', 1))
        page_size = int(receiver.get('size', 10))
        en_name_filter = receiver.get('en_name', None)
        name_filter = receiver.get('name', None)
        category = receiver.get('category', None)
        time = receiver.get('time', None)

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

        data, total = interface.standard_list(skip_count, page_size, en_name_filter,
                                              name_filter, category, time)

        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)


# 数据标准的图谱（血缘关系Kinship+影响关系Impact+所有关系all）
@app.route('/data/standard/graph/all', methods=['POST'])
def data_standard_graph_all():
    try:
        # 传入请求参数
        receiver = request.get_json()
        nodeid = receiver['id']
        type = receiver['type']  # kinship/impact/all
        if type == 'kinship':
            result = interface.standard_kinship_graph(nodeid)
        elif type == 'impact':
            result = interface.standard_impact_graph(nodeid)
        else:
            result = interface.standard_all_graph(nodeid)
        return json.dumps(success(result, "success"), ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        res = failed({}, {"error": f"{e}"})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


# 数据标签新增 data_label
@app.route('/data/label/add', methods=['POST'])
def data_label_add():
    try:
        # 传入请求参数
        receiver = request.get_json()
        name = receiver['name']  # 名称
        en_name = translate_and_parse(name)  # 英文名
        receiver['en_name'] = en_name[0]
        receiver['time'] = get_formatted_time()
        create_or_get_node('data_label', **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)


# 数据标签详情 data_label
@app.route('/data/label/detail', methods=['POST'])
def data_label_detail():
    try:
        # 传入请求参数
        receiver = request.get_json()
        nodeid = receiver['id']  # id

        cql = """MATCH (n:data_label) where id(n) = $nodeId 
                  RETURN properties(n) as property"""
        with connect_graph().session() as session:
            property = session.run(cql, nodeId=nodeid).evaluate()
            if "describe" not in property:
                property["describe"] = None
            res = success(property, "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)


# 数据标签列表展示（分类，名称，时间检索）
@app.route('/data/label/list', methods=['POST'])
def data_label_list():
    try:
        # 传入请求参数
        receiver = request.get_json()
        page = int(receiver.get('current', 1))
        page_size = int(receiver.get('size', 10))
        en_name_filter = receiver.get('en_name', None)
        name_filter = receiver.get('name', None)
        category = receiver.get('category', None)
        group = receiver.get('group', None)

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

        data, total = interface.label_list(skip_count, page_size, en_name_filter,
                                           name_filter, category, group)

        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)


# 24.11.19 数据标签动态识别分组
@app.route('/data/label/dynamic/identify', methods=['POST'])
def data_label_dynamic_identify():
    try:
        # 传入请求参数
        receiver = request.get_json()
        name_filter = receiver.get('content', None)

        data = interface.dynamic_label_list(name_filter)

        res = success(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)


# 数据标签的图谱（血缘关系Kinship+影响关系Impact+所有关系all）
@app.route('/data/label/graph/all', methods=['POST'])
def data_label_graph():
    try:
        # 传入请求参数
        receiver = request.get_json()
        nodeid = receiver['id']
        type = receiver['type']  # kinship/impact/all
        if type == 'kinship':
            result = interface.label_kinship_graph(nodeid)
        elif type == 'impact':
            result = interface.label_impact_graph(nodeid)
        else:
            result = interface.label_kinship_graph(nodeid)  # 对于标签，将all和kinship都视为相同处理
        return json.dumps(success(result, "success"), ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        res = failed({}, {"error": f"{e}"})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


# 删除标签、标准、指标间的关系
@app.route('/metric/label/standard/delete', methods=['POST'])
def metric_label_standard_delete():
    try:
        # 传入请求参数
        receiver = request.get_json()
        sourceid = receiver['sourceid']
        targetid = receiver['targetid']

        # 查询语句，查询两个节点之间的关系
        cql = """
        MATCH (source)-[r]-(target)
        WHERE id(source) = $sourceid AND id(target) = $targetid
        DELETE r
        """
        with connect_graph().session() as session:
            result = session.run(cql, sourceid=sourceid, targetid=targetid)

        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)