如何用多进程解析文本提取URL、多线程检测可用性并修正代码

import concurrent.futures
import pickle
import requests
from urlextract import URLExtract

# 配置参数
CONNECTIONS = 100
TIMEOUT = 5
PROCESS_COUNT = 4 # 可根据自身CPU核心数调整

# 初始化URL提取器
extractor = URLExtract()

def extract_urls(text: str) -> list:
    """单进程URL提取逻辑，供多进程调度调用"""
    return list(extractor.gen_urls(text))

def check_url_availability(url: str, timeout: int) -> tuple[str, int|str]:
    """检测单个URL可用性，返回URL和对应状态码/异常信息"""
    try:
        # 增加allow_redirects=True自动跟随重定向，不需要可以删除
        resp = requests.head(url, timeout=timeout, allow_redirects=True)
        return url, resp.status_code
    except Exception as exc:
        return url, type(exc).__name__

if __name__ == "__main__":
    # 1. 加载原始数据
    with open('messages_to_parse.dat', 'rb') as f:
        raw_data = pickle.load(f, encoding='latin1')
    
    # 2. 多进程提取URL，自动适配原始数据格式
    if isinstance(raw_data, str):
        # 整段文本拆分为等长段落分配给多进程
        chunk_size = len(raw_data) // PROCESS_COUNT + 1
        text_chunks = [raw_data[i:i+chunk_size] for i in range(0, len(raw_data), chunk_size)]
    else:
        # 若原始数据为消息列表，直接把单条消息作为处理单元
        text_chunks = [str(msg) for msg in raw_data]
    
    all_urls = set()
    with concurrent.futures.ProcessPoolExecutor(max_workers=PROCESS_COUNT) as process_executor:
        for result in process_executor.map(extract_urls, text_chunks):
            for url in result:
                all_urls.add(url)
    urls = list(all_urls)
    print(f"共提取到{len(urls)}个去重后的URL")

    # 3. 多线程检测URL可用性
    url_status = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=CONNECTIONS) as thread_executor:
        futures = [thread_executor.submit(check_url_availability, url, TIMEOUT) for url in urls]
        for idx, future in enumerate(concurrent.futures.as_completed(futures), 1):
            url, status = future.result()
            url_status[url] = status
            print(f"检测进度：{idx}/{len(urls)}", end="\r")

    # 4. 输出最终结果字典
    print("\n检测完成，结果如下：")
    print(url_status)

• 补全了多进程提取URL的需求逻辑，支持整段文本、消息列表两种原始数据格式，提取后自动对URL去重，避免重复检测浪费资源
• 删除了原代码中错误的print(load_url(urls, 5))调用，该调用直接把URL列表传入需要单个URL参数的函数，会直接触发类型错误
• 调整了线程池的代码结构，将结果遍历逻辑放到ThreadPoolExecutor的上下文管理器内部，避免线程池提前关闭导致任务未提交就终止
• 新增了URL与状态码的绑定逻辑，不再仅存储状态码列表，最终直接生成符合要求的URL为键、状态码为值的字典
• 优化了异常返回格式，将异常类型转为可读的类名字符串，方便后续排查无效URL的报错原因

内容的提问来源于stack exchange，提问作者Yuliia