data-dump/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java

package com.celnet.datadump.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.celnet.datadump.config.SalesforceConnect;
import com.celnet.datadump.config.SalesforceExecutor;
import com.celnet.datadump.config.SalesforceTargetConnect;
import com.celnet.datadump.entity.DataField;
import com.celnet.datadump.entity.DataObject;
import com.celnet.datadump.enums.FileType;
import com.celnet.datadump.global.Const;
import com.celnet.datadump.param.DataDumpParam;
import com.celnet.datadump.param.FileTransformParam;
import com.celnet.datadump.global.SystemConfigCode;
import com.celnet.datadump.mapper.CustomMapper;
import com.celnet.datadump.service.*;
import com.celnet.datadump.util.DataUtil;
import com.celnet.datadump.util.EmailUtil;
import com.celnet.datadump.util.HttpUtil;
import com.celnet.datadump.util.OssUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sforce.soap.partner.PartnerConnection;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import okhttp3.Response;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.BooleanUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.mime.MultipartEntityBuilder;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.*;
import java.net.URLEncoder;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * @author Red
 * @description
 * @date 2022/12/26
 */
@Service
@Slf4j
public class FileServiceImpl implements FileService {

    @Autowired
    private CustomMapper customMapper;
    @Autowired
    private DataFieldService dataFieldService;
    @Autowired
    private DataObjectService dataObjectService;
    @Autowired
    private SalesforceConnect salesforceConnect;
    @Autowired
    private SalesforceTargetConnect salesforceTargetConnect;
    @Autowired
    private SalesforceExecutor salesforceExecutor;
    @Autowired
    private SystemConfigService systemConfigService;
    @Autowired
    private CommonService commonService;

    @Override
    public void verifyFile(String api, String field) {
        log.info("verify file api:{}, field:{}", api, field);
        String extraSql = "";
        if (dataFieldService.hasDeleted(api)) {
            extraSql += "AND IsDeleted = false ";
        }
        long num = 0L, successNum = 0L;
        String maxId = null;
        while (true) {
            String extraSqlTmp = extraSql;
            if (StringUtils.isNotEmpty(maxId)) {
                extraSqlTmp += "AND Id > '" + maxId + "'";
            }
            // 获取未存储的附件id
            List<Map<String, Object>> list = customMapper.list("Id, url", api, extraSqlTmp + " AND is_dump = 1 order by id asc limit 200");
            if (CollectionUtils.isEmpty(list)) {
                break;
            }
            for (Map<String, Object> map : list) {
                String id = (String) map.get("Id");
                maxId = id;
                String path = (String) map.get("url");
                switch (Const.FILE_TYPE) {
                    case SERVER:
                        path = Const.SERVER_FILE_PATH + "/" + path;
                        log.info("check file on server, id:{}, path:{}", id, path);
                        File file = new File(path);
                        boolean exists = file.exists();
                        if (!exists) {
                            num++;
                            List<Map<String, Object>> maps = Lists.newArrayList();
                            Map<String, Object> paramMap = Maps.newHashMap();
                            paramMap.put("key", "is_dump");
                            paramMap.put("value", 0);
                            maps.add(paramMap);
                            customMapper.updateById(api, maps, id);
                        }
                        break;
                    default:
                        log.error("id: {}, no mapping dump type", id);
                }
            }
            successNum += list.size();
        }
        log.info("dump file verify success, api:{}, success num:{}, retry dump num:{}", api, successNum, num);
        EmailUtil.send("DumpFile Verify Success", "api: " + api + ", success num: " + successNum + ", retry dump num: " + num);
        // 存在需重下的文件 执行一遍
        if (num > 0) {
            dumpFile(api, field, true);
        }
    }

