diff --git a/sql/mid_ftp_processed_file.sql b/sql/mid_ftp_processed_file.sql new file mode 100644 index 0000000..40702c3 --- /dev/null +++ b/sql/mid_ftp_processed_file.sql @@ -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 '创建时间'; diff --git a/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java b/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java index b171700..8bfed53 100644 --- a/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java +++ b/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java @@ -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 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 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 loadProcessedFileNames(String cityCode) { + try { + List 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()); + } + } + /** * 去除电话号码的区号前缀 * diff --git a/src/main/java/com/threecloud/dataserviceyy/util/FileCleaner.java b/src/main/java/com/threecloud/dataserviceyy/util/FileCleaner.java new file mode 100644 index 0000000..6b1e9b7 --- /dev/null +++ b/src/main/java/com/threecloud/dataserviceyy/util/FileCleaner.java @@ -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(); + } + } +} diff --git a/src/main/java/com/threecloud/dataserviceyy/util/FileUploadUtil.java b/src/main/java/com/threecloud/dataserviceyy/util/FileUploadUtil.java index eefc216..1663293 100644 --- a/src/main/java/com/threecloud/dataserviceyy/util/FileUploadUtil.java +++ b/src/main/java/com/threecloud/dataserviceyy/util/FileUploadUtil.java @@ -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);