kris/data-dump
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
823f810702
data-dump/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java

774 lines
37 KiB
Java
Raw Normal View History

【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
package com.celnet.datadump.service.impl;
【feat】调整文件上传下载，将上传下载链接配置在数据库
2025-06-17 15:41:28 +08:00
import com.alibaba.fastjson.JSON;
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
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.DataObject;
import com.celnet.datadump.enums.FileType;
import com.celnet.datadump.global.Const;
【feat】调整文件上传下载，将上传下载链接配置在数据库
2025-06-17 15:41:28 +08:00
import com.celnet.datadump.param.DataDumpParam;
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
import com.celnet.datadump.param.FileTransformParam;
import com.celnet.datadump.global.SystemConfigCode;
import com.celnet.datadump.mapper.CustomMapper;
import com.celnet.datadump.service.*;
【feat】调整文件上传下载，将上传下载链接配置在数据库
2025-06-17 15:41:28 +08:00
import com.celnet.datadump.util.EmailUtil;
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
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.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
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 = true order by id asc limit 10");
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++;
log.warn("file on server, id:{}, path:{} doesn't exist retry dump later", id, path);
List<Map<String, Object>> maps = Lists.newArrayList();
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", "is_dump");
paramMap.put("value", false);
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);
XxlJobLogger.log("dump file verify success, api:{}, success num:{}, retry dump num:{}", api, successNum, num);
// 存在需重下的文件 执行一遍
if (num > 0) {
dumpFile(api, field, true);
}
}
@Override
public void dumpFile(String api, String field, Boolean singleThread) {
【feat】调整文件上传下载，将上传下载链接配置在数据库
2025-06-17 15:41:28 +08:00
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");
}
}
【feat】优化文件上传异常信息提示 (cherry picked from commit ba34877b02088baac0c812570a086aaaed0cbf45)
2025-06-24 18:57:34 +08:00
if (StringUtils.isEmpty(downloadUrl)) {
【feat】优化文件地址配置失效提示方式
2025-06-24 10:52:55 +08:00
EmailUtil.send("DumpFile ERROR", "文件下载失败！下载地址未配置");
return;
}
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
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) {
// 检测路径是否存在 不存在则创建
File excel = new File(Const.SERVER_FILE_PATH + "/" + api);
if (!excel.exists()) {
boolean mkdir = excel.mkdir();
}
}
try {
String name = getName(api);
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, " + name, api, extraSql + " AND is_dump = false limit 1000");
if (CollectionUtils.isEmpty(list)) {
break;
}
String finalName = name;
for (Map<String, Object> map : list) {
【feat】调整文件上传下载，将上传下载链接配置在数据库
2025-06-17 15:41:28 +08:00
String finalDownloadUrl = downloadUrl;
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
Future<?> future = salesforceExecutor.execute(() -> {
String id = null;
// 上传完毕 更新附件信息
List<Map<String, Object>> maps = Lists.newArrayList();
boolean isDump = true;
int failCount = 0;
while (true) {
try {
id = (String) map.get("Id");
String fileName = (String) map.get(finalName);
log.info("------------文件名:"+id + "_" + fileName);
// 判断路径是否为空
if (StringUtils.isNotBlank(fileName)) {
String filePath = api + "/" + id + "_" + fileName;
// 拼接url
【feat】调整文件上传下载，将上传下载链接配置在数据库
2025-06-17 15:41:28 +08:00
String url = finalDownloadUrl + String.format(Const.SF_FILE_URL, api, id, field);
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
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();
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)) {
log.warn("api:{} id {} not exist", api, id);
return;
}
Map<String, Object> data = datas.get(0);
Boolean isDump = (Boolean) data.get("is_dump");
if (BooleanUtils.isTrue(isDump)) {
log.warn("api:{} id {} isDump continue", api, id);
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;
}
log.warn("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();
log.warn("reconnect success, id:{} skip {}", id, offset);
}
}
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 = "PathOnClient";
}
return name;
}
@Override
public void uploadFile(String api, String field, Boolean singleThread) {
【feat】调整文件上传下载，将上传下载链接配置在数据库
2025-06-17 15:41:28 +08:00
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");
}
}
【feat】优化文件地址配置失效提示方式
2025-06-24 10:52:55 +08:00
if (StringUtils.isBlank(uploadUrl)) {
EmailUtil.send("UploadFile ERROR", "文件上传失败！上传地址未配置");
return;
}
【feat】调整文件上传下载，将上传下载链接配置在数据库
2025-06-17 15:41:28 +08:00
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
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("找不到文件路径");
}
}
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("Id, url, PathOnClient, Title, ContentDocumentId, VersionNumber", api, "ContentDocumentId = '" + documentId + "' and new_id is null ORDER BY VersionNumber ASC");
if (CollectionUtils.isEmpty(list)) {
continue;
}
【feat】调整文件上传下载，将上传下载链接配置在数据库
2025-06-17 15:41:28 +08:00
String finalUploadUrl = uploadUrl;
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
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");
Integer versionNumber = Integer.valueOf(map.get("VersionNumber").toString());
// 判断路径是否为空
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
【feat】调整文件上传下载，将上传下载链接配置在数据库
2025-06-17 15:41:28 +08:00
String url = finalUploadUrl + String.format(Const.SF_UPLOAD_FILE_URL, api);
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
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);
}
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");
【feat】优化文件上传异常信息提示 (cherry picked from commit ba34877b02088baac0c812570a086aaaed0cbf45)
2025-06-24 18:57:34 +08:00
String newId = String.valueOf(JSONObject.parseObject(respContent).get("Id"));
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
if (StringUtils.isBlank(newId)) {
【feat】优化文件上传异常信息提示 (cherry picked from commit ba34877b02088baac0c812570a086aaaed0cbf45)
2025-06-24 18:57:34 +08:00
log.error("文件上传错误，返回实体信息：" + respContent);
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
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 {
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) {
【feat】对接SFBulk，大批量Create数据方法增加
2025-06-06 10:29:08 +08:00
log.error("exception message", e);
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
}
}
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) {
【feat】调整文件上传下载，将上传下载链接配置在数据库
2025-06-17 15:41:28 +08:00
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");
}
}
【feat】优化文件地址配置失效提示方式
2025-06-24 10:52:55 +08:00
if (StringUtils.isBlank(uploadUrl)) {
EmailUtil.send("UploadFile ERROR", "文件上传失败！上传地址未配置");
return;
}
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
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("找不到文件路径");
}
}
try {
// 获取未存储的附件id
List<Map<String, Object>> list = customMapper.list("Id, ParentId, Name, url, Description, Parent_type", api, "is_upload = 0");
for (Map<String, Object> map : list) {
【feat】调整文件上传下载，将上传下载链接配置在数据库
2025-06-17 15:41:28 +08:00
String finalUploadUrl = uploadUrl;
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
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");
// 判断路径是否为空
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;
}
String fileName = file.getName();
// dataObject查询
QueryWrapper<DataObject> qw = new QueryWrapper<>();
qw.eq("name", parentType);
List<DataObject> objects = dataObjectService.list(qw);
if (objects.size() == 0) {
log.info("关联对象不存在： {}", parentType);
break;
}
Map<String, Object> lMap = customMapper.getById("new_id",parentType, parentId);
// 拼接url
【feat】调整文件上传下载，将上传下载链接配置在数据库
2025-06-17 15:41:28 +08:00
String url = finalUploadUrl + String.format(Const.SF_UPLOAD_FILE_URL, api);
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
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);
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 {
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) {
【feat】对接SFBulk，大批量Create数据方法增加
2025-06-06 10:29:08 +08:00
log.error("exception message", e);
【feat】项目代码初始化
2025-03-28 17:38:34 +08:00
}
}
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");
}
}
}
Reference in New Issue Copy Permalink