Asyncio嵌套循环实现网站多板块多页面异步解析问题

嘿，刚上手Asyncio就搞双层异步爬取，思路很对！我帮你把代码补全并优化，实现外层异步遍历板块、内层异步处理每个板块下的页面，最大化抓取效率。

先给你完整的可运行代码，然后拆解关键部分：

import asyncio
import aiohttp
from bs4 import BeautifulSoup

class FinViz():
    def __init__(self):
        self.base_url = 'https://finviz.com/screener.ashx?v=160&s='
        # 补全你提到的6个板块（这里填了示例，你替换成自己的）
        self.signals = {
            'Earnings_Before': 'n_earningsbefore',
            'Earnings_After': 'n_earningsafter',
            'High_ROE': 'n_highroe',
            'Low_PE': 'n_lowpe',
            'Growth_Stocks': 'n_growth',
            'Value_Stocks': 'n_value'
        }
        # 每个板块的页面数（如果不知道总数，可以先爬第一页解析分页栏获取）
        self.pages_per_signal = 5
        # 反爬用：限制并发请求数，避免被封
        self.semaphore = asyncio.Semaphore(5)

    async def fetch_page(self, session, url):
        """异步获取单页HTML，带并发限制和反爬延迟"""
        async with self.semaphore:
            try:
                await asyncio.sleep(0.4)  # 加小延迟，模拟人类访问
                headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
                async with session.get(url, headers=headers) as response:
                    if response.status == 200:
                        return await response.text()
                    print(f"⚠️ 请求失败，状态码 {response.status} | URL: {url}")
                    return None
            except Exception as e:
                print(f"❌ 请求异常: {str(e)} | URL: {url}")
                return None

    async def parse_table(self, html):
        """解析页面中的表格数据，返回结构化列表"""
        if not html:
            return []
        soup = BeautifulSoup(html, 'html.parser')
        # 找到目标表格（FinViz的表格类是table-light，可根据实际调整）
        table = soup.find('table', class_='table-light')
        if not table:
            print("⚠️ 未找到目标表格")
            return []
        
        # 提取表头和数据行
        headers = [th.get_text(strip=True) for th in table.find_all('th')]
        rows = []
        for tr in table.find_all('tr')[1:]:  # 跳过表头行
            row_data = [td.get_text(strip=True) for td in tr.find_all('td')]
            if row_data:
                rows.append(dict(zip(headers, row_data)))  # 转成字典更易处理
        return rows

    async def process_single_signal(self, session, signal_name, signal_code):
        """异步处理单个板块下的所有页面"""
        print(f"🚀 开始处理板块: {signal_name}")
        # 生成该板块下所有页面的URL（FinViz分页用r参数，起始条目数）
        page_tasks = []
        for page_num in range(1, self.pages_per_signal + 1):
            start_item = (page_num - 1) * 20 + 1  # 每页20条，第一页r=1，第二页r=21
            page_url = f"{self.base_url}{signal_code}&r={start_item}"
            page_tasks.append(asyncio.create_task(self.fetch_page(session, page_url)))
        
        # 并发获取所有页面的HTML
        page_htmls = await asyncio.gather(*page_tasks)
        
        # 并发解析所有页面的表格
        parse_tasks = [asyncio.create_task(self.parse_table(html)) for html in page_htmls if html]
        all_page_data = await asyncio.gather(*parse_tasks)
        
        # 合并当前板块的所有数据
        merged_data = [item for sublist in all_page_data for item in sublist]
        print(f"✅ 完成板块: {signal_name} | 共抓取 {len(merged_data)} 条数据")
        return {signal_name: merged_data}

    async def run(self):
        """主入口：并发处理所有板块"""
        async with aiohttp.ClientSession() as session:
            # 为每个板块创建异步任务
            signal_tasks = []
            for signal_name, signal_code in self.signals.items():
                task = asyncio.create_task(self.process_single_signal(session, signal_name, signal_code))
                signal_tasks.append(task)
            
            # 等待所有板块任务完成，收集结果
            all_results = await asyncio.gather(*signal_tasks)
            
            # 合并所有板块的数据到一个字典
            final_data = {}
            for result in all_results:
                final_data.update(result)
            return final_data

if __name__ == "__main__":
    finviz_scraper = FinViz()
    try:
        all_data = asyncio.run(finviz_scraper.run())
        print("\n🎉 所有板块抓取完成！")
        # 这里可以把数据保存到CSV/JSON，比如：
        # import json
        # with open('finviz_data.json', 'w', encoding='utf-8') as f:
        #     json.dump(all_data, f, indent=2)
    except KeyboardInterrupt:
        print("\n⏹️ 用户中断程序")

关键细节说明 #

1. 双层异步逻辑

   • 外层：run()方法为每个板块创建异步任务，用asyncio.gather()并发执行，实现板块间的异步处理。
   • 内层：process_single_signal()为当前板块的所有页面创建抓取任务，同样用gather()并发获取和解析，实现页面间的异步处理。

2. 反爬优化

   • 用asyncio.Semaphore限制并发请求数（这里设为5），避免瞬间请求过多被封IP。
   • 加await asyncio.sleep(0.4)模拟人类访问间隔。
   • 自定义User-Agent，避免被识别为爬虫。

3. 表格解析

   • 用BeautifulSoup解析表格，把每行数据转成字典（表头为键，单元格内容为值），后续处理更方便。
   • 处理了页面请求失败、未找到表格等异常情况，避免程序崩溃。

4. 分页参数注意
   FinViz的分页参数r是起始条目数，每页显示20条，所以第N页的r值是(N-1)*20 + 1，别写错了哦！

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