[Bug]: 图片请求失败的bug以及解决方法，[ERROR] : 400, message='Bad Request', url='http://multimedia.nt.qq.com.cn/download'

[sanxianxiaohuntun]

消息平台适配器

aiocqhttp（使用 OneBot 协议接入的）

运行环境

Windows

LangBot 版本

最新版本

异常情况

[12-17 22:47:09.652] process.py (42) - [INFO] : 处理 person_ 的请求(3): [图片]
[12-17 22:47:11.769] chat.py (94) - [ERROR] : 对话(3)请求失败: ClientResponseError 400, message='Bad Request', url='http://multimedia.nt.qq.com.cn/download'
[12-17 22:47:12.081] controller.py (98) - [ERROR] : 400, message='Bad Request', url='http://multimedia.nt.qq.com.cn/download'

启用的插件

无
pkg\utils\image.py出现的错误

源文件

import base64
import typing
from urllib.parse import urlparse, parse_qs
import ssl

import aiohttp


def get_qq_image_downloadable_url(image_url: str) -> tuple[str, dict]:
    """获取QQ图片的下载链接"""
    parsed = urlparse(image_url)
    query = parse_qs(parsed.query)
    return f"http://{parsed.netloc}{parsed.path}", query


async def get_qq_image_bytes(image_url: str) -> tuple[bytes, str]:
    """获取QQ图片的bytes"""
    image_url, query = get_qq_image_downloadable_url(image_url)
    ssl_context = ssl.create_default_context()
    ssl_context.check_hostname = False
    ssl_context.verify_mode = ssl.CERT_NONE
    async with aiohttp.ClientSession(trust_env=False) as session:
        async with session.get(image_url, params=query, ssl=ssl_context) as resp:
            resp.raise_for_status()
            file_bytes = await resp.read()
            content_type = resp.headers.get('Content-Type')
            if not content_type or not content_type.startswith('image/'):
                image_format = 'jpeg'
            else: 
                image_format = content_type.split('/')[-1]
            return file_bytes, image_format


async def qq_image_url_to_base64(
    image_url: str
) -> typing.Tuple[str, str]:
    """将QQ图片URL转为base64，并返回图片格式

    Args:
        image_url (str): QQ图片URL

    Returns:
        typing.Tuple[str, str]: base64编码和图片格式
    """
    image_url, query = get_qq_image_downloadable_url(image_url)

    # Flatten the query dictionary
    query = {k: v[0] for k, v in query.items()}

    file_bytes, image_format = await get_qq_image_bytes(image_url)

    base64_str = base64.b64encode(file_bytes).decode()

    return base64_str, image_format

解决方法

import base64
import typing
from urllib.parse import urlparse, parse_qs
import ssl
import logging
import aiohttp


logger = logging.getLogger(__name__)


def get_qq_image_downloadable_url(image_url: str) -> tuple[str, dict]:
    """获取QQ图片的下载链接"""
    parsed = urlparse(image_url)
    query = parse_qs(parsed.query)
    return f"http://{parsed.netloc}{parsed.path}", query


async def get_qq_image_bytes(image_url: str) -> tuple[bytes, str]:
    """获取QQ图片的bytes"""
    image_url, query = get_qq_image_downloadable_url(image_url)
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36',
        'Referer': 'https://multimedia.nt.qq.com.cn/'
    }
    try:
        async with aiohttp.ClientSession(trust_env=True) as session:  # 尝试开启代理设置
            async with session.get(image_url, params=query, headers=headers) as resp:
                resp.raise_for_status()
                file_bytes = await resp.read()
                content_type = resp.headers.get('Content-Type')
                if not content_type or not content_type.startswith('image/'):
                   image_format = 'jpeg'
                else:
                  image_format = content_type.split('/')[-1]

                return file_bytes, image_format

    except aiohttp.ClientResponseError as e:
        logger.error(f"Error downloading image from {image_url}: {e}")
        logger.error(f"Response status code: {e.status}")
        try:
            content = await resp.text()
        except Exception:
            content = "<Response content is not text>"
        logger.error(f"Response content: {content}")
        raise
    except Exception as e:
        logger.error(f"An unexpected error occurred while downloading image from {image_url}: {e}")
        raise


async def qq_image_url_to_base64(
    image_url: str
) -> typing.Tuple[str, str]:
    """将QQ图片URL转为base64，并返回图片格式

    Args:
        image_url (str): QQ图片URL

    Returns:
        typing.Tuple[str, str]: base64编码和图片格式
    """
    logger.debug(f"Converting image URL to base64: {image_url}")
    file_bytes, image_format = await get_qq_image_bytes(image_url)

    base64_str = base64.b64encode(file_bytes).decode()

    return base64_str, image_format

新增部分

    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36',
        'Referer': 'https://multimedia.nt.qq.com.cn/'
    }

模拟了浏览器请求之后请求成功

[12-17 23:20:49.937] process.py (42) - [INFO] : 处理 person_ 的请求(0): [图片]
[12-17 23:21:02.949] chat.py (80) - [INFO] : 对话(0)响应: assistant: 哇，好可爱的小女孩