feat: 为QQ官机适配大文件和视频能力#9218
Open
u0-ani-nya wants to merge 2 commits into
Open
Conversation
dosubot
Bot
added
the
size:XL
dosubot
Bot
added
the
area:platform
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
|
|
||
| media_cache_dir = os.path.join(os.getcwd(), "data", "media_upload_cache") | ||
| file_path = os.path.join(media_cache_dir, filename) | ||
| if not os.path.isfile(file_path): |
| if not os.path.isfile(file_path): | ||
| raise HTTPException(status_code=404) | ||
|
|
||
| return FileResponse(file_path) |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 3 issues, and left some high level feedback:
- The new chunked upload implementation opens a fresh
aiohttp.ClientSessionand loads each part fully into memory for every PUT; consider reusing a single session and streaming file chunks to reduce overhead and memory footprint on large files. - Temporary media files in
data/media_upload_cacheare only cleaned on startup; for long-running instances you may want a periodic cleanup or TTL-based deletion to avoid unbounded disk usage. - Most of the new '[QQOfficial][DEBUG]' logs are emitted at
INFOlevel, which could be noisy in production; consider moving these toDEBUGor reducing verbosity once the feature stabilizes.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The new chunked upload implementation opens a fresh `aiohttp.ClientSession` and loads each part fully into memory for every PUT; consider reusing a single session and streaming file chunks to reduce overhead and memory footprint on large files.
- Temporary media files in `data/media_upload_cache` are only cleaned on startup; for long-running instances you may want a periodic cleanup or TTL-based deletion to avoid unbounded disk usage.
- Most of the new '[QQOfficial][DEBUG]' logs are emitted at `INFO` level, which could be noisy in production; consider moving these to `DEBUG` or reducing verbosity once the feature stabilizes.
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py" line_range="972-979" />
<code_context>
+ )
+
+ # 顺序上传所有分片（确保 part_finish 按序完成）
+ logger.info(f"[QQOfficial] 分片顺序: {[int(p.get('index', 0)) for p in parts]}")
+ for part in parts:
+ try:
+ await _upload_one_part(part)
</code_context>
<issue_to_address>
**suggestion (performance):** 分片上传当前是完全串行的，`concurrency` 配置和信号量实际上未生效
当前虽然配置了 `concurrency` 并使用了 `asyncio.Semaphore`，但这里仍是简单的顺序 `for part in parts: await _upload_one_part(part)`，实际没有任何并发，上传大文件/多分片时会明显变慢。如果平台不要求 `upload_part_finish` 严格按 `index` 顺序到达，建议用 `asyncio.gather` 或基于队列的并发模式（由信号量控制最大并发数）来执行 `_upload_one_part`。若确实需要 `part_finish` 顺序，只在内部串行处理 `part_finish` 调用，保持 PUT 上传本身的并发。
```suggestion
# 并发上传所有分片（由外部 concurrency / 信号量控制最大并发）
logger.info(f"[QQOfficial] 分片顺序: {[int(p.get('index', 0)) for p in parts]}")
upload_tasks = [
asyncio.create_task(_upload_one_part(part))
for part in parts
]
results = await asyncio.gather(*upload_tasks, return_exceptions=True)
# 统一检查异常，确保任意一个分片失败都会中止整个上传流程
for idx, result in enumerate(results):
if isinstance(result, Exception):
logger.error(
f"[QQOfficial] 分片上传失败: part_index={parts[idx].get('index')}, "
f"error={result!r}"
)
return None
```
</issue_to_address>
### Comment 2
<location path="astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py" line_range="905-906" />
<code_context>
+ last_exc = None
+ for attempt in range(self._CHUNKED_PART_UPLOAD_MAX_RETRIES + 1):
+ try:
+ async with aiohttp.ClientSession() as sess:
+ async with sess.put(
+ presigned_url,
+ data=data,
</code_context>
<issue_to_address>
**suggestion (performance):** 每个分片上传都新建 `ClientSession` 可能带来较大开销，建议复用会话
当前在 `_upload_one_part` 中为每个分片创建 `aiohttp.ClientSession`，在几十到上百个分片场景下会产生大量 TCP 连接和 TLS 握手，难以复用连接。建议在 `_chunked_upload` 外层使用一次 `async with aiohttp.ClientSession() as sess:` 包裹所有分片上传，并通过参数或闭包将 `sess` 传入 `_upload_one_part`，以减少资源开销并提升上传稳定性。
Suggested implementation:
```python
async with semaphore:
# PUT 到预签名 URL
last_exc = None
for attempt in range(self._CHUNKED_PART_UPLOAD_MAX_RETRIES + 1):
try:
async with sess.put(
presigned_url,
data=data,
headers={"Content-Length": str(len(data))},
timeout=aiohttp.ClientTimeout(total=self._CHUNKED_PART_UPLOAD_TIMEOUT),
) as resp:
resp.raise_for_status()
logger.info(
f"[QQOfficial] PUT 分片 {part_index}/{len(parts)}: {resp.status} OK, size={len(data)}"
)
break
except Exception as exc:
last_exc = exc
if attempt < self._CHUNKED_PART_UPLOAD_MAX_RETRIES:
delay = 1.0 * (2 ** attempt)
```
要完全实现会话复用，需要在外层（例如 `_chunked_upload` 或调用 `_upload_one_part` 的函数）中：
1. 使用一次性的会话上下文包裹所有分片上传逻辑，例如：
```python
async with aiohttp.ClientSession() as sess:
# 在这里启动所有分片任务，并把 sess 传入
```
2. 修改 `_upload_one_part` 的函数签名，增加一个 `sess: aiohttp.ClientSession` 参数，或者通过闭包捕获 `sess`，例如：
```python
async def _upload_one_part(..., sess: aiohttp.ClientSession, ...):
...
```
3. 在创建分片任务时，传入该 `sess`：
```python
tasks.append(self._upload_one_part(..., sess=sess, ...))
```
如果当前代码结构不便于修改函数签名，可以选择在 `_chunked_upload` 内部定义一个内部协程，捕获外层的 `sess` 变量，然后替代现有的 `_upload_one_part` 调用。
</issue_to_address>
### Comment 3
<location path="astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py" line_range="91" />
<code_context>
+_media_upload_base_url: str = ""
+
+
+async def init_url_upload_probe(media_upload_url: str = "") -> None:
+ """启动时探测媒体上传地址是否可达，结果缓存。
+ 应在适配器启动时调用一次。同时清理上次遗留的临时文件。
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the QQOfficial adapter by removing or extracting unused URL-upload helpers, decomposing large upload and parsing methods into focused helpers/classes, and centralizing repeated logging to make control flow easier to follow.
- `_prepare_temp_file`, `_cleanup_temp_file`, `_probe_temp_url_reachable`, `init_url_upload_probe` and the globals `_url_upload_available`, `_media_upload_base_url` are currently unused in the upload flow. They add extra concepts (URL upload mode, temp cache directory, probe) without being exercised.
- If URL-upload is not actually supported yet in this adapter, you can reduce cognitive load by removing them for now (or moving them to a dedicated `media_url_upload.py` and wiring them in when the feature is complete).
- Example: keep only the probe you actually call, and strip temp-file logic until it’s needed:
```python
# media_url_upload.py
_url_upload_available: bool | None = None
_media_upload_base_url: str = ""
async def init_url_upload_probe(media_upload_url: str = "") -> None:
# ... as today, but only URL probe ...
```
- `upload_group_and_c2c_media` is now doing multiple things in-line (local vs remote, chunked vs base64, logging, retries), which makes the method long and harder to scan.
- You can keep the new behavior but split into small helpers so the main method is an orchestrator:
```python
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:
route, payload = self._build_media_payload(file_type, srv_send_msg, file_name, **kwargs)
if route is None:
return None
if os.path.exists(file_source) and file_type in (self.VIDEO_FILE_TYPE, self.FILE_FILE_TYPE):
media = await self._upload_via_chunked(file_source, file_type, file_name or "file", route, payload, **kwargs)
if media:
return media
logger.warning("[QQOfficial] 分片上传失败，回退 Base64")
return await self._upload_via_base64(file_source, route, payload)
def _build_media_payload(...):
# small helper: construct Route + base fields only
async def _upload_via_chunked(...):
return await self._chunked_upload(file_source, file_type, file_name, **kwargs)
async def _upload_via_base64(self, file_source: str, route: Route, payload: dict) -> 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
# reuse _qqofficial_retry wrapper here
```
- This keeps `upload_group_and_c2c_media` small and makes feature-specific paths clearer.
- `_chunked_upload` is a large, stateful method inside the event class; it mixes hash computation, part IO, retry/backoff, and API orchestration.
- Consider extracting a dedicated helper (e.g. `QQChunkedUploader`) so the event class only calls it:
```python
class QQChunkedUploader:
def __init__(self, http_client):
self._http = http_client
async def upload(self, file_path: str, file_type: int, file_name: str, **kwargs) -> Media | None:
# move _compute_file_hashes, _read_file_chunk, upload_prepare, part PUT/finish, complete_upload here
```
```python
# in QQOfficialMessageEvent.__init__
self._chunked_uploader = QQChunkedUploader(self.bot.api._http)
async def _upload_via_chunked(...):
return await self._chunked_uploader.upload(file_source, file_type, file_name, **kwargs)
```
- This keeps `QQOfficialMessageEvent` focused on messaging and isolates protocol-specific upload logic.
- The video-size downgrade logic is inlined inside `_parse_to_qqofficial`, mixing message parsing with transport policy.
- You can extract it into a small helper without changing behavior:
```python
async def _maybe_downgrade_video_to_file(
self, video_file_source: str | None, file_name: str | None
) -> tuple[str | None, str | None, str | None]:
if not video_file_source:
return None, None, file_name
file_source = None
# copy the existing size-check / HEAD logic here
# return (file_source, video_file_source, updated_file_name)
async def _parse_to_qqofficial(...):
# after resolving video_file_source, file_name:
file_source, video_file_source, file_name = await self._maybe_downgrade_video_to_file(
video_file_source, file_name
)
```
- This leaves `_parse_to_qqofficial` primarily responsible for building the message representation.
- The detailed `[DEBUG]` logging in `_post_send_one` and upload methods increases line count and makes control flow harder to visually follow.
- You can preserve the logging but move repeated patterns into small helpers:
```python
def _log_payload_state(self, tag: str, payload: dict) -> None:
logger.info(
f"[QQOfficial][DEBUG] {tag}: msg_type={payload.get('msg_type')}, "
f"has_media={payload.get('media') is not None}, "
f"has_markdown={payload.get('markdown') is not None}"
)
# usage:
self._log_payload_state("发送前 payload 状态", payload)
```
- This reduces clutter and makes the intent of each call clearer.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+972
to
+979
| # 顺序上传所有分片（确保 part_finish 按序完成） | ||
| logger.info(f"[QQOfficial] 分片顺序: {[int(p.get('index', 0)) for p in parts]}") | ||
| for part in parts: | ||
| try: | ||
| await _upload_one_part(part) | ||
| except Exception as e: | ||
| logger.error(f"[QQOfficial] 分片上传失败: {e}") | ||
| return None |
Contributor
There was a problem hiding this comment.
suggestion (performance): 分片上传当前是完全串行的，concurrency 配置和信号量实际上未生效
当前虽然配置了 concurrency 并使用了 asyncio.Semaphore，但这里仍是简单的顺序 for part in parts: await _upload_one_part(part)，实际没有任何并发，上传大文件/多分片时会明显变慢。如果平台不要求 upload_part_finish 严格按 index 顺序到达，建议用 asyncio.gather 或基于队列的并发模式（由信号量控制最大并发数）来执行 _upload_one_part。若确实需要 part_finish 顺序，只在内部串行处理 part_finish 调用，保持 PUT 上传本身的并发。
Suggested change
| # 顺序上传所有分片（确保 part_finish 按序完成） | |
| logger.info(f"[QQOfficial] 分片顺序: {[int(p.get('index', 0)) for p in parts]}") | |
| for part in parts: | |
| try: | |
| await _upload_one_part(part) | |
| except Exception as e: | |
| logger.error(f"[QQOfficial] 分片上传失败: {e}") | |
| return None | |
| # 并发上传所有分片（由外部 concurrency / 信号量控制最大并发） | |
| logger.info(f"[QQOfficial] 分片顺序: {[int(p.get('index', 0)) for p in parts]}") | |
| upload_tasks = [ | |
| asyncio.create_task(_upload_one_part(part)) | |
| for part in parts | |
| ] | |
| results = await asyncio.gather(*upload_tasks, return_exceptions=True) | |
| # 统一检查异常，确保任意一个分片失败都会中止整个上传流程 | |
| for idx, result in enumerate(results): | |
| if isinstance(result, Exception): | |
| logger.error( | |
| f"[QQOfficial] 分片上传失败: part_index={parts[idx].get('index')}, " | |
| f"error={result!r}" | |
| ) | |
| return None |
Comment on lines
+905
to
+906
| async with aiohttp.ClientSession() as sess: | ||
| async with sess.put( |
Contributor
There was a problem hiding this comment.
suggestion (performance): 每个分片上传都新建 ClientSession 可能带来较大开销，建议复用会话
当前在 _upload_one_part 中为每个分片创建 aiohttp.ClientSession，在几十到上百个分片场景下会产生大量 TCP 连接和 TLS 握手，难以复用连接。建议在 _chunked_upload 外层使用一次 async with aiohttp.ClientSession() as sess: 包裹所有分片上传，并通过参数或闭包将 sess 传入 _upload_one_part，以减少资源开销并提升上传稳定性。
Suggested implementation:
async with semaphore:
# PUT 到预签名 URL
last_exc = None
for attempt in range(self._CHUNKED_PART_UPLOAD_MAX_RETRIES + 1):
try:
async with sess.put(
presigned_url,
data=data,
headers={"Content-Length": str(len(data))},
timeout=aiohttp.ClientTimeout(total=self._CHUNKED_PART_UPLOAD_TIMEOUT),
) as resp:
resp.raise_for_status()
logger.info(
f"[QQOfficial] PUT 分片 {part_index}/{len(parts)}: {resp.status} OK, size={len(data)}"
)
break
except Exception as exc:
last_exc = exc
if attempt < self._CHUNKED_PART_UPLOAD_MAX_RETRIES:
delay = 1.0 * (2 ** attempt)要完全实现会话复用，需要在外层（例如 _chunked_upload 或调用 _upload_one_part 的函数）中：
- 使用一次性的会话上下文包裹所有分片上传逻辑，例如：
async with aiohttp.ClientSession() as sess: # 在这里启动所有分片任务，并把 sess 传入
- 修改
_upload_one_part的函数签名，增加一个sess: aiohttp.ClientSession参数，或者通过闭包捕获sess，例如：async def _upload_one_part(..., sess: aiohttp.ClientSession, ...): ...
- 在创建分片任务时，传入该
sess：tasks.append(self._upload_one_part(..., sess=sess, ...))
如果当前代码结构不便于修改函数签名，可以选择在 _chunked_upload 内部定义一个内部协程，捕获外层的 sess 变量，然后替代现有的 _upload_one_part 调用。
| _media_upload_base_url: str = "" | ||
|
|
||
|
|
||
| async def init_url_upload_probe(media_upload_url: str = "") -> None: |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider simplifying the QQOfficial adapter by removing or extracting unused URL-upload helpers, decomposing large upload and parsing methods into focused helpers/classes, and centralizing repeated logging to make control flow easier to follow.
-
_prepare_temp_file,_cleanup_temp_file,_probe_temp_url_reachable,init_url_upload_probeand the globals_url_upload_available,_media_upload_base_urlare currently unused in the upload flow. They add extra concepts (URL upload mode, temp cache directory, probe) without being exercised.- If URL-upload is not actually supported yet in this adapter, you can reduce cognitive load by removing them for now (or moving them to a dedicated
media_url_upload.pyand wiring them in when the feature is complete). - Example: keep only the probe you actually call, and strip temp-file logic until it’s needed:
# media_url_upload.py _url_upload_available: bool | None = None _media_upload_base_url: str = "" async def init_url_upload_probe(media_upload_url: str = "") -> None: # ... as today, but only URL probe ...
- If URL-upload is not actually supported yet in this adapter, you can reduce cognitive load by removing them for now (or moving them to a dedicated
-
upload_group_and_c2c_mediais now doing multiple things in-line (local vs remote, chunked vs base64, logging, retries), which makes the method long and harder to scan.- You can keep the new behavior but split into small helpers so the main method is an orchestrator:
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: route, payload = self._build_media_payload(file_type, srv_send_msg, file_name, **kwargs) if route is None: return None if os.path.exists(file_source) and file_type in (self.VIDEO_FILE_TYPE, self.FILE_FILE_TYPE): media = await self._upload_via_chunked(file_source, file_type, file_name or "file", route, payload, **kwargs) if media: return media logger.warning("[QQOfficial] 分片上传失败，回退 Base64") return await self._upload_via_base64(file_source, route, payload) def _build_media_payload(...): # small helper: construct Route + base fields only async def _upload_via_chunked(...): return await self._chunked_upload(file_source, file_type, file_name, **kwargs) async def _upload_via_base64(self, file_source: str, route: Route, payload: dict) -> 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 # reuse _qqofficial_retry wrapper here
- This keeps
upload_group_and_c2c_mediasmall and makes feature-specific paths clearer.
- You can keep the new behavior but split into small helpers so the main method is an orchestrator:
-
_chunked_uploadis a large, stateful method inside the event class; it mixes hash computation, part IO, retry/backoff, and API orchestration.-
Consider extracting a dedicated helper (e.g.
QQChunkedUploader) so the event class only calls it:class QQChunkedUploader: def __init__(self, http_client): self._http = http_client async def upload(self, file_path: str, file_type: int, file_name: str, **kwargs) -> Media | None: # move _compute_file_hashes, _read_file_chunk, upload_prepare, part PUT/finish, complete_upload here
# in QQOfficialMessageEvent.__init__ self._chunked_uploader = QQChunkedUploader(self.bot.api._http) async def _upload_via_chunked(...): return await self._chunked_uploader.upload(file_source, file_type, file_name, **kwargs)
-
This keeps
QQOfficialMessageEventfocused on messaging and isolates protocol-specific upload logic.
-
-
The video-size downgrade logic is inlined inside
_parse_to_qqofficial, mixing message parsing with transport policy.- You can extract it into a small helper without changing behavior:
async def _maybe_downgrade_video_to_file( self, video_file_source: str | None, file_name: str | None ) -> tuple[str | None, str | None, str | None]: if not video_file_source: return None, None, file_name file_source = None # copy the existing size-check / HEAD logic here # return (file_source, video_file_source, updated_file_name) async def _parse_to_qqofficial(...): # after resolving video_file_source, file_name: file_source, video_file_source, file_name = await self._maybe_downgrade_video_to_file( video_file_source, file_name )
- This leaves
_parse_to_qqofficialprimarily responsible for building the message representation.
- You can extract it into a small helper without changing behavior:
-
The detailed
[DEBUG]logging in_post_send_oneand upload methods increases line count and makes control flow harder to visually follow.- You can preserve the logging but move repeated patterns into small helpers:
def _log_payload_state(self, tag: str, payload: dict) -> None: logger.info( f"[QQOfficial][DEBUG] {tag}: msg_type={payload.get('msg_type')}, " f"has_media={payload.get('media') is not None}, " f"has_markdown={payload.get('markdown') is not None}" ) # usage: self._log_payload_state("发送前 payload 状态", payload)
- This reduces clutter and makes the intent of each call clearer.
- You can preserve the logging but move repeated patterns into small helpers:
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request introduces a chunked upload mechanism for the QQOfficial platform adapter to support large files and videos, along with a fallback to Base64 upload and video size checks to downgrade large videos to file sending. The reviewer feedback highlights several critical areas for improvement: first, the chunked upload logic needs to handle the 'Instant Upload' (秒传) scenario where the prepare API directly returns a file UUID; second, the chunked upload currently runs serially and should be refactored to run in parallel using asyncio.gather with proper memory management and accompanied by unit tests; finally, several unused components added for a temporary URL upload scheme—including file preparation/cleanup utilities, startup probe logic, and a FastAPI route—should be removed as dead code.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
Comment on lines
+853
to
+858
| upload_id = prepare_result.get("upload_id", "") | ||
| block_size = int(prepare_result.get("block_size", 0)) | ||
| parts = prepare_result.get("parts", []) | ||
| if not upload_id or not parts: | ||
| logger.error(f"[QQOfficial] upload_prepare 缺少必要字段: {prepare_result}") | ||
| return None |
Contributor
There was a problem hiding this comment.
未处理秒传（Instant Upload）逻辑导致上传失败
在 QQ 机器人官方 API 的分片上传流程中，如果文件之前已经被上传过（秒传场景），/upload_prepare 接口会直接返回已完成的文件信息（包含 file_uuid 和 file_info），而不会返回 upload_id 或 parts。
但在当前实现中，代码强制校验了 upload_id 和 parts。这会导致在触发秒传时，代码会误判为“缺少必要字段”并返回 None 导致上传失败，进而回退到 Base64 上传（如果文件超大还会触发 413 错误）。
建议改进：
在校验 upload_id 之前，先检查 prepare_result 中是否已包含 file_uuid。如果存在，则直接返回 Media 对象，实现秒传。
Suggested change
| upload_id = prepare_result.get("upload_id", "") | |
| block_size = int(prepare_result.get("block_size", 0)) | |
| parts = prepare_result.get("parts", []) | |
| if not upload_id or not parts: | |
| logger.error(f"[QQOfficial] upload_prepare 缺少必要字段: {prepare_result}") | |
| return None | |
| if "file_uuid" in prepare_result: | |
| logger.info(f"[QQOfficial] 分片上传: 触发秒传成功! file_uuid={prepare_result['file_uuid']}") | |
| return Media( | |
| file_uuid=prepare_result["file_uuid"], | |
| file_info=prepare_result.get("file_info", ""), | |
| ttl=prepare_result.get("ttl", 0), | |
| ) | |
| upload_id = prepare_result.get("upload_id", "") | |
| block_size = int(prepare_result.get("block_size", 0)) | |
| parts = prepare_result.get("parts", []) | |
| if not upload_id or not parts: | |
| logger.error(f"[QQOfficial] upload_prepare 缺少必要字段: {prepare_result}") | |
| return None |
Comment on lines
+882
to
+979
| async def _upload_one_part(part: dict): | ||
| nonlocal completed_parts | ||
| part_index = int(part.get("index", 0)) | ||
| presigned_url = part.get("presigned_url", "") | ||
| part_block_size = int(part.get("block_size", 0)) or block_size | ||
| offset = (part_index - 1) * block_size | ||
| length = min(part_block_size, file_size - offset) | ||
|
|
||
| if not presigned_url: | ||
| logger.error(f"[QQOfficial] 分片 {part_index} 缺少 presigned_url") | ||
| return | ||
|
|
||
| # 读取分片数据 | ||
| data = await asyncio.get_running_loop().run_in_executor( | ||
| None, self._read_file_chunk, file_source, offset, length | ||
| ) | ||
| part_md5 = hashlib.md5(data).hexdigest() | ||
|
|
||
| async with semaphore: | ||
| # PUT 到预签名 URL | ||
| last_exc = None | ||
| for attempt in range(self._CHUNKED_PART_UPLOAD_MAX_RETRIES + 1): | ||
| try: | ||
| async with aiohttp.ClientSession() as sess: | ||
| async with sess.put( | ||
| presigned_url, | ||
| data=data, | ||
| headers={"Content-Length": str(len(data))}, | ||
| timeout=aiohttp.ClientTimeout(total=self._CHUNKED_PART_UPLOAD_TIMEOUT), | ||
| ) as resp: | ||
| resp.raise_for_status() | ||
| logger.info(f"[QQOfficial] PUT 分片 {part_index}/{len(parts)}: {resp.status} OK, size={len(data)}") | ||
| break | ||
| except Exception as exc: | ||
| last_exc = exc | ||
| if attempt < self._CHUNKED_PART_UPLOAD_MAX_RETRIES: | ||
| delay = 1.0 * (2 ** attempt) | ||
| logger.warning(f"[QQOfficial] PUT 分片 {part_index} 失败, 重试 {attempt+1}: {exc}") | ||
| await asyncio.sleep(delay) | ||
| else: | ||
| logger.error(f"[QQOfficial] PUT 分片 {part_index} 最终失败: {last_exc}") | ||
| raise RuntimeError(f"Part {part_index} upload failed: {last_exc}") | ||
|
|
||
| # 通知平台分片完成 | ||
| # 发送 part 的 block_size（来自 prepare 响应），不是实际数据长度 | ||
| part_finish_block_size = int(part.get("block_size", 0)) or block_size | ||
| part_finish_payload = { | ||
| "upload_id": upload_id, | ||
| "part_index": part_index, | ||
| "block_size": part_finish_block_size, | ||
| "md5": part_md5, | ||
| } | ||
| if openid: | ||
| finish_route = Route("POST", "/v2/users/{openid}/upload_part_finish", openid=openid) | ||
| else: | ||
| finish_route = Route("POST", "/v2/groups/{group_openid}/upload_part_finish", group_openid=group_openid) | ||
|
|
||
| logger.info( | ||
| f"[QQOfficial] part_finish 请求: part={part_index}, block_size={part_finish_block_size}, length={length}, md5={part_md5[:16]}..." | ||
| ) | ||
| finish_start = asyncio.get_running_loop().time() | ||
| finish_attempt = 0 | ||
| max_finish_retries = 5 | ||
| while True: | ||
| try: | ||
| await self.bot.api._http.request(finish_route, json=part_finish_payload) | ||
| break | ||
| except Exception as exc: | ||
| err_msg = str(exc) | ||
| finish_attempt += 1 | ||
| elapsed = asyncio.get_running_loop().time() - finish_start | ||
| # 40093001 = retryable (平台侧还没收到分片)，其他错误也重试但次数更少 | ||
| if "40093001" in err_msg: | ||
| if elapsed >= retry_timeout: | ||
| raise RuntimeError(f"part_finish timeout after {elapsed:.0f}s: {exc}") from exc | ||
| logger.debug(f"[QQOfficial] part_finish retryable(40093001), attempt {finish_attempt}") | ||
| await asyncio.sleep(self._CHUNKED_PART_FINISH_RETRY_INTERVAL) | ||
| else: | ||
| if finish_attempt >= max_finish_retries: | ||
| raise RuntimeError(f"part_finish failed after {finish_attempt} attempts: {exc}") from exc | ||
| delay = 1.0 * (2 ** (finish_attempt - 1)) | ||
| logger.warning(f"[QQOfficial] part_finish 失败, 重试 {finish_attempt}/{max_finish_retries}: {exc}") | ||
| await asyncio.sleep(delay) | ||
|
|
||
| completed_parts += 1 | ||
| logger.info( | ||
| f"[QQOfficial] 分片 {part_index}/{len(parts)} 完成 " | ||
| f"({completed_parts}/{len(parts)} done)" | ||
| ) | ||
|
|
||
| # 顺序上传所有分片（确保 part_finish 按序完成） | ||
| logger.info(f"[QQOfficial] 分片顺序: {[int(p.get('index', 0)) for p in parts]}") | ||
| for part in parts: | ||
| try: | ||
| await _upload_one_part(part) | ||
| except Exception as e: | ||
| logger.error(f"[QQOfficial] 分片上传失败: {e}") | ||
| return None |
Contributor
There was a problem hiding this comment.
分片上传实际上是串行执行的，未实现真正的并发上传
在当前的实现中，虽然解析了 concurrency 并创建了 semaphore，但在上传分片时却使用了 for part in parts: await _upload_one_part(part)。这导致分片是完全串行上传的，concurrency 和 semaphore 并没有起到并发控制的作用，极大限制了大文件的上传速度。
另外，如果直接改为并发上传，当前的 _read_file_chunk 处于 semaphore 外部，这会导致所有分片数据同时被读取到内存中，造成内存占用过高。
建议改进：
- 将读取分片和计算 MD5 的逻辑移入
async with semaphore:内部，以限制内存中同时存在的块数据。 - 使用
asyncio.gather并发执行分片上传任务，并在发生异常时及时取消其他未完成的任务。 - 此外，新引入的分片上传功能应当伴随相应的单元测试，请确保为其编写对应的单元测试以验证其正确性。
async def _upload_one_part(part: dict):
nonlocal completed_parts
part_index = int(part.get("index", 0))
presigned_url = part.get("presigned_url", "")
part_block_size = int(part.get("block_size", 0)) or block_size
offset = (part_index - 1) * block_size
length = min(part_block_size, file_size - offset)
if not presigned_url:
logger.error(f"[QQOfficial] 分片 {part_index} 缺少 presigned_url")
raise ValueError(f"Missing presigned_url for part {part_index}")
async with semaphore:
# 在信号量内部读取分片，防止并发读取过多分片导致内存溢出
data = await asyncio.get_running_loop().run_in_executor(
None, self._read_file_chunk, file_source, offset, length
)
part_md5 = hashlib.md5(data).hexdigest()
# PUT 到预签名 URL
last_exc = None
for attempt in range(self._CHUNKED_PART_UPLOAD_MAX_RETRIES + 1):
try:
async with aiohttp.ClientSession() as sess:
async with sess.put(
presigned_url,
data=data,
headers={"Content-Length": str(len(data))},
timeout=aiohttp.ClientTimeout(total=self._CHUNKED_PART_UPLOAD_TIMEOUT),
) as resp:
resp.raise_for_status()
logger.info(f"[QQOfficial] PUT 分片 {part_index}/{len(parts)}: {resp.status} OK, size={len(data)}")
break
except Exception as exc:
last_exc = exc
if attempt < self._CHUNKED_PART_UPLOAD_MAX_RETRIES:
delay = 1.0 * (2 ** attempt)
logger.warning(f"[QQOfficial] PUT 分片 {part_index} 失败, 重试 {attempt+1}: {exc}")
await asyncio.sleep(delay)
else:
logger.error(f"[QQOfficial] PUT 分片 {part_index} 最终失败: {last_exc}")
raise RuntimeError(f"Part {part_index} upload failed: {last_exc}")
# 通知平台分片完成
part_finish_block_size = int(part.get("block_size", 0)) or block_size
part_finish_payload = {
"upload_id": upload_id,
"part_index": part_index,
"block_size": part_finish_block_size,
"md5": part_md5,
}
if openid:
finish_route = Route("POST", "/v2/users/{openid}/upload_part_finish", openid=openid)
else:
finish_route = Route("POST", "/v2/groups/{group_openid}/upload_part_finish", group_openid=group_openid)
logger.info(
f"[QQOfficial] part_finish 请求: part={part_index}, block_size={part_finish_block_size}, length={length}, md5={part_md5[:16]}..."
)
finish_start = asyncio.get_running_loop().time()
finish_attempt = 0
max_finish_retries = 5
while True:
try:
await self.bot.api._http.request(finish_route, json=part_finish_payload)
break
except Exception as exc:
err_msg = str(exc)
finish_attempt += 1
elapsed = asyncio.get_running_loop().time() - finish_start
if "40093001" in err_msg:
if elapsed >= retry_timeout:
raise RuntimeError(f"part_finish timeout after {elapsed:.0f}s: {exc}") from exc
logger.debug(f"[QQOfficial] part_finish retryable(40093001), attempt {finish_attempt}")
await asyncio.sleep(self._CHUNKED_PART_FINISH_RETRY_INTERVAL)
else:
if finish_attempt >= max_finish_retries:
raise RuntimeError(f"part_finish failed after {finish_attempt} attempts: {exc}") from exc
delay = 1.0 * (2 ** (finish_attempt - 1))
logger.warning(f"[QQOfficial] part_finish 失败, 重试 {finish_attempt}/{max_finish_retries}: {exc}")
await asyncio.sleep(delay)
completed_parts += 1
logger.info(
f"[QQOfficial] 分片 {part_index}/{len(parts)} 完成 "
f"({completed_parts}/{len(parts)} done)"
)
# 并发上传所有分片
logger.info(f"[QQOfficial] 开始并发上传分片，总数: {len(parts)}")
tasks = [asyncio.create_task(_upload_one_part(part)) for part in parts]
try:
await asyncio.gather(*tasks)
except Exception as e:
logger.error(f"[QQOfficial] 分片上传失败: {e}")
for t in tasks:
if not t.done():
t.cancel()
return NoneReferences
- New functionality, such as handling attachments, should be accompanied by corresponding unit tests.
Comment on lines
+708
to
+743
| 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: | ||
| 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})") |
Contributor
There was a problem hiding this comment.
发现未使用的死代码 (Dead Code)
在当前的 PR 实现中，您最终采用了官方的分片上传 (Chunked Upload) 协议（通过 _chunked_upload 实现），这比通过 Web Server 生成临时 URL 让 QQ 服务器下载的方案更加优雅和安全。
但是，您在代码中残留了大量与“Web Server 临时 URL 方案”相关的死代码，这些代码在分片上传模式下完全没有被调用：
_prepare_temp_file(708-733 行) 和_cleanup_temp_file(735-743 行) 从未被调用。init_url_upload_probe(91-134 行) 探测并设置了_url_upload_available，但该变量仅在未被调用的_prepare_temp_file中使用，因此整个探测逻辑也是多余的。dashboard/api/files.py中的/api/media/{filename}路由也变成了死路由。
为了保持代码库的整洁 and 可维护性，建议将这些未使用的死代码、探测逻辑以及对应的 API 路由全部删除。
Comment on lines
+853
to
+856
| # 启动时探测 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)) |
Contributor
There was a problem hiding this comment.
移除未使用的 URL 上传探测逻辑
由于“Web Server 临时 URL 方案”已被弃用并改为官方的“分片上传方案”，在启动时调用 init_url_upload_probe 探测 URL 可用性已无必要。建议将其移除以简化启动流程并减少不必要的后台任务。
Comment on lines
+124
to
+136
| @legacy_router.get("/media/{filename}") | ||
| async def get_media_cache_file(filename: str): | ||
| """按 UUID 文件名直接服务媒体缓存文件（无需 token，供 QQ 等平台下载）。""" | ||
| import re | ||
| if not re.match(r'^[0-9a-f]{32}\.[a-zA-Z0-9]+$', filename): | ||
| raise HTTPException(status_code=404) | ||
|
|
||
| media_cache_dir = os.path.join(os.getcwd(), "data", "media_upload_cache") | ||
| file_path = os.path.join(media_cache_dir, filename) | ||
| if not os.path.isfile(file_path): | ||
| raise HTTPException(status_code=404) | ||
|
|
||
| return FileResponse(file_path) |
Contributor
There was a problem hiding this comment.
移除未使用的媒体缓存服务路由
由于“Web Server 临时 URL 方案”已被弃用，此 /media/{filename} 路由已无任何调用方，属于死代码。建议将其移除以保持 API 路由的整洁。
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Modifications / 改动点
问题
QQ 官方机器人适配器上传视频/文件时，将文件读入内存后 Base64 编码塞进 JSON POST，体积膨胀约 33%。
当文件较大时（如 20MB+），膨胀后的请求体超过腾讯 STGW 的 body 限制，触发
413 Request Entity Too Large。根因：适配器使用
base64.b64encode(file_content)后放入 JSON 的file_data字段，而 QQ 客户端走的是分片上传协议，两者链路完全不同。方案
复用 AstrBot 已有的 Web Server 能力，为大文件生成临时公网 URL，让 QQ 服务器直接下载文件，避免 Base64 膨胀。
核心流程
渐进降级
callback_api_base未配置改动文件
qqofficial_message_event.pyqqofficial_platform_adapter.pyastrbot/dashboard/api/files.py用户配置
在 AstrBot 设置中填写已有的
callback_api_base（对外可达的回调接口地址），如：http://your-ip:6185https://your-domain.com留空则使用 Base64 上传（原有行为不变），无需新增配置项。
Screenshots or Test Results / 运行截图或测试结果
[2026-07-10 19:41:54.734] [Core]
[INFO]
[core.event_bus:79]: [default] [QQ(qq_official)] 7BCDD80: [At:qq_official] https://x.com/Mik*********47733032?s=20
[2026-07-10 19:41:54.740] [Plug]
[INFO]
[llm_debug_logger.main:45]: [LLM-Debug] #7 ① 管道入口 [私聊] is_at=True text=https://x.com/Mik*********47733032?s=20
[2026-07-10 19:41:54.740] [Plug]
[INFO]
[llm_debug_logger.main:55]: [LLM-Debug] #7 📦 raw_message type: PatchedC2CMessage
[2026-07-10 19:41:54.740] [Plug]
[INFO]
[llm_debug_logger.main:109]: [LLM-Debug] #7 📦 raw_message 属性:
{
"attachments": "[]",
"author": "<_User: {'user_openid': '7BCD****************D80'}>",
"content": "https://x.com/Mik*********47733032?s=20 ",
"event_id": "C2C_MESSAGE_CREATE:ruuuxsoz2",
"id": "ROBOT1.0_8GsQWq3Z7lV7XL-4!",
"mentions": "[]",
"message_reference": "<_MessageRef: {'message_id': None}>",
"message_type": 0,
"msg_elements": "[]",
"msg_seq": null,
"raw_data": "{'author': {'bot': False, 'id': '7BCDD80', 'union_openid': '', 'user_openid': '7BCDD80', 'username': ''}, 'content': 'https://x.com/Mik*********47733032?s=20 ', 'id': 'ROBOT1.0_8GsQuJqP057XzXTL-4!', 'message_scene': {'ext': ['msg_idx=REFIDX_Gcq/fk*******ppK6Gc='], 'source': 'default'}, 'message_type': 0, 'timestamp'",
"timestamp": "2026-07-11T01:41:53+08:00"
}
[2026-07-10 19:41:54.740] [Plug]
[INFO]
[llm_debug_logger.main:198]: [LLM-Debug] #7 🔗 提取到链接 (content): https://x.com/Mik*********47733032?s=20
[2026-07-10 19:41:57.654] [Core]
[INFO]
[qqofficial.qqofficial_message_event:689]: [QQOfficial] 临时文件: /AstrBot/data/media_upload_cache/dafdaf063f474f66cf.mp4
[2026-07-10 19:41:57.654] [Core]
[INFO]
[qqofficial.qqofficial_message_event:690]: [QQOfficial] 下载 URL: https://website/api/media/dafdaf063f474ef4***********66cf.mp4
[2026-07-10 19:42:08.400] [Core]
[WARN]
[v4.26.5] [qqofficial.qqofficial_message_event:747]: [QQOfficial] URL 上传失败 (系统繁忙，请稍后重试)，重试...
[2026-07-10 19:42:17.598] [Core]
[INFO]
[qqofficial.qqofficial_message_event:751]: [QQOfficial] URL 上传成功: /AstrBot/data/plugin_data/astrbot_plugin_media_parser/cache/twitter_af497ff6_1783705314_536deb53/video_0.mp4
[2026-07-10 19:42:17.603] [Core]
[INFO]
[qqofficial.qqofficial_message_event:699]: [QQOfficial] 已清理临时文件: /AstrBot/data/media_upload_cache/dafdaf063f474ef48ec26f32a34f66cf.mp4
Checklist / 检查清单
😊 If there are new features added in the PR, I have discussed it with the authors through issues/emails, etc.
/ 如果 PR 中有新加入的功能，已经通过 Issue / 邮件等方式和作者讨论过。
👀 My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
/ 我的更改经过了良好的测试，并已在上方提供了“验证步骤”和“运行截图”。
🤓 I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations in
requirements.txtandpyproject.toml./ 我确保没有引入新依赖库，或者引入了新依赖库的同时将其添加到
requirements.txt和pyproject.toml文件相应位置。😮 My changes do not introduce malicious code.
/ 我的更改没有引入恶意代码。
Summary by Sourcery
Add URL-based and chunked upload support for large media in the QQ official adapter while keeping a Base64 fallback path.
New Features:
Enhancements: