change parent_id to log_group_id for user activity logs

install.sh

@@ -15,6 +15,8 @@ docker cp scripts/demo/defaultworkflow.sql $(docker-compose -f docker-compose2.y
 docker-compose -f docker-compose2.yml exec postgresql psql -U invenio -d invenio -f /tmp/defaultworkflow.sql
 docker cp scripts/demo/doi_identifier.sql $(docker-compose -f docker-compose2.yml ps -q postgresql):/tmp/doi_identifier.sql
 docker-compose -f docker-compose2.yml exec postgresql psql -U invenio -d invenio -f /tmp/doi_identifier.sql
+docker cp postgresql/ddl/W-OA-user_activity_log.sql $(docker-compose -f docker-compose2.yml ps -q postgresql):/tmp/W-OA-user_activity_log.sql
+docker-compose -f docker-compose2.yml exec postgresql psql -U invenio -d invenio -f /tmp/W-OA-user_activity_log.sql
 
 docker-compose -f docker-compose2.yml run --rm web invenio assets build
 docker-compose -f docker-compose2.yml run --rm web invenio collect -v

modules/weko-authors/weko_authors/admin.py

@@ -425,6 +425,12 @@ def import_authors(self) -> jsonify:
         group_tasks = []
         count=0
 
+        # get the request info for logging
+        request_info = UserActivityLogger.get_summary_from_request()
+        if not UserActivityLogger.get_log_group_id(request_info):
+            UserActivityLogger.issue_log_group_id(None)
+        request_info["log_group_id"] = UserActivityLogger.get_log_group_id(request_info)
+
         if is_target == "id_prefix":
             for id_prefix in records:
                 group_tasks.append(import_id_prefix.s(id_prefix))
@@ -476,7 +482,6 @@ def import_authors(self) -> jsonify:
             records, reached_point, count = prepare_import_data(max_page_for_import_tab)
             task_ids =[]
 
-            request_info = UserActivityLogger.get_summary_from_request()
             for author in records:
                 group_tasks.append(import_author.s(author, force_change_mode, request_info))
         else:
@@ -513,7 +518,9 @@ def import_authors(self) -> jsonify:
                     force_change_mode,
                     current_app.config.get("WEKO_AUTHORS_CACHE_TTL")
                 )
-                task = import_author_over_max.delay(reached_point ,task_ids, max_page_for_import_tab)
+                task = import_author_over_max.delay(
+                    reached_point, task_ids, max_page_for_import_tab,
+                    request_info=request_info)
                 update_cache_data(
                     current_app.config.get("WEKO_AUTHORS_IMPORT_CACHE_OVER_MAX_TASK_KEY"),
                     task.id,

modules/weko-authors/weko_authors/tasks.py

@@ -141,7 +141,7 @@ def import_affiliation_id(affiliation_id):
     return result
 
 @shared_task
-def import_author_over_max(reached_point, task_ids ,max_part):
+def import_author_over_max(reached_point, task_ids ,max_part, request_info=None):
     """
     WEKO_AUTHORS_IMPORT_MAX_NUM_OF_DISPLAYSを超えた著者をインポートする場合の処理です。
     先に行っているインポートタスクが終了次第、reached_pointから一時ファイルを用いて
@@ -152,6 +152,7 @@ def import_author_over_max(reached_point, task_ids ,max_part):
                 データ例:{"part_number": 101, "count": 3}
         task_ids: 先に行っているmax_diplay分のタスクID.
         max_part: パート数の最大値
+        request_info: リクエスト情報
     """
 
     # task_idsの全てのtaskが終了するまで待つ
@@ -162,7 +163,7 @@ def import_author_over_max(reached_point, task_ids ,max_part):
     current_app.logger.info('import_author_over_max is start')
     result = {'start_date': datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
     try:
-        import_authors_from_temp_files(reached_point, max_part)
+        import_authors_from_temp_files(reached_point, max_part, request_info=request_info)
         result['status'] = states.SUCCESS
     except Exception as ex:
         current_app.logger.error(ex)
@@ -178,14 +179,15 @@ def import_author_over_max(reached_point, task_ids ,max_part):
     return result
 
 
-def import_authors_from_temp_files(reached_point, max_part):
+def import_authors_from_temp_files(reached_point, max_part, request_info=None):
     """
     一時ファイルから著者データを読み込み、インポートする処理を行います。
     Args:
         reached_point: 一時ファイルにおいてmax_displayに達した位置
                 part_numberが一時ファイルのpart数で、countが一時ファイルの再開位置
                 データ例:{"part_number": 101, "count": 3}
         max_part: インポートする最大のpart数
+        request_info: リクエスト情報
     """
 
     # 結果ファイルのDL用に一時ファイルを作成
@@ -225,7 +227,7 @@ def import_authors_from_temp_files(reached_point, max_part):
                         authors.append(item)
         # authorsが長さWEKO_AUTHORS_IMPORT_BATCH_SIZEを超えた時点でインポート
         if len(authors) >= current_app.config.get("WEKO_AUTHORS_IMPORT_BATCH_SIZE"):
-            import_authors_for_over_max(authors)
+            import_authors_for_over_max(authors, request_info=request_info)
             authors = []
 
         # 一時ファイルの削除
@@ -237,20 +239,24 @@ def import_authors_from_temp_files(reached_point, max_part):
             current_app.logger.error(f"Error deleting {check_file_part_path}: {e}")
     # authorsが残っている場合
     if authors:
-        import_authors_for_over_max(authors)
+        import_authors_for_over_max(authors, request_info=request_info)
         authors = []
         gc.collect()
 
-def import_authors_for_over_max(authors):
+def import_authors_for_over_max(authors, request_info=None):
     group_tasks = []
     tasks = []
     task_ids = []
     force_change_mode = current_cache.get(\
         current_app.config.get("WEKO_AUTHORS_IMPORT_CACHE_FORCE_CHANGE_MODE_KEY", False)
         )
-    request_info = UserActivityLogger.get_summary_from_request()
+    _request_info = request_info or UserActivityLogger.get_summary_from_request() 
+    if not UserActivityLogger.get_log_group_id(request_info):
+        UserActivityLogger.issue_log_group_id(None)
+    _request_info["log_group_id"] = UserActivityLogger.get_log_group_id(request_info)
+
     for author in authors:
-        group_tasks.append(import_author.s(author, force_change_mode, request_info))
+        group_tasks.append(import_author.s(author, force_change_mode, _request_info))
 
     # group_tasksを実行
     import_task = group(group_tasks).apply_async()

modules/weko-itemtypes-ui/weko_itemtypes_ui/admin.py

@@ -280,7 +280,6 @@ def register(self, item_type_id=0):
             )
 
             db.session.commit()
-            next_id = UserActivityLogger.get_next_parent_id(db.session)
             if item_type_id == 0:
                 UserActivityLogger.info(
                     operation="ITEM_TYPE_CREATE",
@@ -295,20 +294,17 @@ def register(self, item_type_id=0):
             if not upgrade_version or item_type_id != record.model.id:
                 UserActivityLogger.info(
                     operation="ITEM_TYPE_MAPING_CREATE",
-                    parent_id=next_id,
                     target_key=record.model.id
                 )
             else:
                 UserActivityLogger.info(
                     operation="ITEM_TYPE_MAPING_UPDATE",
-                    parent_id=next_id,
                     target_key=item_type_id
                 )
                 workflow_list = WorkFlow().get_workflow_by_itemtype_id(item_type_id)
                 for wf in workflow_list:
                     UserActivityLogger.info(
                         operation="WORKFLOW_UPDATE",
-                        parent_id=next_id,
                         target_key=wf.id
                     )
         except Exception as ex:
@@ -482,6 +478,11 @@ def item_type_import(self):
             current_app.logger.debug(input_file.mimetype)
             return jsonify(msg=_('Illegal mimetype Error'))
 
+        # Issue log group ID
+        if not UserActivityLogger.issue_log_group_id(db.session):
+            current_app.logger.error('Failed to issue log group ID.')
+            return jsonify(msg=_('Failed to issue log group ID.'))
+
         try:
             readable_files = ["ItemType.json", "ItemTypeName.json", "ItemTypeMapping.json", "ItemTypeProperty.json"]
             import_data = {
@@ -545,15 +546,13 @@ def item_type_import(self):
             )
 
             db.session.commit()
-            next_id = UserActivityLogger.get_next_parent_id(db.session)
             UserActivityLogger.info(
                 operation="ITEM_TYPE_CREATE",
                 target_key=item_type_id
             )
             # log item type mapping and workflow
             UserActivityLogger.info(
                 operation="ITEM_TYPE_MAPING_CREATE",
-                parent_id=next_id,
                 target_key=item_type_id
             )
         except Exception as ex:

modules/weko-logging/weko_logging/activity_logger.py

@@ -7,7 +7,7 @@
 # under the terms of the MIT License; see LICENSE file for more details.
 """Weko logging user activity logger wrapper."""
 
-from flask import current_app
+from flask import current_app, g, has_request_context
 from werkzeug.local import LocalProxy
 
 from weko_logging.models import UserActivityLog
@@ -27,24 +27,25 @@ def __init__(self, app):
         self.app = app
 
     @classmethod
-    def error(cls, operation=None, parent_id=None,
-              target_key=None, request_info=None, remarks=None):
+    def error(cls, operation=None, target_key=None,
+              request_info=None, remarks=None):
         """Output as error log.
 
         Args:
             operation (str): User operation type.
-            parent_id (int): Parent log id.
             target_key (str): Operation target key (e.g. id).
             request_info (dict): Request information (Required if called by shared task).
             remarks (str): Remarks.
         """
         user_id = UserActivityLogHandler.get_user_id()
         community_id = UserActivityLogHandler.get_community_id_from_path(request_info)
 
-        error_message = f"Error occurred: operation={operation}, parent_id={parent_id}, target_key={target_key}, "
+        log_group_id = cls.get_log_group_id(request_info)
+
+        error_message = f"Error occurred: operation={operation}, log_group_id={log_group_id}, target_key={target_key}, "
         error_message += f"user_id={user_id}, community_id={community_id}"
         _logger.error(error_message, extra={
-            "parent_id": parent_id,
+            "log_group_id": log_group_id,
             "operation": operation,
             "target_key": target_key,
             "request_info": request_info,
@@ -54,14 +55,12 @@ def error(cls, operation=None, parent_id=None,
         })
 
     @classmethod
-    def info(cls, operation=None, parent_id=None,
-             target_key=None, request_info=None, 
+    def info(cls, operation=None, target_key=None, request_info=None, 
              required_commit=True, remarks=None):
         """Output as info log.
 
         Args:
             operation (str): User operation type.
-            parent_id (int): Parent log id.
             target_key (str): Operation target key (e.g. id).
             request_info (dict): Request information (Required if called by shared task).
             required_commit (bool): Whether to commit the log.
@@ -70,30 +69,59 @@ def info(cls, operation=None, parent_id=None,
         user_id = UserActivityLogHandler.get_user_id()
         community_id = UserActivityLogHandler.get_community_id_from_path(request_info)
 
-        error_message = f"Info: operation={operation}, parent_id={parent_id}, target_key={target_key}, "
-        error_message += f"user_id={user_id}, community_id={community_id}"
+        log_group_id = cls.get_log_group_id(request_info)
+        # If log_group_id is None, issue a new one
+        if log_group_id is None:
+            cls.issue_log_group_id(None)
+            log_group_id = cls.get_log_group_id(request_info)
+
+        message = f"Info: operation={operation}, log_group_id={log_group_id}, target_key={target_key}, "
+        message += f"user_id={user_id}, community_id={community_id}"
 
-        _logger.info(error_message, extra={
-            "parent_id": parent_id,
+        _logger.info(message, extra={
+            "log_group_id": log_group_id,
             "operation": operation,
             "target_key": target_key,
             "request_info": request_info,
             "community_id": community_id,
             "required_commit": required_commit,
             "remarks": remarks,
         })
+
+    @classmethod
+    def get_log_group_id(cls, request_info):
+        """Get log group id
+
+        Args:
+            request_info (dict): The request information.
+
+        Returns:
+            int: The log group ID.
+        """
+        log_group_id = None
+        if request_info and "log_group_id" in request_info:
+            log_group_id = request_info["log_group_id"]
+        elif has_request_context() and hasattr(g, "log_group_id"):
+            log_group_id = g.log_group_id
+
+        return log_group_id
 
     @classmethod
-    def get_next_parent_id(cls, session):
-        """Get the next parent ID.
+    def issue_log_group_id(cls, session):
+        """Issue a new log group ID and store request context.
 
         Args:
             session: The database session.
 
         Returns:
-            int: The next log ID.
+            bool: True if the log group ID was successfully issued, False otherwise.
         """
-        return UserActivityLog.get_sequence(session)
+        if not has_request_context():
+            return False
+
+        log_group_id = UserActivityLog.get_log_group_sequence(session)
+        g.log_group_id = log_group_id
+        return True
 
     @classmethod
     def get_summary_from_request(cls):

modules/weko-logging/weko_logging/handler.py

@@ -53,10 +53,10 @@ def emit(self, record):
             current_app.logger.error(f"Invalid operation: {operation}")
             raise ValueError(f"Invalid operation: {operation}")
 
-        # get log group uuid
-        parent_id = None
-        if hasattr(record, "parent_id"):
-            parent_id = record.parent_id or None
+        # get log group id
+        log_group_id = None
+        if hasattr(record, "log_group_id"):
+            log_group_id = record.log_group_id or None
 
         # get user_id from current_user
         user_id = self.get_user_id()
@@ -109,7 +109,7 @@ def emit(self, record):
         user_activity_log = UserActivityLog(
             user_id=user_id,
             log={},
-            parent_id=parent_id,
+            log_group_id=log_group_id,
             community_id=community_id,
             remarks=remarks,
         )
@@ -123,7 +123,7 @@ def emit(self, record):
             "client_id": client_id,
             "community_id": community_id or "",
             "source": source,
-            "parent_id": parent_id,
+            "log_group_id": log_group_id,
             "operation_type_id": operation_type_id,
             "operation_id": operation_id,
             "target": target,
@@ -164,7 +164,7 @@ def get_community_id_from_path(cls, request_info):
             return None
         community_id = None
         path_info = urllib.parse.urlparse(request_path)
-        if "/c/" in path_info.path:
+        if "/c/" in str(path_info.path):
             community_path = path_info.path.split("/c/")[1]
             community_id = community_path.split("/")[0]
         elif "community" in request_args: