diff --git a/config/application-external.yml b/config/application-external.yml index 6a473ce..cc87555 100644 --- a/config/application-external.yml +++ b/config/application-external.yml @@ -58,7 +58,7 @@ ftp-sync: ftp-port: 9979 ftp-username: yyfile ftp-password: yyfile - ftp-source-dir: /voice_record/ + ftp-source-dir: /sound_log/ ftp-record-dir: /rec/ ftp-archive-dir: /voice_record/processed/ phone-area-code: "0554" diff --git a/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java b/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java index b44caca..cf282e3 100644 --- a/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java +++ b/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java @@ -132,18 +132,16 @@ public class FtpSyncService { city.getFtpUsername(), city.getFtpPassword(), city.isPassiveMode()); logger.info("【FTP地市】FTP连接成功"); - FTPFile[] files = FtpUtil.listFiles(ftp, city.getFtpSourceDir()); - int txtCount = FtpUtil.countTxtFiles(files); - logger.info("【FTP地市】目录 {} 下有 {} 个txt文件", city.getFtpSourceDir(), txtCount); + List candidates = new ArrayList<>(); + collectTxtCandidates(ftp, city.getFtpSourceDir(), candidates, true); + logger.info("【FTP地市】源目录共发现 {} 个txt文件", candidates.size()); // 批量查询已处理的文件名，避免重复处理 Set processedFiles = loadProcessedFileNames(city.getCityCode()); logger.info("【FTP地市】已处理文件记录: {} 个", processedFiles.size()); // 按文件名升序排序，确保按时间先后顺序依次处理 - if (files != null && files.length > 0) { - Arrays.sort(files, Comparator.comparing(FTPFile::getName)); - } + candidates.sort(Comparator.comparing(candidate -> candidate.file.getName())); int fileSuccess = 0; int fileFail = 0; @@ -151,16 +149,21 @@ public class FtpSyncService { int filePending = 0; String todayStr = DateUtil.formatDate(new Date(), "yyyyMMdd"); - logger.info("【FTP地市】当前同步日期: {} (仅处理今日数据文件)", todayStr); - - for (FTPFile file : files) { + Calendar calendar = Calendar.getInstance(); + calendar.add(Calendar.DAY_OF_MONTH, -1); + String yesterdayStr = DateUtil.formatDate(calendar.getTime(), "yyyyMMdd"); + logger.info("【FTP地市】当前同步日期: {}, {} (处理今天和昨天数据文件)", + yesterdayStr, todayStr); + + for (TxtCandidate candidate : candidates) { + FTPFile file = candidate.file; if (!FtpUtil.isTxtFile(file)) { continue; } String fileName = file.getName(); - // 1. 只处理今天的数据文件（例如 20260819*.txt） - if (!fileName.startsWith(todayStr)) { - logger.debug("【FTP文件】非今日数据文件，跳过: {}", fileName); + // 1. 只处理今天和昨天的数据文件（例如 20260819*.txt） + if (!fileName.startsWith(todayStr) && !fileName.startsWith(yesterdayStr)) { + logger.debug("【FTP文件】非今天/昨天数据文件，跳过: {}", fileName); continue; } // 2. 检查是否为有效的数据TXT文件 @@ -172,13 +175,16 @@ public class FtpSyncService { // 3. 检查是否已处理过 if (processedFiles.contains(fileName)) { logger.debug("【FTP文件】已处理过，跳过: {}", fileName); + archiveProcessedTxt(ftp, candidate, city); fileSkipped++; continue; } try { - int txtResult = processTxtFile(ftp, file, city, existingIds); + int txtResult = processTxtFile(ftp, candidate.directory, file, city, existingIds); if (txtResult == 1) { fileSuccess++; + processedFiles.add(fileName); + archiveProcessedTxt(ftp, candidate, city); } else if (txtResult == 2) { filePending++; // 录音未就绪，留待下次定时任务 } else { @@ -193,9 +199,11 @@ public class FtpSyncService { logger.info("【FTP地市】重新连接 FTP 成功，正在重试处理刚才失败的文件: {}", fileName); // 重新尝试处理文件 - int txtResult = processTxtFile(ftp, file, city, existingIds); + int txtResult = processTxtFile(ftp, candidate.directory, file, city, existingIds); if (txtResult == 1) { fileSuccess++; + processedFiles.add(fileName); + archiveProcessedTxt(ftp, candidate, city); } else if (txtResult == 2) { filePending++; } else { @@ -221,10 +229,10 @@ public class FtpSyncService { * * @return 1=全部成功处理完毕, 2=有录音暂未就绪(下次重试), 0=解析失败 */ - private int processTxtFile(FTPClient ftp, FTPFile file, FtpCityConfig city, + private int processTxtFile(FTPClient ftp, String txtDirectory, FTPFile file, FtpCityConfig city, Set existingIds) throws Exception { String fileName = file.getName(); - String sourcePath = joinPath(city.getFtpSourceDir(), fileName); + String sourcePath = joinPath(txtDirectory, fileName); long fileSize = file.getSize(); long fileStartTime = System.currentTimeMillis(); logger.info("【FTP文件】开始处理: {} ({} 字节)", fileName, fileSize); @@ -511,6 +519,73 @@ public class FtpSyncService { } } + /** + * 收集指定FTP目录中的文件。源目录不可访问时抛错，历史目录不可访问时降级跳过。 + */ + private void collectTxtCandidates(FTPClient ftp, String directory, + List candidates, boolean required) throws Exception { + if (!StringUtils.hasText(directory)) { + return; + } + try { + FTPFile[] files = FtpUtil.listFiles(ftp, directory); + int count = 0; + if (files != null) { + for (FTPFile file : files) { + if (FtpUtil.isTxtFile(file)) { + candidates.add(new TxtCandidate(directory, file)); + count++; + } + } + } + logger.info("【FTP地市】目录 {} 下有 {} 个txt文件", directory, count); + } catch (Exception e) { + if (required) { + throw e; + } + logger.warn("【FTP地市】历史目录访问失败，本轮跳过: {}, 原因={}", directory, e.getMessage()); + } + } + + /** + * 将已处理完成的TXT移入按日期分目录的备份目录。 + * 待就绪或失败的TXT不会调用此方法，会继续留在源目录中等待下次同步。 + */ + private void archiveProcessedTxt(FTPClient ftp, TxtCandidate candidate, FtpCityConfig city) { + if (!StringUtils.hasText(city.getFtpArchiveDir())) { + logger.warn("【FTP归档】未配置备份目录，已处理TXT保留在源目录: {}", candidate.file.getName()); + return; + } + String fileName = candidate.file.getName(); + String dateDir = fileName.length() >= 8 && fileName.substring(0, 8).matches("\\d{8}") + ? fileName.substring(0, 8) + : "unknown-date"; + String archiveRoot = city.getFtpArchiveDir(); + String archiveDateDir = joinPath(archiveRoot, dateDir); + String fromPath = joinPath(candidate.directory, fileName); + String toPath = joinPath(archiveDateDir, fileName); + try { + FtpUtil.ensureDir(ftp, archiveRoot); + FtpUtil.ensureDir(ftp, archiveDateDir); + FtpUtil.rename(ftp, fromPath, toPath); + logger.info("【FTP归档】已处理TXT已移入备份目录: {} -> {}", fromPath, toPath); + } catch (Exception e) { + // 归档失败不影响已完成的业务入库；文件留在源目录，下轮会再次尝试归档。 + logger.warn("【FTP归档】移动失败，下轮继续尝试: {} -> {}, 原因={}", + fromPath, toPath, e.getMessage()); + } + } + + private static class TxtCandidate { + private final String directory; + private final FTPFile file; + + private TxtCandidate(String directory, FTPFile file) { + this.directory = directory; + this.file = file; + } + } + /** * 判断是否为有效且已封口的TXT数据文件（支持跨天、多天、一天多个时间切片文件） * 兼容格式：