Skip to content

QQ官机适配器优化 #9204

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
u0-ani-nya wants to merge 3 commits into
base: master
Choose a base branch
from
+960 −44
Open

QQ官机适配器优化 #9204

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
42a3a4e
Add files via upload
u0-ani-nya Jul 10, 2026
a3d7a69
Add files via upload
u0-ani-nya Jul 10, 2026
b724d10
Create files.py
u0-ani-nya Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
203 changes: 160 additions & 43 deletions astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from typing import cast

import aiofiles
import aiohttp
import botpy
import botpy.errors
import botpy.message
Expand All @@ -27,11 +28,8 @@
from astrbot.api.event import AstrMessageEvent, MessageChain
from astrbot.api.message_components import File, Image, Plain, Record, Video
from astrbot.api.platform import AstrBotMessage, PlatformMetadata
from astrbot.core.utils.media_utils import MediaResolver, file_uri_to_path, is_file_uri


class APIReturnNoneError(Exception):
pass
from astrbot.core.utils.media_utils import MediaResolver, file_uri_to_path, is_file_uri


def _patch_qq_botpy_formdata() -> None:
Expand All @@ -51,6 +49,10 @@ def _patch_qq_botpy_formdata() -> None:
logger.debug("[QQOfficial] Skip botpy FormData patch.")


class APIReturnNoneError(Exception):
pass


_patch_qq_botpy_formdata()


Expand All @@ -72,7 +74,6 @@ def _qqofficial_retry(max_attempts: int = 5):
reraise=True,
)


_QQOFFICIAL_SEND_API_ERRORS = (
botpy.errors.ForbiddenError,
botpy.errors.MethodNotAllowedError,
Expand All @@ -81,6 +82,55 @@ def _qqofficial_retry(max_attempts: int = 5):
botpy.errors.ServerError,
)

# URL 上传模式可用性缓存

@sourcery-ai sourcery-ai Bot Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (complexity): Consider encapsulating the URL-upload behavior into a dedicated strategy with shared retry helpers and smaller upload helpers to make upload_group_and_c2c_media and temp-file handling easier to understand and maintain.

You can keep all behavior while reducing complexity by:

  1. Encapsulating URL-upload state and helpers
  2. Unifying retry logic for URL vs Base64
  3. Flattening upload_group_and_c2c_media into smaller helpers

1. Encapsulate URL-upload state instead of globals

Move _url_upload_available / _media_upload_base_url and probe/prepare/cleanup into a small helper or into QQOfficialMessageEvent as a cohesive unit. This keeps related logic together and avoids scattered globals.

For example, extract a minimal helper:

class MediaUploadStrategy:
    def __init__(self) -> None:
        self.available: bool | None = None
        self.base_url: str = ""

    async def init_probe(self, media_upload_url: str = "") -> None:
        await asyncio.sleep(10)
        await self._cleanup_stale_temp_files()

        if not media_upload_url:
            self.available = False
            logger.info("[QQOfficial] 未配置 media_upload_url，URL 上传模式不可用，将使用 Base64")
            return

        self.base_url = media_upload_url.rstrip('/')
        for attempt in range(3):
            self.available = await QQOfficialMessageEvent._probe_temp_url_reachable(self.base_url)
            if self.available:
                break
            if attempt < 2:
                await asyncio.sleep(3)

        if self.available:
            logger.info(f"[QQOfficial] URL 上传模式已启用 (callback_api_base={self.base_url})")
        else:
            logger.warning(f"[QQOfficial] callback_api_base 不可达 ({self.base_url})，回退 Base64 模式")

    async def _cleanup_stale_temp_files(self) -> None:
        temp_dir = os.path.join(os.getcwd(), "data", "media_upload_cache")
        # ... keep your existing cleanup logic ...

Then hang an instance off QQOfficialMessageEvent (or the adapter) instead of using module-level globals:

class QQOfficialMessageEvent(AstrMessageEvent):
    media_upload_strategy = MediaUploadStrategy()

    @staticmethod
    async def _prepare_temp_file(file_source: str) -> tuple[str | None, str | None]:
        strategy = QQOfficialMessageEvent.media_upload_strategy
        if not strategy.available or not strategy.base_url:
            return None, None
        # ... existing copy/shutil logic using strategy.base_url ...

