Python异步解析本地商店数据不稳定：数据缺失排查与日志方案

问题描述 #

编写的Python异步商品解析器结果不稳定，预期获取约11000条商品数据，但会随机只拿到部分数据，且无任何异常抛出。相关代码如下：

URL = 'https://shop.samberi.com'

HEADERS = {
    'Accept': '*/*',
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
                  'AppleWebKit/537.36 (KHTML, like Gecko) '
                  'Chrome/101.0.4951.54 Safari/537.36'
}

all_products = []

async def get_products(url):
    async with aiohttp.ClientSession() as session:
        res = await session.get(url=url, headers=HEADERS)
        bs = BeautifulSoup(await res.text(), 'lxml')
        cats = [URL + cat.get('href') + '?SHOWALL_1=1'
                for cat in bs.find('ul', id='vertical-multilevel-menu')
                .find_all('a', class_='parent')] + [
            #Костыль, не могу получить эти ссылки автоматически(
            'https://shop.samberi.com/catalog/aziya/?SHOWALL_1=1',
            'https://shop.samberi.com/catalog/sportivnye_tovary/?SHOWALL_1=1',
            'https://shop.samberi.com/catalog/upakovka/?SHOWALL_1=1'
        ]
        tasks = [asyncio.shield(parse_page(session, url, max_s)) for url in cats]

        await asyncio.gather(*tasks)

async def parse_page(session, cat_url, max_s):
    async with session.get(url=cat_url, headers=HEADERS) as res:
        res_text = await res.text()
        pagebs = BeautifulSoup(res_text, 'lxml')
        products_on_page = pagebs.find_all('div', class_='product-item')
        for product in products_on_page:
            name = product.find('div', class_='product-item-title').text.strip()
            price = product.find('span', class_='product-item-price-current')\
                .text.strip().strip('₽').strip()
            all_products.append([name, price])
def main():
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(get_products(URL))

可能的问题点 #

1. 未定义变量错误：代码中get_products创建任务时传入了max_s参数，但该变量从未定义，理论上会触发NameError；若实际运行未报错，可能是代码粘贴遗漏或运行环境存在同名变量导致隐性问题。
2. 协程安全问题：全局列表all_products不是协程安全容器，多个异步任务同时调用append时，可能出现数据覆盖或丢失，导致最终结果不全。
3. 无限制并发触发限流：一次性发起所有分类页面请求，网站可能通过静默返回空页面/部分内容的方式限流，不会返回HTTP错误码，导致解析商品数量不足。
4. 网络请求无超时处理：未设置请求超时，部分请求可能因网络问题长时间挂起，或网站故意延迟响应，导致任务未完成但未触发异常。
5. 解析逻辑无异常防护：直接调用find结果的text属性，若页面中不存在目标元素（如商品价格）会触发AttributeError，但用户反馈无异常，可能是网站返回的页面结构随机变化，部分页面无商品元素但异常被隐性吞掉。

捕获与记录问题的方案 #

1. 修复代码显性错误 #

删除parse_page函数中无用的max_s参数，修正任务创建代码：

tasks = [asyncio.shield(parse_page(session, url)) for url in cats]

# 同时修改parse_page定义：
async def parse_page(session, cat_url):

2. 保证协程数据安全 #

使用asyncio.Lock保护全局列表的写入操作，避免并发冲突：

import asyncio

all_products = []
products_lock = asyncio.Lock()

# 在parse_page的append处添加锁：
async with products_lock:
    all_products.append([name, price])

3. 控制并发数并添加超时 #

用信号量限制并发请求数量，同时给每个请求设置超时，避免被限流或请求挂起：

async def get_products(url):
    async with aiohttp.ClientSession() as session:
        # 限制并发数为5，可根据网站情况调整
        semaphore = asyncio.Semaphore(5)
        
        # 包装请求函数，添加信号量和异常捕获
        async def safe_parse(cat_url):
            async with semaphore:
                try:
                    await parse_page(session, cat_url)
                except Exception as e:
                    logging.error(f"处理页面 {cat_url} 出错: {str(e)}")
        
        cats = [...]  # 原分类列表逻辑
        tasks = [asyncio.shield(safe_parse(url)) for url in cats]
        await asyncio.gather(*tasks)

# 修改parse_page中的请求：
async def parse_page(session, cat_url):
    timeout = aiohttp.ClientTimeout(total=15)  # 设置15秒超时
    async with session.get(url=cat_url, headers=HEADERS, timeout=timeout) as res:
        # 记录请求状态码
        logging.info(f"请求 {cat_url} 状态码: {res.status}")
        res_text = await res.text()
        # 记录返回内容长度，判断是否为空
        logging.info(f"请求 {cat_url} 返回内容长度: {len(res_text)}")
        pagebs = BeautifulSoup(res_text, 'lxml')
        products_on_page = pagebs.find_all('div', class_='product-item')
        logging.info(f"页面 {cat_url} 解析到商品数: {len(products_on_page)}")
        for product in products_on_page:
            name_elem = product.find('div', class_='product-item-title')
            price_elem = product.find('span', class_='product-item-price-current')
            # 检查元素是否存在，避免异常
            if not name_elem or not price_elem:
                logging.warning(f"页面 {cat_url} 存在无效商品元素")
                continue
            name = name_elem.text.strip()
            price = price_elem.text.strip().strip('₽').strip()
            async with products_lock:
                all_products.append([name, price])

4. 增加日志记录 #

使用Python标准库logging将请求状态、解析结果、异常信息写入日志文件，方便后续排查：

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    filename='parser.log'
)

5. 结果校验 #

在main函数末尾添加结果数量校验，对比预期值并记录差异：

def main():
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(get_products(URL))
    expected_count = 11000
    actual_count = len(all_products)
    logging.info(f"最终获取商品数: {actual_count}, 预期: {expected_count}")
    if actual_count != expected_count:
        logging.warning(f"商品数量不符，缺失 {expected_count - actual_count} 条")

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