Python large data volume of text files efficient parsing scheme code implementation of the whole process

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import re
import time
from datetime import datetime
from joblib import Parallel, delayed, parallel_backend
from collections import deque
from multiprocessing import cpu_count
import threading


class LogParser(object):
    def __init__(self, chunk_size=1024*1024*10, process_num_for_log_parsing=cpu_count()):
        self.log_unparsed_queue = deque() # Used to store unresolved logs
        self.log_line_parsed_queue = deque()  # Used to store parsed log lines
        self.is_all_files_read = False  # Identifies whether all log files have been read
        self.process_num_for_log_parsing = process_num_for_log_parsing # of processes concurrently parsing log files
        self.chunk_size = chunk_size # Log block size per log read
        self.files_read_list = [] # Store read log files
        self.log_parsing_finished = False # Identifies whether log parsing is complete


    def read_in_chunks(self, filePath, chunk_size=1024*1024):
        """
        Inert function (generator) for reading files block by block.
        Default block size: 1M
        """

        with open(filePath, 'r', encoding='utf-8') as f:            
            while True:
                chunk_data = (chunk_size)
                if not chunk_data:
                    break
                yield chunk_data


    def read_log_file(self, logfile_path):
        '''
        Reading log files
        It is assumed here that the log files are all text files, and after reading them in chunks, they can be sliced and diced twice by line breaks in order to get the line logs
        '''

        temp_list = []  # After the second cut, the header and footer logs may be incomplete, so you need to connect the header and footer logs of the log block and splice them together.
        for chunk in self.read_in_chunks(logfile_path, self.chunk_size):
            log_chunk = ('\n')
            temp_list.extend([log_chunk[0], '\n'])
            temp_list.append(log_chunk[-1])
            self.log_unparsed_queue.append(log_chunk[1:-1])
        self.log_unparsed_queue.append(''.join(temp_list).split('\n'))
        self.files_read_list.remove(logfile_path)


    def start_processes_for_log_parsing(self):
        '''Start log parsing process'''

        with parallel_backend("multiprocessing", n_jobs=self.process_num_for_log_parsing):
            Parallel(require='sharedmem')(delayed(self.parse_logs)() for i in range(self.process_num_for_log_parsing))

        self.log_parsing_finished = True

    def parse_logs(self):
        '''Parsing Logs'''

        method_url_re_pattern = ('(HEAD|POST|GET)\s+([^\s]+?)\s+',)
        url_time_taken_extractor = ('HTTP/1\.1.+\|(.+)\|\d+\|', )

        while self.log_unparsed_queue or self.files_read_list:
            if not self.log_unparsed_queue:
                continue
            log_line_list = self.log_unparsed_queue.popleft()
            for log_line in log_line_list:
                #### do something with log_line
                if not log_line.strip():
                    continue

                res = method_url_re_pattern.findall(log_line)
                if not res:
                    print('The log did not match the request URL and has been ignored:\n%s' % log_line)
                    continue
                method = res[0][0]
                url = res[0][1].split('?')[0]  # Removed ? and the following url argument

                # Extraction time consuming
                res = url_time_taken_extractor.findall(log_line)
                if res:
                    time_taken = float(res[0])
                else:
                    print('Request elapsed time not extracted from logs, logs have been ignored:\n%s' % log_line)
                    continue

                # Store parsed log messages
                self.log_line_parsed_queue.append({'method': method,
                                                   'url': url,
                                                   'time_taken': time_taken,
                                                   })


    def collect_statistics(self):
        '''Collecting statistical data'''

        def _collect_statistics():
            while self.log_line_parsed_queue or not self.log_parsing_finished:
                if not self.log_line_parsed_queue:
                    continue
                log_info = self.log_line_parsed_queue.popleft()
                # do something with log_info
       
        with parallel_backend("multiprocessing", n_jobs=1):
            Parallel()(delayed(_collect_statistics)() for i in range(1))

    def run(self, file_path_list):
        # Multi-threaded reading of log files
        for file_path in file_path_list:
            thread = (target=self.read_log_file,
                                      name="read_log_file",
                                      args=(file_path,))
            ()
            self.files_read_list.append(file_path)

        # Start the log parsing process
        thread = (target=self.start_processes_for_log_parsing, name="start_processes_for_log_parsing")
        ()

        # Initiate the log statistics collection process
        thread = (target=self.collect_statistics, name="collect_statistics")
        ()

        start = ()
        while threading.active_count() > 1:
            print('The program is trying to parse the logs...')
            (0.5)

        end = ()
        print('Parsing complete', 'start', start, 'end', end, 'Time-consuming', end - start)

if __name__ == "__main__":
    log_parser = LogParser()
    log_parser.run(['', ''])