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

import json

from flask import Response, current_app, jsonify, request

from app import db
from app.api.data_interface import bp
from app.core.data_interface import interface
from app.core.data_rules.repository import DataRuleRepository
from app.core.graph.graph_operations import (
    MyEncoder,
    connect_graph,
    create_or_get_node,
)
from app.core.meta_data import get_formatted_time, translate_and_parse
from app.models.result import failed, success


def _repository():
    configured = current_app.extensions.get("data_rule_repository")
    if configured is not None:
        return configured
    return DataRuleRepository(db.session)


_STANDARD_WRITE_FIELDS = {
    "category",
    "describe",
    "id",
    "input",
    "migration_metadata",
    "name_zh",
    "output",
    "rule_version_id",
    "scope",
    "status",
    "tag",
}


def _legacy_standard_exists(node_id):
    if isinstance(node_id, bool) or not isinstance(node_id, int):
        return False
    query = (
        "MATCH (n:data_standard) WHERE id(n) = $node_id "
        "RETURN n.rule_version_id AS rule_version_id"
    )
    with connect_graph().session() as session:
        row = session.run(query, node_id=node_id).single()
    return row is not None and row.get("rule_version_id") is None


def _prepare_standard_write(value, *, is_new):
    if not isinstance(value, dict):
        raise ValueError("request body must be an object")
    if set(value) - _STANDARD_WRITE_FIELDS:
        raise ValueError("data standard request contains unsupported fields")
    receiver = dict(value)
    rule_version_id = receiver.get("rule_version_id")
    if rule_version_id is None:
        if is_new or not _legacy_standard_exists(receiver.get("id")):
            raise ValueError(
                "new data standard requires a published rule_version_id"
            )
    else:
        _repository().require_published_rule_version(rule_version_id)
    return receiver


@bp.route("/data/standard/add", methods=["POST"])
def data_standard_add():
    try:
        receiver = _prepare_standard_write(
            request.get_json(silent=True), is_new=True
        )
        name_zh = receiver["name_zh"]
        name_en = translate_and_parse(name_zh)
        receiver["name_en"] = name_en[0]
        receiver["create_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 (KeyError, TypeError, ValueError) as e:
        res = failed(str(e), 400, {})
        return jsonify(res), 400
    except Exception as e:
        res = failed(str(e), 500, {})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.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:
            result = session.run(cql, nodeId=nodeid).single()
            property = result["property"] if result else {}
            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(str(e), 500, {})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/data/standard/code", methods=["POST"])
def data_standard_code():
    return (
        jsonify(
            failed(
                "旧数据标准代码生成入口已关闭，请发布 RuleVersion 后绑定",
                410,
                {"semantics": "read_only_migration"},
            )
        ),
        410,
    )


@bp.route("/data/standard/update", methods=["POST"])
def data_standard_update():
    try:
        receiver = _prepare_standard_write(
            request.get_json(silent=True), is_new=False
        )
        name_zh = receiver["name_zh"]
        name_en = translate_and_parse(name_zh)
        receiver["name_en"] = name_en[0]
        receiver["create_time"] = get_formatted_time()

        create_or_get_node("data_standard", **receiver)

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

    except (KeyError, TypeError, ValueError) as e:
        res = failed(str(e), 400, {})
        return jsonify(res), 400
    except Exception as e:
        res = failed(str(e), 500, {})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.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))
        name_en_filter = receiver.get("name_en", None)
        name_zh_filter = receiver.get("name_zh", 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,
            name_en_filter,
            name_zh_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(str(e), 500, {})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.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(str(e), 500, {})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/data/label/add", methods=["POST"])
def data_label_add():
    try:
        receiver = request.get_json()
        name_zh = receiver["name_zh"]
        name_en = translate_and_parse(name_zh)
        receiver["name_en"] = name_en[0]
        receiver["create_time"] = get_formatted_time()
        create_or_get_node("DataLabel", **receiver)

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

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


@bp.route("/data/label/detail", methods=["POST"])
def data_label_detail():
    try:
        receiver = request.get_json()
        nodeid = receiver["id"]  # id

        cql = """MATCH (n:DataLabel) where id(n) = $nodeId
                  RETURN properties(n) as property"""
        with connect_graph().session() as session:
            result = session.run(cql, nodeId=nodeid).single()
            property = result["property"] if result else {}
            if "describe" not in property:
                property["describe"] = None
            # 将传入的节点ID添加到返回信息中
            property["id"] = nodeid
            res = success(property, "success")
            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)

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


@bp.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))
        name_en_filter = receiver.get("name_en", None)
        name_zh_filter = receiver.get("name_zh", 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,
            name_en_filter,
            name_zh_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(str(e), 500, {})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/labellist", methods=["POST"])
def interface_label_list():
    """获取 DataLabel 列表（支持多条件 category_filter 过滤）"""
    try:
        receiver = request.get_json() or {}
        page = int(receiver.get("current", 1))
        page_size = int(receiver.get("size", 10))
        name_en_filter = receiver.get("name_en")
        name_zh_filter = receiver.get("name_zh")
        category_filter = receiver.get("category_filter")
        group_filter = receiver.get("group")

        skip_count = (page - 1) * page_size
        data, total = interface.label_list(
            skip_count,
            page_size,
            name_en_filter,
            name_zh_filter,
            category_filter,
            group_filter,
        )

        # 只保留 id, name_zh, name_en 三个字段
        records = [
            {
                "id": item.get("id"),
                "name_zh": item.get("name_zh"),
                "name_en": item.get("name_en"),
            }
            for item in data
        ]

        response_data = {
            "records": records,
            "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(str(e), 500, {})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/graphall", methods=["POST"])
def interface_graph_all():
    """获取完整关系图谱"""
    try:
        receiver = request.get_json() or {}
        domain_id = receiver.get("id")
        include_meta = receiver.get("meta", True)

        if domain_id is None:
            res = failed("节点ID不能为空")
            return Response(
                json.dumps(res, ensure_ascii=False, cls=MyEncoder),
                mimetype="application/json",
            )

        try:
            domain_id = int(domain_id)
        except (ValueError, TypeError):
            res = failed(f"节点ID必须为整数, 收到的是: {domain_id}")
            return Response(
                json.dumps(res, ensure_ascii=False, cls=MyEncoder),
                mimetype="application/json",
            )

        graph_data = interface.graph_all(domain_id, include_meta)
        res = success(graph_data)
        return Response(
            json.dumps(res, ensure_ascii=False, cls=MyEncoder),
            mimetype="application/json",
        )
    except Exception as e:
        res = failed("获取图谱失败", error=str(e))
        return Response(
            json.dumps(res, ensure_ascii=False, cls=MyEncoder),
            mimetype="application/json",
        )


@bp.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(str(e), 500, {})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.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)
        return json.dumps(
            success(result, "success"),
            ensure_ascii=False,
            cls=MyEncoder,
        )
    except Exception as e:
        res = failed(str(e), 500, {})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.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:
            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(str(e), 500, {})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)


@bp.route("/data/label/delete", methods=["POST"])
def data_label_delete():
    """Delete data label node"""
    try:
        receiver = request.get_json()
        node_id = receiver.get("id") if receiver else None

        if not node_id:
            return jsonify(failed("node id is required", 400, {}))

        try:
            node_id = int(node_id)
        except (ValueError, TypeError):
            return jsonify(failed("node id must be an integer", 400, {}))

        delete_result = interface.node_delete(node_id)
        message = delete_result.get("message") or ""

        if delete_result.get("success"):
            res = success(
                {
                    "id": node_id,
                    "message": message,
                },
                "delete success",
            )
        else:
            res = failed(
                message,
                500,
                {
                    "id": node_id,
                    "message": message,
                },
            )

        return jsonify(res)
    except Exception as e:
        return jsonify(failed(f"delete failed: {str(e)}", 500, {}))