使用asyncio+aiofiles批量解析HTML生成DataFrame遇await错误求助

我来帮你解决这个问题！你遇到的TypeError: object dict can't be used in 'await' expression，核心原因是你在尝试await一个普通的同步解析函数——只有协程（用async def定义的函数）或者可等待对象才能被await。下面我一步步给你讲清楚怎么改造代码，让异步处理真正跑起来：

先明确异步处理的核心逻辑 #

你的场景里，文件读取是IO密集型操作（适合用异步提升效率），但BeautifulSoup解析HTML是CPU密集型操作（不能直接用异步协程，会阻塞事件循环）。所以正确的思路是：

1. 用aiofiles异步读取文件内容
2. 把CPU密集的解析任务放到线程池（或进程池）里执行，避免阻塞异步事件循环

完整改造代码示例 #

import asyncio
import aiofiles
from bs4 import BeautifulSoup
import pandas as pd
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

# 1. 保留你的同步解析函数（不用改！）
# 这部分是CPU密集型，我们后面用线程池来异步调用它
def parse_html(html_content):
    soup = BeautifulSoup(html_content, 'html.parser')
    # 替换成你实际的解析逻辑，返回字典
    return {
        'title': soup.title.string.strip() if soup.title else None,
        # 其他需要提取的字段...
    }

# 2. 异步处理单个文件：异步读文件 + 线程池跑解析
async def process_single_file(file_path, executor):
    try:
        # 异步读取文件（IO操作，适合异步）
        async with aiofiles.open(file_path, 'r', encoding='utf-8') as f:
            html_content = await f.read()
        
        # 用线程池执行同步解析函数，避免阻塞事件循环
        parsed_result = await asyncio.get_event_loop().run_in_executor(
            executor, parse_html, html_content
        )
        return parsed_result
    except Exception as e:
        print(f"处理文件失败 {file_path}: {str(e)}")
        return None  # 出错时返回标记，后续可以过滤

# 3. 主异步函数：分批次处理所有文件
async def main(html_dir, batch_size=100):
    # 获取所有HTML文件路径（替换成你的目录）
    file_paths = list(Path(html_dir).glob("*.html"))
    total_files = len(file_paths)
    print(f"共发现 {total_files} 个HTML文件")

    all_results = []
    # 创建线程池，大小根据CPU核心数调整（比如CPU核心数*2）
    with ThreadPoolExecutor(max_workers=8) as executor:
        # 分批次处理，避免一次性创建4万个任务导致内存爆炸
        for i in range(0, total_files, batch_size):
            batch = file_paths[i:i+batch_size]
            # 为批次内每个文件创建异步任务
            tasks = [process_single_file(path, executor) for path in batch]
            # 等待批次内所有任务完成
            batch_results = await asyncio.gather(*tasks)
            # 过滤掉出错的None结果
            all_results.extend([res for res in batch_results if res is not None])
            print(f"已处理 {min(i+batch_size, total_files)}/{total_files} 个文件")
    
    # 转成DataFrame
    df = pd.DataFrame(all_results)
    return df

if __name__ == "__main__":
    # 替换成你的HTML文件所在目录
    df = asyncio.run(main("./your_html_files_dir"))
    # 保存结果
    df.to_csv("parsed_html_results.csv", index=False)

关键细节解释 #

1. 为什么不能直接await同步解析函数？
   你之前的代码应该是写了result = await parse_html(...)，但parse_html是普通的同步函数（没有async def修饰），不是协程，所以不能用await。我们用run_in_executor把它包装成可等待对象，让事件循环在处理解析时可以去处理其他文件的读取。

2. 为什么要分批次处理？
   一次性给4万个文件创建异步任务会占用大量内存，分批次（比如每次处理100个）可以控制内存占用，同时保证效率。你可以根据自己的机器性能调整batch_size。

3. 线程池 vs 进程池？
   如果你的解析逻辑非常耗时，也可以把ThreadPoolExecutor换成ProcessPoolExecutor（进程池），绕过Python的GIL限制，但进程池的开销比线程池大，需要根据实际情况权衡。

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