* make log summarizations after OPs are done.

data_juicer/core/data.py

@@ -21,6 +21,7 @@
                                         cleanup_compressed_cache_files,
                                         compress, decompress)
 from data_juicer.utils.fingerprint_utils import generate_fingerprint
+from data_juicer.utils.logger_utils import make_log_summarization
 from data_juicer.utils.process_utils import setup_mp
 
 
@@ -258,6 +259,9 @@ def process(
             if work_dir and enable_insight_mining:
                 logger.info('Insight mining for each OP...')
                 adapter.insight_mining()
+            # make summarization on the error/warning logs
+            if work_dir:
+                make_log_summarization()
         return dataset
 
     def update_args(self, args, kargs, is_filter=False):

data_juicer/utils/logger_utils.py

@@ -176,6 +176,47 @@ def setup_logger(save_dir,
     LOGGER_SETUP = True
 
 
+def make_log_summarization(max_show_item=10):
+    error_pattern = r'^An error occurred in mapper operation when ' \
+                    r'processing samples? \"(.*?)\" -- (.*?): (.*?)$'
+    log_file = get_log_file_path()
+    error_log_file = add_suffix_to_filename(log_file, '_ERROR')
+    warning_log_file = add_suffix_to_filename(log_file, '_WARNING')
+
+    import jsonlines as jl
+    import regex as re
+
+    # make error summarization
+    error_dict = {}
+    with jl.open(error_log_file) as reader:
+        for error_log in reader:
+            error_msg = error_log['record']['message']
+            find_res = re.findall(error_pattern, error_msg)
+            if len(find_res) > 0:
+                sample, error_type, error_msg = find_res[0]
+                error = f'{error_type} -- \'{error_msg}\''
+                error_dict.setdefault(error, 0)
+                error_dict[error] += 1
+    total_error_count = sum(error_dict.values())
+    # make warning summarization
+    warning_count = 0
+    with jl.open(warning_log_file) as reader:
+        for _ in reader:
+            warning_count += 1
+    # make summary log
+    summary = f'Processing finished with:\n' \
+              f'<yellow>Warnings</yellow>: {warning_count}\n' \
+              f'<red>Errors</red>: {total_error_count}\n'
+    error_items = list(error_dict.items())
+    error_items.sort(key=lambda it: it[1], reverse=True)
+    error_items = error_items[:max_show_item]
+    for key, num in error_items:
+        summary += f'- {key}: {num}\n'
+    summary += f'Error/Warning details can be found in the log file ' \
+               f'[{log_file}] and its related log files.'
+    logger.opt(ansi=True).info(summary)
+
+
 class HiddenPrints:
     """Define a range that hide the outputs within this range."""