    @Override
    public void dumpFile(String api, String field, Boolean singleThread) {
        String downloadUrl = null;
        List<Map<String, Object>> poll = customMapper.list("code,value","org_config",null);
        for (Map<String, Object> map1 : poll) {
            if ("FILE_DOWNLOAD_URL".equals(map1.get("code"))) {
                downloadUrl = (String) map1.get("value");
            }
        }
        if (StringUtils.isEmpty(downloadUrl)) {
            EmailUtil.send("DumpFile ERROR", "文件下载失败！下载地址未配置");
            return;
        }
        log.info("dump file api:{}, field:{}", api, field);
        PartnerConnection connect = salesforceConnect.createConnect();
        String token = connect.getSessionHeader().getSessionId();

        List<Future<?>> futures = Lists.newArrayList();
        String extraSql = "";
        if (dataFieldService.hasDeleted(api)) {
            extraSql += "AND IsDeleted = false ";
        }
        if (Const.FILE_TYPE == FileType.SERVER) {
            Date defaultBeginDate = DataUtil.DEFAULT_BEGIN_DATE;
            Calendar calendar = Calendar.getInstance();
            int currentYear = calendar.get(Calendar.YEAR);
            calendar.setTime(defaultBeginDate);
            int beginYear = calendar.get(Calendar.YEAR);
            
            // 检测路径是否存在 不存在则创建
            File baseDir = new File(Const.SERVER_FILE_PATH + "/" + api);
            if (!baseDir.exists()) {
                baseDir.mkdirs();
            }
            
            // 创建从开始年份到当前年份的子目录
            for (int year = beginYear; year <= currentYear; year++) {
                File yearDir = new File(Const.SERVER_FILE_PATH + "/" + api + "-" + year);
                if (!yearDir.exists()) {
                    yearDir.mkdir();
                }
            }
        }
        try {
            String name = "Name";
            if (Const.CONTENT_VERSION.equalsIgnoreCase(api)) {
                name = "Title,FileExtension";
            }
            Map<String, String> headers = Maps.newHashMap();
            headers.put("Authorization", "Bearer " + token);
            headers.put("connection", "keep-alive");
            long num = 0;
            while (true) {
                // 获取未存储的附件id
                List<Map<String, Object>> list = customMapper.list("Id,CreatedDate," + name, api, extraSql + " AND is_dump = 0 limit 1000");
                if (CollectionUtils.isEmpty(list)) {
                    log.info("无需要下载文件数据！！！");
                    break;
                }
                for (Map<String, Object> map : list) {
                    String finalDownloadUrl = downloadUrl;
                    Future<?> future = salesforceExecutor.execute(() -> {
                        String id = null;
                        // 上传完毕 更新附件信息
                        List<Map<String, Object>> maps = Lists.newArrayList();
                        boolean isDump = true;
                        int failCount = 0;
                        while (true) {
                            try {
                                String fileName;
                                id = (String) map.get("Id");
                                if (Const.CONTENT_VERSION.equalsIgnoreCase(api)) {
                                    fileName = map.get("Title") + "." + map.get("FileExtension");
                                }else {
                                    fileName = (String) map.get("Name");
                                }
                                log.info("------------文件名:"+id + "_" + fileName);
                                Object createdDateObj = map.get("CreatedDate");
                                String year = createdDateObj.toString().substring(0, 4);

                                // 判断路径是否为空
                                if (StringUtils.isNotBlank(fileName)) {
                                    String filePath = api + "-" + year + "/" + id + "_" + fileName;
                                    // 拼接url
                                    String url = finalDownloadUrl + String.format(Const.SF_FILE_URL, api, id, field);
                                    Response response = HttpUtil.doGet(url, null, headers);
                                    if (response.body() != null) {
                                        InputStream inputStream = response.body().byteStream();
                                        switch (Const.FILE_TYPE) {
                                            case OSS:
                                                // 上传到oss
                                                OssUtil.upload(inputStream, filePath);
                                                break;
                                            case SERVER:
                                                dumpToServer(headers, id, filePath, url, response, inputStream);
                                                break;
                                            default:
                                                log.error("id: {}, no mapping dump type", id);
                                        }
                                        Map<String, Object> paramMap = Maps.newHashMap();
                                        if ("Document".equals(api)){
                                            paramMap.put("key", "localUrl");
                                        }else {
                                            paramMap.put("key", "url");
                                        }
                                        paramMap.put("value", filePath);
                                        maps.add(paramMap);
                                    }
                                }
                                Map<String, Object> paramMap = Maps.newHashMap();
                                paramMap.put("key", "is_dump");
                                paramMap.put("value", isDump);
                                maps.add(paramMap);
                                customMapper.updateById(api, maps, id);
                                TimeUnit.MILLISECONDS.sleep(1);
                                break;
                            } catch (Throwable throwable) {
                                log.error("dump file error, id: {}", id, throwable);
                                failCount++;
                                if (Const.MAX_FAIL_COUNT < failCount) {
                                    {
                                        Map<String, Object> paramMap = Maps.newHashMap();
                                        paramMap.put("key", "is_dump");
                                        paramMap.put("value", 2);
                                        maps.add(paramMap);
                                    }
                                    customMapper.updateById(api, maps, id);
                                    break;
                                }
                                try {
                                    TimeUnit.SECONDS.sleep(30);
                                } catch (InterruptedException e) {
                                    throw new RuntimeException(e);
                                }
                            }
                        }
                    }, 0, 0);
                    futures.add(future);
                    // 单线程
                    if (singleThread) {
                        salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
                    }
                }
                // 多线程
                if (!singleThread) {
                    salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
                }
                num += list.size();
                log.info("dump file count api:{}, field:{}, num:{}", api, field, num);
            }
            log.info("dump file success api:{}, field:{}, num:{}", api, field, num);
        } catch (Throwable throwable) {
            log.error("dump file error", throwable);
            salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
        } finally {
            // 把is_dump为2的重置为0
            List<Map<String, Object>> maps = Lists.newArrayList();
            Map<String, Object> paramMap = Maps.newHashMap();
            paramMap.put("key", "is_dump");
            paramMap.put("value", 0);
            maps.add(paramMap);
            customMapper.update(maps, api, "is_dump = 2");
        }
    }

