fix(aiocqhttp): resolve bare record file references#9219
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a mechanism to resolve file-only OneBot record segments via the get_record action when direct media sources are not present, accompanied by comprehensive unit tests. The review feedback is highly constructive and suggests refining the legacy base64 detection by checking for audio magic headers to avoid false positives on hex-like filenames, catching ValueError alongside OSError during path checks to ensure Windows compatibility, and adding a corresponding unit test for the hex-like filename scenario.
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.
| if not has_direct_source: | ||
| compact = "".join(file_ref.split()) | ||
| padding = len(compact) % 4 | ||
| if padding: | ||
| compact += "=" * (4 - padding) | ||
| try: | ||
| has_direct_source = bool(base64.b64decode(compact, validate=True)) | ||
| except (binascii.Error, ValueError): | ||
| pass |
There was a problem hiding this comment.
Detecting legacy bare base64 payloads solely by attempting to decode them can lead to false positives. For example, a bare filename that is a hex string (such as an MD5 hash like 0d2bb1468a87d64414f8e563cc61c33c or a UUID) contains only valid base64 characters and has a length divisible by 4. It will successfully decode, causing has_direct_source to be incorrectly set to True, which bypasses the get_record call entirely.
To prevent this, we should verify that the decoded bytes actually start with a known audio magic header (matching the formats supported by _get_audio_magic_type in media_utils.py).
if not has_direct_source:
compact = "".join(file_ref.split())
padding = len(compact) % 4
if padding:
compact += "=" * (4 - padding)
try:
decoded = base64.b64decode(compact, validate=True)
has_direct_source = (
decoded.startswith(b"RIFF")
or decoded.startswith(b"#!AM")
or decoded.startswith(b"OggS")
or decoded.startswith(b"fLa")
or decoded.startswith(b"ID3")
or decoded.startswith(bytes([255, 251]))
or decoded.startswith(bytes([255, 243]))
or decoded.startswith(bytes([255, 242]))
or decoded.startswith(b"ftyp")
or decoded.startswith(b"#!SILK_V3")
or decoded.startswith(bytes([2]) + b"#!SILK_V3")
)
except (binascii.Error, ValueError):
pass| try: | ||
| has_direct_source = has_direct_source or Path(file_ref).is_file() | ||
| except OSError: | ||
| pass |
There was a problem hiding this comment.
On Windows, passing a string with invalid path characters (such as colons in URLs like http://... or base64://...) to Path().is_file() can raise a ValueError rather than an OSError. Catching both OSError and ValueError ensures robust handling across all platforms.
| try: | |
| has_direct_source = has_direct_source or Path(file_ref).is_file() | |
| except OSError: | |
| pass | |
| try: | |
| has_direct_source = has_direct_source or Path(file_ref).is_file() | |
| except (OSError, ValueError): | |
| pass |
| try: | ||
| if Path(file_uri_to_path(converted_file)).is_file(): | ||
| normalized["file"] = converted_file | ||
| return normalized | ||
| except OSError: | ||
| pass |
There was a problem hiding this comment.
Similarly, catching ValueError in addition to OSError here prevents potential unhandled exceptions on Windows if converted_file contains invalid path characters.
| try: | |
| if Path(file_uri_to_path(converted_file)).is_file(): | |
| normalized["file"] = converted_file | |
| return normalized | |
| except OSError: | |
| pass | |
| try: | |
| if Path(file_uri_to_path(converted_file)).is_file(): | |
| normalized["file"] = converted_file | |
| return normalized | |
| except (OSError, ValueError): | |
| pass |
| record = message.message[0] | ||
| assert isinstance(record, Record) | ||
| assert record.file == "voice.amr" | ||
| bot.call_action.assert_not_awaited() |
There was a problem hiding this comment.
It would be beneficial to add a test case to verify that a bare filename consisting of valid base64 characters (such as a hex hash or UUID) is correctly resolved through get_record rather than being bypassed as a legacy base64 payload.
bot.call_action.assert_not_awaited()
@pytest.mark.asyncio
async def test_record_with_hex_filename_is_resolved_through_get_record(tmp_path):
converted_file = tmp_path / "voice.wav"
with wave.open(str(converted_file), "wb") as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(8000)
wav_file.writeframes(bytes([0, 0]))
bot = AsyncMock()
bot.call_action.return_value = {"file": str(converted_file)}
event = _record_event({"file": "0d2bb1468a87d64414f8e563cc61c33c"})
message = await _adapter(bot)._convert_handle_message_event(event)
assert len(message.message) == 1
record = message.message[0]
assert isinstance(record, Record)
assert record.file == str(converted_file)
bot.call_action.assert_awaited_once_with(
action="get_record",
file="0d2bb1468a87d64414f8e563cc61c33c",
out_format="wav",
self_id=300,
)References
- New functionality, such as handling attachments, should be accompanied by corresponding unit tests.
dosubot
Bot
added
size:L
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 1 issue, and left some high level feedback:
- The
_resolve_record_datamethod has grown quite large and mixes several concerns (normalization, source detection, base64 signature parsing, and OneBot RPC); consider breaking it into smaller helpers (e.g.,has_direct_source,is_supported_bare_base64) to improve readability and make the heuristics easier to reuse or adjust. - The bare base64 audio detection logic (signature checks and manual padding) is fairly specialized; if there are existing utilities in
media_utilsor elsewhere for identifying/normalizing audio blobs, it would be beneficial to reuse or centralize this logic to avoid divergence and reduce maintenance cost.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The `_resolve_record_data` method has grown quite large and mixes several concerns (normalization, source detection, base64 signature parsing, and OneBot RPC); consider breaking it into smaller helpers (e.g., `has_direct_source`, `is_supported_bare_base64`) to improve readability and make the heuristics easier to reuse or adjust.
- The bare base64 audio detection logic (signature checks and manual padding) is fairly specialized; if there are existing utilities in `media_utils` or elsewhere for identifying/normalizing audio blobs, it would be beneficial to reuse or centralize this logic to avoid divergence and reduce maintenance cost.
## Individual Comments
### Comment 1
<location path="astrbot/core/platform/sources/aiocqhttp/aiocqhttp_platform_adapter.py" line_range="147" />
<code_context>
return abm
+ async def _resolve_record_data(
+ self,
+ data: dict[str, Any],
</code_context>
<issue_to_address>
**issue (complexity):** Consider splitting `_resolve_record_data` into smaller helpers for normalization, media-source detection, base64-audio sniffing, and `get_record` handling to keep the main flow linear and readable.
You can keep all current behavior but reduce complexity by extracting the clearly independent concerns into small helpers. This will also make it easier to test each piece.
### 1. Extract normalization and direct-source detection
Move the normalization and “has direct source” logic into helpers to clarify the high‑level flow:
```python
def _normalize_record_data(self, data: dict[str, Any]) -> dict[str, Any]:
normalized = dict(data)
for key in ("file", "url", "path"):
if isinstance(normalized.get(key), str):
normalized[key] = normalized[key].strip()
return normalized
def _has_direct_media_source(self, normalized: dict[str, Any]) -> bool:
file_ref = normalized.get("file")
if not isinstance(file_ref, str) or not file_ref:
return False
if normalized.get("url") or normalized.get("path"):
return True
if is_file_uri(file_ref) or file_ref.startswith(("http://", "https://", "base64://", "data:")):
return True
try:
if Path(file_ref).is_file():
return True
except (OSError, ValueError):
pass
return self._is_probable_base64_audio(file_ref)
```
Then `_resolve_record_data` becomes more linear and easier to read:
```python
async def _resolve_record_data(
self,
data: dict[str, Any],
routing_params: dict[str, Any],
) -> dict[str, Any]:
normalized = self._normalize_record_data(data)
file_ref = normalized.get("file")
if not isinstance(file_ref, str) or not file_ref:
return normalized
if self._has_direct_media_source(normalized):
return normalized
record = await self._resolve_record_via_get_record(file_ref, routing_params)
if record is not None:
return record
return normalized
```
### 2. Extract base64 audio detection
The inline “mini format sniffer” can be encapsulated into a single helper. That keeps `_has_direct_media_source` readable and prevents further format heuristics from bloating `_resolve_record_data`:
```python
def _is_probable_base64_audio(self, file_ref: str) -> bool:
compact = "".join(file_ref.split())
padding = len(compact) % 4
if padding:
compact += "=" * (4 - padding)
try:
decoded = base64.b64decode(compact, validate=True)
except (binascii.Error, ValueError):
return False
if (
len(decoded) >= 12
and decoded[:4] == b"RIFF"
and decoded[8:12] == b"WAVE"
):
return True
if len(decoded) >= 12 and decoded[4:8] == b"ftyp":
return True
return decoded.startswith(
(
b"#!AMR\n",
b"#!AMR-WB\n",
b"OggS",
b"fLaC",
b"ID3",
b"\xff\xfb",
b"\xff\xfa",
b"\xff\xf3",
b"\xff\xf2",
b"\xff\xe3",
b"\xff\xe2",
b"#!SILK_V3",
b"\x02#!SILK_V3",
)
)
```
If you already have or can add `media_utils`, this helper can be moved there, and `_has_direct_media_source` can call something like `is_base64_audio(file_ref)` instead.
### 3. Extract the `get_record` call / post-processing
The call to `self.bot.call_action` and its result handling can be factored out, isolating routing/bot coupling:
```python
async def _resolve_record_via_get_record(
self,
file_ref: str,
routing_params: dict[str, Any],
) -> dict[str, Any] | None:
try:
result = await self.bot.call_action(
action="get_record",
file=file_ref,
out_format="wav",
**routing_params,
)
except Exception as exc:
logger.warning(
"OneBot get_record failed; preserving the original record segment: %s",
exc,
)
return None
if not isinstance(result, dict):
logger.warning(
"OneBot get_record returned no file path; preserving the original record segment."
)
return None
converted_file = result.get("file")
if not isinstance(converted_file, str) or not converted_file.strip():
logger.warning(
"OneBot get_record returned no accessible media source; "
"preserving the original record segment."
)
return None
converted_file = converted_file.strip()
try:
if Path(file_uri_to_path(converted_file)).is_file():
return {"file": converted_file}
except (OSError, ValueError):
pass
logger.warning(
"OneBot get_record returned no accessible media source; "
"preserving the original record segment."
)
return None
```
This preserves all existing behavior (including logging and fallbacks) while making `_resolve_record_data` a thin orchestrator, which addresses the complexity concerns without reverting your feature.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
|
||
| return abm | ||
|
|
||
| async def _resolve_record_data( |
There was a problem hiding this comment.
issue (complexity): Consider splitting _resolve_record_data into smaller helpers for normalization, media-source detection, base64-audio sniffing, and get_record handling to keep the main flow linear and readable.
You can keep all current behavior but reduce complexity by extracting the clearly independent concerns into small helpers. This will also make it easier to test each piece.
1. Extract normalization and direct-source detection
Move the normalization and “has direct source” logic into helpers to clarify the high‑level flow:
def _normalize_record_data(self, data: dict[str, Any]) -> dict[str, Any]:
normalized = dict(data)
for key in ("file", "url", "path"):
if isinstance(normalized.get(key), str):
normalized[key] = normalized[key].strip()
return normalized
def _has_direct_media_source(self, normalized: dict[str, Any]) -> bool:
file_ref = normalized.get("file")
if not isinstance(file_ref, str) or not file_ref:
return False
if normalized.get("url") or normalized.get("path"):
return True
if is_file_uri(file_ref) or file_ref.startswith(("http://", "https://", "base64://", "data:")):
return True
try:
if Path(file_ref).is_file():
return True
except (OSError, ValueError):
pass
return self._is_probable_base64_audio(file_ref)Then _resolve_record_data becomes more linear and easier to read:
async def _resolve_record_data(
self,
data: dict[str, Any],
routing_params: dict[str, Any],
) -> dict[str, Any]:
normalized = self._normalize_record_data(data)
file_ref = normalized.get("file")
if not isinstance(file_ref, str) or not file_ref:
return normalized
if self._has_direct_media_source(normalized):
return normalized
record = await self._resolve_record_via_get_record(file_ref, routing_params)
if record is not None:
return record
return normalized2. Extract base64 audio detection
The inline “mini format sniffer” can be encapsulated into a single helper. That keeps _has_direct_media_source readable and prevents further format heuristics from bloating _resolve_record_data:
def _is_probable_base64_audio(self, file_ref: str) -> bool:
compact = "".join(file_ref.split())
padding = len(compact) % 4
if padding:
compact += "=" * (4 - padding)
try:
decoded = base64.b64decode(compact, validate=True)
except (binascii.Error, ValueError):
return False
if (
len(decoded) >= 12
and decoded[:4] == b"RIFF"
and decoded[8:12] == b"WAVE"
):
return True
if len(decoded) >= 12 and decoded[4:8] == b"ftyp":
return True
return decoded.startswith(
(
b"#!AMR\n",
b"#!AMR-WB\n",
b"OggS",
b"fLaC",
b"ID3",
b"\xff\xfb",
b"\xff\xfa",
b"\xff\xf3",
b"\xff\xf2",
b"\xff\xe3",
b"\xff\xe2",
b"#!SILK_V3",
b"\x02#!SILK_V3",
)
)If you already have or can add media_utils, this helper can be moved there, and _has_direct_media_source can call something like is_base64_audio(file_ref) instead.
3. Extract the get_record call / post-processing
The call to self.bot.call_action and its result handling can be factored out, isolating routing/bot coupling:
async def _resolve_record_via_get_record(
self,
file_ref: str,
routing_params: dict[str, Any],
) -> dict[str, Any] | None:
try:
result = await self.bot.call_action(
action="get_record",
file=file_ref,
out_format="wav",
**routing_params,
)
except Exception as exc:
logger.warning(
"OneBot get_record failed; preserving the original record segment: %s",
exc,
)
return None
if not isinstance(result, dict):
logger.warning(
"OneBot get_record returned no file path; preserving the original record segment."
)
return None
converted_file = result.get("file")
if not isinstance(converted_file, str) or not converted_file.strip():
logger.warning(
"OneBot get_record returned no accessible media source; "
"preserving the original record segment."
)
return None
converted_file = converted_file.strip()
try:
if Path(file_uri_to_path(converted_file)).is_file():
return {"file": converted_file}
except (OSError, ValueError):
pass
logger.warning(
"OneBot get_record returned no accessible media source; "
"preserving the original record segment."
)
return NoneThis preserves all existing behavior (including logging and fallbacks) while making _resolve_record_data a thin orchestrator, which addresses the complexity concerns without reverting your feature.
1 participant
Fixes #9199.
OneBot v11 allows an incoming
record.data.filevalue to be animplementation-owned voice filename rather than a directly accessible path or
URL. The aiocqhttp adapter previously passed that value directly to
Record,leaving downstream media resolution and STT with an unusable reference.
This change follows the official OneBot v11
get_recordcontract.
Modifications / 改动点
Resolve file-only incoming record segments through
get_record, requestingWAV output for the downstream audio pipeline.
Preserve existing behavior when a record already has a URL, path, URI, local
file, or supported base64 reference.
Preserve the original record segment when
get_recordfails or returns apath that AstrBot cannot access.
Keep the raw OneBot event unchanged and retain
self_idrouting formulti-account setups.
Add regression tests for successful resolution, direct-source bypasses,
legacy base64, account routing, and fallback behavior.
This is NOT a breaking change. / 这不是一个破坏性变更。
Screenshots or Test Results / 运行截图或测试结果
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
Resolve OneBot v11 record segments that contain only implementation-owned file identifiers by integrating the
get_recordAPI in the aiocqhttp adapter while preserving existing behavior and fallbacks.Bug Fixes:
filereferences are converted into accessible audio sources viaget_recordfor downstream media resolution and STT.get_recordcalls when a record already has a usable URL, path, local file, file URI, or recognizable base64-encoded audio payload.get_recordfails or returns an inaccessible file path so events and routing remain stable.Enhancements:
Tests:
get_record, handling of legacy and signature-based bare base64 audio, bypass behavior for direct sources, failure and inaccessible-path fallbacks, and correct account routing viaself_id.