This keeps state and behavior together and makes the lifecycle clearer.

2. Unify retry behavior for URL and Base64

You currently have:

  • _qqofficial_retry decorator for base64 path.
  • Manual for _attempt in range(3) loop for URL path.

You can reuse _qqofficial_retry by parameterizing max attempts and keeping URL-specific logic in a small helper:

async def _request_with_retry(self, route: Route, payload: dict, max_attempts: int = 5):
    @_qqofficial_retry(max_attempts=max_attempts)
    async def _do():
        return await self.bot.api._http.request(route, json=payload)
    return await _do()

Then in upload_group_and_c2c_media:

if temp_url:
    payload["url"] = temp_url
    try:
        url_result = await self._request_with_retry(route, payload, max_attempts=3)
    except Exception as e:
        logger.warning(f"[QQOfficial] URL 上传失败 ({e})，回退 Base64")
        payload.pop("url", None)
    else:
        if isinstance(url_result, dict):
            self._cleanup_temp_file(temp_path)
            temp_path = None
            return Media(
                file_uuid=url_result["file_uuid"],
                file_info=url_result["file_info"],
                ttl=url_result.get("ttl", 0),
            )
        logger.warning("[QQOfficial] URL 上传响应格式错误，回退 Base64")
        payload.pop("url", None)

This removes the manual for loop and aligns retries with the rest of the API.

3. Split upload_group_and_c2c_media into focused helpers

Right now upload_group_and_c2c_media decides route, temp-file URL path with retries, base64 fallback, and result-to-Media. You can make the top-level method linear by moving responsibilities into small functions:

async def upload_group_and_c2c_media(
    self,
    file_source: str,
    file_type: int,
    srv_send_msg: bool = False,
    file_name: str | None = None,
    **kwargs,
) -> Media | None:
    payload = self._build_base_payload(file_type, srv_send_msg, file_name)
    route = self._build_route(payload, **kwargs)
    if route is None:
        return None

    if self._should_try_url_upload(file_source, file_type):
        media = await self._attempt_url_upload(file_source, payload, route)
        if media:
            return media

    media = await self._attempt_base64_upload(file_source, payload, route)
    return media

Helper implementations stay small and focused:

def _should_try_url_upload(self, file_source: str, file_type: int) -> bool:
    return os.path.exists(file_source) and file_type in (
        self.VIDEO_FILE_TYPE,
        self.FILE_FILE_TYPE,
    )

async def _attempt_base64_upload(
    self, file_source: str, payload: dict, route: Route
) -> Media | None:
    if os.path.exists(file_source):
        async with aiofiles.open(file_source, "rb") as f:
            file_content = await f.read()
            payload["file_data"] = base64.b64encode(file_content).decode("utf-8")
    else:
        payload["url"] = file_source

    try:
        result = await self._request_with_retry(route, payload)
        if not isinstance(result, dict):
            logger.error(f"上传文件响应格式错误: {result}")
            return None
        return Media(
            file_uuid=result["file_uuid"],
            file_info=result["file_info"],
            ttl=result.get("ttl", 0),
        )
    except (botpy.errors.ServerError, botpy.errors.SequenceNumberError):
        logger.error(f"上传媒体文件失败，共尝试5次后放弃: {file_source}")
    except Exception as e:
        logger.error(f"上传请求错误: {e}")
    return None

This keeps upload_group_and_c2c_media readable and makes the URL/base64 behavior easier to reason about.

4. Make temp-file lifecycle rules explicit

Instead of relying on comments and implicit “leave some files, cleanup on startup”, codify the policy with a flag or clear branch. For example:

DEBUG_KEEP_TEMP = False  # or from config

def _cleanup_temp_file(self, temp_path: str | None, *, on_success: bool = False) -> None:
    if DEBUG_KEEP_TEMP and not on_success:
        logger.info(f"[QQOfficial] 调试模式保留临时文件: {temp_path}")
        return
    if temp_path and os.path.exists(temp_path):
        try:
            os.unlink(temp_path)
            logger.info(f"[QQOfficial] 已清理临时文件: {temp_path}")
        except OSError as e:
            logger.warning(f"[QQOfficial] 清理临时文件失败: {temp_path} ({e})")

