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.mapper.MidVoiceCallRecordMapper; import com.threecloud.dataserviceyy.mapper.MidVoiceDeviceLogMapper; 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.net.URLEncoder; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; import java.util.*; /** * VAA录音盒定时同步服务 * * 【功能】从EBOX录音盒拉取录音 → 下载文件 → 上传OSS → 保存通话记录 * 【策略】增量同步、无重试、防重复、失败记录到日志表 * 【账户】每个设备的账号密码从 mid_voice_device_config 表读取 */ @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 VaaHttpUtil vaaHttpUtil; @Autowired private FileUploadUtil fileUploadUtil; @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/2 * * ?}") 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); } } /** * 同步单个设备 * * 流程:登录 → 获取分机号码 → 查录音列表 → 逐条下载上传保存 * 任何步骤失败直接记录日志,跳过该设备 */ 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 { String loginUrl = String.format("http://%s/authorize?username=%s&password=%s", deviceHost, urlEncode(username), urlEncode(password)); authToken = vaaHttpUtil.httpLogin(loginUrl); 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:获取录音列表(无重试) JSONArray records; try { String recordUrl = String.format("http://%s/service/record/~/time[%d,%d]", deviceHost, startEpoch, endEpoch); String recordData = vaaHttpUtil.httpVisit(recordUrl, authToken); 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()); // 步骤5:逐条处理录音 int successCount = 0; int failCount = 0; Date latestCallTime = null; for (int i = 0; i < records.size(); i++) { JSONObject rec = records.getJSONObject(i); try { Date callTime = processSingleRecord(rec, deviceNo, cityName, cityCode, orgCode, deviceHost, authToken, extNumbers); if (callTime != null) { successCount++; if (latestCallTime == null || callTime.after(latestCallTime)) { latestCallTime = callTime; } } } catch (Exception e) { failCount++; logger.error("【录音】处理失败: {}, 原因={}", rec.getString("id"), e.getMessage()); } } // 步骤6:保存同步时间 if (latestCallTime != null) { SyncTimeUtil.writeLastSyncTime(downloadPath, deviceId, latestCallTime); } logger.info("【设备】同步完成: ID={}, 成功{}条, 失败{}条", deviceId, successCount, failCount); } /** * 处理单条录音记录 * * @return 通话开始时间(成功时),null表示跳过 */ private Date processSingleRecord(JSONObject rec, String deviceNo, String cityName, String cityCode, String orgCode, String deviceHost, String authToken, Map extNumbers) 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; } // 防重复:根据 device_no + record_id 判断 String callRecordId = deviceNo + "_" + recordId; if (callRecordMapper.selectByCallRecordId(callRecordId) != null) { 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 ossPath = FilePathUtil.buildOssPath(cityCode, callStartTime, fileName); // 获取通道绑定的电话号码(从EBOX分机号码配置) 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); // 下载录音文件(无重试) if (!Files.exists(localFile) || Files.size(localFile) == 0) { 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 byte[] wavData = Files.readAllBytes(localFile); String ossUrl = fileUploadUtil.uploadWav(cityName, ossPath, fileName, wavData); callRecord.setRecordingFilePath(ossUrl); logger.info("【录音】上传OSS成功: {}", ossUrl); // 保存到数据库 callRecordMapper.insert(callRecord); logger.info("【录音】保存成功: callRecordId={}", callRecordId); return callStartTime; } /** * 构建通话记录实体 * * 号码解析规则: * - 呼出(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; } /** * 获取分机号码配置 * 调用 EBOX 接口 GET /service/ext/number * 返回 Map<通道号, 电话号码>,如 {"1":"8001","2":"8002"} */ 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; } } private String urlEncode(String value) throws Exception { return URLEncoder.encode(value, "UTF-8"); } }