Merge pull request kubeagi#562 from wangxinbiao/main

task fails, update the 'message' information in the dataset for cr

apiserver/graph/schema/dataprocessing.graphqls

@@ -49,7 +49,6 @@ input AddDataProcessInput {
   post_data_set_name: String!
   post_data_set_version: String!
   data_process_config_info: [DataProcessConfigItem!]
-  bucket_name: String!
   version_data_set_name: String!
   namespace: String!
   creator: String!

data-processing/data_manipulation/controller/data_process_controller.py

@@ -54,13 +54,14 @@ async def add(request):
         "post_data_set_name": "dataset1",
         "post_data_set_version": "v2",
         "version_data_set_name": "dataset1-v2",
-        "bucket_name": "system-tce",
         "file_names": [
             {
                 "name": "数据处理文件_小T.pdf"
             }
         ],
-        "data_process_config_info": []
+        "data_process_config_info": [],
+        "creator": "",
+        "namespace": "abc"
     }    
     """
     res = data_process_service.add(

data-processing/data_manipulation/data_store_process/minio_store_process.py

@@ -52,7 +52,7 @@ def text_manipulate(
     req_json is a dictionary object. 
     """
 
-    bucket_name = req_json['bucket_name']
+    namespace = req_json['namespace']
     support_type = req_json['data_process_config_info']
     file_names = req_json['file_names']
 
