fu-hsi-service/src/main/java/com/supervision/neo4j/service/impl/Neo4jServiceImpl.java

package com.supervision.neo4j.service.impl;

import cn.hutool.core.lang.Pair;
import cn.hutool.core.util.StrUtil;
import cn.hutool.poi.excel.ExcelReader;
import cn.hutool.poi.excel.ExcelUtil;
import com.supervision.common.domain.R;
import com.supervision.neo4j.domain.CaseNode;
import com.supervision.neo4j.domain.Rel;
import com.supervision.neo4j.dto.WebRelDTO;
import com.supervision.neo4j.service.Neo4jService;
import com.supervision.neo4j.utils.Neo4jUtils;
import com.supervision.police.vo.GraphReqVO;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.neo4j.driver.Record;
import org.neo4j.driver.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;

/**
 * @author qmy
 * @since 2023-10-26
 */
@Slf4j
@Service
public class Neo4jServiceImpl implements Neo4jService {

    private final Driver driver;

    @Autowired
    private Neo4jServiceImpl(Driver driver) {
        this.driver = driver;
    }

    @Override
    public CaseNode save(CaseNode caseNode) {
        if (StringUtils.isEmpty(caseNode.getName()) || StringUtils.isEmpty(caseNode.getNodeType())) {
            throw new RuntimeException("未传节点名称或节点类型或图谱类型！");
        }
        List<CaseNode> byName = findByName(caseNode.getCaseId(), caseNode.getRecordId(), caseNode.getNodeType(), caseNode.getName(), caseNode.getPicType());
        if (byName != null && !byName.isEmpty()) {
            throw new RuntimeException("名称已存在！");
        }
        CaseNode res = null;
        try {
            Session session = driver.session();
            StringBuilder cql = new StringBuilder();
            Map<String, Object> params = new HashMap<>();
            cql.append("CREATE (n:").append(caseNode.getNodeType()).append("{name:$name");
            params.put("name", caseNode.getName());
            if (StringUtils.isNotEmpty(caseNode.getRecordId())) {
                cql.append(", recordSplitId:$recordSplitId");
                params.put("recordSplitId", caseNode.getRecordSplitId());
            }
            if (StringUtils.isNotEmpty(caseNode.getRecordId())) {
                cql.append(", recordId:$recordId");
                params.put("recordId", caseNode.getRecordId());
            }
            if (StringUtils.isNotEmpty(caseNode.getCaseId())) {
                cql.append(", caseId:$caseId");
                params.put("caseId", caseNode.getCaseId());
            }
            if (StringUtils.isNotEmpty(caseNode.getPicType())) {
                cql.append(", picType:$picType");
                params.put("picType", caseNode.getPicType());
            }
            cql.append("})").append(Neo4jUtils.NODE_RETURN);
            Result run = session.run(cql.toString(), params);
            res = Neo4jUtils.getOneNode(run);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return res;
    }

    /**
     * 删除节点,注意,删除节点的时候,要先把边删除掉,不然会报错
     *
     * @param id 点的ID
     */
    @Override
    public void delNode(Long id) {
        try {
            Session session = driver.session();
            StringBuilder cql = new StringBuilder();
            cql.append("MATCH (n) where id(n) = ").append(id).append(" DELETE n");
            log.info(cql.toString());
            Result run = session.run(cql.toString());
            while (run.hasNext()) {
                run.next();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    @Override
    public void deleteNoRelationNode(Long id) {
        try {
            Session session = driver.session();
            StringBuilder cql = new StringBuilder();
            cql.append("MATCH (n) WHERE n.id = ").append(id).append(" AND NOT (n)--() DELETE n");
            log.info(cql.toString());
            Result run = session.run(cql.toString());
            while (run.hasNext()) {
                run.next();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 删除边
     *
     * @param relId 边ID
     */
    public void deleteRel(Long relId) {
        try {
            Session session = driver.session();
            StringBuilder cql = new StringBuilder();
            cql.append("MATCH ()-[r]->() WHERE id(r) = ").append(relId).append(" DELETE r");
            log.info(cql.toString());
            Result run = session.run(cql.toString());
            while (run.hasNext()) {
                run.next();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    @Override
    public CaseNode findById(Long id) {
        CaseNode node = null;
        try {
            Session session = driver.session();
            Result run = session.run("MATCH (n) where id(n) = " + id + Neo4jUtils.NODE_RETURN);
            node = Neo4jUtils.getOneNode(run);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return node;
    }

    @Override
    public List<CaseNode> findByName(String caseId, String recordId, String nodeType, String name, String picType) {
        List<CaseNode> list = new ArrayList<>();
        try {
            Session session = driver.session();
            StringBuilder cql = new StringBuilder();
            cql.append("MATCH (n");
            if (StringUtils.isNotEmpty(nodeType)) {
                cql.append(":");
                cql.append(nodeType);
            }
            cql.append(") where 1 = 1");
            if (StringUtils.isNotEmpty(caseId)) {
                cql.append(" and n.caseId = ");
                cql.append(caseId);
            }
            if (StringUtils.isNotEmpty(recordId)) {
                cql.append(" and n.recordId = ");
                cql.append(recordId);
            }
            if (StringUtils.isNotEmpty(name)) {
                cql.append(" and n.name = '");
                cql.append(name);
                cql.append("'");
            }
            if (StringUtils.isNotEmpty(picType)) {
                cql.append(" and n.picType = ");
                cql.append(picType);
            }
            cql.append(Neo4jUtils.NODE_RETURN);
            Result run = session.run(cql.toString());
            list = Neo4jUtils.getNodeList(run);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return list;
    }

    @Override
    public CaseNode findOneByName(String caseId, String recordId, String nodeType, String name, String picType) {
        CaseNode node = null;
        try {
            Session session = driver.session();
            StringBuilder cql = new StringBuilder();
            Map<String, Object> params = new HashMap<>();
            cql.append("MATCH (n");
            if (StringUtils.isNotEmpty(nodeType)) {
                cql.append(":");
                cql.append(nodeType);
            }
            cql.append(") where 1 = 1");
            if (StringUtils.isNotEmpty(caseId)) {
                cql.append(" and n.caseId = $caseId");
                params.put("caseId", caseId);
            }
            if (StringUtils.isNotEmpty(recordId)) {
                cql.append(" and n.recordId = $recordId");
                params.put("recordId", recordId);
            }
            if (StringUtils.isNotEmpty(name)) {
                cql.append(" and n.name = $name");
                params.put("name", name);
            }
            if (StringUtils.isNotEmpty(picType)) {
                cql.append(" and n.picType = $picType");
                params.put("picType", picType);
            }
            cql.append(Neo4jUtils.NODE_RETURN);
            Result run = session.run(cql.toString(), params);
            node = Neo4jUtils.getOneNode(run);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return node;
    }

    @Override
    public Rel findRelation(Rel rel) {
        try {
            Session session = driver.session();
            Map<String, Object> params = new HashMap<>();
            String cql = "MATCH (a)-[rel:" + rel.getName() + "]->(b) where id(a) = $sourceId and id(b) = $targetId" +
                    Neo4jUtils.REL_RETURN;
            params.put("sourceId", rel.getSourceId());
            params.put("targetId", rel.getTargetId());

            Result run = session.run(cql, params);
            rel = Neo4jUtils.getOneRel(run);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        if (rel != null && rel.getId() != null) {
            return rel;
        } else {
            return null;
        }
    }

    @Override
    public Rel saveRelation(Rel rel) {
        try {
            Session session = driver.session();
            Map<String, Object> params = new HashMap<>();
            String cql = "MATCH (a), (b) where id(a) = $sourceId and id(b) = $targetId CREATE(a)-[rel:" + rel.getName() +
                    "]->(b) " + Neo4jUtils.REL_RETURN;
            params.put("sourceId", rel.getSourceId());
            params.put("targetId", rel.getTargetId());

            Result run = session.run(cql, params);
            rel = Neo4jUtils.getOneRel(run);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return rel;
    }

    @Override
    public R<?> getCaseGraph(GraphReqVO graphReqVO) {
        Map<String, Object> map = new HashMap<>();
        List<WebRelDTO> list = new ArrayList<>();
        List<Map<String, String>> nodes = new ArrayList<>();
        try {
            Session session = driver.session();
            StringBuilder relQuery = new StringBuilder("MATCH (n)-[rel]->(m) " +
                    "WHERE n.picType = $picType " +
                    "AND n.caseId = $caseId ");
            Map<String, Object> params = new HashMap<>();
            params.put("picType", graphReqVO.getPicType());
            params.put("caseId", graphReqVO.getCaseId());
            if (StringUtils.isNotEmpty(graphReqVO.getQueryStr())) {
                params.put("queryStr", graphReqVO.getQueryStr());
                relQuery.append("AND (n.name CONTAINS $queryStr OR type(rel) CONTAINS $queryStr OR m.name CONTAINS $queryStr) ");
            }
            if (graphReqVO.getNodeLabels() != null && !graphReqVO.getNodeLabels().isEmpty()) {
                params.put("nodeLabels", graphReqVO.getNodeLabels());
                relQuery.append("AND (ANY(label IN $nodeLabels WHERE label IN labels(n)) " +
                        "OR ANY(label IN $nodeLabels WHERE label IN labels(m))) ");
            }
            if (graphReqVO.getRelTypes() != null && !graphReqVO.getRelTypes().isEmpty()) {
                params.put("relTypes", graphReqVO.getRelTypes());
                relQuery.append("AND ANY(type IN $relTypes WHERE type = type(rel)) ");
            }
            relQuery.append("RETURN id(rel) as id, n.name as source, id(n) as sourceId, n.name as sourceName, type(rel) as relName, m.name as target, id(m) as targetId, m.name as targetName");
            log.info("relQuery:{}", relQuery);
            Result run = session.run(relQuery.toString(), params);
            while (run.hasNext()) {
                Record record = run.next();
                // 组织边
                long sourceId = record.get("sourceId").asLong();
                String relName = record.get("relName").asString();
                long targetId = record.get("targetId").asLong();
                list.add(new WebRelDTO(sourceId, targetId, relName));

                // 组织节点
                Map<String, String> sourceNodeMap = new HashMap<>();
                sourceNodeMap.put("name", record.get("sourceName").asString());
                sourceNodeMap.put("id", String.valueOf(sourceId));
                nodes.add(sourceNodeMap);
                Map<String, String> targetNodeMap = new HashMap<>();
                targetNodeMap.put("name", record.get("targetName").asString());
                targetNodeMap.put("id", String.valueOf(targetId));
                nodes.add(targetNodeMap);
            }
        } catch (Exception e) {
            log.error("查询失败", e);
        }

        // 根据节点ID去重
        List<Map<String, String>> distinctNodes = new ArrayList<>(nodes.stream()
                .collect(Collectors.toMap(
                        node -> node.get("id"),  // 以 ID 为唯一键
                        node -> node,            // 保留整个 Map 作为值
                        (existing, replacement) -> existing)) // 如果有重复 ID，保留第一个
                .values());
        // 节点和关系合并
        Pair<List<WebRelDTO>, List<Map<String, String>>> pair = mergeRecord(distinctNodes, list);

        map.put("list", pair.getKey());
        map.put("nodes", pair.getValue());
        return R.ok(map);
    }

    public Pair<List<WebRelDTO>, List<Map<String, String>>> mergeRecord(List<Map<String, String>> nodes, List<WebRelDTO> relDTOS) {
        Map<String, NodeMapRecord> nodeRecordMap = electNodeRecord(nodes);

         return Pair.of(mergerWebRel(relDTOS, nodeRecordMap), mergeNode(nodes, nodeRecordMap));
    }

    @Override
    public R<?> getNodeAndRelationListByCaseId(String picType, String caseId) {
        Map<String, Set<String>> map = new HashMap<>();
        Set<String> nodeLabels = new HashSet<>();
        Set<String> relTypes = new HashSet<>();
        try (Session session = driver.session()) {
            // 查询所有匹配caseId的节点及其关联关系
            String query = "MATCH (n)-[r]->(m) WHERE n.caseId = $caseId AND m.caseId = $caseId RETURN labels(n) as sourceLabels, type(r) as relName, labels(m) as targetLabels";
            session.executeRead(tx -> {
                Result result = tx.run(query, Values.parameters("caseId", caseId));
                while (result.hasNext()) {
                    Record record = result.next();
                    List<String> sourceLabels = record.get("sourceLabels").asList(Value::asString);
                    List<String> targetLabels = record.get("targetLabels").asList(Value::asString);
                    if (!sourceLabels.isEmpty()) {
                        nodeLabels.add(sourceLabels.get(0));
                    }
                    if (!targetLabels.isEmpty()) {
                        nodeLabels.add(targetLabels.get(0));
                    }
                    relTypes.add(record.get("relName").asString());
                }
                return null;
            });
        } catch (Exception e) {
            log.error("查询失败", e);
        }
        log.info("查询到的节点类型{}个：{}", nodeLabels.size(), nodeLabels);
        log.info("查询到的关系类型{}个：{}", relTypes.size(), relTypes);
        map.put("nodeLabels", nodeLabels);
        map.put("relTypes", relTypes);
        return R.ok(map);
    }


    record NodeMapRecord(String name, String id, Set<String> idSet) {
    }


    /**
     * 推选出代表节点信息
     *
     * @param nodes key: name ,entityName,id 节点信息
     * @return key: name ,value: NodeMapRecord
     */
    private Map<String, NodeMapRecord> electNodeRecord(List<Map<String, String>> nodes) {
        Map<String, NodeMapRecord> nodeRecordMap = new HashMap<>();
        for (Map<String, String> node : nodes) {
            String name = node.get("name");
            String id = node.get("id");
            NodeMapRecord nodeMapRecord = nodeRecordMap.get(name);
            if (nodeMapRecord == null) {
                Set<String> idSet = new HashSet<>();
                idSet.add(id);
                nodeRecordMap.put(name, new NodeMapRecord(name, id, idSet));
            } else {
                nodeMapRecord.idSet.add(id);
            }
        }
        return nodeRecordMap;
    }

    /**
     * 合并节点信息
     * 合并依据：
     * name为唯一标识
     *
     * @param nodes         key: name ,entityName,id
     * @param nodeRecordMap 代表节点信息
     * @return 合并后的节点信息
     */
    private List<Map<String, String>> mergeNode(List<Map<String, String>> nodes, Map<String, NodeMapRecord> nodeRecordMap) {

        return nodes.stream().map(map -> {
                    Map<String, String> nodeMap = new HashMap<>();
                    nodeMap.put("name", map.get("name"));
                    NodeMapRecord nodeMapRecord = nodeRecordMap.get(map.get("name"));
                    if (null == nodeMapRecord) {
                        log.warn("mergeNode:节点信息异常，nodeRecordMap中不存在节点名称为：{}的NodeMapRecord", map.get("name"));
                        return nodeMap;
                    }
                    if (!nodeMapRecord.idSet.contains(map.get("id"))) {
                        log.warn("mergeNode:节点信息异常，nodeMapRecord.idSet中不包含节点id:{},节点名称为：{}", map.get("id"), map.get("name"));
                        return nodeMap;
                    }
                    nodeMap.put("id", nodeMapRecord.id);
                    return nodeMap;
                }).filter(map -> StrUtil.isNotEmpty(map.get("id")))
                .filter(distinctPredicate(m -> m.get("id"))).collect(Collectors.toList());
    }

    /**
     * 合并关系信息
     *
     * @param webRelDTOList 关系信息
     * @param nodeRecordMap 代表节点信息
     * @return 合并后的关系信息
     */
    private List<WebRelDTO> mergerWebRel(List<WebRelDTO> webRelDTOList, Map<String, NodeMapRecord> nodeRecordMap) {

        Map<String, NodeMapRecord> idNodeRecordMap = nodeRecordMap.entrySet().stream()
                .collect(Collectors.toMap(entry -> entry.getValue().id, Map.Entry::getValue));

        return webRelDTOList.stream().map(webRelDTO -> {
                    String target = webRelDTO.getTarget();
                    String source = webRelDTO.getSource();
                    String name = webRelDTO.getName();

                    String sourceNew = idNodeRecordMap.entrySet().stream()
                            .filter(entry -> entry.getValue().idSet.contains(source))
                            .findAny().map(Map.Entry::getKey).orElse("");

                    String targetNew = idNodeRecordMap.entrySet().stream()
                            .filter(entry -> entry.getValue().idSet.contains(target))
                            .findAny().map(Map.Entry::getKey).orElse("");

                    if (StrUtil.isEmpty(sourceNew) || StrUtil.isEmpty(targetNew)) {
                        log.warn("mergerWebRel:关系信息异常，nodeRecordMap中不存在节点id:{}或节点id:{}信息,节点名称为：{}", source, target, name);
                    }

                    return new WebRelDTO(sourceNew, targetNew, name);
                }).filter(webRelDTO -> StrUtil.isNotEmpty(webRelDTO.getSource()) && StrUtil.isNotEmpty(webRelDTO.getTarget()))
                .filter(distinctPredicate(rel -> rel.getSource() + rel.getTarget())).toList();
    }

    private <K> Predicate<K> distinctPredicate(Function<K, Object> function) {
        ConcurrentHashMap<Object, Boolean> map = new ConcurrentHashMap<>();
        return (t) -> null == map.putIfAbsent(function.apply(t), true);

    }


    @Override
    public void createAbstractGraph(String path, String sheetName) {
        // 首先从数据库中读到数据
        ExcelReader reader = ExcelUtil.getReader(path, sheetName);
        List<AbstractGraphExcelHeader> abstractGraphExcelHeaders = reader.readAll(AbstractGraphExcelHeader.class);
        Map<String, CaseNode> nodeMap = new HashMap<>();
        Map<String, Rel> relMap = new HashMap<>();
        for (AbstractGraphExcelHeader abstractGraphExcelHeader : abstractGraphExcelHeaders) {
            // from
            if (!nodeMap.containsKey(abstractGraphExcelHeader.getFrom())) {
                CaseNode caseNode = new CaseNode(abstractGraphExcelHeader.getFrom(), abstractGraphExcelHeader.getFrom(), "0");
                log.info("点:{}插入成功", abstractGraphExcelHeader.getFrom());
                CaseNode save = save(caseNode);
                nodeMap.put(abstractGraphExcelHeader.getFrom(), save);
            }
            // to
            if (!nodeMap.containsKey(abstractGraphExcelHeader.getTo())) {
                CaseNode caseNode = new CaseNode(abstractGraphExcelHeader.getTo(), abstractGraphExcelHeader.getTo(), "0");
                CaseNode save = save(caseNode);
                log.info("点:{}插入成功", abstractGraphExcelHeader.getTo());
                nodeMap.put(abstractGraphExcelHeader.getTo(), save);
            }
            // relation
            if (!relMap.containsKey(abstractGraphExcelHeader.getFrom() + "->" + abstractGraphExcelHeader.getRelation() + "->" + abstractGraphExcelHeader.getTo())) {
                Rel rel = new Rel(nodeMap.get(abstractGraphExcelHeader.getFrom()).getId(), abstractGraphExcelHeader.getRelation(), nodeMap.get(abstractGraphExcelHeader.getTo()).getId(), "0");
                saveRelation(rel);
                log.info("关系:{}插入成功", (abstractGraphExcelHeader.getFrom() + "->" + abstractGraphExcelHeader.getRelation() + "->" + abstractGraphExcelHeader.getTo()));
                relMap.put(abstractGraphExcelHeader.getFrom() + "->" + abstractGraphExcelHeader.getRelation() + "->" + abstractGraphExcelHeader.getTo(), rel);
            }
        }


    }

    public void deleteAbstractGraph() {
        Session session = driver.session();
        // 首先查出来所有的抽象节点
        Result run = session.run("MATCH (n) WHERE n.picType = '0' OPTIONAL MATCH (n)-[r]-() RETURN id(n) as nodeId, id(r) as relId");
        Set<String> nodeIdSet = new HashSet<>();
        HashSet<String> relIdSet = new HashSet<>();
        while (run.hasNext()) {
            Record record = run.next();
            String nodeId = Neo4jUtils.valueTransportString(record.get("nodeId"));
            nodeIdSet.add(nodeId);
            String relId = Neo4jUtils.valueTransportString(record.get("relId"));
            relIdSet.add(relId);
        }
        // 删除边
        for (String s : relIdSet) {
            long relId = Long.parseLong(s);
            deleteRel(relId);
            log.info("删除边:{} 成功", relId);
        }
        // 删除节点
        for (String s : nodeIdSet) {
            long nodeId = Long.parseLong(s);
            delNode(nodeId);
            log.info("删除节点:{} 成功", nodeId);
        }


    }

    @Data
    private static class AbstractGraphExcelHeader {
        private String from;
        private String relation;
        private String to;
    }

    @Data
    private static class MockDataGraphExcelHeader {
        private String fromType;
        private String from;
        private String relation;
        private String to;
        private String toType;
    }

    @Override
    public void mockTestGraph(String path, String sheetName, String recordId, String recordSplitId, String caseId) {
        // 首先从数据库中读到数据
        ExcelReader reader = ExcelUtil.getReader(path, sheetName);
        List<MockDataGraphExcelHeader> mockDataGraphExcelList = reader.readAll(MockDataGraphExcelHeader.class);
        Map<String, CaseNode> nodeMap = new HashMap<>();
        Map<String, Rel> relMap = new HashMap<>();
        for (MockDataGraphExcelHeader mockData : mockDataGraphExcelList) {
            // from
            if (!nodeMap.containsKey(mockData.getFrom())) {
                CaseNode caseNode = new CaseNode(mockData.getFrom(), mockData.getFromType(), recordSplitId, recordId, caseId, "1");
                log.info("点:{}插入成功", mockData.getFrom());
                CaseNode save = save(caseNode);
                nodeMap.put(mockData.getFrom(), save);
            }
            // to
            if (!nodeMap.containsKey(mockData.getTo())) {
                CaseNode caseNode = new CaseNode(mockData.getTo(), mockData.getToType(), recordSplitId, recordId, caseId, "1");
                CaseNode save = save(caseNode);
                log.info("点:{}插入成功", mockData.getTo());
                nodeMap.put(mockData.getTo(), save);
            }
            // relation
            if (!relMap.containsKey(mockData.getFrom() + "->" + mockData.getRelation() + "->" + mockData.getTo())) {
                Rel rel = new Rel(nodeMap.get(mockData.getFrom()).getId(), mockData.getRelation(), nodeMap.get(mockData.getTo()).getId(), "1");
                saveRelation(rel);
                log.info("关系:{}插入成功", (mockData.getFrom() + "->" + mockData.getRelation() + "->" + mockData.getTo()));
                relMap.put(mockData.getFrom() + "->" + mockData.getRelation() + "->" + mockData.getTo(), rel);
            }
        }
    }

    @Override
    public List<Record> executeCypher(String cypher,Map<String, Object> parameters) {
        return this.driver.session().run(cypher, parameters).list();
    }
}