    @Override
    public ReturnT<String> transform(FileTransformParam param) {
        List<DataObject> list;
        // 判断api 为空则取出blob_field不为空的
        if (StringUtils.isBlank(param.getApi())) {
            LambdaQueryWrapper<DataObject> qw = Wrappers.lambdaQuery(DataObject.class);
            qw.isNotNull(DataObject::getBlobField);
            list = dataObjectService.list(qw);
        } else {
            // 不为为空则查询出api中blob_field不为空的
            LambdaQueryWrapper<DataObject> qw = Wrappers.lambdaQuery(DataObject.class);
            qw.in(DataObject::getName, param.getApi())
                    .isNotNull(DataObject::getBlobField);
            list = dataObjectService.list(qw);
        }
        // 获取转换根目录
        String path = param.getPath();
        if (StringUtils.isBlank(path)) {
            path = systemConfigService.getById(SystemConfigCode.FILE_TRANSFORM_PATH).getValue();
        }
        log.info("transform path:{}", path);
        for (DataObject dataObject : list) {
            String api = dataObject.getName();
            String apiPath = path + "/" + api;
            File root = FileUtils.getFile(apiPath);
            log.info("transform api:{}, apiPath:{}", api, apiPath);
            File[] files = root.listFiles();
            // 不为空 说明有数据
            if (ArrayUtils.isNotEmpty(files)) {
                String name = getName(api);
                assert files != null;
                Arrays.stream(files).parallel().forEach(file -> {
                    moveFile(api, name, file);
                });
            }
        }
        return ReturnT.SUCCESS;
    }

