feat: 增加上报时间

main/manager-api/pom.xml

@@ -258,4 +258,4 @@
             </plugin>
         </plugins>
     </build>
-</project>
+</project>

main/manager-api/src/main/java/xiaozhi/modules/agent/dto/AgentChatHistoryReportDTO.java

@@ -28,4 +28,6 @@ public class AgentChatHistoryReportDTO {
     private String content;
     @Schema(description = "base64编码的opus音频数据", example = "")
     private String audioBase64;
+    @Schema(description = "上报时间，十位时间戳，空时默认使用当前时间", example = "1745657732")
+    private Long reportTime;
 }

main/manager-api/src/main/java/xiaozhi/modules/agent/service/AgentChatAudioService.java

@@ -27,4 +27,4 @@ public interface AgentChatAudioService extends IService<AgentChatAudioEntity> {
      * @return 音频数据
      */
     byte[] getAudio(String audioId);
-}
+}

main/manager-api/src/main/java/xiaozhi/modules/agent/service/biz/impl/AgentChatHistoryBizServiceImpl.java

@@ -47,7 +47,8 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic
     public Boolean report(AgentChatHistoryReportDTO report) {
         String macAddress = report.getMacAddress();
         Byte chatType = report.getChatType();
-        log.info("小智设备聊天上报请求: macAddress={}, type={}", macAddress, chatType);
+        Long reportTimeMillis = null != report.getReportTime() ? report.getReportTime() * 1000 : System.currentTimeMillis();
+        log.info("小智设备聊天上报请求: macAddress={}, type={} reportTime={}", macAddress, chatType, reportTimeMillis);
 
         // 根据设备MAC地址查询对应的默认智能体，判断是否需要上报
         AgentEntity agentEntity = agentService.getDefaultAgentByMacAddress(macAddress);
@@ -59,10 +60,10 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic
         String agentId = agentEntity.getId();
 
         if (Objects.equals(chatHistoryConf, Constant.ChatHistoryConfEnum.RECORD_TEXT.getCode())) {
-            saveChatText(report, agentId, macAddress, null);
+            saveChatText(report, agentId, macAddress, null, reportTimeMillis);
         } else if (Objects.equals(chatHistoryConf, Constant.ChatHistoryConfEnum.RECORD_TEXT_AUDIO.getCode())) {
             String audioId = saveChatAudio(report);
-            saveChatText(report, agentId, macAddress, audioId);
+            saveChatText(report, agentId, macAddress, audioId, reportTimeMillis);
         }
 
         // 更新设备最后对话时间
@@ -92,8 +93,7 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic
     /**
      * 组装上报数据
      */
-    private void saveChatText(AgentChatHistoryReportDTO report, String agentId, String macAddress, String audioId) {
+    private void saveChatText(AgentChatHistoryReportDTO report, String agentId, String macAddress, String audioId, Long reportTime) {
-
         // 构建聊天记录实体
         AgentChatHistoryEntity entity = AgentChatHistoryEntity.builder()
                 .macAddress(macAddress)
@@ -102,6 +102,8 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic
                 .chatType(report.getChatType())
                 .content(report.getContent())
                 .audioId(audioId)
+                .createdAt(new Date(reportTime))
+                // NOTE(haotian): 2025/5/26 updateAt可以不设置，重点是createAt，而且这样可以看到上报延迟
                 .build();
 
         // 保存数据

main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentChatAudioServiceImpl.java

@@ -31,4 +31,4 @@ public class AgentChatAudioServiceImpl extends ServiceImpl<AiAgentChatAudioDao,
         AgentChatAudioEntity entity = getById(audioId);
         return entity != null ? entity.getAudio() : null;
     }
-}
+}

main/manager-api/src/main/resources/application-dev.yml

@@ -48,4 +48,4 @@ spring:
           max-active: 8      # 连接池最大连接数（使用负值表示没有限制）
           max-idle: 8        # 连接池中的最大空闲连接
           min-idle: 0        # 连接池中的最小空闲连接
-        shutdown-timeout: 100ms  # 客户端优雅关闭的等待时间
+        shutdown-timeout: 100ms  # 客户端优雅关闭的等待时间

main/xiaozhi-server/config/manage_api_client.py

@@ -160,7 +160,7 @@ def save_mem_local_short(mac_address: str, short_momery: str) -> Optional[Dict]:
 
 
 def report(
-    mac_address: str, session_id: str, chat_type: int, content: str, audio
+    mac_address: str, session_id: str, chat_type: int, content: str, audio, report_time
 ) -> Optional[Dict]:
     """带熔断的业务方法示例"""
     if not content or not ManageApiClient._instance:
