如何在asyncio循环中过滤引发JSON解码错误的Virustotal API请求

针对你遇到的500内部服务器错误导致JSON解码失败的问题，可以通过以下几个步骤在asyncio循环中规避这类错误：

1. 先校验响应状态码再解码JSON #

在尝试解析响应为JSON之前，先检查HTTP状态码。对于4xx/5xx这类错误状态码，直接跳过解码逻辑，记录错误信息即可，避免无效的JSON解码操作。

示例代码片段：

async def fetch_vt(session, url, api_key):
    headers = {"x-apikey": api_key}
    vt_endpoint = f"https://www.virustotal.com/api/v3/urls/{url}"
    try:
        async with session.get(vt_endpoint, headers=headers) as resp:
            # 仅处理成功的响应（2xx状态码）
            if resp.status >= 400:
                print(f"请求URL {url}失败，状态码: {resp.status}")
                return None
            # 后续再尝试解码JSON
            data = await resp.json()
            return data
    except aiohttp.ClientError as e:
        print(f"URL {url}请求异常: {str(e)}")
        return None

2. 捕获JSON解码异常 #

即使状态码显示成功，也可能存在响应非JSON的情况（比如服务器返回错误页面）。此时需要将resp.json()调用包裹在try-except块中，捕获JSONDecodeError，避免整个任务崩溃。

修改后的代码：

from json.decoder import JSONDecodeError

async def fetch_vt(session, url, api_key):
    headers = {"x-apikey": api_key}
    vt_endpoint = f"https://www.virustotal.com/api/v3/urls/{url}"
    try:
        async with session.get(vt_endpoint, headers=headers) as resp:
            if resp.status >= 400:
                # 可以获取错误响应的文本内容用于排查
                error_content = await resp.text()
                print(f"URL {url}请求失败，状态码{resp.status}，响应内容: {error_content[:200]}")
                return None
            try:
                data = await resp.json()
                return data
            except JSONDecodeError:
                html_content = await resp.text()
                print(f"URL {url}返回非JSON响应，内容片段: {html_content[:200]}")
                return None
    except aiohttp.ClientError as e:
        print(f"URL {url}请求异常: {str(e)}")
        return None

3. 给失败请求加重试机制 #

500错误往往是服务器临时故障，针对这类情况可以添加重试逻辑，避免单次失败就丢弃请求。可以用tenacity库实现异步重试，或者自己编写简单的重试逻辑。

示例（使用tenacity）：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_result

@retry(
    stop=stop_after_attempt(3),  # 最多重试3次
    wait=wait_exponential(multiplier=1, min=2, max=10),  # 指数退避等待
    retry=retry_if_result(lambda res: res is None)  # 仅当返回None时重试
)
async def fetch_vt(session, url, api_key):
    # 上面的fetch_vt代码

4. 控制请求速率和并发数 #

Virustotal API有严格的速率限制，过度并发可能触发服务器错误。可以用asyncio.Semaphore控制并发请求数，同时添加适当的请求间隔，避免触发限流或服务器错误。

示例：

async def fetch_vt(session, url, api_key, semaphore):
    async with semaphore:
        await asyncio.sleep(1)  # 每次请求间隔1秒
        # 上面的fetch_vt核心逻辑

async def main(url_list, api_key):
    # 限制并发数为5
    semaphore = asyncio.Semaphore(5)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_vt(session, url, api_key, semaphore) for url in url_list]
        results = await asyncio.gather(*tasks)
        # 过滤掉无效结果
        valid_data = [res for res in results if res is not None]
        print(f"成功处理{len(valid_data)}/{len(url_list)}条URL")

把这些策略结合起来，就能有效规避请求失败导致的JSON解码错误，同时提升批量请求的稳定性。

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