@@ -72,9 +72,10 @@ def text_manipulate(
 
     # update the dataset status
     update_dataset = _update_dateset_status(
-        bucket_name=req_json['bucket_name'],
+        namespace=req_json['namespace'],
         version_data_set_name=req_json['version_data_set_name'],
         reason='processing',
+        message='Data processing in progress',
         task_id=id,
         log_id=log_id,
         creator=req_json.get('creator'),
@@ -130,7 +131,7 @@ def text_manipulate(
         # 将文件下载到本地
         minio_store_client.download(
             minio_client,
-            bucket_name=bucket_name,
+            bucket_name=namespace,
             folder_prefix=folder_prefix,
             file_name=file_name
         )
@@ -328,17 +329,18 @@ def text_manipulate(
     minio_store_client.upload_files_to_minio_with_tags(
         minio_client=minio_client,
         local_folder=file_path + 'final',
-        minio_bucket=bucket_name,
+        minio_bucket=namespace,
         minio_prefix=folder_prefix,
         support_type=support_type,
         data_volumes_file=data_volumes_file
     )
 
     # update the dataset status
     update_dataset = _update_dateset_status(
-        bucket_name=req_json['bucket_name'],
+        namespace=req_json['namespace'],
         version_data_set_name=req_json['version_data_set_name'],
         reason=task_status,
+        message=error_msg,
         task_id=id,
         log_id=log_id,
         creator=req_json.get('creator'),
@@ -417,9 +419,10 @@ def text_manipulate_retry(
 
         # 更新数据集状态
         update_dataset = _update_dateset_status(
-            bucket_name=task_info_dict.get('namespace'),
+            namespace=task_info_dict.get('namespace'),
             version_data_set_name=task_info_dict.get('pre_version_data_set_name'),
             reason='processing',
+            message='Data processing in progress',
             task_id=task_id,
             log_id=log_id,
             creator=creator,
@@ -515,17 +518,18 @@ def text_manipulate_retry(
             minio_store_client.upload_files_to_minio_with_tags(
                 minio_client=minio_client,
                 local_folder=file_path + 'final',
-                minio_bucket=task_info_dict.get('bucket_name'),
+                minio_bucket=task_info_dict.get('namespace'),
                 minio_prefix=folder_prefix,
                 support_type=task_info_dict.get('data_process_config_info'),
                 data_volumes_file=data_volumes_file
             )
 
         # 更新数据集状态
         update_dataset = _update_dateset_status(
-            bucket_name=task_info_dict.get('namespace'),
+            namespace=task_info_dict.get('namespace'),
             version_data_set_name=task_info_dict.get('pre_version_data_set_name'),
             reason=task_status,
+            message=error_msg,
             task_id=task_id,
             log_id=log_id,
             creator=creator,
@@ -597,9 +601,10 @@ def _remove_local_file(file_name):
         }
 
 def _update_dateset_status(
-    bucket_name,
+    namespace,
     version_data_set_name,
     reason,
+    message,
     task_id,
     log_id,
     creator,
@@ -608,21 +613,22 @@ def _update_dateset_status(
     logger.debug(''.join([
         f"{log_tag_const.MINIO_STORE_PROCESS} update dataset status \n",
         f"task_id: {task_id}\n",
-        f"bucket_name: {bucket_name}\n",
+        f"namespace: {namespace}\n",
         f"version_data_set_name: {version_data_set_name}\n",
         f"reason: {reason}"
     ]))
     update_dataset = dataset_cr.update_dataset_k8s_cr(
-        bucket_name=bucket_name,
+        namespace=namespace,
         version_data_set_name=version_data_set_name,
-        reason=reason
+        reason=reason,
+        message=message
     )
 
     if update_dataset['status'] != 200:
         logger.error(''.join([
             f"{log_tag_const.MINIO_STORE_PROCESS} update dataset status \n",
             f"task_id: {task_id}\n",
-            f"bucket_name: {bucket_name}\n",
+            f"namespace: {namespace}\n",
             f"version_data_set_name: {version_data_set_name}\n",
             f"reason: {reason}"
         ]))
@@ -1019,7 +1025,7 @@ def _text_manipulate_retry_for_document(
         # 将文件下载到本地
         minio_store_client.download(
             minio_client,
-            bucket_name=task_info.get('bucket_name'),
+            bucket_name=task_info.get('namespace'),
             folder_prefix=folder_prefix,
             file_name=file_name
         )

data-processing/data_manipulation/database_operate/data_process_db_operate.py

@@ -120,7 +120,6 @@ def add(
         'file_type': req_json['file_type'],
         'status': 'processing',
         'namespace': req_json['namespace'],
-        'bucket_name': req_json['bucket_name'],
         'pre_data_set_name': req_json['pre_data_set_name'],
         'pre_data_set_version': req_json['pre_data_set_version'],
         'pre_version_data_set_name': req_json['version_data_set_name'],
@@ -144,7 +143,6 @@ def add(
           file_type,
           status,
           namespace,
-          bucket_name,
           pre_data_set_name,
           pre_data_set_version,
           file_names,
@@ -166,7 +164,6 @@ def add(
           %(file_type)s,
           %(status)s,
           %(namespace)s,
-          %(bucket_name)s,
           %(pre_data_set_name)s,
           %(pre_data_set_version)s,
           %(file_names)s,
@@ -246,7 +243,6 @@ def info_by_id(
           dpt.data_process_config_info,
           dpt.start_datetime,
           dpt.end_datetime,
-          dpt.bucket_name,
           dpt.namespace,
           dpt.pre_version_data_set_name,
           dpt.create_user,

data-processing/data_manipulation/database_operate/data_process_detail_db_operate.py

@@ -457,16 +457,21 @@ def query_question_answer_list(
 
     sql = """
       select
-        id,
-        task_id,
-        document_id,
-        document_chunk_id,
-        file_name,
-        question,
-        answer
-      from public.data_process_task_question_answer_clean
+        dptqa.id,
+        dptqa.task_id,
+        dptqa.document_id,
+        dptqa.document_chunk_id,
+        dptqa.file_name,
+        dptqa.question,
+        dptqa.answer,
+        dptdc.content,
+        dptdc.page_number
+      from public.data_process_task_question_answer dptqa
+      left join public.data_process_task_document_chunk dptdc
+      on
+        dptdc.id = dptqa.document_chunk_id
       where 
-        document_id = %(document_id)s
+        dptqa.document_id = %(document_id)s
     """.strip()
 
     res = postgresql_pool_client.execute_query(pool, sql, params)

data-processing/data_manipulation/file_handle/common_handle.py

@@ -132,11 +132,14 @@ def text_manipulate(
             pool=conn_pool
         )
 
-        qa_data_dict = [['q', 'a']]
+        qa_data_dict = [['q', 'a', 'file_name', 'page_number', 'chunk_content']]
         for item in qa_list.get('data'):
             qa_data_dict.append([
                 item.get('question'),
-                item.get('answer')
+                item.get('answer'),
+                item.get('file_name'),
+                item.get('page_number'),
+                item.get('content')
             ])
 
         # Save the csv file.        

data-processing/data_manipulation/kube/dataset_cr.py

@@ -21,21 +21,22 @@
 logger = logging.getLogger(__name__)
 
 def update_dataset_k8s_cr(
-    bucket_name,
+    namespace,
     version_data_set_name,
-    reason
+    reason,
+    message
 ):
     """ Update the condition info for the dataset.
     
-    bucket_name: bucket name;
+    namespace: namespace;
     version_data_set_name: version dataset name;
     reason: the update reason;
     """
     try:
         kube = client.KubeEnv()
 
         one_cr_datasets = kube.get_versioneddatasets_status(
-                                bucket_name, 
+                                namespace, 
                                 version_data_set_name
                             )
 
@@ -56,18 +57,20 @@ def update_dataset_k8s_cr(
                 'lastTransitionTime': now_utc_str,
                 'reason': reason,
                 'status': "True",
-                "type": "DataProcessing"
+                "type": "DataProcessing",
+                "message": message
             })
         else:
             conditions[found_index] = {
                 'lastTransitionTime': now_utc_str,
                 'reason': reason,
                 'status': "True",
-                "type": "DataProcessing"
+                "type": "DataProcessing",
+                "message": message
             }
 
         kube.patch_versioneddatasets_status(
-            bucket_name, 
+            namespace, 
             version_data_set_name,
             {
                 'status': {
@@ -90,20 +93,20 @@ def update_dataset_k8s_cr(
         }
 
 def get_dataset_status_k8s_cr(
-    bucket_name,
+    namespace,
     version_data_set_name
 ):
     """ get the condition info for the dataset.
     
-    bucket_name: bucket name;
+    namespace: namespace;
     version_data_set_name: version dataset name;
     """
     try:
         dataset_status = None
         kube = client.KubeEnv()
 
         one_cr_datasets = kube.get_versioneddatasets_status(
-                                bucket_name, 
+                                namespace, 
                                 version_data_set_name
                             )
 

data-processing/data_manipulation/service/data_process_service.py

@@ -65,7 +65,7 @@ def add(
         "post_data_set_name": "dataset1",
         "post_data_set_version": "v2",
         "version_data_set_name": "dataset1-v2",
-        "bucket_name": "system-tce",
+        "namespace": "system-tce",
         "file_names": [
             {
                 "name": "数据处理文件_小T.pdf"

data-processing/db-scripts/init-database-schema.sql

@@ -25,12 +25,10 @@
         update_user character varying(32) COLLATE pg_catalog."default",
         update_program character varying(64) COLLATE pg_catalog."default",
         namespace character varying(64) COLLATE pg_catalog."default",
-        bucket_name character varying(64) COLLATE pg_catalog."default",
         current_log_id character varying(32) COLLATE pg_catalog."default",
         CONSTRAINT data_process_task_pkey PRIMARY KEY (id)
     );
 
-    COMMENT ON COLUMN public.data_process_task.bucket_name IS 'bucket name';
     COMMENT ON COLUMN public.data_process_task.current_log_id IS '当前日志Id';
     COMMENT ON COLUMN public.data_process_task.pre_version_data_set_name IS '处理前数据集版本信息';
 

deploy/charts/arcadia/Chart.yaml

@@ -2,7 +2,7 @@ apiVersion: v2
 name: arcadia
 description: A Helm chart(KubeBB Component) for KubeAGI Arcadia
 type: application
-version: 0.2.10
+version: 0.2.11
 appVersion: "0.1.0"
 
 keywords:

deploy/charts/arcadia/templates/pg-init-data-configmap.yaml

@@ -29,12 +29,10 @@ data:
             update_user character varying(32) COLLATE pg_catalog."default",
             update_program character varying(64) COLLATE pg_catalog."default",
             namespace character varying(64) COLLATE pg_catalog."default",
-            bucket_name character varying(64) COLLATE pg_catalog."default",
             current_log_id character varying(32) COLLATE pg_catalog."default",
             CONSTRAINT data_process_task_pkey PRIMARY KEY (id)
         );
 
-        COMMENT ON COLUMN public.data_process_task.bucket_name IS 'bucket name';
         COMMENT ON COLUMN public.data_process_task.current_log_id IS '当前日志Id';
         COMMENT ON COLUMN public.data_process_task.pre_version_data_set_name IS '处理前数据集版本信息';