@@ -174,6 +174,7 @@ def report(
                 "sessionId": session_id,
                 "chatType": chat_type,
                 "content": content,
+                "reportTime": report_time,
                 "audioBase64": (
                     base64.b64encode(audio).decode("utf-8") if audio else None
                 ),

main/xiaozhi-server/core/connection.py

@@ -890,11 +890,11 @@ class ConnectionHandler:
                 if item is None:  # 检测毒丸对象
                     break
 
-                type, text, audio_data = item
+                type, text, audio_data, report_time = item
 
                 try:
                     # 提交任务到线程池
-                    self.report_thread_pool.submit(self._process_report, type, text, audio_data)
+                    self.report_thread_pool.submit(self._process_report, type, text, audio_data, report_time)
                 except Exception as e:
                     self.logger.bind(tag=TAG).error(f"聊天记录上报线程异常: {e}")
             except queue.Empty:
@@ -904,11 +904,11 @@ class ConnectionHandler:
 
         self.logger.bind(tag=TAG).info("聊天记录上报线程已退出")
 
-    def _process_report(self, type, text, audio_data):
+    def _process_report(self, type, text, audio_data, report_time):
         """处理上报任务"""
         try:
             # 执行上报（传入二进制数据）
-            report(self, type, text, audio_data)
+            report(self, type, text, audio_data, report_time)
         except Exception as e:
             self.logger.bind(tag=TAG).error(f"上报处理异常: {e}")
         finally:
@@ -973,6 +973,12 @@ class ConnectionHandler:
             self.executor.shutdown(wait=False)
             self.executor = None
 
+        # 关闭上报线程池
+        if self.report_thread_pool:
+            self.report_thread_pool.shutdown(wait=False)
+            self.report_thread_pool = None
+            self.logger.bind(tag=TAG).info("上报线程池已关闭")
+
         self.logger.bind(tag=TAG).info("连接资源已释放")
 
     def clear_queues(self):

main/xiaozhi-server/core/handle/reportHandle.py

@@ -8,6 +8,7 @@ TTS上报功能已集成到ConnectionHandler类中。
 
 具体实现请参考core/connection.py中的相关代码。
 """
+import time
 
 import opuslib_next
 
@@ -16,7 +17,7 @@ from config.manage_api_client import report as manage_report
 TAG = __name__
 
 
-def report(conn, type, text, opus_data):
+def report(conn, type, text, opus_data, report_time):
     """执行聊天记录上报操作
 
     Args:
@@ -24,6 +25,7 @@ def report(conn, type, text, opus_data):
         type: 上报类型，1为用户，2为智能体
         text: 合成文本
         opus_data: opus音频数据
+        report_time: 上报时间
     """
     try:
         if opus_data:
@@ -37,6 +39,7 @@ def report(conn, type, text, opus_data):
             chat_type=type,
             content=text,
             audio=audio_data,
+            report_time=report_time,
         )
     except Exception as e:
         conn.logger.bind(tag=TAG).error(f"聊天记录上报失败: {e}")
@@ -104,12 +107,12 @@ def enqueue_tts_report(conn, text, opus_data):
     try:
         # 使用连接对象的队列，传入文本和二进制数据而非文件路径
         if conn.chat_history_conf == 2:
-            conn.report_queue.put((2, text, opus_data))
+            conn.report_queue.put((2, text, opus_data, int(time.time())))
             conn.logger.bind(tag=TAG).debug(
                 f"TTS数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} "
             )
         else:
-            conn.report_queue.put((2, text, None))
+            conn.report_queue.put((2, text, None, int(time.time())))
             conn.logger.bind(tag=TAG).debug(
                 f"TTS数据已加入上报队列: {conn.device_id}, 不上报音频"
             )
@@ -132,12 +135,12 @@ def enqueue_asr_report(conn, text, opus_data):
     try:
         # 使用连接对象的队列，传入文本和二进制数据而非文件路径
         if conn.chat_history_conf == 2:
-            conn.report_queue.put((1, text, opus_data))
+            conn.report_queue.put((1, text, opus_data, int(time.time())))
             conn.logger.bind(tag=TAG).debug(
                 f"ASR数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} "
             )
         else:
-            conn.report_queue.put((1, text, None))
+            conn.report_queue.put((1, text, None, int(time.time())))
             conn.logger.bind(tag=TAG).debug(
                 f"ASR数据已加入上报队列: {conn.device_id}, 不上报音频"
             )