QQ官机适配器优化#9204
Open
u0-ani-nya wants to merge 3 commits into
Open
Conversation
dosubot
Bot
added
size:XXL
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| raise HTTPException(status_code=404) | ||
|
|
||
| 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) |
| "Set dashboard.host in data/cmd_config.json to enable remote access.\n" | ||
| ) | ||
|
|
||
| logger.info(display) |
![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 2 issues, and left some high level feedback:
- The dashboard startup logic in
AstrBotDashboard.runuses blocking socket checks (check_port_in_use/get_process_using_port) and psutil calls directly in the async path; consider offloading these to a thread executor or performing them before entering the event loop to avoid blocking the server startup coroutine. init_url_upload_probecurrently hardcodes a 10-second sleep and media cache directory relative toos.getcwd(); making the delay and cache path configurable (and based on existing config/path helpers) would make this probe more predictable across different deployment layouts.- The legacy
/media/{filename}endpoint indashboard/api/files.pyrecompiles the UUID filename regex on every request; extracting a precompiled pattern at module scope would reduce overhead and simplify the per-request logic.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The dashboard startup logic in `AstrBotDashboard.run` uses blocking socket checks (`check_port_in_use` / `get_process_using_port`) and psutil calls directly in the async path; consider offloading these to a thread executor or performing them before entering the event loop to avoid blocking the server startup coroutine.
- `init_url_upload_probe` currently hardcodes a 10-second sleep and media cache directory relative to `os.getcwd()`; making the delay and cache path configurable (and based on existing config/path helpers) would make this probe more predictable across different deployment layouts.
- The legacy `/media/{filename}` endpoint in `dashboard/api/files.py` recompiles the UUID filename regex on every request; extracting a precompiled pattern at module scope would reduce overhead and simplify the per-request logic.
## Individual Comments
### Comment 1
<location path="dashboard/api/files.py" line_range="18-27" />
<code_context>
+_MEDIA_CACHE_DIR = os.path.join(os.getcwd(), "data", "media_upload_cache")
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Using `os.getcwd()` for the media cache directory may break if the working directory differs from the AstrBot data path.
Both the QQOfficial upload logic and this dashboard endpoint hard-code `os.getcwd()/data/media_upload_cache`. If AstrBot is started from a different working directory or uses a configured data path (e.g. via `get_astrbot_data_path`), uploads may go to one directory while this API reads from another, leading to 404s for valid media. Please derive `_MEDIA_CACHE_DIR` from the same data-path helper or shared config so the cache location doesn’t depend on the current working directory.
Suggested implementation:
```python
from astrbot.dashboard.async_utils import run_maybe_async
from astrbot.dashboard.responses import error, ok
from astrbot.dashboard.services.chat_service import ChatService, ChatServiceError
from astrbot.dashboard.services.file_service import FileService, FileServiceError
from astrbot.utils.paths import get_astrbot_data_path
from .auth import AuthContext, require_scope
from .multipart import UploadFileAdapter
# 媒体上传缓存目录（与 qqofficial_message_event.py 一致），基于 AstrBot 数据路径
_MEDIA_CACHE_DIR = os.path.join(get_astrbot_data_path(), "media_upload_cache")
```
1. Ensure `get_astrbot_data_path` is imported from the correct module (the example uses `astrbot.utils.paths`; adjust the import path if your project defines it elsewhere).
2. Update the QQOfficial upload logic (e.g. in `qqofficial_message_event.py` or similar) to derive its media upload cache directory from `get_astrbot_data_path()` in the same way, so both the uploader and this API endpoint use the same configured data path.
</issue_to_address>
### Comment 2
<location path="astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py" line_range="85" />
<code_context>
botpy.errors.ServerError,
)
+# URL 上传模式可用性缓存
+_url_upload_available: bool | None = None
+_media_upload_base_url: str = ""
</code_context>
<issue_to_address>
**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:
```python
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:
```python
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:
```python
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`:
```python
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:
```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:
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:
```python
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:
```python
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 “无需清理 / 兜底清理”.
</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
+18
to
+27
| _MEDIA_CACHE_DIR = os.path.join(os.getcwd(), "data", "media_upload_cache") | ||
|
|
||
| router = APIRouter(tags=["Files"]) | ||
| legacy_router = APIRouter(prefix="/api", include_in_schema=False) | ||
|
|
||
|
|
||
| def get_service(request: Request) -> FileService: | ||
| return request.app.state.services.files | ||
|
|
||
|
|
Contributor
There was a problem hiding this comment.
suggestion (bug_risk): Using os.getcwd() for the media cache directory may break if the working directory differs from the AstrBot data path.
Both the QQOfficial upload logic and this dashboard endpoint hard-code os.getcwd()/data/media_upload_cache. If AstrBot is started from a different working directory or uses a configured data path (e.g. via get_astrbot_data_path), uploads may go to one directory while this API reads from another, leading to 404s for valid media. Please derive _MEDIA_CACHE_DIR from the same data-path helper or shared config so the cache location doesn’t depend on the current working directory.
Suggested implementation:
from astrbot.dashboard.async_utils import run_maybe_async
from astrbot.dashboard.responses import error, ok
from astrbot.dashboard.services.chat_service import ChatService, ChatServiceError
from astrbot.dashboard.services.file_service import FileService, FileServiceError
from astrbot.utils.paths import get_astrbot_data_path
from .auth import AuthContext, require_scope
from .multipart import UploadFileAdapter
# 媒体上传缓存目录（与 qqofficial_message_event.py 一致），基于 AstrBot 数据路径
_MEDIA_CACHE_DIR = os.path.join(get_astrbot_data_path(), "media_upload_cache")- Ensure
get_astrbot_data_pathis imported from the correct module (the example usesastrbot.utils.paths; adjust the import path if your project defines it elsewhere). - Update the QQOfficial upload logic (e.g. in
qqofficial_message_event.pyor similar) to derive its media upload cache directory fromget_astrbot_data_path()in the same way, so both the uploader and this API endpoint use the same configured data path.
| botpy.errors.ServerError, | ||
| ) | ||
|
|
||
| # URL 上传模式可用性缓存 |
Contributor
There was a problem hiding this comment.
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:
- Encapsulating URL-upload state and helpers
- Unifying retry logic for URL vs Base64
- Flattening
upload_group_and_c2c_mediainto 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_retrydecorator 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 mediaHelper 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 NoneThis 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 “无需清理 / 兜底清理”.
![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 dashboard server using Hypercorn, adds file-handling endpoints, and implements a URL upload mode with Base64 fallback for QQ official media files. Key feedback points out that the message reply fallback and upload retry mechanisms are broken due to the accidental removal of msg_id popping and APIReturnNoneError raising. Additionally, it is recommended to wrap the synchronous shutil.copy2 call in asyncio.to_thread to prevent blocking the event loop, and to use a direct bind attempt instead of connect_ex for more reliable port-in-use detection.
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
553
to
557
| 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] 使用主动发送接口发送成功。") |
Contributor
There was a problem hiding this comment.
在重构中移除了 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] 使用主动发送接口发送成功。') |
Comment on lines
+773
to
+774
| @_qqofficial_retry() | ||
| async def _do_upload(): |
Contributor
There was a problem hiding this comment.
与图片上传类似，这里在重构 _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 |
Comment on lines
+682
to
+686
| try: | ||
| shutil.copy2(file_source, temp_path) | ||
| except Exception as e: | ||
| logger.warning(f"[QQOfficial] 创建临时文件失败: {e}") | ||
| return None, None |
Contributor
There was a problem hiding this comment.
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 |
Comment on lines
+432
to
+447
| def check_port_in_use(self, port: int) -> bool: | ||
| """跨平台检测端口是否被占用""" | ||
| try: | ||
| # 创建 IPv4 TCP Socket | ||
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | ||
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) | ||
| # 设置超时时间 | ||
| sock.settimeout(2) | ||
| result = sock.connect_ex(("127.0.0.1", port)) | ||
| sock.close() | ||
| # result 为 0 表示端口被占用 | ||
| return result == 0 | ||
| except Exception as e: | ||
| logger.warning(f"检查端口 {port} 时发生错误: {e!s}") | ||
| # 如果出现异常，保守起见认为端口可能被占用 | ||
| return True |
Contributor
There was a problem hiding this comment.
使用 connect_ex 来检测端口是否被占用是不够可靠的。它只能检测当前是否有服务在 127.0.0.1 上监听并接受连接。如果端口处于 TIME_WAIT 状态，或者 socket 已绑定但未监听，connect_ex 可能会返回非零值（表示未占用），但随后尝试 bind 时仍会抛出 Address already in use 错误。更可靠的做法是直接尝试 bind 该端口，若抛出 OSError 则说明端口已被占用。
Suggested change
| def check_port_in_use(self, port: int) -> bool: | |
| """跨平台检测端口是否被占用""" | |
| try: | |
| # 创建 IPv4 TCP Socket | |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) | |
| # 设置超时时间 | |
| sock.settimeout(2) | |
| result = sock.connect_ex(("127.0.0.1", port)) | |
| sock.close() | |
| # result 为 0 表示端口被占用 | |
| return result == 0 | |
| except Exception as e: | |
| logger.warning(f"检查端口 {port} 时发生错误: {e!s}") | |
| # 如果出现异常，保守起见认为端口可能被占用 | |
| return True | |
| def check_port_in_use(self, port: int) -> bool: | |
| """跨平台检测端口是否被占用""" | |
| try: | |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) | |
| sock.bind(('127.0.0.1', port)) | |
| sock.close() | |
| return False | |
| except OSError: | |
| return True | |
| except Exception as e: | |
| logger.warning(f'检查端口 {port} 时发生错误: {e!s}') | |
| return True |
Contributor
There was a problem hiding this comment.
Pull request overview
This PR aims to improve the QQ Official platform adapter (adding URL-based media upload with fallback) and introduce/extend dashboard-side file/media serving to support platform downloads.
Changes:
- Add URL-upload probing + URL-first upload flow for QQ Official media, with Base64 fallback and temp-file caching.
- Add a legacy dashboard endpoint intended to serve cached media files (
/api/media/{filename}) for QQ/platform downloads. - Introduce a FastAPI/Hypercorn-based dashboard server implementation (as added under repo-root
dashboard/).
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 9 comments.
| File | Description |
|---|---|
| dashboard/server.py | Adds a dashboard server module under repo-root dashboard/ (currently appears to be an accidental duplicate / not importable as wired). |
| dashboard/api/files.py | Adds file APIs and a legacy /api/media/{filename} endpoint (currently not wired into the actual backend ASGI app). |
| astrbot/core/platform/sources/qqofficial/qqofficial_platform_adapter.py | Schedules URL-upload probing on adapter startup (needs a safer scheduling mechanism than get_event_loop().create_task). |
| astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py | Implements URL upload probing, temp-file generation/cleanup, and URL-first upload with Base64 fallback (several correctness issues found). |
Comment on lines
+36
to
+38
| from .api.app import create_dashboard_asgi_app | ||
| from .plugin_page_auth import PluginPageAuth | ||
| from .services.auth_service import DASHBOARD_JWT_COOKIE_NAME |
Comment on lines
+14
to
+16
| from .auth import AuthContext, require_scope | ||
| from .multipart import UploadFileAdapter | ||
|
|
Comment on lines
+853
to
857
| # 启动时探测 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
553
to
556
| if payload.get("msg_id"): | ||
| fallback_payload = payload.copy() | ||
| fallback_payload.pop("msg_id", None) | ||
| try: | ||
| ret = await send_func(fallback_payload) |
| try: | ||
| async with aiohttp.ClientSession() as session: | ||
| async with session.get( | ||
| base_url, |
| await asyncio.sleep(10) | ||
|
|
||
| # 清理上次遗留的临时文件 | ||
| temp_dir = os.path.join(os.getcwd(), "data", "media_upload_cache") |
|
|
||
| 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") |
| for _attempt in range(3): | ||
| try: | ||
| url_result = await self.bot.api._http.request(route, json=payload) | ||
| break |
Comment on lines
+773
to
+775
| @_qqofficial_retry() | ||
| async def _do_upload(): | ||
| return await self.bot.api._http.request(route, json=payload) |
Contributor
Author
|
虽说比较屎山代码 不过它似乎可以跑（x 嘛 反正qq官机好久没人修了（） |
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.pydashboard/api/files.py/api/media/{filename}端点dashboard/server.py/api/media用户配置
在 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.
/ 我的更改没有引入恶意代码。