package com.threecloud.dataserviceyy.service; 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; import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPFile; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import org.springframework.util.StringUtils; import java.io.BufferedReader; import java.io.ByteArrayInputStream; import java.io.InputStreamReader; import java.util.*; /** * FTP 录音同步服务(参考淮南 yydc-tb-server 的 MineService 重写) * * 【功能】从FTP读取录音数据txt → 解析 → 下载录音文件 → 上传OSS → 入库 * 【策略】增量同步、无重试、数据库批量防重、txt处理完归档 * 【部署】每个地市单独部署一份,每个服务只处理一个地市的FTP * * txt文件格式(GBK编码,※分隔): * kssj※jssj※hjls※hjzls※zjhm※bjhm※thsc※...※thfx * * kssj : 开始时间 yyyyMMddHHmmss * jssj : 结束时间 yyyyMMddHHmmss * hjls : 呼叫流水(录音文件主名) * hjzls : 呼叫主/被叫流水 * zjhm : 主叫号码 * bjhm : 被叫号码 * thsc : 通话时长(秒) * thfx : 通话方向 1=呼入 0=呼出 * * 录音文件命名:{hjls}_{hjzls}.mp3 * 录音文件路径:{ftpRecordDir}/{yyyyMMdd}/{hjls}_{hjzls}.mp3 */ @Service 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 private MidVoiceCallRecordMapper callRecordMapper; @Autowired private FileUploadUtil fileUploadUtil; /** * 定时同步任务 */ @Scheduled(cron = "${ftp-sync.sync-interval-cron:0 0 0/2 * * ?}") public void scheduledSync() { logger.info("【FTP定时】========== FTP录音同步开始 =========="); long startTime = System.currentTimeMillis(); try { executeSync(); } catch (Exception e) { logger.error("【FTP定时】同步任务异常: {}", e.getMessage(), e); } long costTime = System.currentTimeMillis() - startTime; logger.info("【FTP定时】========== FTP录音同步结束,耗时 {} 秒 ==========", costTime / 1000); } /** * 执行同步(主入口) */ public void executeSync() { // 清理过期临时文件 FileCleaner.clean(properties.getTempPath(), properties.getRetainDays()); List cities = properties.getCities(); if (cities == null || cities.isEmpty()) { logger.warn("【FTP主流程】未配置任何地市FTP,跳过同步"); return; } logger.info("【FTP主流程】配置了 {} 个地市", cities.size()); for (FtpCityConfig city : cities) { try { syncCity(city); } catch (Exception e) { logger.error("【FTP主流程】地市 {} 同步失败: {}", city.getCityName(), e.getMessage(), e); } } } /** * 同步单个地市 */ private void syncCity(FtpCityConfig city) throws Exception { logger.info("【FTP地市】═══════════════════════════════════════"); logger.info("【FTP地市】开始同步: {}({}), FTP={}:{}", city.getCityName(), city.getCityCode(), city.getFtpHost(), city.getFtpPort()); // 参数校验 if (!StringUtils.hasText(city.getFtpHost()) || !StringUtils.hasText(city.getFtpUsername())) { logger.warn("【FTP地市】{} FTP未配置完整,跳过", city.getCityName()); return; } // 批量查询数据库已存在的call_record_id(防重) String prefix = city.getCityCode() + "_FTP_"; Set existingIds = loadExistingCallRecordIds(prefix); FTPClient ftp = null; try { ftp = FtpUtil.connect(city.getFtpHost(), city.getFtpPort(), city.getFtpUsername(), city.getFtpPassword()); logger.info("【FTP地市】FTP连接成功"); FTPFile[] files = FtpUtil.listFiles(ftp, city.getFtpSourceDir()); int txtCount = FtpUtil.countTxtFiles(files); logger.info("【FTP地市】目录 {} 下有 {} 个txt文件", city.getFtpSourceDir(), txtCount); int fileSuccess = 0; int fileFail = 0; for (FTPFile file : files) { if (!FtpUtil.isTxtFile(file)) { continue; } try { if (processTxtFile(ftp, file, city, existingIds)) { fileSuccess++; } else { fileFail++; } } catch (Exception e) { fileFail++; logger.error("【FTP地市】处理文件失败: {}, 原因={}", file.getName(), e.getMessage()); } } logger.info("【FTP地市】{} 同步完成: 成功{}个文件, 失败{}个文件", city.getCityName(), fileSuccess, fileFail); } finally { FtpUtil.disconnect(ftp); } } /** * 处理单个txt文件 * * @return true=成功,false=失败 */ private boolean processTxtFile(FTPClient ftp, FTPFile file, FtpCityConfig city, Set existingIds) throws Exception { String fileName = file.getName(); String sourcePath = city.getFtpSourceDir() + "/" + fileName; logger.info("【FTP文件】开始处理: {}", fileName); byte[] txtBytes = FtpUtil.downloadFile(ftp, sourcePath); int lineCount = 0; int successCount = 0; int skipCount = 0; int failCount = 0; try (BufferedReader reader = new BufferedReader( new InputStreamReader(new ByteArrayInputStream(txtBytes), properties.getTxtEncoding()))) { String line; while ((line = reader.readLine()) != null) { if (line.indexOf(properties.getFieldSeparator()) <= 0) { continue; } lineCount++; try { if (processLine(line, city, existingIds, ftp)) { successCount++; } else { skipCount++; } } catch (Exception e) { failCount++; logger.error("【FTP文件】解析单行失败: {}, 原因={}", line, e.getMessage()); } } } logger.info("【FTP文件】{} 解析完成: 共{}行, 新增{}条, 跳过{}条, 失败{}条", fileName, lineCount, successCount, skipCount, failCount); // 移动到归档目录 try { String archivePath = 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 true; } /** * 处理单行记录 * * @return true=成功入库,false=跳过 */ private boolean processLine(String line, FtpCityConfig city, Set existingIds, FTPClient ftp) throws Exception { // 按分隔符拆分字段(老淮南FTP数据格式:※分隔) String[] fields = line.split(properties.getFieldSeparator(), -1); if (fields.length < MIN_FIELD_COUNT) { logger.debug("【FTP行】字段数不足({}<{}),跳过: {}", fields.length, MIN_FIELD_COUNT, line); return false; } // ========== 字段解析(老淮南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; } // 时间字符串转 Date 对象 Date callStartTime = DateUtil.parseDate14(kssj); Date callEndTime = DateUtil.parseDate14(jssj); if (callStartTime == null) { return false; } 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"; // 唯一标识 = 地市编码_FTP_呼叫流水(防重键) String callRecordId = city.getCityCode() + "_FTP_" + hjls; // 防重复检查(内存Set比对,已存在则跳过) if (existingIds.contains(callRecordId)) { logger.debug("【FTP行】已存在,跳过: {}", callRecordId); return false; } // 主叫/被叫处理:去除区号前缀 // 配置不存在的区号(如"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 remoteRecordPath = city.getFtpRecordDir() + "/" + dateStr + "/" + recordFileName; byte[] recordBytes = FtpUtil.downloadFile(ftp, remoteRecordPath); logger.info("【FTP行】下载录音: {} ({} 字节)", recordFileName, recordBytes.length); // 上传到OSS(获取可访问的URL) String ossUrl = fileUploadUtil.uploadWav(city.getCityName(), ossPath, recordFileName, recordBytes); logger.info("【FTP行】上传OSS: {}", ossUrl); // 构建实体对象并入库 MidVoiceCallRecord rec = buildCallRecord(city, callRecordId, callStartTime, callEndTime, thsc, isOutgoing, zjhm, bjhm, recordFileName, recordBytes.length, ossUrl); callRecordMapper.insert(rec); 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(防重) */ private Set loadExistingCallRecordIds(String prefix) { try { List ids = callRecordMapper.selectCallRecordIdsByPrefix(prefix); logger.info("【FTP防重】已加载 {} 条历史记录(prefix={})", ids.size(), prefix); return new HashSet<>(ids); } catch (Exception e) { logger.error("【FTP防重】加载历史记录失败,将全部当作新数据: {}", e.getMessage()); return new HashSet<>(); } } /** * 去除电话号码的区号前缀 * * @param phone 原始号码 * @param areaCode 区号前缀(如合肥 0551),为空或配置不存在的值则不去除 * @return 处理后的号码 */ private String stripAreaCode(String phone, String areaCode) { if (phone == null || areaCode == null || areaCode.isEmpty()) { return phone; } if (phone.startsWith(areaCode)) { return phone.substring(areaCode.length()); } return phone; } /** * 安全转 int */ private int parseInt(String s, int defaultValue) { if (s == null || s.isEmpty()) return defaultValue; try { return Integer.parseInt(s); } catch (NumberFormatException e) { return defaultValue; } } }