|
|
@ -132,18 +132,16 @@ public class FtpSyncService { |
|
|
city.getFtpUsername(), city.getFtpPassword(), city.isPassiveMode()); |
|
|
city.getFtpUsername(), city.getFtpPassword(), city.isPassiveMode()); |
|
|
logger.info("【FTP地市】FTP连接成功"); |
|
|
logger.info("【FTP地市】FTP连接成功"); |
|
|
|
|
|
|
|
|
FTPFile[] files = FtpUtil.listFiles(ftp, city.getFtpSourceDir()); |
|
|
List<TxtCandidate> candidates = new ArrayList<>(); |
|
|
int txtCount = FtpUtil.countTxtFiles(files); |
|
|
collectTxtCandidates(ftp, city.getFtpSourceDir(), candidates, true); |
|
|
logger.info("【FTP地市】目录 {} 下有 {} 个txt文件", city.getFtpSourceDir(), txtCount); |
|
|
logger.info("【FTP地市】源目录共发现 {} 个txt文件", candidates.size()); |
|
|
|
|
|
|
|
|
// 批量查询已处理的文件名,避免重复处理
|
|
|
// 批量查询已处理的文件名,避免重复处理
|
|
|
Set<String> processedFiles = loadProcessedFileNames(city.getCityCode()); |
|
|
Set<String> processedFiles = loadProcessedFileNames(city.getCityCode()); |
|
|
logger.info("【FTP地市】已处理文件记录: {} 个", processedFiles.size()); |
|
|
logger.info("【FTP地市】已处理文件记录: {} 个", processedFiles.size()); |
|
|
|
|
|
|
|
|
// 按文件名升序排序,确保按时间先后顺序依次处理
|
|
|
// 按文件名升序排序,确保按时间先后顺序依次处理
|
|
|
if (files != null && files.length > 0) { |
|
|
candidates.sort(Comparator.comparing(candidate -> candidate.file.getName())); |
|
|
Arrays.sort(files, Comparator.comparing(FTPFile::getName)); |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
int fileSuccess = 0; |
|
|
int fileSuccess = 0; |
|
|
int fileFail = 0; |
|
|
int fileFail = 0; |
|
|
@ -151,16 +149,21 @@ public class FtpSyncService { |
|
|
int filePending = 0; |
|
|
int filePending = 0; |
|
|
|
|
|
|
|
|
String todayStr = DateUtil.formatDate(new Date(), "yyyyMMdd"); |
|
|
String todayStr = DateUtil.formatDate(new Date(), "yyyyMMdd"); |
|
|
logger.info("【FTP地市】当前同步日期: {} (仅处理今日数据文件)", todayStr); |
|
|
Calendar calendar = Calendar.getInstance(); |
|
|
|
|
|
calendar.add(Calendar.DAY_OF_MONTH, -1); |
|
|
for (FTPFile file : files) { |
|
|
String yesterdayStr = DateUtil.formatDate(calendar.getTime(), "yyyyMMdd"); |
|
|
|
|
|
logger.info("【FTP地市】当前同步日期: {}, {} (处理今天和昨天数据文件)", |
|
|
|
|
|
yesterdayStr, todayStr); |
|
|
|
|
|
|
|
|
|
|
|
for (TxtCandidate candidate : candidates) { |
|
|
|
|
|
FTPFile file = candidate.file; |
|
|
if (!FtpUtil.isTxtFile(file)) { |
|
|
if (!FtpUtil.isTxtFile(file)) { |
|
|
continue; |
|
|
continue; |
|
|
} |
|
|
} |
|
|
String fileName = file.getName(); |
|
|
String fileName = file.getName(); |
|
|
// 1. 只处理今天的数据文件(例如 20260819*.txt)
|
|
|
// 1. 只处理今天和昨天的数据文件(例如 20260819*.txt)
|
|
|
if (!fileName.startsWith(todayStr)) { |
|
|
if (!fileName.startsWith(todayStr) && !fileName.startsWith(yesterdayStr)) { |
|
|
logger.debug("【FTP文件】非今日数据文件,跳过: {}", fileName); |
|
|
logger.debug("【FTP文件】非今天/昨天数据文件,跳过: {}", fileName); |
|
|
continue; |
|
|
continue; |
|
|
} |
|
|
} |
|
|
// 2. 检查是否为有效的数据TXT文件
|
|
|
// 2. 检查是否为有效的数据TXT文件
|
|
|
@ -172,13 +175,16 @@ public class FtpSyncService { |
|
|
// 3. 检查是否已处理过
|
|
|
// 3. 检查是否已处理过
|
|
|
if (processedFiles.contains(fileName)) { |
|
|
if (processedFiles.contains(fileName)) { |
|
|
logger.debug("【FTP文件】已处理过,跳过: {}", fileName); |
|
|
logger.debug("【FTP文件】已处理过,跳过: {}", fileName); |
|
|
|
|
|
archiveProcessedTxt(ftp, candidate, city); |
|
|
fileSkipped++; |
|
|
fileSkipped++; |
|
|
continue; |
|
|
continue; |
|
|
} |
|
|
} |
|
|
try { |
|
|
try { |
|
|
int txtResult = processTxtFile(ftp, file, city, existingIds); |
|
|
int txtResult = processTxtFile(ftp, candidate.directory, file, city, existingIds); |
|
|
if (txtResult == 1) { |
|
|
if (txtResult == 1) { |
|
|
fileSuccess++; |
|
|
fileSuccess++; |
|
|
|
|
|
processedFiles.add(fileName); |
|
|
|
|
|
archiveProcessedTxt(ftp, candidate, city); |
|
|
} else if (txtResult == 2) { |
|
|
} else if (txtResult == 2) { |
|
|
filePending++; // 录音未就绪,留待下次定时任务
|
|
|
filePending++; // 录音未就绪,留待下次定时任务
|
|
|
} else { |
|
|
} else { |
|
|
@ -193,9 +199,11 @@ public class FtpSyncService { |
|
|
logger.info("【FTP地市】重新连接 FTP 成功,正在重试处理刚才失败的文件: {}", fileName); |
|
|
logger.info("【FTP地市】重新连接 FTP 成功,正在重试处理刚才失败的文件: {}", fileName); |
|
|
|
|
|
|
|
|
// 重新尝试处理文件
|
|
|
// 重新尝试处理文件
|
|
|
int txtResult = processTxtFile(ftp, file, city, existingIds); |
|
|
int txtResult = processTxtFile(ftp, candidate.directory, file, city, existingIds); |
|
|
if (txtResult == 1) { |
|
|
if (txtResult == 1) { |
|
|
fileSuccess++; |
|
|
fileSuccess++; |
|
|
|
|
|
processedFiles.add(fileName); |
|
|
|
|
|
archiveProcessedTxt(ftp, candidate, city); |
|
|
} else if (txtResult == 2) { |
|
|
} else if (txtResult == 2) { |
|
|
filePending++; |
|
|
filePending++; |
|
|
} else { |
|
|
} else { |
|
|
@ -221,10 +229,10 @@ public class FtpSyncService { |
|
|
* |
|
|
* |
|
|
* @return 1=全部成功处理完毕, 2=有录音暂未就绪(下次重试), 0=解析失败 |
|
|
* @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<String> existingIds) throws Exception { |
|
|
Set<String> existingIds) throws Exception { |
|
|
String fileName = file.getName(); |
|
|
String fileName = file.getName(); |
|
|
String sourcePath = joinPath(city.getFtpSourceDir(), fileName); |
|
|
String sourcePath = joinPath(txtDirectory, fileName); |
|
|
long fileSize = file.getSize(); |
|
|
long fileSize = file.getSize(); |
|
|
long fileStartTime = System.currentTimeMillis(); |
|
|
long fileStartTime = System.currentTimeMillis(); |
|
|
logger.info("【FTP文件】开始处理: {} ({} 字节)", fileName, fileSize); |
|
|
logger.info("【FTP文件】开始处理: {} ({} 字节)", fileName, fileSize); |
|
|
@ -511,6 +519,73 @@ public class FtpSyncService { |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
|
* 收集指定FTP目录中的文件。源目录不可访问时抛错,历史目录不可访问时降级跳过。 |
|
|
|
|
|
*/ |
|
|
|
|
|
private void collectTxtCandidates(FTPClient ftp, String directory, |
|
|
|
|
|
List<TxtCandidate> 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数据文件(支持跨天、多天、一天多个时间切片文件) |
|
|
* 判断是否为有效且已封口的TXT数据文件(支持跨天、多天、一天多个时间切片文件) |
|
|
* 兼容格式: |
|
|
* 兼容格式: |
|
|
|