    /**
     * 文件转移并更新
     *
     * @param api  表名
     * @param name 字段名
     * @param file 目标文件
     */
    private void moveFile(String api, String name, File file) {
        String id = file.getName();
        List<Map<String, Object>> datas = customMapper.list("Id, url, is_dump," + name, api, "Id = '" + id + "'");
        // 没有这条数据
        if (CollectionUtils.isEmpty(datas)) {
            return;
        }
        Map<String, Object> data = datas.get(0);
        Boolean isDump = (Boolean) data.get("is_dump");
        if (BooleanUtils.isTrue(isDump)) {
            return;
        }
        String fileName = (String) data.get(name);
        String filePath = api + "/" + id + "_" + fileName;
        switch (Const.FILE_TYPE) {
            case OSS:
                // 上传到oss
                OssUtil.upload(file, filePath);
                break;
            case SERVER:
                try {
                    // 转移文件并改名
                    FileUtils.copyFile(file, new File(Const.SERVER_FILE_PATH + "/" + filePath));
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                break;
            default:
                log.error("id: {}, no mapping dump type", id);
        }
        // 完成后删除
        boolean delete = file.delete();
        List<Map<String, Object>> maps = Lists.newArrayList();
        {
            Map<String, Object> paramMap = Maps.newHashMap();
            paramMap.put("key", "is_dump");
            paramMap.put("value", 1);
            maps.add(paramMap);
        }
        {
            Map<String, Object> paramMap = Maps.newHashMap();
            paramMap.put("key", "url");
            paramMap.put("value", filePath);
            maps.add(paramMap);
        }
        customMapper.updateById(api, maps, id);
        try {
            TimeUnit.MILLISECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 下载文件到服务器
     *
     * @param headers     请求头
     * @param id          id
     * @param filePath    文件路径
     * @param url         链接
     * @param response    响应
     * @param inputStream 流
     * @throws IOException exception
     */
    private void dumpToServer(Map<String, String> headers, String id, String filePath, String url, Response response, InputStream inputStream) throws IOException {
        String path = Const.SERVER_FILE_PATH + "/" + filePath;
        log.info("--------文件名称："+path);
        long offset = 0L;
        RandomAccessFile accessFile = new RandomAccessFile(path, "rw");
        while (true) {
            try {
                // 保存到本地
                byte[] buf = new byte[8192];
                int len = 0;
                if (offset > 0) {
                    inputStream.skip(offset);
                    accessFile.seek(offset);
                }
                while ((len = inputStream.read(buf)) != -1) {
                    accessFile.write(buf, 0, len);
                    offset += len;
                }
                break;
            } catch (Exception e) {
                if (offset <= 0) {
                    throw e;
                }
                System.out.println("file dump to server EOF ERROR try to reconnect");
                response.close();
                response = HttpUtil.doGet(url, null, headers);
                assert response.body() != null;
                inputStream = response.body().byteStream();
            }
        }
        accessFile.close();
    }

    /**
     * 获取附件名称
     *
     * @param api 表明
     * @return name
     */
    private static String getName(String api) {
        // 默认name
        String name = "Name";
        // contentVersion使用PathOnClient
        if (Const.CONTENT_VERSION.equalsIgnoreCase(api)) {
            name = "Title,FileExtension";
        }
        return name;
    }

    @Override
    public void uploadFile(String api, String field, Boolean singleThread) {
        String uploadUrl = null;
        List<Map<String, Object>> poll = customMapper.list("code,value","org_config",null);
        for (Map<String, Object> map1 : poll) {
            if ("FILE_UPLOAD_URL".equals(map1.get("code"))) {
                uploadUrl = (String) map1.get("value");
            }
        }
        if (StringUtils.isBlank(uploadUrl)) {
            EmailUtil.send("UploadFile ERROR", "文件上传失败！上传地址未配置");
            return;
        }

        log.info("upload file api:{}, field:{}", api, field);
        PartnerConnection connect = salesforceTargetConnect.createConnect();
        String token = connect.getSessionHeader().getSessionId();

        List<Future<?>> futures = Lists.newArrayList();
        String extraSql = "";
        if (dataFieldService.hasDeleted(api)) {
            extraSql += "AND IsDeleted = false ";
        }
        if (Const.FILE_TYPE == FileType.SERVER) {
            // 检测路径是否存在
            File excel = new File(Const.SERVER_FILE_PATH + "/" + api);
            if (!excel.exists()) {
                throw new RuntimeException("找不到文件路径");
            }
        }
        String selectField = "Id, url, PathOnClient, Title, ContentDocumentId, VersionNumber";

        QueryWrapper<DataField> wrapper = new QueryWrapper<>();
        wrapper.eq("api", api);
        wrapper.eq("is_createable",1);
        List<DataField> dataFields = dataFieldService.list(wrapper);

        ArrayList<String> addFieldList = new ArrayList<>();
        for (DataField dataField : dataFields) {
            if (!selectField.contains(dataField.getField())){
                selectField += "," + dataField.getField();
                addFieldList.add(dataField.getField());
            }
        }

        try {
            List<Map<String, Object>> documentList = customMapper.list("Id", "ContentDocument", "new_id is null");
            for (Map<String, Object> documentMap : documentList) {
                boolean index = false;
                String documentId = (String) documentMap.get("Id");

                // 获取未存储的附件id
                List<Map<String, Object>> list = customMapper.list(selectField, api, "ContentDocumentId = '" + documentId + "' and new_id is null ORDER BY VersionNumber ASC");
                if (CollectionUtils.isEmpty(list)) {
                    continue;
                }
                String finalUploadUrl = uploadUrl;
                Future<?> future = salesforceExecutor.execute(() -> {
                    String newDocumentId = null;
                    for (Map<String, Object> map : list) {
                        String id = null;
                        // 上传完毕 更新附件信息
                        List<Map<String, Object>> maps = Lists.newArrayList();
                        boolean isUpload = true;
                        int failCount = 0;
                        CloseableHttpResponse response = null;
                        String respContent = null;
                        while (true) {
                            try {
                                id = (String) map.get("Id");
                                String url_fileName = (String) map.get("url");
                                String fileName = (String) map.get("PathOnClient");
                                String title = (String) map.get("Title");
                                String oldDocumentId = (String) map.get("ContentDocumentId");
                                int versionNumber = Integer.parseInt(map.get("VersionNumber").toString());
                                log.info("文件名称：" + fileName);

                                // 判断路径是否为空
                                if (StringUtils.isNotBlank(url_fileName)) {
                                    String filePath = Const.SERVER_FILE_PATH + "/" + url_fileName;
                                    File file = new File(filePath);
                                    boolean exists = file.exists();
                                    if (!exists) {
                                        log.info("文件不存在");
                                        break;
                                    }
                                    // 拼接url
                                    String url = finalUploadUrl + String.format(Const.SF_UPLOAD_FILE_URL, api);
                                    HttpPost httpPost = new HttpPost(url);
                                    httpPost.setHeader("Authorization", "Bearer " + token);
                                    httpPost.setHeader("connection", "keep-alive");

                                    JSONObject credentialsJsonParam = new JSONObject();
                                    credentialsJsonParam.put("title", title);
                                    credentialsJsonParam.put("pathOnClient", fileName);
                                    if (newDocumentId != null) {
                                        credentialsJsonParam.put("ContentDocumentId", newDocumentId);
                                    }
                                    if (!addFieldList.isEmpty()){
                                        for (String s : addFieldList) {
                                            credentialsJsonParam.put(s, map.get(s));
                                        }
                                    }
                                    MultipartEntityBuilder builder = MultipartEntityBuilder.create();
                                    builder.addTextBody("entity_content", credentialsJsonParam.toJSONString(), ContentType.APPLICATION_JSON);
                                    builder.addBinaryBody("VersionData", new FileInputStream(file), ContentType.APPLICATION_OCTET_STREAM, fileName);
                                    HttpEntity entity = builder.build();
                                    httpPost.setEntity(entity);

                                    CloseableHttpClient httpClient = HttpClients.createDefault();
                                    response = httpClient.execute(httpPost);
                                    if (response != null && response.getStatusLine() != null && response.getStatusLine().getStatusCode() < 400) {
                                        HttpEntity he = response.getEntity();
                                        if (he != null) {
                                            respContent = EntityUtils.toString(he, "UTF-8");
                                            String newId = String.valueOf(JSONObject.parseObject(respContent).get("id"));
                                            if (StringUtils.isBlank(newId)) {
                                                log.error("文件上传错误，返回实体信息：" + respContent);
                                                break;
                                            }
                                            Map<String, Object> paramMap = Maps.newHashMap();
                                            paramMap.put("key", "new_id");
                                            paramMap.put("value", newId);
                                            maps.add(paramMap);

                                            if (versionNumber == 1) {
                                                // 更新documentId
                                                String dId = commonService.getDocumentId(connect, newId);
                                                List<Map<String, Object>> dList = new ArrayList<>();
                                                Map<String, Object> dMap = Maps.newHashMap();
                                                dMap.put("key", "new_id");
                                                dMap.put("value", dId);
                                                dList.add(dMap);
                                                customMapper.updateById("ContentDocument", dList, oldDocumentId);

                                                newDocumentId = dId;
                                            }
                                        }else {
                                            break;
                                        }
                                    } else {
                                        log.info("文件上传错误，返回：" + JSON.toJSONString(response));
                                        throw new RuntimeException();
                                    }
                                }else {
                                    log.info("文件路径URL为空，数据库执行SQL！！！");
                                }
                                Map<String, Object> paramMap = Maps.newHashMap();
                                paramMap.put("key", "is_upload");
                                paramMap.put("value", isUpload);
                                maps.add(paramMap);
                                customMapper.updateById(api, maps, id);
                                TimeUnit.MILLISECONDS.sleep(1);
                                break;
                            } catch (Exception throwable) {
                                log.error("upload file error, id: {}, 返回实体信息：{}", id,respContent, throwable);
                                failCount++;
                                if (Const.MAX_FAIL_COUNT < failCount) {
                                    {
                                        Map<String, Object> paramMap = Maps.newHashMap();
                                        paramMap.put("key", "is_upload");
                                        paramMap.put("value", 2);
                                        maps.add(paramMap);
                                    }
                                    customMapper.updateById(api, maps, id);
                                    break;
                                }
                                if (response != null) {
                                    try {
                                        response.close();
                                    } catch (IOException e) {
                                        log.error("exception message", e); 
                                    }
                                }
                                try {
                                    TimeUnit.SECONDS.sleep(30);
                                } catch (InterruptedException e) {
                                    throw new RuntimeException(e);
                                }
                            }
                        }
                }
            }, 0, 0);
                futures.add(future);
                // 单线程
                if (singleThread) {
                    salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
                }
            }
            // 多线程
            if (!singleThread) {
                salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
            }

            log.info("upload file success api:{}, field:{}", api, field);
        } catch (Throwable throwable) {
            log.error("upload file error", throwable);
            salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
        } finally {
            // 把is_dump为2的重置为0
            List<Map<String, Object>> maps = Lists.newArrayList();
            Map<String, Object> paramMap = Maps.newHashMap();
            paramMap.put("key", "is_upload");
            paramMap.put("value", 0);
            maps.add(paramMap);
            customMapper.update(maps, api, "is_upload = 2");
        }
    }

    @Override
    public void uploadFileToAttachment(String api, String field, Boolean singleThread) {
        String uploadUrl = null;
        List<Map<String, Object>> poll = customMapper.list("code,value","org_config",null);
        for (Map<String, Object> map1 : poll) {
            if ("FILE_UPLOAD_URL".equals(map1.get("code"))) {
                uploadUrl = (String) map1.get("value");
            }
        }
        if (StringUtils.isBlank(uploadUrl)) {
            EmailUtil.send("UploadFile ERROR", "文件上传失败！上传地址未配置");
            return;
        }
        log.info("upload file api:{}, field:{}", api, field);
        PartnerConnection connect = salesforceTargetConnect.createConnect();
        String token = connect.getSessionHeader().getSessionId();

        List<Future<?>> futures = Lists.newArrayList();
        String extraSql = "";
        if (dataFieldService.hasDeleted(api)) {
            extraSql += "AND IsDeleted = false ";
        }
        if (Const.FILE_TYPE == FileType.SERVER) {
            // 检测路径是否存在
            File excel = new File(Const.SERVER_FILE_PATH + "/" + api);
            if (!excel.exists()) {
                throw new RuntimeException("找不到文件路径");
            }
        }

        String selectField = "Id,url,ParentId,Parent_Type,OwnerId,Description,CreatedById,Name";

        QueryWrapper<DataField> wrapper = new QueryWrapper<>();
        wrapper.eq("api", api);
        wrapper.eq("is_createable",1);
        List<DataField> dataFields = dataFieldService.list(wrapper);

        ArrayList<String> addFieldList = new ArrayList<>();
        for (DataField dataField : dataFields) {
            if (!selectField.contains(dataField.getField())){
                selectField += "," + dataField.getField();
                addFieldList.add(dataField.getField());
            }
        }

        try {
            // 获取未存储的附件id
            List<Map<String, Object>> list = customMapper.list(selectField, api, "is_upload = 0");
            for (Map<String, Object> map : list) {
                String finalUploadUrl = uploadUrl;
                Future<?> future = salesforceExecutor.execute(() -> {
                    String id = null;
                    // 上传完毕 更新附件信息
                    List<Map<String, Object>> maps = Lists.newArrayList();
                    boolean isUpload = true;
                    int failCount = 0;
                    CloseableHttpResponse response = null;
                    String respContent = null;
                    while (true) {
                        try {
                            id = (String) map.get("Id");
                            String url_fileName = (String) map.get("url");
                            String parentId = (String) map.get("ParentId");
                            String parentType = (String) map.get("Parent_Type");
                            String ownerId = (String) map.get("OwnerId");
                            String description = (String) map.get("Description");
                            String createdById = (String) map.get("CreatedById");
                            String fileName = (String) map.get("Name");

                            // 判断路径是否为空
                            if (StringUtils.isNotBlank(url_fileName)) {
                                String filePath = Const.SERVER_FILE_PATH + "/" + url_fileName;
                                File file = new File(filePath);
                                boolean exists = file.exists();
                                if (!exists) {
                                    log.info("文件不存在");
                                    break;
                                }

                                // dataObject查询
                                QueryWrapper<DataObject> qw = new QueryWrapper<>();
                                qw.eq("name", parentType);
                                List<DataObject> objects = dataObjectService.list(qw);
                                if (objects.isEmpty()) {
                                    String format = "文件ID：" + id + "关联对象不存在：" + parentType;
                                    log.info(format);
                                    Map<String, Object> paramMap = Maps.newHashMap();
                                    paramMap.put("key", "is_upload");
                                    paramMap.put("value", 2);
                                    maps.add(paramMap);
                                    Map<String, Object> paramMap1 = Maps.newHashMap();
                                    paramMap1.put("key", "error_message");
                                    paramMap1.put("value",format);
                                    maps.add(paramMap1);
                                    customMapper.updateById(api, maps, id);
                                    break;
                                }

                                Map<String, Object> lMap = customMapper.getById("new_id",parentType, String.valueOf(parentId));
                                if(lMap == null) {
                                    String format = "文件ID：" + id + ",对应关联对象类型：" + parentType + "，对应关联对象ID：" + parentId + "数据不存在！！！！";
                                    log.info(format);
                                    Map<String, Object> paramMap = Maps.newHashMap();
                                    paramMap.put("key", "is_upload");
                                    paramMap.put("value", 2);
                                    maps.add(paramMap);
                                    Map<String, Object> paramMap1 = Maps.newHashMap();
                                    paramMap1.put("key", "error_message");
                                    paramMap1.put("value",format);
                                    maps.add(paramMap1);
                                    customMapper.updateById(api, maps, id);
                                    break;
                                }

                                Map<String, Object> uMap = customMapper.getById("new_id","User", ownerId);
                                if(uMap == null) {
                                    String format = "文件ID：" + id + ",对应用户ID：" + ownerId + "数据不存在！！！！";
                                    log.info(format);
                                    Map<String, Object> paramMap = Maps.newHashMap();
                                    paramMap.put("key", "is_upload");
                                    paramMap.put("value", 2);
                                    maps.add(paramMap);
                                    Map<String, Object> paramMap1 = Maps.newHashMap();
                                    paramMap1.put("key", "error_message");
                                    paramMap1.put("value",format);
                                    maps.add(paramMap1);
                                    customMapper.updateById(api, maps, id);
                                    break;
                                }

                                Map<String, Object> cMap = customMapper.getById("new_id","User", createdById);
                                if(cMap == null) {
                                    String format = "文件ID：" + id + ",对应用户ID：" + createdById + "数据不存在！！！！";
                                    log.info(format);
                                    Map<String, Object> paramMap = Maps.newHashMap();
                                    paramMap.put("key", "is_upload");
                                    paramMap.put("value", 2);
                                    maps.add(paramMap);
                                    Map<String, Object> paramMap1 = Maps.newHashMap();
                                    paramMap1.put("key", "error_message");
                                    paramMap1.put("value",format);
                                    maps.add(paramMap1);
                                    customMapper.updateById(api, maps, id);
                                    break;
                                }

                                // 拼接url
                                String url = finalUploadUrl + String.format(Const.SF_UPLOAD_FILE_URL, api);
                                HttpPost httpPost = new HttpPost(url);
                                httpPost.setHeader("Authorization", "Bearer " + token);
                                httpPost.setHeader("connection", "keep-alive");

                                JSONObject credentialsJsonParam = new JSONObject();
                                credentialsJsonParam.put("parentId", lMap.get("new_id"));
                                credentialsJsonParam.put("Name", fileName);
                                credentialsJsonParam.put("Description", description);
                                credentialsJsonParam.put("OwnerId", uMap.get("new_id"));
//                                credentialsJsonParam.put("CreatedDate", map.get("CreatedDate"));
                                credentialsJsonParam.put("CreatedById", cMap.get("new_id"));

                                if (!addFieldList.isEmpty()){
                                    for (String s : addFieldList) {
                                        credentialsJsonParam.put(s, map.get(s));
                                    }
                                }

                                MultipartEntityBuilder builder = MultipartEntityBuilder.create();
                                builder.addTextBody("data", credentialsJsonParam.toJSONString(), ContentType.APPLICATION_JSON);
                                builder.addBinaryBody("Body", new FileInputStream(file), ContentType.APPLICATION_OCTET_STREAM, fileName);
                                HttpEntity entity = builder.build();
                                httpPost.setEntity(entity);

                                CloseableHttpClient httpClient = HttpClients.createDefault();
                                response = httpClient.execute(httpPost);
                                if (response != null && response.getStatusLine() != null && response.getStatusLine().getStatusCode() < 400) {
                                    HttpEntity he = response.getEntity();
                                    if (he != null) {
                                        respContent = EntityUtils.toString(he, "UTF-8");
                                        String newId = JSONObject.parseObject(respContent).get("id").toString();
                                        Map<String, Object> paramMap = Maps.newHashMap();
                                        paramMap.put("key", "new_id");
                                        paramMap.put("value", newId);
                                        maps.add(paramMap);
                                    }
                                } else {
                                    {
                                        Map<String, Object> paramMap = Maps.newHashMap();
                                        paramMap.put("key", "is_upload");
                                        paramMap.put("value", 2);
                                        maps.add(paramMap);
                                        Map<String, Object> paramMap1 = Maps.newHashMap();
                                        paramMap1.put("key", "error_message");
                                        paramMap1.put("value", response == null?"上传失败" :EntityUtils.toString(response.getEntity(), "UTF-8"));
                                        maps.add(paramMap1);
                                    }
                                    customMapper.updateById(api, maps, id);
                                    log.info("---------文件ID：" + id + "上传失败--------" );
                                    throw new RuntimeException();
                                }
                            }
                            Map<String, Object> paramMap = Maps.newHashMap();
                            paramMap.put("key", "is_upload");
                            paramMap.put("value", isUpload);
                            maps.add(paramMap);
                            customMapper.updateById(api, maps, id);
                            TimeUnit.MILLISECONDS.sleep(1);
                            break;
                        } catch (Exception throwable) {
                            log.error("upload file error, id: {}", id, throwable);
                            failCount++;
                            if (Const.MAX_FAIL_COUNT < failCount) {
                                Map<String, Object> paramMap = Maps.newHashMap();
                                paramMap.put("key", "is_upload");
                                paramMap.put("value", 2);
                                maps.add(paramMap);
                                customMapper.updateById(api, maps, id);
                                break;
                            }
                            if (response != null) {
                                try {
                                    response.close();
                                } catch (IOException e) {
                                    log.error("exception message", e);
                                }
                            }
                            try {
                                TimeUnit.SECONDS.sleep(30);
                            } catch (InterruptedException e) {
                                throw new RuntimeException(e);
                            }
                        }
                    }

                }, 0, 0);
                futures.add(future);
            }

            salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));

            log.info("upload file success api:{}, field:{}", api, field);
        } catch (Throwable throwable) {
            log.error("upload file error", throwable);
            salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
        } finally {
            // 把is_upload为2的重置为0
            List<Map<String, Object>> maps = Lists.newArrayList();
            Map<String, Object> paramMap = Maps.newHashMap();
            paramMap.put("key", "is_upload");
            paramMap.put("value", 0);
            maps.add(paramMap);
            customMapper.update(maps, api, "is_upload = 2");
        }
    }

}