zhang_crawl_bio/ui/backend/tests/mytests/t_process_mgr.py

import asyncio
from pathlib import Path
import sys
import time
sys.path.append(str(Path(r'G:\code\upwork\zhang_crawl_bio\ui\backend')))
from utils.process_mgr import ProcessManager
from utils.logu import get_logger
logger = get_logger('mytests', file=True)
async def main():
    # 单例实例
    process_manager = ProcessManager()
    bin_path = r'G:\code\upwork\zhang_crawl_bio\download\proxy_pool\mihomo-windows-amd64-go120.exe'
    file1 = r'G:\code\upwork\zhang_crawl_bio\download\proxy_pool\yfjc_9361.yaml'
    file2 = r'G:\code\upwork\zhang_crawl_bio\download\proxy_pool\yfjc_9363.yaml'
    file3 = r'G:\code\upwork\zhang_crawl_bio\download\proxy_pool\yfjc_9365.yaml'
    file4 = r'G:\code\upwork\zhang_crawl_bio\download\proxy_pool\yfjc_9367.yaml'
    file5 = r'G:\code\upwork\zhang_crawl_bio\download\proxy_pool\yfjc_9369.yaml'
    list_files = [file1,file2,file3,file4,file5]
    for file in list_files:
        pid = await process_manager.start_process(f"mimo_{Path(file).stem}", [bin_path, '-f' ,file], )
        logger.info(f"start {pid}, file {file}")

    logger.info(f"processes {process_manager.processes}")
    try:
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        logger.info("KeyboardInterrupt")
        success = await process_manager.cleanup()

if __name__ == "__main__":
    asyncio.run(main())