diff --git a/config/application-external.yml b/config/application-external.yml index 525b054..8ee4a35 100644 --- a/config/application-external.yml +++ b/config/application-external.yml @@ -24,11 +24,16 @@ server: oss: base-url: http://53.1.211.7/apiOss upload-path: voice/record/ + # 网络中断或OSS 5xx时最多尝试次数；4xx不会重试 + upload-max-attempts: 3 + # 线性退避基础时间：第1次失败等1秒，第2次失败等2秒 + upload-retry-delay-ms: 1000 # ==================== VAA语音盒同步配置 ==================== vaa-sync: download-path: ./vaa-recordings retain-days: 10 + # 每次回看昨天00:00到当前时间，防止录音盒延迟生成文件造成漏拉 sync-interval-cron: "0 0 0/8 * * ?" # ==================== 日志配置 ==================== diff --git a/src/main/java/com/threecloud/dataserviceyy/mapper/MidVoiceResyncLogMapper.java b/src/main/java/com/threecloud/dataserviceyy/mapper/MidVoiceResyncLogMapper.java index 27b8e07..ada6fa7 100644 --- a/src/main/java/com/threecloud/dataserviceyy/mapper/MidVoiceResyncLogMapper.java +++ b/src/main/java/com/threecloud/dataserviceyy/mapper/MidVoiceResyncLogMapper.java @@ -14,10 +14,10 @@ public interface MidVoiceResyncLogMapper { * 插入重新同步操作记录 */ @Insert("INSERT INTO mid_voice_resync_log (" + - "id, call_record_id, device_no, recording_file_name, oss_file_path, " + + "call_record_id, device_no, recording_file_name, oss_file_path, " + "resync_status, fail_reason, create_time" + ") VALUES (" + - "#{id}, #{callRecordId}, #{deviceNo}, #{recordingFileName}, #{ossFilePath}, " + + "#{callRecordId}, #{deviceNo}, #{recordingFileName}, #{ossFilePath}, " + "#{resyncStatus}, #{failReason}, #{createTime}" + ")") int insert(MidVoiceResyncLog log); diff --git a/src/main/java/com/threecloud/dataserviceyy/service/VaaSyncService.java b/src/main/java/com/threecloud/dataserviceyy/service/VaaSyncService.java index f8c6c36..e27ccca 100644 --- a/src/main/java/com/threecloud/dataserviceyy/service/VaaSyncService.java +++ b/src/main/java/com/threecloud/dataserviceyy/service/VaaSyncService.java @@ -31,7 +31,7 @@ import java.util.*; * VAA录音盒定时同步服务 * * 【功能】从EBOX录音盒拉取录音 → 下载文件 → 上传OSS → 保存通话记录 - * 【策略】增量同步、无重试、防重复、失败记录到日志表 + * 【策略】增量同步、防重复、上传有限重试、失败记录保留补拉 * 【账户】每个设备的账号密码从 mid_voice_device_config 表读取 * 【OSS存储路径】voice/{地市编码}/{yyyy-MM-dd}/{文件名} */ @@ -39,6 +39,7 @@ import java.util.*; public class VaaSyncService { private static final Logger logger = LoggerFactory.getLogger(VaaSyncService.class); + private static long lastGeneratedId; @Autowired private VoiceSyncMapper voiceSyncMapper; @@ -221,7 +222,7 @@ public class VaaSyncService { * 上传录音文件到 OSS；G.729 格式自动转码为 PCM 后上传 * * 【降级】ffmpeg 不存在或转码失败时，回退为上传原文件，保证数据不丢失。 - * 转码生成的临时文件在上传后清理。 + * 转码文件在上传成功后清理；上传失败时保留，供下一轮直接复用。 * * @param localFile 本地录音文件 * @param fileName OSS 上的文件名（保持原文件名，业务表无需更新） @@ -231,11 +232,25 @@ public class VaaSyncService { private OssFileResponse uploadRecording(File localFile, String fileName, String ossPath) { File uploadFile = localFile; File transcodedFile = null; + boolean uploadSucceeded = false; try { if (g729Converter.isG729(localFile)) { try { - logger.info("【录音】检测到 G.729 编码，转码为 PCM: {}", fileName); - transcodedFile = g729Converter.convertToPcm(localFile); + transcodedFile = new File(localFile.getParentFile(), localFile.getName() + ".pcm.wav"); + if (transcodedFile.isFile()) { + try { + vaaHttpUtil.validateAudioFile(transcodedFile); + logger.info("【录音】复用上次已完成的PCM转码文件: {} ({} 字节)", + fileName, transcodedFile.length()); + } catch (Exception invalidCache) { + Files.deleteIfExists(transcodedFile.toPath()); + transcodedFile = null; + } + } + if (transcodedFile == null) { + logger.info("【录音】检测到 G.729 编码，转码为 PCM: {}", fileName); + transcodedFile = g729Converter.convertToPcm(localFile); + } uploadFile = transcodedFile; logger.info("【录音】转码完成: {} ({} -> {} 字节)", fileName, localFile.length(), transcodedFile.length()); @@ -244,9 +259,12 @@ public class VaaSyncService { uploadFile = localFile; } } - return ossFileService.uploadFile(uploadFile, fileName, ossPath); + OssFileResponse response = ossFileService.uploadFile(uploadFile, fileName, ossPath); + uploadSucceeded = true; + return response; } finally { - if (transcodedFile != null && uploadFile == transcodedFile && !transcodedFile.delete()) { + // 上传失败时保留转码结果供下轮直接重试，避免同一录音反复执行ffmpeg。 + if (uploadSucceeded && transcodedFile != null && uploadFile == transcodedFile && !transcodedFile.delete()) { transcodedFile.deleteOnExit(); } } @@ -355,11 +373,15 @@ public class VaaSyncService { 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); + // 每次回看到昨天00:00，兼容录音盒延迟生成/延迟可见的录音。 + // 已入库数据由 call_record_id 批量去重，不会重复下载和入库。 + Date lastSyncTime = startOfYesterday(); + Date savedSyncTime = SyncTimeUtil.readLastSyncTime(downloadPath, deviceId); + // 若历史上有失败记录尚未跨过，保留更早的补拉起点，直到后续处理成功。 + if (savedSyncTime != null && savedSyncTime.before(lastSyncTime)) { + lastSyncTime = savedSyncTime; } + Date now = new Date(); long startEpoch = lastSyncTime.getTime() / 1000; long endEpoch = now.getTime() / 1000; @@ -379,6 +401,7 @@ public class VaaSyncService { if (records == null || records.isEmpty()) { logger.info("【设备】无新录音: ID={}", deviceId); + SyncTimeUtil.writeLastSyncTime(downloadPath, deviceId, now); return; } logger.info("【设备】获取到 {} 条录音记录", records.size()); @@ -395,6 +418,7 @@ public class VaaSyncService { int failCount = 0; int skipCount = 0; Date latestCallTime = null; + Date earliestFailedTime = null; for (int i = 0; i < records.size(); i++) { JSONObject rec = records.getJSONObject(i); @@ -417,16 +441,32 @@ public class VaaSyncService { } } catch (Exception e) { failCount++; + Long failedEndTime = RecordParser.parseEndTime(rec); + if (failedEndTime != null) { + Date failedTime = new Date(failedEndTime * 1000); + if (earliestFailedTime == null || failedTime.before(earliestFailedTime)) { + earliestFailedTime = failedTime; + } + } logger.error("【录音】处理失败: {}, 原因={}", rec.getString("id"), e.getMessage()); } } - // 只有本轮全部处理成功才推进游标，避免失败录音被同步时间跨过去而永久漏拉。 - // 下轮重复查询到的成功记录会通过 callRecordId 防重跳过。 + // 全部成功时推进到最新记录；部分失败时推进到最早失败记录前1秒。 + // 这样既不会跨过失败录音，也避免每次从旧游标重新扫描整段历史数据。 if (latestCallTime != null && failCount == 0) { SyncTimeUtil.writeLastSyncTime(downloadPath, deviceId, latestCallTime); } else if (failCount > 0) { - logger.warn("【设备】本轮有{}条录音失败，不推进同步时间，下轮将重新查询", failCount); + if (earliestFailedTime != null) { + Date retryFrom = new Date(Math.max(lastSyncTime.getTime(), earliestFailedTime.getTime() - 1000L)); + if (retryFrom.after(lastSyncTime)) { + SyncTimeUtil.writeLastSyncTime(downloadPath, deviceId, retryFrom); + } + logger.warn("【设备】本轮有{}条录音失败，游标推进至最早失败记录前: {}，下轮从此处补拉", + failCount, retryFrom); + } else { + logger.warn("【设备】本轮有{}条录音失败且无法解析失败时间，不推进同步时间", failCount); + } } logger.info("【设备】同步完成: ID={}, 成功{}条, 失败{}条", deviceId, successCount, failCount); @@ -563,8 +603,7 @@ public class VaaSyncService { int duration, String channelPhone, String remotePhone, boolean isOutgoing, boolean isAnswered) { MidVoiceCallRecord record = new MidVoiceCallRecord(); - long generatedId = System.currentTimeMillis() * 1000L + (long)(Math.random() * 1000L); - record.setId(generatedId); + record.setId(nextGeneratedId()); record.setCallRecordId(callRecordId); record.setDeviceNo(deviceNo); record.setCityCode(cityCode); @@ -595,10 +634,22 @@ public class VaaSyncService { * 获取分机号码配置 */ private Map getExtensionNumbers(String deviceHost, String authToken) { + String standardUrl = "http://" + deviceHost + "/service/setting/ext_number"; try { - String extUrl = "http://" + deviceHost + "/service/ext/number"; - return vaaHttpUtil.getExtensionNumbers(extUrl, authToken); + return vaaHttpUtil.getExtensionNumbers(standardUrl, authToken); } catch (Exception e) { + // 不同批次固件的分机接口路径不一致；标准接口不存在时兼容旧路径。 + if (e.getMessage() != null && e.getMessage().contains("HTTP状态码: 404")) { + String compatibleUrl = "http://" + deviceHost + "/service/ext/number"; + try { + logger.debug("【设备】标准分机接口不存在，回退兼容接口: {}", compatibleUrl); + return vaaHttpUtil.getExtensionNumbers(compatibleUrl, authToken); + } catch (Exception fallbackError) { + logger.warn("【设备】获取分机号码失败，将无法解析本机号码: 标准接口={}, 兼容接口={}", + e.getMessage(), fallbackError.getMessage()); + return Collections.emptyMap(); + } + } logger.warn("【设备】获取分机号码失败，将无法解析本机号码: {}", e.getMessage()); return Collections.emptyMap(); } @@ -612,9 +663,7 @@ public class VaaSyncService { String connectStatus, String failReason) { try { MidVoiceDeviceLog log = new MidVoiceDeviceLog(); - // 代码生成唯一分布式数值 ID (雪花算法/时间戳+序列号)，避免数据库无自增序列导致非空约束报错 - long generatedId = System.currentTimeMillis() * 1000L + (long)(Math.random() * 1000L); - log.setId(generatedId); + log.setId(nextGeneratedId()); log.setDeviceId(deviceId); log.setDeviceNo(deviceNo); log.setCityCode(cityCode); @@ -633,6 +682,31 @@ public class VaaSyncService { // ==================== 工具方法 ==================== + /** + * 单实例内单调递增的数值ID，避免同一毫秒内使用随机数造成主键碰撞。 + */ + private static synchronized long nextGeneratedId() { + long candidate = System.currentTimeMillis() * 1000L; + if (candidate <= lastGeneratedId) { + candidate = lastGeneratedId + 1; + } + lastGeneratedId = candidate; + return candidate; + } + + /** + * 返回服务器当地时区的昨天00:00:00。 + */ + private Date startOfYesterday() { + Calendar calendar = Calendar.getInstance(); + calendar.add(Calendar.DAY_OF_MONTH, -1); + calendar.set(Calendar.HOUR_OF_DAY, 0); + calendar.set(Calendar.MINUTE, 0); + calendar.set(Calendar.SECOND, 0); + calendar.set(Calendar.MILLISECOND, 0); + return calendar.getTime(); + } + private String buildHost(String ip, Integer port) { return ip + (port != null && port != 80 ? ":" + port : ""); } diff --git a/src/main/java/com/threecloud/dataserviceyy/util/OssFileService.java b/src/main/java/com/threecloud/dataserviceyy/util/OssFileService.java index b19cbe3..b4288ab 100644 --- a/src/main/java/com/threecloud/dataserviceyy/util/OssFileService.java +++ b/src/main/java/com/threecloud/dataserviceyy/util/OssFileService.java @@ -11,6 +11,7 @@ import org.springframework.stereotype.Component; import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; import org.springframework.web.client.RestTemplate; +import org.springframework.web.client.HttpClientErrorException; import javax.annotation.PostConstruct; import javax.net.ssl.HostnameVerifier; @@ -45,6 +46,12 @@ public class OssFileService { @Value("${oss.upload-path:voice/record/}") private String defaultUploadPath; + @Value("${oss.upload-max-attempts:3}") + private int uploadMaxAttempts; + + @Value("${oss.upload-retry-delay-ms:1000}") + private long uploadRetryDelayMs; + private RestTemplate restTemplate; @PostConstruct @@ -80,6 +87,42 @@ public class OssFileService { String url = ossBaseUrl + "/oss/upload"; String path = uploadPath.endsWith("/") ? uploadPath : uploadPath + "/"; + if (file == null || !file.isFile() || file.length() <= 0) { + throw new IllegalArgumentException("待上传文件不存在或为空: " + (file == null ? "null" : file.getPath())); + } + + int attempts = Math.max(1, uploadMaxAttempts); + Exception lastError = null; + for (int attempt = 1; attempt <= attempts; attempt++) { + try { + return doUpload(file, fileName, path, url); + } catch (HttpClientErrorException e) { + // 4xx 属于请求参数、权限或大小限制等确定性错误，重复请求没有意义。 + logger.error("文件上传被OSS拒绝，不再重试: fileName={}, status={}", fileName, e.getStatusCode()); + throw new RuntimeException("文件上传失败: " + e.getMessage(), e); + } catch (Exception e) { + lastError = e; + if (attempt >= attempts) { + break; + } + long delay = Math.max(0L, uploadRetryDelayMs) * attempt; + logger.warn("文件上传失败，准备重试: fileName={}, attempt={}/{}, delay={}ms, 原因={}", + fileName, attempt, attempts, delay, e.getMessage()); + try { + Thread.sleep(delay); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw new RuntimeException("文件上传重试被中断", interrupted); + } + } + } + + logger.error("文件上传最终失败: fileName={}, attempts={}", fileName, attempts, lastError); + throw new RuntimeException("文件上传失败: " + (lastError == null ? "未知异常" : lastError.getMessage()), lastError); + } + + private OssFileResponse doUpload(File file, String fileName, String path, String url) { + try { HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.MULTIPART_FORM_DATA); @@ -115,9 +158,8 @@ public class OssFileService { } else { throw new RuntimeException("上传失败, HTTP状态码: " + response.getStatusCode()); } - } catch (Exception e) { - logger.error("文件上传异常: fileName={}", fileName, e); - throw new RuntimeException("文件上传失败: " + e.getMessage(), e); + } catch (RuntimeException e) { + throw e; } } @@ -184,4 +226,4 @@ public class OssFileService { requestFactory.setBufferRequestBody(false); return new RestTemplate(requestFactory); } -} \ No newline at end of file +} diff --git a/src/main/java/com/threecloud/dataserviceyy/util/SyncTimeUtil.java b/src/main/java/com/threecloud/dataserviceyy/util/SyncTimeUtil.java index b145ec2..ac5c474 100644 --- a/src/main/java/com/threecloud/dataserviceyy/util/SyncTimeUtil.java +++ b/src/main/java/com/threecloud/dataserviceyy/util/SyncTimeUtil.java @@ -5,6 +5,8 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; import java.util.Date; /** @@ -53,18 +55,35 @@ public class SyncTimeUtil { public static void writeLastSyncTime(String basePath, String deviceId, Date time) { String markerDir = FilePathUtil.buildSyncMarkerDir(basePath); File dir = new File(markerDir); - if (!dir.exists()) { - dir.mkdirs(); + if (!dir.exists() && !dir.mkdirs() && !dir.exists()) { + logger.warn("创建同步标记目录失败: {}", markerDir); + return; } String markerPath = FilePathUtil.buildSyncMarkerPath(basePath, deviceId); File markerFile = new File(markerPath); + Path tempFile = null; try { - Files.write(markerFile.toPath(), String.valueOf(time.getTime()).getBytes()); + tempFile = Files.createTempFile(dir.toPath(), deviceId + ".", ".tmp"); + Files.write(tempFile, String.valueOf(time.getTime()).getBytes("UTF-8")); + try { + Files.move(tempFile, markerFile.toPath(), + StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); + } catch (java.nio.file.AtomicMoveNotSupportedException e) { + Files.move(tempFile, markerFile.toPath(), StandardCopyOption.REPLACE_EXISTING); + } logger.debug("保存同步时间成功: deviceId={}, time={}", deviceId, time); } catch (Exception e) { logger.warn("保存同步时间失败: deviceId={}, error={}", deviceId, e.getMessage()); + } finally { + if (tempFile != null) { + try { + Files.deleteIfExists(tempFile); + } catch (Exception ignored) { + // 临时文件清理由下次目录清理兜底。 + } + } } } } diff --git a/src/main/java/com/threecloud/dataserviceyy/util/VaaHttpUtil.java b/src/main/java/com/threecloud/dataserviceyy/util/VaaHttpUtil.java index 76dfc65..f5618cf 100644 --- a/src/main/java/com/threecloud/dataserviceyy/util/VaaHttpUtil.java +++ b/src/main/java/com/threecloud/dataserviceyy/util/VaaHttpUtil.java @@ -44,11 +44,12 @@ public class VaaHttpUtil { * @throws Exception 登录失败直接抛出异常 */ public String httpLogin(String loginUrl) throws Exception { - logger.debug("正在登录录音盒: {}", loginUrl); + URL url = new URL(loginUrl); + logger.debug("正在登录录音盒: {}://{}{}（账号密码已脱敏）", + url.getProtocol(), url.getAuthority(), url.getPath()); HttpURLConnection conn = null; try { - URL url = new URL(loginUrl); conn = (HttpURLConnection) url.openConnection(); conn.setInstanceFollowRedirects(false); conn.setConnectTimeout(CONNECT_TIMEOUT); @@ -306,7 +307,8 @@ public class VaaHttpUtil { /** * 获取分机号码配置 - * EBOX接口: GET /service/ext/number + * 标准EBOX接口: GET /service/setting/ext_number + * 兼容部分固件接口: GET /service/ext/number * @param extUrl 接口地址 * @param authorization 登录后返回的Authorization Cookie值 * @return 通道-分机号映射 Map @@ -317,11 +319,22 @@ public class VaaHttpUtil { return new java.util.HashMap<>(); } try { - com.alibaba.fastjson2.JSONObject jsonObj = JSON.parseObject(jsonData); Map result = new java.util.HashMap<>(); - for (String key : jsonObj.keySet()) { - String value = jsonObj.getString(key); - result.put(key, value != null ? value : ""); + String trimmed = jsonData.trim(); + if (trimmed.startsWith("[")) { + // 手册标准响应为数组，下标从0开始，对应通道1、2、3…… + JSONArray numbers = JSON.parseArray(trimmed); + for (int i = 0; i < numbers.size(); i++) { + Object value = numbers.get(i); + result.put(String.valueOf(i + 1), value == null ? "" : String.valueOf(value)); + } + } else { + // 兼容返回 {"1":"801","2":"802"} 的固件。 + com.alibaba.fastjson2.JSONObject jsonObj = JSON.parseObject(trimmed); + for (String key : jsonObj.keySet()) { + String value = jsonObj.getString(key); + result.put(key, value != null ? value : ""); + } } logger.info("获取到分机号码配置: {} 条", result.size()); return result;