Then call self._cleanup_temp_file(temp_path, on_success=True) on successful URL upload, and on_success=False on failure. This makes behavior explicit and reduces subtlety around “无需清理 / 兜底清理”.

_url_upload_available: bool | None = None
_media_upload_base_url: str = ""


async def init_url_upload_probe(media_upload_url: str = "") -> None:
"""启动时探测媒体上传地址是否可达，结果缓存。
应在适配器启动时调用一次。同时清理上次遗留的临时文件。
"""
global _url_upload_available, _media_upload_base_url

# 等待 Web Server 启动完成
await asyncio.sleep(10)

# 清理上次遗留的临时文件
temp_dir = os.path.join(os.getcwd(), "data", "media_upload_cache")
if os.path.isdir(temp_dir):
try:
import glob
stale = glob.glob(os.path.join(temp_dir, "*"))
for f in stale:
try:
os.unlink(f)
except OSError:
pass
if stale:
logger.info(f"[QQOfficial] 清理遗留临时文件: {len(stale)} 个")
except Exception:
pass

if not media_upload_url:
_url_upload_available = False
logger.info("[QQOfficial] 未配置 media_upload_url，URL 上传模式不可用，将使用 Base64")
return

_media_upload_base_url = media_upload_url.rstrip('/')

for attempt in range(3):
_url_upload_available = await QQOfficialMessageEvent._probe_temp_url_reachable(_media_upload_base_url)
if _url_upload_available:
break
if attempt < 2:
await asyncio.sleep(3)

if _url_upload_available:
logger.info(f"[QQOfficial] URL 上传模式已启用 (callback_api_base={_media_upload_base_url})")
else:
logger.warning(f"[QQOfficial] callback_api_base 不可达 ({_media_upload_base_url})，回退 Base64 模式")


