kafka-neo4j/src/main/java/com/winhc/service/impl/PersonMergeV2Impl.java

package com.winhc.service.impl;

import com.winhc.common.enums.CompanyEnum;
import com.winhc.config.ConfigConstant;
import com.winhc.kafka.KafkaProduce;
import com.winhc.pojo.MergePerson;
import com.winhc.service.RelationService;
import com.winhc.utils.CompanyUtils;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.neo4j.driver.Driver;
import org.neo4j.driver.Session;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author π
 * @Description:人员合并
 * @date 2021/1/11 10:03
 */
@Slf4j
@Service("personMergeV2Impl")
@AllArgsConstructor
public class PersonMergeV2Impl implements RelationService {

    @Autowired
    @Qualifier("DriverV1")
    Driver driver;
    @Autowired
    KafkaProduce kafkaProduce;
    @Autowired
    ConfigConstant configConstant;

    @Override
    public String save(List<Map<String, Object>> batch_list) {
        if (batch_list.isEmpty()) return null;
        long start = System.currentTimeMillis();
        Session session = driver.session();

        final String cql =
                "\nWITH  {batch_list} AS batch_list \n" +
                        "UNWIND batch_list AS row \n" +
                        "MATCH (p:" + CompanyEnum.Lable.PERSON.code + "{person_id: row.person_id})-[*0..3]-(q:" + CompanyEnum.Lable.PERSON.code + "{name:row.name})\n" +
                        "WHERE ID(p) <> ID(q)\n" +
                        "WITH p.person_id as person_id, p as first_node, apoc.coll.sort(collect(distinct ID(q))) as all_ids\n" +
                        "UNWIND all_ids as all_id\n" +
                        "MATCH(m:" + CompanyEnum.Lable.PERSON.code + ")-[r]-(x:" + CompanyEnum.Lable.COMPANY.code + ")\n" +
                        "WHERE ID(m) = all_id\n" +
                        "WITH person_id, first_node, m as other_node, x, r\n" +
                        "CALL apoc.merge.relationship.eager(first_node, TYPE(r), properties(r),{}, x,{}) YIELD rel\n" +
                        "SET first_node:" + CompanyUtils.getIncrPersonLabelV2("新增", CompanyEnum.SPLIT_HOUR) + "\n" +
                        "SET other_node:" + CompanyUtils.getIncrPersonLabelV2("删除", CompanyEnum.SPLIT_HOUR) + "\n" +
                        "DELETE r" + "\n" +
                        "RETURN first_node.person_id as new_human_pid,first_node.name as new_human_name,other_node.person_id as old_human_pid,other_node.name as old_human_name" + "\n";
        log.info("consumer size: {}, cql:{}", batch_list.size(), cql);
        String data = CompanyUtils.writeNeo4j2(session, cql, new HashMap<String, Object>() {{
            put("batch_list", batch_list);
        }});
        session.close();
        log.info("class:{} | save size:{} | cost:{}", PersonMergeV2Impl.class.getSimpleName(), batch_list.size(), (System.currentTimeMillis() - start));
        //发送变更人员记录
        if (StringUtils.isNotBlank(data)) {
            kafkaProduce.produce(configConstant.topic_pid_update_v1, data);
        }
        return data;
    }
}