Feat/2.4.0 beta1 (#1906)

Author: zgqgit

docker/docker-compose.yml

@@ -40,7 +40,7 @@ services:
 
   backend:
     container_name: bisheng-backend
-    image: dataelement/bisheng-backend:v2.3.0
+    image: dataelement/bisheng-backend:v2.4.0-beta1
     ports:
       - "7860:7860"
     environment:
@@ -78,7 +78,7 @@ services:
 
   backend_worker:
     container_name: bisheng-backend-worker
-    image: dataelement/bisheng-backend:v2.3.0
+    image: dataelement/bisheng-backend:v2.4.0-beta1
     environment:
       TZ: Asia/Shanghai
       BS_MILVUS_CONNECTION_ARGS: '{"host":"milvus","port":"19530","user":"","password":"","secure":false}'
@@ -109,7 +109,7 @@ services:
 
   frontend:
     container_name: bisheng-frontend
-    image: dataelement/bisheng-frontend:v2.3.0
+    image: dataelement/bisheng-frontend:v2.4.0-beta1
     ports:
       - "3001:3001"
     environment:

src/backend/bisheng/__init__.py

@@ -7,7 +7,7 @@
 
 try:
     # SetujuciGo to automatic modification
-    __version__ = '2.3.0'
+    __version__ = '2.4.0-beta1'
 except metadata.PackageNotFoundError:
     # Case where package metadata is not available.
     __version__ = ''

src/backend/bisheng/api/services/audit_log.py

@@ -35,7 +35,7 @@
 from bisheng.user.domain.models.user import UserDao
 from bisheng.utils import generate_uuid
 from bisheng.worker.workflow.redis_callback import RedisCallback
-from bisheng.worker.workflow.tasks import execute_workflow, continue_workflow
+from bisheng.worker.workflow.tasks import execute_workflow, continue_workflow, workflow_stateful_worker
 from bisheng.workflow.common.workflow import WorkflowStatus
 
 expire = 600
@@ -145,7 +145,7 @@ def upload_file(cls, file: UploadFile):
 
         file_ext = os.path.basename(file.filename).split('.')[-1]
         file_path = f'evaluation/dataset/{file_id}.{file_ext}'
-        minio_client.put_object_sync(bucket_name=minio_client.bucket, object_name=file_path, file=file.file.read(),
+        minio_client.put_object_sync(bucket_name=minio_client.bucket, object_name=file_path, file=file.file,
                                      content_type=file.content_type)
         return file_name, file_path
 
@@ -247,7 +247,10 @@ def execute_workflow_get_answer(workflow_info: FlowVersion, evaluation: Evaluati
     workflow = RedisCallback(unique_id, workflow_id, chat_id, user_id)
     workflow.set_workflow_data(workflow_info.data)
     workflow.set_workflow_status(WorkflowStatus.WAITING.value)
-    execute_workflow.delay(unique_id, workflow_id, chat_id, user_id)
+    hash_key = generate_uuid()
+    worker_node = workflow_stateful_worker.find_task_node_sync(hash_key)
+
+    execute_workflow.apply_async([unique_id, workflow_id, chat_id, user_id], queue=worker_node)
 
     # Listen for execution results of workflows
     input_event = None
@@ -265,7 +268,8 @@ def execute_workflow_get_answer(workflow_info: FlowVersion, evaluation: Evaluati
         # Only workflows entered in dialog boxes are entered by default
         workflow.set_user_input({input_event.message.get('node_id'): {"user_input": question}})
         workflow.set_workflow_status(WorkflowStatus.INPUT_OVER.value)
-        continue_workflow.delay(unique_id, workflow_id, chat_id, user_id)
+        worker_node = workflow_stateful_worker.find_task_node_sync(hash_key)
+        continue_workflow.apply_async([unique_id, workflow_id, chat_id, user_id], queue=worker_node)
         events = []
         for event in workflow.sync_get_response_until_break():
             events.append(event)

src/backend/bisheng/api/services/evaluation.py

@@ -204,7 +204,7 @@ async def update_version_info(cls, request: Request, user: UserPayload, version_
 
         flow_version = await FlowVersionDao.aupdate_version(version_info)
 
-        if flow_version.flow_type == FlowType.FLOW.value:
+        if flow_info.flow_type == FlowType.FLOW.value:
             try:
                 # Refresh this version of the form data
                 if not get_L2_param_from_flow(flow_version.data, flow_version.flow_id, flow_version.id):

src/backend/bisheng/api/services/flow.py

@@ -93,6 +93,9 @@ def unmerge_and_read_sheet(sheet_obj):
     for merged_range in merged_cell_ranges:
         min_col, min_row, max_col, max_row = merged_range.bounds
         top_left_cell_value = sheet_obj.cell(row=min_row, column=min_col).value
+        # ignore empty rows
+        if min_row > len(data_grid):
+            continue
         for r in range(min_row, max_row + 1):
             for c in range(min_col, max_col + 1):
                 data_grid[r - 1][c - 1] = top_left_cell_value
@@ -184,10 +187,12 @@ def process_dataframe_to_markdown_files(
     if append_header:
         # Handle header index outliers based on user rules
         if start_header_idx >= rows:
-            logger.warning(f"  Table Header Start Row {start_header_idx} Total lines exceeded {rows}. The first row will be used as the table header.")
+            logger.warning(
+                f"Table Header Start Row {start_header_idx} Total lines exceeded {rows}. The first row will be used as the table header.")
             start_header_idx, end_header_idx = 0, 0
         elif end_header_idx >= rows:
-            logger.warning(f"  Table Header End Row {end_header_idx} Total lines exceeded {rows}. will be truncated to the last line.")
+            logger.warning(
+                f"Table Header End Row {end_header_idx} Total lines exceeded {rows}. will be truncated to the last line.")
             end_header_idx = rows - 1
 
         # Make sure the index is legitimate

src/backend/bisheng/api/services/md_from_excel.py

@@ -374,22 +374,22 @@ async def get_group_dashboards(self, group_id: int, keyword: str, page_size: int
             return paged_res, len(res)
         return res, len(res)
 
-    def get_manage_resources(self, login_user: UserPayload, keyword: str, page: int, page_size: int) -> (list, int):
+    async def get_manage_resources(self, login_user: UserPayload, keyword: str, page: int, page_size: int) -> (list, int):
         """ Get a list of apps under a user group managed by a user Contains skills, assistants, workflows"""
         groups = []
         if not login_user.is_admin():
-            groups = [str(one.group_id) for one in UserGroupDao.get_user_admin_group(login_user.user_id)]
+            groups = [str(one.group_id) for one in await UserGroupDao.aget_user_admin_group(login_user.user_id)]
             if not groups:
                 return [], 0
 
         resource_ids = []
         # Description is a user group administrator, need to filter to get the resources under the corresponding group
         if groups:
-            group_resources = GroupResourceDao.get_groups_resource(groups, resource_types=[ResourceTypeEnum.FLOW,
+            group_resources = await GroupResourceDao.get_groups_resource(groups, resource_types=[ResourceTypeEnum.FLOW,
                                                                                            ResourceTypeEnum.ASSISTANT,
                                                                                            ResourceTypeEnum.WORK_FLOW])
             if not group_resources:
                 return [], 0
             resource_ids = [one.third_id for one in group_resources]
 
-        return FlowDao.get_all_apps(keyword, id_list=resource_ids, page=page, limit=page_size)
+        return await FlowDao.aget_all_apps(keyword, id_list=resource_ids, page=page, limit=page_size)

src/backend/bisheng/api/services/role_group_service.py

@@ -11,20 +11,20 @@
 
 
 @router.get('')
-def get_audit_logs(*,
-                   group_ids: Optional[List[str]] = Query(default=[], description='GroupingidVertical'),
-                   operator_ids: Optional[List[int]] = Query(default=[], description='WhoidVertical'),
-                   start_time: Optional[datetime] = Query(default=None, description='Start when'),
-                   end_time: Optional[datetime] = Query(default=None, description='End time'),
-                   system_id: Optional[str] = Query(default=None, description='Module Item'),
-                   event_type: Optional[str] = Query(default=None, description='Operation behaviors'),
-                   page: Optional[int] = Query(default=0, description='Page'),
-                   limit: Optional[int] = Query(default=0, description='Listings Per Page'),
-                   login_user: UserPayload = Depends(UserPayload.get_login_user)):
+async def get_audit_logs(*,
+                         group_ids: Optional[List[str]] = Query(default=[], description='GroupingidVertical'),
+                         operator_ids: Optional[List[int]] = Query(default=[], description='WhoidVertical'),
+                         start_time: Optional[datetime] = Query(default=None, description='Start when'),
+                         end_time: Optional[datetime] = Query(default=None, description='End time'),
+                         system_id: Optional[str] = Query(default=None, description='Module Item'),
+                         event_type: Optional[str] = Query(default=None, description='Operation behaviors'),
+                         page: Optional[int] = Query(default=0, description='Page'),
+                         limit: Optional[int] = Query(default=0, description='Listings Per Page'),
+                         login_user: UserPayload = Depends(UserPayload.get_login_user)):
     group_ids = [one for one in group_ids if one]
     operator_ids = [one for one in operator_ids if one]
-    return AuditLogService.get_audit_log(login_user, group_ids, operator_ids,
-                                         start_time, end_time, system_id, event_type, page, limit)
+    return await AuditLogService.get_audit_log(login_user, group_ids, operator_ids,
+                                               start_time, end_time, system_id, event_type, page, limit)
 
 
 @router.get('/operators')
@@ -36,56 +36,63 @@ def get_all_operators(*, login_user: UserPayload = Depends(UserPayload.get_login
 
 
 @router.get('/session')
-def get_session_list(login_user: UserPayload = Depends(UserPayload.get_login_user),
-                     flow_ids: Optional[List[str]] = Query(default=[], description='ApplicationsidVertical'),
-                     user_ids: Optional[List[int]] = Query(default=[], description='UsersidVertical'),
-                     group_ids: Optional[List[int]] = Query(default=[], description='User GroupsidVertical'),
-                     start_date: Optional[datetime] = Query(default=None, description='Start when'),
-                     end_date: Optional[datetime] = Query(default=None, description='End time'),
-                     feedback: Optional[str] = Query(default=None, description='like LikedislikeUnlikecopiedCopy:'),
-                     sensitive_status: Optional[int] = Query(default=None, description='Sensitive word review status'),
-                     page: Optional[int] = Query(default=1, description='Page'),
-                     page_size: Optional[int] = Query(default=10, description='Listings Per Page')):
+async def get_session_list(login_user: UserPayload = Depends(UserPayload.get_login_user),
+                           flow_ids: Optional[List[str]] = Query(default=[], description='ApplicationsidVertical'),
+                           user_ids: Optional[List[int]] = Query(default=[], description='UsersidVertical'),
+                           group_ids: Optional[List[int]] = Query(default=[], description='User GroupsidVertical'),
+                           start_date: Optional[datetime] = Query(default=None, description='Start when'),
+                           end_date: Optional[datetime] = Query(default=None, description='End time'),
+                           feedback: Optional[str] = Query(default=None,
+                                                           description='like LikedislikeUnlikecopiedCopy:'),
+                           sensitive_status: Optional[int] = Query(default=None,
+                                                                   description='Sensitive word review status'),
+                           page: Optional[int] = Query(default=1, description='Page'),
+                           page_size: Optional[int] = Query(default=10, description='Listings Per Page')):
     """ Filter all session lists """
-    data, total = AuditLogService.get_session_list(login_user, flow_ids, user_ids, group_ids, start_date, end_date,
-                                                   feedback, sensitive_status, page, page_size)
+    data, total = await AuditLogService.get_session_list(login_user, flow_ids, user_ids, group_ids, start_date,
+                                                         end_date,
+                                                         feedback, sensitive_status, page, page_size)
     return resp_200(data={
         'data': data,
         'total': total
     })
 
 
 @router.get('/session/export')
-def export_session_messages(login_user: UserPayload = Depends(UserPayload.get_login_user),
-                            flow_ids: Optional[List[str]] = Query(default=[], description='ApplicationsidVertical'),
-                            user_ids: Optional[List[int]] = Query(default=[], description='UsersidVertical'),
-                            group_ids: Optional[List[int]] = Query(default=[], description='User GroupsidVertical'),
-                            start_date: Optional[datetime] = Query(default=None, description='Start when'),
-                            end_date: Optional[datetime] = Query(default=None, description='End time'),
-                            feedback: Optional[str] = Query(default=None,
-                                                            description='like LikedislikeUnlikecopiedCopy:'),
-                            sensitive_status: Optional[int] = Query(default=None, description='Sensitive word review status')):
+async def export_session_messages(login_user: UserPayload = Depends(UserPayload.get_login_user),
+                                  flow_ids: Optional[List[str]] = Query(default=[],
+                                                                        description='ApplicationsidVertical'),
+                                  user_ids: Optional[List[int]] = Query(default=[], description='UsersidVertical'),
+                                  group_ids: Optional[List[int]] = Query(default=[],
+                                                                         description='User GroupsidVertical'),
+                                  start_date: Optional[datetime] = Query(default=None, description='Start when'),
+                                  end_date: Optional[datetime] = Query(default=None, description='End time'),
+                                  feedback: Optional[str] = Query(default=None,
+                                                                  description='like LikedislikeUnlikecopiedCopy:'),
+                                  sensitive_status: Optional[int] = Query(default=None,
+                                                                          description='Sensitive word review status')):
     """ Exporting a list of session detailscsvDoc. """
-    url = AuditLogService.export_session_messages(login_user, flow_ids, user_ids, group_ids, start_date, end_date,
-                                                  feedback, sensitive_status)
+    url = await AuditLogService.export_session_messages(login_user, flow_ids, user_ids, group_ids, start_date, end_date,
+                                                        feedback, sensitive_status)
     return resp_200(data={
         'url': url
     })
 
 
 @router.get('/session/export/data')
-def get_session_messages(login_user: UserPayload = Depends(UserPayload.get_login_user),
-                         flow_ids: Optional[List[str]] = Query(default=[], description='ApplicationsidVertical'),
-                         user_ids: Optional[List[int]] = Query(default=[], description='UsersidVertical'),
-                         group_ids: Optional[List[int]] = Query(default=[], description='User GroupsidVertical'),
-                         start_date: Optional[datetime] = Query(default=None, description='Start when'),
-                         end_date: Optional[datetime] = Query(default=None, description='End time'),
-                         feedback: Optional[str] = Query(default=None,
-                                                         description='like LikedislikeUnlikecopiedCopy:'),
-                         sensitive_status: Optional[int] = Query(default=None, description='Sensitive word review status')):
+async def get_session_messages(login_user: UserPayload = Depends(UserPayload.get_login_user),
+                               flow_ids: Optional[List[str]] = Query(default=[], description='ApplicationsidVertical'),
+                               user_ids: Optional[List[int]] = Query(default=[], description='UsersidVertical'),
+                               group_ids: Optional[List[int]] = Query(default=[], description='User GroupsidVertical'),
+                               start_date: Optional[datetime] = Query(default=None, description='Start when'),
+                               end_date: Optional[datetime] = Query(default=None, description='End time'),
+                               feedback: Optional[str] = Query(default=None,
+                                                               description='like LikedislikeUnlikecopiedCopy:'),
+                               sensitive_status: Optional[int] = Query(default=None,
+                                                                       description='Sensitive word review status')):
     """ Export data for a list of session details """
-    result = AuditLogService.get_session_messages(login_user, flow_ids, user_ids, group_ids, start_date, end_date,
-                                                  feedback, sensitive_status)
+    result = await AuditLogService.get_session_messages(login_user, flow_ids, user_ids, group_ids, start_date, end_date,
+                                                        feedback, sensitive_status)
     return resp_200(data={
         'data': result
     })