语音拉取优化

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 * * ?"
 
 # ==================== 日志配置 ====================

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);

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<String, String> 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<String, String> 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 : "");
     }

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);
     }
-}
+}

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) {
+                    // 临时文件清理由下次目录清理兜底。
+                }
+            }
         }
     }
 }

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<channel, phoneNumber>
@@ -317,11 +319,22 @@ public class VaaHttpUtil {
             return new java.util.HashMap<>();
         }
         try {
-            com.alibaba.fastjson2.JSONObject jsonObj = JSON.parseObject(jsonData);
             Map<String, String> 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;