wzh
/
dataservice-yy
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

优化上传

ftpmain
wang 3 weeks ago
parent
a83572f109
commit
8bbe0d2c04
4 changed files with 227 additions and 24 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 54
      sql/mid_ftp_processed_file.sql
  2. 142
      src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java
  3. 44
      src/main/java/com/threecloud/dataserviceyy/util/FileCleaner.java
  4. 11
      src/main/java/com/threecloud/dataserviceyy/util/FileUploadUtil.java

54
sql/mid_ftp_processed_file.sql
View File

@ -0,0 +1,54 @@
-- ============================================
-- FTP已处理文件记录表（普通表）
-- 用于记录已处理过的FTP txt文件，避免重复处理
-- ============================================
CREATE TABLE "mid_voice"."mid_ftp_processed_file" (
"id" int8 NOT NULL GENERATED ALWAYS AS IDENTITY (
INCREMENT 1
MINVALUE 1
MAXVALUE 9223372036854775807
START 1
CACHE 1
),
"city_code" varchar(20) COLLATE "pg_catalog"."default" NOT NULL,
"city_name" varchar(100) COLLATE "pg_catalog"."default",
"file_name" varchar(255) COLLATE "pg_catalog"."default" NOT NULL,
"file_size" int8,
"process_status" varchar(10) COLLATE "pg_catalog"."default" NOT NULL DEFAULT '1',
"record_count" int4 DEFAULT 0,
"success_count" int4 DEFAULT 0,
"fail_count" int4 DEFAULT 0,
"process_time" int8 DEFAULT 0,
"error_msg" text COLLATE "pg_catalog"."default",
"create_time" timestamp(6) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "mid_ftp_processed_file_pkey" PRIMARY KEY ("id"),
CONSTRAINT "uk_ftp_file_city_name" UNIQUE ("city_code", "file_name")
);
ALTER TABLE "mid_voice"."mid_ftp_processed_file"
OWNER TO "dcms_dev";
-- 创建索引
CREATE INDEX "idx_ftp_processed_file_name" ON "mid_voice"."mid_ftp_processed_file" USING btree (
"file_name" COLLATE "pg_catalog"."default" "pg_catalog"."text_ops" ASC NULLS LAST
);
CREATE INDEX "idx_ftp_processed_create_time" ON "mid_voice"."mid_ftp_processed_file" USING btree (
"create_time" "pg_catalog"."timestamp_ops" ASC NULLS LAST
);
-- 添加注释
COMMENT ON TABLE "mid_voice"."mid_ftp_processed_file" IS 'FTP已处理文件记录表';
COMMENT ON COLUMN "mid_voice"."mid_ftp_processed_file"."id" IS '自增ID主键';
COMMENT ON COLUMN "mid_voice"."mid_ftp_processed_file"."city_code" IS '地市编码';
COMMENT ON COLUMN "mid_voice"."mid_ftp_processed_file"."city_name" IS '地市名称';
COMMENT ON COLUMN "mid_voice"."mid_ftp_processed_file"."file_name" IS 'FTP文件名';
COMMENT ON COLUMN "mid_voice"."mid_ftp_processed_file"."file_size" IS '文件大小（字节）';
COMMENT ON COLUMN "mid_voice"."mid_ftp_processed_file"."process_status" IS '处理状态：0失败，1成功';
COMMENT ON COLUMN "mid_voice"."mid_ftp_processed_file"."record_count" IS '记录总数';
COMMENT ON COLUMN "mid_voice"."mid_ftp_processed_file"."success_count" IS '成功入库数';
COMMENT ON COLUMN "mid_voice"."mid_ftp_processed_file"."fail_count" IS '失败数';
COMMENT ON COLUMN "mid_voice"."mid_ftp_processed_file"."process_time" IS '处理耗时（毫秒）';
COMMENT ON COLUMN "mid_voice"."mid_ftp_processed_file"."error_msg" IS '错误信息';
COMMENT ON COLUMN "mid_voice"."mid_ftp_processed_file"."create_time" IS '创建时间';

142
src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java
View File

