From 050a2bea11ad45967f7952f3e6de56dd8da12ee0 Mon Sep 17 00:00:00 2001 From: wang Date: Fri, 5 Jun 2026 10:30:11 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E4=BB=A3=E7=A0=81=E9=80=BB?= =?UTF-8?q?=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/application-external.yml | 7 + .../config/FtpSyncProperties.java | 10 + .../dataserviceyy/service/FtpSyncService.java | 258 ++++++++---------- .../dataserviceyy/util/DateUtil.java | 20 +- .../dataserviceyy/util/FtpUtil.java | 20 +- 5 files changed, 176 insertions(+), 139 deletions(-) diff --git a/config/application-external.yml b/config/application-external.yml index 14142c6..713549b 100644 --- a/config/application-external.yml +++ b/config/application-external.yml @@ -67,6 +67,13 @@ ftp-sync: ftp-source-dir: /340100/record/ ftp-record-dir: /340100/recordfile/ ftp-archive-dir: /340100/processed/ + # 电话区号前缀（处理主被叫号码时去除） + # - 配正确区号：号码以这个区号开头则去除 + # - 配不存在的区号（如 "0000"）：号码匹配不上，保留原始号码 + # - 留空：保留原始号码 + # 合肥: 0551, 芜湖: 0553, 蚌埠: 0552, 淮南: 0554, 马鞍山: 0555 + # 当前配置 "0000" 表示不去除区号（保留原始号码） + phone-area-code: "0000" # ==================== 服务端口 ==================== server: diff --git a/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java b/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java index abf3d99..d9fc66b 100644 --- a/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java +++ b/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java @@ -88,6 +88,13 @@ public class FtpSyncProperties { private String ftpRecordDir; /** 归档目录（处理完的txt移到这里） */ private String ftpArchiveDir; + /** + * 电话区号前缀（处理主被叫号码时去除） + * - 配正确区号（如合肥 0551）：号码以 0551 开头则去除 + * - 配不存在的区号（如 "0000"）：号码匹配不上，保留原始号码 + * - 留空：保留原始号码 + */ + private String phoneAreaCode; public String getCityCode() { return cityCode; } public void setCityCode(String cityCode) { this.cityCode = cityCode; } @@ -115,5 +122,8 @@ public class FtpSyncProperties { public String getFtpArchiveDir() { return ftpArchiveDir; } public void setFtpArchiveDir(String ftpArchiveDir) { this.ftpArchiveDir = ftpArchiveDir; } + + public String getPhoneAreaCode() { return phoneAreaCode; } + public void setPhoneAreaCode(String phoneAreaCode) { this.phoneAreaCode = phoneAreaCode; } } } diff --git a/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java b/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java index 763236a..782977a 100644 --- a/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java +++ b/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java @@ -4,6 +4,8 @@ import com.threecloud.dataserviceyy.config.FtpSyncProperties; import com.threecloud.dataserviceyy.config.FtpSyncProperties.FtpCityConfig; import com.threecloud.dataserviceyy.entity.MidVoiceCallRecord; import com.threecloud.dataserviceyy.mapper.MidVoiceCallRecordMapper; +import com.threecloud.dataserviceyy.util.DateUtil; +import com.threecloud.dataserviceyy.util.FileCleaner; import com.threecloud.dataserviceyy.util.FilePathUtil; import com.threecloud.dataserviceyy.util.FileUploadUtil; import com.threecloud.dataserviceyy.util.FtpUtil; @@ -18,9 +20,7 @@ import org.springframework.util.StringUtils; import java.io.BufferedReader; import java.io.ByteArrayInputStream; -import java.io.File; import java.io.InputStreamReader; -import java.text.SimpleDateFormat; import java.util.*; /** @@ -50,6 +50,14 @@ public class FtpSyncService { private static final Logger logger = LoggerFactory.getLogger(FtpSyncService.class); + /** 通话状态：1=正常接通 */ + private static final String CALL_STATUS_OK = "1"; + /** 通话方向：1=呼入，2=呼出（数据库规范） */ + private static final String DIR_INCOMING = "1"; + private static final String DIR_OUTGOING = "2"; + /** 字段数最小阈值 */ + private static final int MIN_FIELD_COUNT = 10; + @Autowired private FtpSyncProperties properties; @Autowired @@ -78,7 +86,7 @@ public class FtpSyncService { */ public void executeSync() { // 清理过期临时文件 - cleanOldFiles(); + FileCleaner.clean(properties.getTempPath(), properties.getRetainDays()); List cities = properties.getCities(); if (cities == null || cities.isEmpty()) { @@ -116,27 +124,27 @@ public class FtpSyncService { FTPClient ftp = null; try { - // 连接FTP ftp = FtpUtil.connect(city.getFtpHost(), city.getFtpPort(), city.getFtpUsername(), city.getFtpPassword()); logger.info("【FTP地市】FTP连接成功"); - // 列出txt源目录所有文件 FTPFile[] files = FtpUtil.listFiles(ftp, city.getFtpSourceDir()); - int txtCount = countTxtFiles(files); + int txtCount = FtpUtil.countTxtFiles(files); logger.info("【FTP地市】目录 {} 下有 {} 个txt文件", city.getFtpSourceDir(), txtCount); int fileSuccess = 0; int fileFail = 0; for (FTPFile file : files) { - if (!file.isFile() || !file.getName().toLowerCase().endsWith(".txt")) { + if (!FtpUtil.isTxtFile(file)) { continue; } try { - boolean ok = processTxtFile(ftp, file, city, existingIds); - if (ok) fileSuccess++; - else fileFail++; + if (processTxtFile(ftp, file, city, existingIds)) { + fileSuccess++; + } else { + fileFail++; + } } catch (Exception e) { fileFail++; logger.error("【FTP地市】处理文件失败: {}, 原因={}", file.getName(), e.getMessage()); @@ -155,12 +163,11 @@ public class FtpSyncService { * @return true=成功，false=失败 */ private boolean processTxtFile(FTPClient ftp, FTPFile file, FtpCityConfig city, - Set existingIds) throws Exception { + Set existingIds) throws Exception { String fileName = file.getName(); String sourcePath = city.getFtpSourceDir() + "/" + fileName; logger.info("【FTP文件】开始处理: {}", fileName); - // 下载txt到内存 byte[] txtBytes = FtpUtil.downloadFile(ftp, sourcePath); int lineCount = 0; int successCount = 0; @@ -176,9 +183,11 @@ public class FtpSyncService { } lineCount++; try { - boolean processed = processLine(line, city, existingIds, ftp); - if (processed) successCount++; - else skipCount++; + if (processLine(line, city, existingIds, ftp)) { + successCount++; + } else { + skipCount++; + } } catch (Exception e) { failCount++; logger.error("【FTP文件】解析单行失败: {}, 原因={}", line, e.getMessage()); @@ -208,110 +217,120 @@ public class FtpSyncService { * @return true=成功入库，false=跳过 */ private boolean processLine(String line, FtpCityConfig city, Set existingIds, - FTPClient ftp) throws Exception { + FTPClient ftp) throws Exception { + // 按分隔符拆分字段（老淮南FTP数据格式：※分隔） String[] fields = line.split(properties.getFieldSeparator(), -1); - if (fields.length < 10) { - logger.debug("【FTP行】字段数不足（{}<10），跳过: {}", fields.length, line); + if (fields.length < MIN_FIELD_COUNT) { + logger.debug("【FTP行】字段数不足（{}<{}），跳过: {}", fields.length, MIN_FIELD_COUNT, line); return false; } - String kssj = trimToNull(fields[0]); // 开始时间 yyyyMMddHHmmss - String jssj = trimToNull(fields[1]); // 结束时间 - String hjls = trimToNull(fields[2]); // 呼叫流水（录音主名） - String hjzls = trimToNull(fields[3]); // 呼叫主/被叫流水 - String zjhm = trimToNull(fields[4]); // 主叫号码 - String bjhm = trimToNull(fields[5]); // 被叫号码 - String thscStr = trimToNull(fields[6]); // 通话时长 - String thfx = trimToNull(fields[9]); // 通话方向 - - if (kssj == null || jssj == null || hjls == null || hjzls == null) { + // ========== 字段解析（老淮南FTP数据格式）========== + String kssj = fields[0].trim(); // kssj : 开始时间 yyyyMMddHHmmss（如：20240101120000） + String jssj = fields[1].trim(); // jssj : 结束时间 yyyyMMddHHmmss + String hjls = fields[2].trim(); // hjls : 呼叫流水号（用于拼录音文件名主名） + String hjzls = fields[3].trim(); // hjzls : 呼叫主/被叫流水号（用于拼录音文件名辅名） + String zjhm = fields[4].trim(); // zjhm : 主叫号码（包含区号，如：055312345678） + String bjhm = fields[5].trim(); // bjhm : 被叫号码 + String thscStr = fields[6].trim();// thsc : 通话时长（秒） + String thfx = fields[9].trim(); // thfx : 通话方向（1=呼入，0=呼出 - 老系统特殊编码） + + // 必填字段校验 + if (kssj.isEmpty() || jssj.isEmpty() || hjls.isEmpty() || hjzls.isEmpty()) { return false; } - int thsc = 0; - try { thsc = thscStr == null ? 0 : Integer.parseInt(thscStr); } catch (NumberFormatException ignored) {} - - // 解析时间 - Date callStartTime = parseDate(kssj); - Date callEndTime = parseDate(jssj); + // 时间字符串转 Date 对象 + Date callStartTime = DateUtil.parseDate14(kssj); + Date callEndTime = DateUtil.parseDate14(jssj); if (callStartTime == null) { return false; } - // 通话方向：1=呼入 0=呼出（老系统thfx字段） - boolean isOutgoing = thfx != null && thfx.contains("0"); - String callDirection = isOutgoing ? "2" : "1"; // 1呼入 2呼出（数据库规范） - - // 录音文件名 + int thsc = parseInt(thscStr, 0); + // thfx 转换：老系统"0"=呼出，其他（含1）=呼入 + // 数据库规范：1=呼入，2=呼出（所以 isOutgoing=true 时存"2"） + boolean isOutgoing = thfx.contains("0"); + // 录音文件名 = hjls_hjzls.mp3（如：12345_67890.mp3） String recordFileName = hjls + "_" + hjzls + ".mp3"; - - // callRecordId 唯一标识 + // 唯一标识 = 地市编码_FTP_呼叫流水（防重键） String callRecordId = city.getCityCode() + "_FTP_" + hjls; - // 防重复检查 + // 防重复检查（内存Set比对，已存在则跳过） if (existingIds.contains(callRecordId)) { logger.debug("【FTP行】已存在，跳过: {}", callRecordId); return false; } - // 主叫/被叫处理（去除 0553 区号前缀 - 参考老代码） - if (zjhm != null && zjhm.startsWith("0553")) { - zjhm = zjhm.substring(4); - } - if (bjhm != null && bjhm.startsWith("0553")) { - bjhm = bjhm.substring(4); - } + // 主叫/被叫处理：去除区号前缀 + // 配置不存在的区号（如"0000"），因为 startsWith 匹配不上，会保留原始号码 + // 后续如需去除区号，把配置改为正确的区号即可，无需重新发包 + zjhm = stripAreaCode(zjhm, city.getPhoneAreaCode()); + bjhm = stripAreaCode(bjhm, city.getPhoneAreaCode()); - // 通话状态：1正常接通 - String callStatus = "1"; - - // 构建本地和OSS路径 - SimpleDateFormat yyyyMMdd = new SimpleDateFormat("yyyyMMdd"); - String dateStr = yyyyMMdd.format(callStartTime); - String localPath = FilePathUtil.buildLocalPath( - properties.getTempPath(), city.getCityCode(), callStartTime, "ftp", recordFileName); + // 构建本地临时路径和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 remoteRecordPath = city.getFtpRecordDir() + "/" + dateStr + "/" + recordFileName; byte[] recordBytes = FtpUtil.downloadFile(ftp, remoteRecordPath); logger.info("【FTP行】下载录音: {} ({} 字节)", recordFileName, recordBytes.length); - // 上传OSS + // 上传到OSS（获取可访问的URL） String ossUrl = fileUploadUtil.uploadWav(city.getCityName(), ossPath, recordFileName, recordBytes); logger.info("【FTP行】上传OSS: {}", ossUrl); - // 构建并保存实体 - MidVoiceCallRecord rec = new MidVoiceCallRecord(); - rec.setCallRecordId(callRecordId); - rec.setDeviceNo("FTP_" + city.getCityCode()); - rec.setCityCode(city.getCityCode()); - rec.setCityName(city.getCityName()); - rec.setOrgCode(city.getCityCode()); - rec.setRecordingFileName(recordFileName); - rec.setCallStartTime(callStartTime); - rec.setCallEndTime(callEndTime); - rec.setCallDuration(thsc); - rec.setCallDirection(callDirection); - - if (isOutgoing) { - rec.setCallTel(zjhm != null ? zjhm : ""); - rec.setCalledTel(bjhm != null ? bjhm : ""); - } else { - rec.setCallTel(bjhm != null ? bjhm : ""); - rec.setCalledTel(zjhm != null ? zjhm : ""); - } - rec.setCallStatus(callStatus); - rec.setRecordingFileSize(recordBytes.length); - rec.setRecordingFilePath(ossUrl); - + // 构建实体对象并入库 + MidVoiceCallRecord rec = buildCallRecord(city, callRecordId, callStartTime, callEndTime, + thsc, isOutgoing, zjhm, bjhm, recordFileName, recordBytes.length, ossUrl); callRecordMapper.insert(rec); - existingIds.add(callRecordId); // 防止同批内重复 + existingIds.add(callRecordId); // 防止同批内重复（本次循环内） logger.info("【FTP行】保存成功: id={}, callRecordId={}", rec.getId(), callRecordId); - return true; } + /** + * 构建通话记录实体 + * + * @param city 地市配置（提供编码、名称） + * @param callRecordId 唯一标识（地市编码_FTP_呼叫流水） + * @param callStartTime 通话开始时间 + * @param callEndTime 通话结束时间 + * @param duration 通话时长（秒） + * @param isOutgoing 是否呼出（true=呼出，false=呼入） + * @param zjhm 主叫号码（已去区号） + * @param bjhm 被叫号码（已去区号） + * @param fileName 录音文件名 + * @param fileSize 录音文件大小（字节） + * @param ossUrl OSS访问URL + */ + private MidVoiceCallRecord buildCallRecord(FtpCityConfig city, String callRecordId, + Date callStartTime, Date callEndTime, + int duration, boolean isOutgoing, + String zjhm, String bjhm, + String fileName, int fileSize, String ossUrl) { + MidVoiceCallRecord rec = new MidVoiceCallRecord(); + rec.setCallRecordId(callRecordId); // 唯一标识 + rec.setDeviceNo("FTP_" + city.getCityCode()); // 设备编码（FTP_地市编码） + rec.setCityCode(city.getCityCode()); // 地市编码（用于数据库分区） + rec.setCityName(city.getCityName()); // 地市名称 + rec.setOrgCode(city.getCityCode()); // 单位代码（用地市编码代替） + rec.setRecordingFileName(fileName); // 录音文件名 + rec.setCallStartTime(callStartTime); // 通话开始时间 + rec.setCallEndTime(callEndTime); // 通话结束时间 + rec.setCallDuration(duration); // 通话时长（秒） + rec.setCallDirection(isOutgoing ? DIR_OUTGOING : DIR_INCOMING); // 通话方向：1呼入/2呼出 + rec.setCallTel(isOutgoing ? zjhm : bjhm); // 主叫号码 + rec.setCalledTel(isOutgoing ? bjhm : zjhm); // 被叫号码 + rec.setCallStatus(CALL_STATUS_OK); // 通话状态：1=正常接通 + rec.setRecordingFileSize(fileSize); // 文件大小（字节） + rec.setRecordingFilePath(ossUrl); // OSS访问URL + return rec; + } + /** * 批量加载已存在的call_record_id（防重） */ @@ -327,66 +346,31 @@ public class FtpSyncService { } /** - * 清理过期临时文件 + * 去除电话号码的区号前缀 + * + * @param phone 原始号码 + * @param areaCode 区号前缀（如合肥 0551），为空或配置不存在的值则不去除 + * @return 处理后的号码 */ - private void cleanOldFiles() { - File tempDir = new File(properties.getTempPath()); - if (!tempDir.exists()) { - return; + private String stripAreaCode(String phone, String areaCode) { + if (phone == null || areaCode == null || areaCode.isEmpty()) { + return phone; } - long retainMillis = (long) properties.getRetainDays() * 24 * 60 * 60 * 1000; - long now = System.currentTimeMillis(); - deleteOldFilesRecursive(tempDir, now - retainMillis); - } - - private void deleteOldFilesRecursive(File dir, long threshold) { - if (!dir.isDirectory()) { - if (dir.lastModified() < threshold) { - dir.delete(); - } - return; - } - File[] files = dir.listFiles(); - if (files == null) return; - for (File f : files) { - deleteOldFilesRecursive(f, threshold); - } - // 删除空目录 - File[] remainFiles = dir.listFiles(); - if (remainFiles == null || remainFiles.length == 0) { - dir.delete(); + if (phone.startsWith(areaCode)) { + return phone.substring(areaCode.length()); } + return phone; } /** - * 统计txt文件数 + * 安全转 int */ - private int countTxtFiles(FTPFile[] files) { - int count = 0; - if (files == null) return 0; - for (FTPFile f : files) { - if (f.isFile() && f.getName().toLowerCase().endsWith(".txt")) { - count++; - } - } - return count; - } - - /** - * 解析日期 yyyyMMddHHmmss - */ - private Date parseDate(String s) { - if (s == null) return null; + private int parseInt(String s, int defaultValue) { + if (s == null || s.isEmpty()) return defaultValue; try { - return new SimpleDateFormat("yyyyMMddHHmmss").parse(s); - } catch (Exception e) { - return null; + return Integer.parseInt(s); + } catch (NumberFormatException e) { + return defaultValue; } } - - private String trimToNull(String s) { - if (s == null) return null; - s = s.trim(); - return s.isEmpty() ? null : s; - } } diff --git a/src/main/java/com/threecloud/dataserviceyy/util/DateUtil.java b/src/main/java/com/threecloud/dataserviceyy/util/DateUtil.java index 3d97b0f..8f9a4be 100644 --- a/src/main/java/com/threecloud/dataserviceyy/util/DateUtil.java +++ b/src/main/java/com/threecloud/dataserviceyy/util/DateUtil.java @@ -1,5 +1,6 @@ package com.threecloud.dataserviceyy.util; +import java.text.ParseException; import java.text.SimpleDateFormat; import java.util.Calendar; import java.util.Date; @@ -9,7 +10,10 @@ import java.util.Date; */ public class DateUtil { - public static final SimpleDateFormat dateFmt4 = new SimpleDateFormat("yyyyMMdd"); + /** yyyyMMdd 格式（日期目录用） */ + public static final SimpleDateFormat DATE_FMT_8 = new SimpleDateFormat("yyyyMMdd"); + /** yyyyMMddHHmmss 格式（FTP时间戳用） */ + public static final SimpleDateFormat DATE_FMT_14 = new SimpleDateFormat("yyyyMMddHHmmss"); /** * 计算两个日期之间的天数差 @@ -54,4 +58,18 @@ public class DateUtil { SimpleDateFormat sdf = new SimpleDateFormat(pattern); return sdf.format(date); } + + /** + * 解析 yyyyMMddHHmmss 格式日期 + * @param s 字符串 + * @return 解析失败返回 null + */ + public static Date parseDate14(String s) { + if (s == null) return null; + try { + return DATE_FMT_14.parse(s); + } catch (ParseException e) { + return null; + } + } } diff --git a/src/main/java/com/threecloud/dataserviceyy/util/FtpUtil.java b/src/main/java/com/threecloud/dataserviceyy/util/FtpUtil.java index f72e651..397b9c0 100644 --- a/src/main/java/com/threecloud/dataserviceyy/util/FtpUtil.java +++ b/src/main/java/com/threecloud/dataserviceyy/util/FtpUtil.java @@ -9,7 +9,6 @@ import org.slf4j.LoggerFactory; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; -import java.io.OutputStream; /** * FTP 工具类 @@ -160,4 +159,23 @@ public class FtpUtil { throw new IOException("切换FTP目录失败: " + dir); } } + + /** + * 判断是否为txt文件 + */ + public static boolean isTxtFile(FTPFile file) { + return file != null && file.isFile() && file.getName().toLowerCase().endsWith(".txt"); + } + + /** + * 统计txt文件数量 + */ + public static int countTxtFiles(FTPFile[] files) { + if (files == null) return 0; + int count = 0; + for (FTPFile f : files) { + if (isTxtFile(f)) count++; + } + return count; + } }