class QQOfficialMessageEvent(AstrMessageEvent):
MARKDOWN_NOT_ALLOWED_ERROR = "不允许发送原生 markdown"
Expand Down Expand Up @@ -502,7 +552,6 @@ async def _send_with_markdown_fallback(
logger.info("[QQOfficial] 回复消息失败: %s, 尝试使用主动发送接口。", err)
if payload.get("msg_id"):
fallback_payload = payload.copy()
fallback_payload.pop("msg_id", None)
try:
ret = await send_func(fallback_payload)
Comment on lines 553 to 556
logger.info("[QQOfficial] 使用主动发送接口发送成功。")
Comment on lines 553 to 557

@gemini-code-assist gemini-code-assist Bot Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

在重构中移除了 fallback_payload.pop('msg_id', None)。这导致 fallback_payload 与 payload 完全一致。当回复消息失败并尝试使用主动发送接口作为 fallback 时，由于 msg_id 仍然存在，API 调用会重复执行完全相同的请求并再次失败，使得 fallback 机制失效。请恢复 pop 操作以移除 msg_id。

Suggested change
if payload.get("msg_id"):
fallback_payload = payload.copy()
fallback_payload.pop("msg_id", None)
try:
ret = await send_func(fallback_payload)
logger.info("[QQOfficial] 使用主动发送接口发送成功。")
if payload.get('msg_id'):
fallback_payload = payload.copy()
fallback_payload.pop('msg_id', None)
try:
ret = await send_func(fallback_payload)
logger.info('[QQOfficial] 使用主动发送接口发送成功。')

Expand Down Expand Up @@ -574,27 +623,19 @@ async def _do_upload():
route = Route(
"POST", "/v2/users/{openid}/files", openid=kwargs["openid"]
)
return await self.bot.api._http.request(route, json=payload)
elif "group_openid" in kwargs:
payload["group_openid"] = kwargs["group_openid"]
route = Route(
"POST",
"/v2/groups/{group_openid}/files",
group_openid=kwargs["group_openid"],
)
return await self.bot.api._http.request(route, json=payload)
else:
raise ValueError("Invalid upload parameters")

result = await self.bot.api._http.request(route, json=payload)
if result is None:
err_msg = "上传图片API返回None，触发重试"
raise APIReturnNoneError(err_msg)
return result

try:
result = await _do_upload()
except APIReturnNoneError:
logger.warning(f"上传图片API返回None，共尝试5次后放弃: {payload}")
raise
result = await _do_upload()

if not isinstance(result, dict):
raise RuntimeError(
Expand All @@ -607,6 +648,58 @@ async def _do_upload():
ttl=result.get("ttl", 0),
)

@staticmethod
async def _probe_temp_url_reachable(base_url: str) -> bool:
"""探测 AstrBot Web Server 是否对外可达（GET 面板根路径）。"""
try:
async with aiohttp.ClientSession() as session:
async with session.get(
base_url,
timeout=aiohttp.ClientTimeout(total=5),
allow_redirects=True,
) as resp:
return resp.status == 200
except Exception:
return False

@staticmethod
async def _prepare_temp_file(file_source: str) -> tuple[str | None, str | None]:
"""将本地文件复制到临时目录（UUID 文件名），返回 (temp_path, temp_url)。
失败返回 (None, None)。
"""
if not _url_upload_available or not _media_upload_base_url:
return None, None

import shutil
import uuid as _uuid

ext = os.path.splitext(file_source)[1]
temp_name = f"{_uuid.uuid4().hex}{ext}"
temp_dir = os.path.join(os.getcwd(), "data", "media_upload_cache")
os.makedirs(temp_dir, exist_ok=True)
temp_path = os.path.join(temp_dir, temp_name)

try:
shutil.copy2(file_source, temp_path)
except Exception as e:
logger.warning(f"[QQOfficial] 创建临时文件失败: {e}")
return None, None
Comment on lines +682 to +686

@gemini-code-assist gemini-code-assist Bot Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

shutil.copy2 是一个同步阻塞的 I/O 操作。在 asyncio 事件循环中直接调用它会阻塞整个主线程，影响其他协程的并发执行（特别是当复制大文件或视频时）。建议使用 asyncio.to_thread 将其放到线程池中执行。

Suggested change
try:
shutil.copy2(file_source, temp_path)
except Exception as e:
logger.warning(f"[QQOfficial] 创建临时文件失败: {e}")
return None, None
try:
await asyncio.to_thread(shutil.copy2, file_source, temp_path)
except Exception as e:
logger.warning(f'[QQOfficial] 创建临时文件失败: {e}')
return None, None


url = f"{_media_upload_base_url}/api/media/{temp_name}"
logger.info(f"[QQOfficial] 临时文件: {temp_path}")
logger.info(f"[QQOfficial] 下载 URL: {url}")
return temp_path, url

@staticmethod
def _cleanup_temp_file(temp_path: str | None) -> None:
"""删除临时文件。"""
if temp_path and os.path.exists(temp_path):
try:
os.unlink(temp_path)
logger.info(f"[QQOfficial] 已清理临时文件: {temp_path}")
except OSError as e:
logger.warning(f"[QQOfficial] 清理临时文件失败: {temp_path} ({e})")

async def upload_group_and_c2c_media(
self,
file_source: str,
Expand All @@ -615,24 +708,16 @@ async def upload_group_and_c2c_media(
file_name: str | None = None,
**kwargs,
) -> Media | None:
"""上传媒体文件"""
# 构建基础payload
"""上传媒体文件。

视频/文件：URL 上传优先，失败回退 Base64。
音频：直接 Base64。
"""
payload: dict = {"file_type": file_type, "srv_send_msg": srv_send_msg}
if file_name:
payload["file_name"] = file_name

# 处理文件数据
if os.path.exists(file_source):
# 读取本地文件
async with aiofiles.open(file_source, "rb") as f:
file_content = await f.read()
# use base64 encode
payload["file_data"] = base64.b64encode(file_content).decode("utf-8")
else:
# 使用URL
payload["url"] = file_source

# 添加接收者信息和确定路由
route: Route | None = None
if "openid" in kwargs:
payload["openid"] = kwargs["openid"]
route = Route("POST", "/v2/users/{openid}/files", openid=kwargs["openid"])
Expand All @@ -646,34 +731,66 @@ async def upload_group_and_c2c_media(
else:
return None

@_qqofficial_retry()
async def _do_upload():
result = await self.bot.api._http.request(route, json=payload)
if result is None:
err_msg = "上传文件API返回None，触发重试"
raise APIReturnNoneError(err_msg)
return result

temp_path: str | None = None
try:
result = await _do_upload()
# 视频/文件：尝试 URL 上传
if os.path.exists(file_source) and file_type in (self.VIDEO_FILE_TYPE, self.FILE_FILE_TYPE):
temp_path, temp_url = await self._prepare_temp_file(file_source)
if temp_url:
payload["url"] = temp_url
url_result = None
for _attempt in range(3):
try:
url_result = await self.bot.api._http.request(route, json=payload)
break
except Exception as e:
logger.warning(f"[QQOfficial] URL 上传失败 ({e})，重试...")
await asyncio.sleep(3)

if url_result and isinstance(url_result, dict):
logger.info(f"[QQOfficial] URL 上传成功: {file_source}")
self._cleanup_temp_file(temp_path)
temp_path = None
return Media(
file_uuid=url_result["file_uuid"],
file_info=url_result["file_info"],
ttl=url_result.get("ttl", 0),
)

logger.warning("[QQOfficial] URL 上传 3 次均失败，回退 Base64")
# 保留临时文件供排查，启动时兜底清理
temp_path = None
payload.pop("url", None)

# Base64 路径（音频、URL 不可用、或 URL 上传失败回退）
if os.path.exists(file_source):
async with aiofiles.open(file_source, "rb") as f:
file_content = await f.read()
payload["file_data"] = base64.b64encode(file_content).decode("utf-8")
else:
payload["url"] = file_source

@_qqofficial_retry()
async def _do_upload():
Comment on lines +773 to +774

@gemini-code-assist gemini-code-assist Bot Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

与图片上传类似，这里在重构 _do_upload 时也移除了 APIReturnNoneError 的抛出逻辑，导致 @_qqofficial_retry() 无法在 API 返回 None 时进行重试。请恢复 None 检查和异常抛出。

Suggested change
@_qqofficial_retry()
async def _do_upload():
async def _do_upload():
result = await self.bot.api._http.request(route, json=payload)
if result is None:
raise APIReturnNoneError('上传文件API返回None，触发重试')
return result

return await self.bot.api._http.request(route, json=payload)
Comment on lines +773 to +775

result = await _do_upload()
if result:
if not isinstance(result, dict):
logger.error(f"上传文件响应格式错误: {result}")
return None

return Media(
file_uuid=result["file_uuid"],
file_info=result["file_info"],
ttl=result.get("ttl", 0),
)
except APIReturnNoneError:
logger.warning(f"上传文件API返回None，共尝试5次后放弃: {file_source}")
except (botpy.errors.ServerError, botpy.errors.SequenceNumberError):
logger.error(f"上传媒体文件失败，共尝试5次后放弃: {file_source}")
except Exception as e:
logger.error(f"上传请求错误: {e}")

# URL 上传成功时 temp_path 已返回保留，此处无需清理。
# URL 上传失败时已在上方清理。残留文件由启动时的 init_url_upload_probe() 兜底清理。
return None

async def post_c2c_message(
Expand Down
6 changes: 5 additions & 1 deletion astrbot/core/platform/sources/qqofficial/qqofficial_platform_adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
from astrbot.core.utils.media_utils import MediaResolver

from ...register import register_platform_adapter
from .qqofficial_message_event import QQOfficialMessageEvent
from .qqofficial_message_event import QQOfficialMessageEvent, init_url_upload_probe

# remove logger handler
for handler in logging.root.handlers[:]:
Expand Down Expand Up @@ -850,6 +850,10 @@ async def _parse_from_qqofficial(
return abm

def run(self):
# 启动时探测 URL 上传可用性（结果缓存，只执行一次）
from astrbot.core import astrbot_config
callback_base = astrbot_config.get("callback_api_base", "")
asyncio.get_event_loop().create_task(init_url_upload_probe(callback_base))
return self.client.start(appid=self.appid, secret=self.secret)
Comment on lines +853 to 857

def get_client(self) -> botClient:
Expand Down
Loading
Loading