diff --git a/config/application-external.yml b/config/application-external.yml
index 50b2a4f..6a473ce 100644
--- a/config/application-external.yml
+++ b/config/application-external.yml
@@ -22,6 +22,16 @@ ftp-sync:
# 是否启用FTP同步
enabled: true
+ # ==================== 定时任务执行周期 (Cron表达式) ====================
+ # 常用配置示例：
+ # "0 0/5 * * * ?" 每 5 分钟执行一次（测试推荐）
+ # "0 0/10 * * * ?" 每 10 分钟执行一次
+ # "0 0/30 * * * ?" 每 30 分钟执行一次
+ # "0 0 0/1 * * ?" 每 1 小时执行一次
+ # "0 0 0/2 * * ?" 每 2 小时执行一次（默认）
+ # "0 0 2 * * ?" 每天凌晨 2:00 执行一次
+ sync-interval-cron: "0 0 0/2 * * ?"
+
# 本地临时文件目录
temp-path: ./vaa-ftp-temp
@@ -43,7 +53,7 @@ ftp-sync:
cities:
# ===== 淮南 =====
- city-code: "340400"
- city-name: "淮南"
+ city-name: "淮南市"
ftp-host: 10.126.129.7
ftp-port: 9979
ftp-username: yyfile
@@ -56,7 +66,7 @@ ftp-sync:
# ===== 蚌埠 =====
- city-code: "340300"
- city-name: "蚌埠"
+ city-name: "蚌埠市"
ftp-host: 10.126.54.104
ftp-port: 9989
ftp-username: ftpadmin
diff --git a/pom.xml b/pom.xml
index 84b1019..33e7df2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -20,7 +20,7 @@
8.0.28
2.2.2
1.4.6
- 8.6.0
+ 8.2.0
21.9.0.0
1.18.24
@@ -71,6 +71,8 @@
cn.com.kingbase
kingbase8
${kingbase.version}
+ system
+ ${project.basedir}/src/main/resources/kingbase8-8.2.0.jar
com.oracle.database.jdbc
@@ -177,6 +179,7 @@
${spring-boot.version}
com.threecloud.dataserviceyy.DataserviceYyApplication
+ true
false
diff --git a/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java b/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java
index bd6f7bf..f43ec43 100644
--- a/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java
+++ b/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java
@@ -42,6 +42,9 @@ public class FtpSyncProperties {
/** OSS服务基础地址（用于mp3直接URL拼接） */
private String ossBaseUrl;
+ /** 定时同步Cron表达式（默认每2小时执行一次） */
+ private String syncIntervalCron = "0 0 0/2 * * ?";
+
/** 地市FTP配置列表 */
private List cities = new ArrayList<>();
@@ -63,6 +66,9 @@ public class FtpSyncProperties {
public String getOssBaseUrl() { return ossBaseUrl; }
public void setOssBaseUrl(String ossBaseUrl) { this.ossBaseUrl = ossBaseUrl; }
+ public String getSyncIntervalCron() { return syncIntervalCron; }
+ public void setSyncIntervalCron(String syncIntervalCron) { this.syncIntervalCron = syncIntervalCron; }
+
public List getCities() { return cities; }
public void setCities(List cities) { this.cities = cities; }
diff --git a/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java b/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java
index d83b436..b44caca 100644
--- a/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java
+++ b/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java
@@ -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,10 +219,10 @@ public class FtpSyncService {
/**
* 处理单个txt文件
*
- * @return true=成功，false=失败
+ * @return 1=全部成功处理完毕, 2=有录音暂未就绪(下次重试), 0=解析失败
*/
- private boolean processTxtFile(FTPClient ftp, FTPFile file, FtpCityConfig city,
- Set existingIds) throws Exception {
+ private int processTxtFile(FTPClient ftp, FTPFile file, FtpCityConfig city,
+ Set existingIds) throws Exception {
String fileName = file.getName();
String sourcePath = joinPath(city.getFtpSourceDir(), fileName);
long fileSize = file.getSize();
@@ -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)) {
- successCount++;
- } else {
- skipCount++;
+ int result = processLine(line, city, existingIds, ftp);
+ switch (result) {
+ case 1:
+ successCount++;
+ 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 existingIds,
- FTPClient ftp) throws Exception {
+ private int processLine(String line, FtpCityConfig city, Set 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;
}
/**
diff --git a/src/main/resources/kingbase8-8.2.0.jar b/src/main/resources/kingbase8-8.2.0.jar
new file mode 100644
index 0000000..aa46cbc
Binary files /dev/null and b/src/main/resources/kingbase8-8.2.0.jar differ
diff --git a/src/main/resources/mapper/FtpProcessedFileMapper.xml b/src/main/resources/mapper/FtpProcessedFileMapper.xml
index 0823425..d389f21 100644
--- a/src/main/resources/mapper/FtpProcessedFileMapper.xml
+++ b/src/main/resources/mapper/FtpProcessedFileMapper.xml
@@ -18,7 +18,7 @@
-
+
INSERT INTO mid_ftp_processed_file (
city_code, city_name, file_name, file_size, process_status,
record_count, success_count, fail_count, process_time, error_msg, create_time
diff --git a/src/main/resources/mapper/MidVoiceCallRecordMapper.xml b/src/main/resources/mapper/MidVoiceCallRecordMapper.xml
index 5feb9fc..f143050 100644
--- a/src/main/resources/mapper/MidVoiceCallRecordMapper.xml
+++ b/src/main/resources/mapper/MidVoiceCallRecordMapper.xml
@@ -27,7 +27,7 @@
-
+
INSERT INTO mid_voice_call_record (
city_code, city_name, call_record_id, call_tel, called_tel,
call_start_time, call_end_time, call_duration, call_direction,