fix: 修复文档状态数据数据错误问题 (#1697)

apps/common/db/sql_execute.py

@@ -36,8 +36,9 @@ def update_execute(sql: str, params):
       """
     with connection.cursor() as cursor:
         cursor.execute(sql, params)
+        affected_rows = cursor.rowcount
         cursor.close()
-        return None
+        return affected_rows
 
 
 def select_list(sql: str, params: List):

apps/common/event/listener_manage.py

@@ -10,11 +10,12 @@ import datetime
 import logging
 import os
 import threading
+import time
 import traceback
 from typing import List
 
 import django.db.models
-from django.db import models
+from django.db import models, transaction
 from django.db.models import QuerySet
 from django.db.models.functions import Substr, Reverse
 from langchain_core.embeddings import Embeddings
@@ -168,6 +169,7 @@ class ListenerManagement:
     @staticmethod
     def get_aggregation_document_status(document_id):
         def aggregation_document_status():
+            pass
             sql = get_file_content(
                 os.path.join(PROJECT_DIR, "apps", "dataset", 'sql', 'update_document_status_meta.sql'))
             native_update({'document_custom_sql': QuerySet(Document).filter(id=document_id)}, sql, with_table_name=True)

apps/common/util/page_utils.py

@@ -18,10 +18,11 @@ def page(query_set, page_size, handler, is_the_task_interrupted=lambda: False):
     @param is_the_task_interrupted: 任务是否被中断
     @return:
     """
+    query = query_set.order_by("id")
     count = query_set.count()
     for i in range(0, ceil(count / page_size)):
         if is_the_task_interrupted():
             return
         offset = i * page_size
-        paragraph_list = query_set[offset: offset + page_size]
+        paragraph_list = query.all()[offset: offset + page_size]
         handler(paragraph_list)

apps/dataset/serializers/document_serializers.py

@@ -613,7 +613,8 @@ class DocumentSerializers(ApiMixin, serializers.Serializer):
             document_id = self.data.get("document_id")
             ListenerManagement.update_status(QuerySet(Document).filter(id=document_id), TaskType.EMBEDDING,
                                              State.PENDING)
-            ListenerManagement.update_status(QuerySet(Paragraph).filter(document_id=document_id), TaskType.EMBEDDING,
+            ListenerManagement.update_status(QuerySet(Paragraph).filter(document_id=document_id),
+                                             TaskType.EMBEDDING,
                                              State.PENDING)
             ListenerManagement.get_aggregation_document_status(document_id)()
             embedding_model_id = get_embedding_model_id_by_dataset_id(dataset_id=self.data.get('dataset_id'))
@@ -708,8 +709,8 @@ class DocumentSerializers(ApiMixin, serializers.Serializer):
 
         @staticmethod
         def post_embedding(result, document_id, dataset_id):
-            model_id = get_embedding_model_id_by_dataset_id(dataset_id)
+            DocumentSerializers.Operate(
-            embedding_by_document.delay(document_id, model_id)
+                data={'dataset_id': dataset_id, 'document_id': document_id}).refresh()
             return result
 
         @staticmethod
@@ -907,8 +908,8 @@ class DocumentSerializers(ApiMixin, serializers.Serializer):
         @staticmethod
         def post_embedding(document_list, dataset_id):
             for document_dict in document_list:
-                model_id = get_embedding_model_id_by_dataset_id(dataset_id)
+                DocumentSerializers.Operate(
-                embedding_by_document.delay(document_dict.get('id'), model_id)
+                    data={'dataset_id': dataset_id, 'document_id': document_dict.get('id')}).refresh()
             return document_list
 
         @post(post_function=post_embedding)

ui/src/views/document/component/Status.vue

@@ -1,51 +1,13 @@
 <template>
-  <el-popover placement="top" :width="450" trigger="hover">
+  <el-popover v-model:visible="visible" placement="top" :width="450" trigger="hover">
-    <template #default>
+    <template #default
-      <el-row :gutter="3" v-for="status in statusTable" :key="status.type">
+      ><StatusTable
-        <el-col :span="4">{{ taskTypeMap[status.type] }} </el-col>
+        v-if="visible"
-        <el-col :span="4">
+        :status="status"
-          <el-text v-if="status.state === State.SUCCESS || status.state === State.REVOKED">
+        :statusMeta="statusMeta"
-            <el-icon class="success"><SuccessFilled /></el-icon>
+        :taskTypeMap="taskTypeMap"
-            {{ stateMap[status.state](status.type) }}
+        :stateMap="stateMap"
-          </el-text>
+      ></StatusTable>
-          <el-text v-else-if="status.state === State.FAILURE">
-            <el-icon class="danger"><CircleCloseFilled /></el-icon>
-            {{ stateMap[status.state](status.type) }}
-          </el-text>
-          <el-text v-else-if="status.state === State.STARTED">
-            <el-icon class="is-loading primary"><Loading /></el-icon>
-            {{ stateMap[status.state](status.type) }}
-          </el-text>
-          <el-text v-else-if="status.state === State.PENDING">
-            <el-icon class="is-loading primary"><Loading /></el-icon>
-            {{ stateMap[status.state](status.type) }}
-          </el-text>
-          <el-text v-else-if="aggStatus?.value === State.REVOKE">
-            <el-icon class="is-loading primary"><Loading /></el-icon>
-            {{ stateMap[aggStatus.value](aggStatus.key) }}
-          </el-text>
-        </el-col>
-        <el-col :span="5">
-          完成
-          {{
-            Object.keys(status.aggs ? status.aggs : {})
-              .filter((k) => k == State.SUCCESS)
-              .map((k) => status.aggs[k])
-              .reduce((x: any, y: any) => x + y, 0)
-          }}/{{
-            Object.values(status.aggs ? status.aggs : {}).reduce((x: any, y: any) => x + y, 0)
-          }}
-        </el-col>
-        <el-col :span="9">
-          {{
-            status.time
-              ? status.time[
-                  status.state == State.REVOKED ? State.REVOKED : State.PENDING
-                ]?.substring(0, 19)
-              : undefined
-          }}
-        </el-col>
-      </el-row>
     </template>
     <template #reference>
       <el-text v-if="aggStatus?.value === State.SUCCESS || aggStatus?.value === State.REVOKED">
@@ -72,11 +34,11 @@
   </el-popover>
 </template>
 <script setup lang="ts">
-import { computed } from 'vue'
+import { computed, ref } from 'vue'
-import { Status, TaskType, State, type TaskTypeInterface } from '@/utils/status'
+import { TaskType, State } from '@/utils/status'
-import { mergeWith } from 'lodash'
+import StatusTable from '@/views/document/component/StatusTable.vue'
 const props = defineProps<{ status: string; statusMeta: any }>()
-
+const visible = ref<boolean>(false)
 const checkList: Array<string> = [
   State.REVOKE,
   State.STARTED,
@@ -112,56 +74,5 @@ const stateMap: any = {
   [State.FAILURE]: (type: number) => '失败',
   [State.SUCCESS]: (type: number) => '成功'
 }
-
-const parseAgg = (agg: { count: number; status: string }) => {
-  const status = new Status(agg.status)
-  return Object.keys(TaskType)
-    .map((key) => {
-      const value = TaskType[key as keyof TaskTypeInterface]
-      return { [value]: { [status.task_status[value]]: agg.count } }
-    })
-    .reduce((x, y) => ({ ...x, ...y }), {})
-}
-
-const customizer: (x: any, y: any) => any = (objValue: any, srcValue: any) => {
-  if (objValue == undefined && srcValue) {
-    return srcValue
-  }
-  if (srcValue == undefined && objValue) {
-    return objValue
-  }
-  // 如果是数组，我们将元素进行聚合
-  if (typeof objValue === 'object' && typeof srcValue === 'object') {
-    // 若是object类型的对象，我们进行递归
-    return mergeWith(objValue, srcValue, customizer)
-  } else {
-    // 否则，单纯的将值进行累加
-    return objValue + srcValue
-  }
-}
-const aggs = computed(() => {
-  return (props.statusMeta.aggs ? props.statusMeta.aggs : [])
-    .map((agg: any) => {
-      return parseAgg(agg)
-    })
-    .reduce((x: any, y: any) => {
-      return mergeWith(x, y, customizer)
-    }, {})
-})
-
-const statusTable = computed(() => {
-  return Object.keys(TaskType)
-    .map((key) => {
-      const value = TaskType[key as keyof TaskTypeInterface]
-      const parseStatus = new Status(props.status)
-      return {
-        type: value,
-        state: parseStatus.task_status[value],
-        aggs: aggs.value[value],
-        time: props.statusMeta.state_time[value]
-      }
-    })
-    .filter((item) => item.state !== State.IGNORED)
-})
 </script>
 <style lang="scss" scoped></style>

ui/src/views/document/component/StatusTable.vue

@@ -0,0 +1,104 @@
+<template>
+  <el-row :gutter="3" v-for="status in statusTable" :key="status.type">
+    <el-col :span="4">{{ taskTypeMap[status.type] }} </el-col>
+    <el-col :span="4">
+      <el-text v-if="status.state === State.SUCCESS || status.state === State.REVOKED">
+        <el-icon class="success"><SuccessFilled /></el-icon>
+        {{ stateMap[status.state](status.type) }}
+      </el-text>
+      <el-text v-else-if="status.state === State.FAILURE">
+        <el-icon class="danger"><CircleCloseFilled /></el-icon>
+        {{ stateMap[status.state](status.type) }}
+      </el-text>
+      <el-text v-else-if="status.state === State.STARTED">
+        <el-icon class="is-loading primary"><Loading /></el-icon>
+        {{ stateMap[status.state](status.type) }}
+      </el-text>
+      <el-text v-else-if="status.state === State.PENDING">
+        <el-icon class="is-loading primary"><Loading /></el-icon>
+        {{ stateMap[status.state](status.type) }}
+      </el-text>
+      <el-text v-else-if="status.state === State.REVOKE">
+        <el-icon class="is-loading primary"><Loading /></el-icon>
+        {{ stateMap[status.state](status.type) }}
+      </el-text>
+    </el-col>
+    <el-col :span="7">
+      完成
+      {{
+        Object.keys(status.aggs ? status.aggs : {})
+          .filter((k) => k == State.SUCCESS)
+          .map((k) => status.aggs[k])
+          .reduce((x: any, y: any) => x + y, 0)
+      }}/{{ Object.values(status.aggs ? status.aggs : {}).reduce((x: any, y: any) => x + y, 0) }}
+    </el-col>
+    <el-col :span="9">
+      {{
+        status.time
+          ? status.time[status.state == State.REVOKED ? State.REVOKED : State.PENDING]?.substring(
+              0,
+              19
+            )
+          : undefined
+      }}
+    </el-col>
+  </el-row>
+</template>
+<script setup lang="ts">
+import { computed } from 'vue'
+import { Status, TaskType, State, type TaskTypeInterface } from '@/utils/status'
+import { mergeWith } from 'lodash'
+const props = defineProps<{ status: string; statusMeta: any; stateMap: any; taskTypeMap: any }>()
+
+const parseAgg = (agg: { count: number; status: string }) => {
+  const status = new Status(agg.status)
+  return Object.keys(TaskType)
+    .map((key) => {
+      const value = TaskType[key as keyof TaskTypeInterface]
+      return { [value]: { [status.task_status[value]]: agg.count } }
+    })
+    .reduce((x, y) => ({ ...x, ...y }), {})
+}
+
+const customizer: (x: any, y: any) => any = (objValue: any, srcValue: any) => {
+  if (objValue == undefined && srcValue) {
+    return srcValue
+  }
+  if (srcValue == undefined && objValue) {
+    return objValue
+  }
+  // 如果是数组，我们将元素进行聚合
+  if (typeof objValue === 'object' && typeof srcValue === 'object') {
+    // 若是object类型的对象，我们进行递归
+    return mergeWith(objValue, srcValue, customizer)
+  } else {
+    // 否则，单纯的将值进行累加
+    return objValue + srcValue
+  }
+}
+const aggs = computed(() => {
+  return (props.statusMeta.aggs ? props.statusMeta.aggs : [])
+    .map((agg: any) => {
+      return parseAgg(agg)
+    })
+    .reduce((x: any, y: any) => {
+      return mergeWith(x, y, customizer)
+    }, {})
+})
+
+const statusTable = computed(() => {
+  return Object.keys(TaskType)
+    .map((key) => {
+      const value = TaskType[key as keyof TaskTypeInterface]
+      const parseStatus = new Status(props.status)
+      return {
+        type: value,
+        state: parseStatus.task_status[value],
+        aggs: aggs.value[value],
+        time: props.statusMeta.state_time[value]
+      }
+    })
+    .filter((item) => item.state !== State.IGNORED)
+})
+</script>
+<style lang="scss"></style>