package com.threecloud.dataserviceyy.service; import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; import com.threecloud.dataserviceyy.entity.MidVoiceCallRecord; import com.threecloud.dataserviceyy.entity.MidVoiceDeviceLog; import com.threecloud.dataserviceyy.entity.MidVoiceResyncLog; import com.threecloud.dataserviceyy.entity.OssFileResponse; import com.threecloud.dataserviceyy.mapper.MidVoiceCallRecordMapper; import com.threecloud.dataserviceyy.mapper.MidVoiceDeviceLogMapper; import com.threecloud.dataserviceyy.mapper.MidVoiceResyncLogMapper; import com.threecloud.dataserviceyy.mapper.VoiceSyncMapper; import com.threecloud.dataserviceyy.util.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import org.springframework.util.StringUtils; import java.io.File; import java.net.URLEncoder; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; import java.text.SimpleDateFormat; import java.util.*; /** * VAA录音盒定时同步服务 * * 【功能】从EBOX录音盒拉取录音 → 下载文件 → 上传OSS → 保存通话记录 * 【策略】增量同步、无重试、防重复、失败记录到日志表 * 【账户】每个设备的账号密码从 mid_voice_device_config 表读取 * 【OSS存储路径】voice/{地市编码}/{yyyy-MM-dd}/{文件名} */ @Service public class VaaSyncService { private static final Logger logger = LoggerFactory.getLogger(VaaSyncService.class); @Autowired private VoiceSyncMapper voiceSyncMapper; @Autowired private MidVoiceCallRecordMapper callRecordMapper; @Autowired private MidVoiceDeviceLogMapper deviceLogMapper; @Autowired private MidVoiceResyncLogMapper resyncLogMapper; @Autowired private VaaHttpUtil vaaHttpUtil; @Autowired private OssFileService ossFileService; @Autowired private G729Converter g729Converter; @Value("${vaa-sync.download-path:./vaa-recordings}") private String downloadPath; @Value("${vaa-sync.retain-days:10}") private int retainDays; /** * 定时同步任务 */ @Scheduled(cron = "${vaa-sync.sync-interval-cron:0 0 0/8 * * ?}") public void scheduledSync() { logger.info("【定时任务】========== VAA录音盒同步开始 =========="); long startTime = System.currentTimeMillis(); executeSync(); long costTime = System.currentTimeMillis() - startTime; logger.info("【定时任务】========== VAA录音盒同步结束,耗时 {} 秒 ==========", costTime / 1000); } /** * 执行同步任务(主入口) */ public void executeSync() { try { // 清理过期本地文件 FileCleaner.cleanOldFiles(downloadPath, retainDays); // 查询在线设备列表(含每个设备的账号密码) List> deviceList = voiceSyncMapper.getAllYysb(); logger.info("【主流程】查询到 {} 个在线语音设备", deviceList.size()); int successCount = 0; int failCount = 0; for (int i = 0; i < deviceList.size(); i++) { Map device = deviceList.get(i); String deviceId = getStr(device, "ID"); try { syncSingleDevice(device); successCount++; } catch (Exception e) { failCount++; logger.error("【异常】设备同步失败: ID={}, 原因={}", deviceId, e.getMessage()); } } logger.info("【主流程】同步完成,成功 {} 个,失败 {} 个", successCount, failCount); } catch (Exception e) { logger.error("【异常】同步任务执行失败: {}", e.getMessage(), e); } } /** * 根据通话记录编号重新同步录音文件 * * 【说明】只重新下载录音文件并上传OSS,不修改 mid_voice_call_record 业务数据。 * 流程:查通话记录 → 查设备 → 登录 → 按时间范围查录音列表 → 匹配记录 → 下载 → 上传OSS → 记录操作日志 * * @param callRecordId 通话记录编号 * @return 成功提示信息 * @throws RuntimeException 重新同步失败(已记录操作日志) */ public String resyncByCallRecordId(String callRecordId) { String deviceNo = null; String fileName = null; String ossFilePath = null; try { if (!StringUtils.hasText(callRecordId)) { throw new IllegalArgumentException("通话记录编号为空"); } // 1. 查询通话记录 MidVoiceCallRecord record = callRecordMapper.selectByCallRecordId(callRecordId); if (record == null) { throw new IllegalArgumentException("通话记录不存在: " + callRecordId); } deviceNo = record.getDeviceNo(); fileName = record.getRecordingFileName(); Date callStartTime = record.getCallStartTime(); Date callEndTime = record.getCallEndTime(); if (callStartTime == null) { throw new IllegalStateException("通话记录缺少开始时间: " + callRecordId); } // 2. 查询设备 Map device = voiceSyncMapper.getDeviceByNo(deviceNo); if (device == null || device.isEmpty()) { throw new IllegalStateException("语音设备不存在或未配置: " + deviceNo); } String cityCode = getStr(device, "ORGAN_ID"); String ip = getStr(device, "IP"); Integer port = getInt(device, "PORT", 80); String username = getStr(device, "USERNAME"); String password = getStr(device, "PASSWORD"); if (!StringUtils.hasText(ip)) { throw new IllegalStateException("设备IP为空: " + deviceNo); } if (!StringUtils.hasText(username) || !StringUtils.hasText(password)) { throw new IllegalStateException("设备账号密码未配置: " + deviceNo); } String deviceHost = buildHost(ip, port); // 3. 登录 String authToken = loginDevice(deviceHost, username, password); // 4. 反推录音盒上的录音ID String recordId = extractRecordId(callRecordId, deviceNo); // 5. 按通话时间前后各扩展1小时查询录音列表 long startEpoch = callStartTime.getTime() / 1000 - 3600; long endEpoch = (callEndTime != null ? callEndTime.getTime() : callStartTime.getTime()) / 1000 + 3600; String recordUrl = String.format("http://%s/service/record/~/time[%d,%d]", deviceHost, startEpoch, endEpoch); String recordData = httpVisitWithReauth(recordUrl, authToken, deviceHost, username, password); JSONArray records = vaaHttpUtil.parseRecordData(recordData); // 6. 匹配录音记录 JSONObject matched = null; for (int i = 0; i < records.size(); i++) { JSONObject rec = records.getJSONObject(i); if (recordId.equals(RecordParser.parseRecordId(rec))) { matched = rec; break; } } if (matched == null) { throw new IllegalStateException("录音盒上未找到该录音: " + recordId); } String filePath = RecordParser.parseFilePath(matched); if (filePath == null || filePath.isEmpty()) { throw new IllegalStateException("录音文件路径为空: " + recordId); } // 7. 强制重新下载录音文件 String localPath = FilePathUtil.buildLocalPath(downloadPath, cityCode, callStartTime, deviceNo, fileName); Path localFile = Paths.get(localPath); Files.deleteIfExists(localFile); String fileUrl = "http://" + deviceHost + filePath; vaaHttpUtil.httpDown(fileUrl, localPath, authToken); logger.info("【重新同步】录音下载完成: {}", fileName); // 8. 上传OSS(复用原OSS文件名覆盖,业务表无需更新) String ossFileName = extractOssFileName(record.getRecordingFilePath(), fileName); String ossPath = "voice/" + cityCode + "/" + new SimpleDateFormat("yyyy-MM-dd").format(callStartTime) + "/"; OssFileResponse ossResponse = uploadRecording(localFile.toFile(), ossFileName, ossPath); String previewUrl = ossResponse.getFileUrl(); if (previewUrl == null || previewUrl.trim().isEmpty()) { throw new IllegalStateException("OSS上传成功但未返回文件地址: " + ossFileName); } ossFilePath = previewUrl; // 9. 记录成功操作日志 saveResyncLog(callRecordId, deviceNo, fileName, previewUrl, "1", null); logger.info("【重新同步】成功: callRecordId={}, ossUrl={}", callRecordId, previewUrl); return "重新同步成功: " + callRecordId + " -> " + previewUrl; } catch (Exception e) { logger.error("【重新同步】失败: callRecordId={}, 原因={}", callRecordId, e.getMessage(), e); saveResyncLog(callRecordId, deviceNo, fileName, ossFilePath, "0", e.getMessage()); if (e instanceof RuntimeException) { throw (RuntimeException) e; } throw new RuntimeException("重新同步失败: " + e.getMessage(), e); } } /** * 上传录音文件到 OSS;G.729 格式自动转码为 PCM 后上传 * * 【降级】ffmpeg 不存在或转码失败时,回退为上传原文件,保证数据不丢失。 * 转码生成的临时文件在上传后清理。 * * @param localFile 本地录音文件 * @param fileName OSS 上的文件名(保持原文件名,业务表无需更新) * @param ossPath OSS 存储目录 * @return OSS 上传结果 */ private OssFileResponse uploadRecording(File localFile, String fileName, String ossPath) { File uploadFile = localFile; File transcodedFile = null; try { if (g729Converter.isG729(localFile)) { try { logger.info("【录音】检测到 G.729 编码,转码为 PCM: {}", fileName); transcodedFile = g729Converter.convertToPcm(localFile); uploadFile = transcodedFile; logger.info("【录音】转码完成: {} ({} -> {} 字节)", fileName, localFile.length(), transcodedFile.length()); } catch (Exception e) { logger.warn("【录音】G.729 转码失败,回退上传原文件: {}, 原因={}", fileName, e.getMessage()); uploadFile = localFile; } } return ossFileService.uploadFile(uploadFile, fileName, ossPath); } finally { if (transcodedFile != null && uploadFile == transcodedFile && !transcodedFile.delete()) { transcodedFile.deleteOnExit(); } } } /** * 从 call_record_id(格式:设备编号_录音ID)中提取录音盒上的录音ID */ private String extractRecordId(String callRecordId, String deviceNo) { if (deviceNo != null && callRecordId.startsWith(deviceNo + "_")) { return callRecordId.substring(deviceNo.length() + 1); } return callRecordId; } /** * 从原录音的 OSS 地址中提取 OSS 文件名(URL 最后一段),用于重新同步时覆盖原文件。 * 若地址为空或无法提取,则回退为录音盒原始文件名。 */ private String extractOssFileName(String recordingFilePath, String fallback) { if (StringUtils.hasText(recordingFilePath)) { String path = recordingFilePath; int queryIdx = path.indexOf('?'); if (queryIdx >= 0) { path = path.substring(0, queryIdx); } int slashIdx = path.lastIndexOf('/'); if (slashIdx >= 0 && slashIdx < path.length() - 1) { String name = path.substring(slashIdx + 1); if (StringUtils.hasText(name)) { return name; } } } return fallback; } /** * 保存重新同步操作记录到 mid_voice_resync_log 表 */ private void saveResyncLog(String callRecordId, String deviceNo, String fileName, String ossFilePath, String resyncStatus, String failReason) { try { MidVoiceResyncLog log = new MidVoiceResyncLog(); log.setCallRecordId(callRecordId); log.setDeviceNo(deviceNo); log.setRecordingFileName(fileName); log.setOssFilePath(ossFilePath); log.setResyncStatus(resyncStatus); log.setFailReason(failReason); log.setCreateTime(new Date()); resyncLogMapper.insert(log); } catch (Exception e) { logger.error("保存重新同步操作日志失败: callRecordId={}, 原因={}", callRecordId, e.getMessage()); } } /** * 同步单个设备 * * 流程:登录 → 获取分机号码 → 查录音列表 → 逐条下载上传保存 * 任何步骤失败直接记录日志,跳过该设备 */ private void syncSingleDevice(Map device) throws Exception { String deviceId = getStr(device, "ID"); String deviceNo = getStr(device, "UUID"); String cityName = getStr(device, "ORGAN_NAME"); String cityCode = getStr(device, "ORGAN_ID"); String ip = getStr(device, "IP"); Integer port = getInt(device, "PORT", 80); String orgCode = getStr(device, "ORG_CODE"); String username = getStr(device, "USERNAME"); String password = getStr(device, "PASSWORD"); logger.info("【设备】同步设备: ID={}, 编号={}, 机构={}, IP={}:{}", deviceId, deviceNo, cityName, ip, port); // 参数校验 if (!StringUtils.hasText(ip)) { logger.warn("【设备】IP为空,跳过: ID={}", deviceId); saveDeviceLog(deviceId, deviceNo, cityCode, cityName, ip, port, "1", "0", "设备IP为空"); return; } if (!StringUtils.hasText(username) || !StringUtils.hasText(password)) { logger.warn("【设备】账号密码为空,跳过: ID={}", deviceId); saveDeviceLog(deviceId, deviceNo, cityCode, cityName, ip, port, "1", "0", "设备账号密码未配置"); return; } if (!StringUtils.hasText(cityName)) { cityName = "unknown_" + deviceId; } String deviceHost = buildHost(ip, port); // 步骤1:登录 String authToken; try { authToken = loginDevice(deviceHost, username, password); saveDeviceLog(deviceId, deviceNo, cityCode, cityName, ip, port, "1", "1", null); } catch (Exception e) { saveDeviceLog(deviceId, deviceNo, cityCode, cityName, ip, port, "1", "0", "登录失败:" + e.getMessage()); logger.error("【设备】登录失败,跳过: IP={}, 原因={}", ip, e.getMessage()); return; } // 步骤2:获取分机号码 Map extNumbers = getExtensionNumbers(deviceHost, authToken); // 步骤3:计算同步时间范围 Date lastSyncTime = SyncTimeUtil.readLastSyncTime(downloadPath, deviceId); Date now = new Date(); if (lastSyncTime == null || DateUtil.getDateDoubleDiff(now, lastSyncTime) > 1.0) { lastSyncTime = DateUtil.addDayByDate(now, -1); } long startEpoch = lastSyncTime.getTime() / 1000; long endEpoch = now.getTime() / 1000; // 步骤4:获取录音列表(Token过期时自动重登录重试一次) JSONArray records; try { String recordUrl = String.format("http://%s/service/record/~/time[%d,%d]", deviceHost, startEpoch, endEpoch); String recordData = httpVisitWithReauth(recordUrl, authToken, deviceHost, username, password); records = vaaHttpUtil.parseRecordData(recordData); saveDeviceLog(deviceId, deviceNo, cityCode, cityName, ip, port, "2", "1", null); } catch (Exception e) { saveDeviceLog(deviceId, deviceNo, cityCode, cityName, ip, port, "2", "0", "查询录音列表失败:" + e.getMessage()); logger.error("【设备】获取录音列表失败: {}", e.getMessage()); return; } if (records == null || records.isEmpty()) { logger.info("【设备】无新录音: ID={}", deviceId); return; } logger.info("【设备】获取到 {} 条录音记录", records.size()); // 批量查询已存在的call_record_id String startTimeStr = DateUtil.formatDate(lastSyncTime, "yyyy-MM-dd HH:mm:ss"); String endTimeStr = DateUtil.formatDate(now, "yyyy-MM-dd HH:mm:ss"); Set existingIds = new HashSet<>(callRecordMapper.selectExistingCallRecordIds( deviceNo, startTimeStr, endTimeStr)); logger.debug("【设备】已存在 {} 条通话记录,将跳过", existingIds.size()); // 步骤5:逐条处理录音 int successCount = 0; int failCount = 0; int skipCount = 0; Date latestCallTime = null; for (int i = 0; i < records.size(); i++) { JSONObject rec = records.getJSONObject(i); try { // 解析时间用于更新同步进度(即使跳过也要更新,防止重复拉取) Long endTime = RecordParser.parseEndTime(rec); if (endTime != null) { Date recEndTime = new Date(endTime * 1000); if (latestCallTime == null || recEndTime.after(latestCallTime)) { latestCallTime = recEndTime; } } Date callTime = processSingleRecord(rec, deviceNo, cityName, cityCode, orgCode, deviceHost, authToken, extNumbers, existingIds, username, password); if (callTime != null) { successCount++; } else { skipCount++; } } catch (Exception e) { failCount++; logger.error("【录音】处理失败: {}, 原因={}", rec.getString("id"), e.getMessage()); } } // 只有本轮全部处理成功才推进游标,避免失败录音被同步时间跨过去而永久漏拉。 // 下轮重复查询到的成功记录会通过 callRecordId 防重跳过。 if (latestCallTime != null && failCount == 0) { SyncTimeUtil.writeLastSyncTime(downloadPath, deviceId, latestCallTime); } else if (failCount > 0) { logger.warn("【设备】本轮有{}条录音失败,不推进同步时间,下轮将重新查询", failCount); } logger.info("【设备】同步完成: ID={}, 成功{}条, 失败{}条", deviceId, successCount, failCount); } /** * 处理单条录音记录 * * OSS存储路径: voice/{cityCode}/{yyyy-MM-dd}/{fileName} * 下载、校验、OSS上传全部成功后才保存数据库记录。 * * @return 通话开始时间(成功时),null表示跳过 */ private Date processSingleRecord(JSONObject rec, String deviceNo, String cityName, String cityCode, String orgCode, String deviceHost, String authToken, Map extNumbers, Set existingIds, String username, String password) throws Exception { // 解析录音信息 String recordId = RecordParser.parseRecordId(rec); String filePath = RecordParser.parseFilePath(rec); Integer channel = RecordParser.parseChannel(rec); String phone = RecordParser.parsePhone(rec); boolean isOutgoing = RecordParser.isOutgoing(rec); boolean isAnswered = RecordParser.isAnswered(rec); Long begTime = RecordParser.parseBegTime(rec); Long endTime = RecordParser.parseEndTime(rec); if (filePath == null || filePath.isEmpty()) { return null; } if (begTime == null || endTime == null) { return null; } // 防重复 String callRecordId = deviceNo + "_" + recordId; if (existingIds.contains(callRecordId)) { logger.debug("【录音】已存在,跳过: {}", callRecordId); return null; } // 准备文件路径 Date callStartTime = new Date(begTime * 1000); String fileName = FilePathUtil.extractFileName(filePath); String localPath = FilePathUtil.buildLocalPath(downloadPath, cityCode, callStartTime, deviceNo, fileName); Path localFile = Paths.get(localPath); // 获取通道绑定的电话号码 String channelPhone = ""; if (channel != null && extNumbers != null) { channelPhone = extNumbers.getOrDefault(String.valueOf(channel), ""); } // 构建通话记录 MidVoiceCallRecord callRecord = buildCallRecord( callRecordId, deviceNo, cityCode, cityName, orgCode, fileName, callStartTime, new Date(endTime * 1000), (int) (endTime - begTime), channelPhone, phone, isOutgoing, isAnswered); // 下载录音文件(无重试) // 文件不存在或完整性校验失败时重新下载。 boolean needDownload = !Files.exists(localFile); if (!needDownload) { try { vaaHttpUtil.validateAudioFile(localFile.toFile()); } catch (Exception e) { logger.warn("【录音】本地文件校验失败,重新下载: {}, 原因={}", fileName, e.getMessage()); needDownload = true; Files.deleteIfExists(localFile); } } if (needDownload) { String fileUrl = "http://" + deviceHost + filePath; vaaHttpUtil.httpDown(fileUrl, localPath, authToken); logger.info("【录音】下载完成: {} ({} 字节)", fileName, Files.size(localFile)); } else { logger.debug("【录音】文件已存在,跳过下载: {}", fileName); } callRecord.setRecordingFileSize((int) Files.size(localFile)); // 上传到OSS(地市+日期目录);上传失败时整条处理失败,下轮重新拉取。 String ossPath = "voice/" + cityCode + "/" + new SimpleDateFormat("yyyy-MM-dd").format(callStartTime) + "/"; OssFileResponse ossResponse = uploadRecording(localFile.toFile(), fileName, ossPath); String previewUrl = ossResponse.getFileUrl(); if (previewUrl == null || previewUrl.trim().isEmpty()) { throw new IllegalStateException("OSS上传成功但未返回文件地址: " + fileName); } callRecord.setRecordingFilePath(previewUrl); logger.info("【录音】OSS上传成功: {}", previewUrl); // 保存到数据库 callRecordMapper.insert(callRecord); logger.info("【录音】保存成功: callRecordId={}", callRecordId); return callStartTime; } /** * 登录录音盒设备 */ private String loginDevice(String deviceHost, String username, String password) throws Exception { String loginUrl = String.format("http://%s/authorize?username=%s&password=%s", deviceHost, URLEncoder.encode(username, "UTF-8"), URLEncoder.encode(password, "UTF-8")); return vaaHttpUtil.httpLogin(loginUrl); } /** * HTTP访问API,Token过期时自动重新登录并重试一次 */ private String httpVisitWithReauth(String apiUrl, String authToken, String deviceHost, String username, String password) throws Exception { try { return vaaHttpUtil.httpVisit(apiUrl, authToken); } catch (Exception e) { if (e.getMessage() != null && e.getMessage().contains("认证失效")) { logger.warn("【设备】Token过期,自动重新登录并重试"); String newToken = loginDevice(deviceHost, username, password); return vaaHttpUtil.httpVisit(apiUrl, newToken); } throw e; } } /** * 构建通话记录实体 * * 号码解析规则: * - 呼出(state=1):主叫=本机号码(通道绑定号码),被叫=对方号码(phone字段) * - 呼入(state=2):主叫=对方号码(phone字段),被叫=本机号码(通道绑定号码) */ private MidVoiceCallRecord buildCallRecord(String callRecordId, String deviceNo, String cityCode, String cityName, String orgCode, String fileName, Date callStartTime, Date callEndTime, int duration, String channelPhone, String remotePhone, boolean isOutgoing, boolean isAnswered) { MidVoiceCallRecord record = new MidVoiceCallRecord(); record.setCallRecordId(callRecordId); record.setDeviceNo(deviceNo); record.setCityCode(cityCode); record.setCityName(cityName); record.setOrgCode(orgCode); record.setRecordingFileName(fileName); record.setCallStartTime(callStartTime); record.setCallEndTime(callEndTime); record.setCallDuration(duration); record.setCallDirection(isOutgoing ? "2" : "1"); // 1呼入,2呼出 String localNum = channelPhone != null ? channelPhone : ""; String remoteNum = remotePhone != null ? remotePhone : ""; if (isOutgoing) { record.setCallTel(localNum); // 主叫:本机 record.setCalledTel(remoteNum); // 被叫:对方 } else { record.setCallTel(remoteNum); // 主叫:对方 record.setCalledTel(localNum); // 被叫:本机 } record.setCallStatus(isAnswered ? "1" : "2"); // 1正常接通,2未接通 return record; } /** * 获取分机号码配置 */ private Map getExtensionNumbers(String deviceHost, String authToken) { try { String extUrl = "http://" + deviceHost + "/service/ext/number"; return vaaHttpUtil.getExtensionNumbers(extUrl, authToken); } catch (Exception e) { logger.warn("【设备】获取分机号码失败,将无法解析本机号码: {}", e.getMessage()); return Collections.emptyMap(); } } /** * 保存设备连接日志到 mid_voice_device_log 表 */ private void saveDeviceLog(String deviceId, String deviceNo, String cityCode, String cityName, String ipAddress, Integer devicePort, String connectType, String connectStatus, String failReason) { try { MidVoiceDeviceLog log = new MidVoiceDeviceLog(); log.setDeviceId(deviceId); log.setDeviceNo(deviceNo); log.setCityCode(cityCode); log.setCityName(cityName); log.setIpAddress(ipAddress); log.setDevicePort(devicePort); log.setConnectType(connectType); log.setConnectStatus(connectStatus); log.setFailReason(failReason); log.setCreateTime(new Date()); deviceLogMapper.insert(log); } catch (Exception e) { logger.error("保存设备连接日志失败: {}", e.getMessage()); } } // ==================== 工具方法 ==================== private String buildHost(String ip, Integer port) { return ip + (port != null && port != 80 ? ":" + port : ""); } private String getStr(Map map, String key) { Object v = map.get(key); return v != null ? v.toString() : null; } private Integer getInt(Map map, String key, Integer defaultVal) { Object v = map.get(key); if (v == null) return defaultVal; try { return Integer.parseInt(v.toString()); } catch (NumberFormatException e) { return defaultVal; } } }