|
|
|
@ -148,27 +148,39 @@ public class FtpSyncService { |
|
|
|
int fileSuccess = 0; |
|
|
|
int fileFail = 0; |
|
|
|
int fileSkipped = 0; |
|
|
|
int filePending = 0; |
|
|
|
|
|
|
|
String todayStr = DateUtil.formatDate(new Date(), "yyyyMMdd"); |
|
|
|
logger.info("【FTP地市】当前同步日期: {} (仅处理今日数据文件)", todayStr); |
|
|
|
|
|
|
|
for (FTPFile file : files) { |
|
|
|
if (!FtpUtil.isTxtFile(file)) { |
|
|
|
continue; |
|
|
|
} |
|
|
|
String fileName = file.getName(); |
|
|
|
// 检查是否为有效的数据TXT文件（支持跨天、一天内多个切片文件；排除空文件或未封口占位文件）
|
|
|
|
// 1. 只处理今天的数据文件（例如 20260819*.txt）
|
|
|
|
if (!fileName.startsWith(todayStr)) { |
|
|
|
logger.debug("【FTP文件】非今日数据文件，跳过: {}", fileName); |
|
|
|
continue; |
|
|
|
} |
|
|
|
// 2. 检查是否为有效的数据TXT文件
|
|
|
|
if (!isValidDataTxtFile(file)) { |
|
|
|
logger.debug("【FTP文件】非有效数据文件，跳过: {} ({} 字节)", |
|
|
|
fileName, file.getSize()); |
|
|
|
continue; |
|
|
|
} |
|
|
|
// 检查是否已处理过
|
|
|
|
// 3. 检查是否已处理过
|
|
|
|
if (processedFiles.contains(fileName)) { |
|
|
|
logger.debug("【FTP文件】已处理过，跳过: {}", fileName); |
|
|
|
fileSkipped++; |
|
|
|
continue; |
|
|
|
} |
|
|
|
try { |
|
|
|
if (processTxtFile(ftp, file, city, existingIds)) { |
|
|
|
int txtResult = processTxtFile(ftp, file, city, existingIds); |
|
|
|
if (txtResult == 1) { |
|
|
|
fileSuccess++; |
|
|
|
} else if (txtResult == 2) { |
|
|
|
filePending++; // 录音未就绪，留待下次定时任务
|
|
|
|
} else { |
|
|
|
fileFail++; |
|
|
|
} |
|
|
|
@ -181,8 +193,11 @@ public class FtpSyncService { |
|
|
|
logger.info("【FTP地市】重新连接 FTP 成功，正在重试处理刚才失败的文件: {}", fileName); |
|
|
|
|
|
|
|
// 重新尝试处理文件
|
|
|
|
if (processTxtFile(ftp, file, city, existingIds)) { |
|
|
|
int txtResult = processTxtFile(ftp, file, city, existingIds); |
|
|
|
if (txtResult == 1) { |
|
|
|
fileSuccess++; |
|
|
|
} else if (txtResult == 2) { |
|
|
|
filePending++; |
|
|
|
} else { |
|
|
|
fileFail++; |
|
|
|
} |
|
|
|
@ -194,8 +209,8 @@ public class FtpSyncService { |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
logger.info("【FTP地市】{} 同步完成: 成功{}个文件, 跳过{}个(已处理), 失败{}个文件", |
|
|
|
city.getCityName(), fileSuccess, fileSkipped, fileFail); |
|
|
|
logger.info("【FTP地市】{} 同步完成: 成功{}个文件, 待就绪(等待下次同步){}个, 跳过{}个(已处理), 失败{}个文件", |
|
|
|
city.getCityName(), fileSuccess, filePending, fileSkipped, fileFail); |
|
|
|
} finally { |
|
|
|
FtpUtil.disconnect(ftp); |
|
|
|
} |
|
|
|
@ -204,9 +219,9 @@ public class FtpSyncService { |
|
|
|
/** |
|
|
|
* 处理单个txt文件 |
|
|
|
* |
|
|
|
* @return true=成功，false=失败 |
|
|
|
* @return 1=全部成功处理完毕, 2=有录音暂未就绪(下次重试), 0=解析失败 |
|
|
|
*/ |
|
|
|
private boolean processTxtFile(FTPClient ftp, FTPFile file, FtpCityConfig city, |
|
|
|
private int processTxtFile(FTPClient ftp, FTPFile file, FtpCityConfig city, |
|
|
|
Set<String> existingIds) throws Exception { |
|
|
|
String fileName = file.getName(); |
|
|
|
String sourcePath = joinPath(city.getFtpSourceDir(), fileName); |
|
|
|
@ -217,7 +232,9 @@ public class FtpSyncService { |
|
|
|
byte[] txtBytes = FtpUtil.downloadFile(ftp, sourcePath); |
|
|
|
int lineCount = 0; |
|
|
|
int successCount = 0; |
|
|
|
int skipCount = 0; |
|
|
|
int existCount = 0; |
|
|
|
int pendingCount = 0; |
|
|
|
int invalidCount = 0; |
|
|
|
int failCount = 0; |
|
|
|
String errorMsg = null; |
|
|
|
|
|
|
|
@ -230,10 +247,20 @@ public class FtpSyncService { |
|
|
|
} |
|
|
|
lineCount++; |
|
|
|
try { |
|
|
|
if (processLine(line, city, existingIds, ftp)) { |
|
|
|
int result = processLine(line, city, existingIds, ftp); |
|
|
|
switch (result) { |
|
|
|
case 1: |
|
|
|
successCount++; |
|
|
|
} else { |
|
|
|
skipCount++; |
|
|
|
break; |
|
|
|
case 2: |
|
|
|
existCount++; |
|
|
|
break; |
|
|
|
case 3: |
|
|
|
pendingCount++; |
|
|
|
break; |
|
|
|
default: |
|
|
|
invalidCount++; |
|
|
|
break; |
|
|
|
} |
|
|
|
} catch (Exception e) { |
|
|
|
// 如果是网络连接重置或其它 IO 严重异常，不在此吞掉，直接向上抛出以触发外部重连重试
|
|
|
|
@ -252,39 +279,40 @@ public class FtpSyncService { |
|
|
|
} |
|
|
|
|
|
|
|
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; |
|
|
|
logger.info("【FTP文件】{} 处理结果: 共{}行, 新增入库{}条, 已存在{}条, 录音待就绪{}条, 失败{}条, 耗时{}ms", |
|
|
|
fileName, lineCount, successCount, existCount, pendingCount, failCount, processTime); |
|
|
|
|
|
|
|
// 关键逻辑：只有当该文件里所有录音均已处理完成（无待就绪、无失败）时，才标记该TXT文件为已处理
|
|
|
|
if (pendingCount == 0 && failCount == 0) { |
|
|
|
recordProcessedFile(city, fileName, fileSize, lineCount, successCount, failCount, processTime, null); |
|
|
|
logger.info("【FTP文件】{} 全部记录已处理完成，标记为已完成", fileName); |
|
|
|
return 1; |
|
|
|
} else if (pendingCount > 0 && failCount == 0) { |
|
|
|
logger.info("【FTP文件】{} 仍有 {} 条录音在FTP未就绪，暂不标记完成，下次定时任务将继续自动重试", fileName, pendingCount); |
|
|
|
return 2; |
|
|
|
} else { |
|
|
|
logger.warn("【FTP文件】{} 存在 {} 行解析失败", fileName, failCount); |
|
|
|
return 0; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* 处理单行记录 |
|
|
|
* 处理单行数据 |
|
|
|
* |
|
|
|
* @return true=成功入库，false=跳过 |
|
|
|
* @param line 单行文本内容 |
|
|
|
* @param city 地市配置 |
|
|
|
* @param existingIds 已存在的通话记录ID集合（用于防重） |
|
|
|
* @param ftp FTP客户端连接 |
|
|
|
* @return 1=成功入库, 2=已存在跳过, 3=录音未就绪跳过, 0=无效行 |
|
|
|
*/ |
|
|
|
private boolean processLine(String line, FtpCityConfig city, Set<String> existingIds, |
|
|
|
private int processLine(String line, FtpCityConfig city, Set<String> existingIds, |
|
|
|
FTPClient ftp) throws Exception { |
|
|
|
// 按分隔符拆分字段（*分隔，需转义正则特殊字符）
|
|
|
|
String separator = java.util.regex.Pattern.quote(properties.getFieldSeparator()); |
|
|
|
String[] fields = line.split(separator, -1); |
|
|
|
if (fields.length < MIN_FIELD_COUNT) { |
|
|
|
logger.debug("【FTP行】字段数不足（{}<{}），跳过: {}", fields.length, MIN_FIELD_COUNT, line); |
|
|
|
return false; |
|
|
|
return 0; |
|
|
|
} |
|
|
|
|
|
|
|
// ========== 字段解析（老淮南FTP数据格式）==========
|
|
|
|
@ -299,14 +327,14 @@ public class FtpSyncService { |
|
|
|
|
|
|
|
// 必填字段校验
|
|
|
|
if (kssj.isEmpty() || jssj.isEmpty() || hjls.isEmpty() || hjzls.isEmpty()) { |
|
|
|
return false; |
|
|
|
return 0; |
|
|
|
} |
|
|
|
|
|
|
|
// 时间字符串转 Date 对象
|
|
|
|
Date callStartTime = DateUtil.parseDate14(kssj); |
|
|
|
Date callEndTime = DateUtil.parseDate14(jssj); |
|
|
|
if (callStartTime == null) { |
|
|
|
return false; |
|
|
|
return 0; |
|
|
|
} |
|
|
|
|
|
|
|
int thsc = parseInt(thscStr, 0); |
|
|
|
@ -320,24 +348,19 @@ public class FtpSyncService { |
|
|
|
|
|
|
|
// 防重复检查（内存Set比对，已存在则跳过）
|
|
|
|
if (existingIds.contains(callRecordId)) { |
|
|
|
logger.debug("【FTP行】已存在，跳过: {}", callRecordId); |
|
|
|
return false; |
|
|
|
logger.debug("【FTP行】已存在数据库中，跳过: {}", callRecordId); |
|
|
|
return 2; |
|
|
|
} |
|
|
|
|
|
|
|
// 主叫/被叫处理：去除区号前缀
|
|
|
|
// 配置不存在的区号（如"0000"），因为 startsWith 匹配不上，会保留原始号码
|
|
|
|
// 后续如需去除区号，把配置改为正确的区号即可，无需重新发包
|
|
|
|
zjhm = stripAreaCode(zjhm, city.getPhoneAreaCode()); |
|
|
|
bjhm = stripAreaCode(bjhm, city.getPhoneAreaCode()); |
|
|
|
|
|
|
|
// 构建本地临时路径和OSS路径
|
|
|
|
// 本地路径示例：./vaa-ftp-temp/340100/20240101/ftp/12345_67890.mp3
|
|
|
|
// OSS路径示例：voice/340100/20240101/12345_67890.mp3
|
|
|
|
String dateStr = DateUtil.formatDate(callStartTime, "yyyyMMdd"); |
|
|
|
String ossPath = FilePathUtil.buildOssPath(city.getCityCode(), callStartTime, recordFileName); |
|
|
|
|
|
|
|
// 下载录音文件（FTP路径：{ftp-record-dir}/{yyyyMMdd}/{filename}）
|
|
|
|
// 使用绝对路径下载，避免切换目录导致的问题
|
|
|
|
String recordDir = city.getFtpRecordDir(); |
|
|
|
if (recordDir != null && recordDir.endsWith("/")) { |
|
|
|
recordDir = recordDir.substring(0, recordDir.length() - 1); |
|
|
|
@ -345,26 +368,27 @@ public class FtpSyncService { |
|
|
|
String remoteRecordPath = recordDir + "/" + dateStr + "/" + recordFileName; |
|
|
|
logger.info("【FTP行】准备下载录音: {}", remoteRecordPath); |
|
|
|
byte[] recordBytes = FtpUtil.downloadFile(ftp, remoteRecordPath); |
|
|
|
String ossUrl = null; |
|
|
|
int fileSize = 0; |
|
|
|
if (recordBytes != null && recordBytes.length > 0) { |
|
|
|
fileSize = recordBytes.length; |
|
|
|
logger.info("【FTP行】下载录音: {} ({} 字节)", recordFileName, fileSize); |
|
|
|
// 上传到OSS（获取可访问的URL）
|
|
|
|
ossUrl = fileUploadUtil.uploadWav(ossPath, recordFileName, recordBytes); |
|
|
|
logger.info("【FTP行】上传OSS: {}", ossUrl); |
|
|
|
} else { |
|
|
|
logger.warn("【FTP行】录音文件不存在或为空，跳过OSS上传: {}", remoteRecordPath); |
|
|
|
|
|
|
|
// 如果录音文件在 FTP 上暂未生成或为空：不插入数据库（绝不放空值），等待下次定时任务
|
|
|
|
if (recordBytes == null || recordBytes.length == 0) { |
|
|
|
logger.info("【FTP行】录音文件在FTP暂未就绪，跳过本次入库，等待下次定时同步: {}", remoteRecordPath); |
|
|
|
return 3; |
|
|
|
} |
|
|
|
|
|
|
|
// 录音文件存在：上传到OSS
|
|
|
|
int fileSize = recordBytes.length; |
|
|
|
logger.info("【FTP行】下载录音成功: {} ({} 字节)", recordFileName, fileSize); |
|
|
|
String ossUrl = fileUploadUtil.uploadWav(ossPath, recordFileName, recordBytes); |
|
|
|
logger.info("【FTP行】上传OSS成功: {}", ossUrl); |
|
|
|
|
|
|
|
// 构建实体对象并入库
|
|
|
|
MidVoiceCallRecord rec = buildCallRecord(city, callRecordId, callStartTime, callEndTime, |
|
|
|
thsc, isOutgoing, zjhm, bjhm, recordFileName, fileSize, ossUrl); |
|
|
|
callRecordMapper.insert(rec); |
|
|
|
existingIds.add(callRecordId); // 防止同批内重复（本次循环内）
|
|
|
|
logger.info("【FTP行】保存成功: id={}, callRecordId={}", rec.getId(), callRecordId); |
|
|
|
existingIds.add(callRecordId); // 防止同批内重复
|
|
|
|
logger.info("【FTP行】保存成功: callRecordId={}", callRecordId); |
|
|
|
|
|
|
|
return true; |
|
|
|
return 1; |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
|
wang
2 weeks ago