fix: sync mid table repeat data

Author: zgqgit

src/backend/bisheng/script/sync_increment_table.py

@@ -3,6 +3,7 @@
 from bisheng.api.services.workflow import WorkFlowService
 from bisheng.knowledge.domain.services.knowledge_service import KnowledgeService
 from bisheng.user.domain.services.user import UserService
+from bisheng.worker import sync_mid_user_interact_dtl
 from bisheng.worker.telemetry.mid_table import sync_mid_user_increment, sync_mid_knowledge_increment, \
     sync_mid_app_increment
 
@@ -46,7 +47,14 @@ def sync_app_increment_table_all():
     sync_mid_app_increment(start_date, end_date)
 
 
+def sync_user_interact_dtl_all():
+    first_date = datetime(2025, 12, 1).isoformat()
+    end_date = datetime.now().isoformat()
+    sync_mid_user_interact_dtl(first_date, end_date)
+
+
 if __name__ == '__main__':
     sync_user_increment_table_all()
     sync_knowledge_increment_table_all()
     sync_app_increment_table_all()
+    sync_user_interact_dtl_all()

src/backend/bisheng/telemetry/domain/mid_table/base.py

@@ -1,8 +1,8 @@
-from typing import Dict, Any, List
+from typing import Dict, Any, List, Optional
 
 from elasticsearch import Elasticsearch, AsyncElasticsearch, exceptions as es_exceptions, helpers
 from loguru import logger
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from bisheng.common.schemas.telemetry.base_telemetry_schema import UserContext
 from bisheng.common.services import telemetry_service
@@ -32,6 +32,7 @@
 
 class BaseRecord(UserContext):
     timestamp: int
+    es_id: Optional[str] = Field(default=None)
 
 
 class BaseMidTable(BaseModel):
@@ -96,14 +97,16 @@ def get_latest_record_time_sync(self) -> int | None:
             return latest_time
         return None
 
-    def insert_records_sync(self, records: List[BaseModel]) -> None:
+    def insert_records_sync(self, records: List[BaseRecord]) -> None:
         """ 批量插入记录 """
         actions = []
         for rec in records:
             action = {
                 "_index": self._index_name,
-                "_source": rec.model_dump()
+                "_source": rec.model_dump(exclude={"es_id"}),
             }
+            if rec.es_id is not None:
+                action["_id"] = rec.es_id
             actions.append(action)
         helpers.bulk(self._es_client_sync, actions)
 

src/backend/bisheng/telemetry/domain/mid_table/user_interact.py

@@ -25,7 +25,7 @@ def get_records_by_time_range_sync(self, start_time: int, end_time: int, page: i
             "bool": {
                 "filter": [
                     {"term": {"event_type": {"value": BaseTelemetryTypeEnum.MESSAGE_FEEDBACK.value}}},
-                    {"range": {"timestamp": {"gte": start_time, "lt": end_time}}}
+                    {"range": {"timestamp": {"gte": start_time * 1000, "lt": end_time * 1000}}}
                 ]
             }
         }

src/backend/bisheng/worker/telemetry/mid_table.py

@@ -69,6 +69,7 @@ def sync_mid_user_increment(start_date: str = None, end_date: str = None):
         records = []
         for user in user_list:
             records.append(UserIncrementRecord(
+                es_id=f"user_{user.user_id}",
                 user_id=user.user_id,
                 user_name=user.user_name,
                 user_group_infos=[UserGroupInfo(user_group_id=group.id, user_group_name=group.group_name)
@@ -121,6 +122,7 @@ def sync_mid_app_increment(start_date: str = None, end_date: str = None):
         for app in app_list:
             user = user_map.get(app['user_id'], None)
             records.append(AppIncrementRecord(
+                es_id=f"app_{app['id']}",
                 user_id=app['user_id'],
                 user_name=user.user_name if user else "",
                 user_group_infos=[UserGroupInfo(user_group_id=group.id, user_group_name=group.group_name)
@@ -167,6 +169,7 @@ def sync_mid_knowledge_increment(start_date: str = None, end_date: str = None):
         for knowledge in knowledge_list:
             user = user_map.get(knowledge.user_id, None)
             records.append(KnowledgeIncrementRecord(
+                es_id=f"knowledge_{knowledge.id}",
                 user_id=knowledge.user_id,
                 user_name=user.user_name if user else "",
                 user_group_infos=[UserGroupInfo(user_group_id=group.id, user_group_name=group.group_name)
@@ -204,16 +207,18 @@ def sync_mid_user_interact_dtl(start_date: str = None, end_date: str = None):
             break
         records = []
         for record in result:
+            es_id = record['_id']
             record = record['_source']
             records.append(UserInteractRecord(
+                es_id=es_id,
                 user_id=record['user_context']['user_id'],
                 user_name=record['user_context']['user_name'],
                 user_group_infos=[UserGroupInfo(user_group_id=group['user_group_id'],
                                                 user_group_name=group['user_group_name'])
                                   for group in record['user_context'].get('user_group_infos', [])],
                 user_role_infos=[UserRoleInfo(role_id=role['role_id'],
                                               role_name=role['role_name'],
-                                              group_id=role['group_id'])
+                                              group_id=role.get('group_id', 0))
                                  for role in record['user_context'].get('user_role_infos', [])],
                 event_id=record['event_id'],
                 timestamp=record['timestamp'],