import logging
import os

# 獲取當(dāng)前文件的絕對(duì)路徑
current_file = os.path.abspath(__file__)

# 配置日志格式
log_format = '%(asctime)s - %(levelname)s - %(pathname)s:%(lineno)d - %(message)s'
logging.basicConfig(format=log_format, level=logging.INFO)

# 創(chuàng)建一個(gè)文件處理器，并將日志輸出到文件
file_handler = logging.FileHandler('app.log')
file_handler.setFormatter(logging.Formatter(log_format))
logging.getLogger().addHandler(file_handler)

import asyncio
import random

DEFAULT_MAX_CONCURRENT_TASKS = 2  # 最大并發(fā)任務(wù)數(shù)
MAX_RETRIES = 3  # 最大重試次數(shù)

class AsyncBatchProcessor:
    def __init__(self, max_concurrent: int = DEFAULT_MAX_CONCURRENT_TASKS):
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def process_single_file(
            self,
            input_file: str,
            retry_count: int = 0
    ) -> None:
        """處理單個(gè)文件的異步方法"""
        async with self.semaphore:  # 使用信號(hào)量控制并發(fā)
            try:
                logging.info(f"Processing file: {input_file}")

                # 模擬文件處理過(guò)程
                await asyncio.sleep(random.uniform(0.5, 2.0))

                logging.info(f"Successfully processed {input_file}")

            except Exception as e:
                logging.error(f"Error processing {input_file} of Attempt {retry_count}: {str(e)}")
                if retry_count < MAX_RETRIES:
                    logging.info(f"Retrying {input_file} (Attempt {retry_count + 1})")
                    await asyncio.sleep(1)
                    await self.process_single_file(input_file, retry_count + 1)
                else:
                    logging.error(f"Failed to process {input_file} after {MAX_RETRIES} attempts")

    async def process_batch(
            self,
            file_list: list
    ) -> None:
        total_files = len(file_list)
        logging.info(f"Found {total_files} files to process")

        # 創(chuàng)建工作隊(duì)列
        queue = asyncio.Queue()

        # 將所有文件放入隊(duì)列
        for file_path in file_list:
            await queue.put(file_path)

        # 創(chuàng)建工作協(xié)程
        async def worker(worker_id: int):
            while True:
                try:
                    # 非阻塞方式獲取任務(wù)
                    input_file_path = await queue.get()
                    logging.info(f"Worker {worker_id} processing: {input_file_path}")

                    try:
                        await self.process_single_file(input_file_path)
                    except Exception as e:
                        logging.error(f"Error processing {input_file_path}: {str(e)}")
                    finally:
                        queue.task_done()

                except asyncio.QueueEmpty:
                    # 隊(duì)列為空，工作結(jié)束
                    break
                except Exception as e:
                    logging.error(f"Worker {worker_id} encountered error: {str(e)}")
                    break

        # 創(chuàng)建工作任務(wù)
        workers = []
        for i in range(self.max_concurrent):
            worker_task = asyncio.create_task(worker(i))
            workers.append(worker_task)

        # 等待隊(duì)列處理完成
        await queue.join()

        # 取消所有仍在運(yùn)行的工作任務(wù)
        for w in workers:
            w.cancel()

        # 等待所有工作任務(wù)完成
        await asyncio.gather(*workers, return_exceptions=True)

async def batch_detect(
        file_list: list,
        max_concurrent: int = DEFAULT_MAX_CONCURRENT_TASKS
):
    """異步批處理入口函數(shù)"""
    processor = AsyncBatchProcessor(max_concurrent)
    await processor.process_batch(file_list)

# 示例調(diào)用
file_list = ["file1.pdf", "file2.pdf", "file3.pdf", "file4.pdf"]
asyncio.run(batch_detect(file_list))