无需Telegram API：优化频道爬虫速度的方法

核心优化方向及实现 #

1. 复用HTTP连接（减少TCP握手开销） #

使用requests.Session()代替单次requests.get()，Session会维护连接池，重复请求同一域名时复用已有连接，大幅降低网络握手的时间损耗。

2. 异步并发请求（突破串行瓶颈） #

原代码采用同步串行请求，每个页面必须等待前一个请求完成才能发起下一个。改用异步框架aiohttp并发请求多个页面，将等待服务器响应的空闲时间充分利用，这是提升爬取效率最显著的手段。

3. 更换更快的HTML解析器 #

html5lib解析速度偏慢，换成lxml（需提前执行pip install lxml安装）或Python内置的html.parser，解析效率可提升数倍。

4. 控制并发数与添加重试机制 #

避免并发过高触发服务器反爬限制，同时添加重试逻辑处理网络波动导致的请求失败，避免中途中断浪费时间。

优化后的完整代码 #

import aiohttp
import asyncio
from bs4 import BeautifulSoup
import pandas as pd
from tenacity import retry, stop_after_attempt, wait_exponential  # 需执行pip install tenacity安装

async def fetch_page(session, url):
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    async def _fetch():
        async with session.get(url) as response:
            response.raise_for_status()
            return await response.text()
    return await _fetch()

async def parse_page(html):
    soup = BeautifulSoup(html, 'lxml')
    # 用列表推导式批量提取数据，比循环append更高效
    texts = [t.text for t in soup.find_all("div", class_="tgme_widget_message_text")]
    times = [j["datetime"] for j in soup.find_all("time", class_="time")]
    return texts, times

async def main():
    base_url = "https://t.me/s/dollar_tehran3bze?before={}"
    start_x = 415568
    end_x = 350000
    step = 20
    # 生成所有待爬取的URL
    urls = [base_url.format(x) for x in range(start_x, end_x, -step)]
    
    # 限制并发数，避免请求过于频繁
    connector = aiohttp.TCPConnector(limit=10)
    async with aiohttp.ClientSession(connector=connector) as session:
        # 并发获取所有页面HTML
        html_tasks = [fetch_page(session, url) for url in urls]
        htmls = await asyncio.gather(*html_tasks)
        
        # 并发解析所有页面
        parse_tasks = [parse_page(html) for html in htmls]
        results = await asyncio.gather(*parse_tasks)
        
        # 合并所有结果
        text1 = []
        time1 = []
        for texts, times in results:
            text1.extend(texts)
            time1.extend(times)
    
    print(f"len(text1): {len(text1)} | len(time1): {len(time1)}")
    df = pd.DataFrame({"time": time1, "text": text1})
    return df

if __name__ == "__main__":
    # Windows系统需取消下方注释
    # asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    df = asyncio.run(main())
    print(df.head())

额外优化建议 #

• 添加请求头：模拟浏览器请求（比如设置User-Agent），避免被服务器识别为爬虫而限制速度或返回错误。
• 增量爬取：记录上次爬取的最后一个x值，下次仅爬取新增内容，避免重复爬取历史数据。
• 分批保存数据：若数据量较大，每爬取一定数量页面就保存一次数据，防止程序崩溃导致数据丢失。

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