@ -2,7 +2,9 @@ package com.threecloud.dataserviceyy.service;
import com.threecloud.dataserviceyy.config.FtpSyncProperties;
import com.threecloud.dataserviceyy.config.FtpSyncProperties.FtpCityConfig;
import com.threecloud.dataserviceyy.entity.FtpProcessedFile;
import com.threecloud.dataserviceyy.entity.MidVoiceCallRecord;
import com.threecloud.dataserviceyy.mapper.FtpProcessedFileMapper;
import com.threecloud.dataserviceyy.mapper.MidVoiceCallRecordMapper;
import com.threecloud.dataserviceyy.util.DateUtil;
import com.threecloud.dataserviceyy.util.FileCleaner;
@ -63,6 +65,8 @@ public class FtpSyncService {
@Autowired
private MidVoiceCallRecordMapper callRecordMapper;
@Autowired
private FtpProcessedFileMapper processedFileMapper;
@Autowired
private FileUploadUtil fileUploadUtil;
/**
@ -132,13 +136,25 @@ public class FtpSyncService {
int txtCount = FtpUtil.countTxtFiles(files);
logger.info("【FTP地市】目录 {} 下有 {} 个txt文件", city.getFtpSourceDir(), txtCount);
// 批量查询已处理的文件名，避免重复处理
Set<String> processedFiles = loadProcessedFileNames(city.getCityCode());
logger.info("【FTP地市】已处理文件记录: {} 个", processedFiles.size());
int fileSuccess = 0;
int fileFail = 0;
int fileSkipped = 0;
for (FTPFile file : files) {
if (!FtpUtil.isTxtFile(file)) {
continue;
}
String fileName = file.getName();
// 检查是否已处理过
if (processedFiles.contains(fileName)) {
logger.debug("【FTP文件】已处理过，跳过: {}", fileName);
fileSkipped++;
continue;
}
try {
if (processTxtFile(ftp, file, city, existingIds)) {
fileSuccess++;
@ -147,11 +163,11 @@ public class FtpSyncService {
}
} catch (Exception e) {
fileFail++;
logger.error("【FTP地市】处理文件失败: {}, 原因={}", file.getName(), e.getMessage());
logger.error("【FTP地市】处理文件失败: {}, 原因={}", fileName, e.getMessage());
}
}
logger.info("【FTP地市】{} 同步完成: 成功{}个文件, 失败{}个文件",
city.getCityName(), fileSuccess, fileFail);
logger.info("【FTP地市】{} 同步完成: 成功{}个文件, 跳过{}个(已处理), 失败{}个文件",
city.getCityName(), fileSuccess, fileSkipped, fileFail);
} finally {
FtpUtil.disconnect(ftp);
}
@ -166,13 +182,16 @@ public class FtpSyncService {
Set<String> existingIds) throws Exception {
String fileName = file.getName();
String sourcePath = joinPath(city.getFtpSourceDir(), fileName);
logger.info("【FTP文件】开始处理: {}", fileName);
long fileSize = file.getSize();
long fileStartTime = System.currentTimeMillis();
logger.info("【FTP文件】开始处理: {} ({} 字节)", fileName, fileSize);
byte[] txtBytes = FtpUtil.downloadFile(ftp, sourcePath);
int lineCount = 0;
int successCount = 0;
int skipCount = 0;
int failCount = 0;
String errorMsg = null;
try (BufferedReader reader = new BufferedReader(
new InputStreamReader(new ByteArrayInputStream(txtBytes), properties.getTxtEncoding()))) {
@ -190,25 +209,31 @@ public class FtpSyncService {
}
} catch (Exception e) {
failCount++;
errorMsg = e.getMessage();
logger.error("【FTP文件】解析单行失败: {}, 原因={}", line, e.getMessage());
}
}
}
logger.info("【FTP文件】{} 解析完成: 共{}行, 新增{}条, 跳过{}条, 失败{}条",
fileName, lineCount, successCount, skipCount, failCount);
// 移动到归档目录
try {
String archivePath = joinPath(city.getFtpArchiveDir(), fileName);
FtpUtil.ensureDir(ftp, city.getFtpArchiveDir());
FtpUtil.rename(ftp, sourcePath, archivePath);
logger.info("【FTP文件】已归档: {} -> {}", sourcePath, archivePath);
} catch (Exception e) {
logger.error("【FTP文件】归档失败: {}, 原因={}", fileName, e.getMessage());
}
return true;
long processTime = System.currentTimeMillis() - fileStartTime;
logger.info("【FTP文件】{} 解析完成: 共{}行, 新增{}条, 跳过{}条, 失败{}条, 耗时{}ms",
fileName, lineCount, successCount, skipCount, failCount, processTime);
// 记录已处理的文件（避免下次重复处理）
recordProcessedFile(city, fileName, fileSize, lineCount, successCount, failCount, processTime, errorMsg);
// 注意：不进行归档操作，文件保留在原位置
// 如需归档，请取消下面代码的注释
// try {
// String archivePath = joinPath(city.getFtpArchiveDir(), fileName);
// FtpUtil.ensureDir(ftp, city.getFtpArchiveDir());
// FtpUtil.rename(ftp, sourcePath, archivePath);
// logger.info("【FTP文件】已归档: {} -> {}", sourcePath, archivePath);
// } catch (Exception e) {
// logger.error("【FTP文件】归档失败: {}, 原因={}", fileName, e.getMessage());
// }
return failCount == 0;
}
/**
@ -275,9 +300,34 @@ public class FtpSyncService {
String ossPath = FilePathUtil.buildOssPath(city.getCityCode(), callStartTime, recordFileName);
// 下载录音文件（FTP路径：{ftp-record-dir}/{yyyyMMdd}/{filename}）
String remoteRecordPath = joinPath(city.getFtpRecordDir(), dateStr, recordFileName);
// 先切换到录音文件根目录，避免工作目录在别处导致下载失败
String recordDir = city.getFtpRecordDir();
// 去掉末尾斜杠，避免双斜杠
if (recordDir != null && recordDir.endsWith("/")) {
recordDir = recordDir.substring(0, recordDir.length() - 1);
}
logger.info("【FTP行】准备切换目录: {}", recordDir);
FtpUtil.changeDir(ftp, recordDir);
String remoteRecordPath = dateStr + "/" + recordFileName;
logger.info("【FTP行】准备下载录音: {}/{}", recordDir, remoteRecordPath);
byte[] recordBytes = FtpUtil.downloadFile(ftp, remoteRecordPath);
logger.info("【FTP行】下载录音: {} ({} 字节)", recordFileName, recordBytes.length);
// 为了方便您测试验证链路，这里加上一段代码，把从FTP读到的录音直接存在本地硬盘
try {
java.nio.file.Path testSaveDir = java.nio.file.Paths.get("download_test_records", dateStr);
if (!java.nio.file.Files.exists(testSaveDir)) {
java.nio.file.Files.createDirectories(testSaveDir);
}
java.nio.file.Path testSaveFile = testSaveDir.resolve(recordFileName);
java.nio.file.Files.write(testSaveFile, recordBytes);
logger.info("【FTP测试】已成功将录音保存到本地硬盘: {}", testSaveFile.toAbsolutePath());
} catch (Exception e) {
logger.error("【FTP测试】保存本地测试文件失败: {}", e.getMessage());
}
// 下载完成后切回txt目录，继续处理下一个文件
FtpUtil.changeDir(ftp, city.getFtpSourceDir());
// 上传到OSS（获取可访问的URL）
String ossUrl = fileUploadUtil.uploadWav(city.getCityName(), ossPath, recordFileName, recordBytes);
@ -289,6 +339,22 @@ public class FtpSyncService {
callRecordMapper.insert(rec);
existingIds.add(callRecordId); // 防止同批内重复（本次循环内）
logger.info("【FTP行】保存成功: id={}, callRecordId={}", rec.getId(), callRecordId);
// 将成功信息记录到本地文件中，方便排查和核对
try {
java.nio.file.Path logDir = java.nio.file.Paths.get("logs");
if (!java.nio.file.Files.exists(logDir)) {
java.nio.file.Files.createDirectories(logDir);
}
java.nio.file.Path logFile = logDir.resolve("upload_success.log");
String logLine = String.format("[%s] 成功同步录音: 流水号=%s, 文件名=%s, OSS地址=%s\n",
DateUtil.formatDate(new Date(), "yyyy-MM-dd HH:mm:ss"), callRecordId, recordFileName, ossUrl);
java.nio.file.Files.write(logFile, logLine.getBytes("UTF-8"),
java.nio.file.StandardOpenOption.CREATE, java.nio.file.StandardOpenOption.APPEND);
} catch (Exception e) {
logger.error("记录成功日志到文件失败", e);
}
return true;
}
@ -345,6 +411,44 @@ public class FtpSyncService {
}
}
/**
* 批量加载已处理的文件名（避免重复处理txt文件）
*/
private Set<String> loadProcessedFileNames(String cityCode) {
try {
List<String> fileNames = processedFileMapper.selectProcessedFileNamesByCity(cityCode);
return new HashSet<>(fileNames);
} catch (Exception e) {
logger.error("【FTP防重】加载已处理文件记录失败: {}", e.getMessage());
return new HashSet<>();
}
}
/**
* 记录已处理的txt文件
*/
private void recordProcessedFile(FtpCityConfig city, String fileName, long fileSize,
int recordCount, int successCount, int failCount,
long processTime, String errorMsg) {
try {
FtpProcessedFile record = new FtpProcessedFile();
record.setCityCode(city.getCityCode());
record.setCityName(city.getCityName());
record.setFileName(fileName);
record.setFileSize(fileSize);
record.setProcessStatus(failCount == 0 ? "1" : "0");
record.setRecordCount(recordCount);
record.setSuccessCount(successCount);
record.setFailCount(failCount);
record.setProcessTime(processTime);
record.setErrorMsg(errorMsg);
processedFileMapper.insert(record);
logger.debug("【FTP文件】已记录处理结果: {}", fileName);
} catch (Exception e) {
logger.error("【FTP文件】记录处理结果失败: {}, 原因={}", fileName, e.getMessage());
}
}
/**
* 去除电话号码的区号前缀
*

44
src/main/java/com/threecloud/dataserviceyy/util/FileCleaner.java
View File

@ -0,0 +1,44 @@
package com.threecloud.dataserviceyy.util;
import java.io.File;
/**
* 文件清理工具
* 用于清理超过指定天数的本地临时文件
*/
public class FileCleaner {
/**
* 清理指定目录下的过期文件
*
* @param tempDir 临时目录
* @param retainDays 保留天数（早于今天 retainDays 天的文件将被删除）
*/
public static void clean(String tempDir, int retainDays) {
File dir = new File(tempDir);
if (!dir.exists()) {
return;
}
long threshold = System.currentTimeMillis() - (long) retainDays * 24 * 60 * 60 * 1000;
deleteRecursive(dir, threshold);
}
private static void deleteRecursive(File file, long threshold) {
if (!file.isDirectory()) {
if (file.lastModified() < threshold) {
file.delete();
}
return;
}
File[] files = file.listFiles();
if (files == null) return;
for (File f : files) {
deleteRecursive(f, threshold);
}
// 空目录也删除
File[] remain = file.listFiles();
if (remain == null || remain.length == 0) {
file.delete();
}
}
}

11
src/main/java/com/threecloud/dataserviceyy/util/FileUploadUtil.java
View File

@ -32,19 +32,19 @@ public class FileUploadUtil {
private static final Logger logger = LoggerFactory.getLogger(FileUploadUtil.class);
// OSS配置（从配置文件读取）
@Value("${vaa-sync.oss.base-url:http://53.1.194.59:9090}")
@Value("${ftp-sync.oss.base-url:http://53.1.194.59:9090}")
private String ossBaseUrl;
@Value("${vaa-sync.oss.upload-url:http://53.1.194.59:9090/apiOss/oss/fileUpload}")
@Value("${ftp-sync.oss.upload-url:http://53.1.194.59:9090/apiOss/oss/fileUpload}")
private String ossUploadUrl;
@Value("${vaa-sync.oss.appcode:dataservice-yy}")
@Value("${ftp-sync.oss.appcode:dataservice-yy}")
private String ossAppcode;
@Value("${vaa-sync.oss.appid:371a3368-e28e-4ba3-95a3-c31c19cf0ad0}")
@Value("${ftp-sync.oss.appid:371a3368-e28e-4ba3-95a3-c31c19cf0ad0}")
private String ossAppid;
@Value("${vaa-sync.oss.appsecret:06a6a80e-f9d2-4b3b-acc0-8d182c876074}")
@Value("${ftp-sync.oss.appsecret:06a6a80e-f9d2-4b3b-acc0-8d182c876074}")
private String ossAppsecret;
private RestTemplate restTemplate;
@ -150,6 +150,7 @@ public class FileUploadUtil {
if (response.getStatusCode().is2xxSuccessful() && response.getBody() != null) {
ResultEntity result = response.getBody();
logger.debug("OSS返回结果: {}", result.toString());
if (result.getCode() == ResultEntity.StatusCode.SUCCESS.getCode()) {
String url = parseUrlFromContent(result.getContent());
logger.debug("OSS返回URL: {}", url);

Write Preview
Loading…
Cancel
Save