Please address the comments from this code review: ## Overall Comments - In `_resolve_onebot_file_reference` you currently `except BaseException`, which will also swallow `asyncio.CancelledError` and system-level exceptions; consider narrowing this to `Exception` (or a more specific subset) so cancellations and fatal errors are not silently ignored. - The `_get_source` helpers in `Record` and `Video` are duplicated and follow the same precedence rules; consider extracting a shared utility or mixin to avoid divergence if the source-selection logic needs to change in the future. ## Individual Comments ### Comment 1 <location path="astrbot/core/platform/sources/aiocqhttp/aiocqhttp_platform_adapter.py" line_range="136-40" /> <code_context> + + file_ref = normalized.get("file") + file_id = normalized.get("file_id") + candidate = file_ref or file_id + if not isinstance(candidate, str) or not candidate.strip(): + usable_source = _pick_usable_media_source(normalized) + if usable_source: + normalized["file"] = usable_source + if seg_type == "file": + normalized.setdefault("url", usable_source) + return normalized + return normalized + if _looks_like_resolved_media_ref(candidate) or os.path.exists(candidate): + normalized["file"] = candidate + return normalized </code_context> <issue_to_address> **issue (bug_risk):** Unstripped `candidate` may contain leading/trailing spaces and fail resolution checks. `candidate` is checked using `candidate.strip()` but the subsequent `_looks_like_resolved_media_ref` and `os.path.exists` calls use the unstripped value. If `file`/`file_id` includes surrounding whitespace, these checks may incorrectly fail. Consider normalizing once (e.g. `candidate = candidate.strip()`) before the checks and later uses. </issue_to_address> ### Comment 2 <location path="astrbot/core/platform/sources/aiocqhttp/aiocqhttp_platform_adapter.py" line_range="98-102" /> <code_context> + ) + + for action, params in actions: + try: + ret = await bot.call_action(action=action, **params) + except BaseException: + continue + data = _unwrap_onebot_action_data(ret) </code_context> <issue_to_address> **suggestion (bug_risk):** Catching `BaseException` here is overly broad and may hide critical errors. This will also suppress `KeyboardInterrupt`, `SystemExit`, and other critical exceptions, which can break shutdown and debugging. Here it’s safer to catch `Exception` (or specific library exceptions) so truly critical errors still propagate. ```suggestion for action, params in actions: try: ret = await bot.call_action(action=action, **params) except Exception: continue ``` </issue_to_address> ### Comment 3 <location path="astrbot/core/platform/sources/aiocqhttp/aiocqhttp_platform_adapter.py" line_range="31" /> <code_context> from .aiocqhttp_message_event import AiocqhttpMessageEvent +def _looks_like_resolved_media_ref(value: str) -> bool: + return value.startswith(("http://", "https://", "file://")) or os.path.isabs(value) + </code_context> <issue_to_address> **issue (complexity):** Consider refactoring the new media resolution helpers into a small `MediaResolver` class with shared helpers and action templates to reduce duplication and make the control flow easier to follow. You can reduce the new complexity without changing behavior by: 1. **Encapsulating shared context and flow in a small resolver object** 2. **Centralizing the “is resolved” check** 3. **De‑duplicating action construction with static templates** ### 1. Encapsulate context into a `MediaResolver` Instead of threading `bot`, `message_type`, `group_id`, `seg_type` through multiple helpers and call sites, wrap them in a small resolver. This keeps the public surface small while keeping functionality identical: ```python class MediaResolver: def __init__(self, bot: CQHttp, message_type: str, group_id: str | int | None): self.bot = bot self.message_type = message_type self.group_id = group_id async def normalize(self, seg_type: str, seg_data: dict[str, Any]) -> dict[str, Any]: # move current _normalize_onebot_media_data body here # but replace direct calls with self._resolve_file_reference(...) ... async def _resolve_file_reference(self, file_ref: str, seg_type: str | None = None) -> str | None: # move current _resolve_onebot_file_reference body here # use self.bot, self.message_type, self.group_id ... ``` Call site becomes simpler and avoids repeated parameters: ```python resolver = MediaResolver(self.bot, event["message_type"], event.get("group_id")) # file segment special case normalized_data = await resolver.normalize("file", m["data"]) ... # generic handler normalized_data = await resolver.normalize(t, m["data"]) a = ComponentTypes[t](**normalized_data) ``` This keeps all behavior but removes a lot of cross‑cutting arguments and makes the flow easier to follow. ### 2. Centralize the “resolved path or URL” check You currently repeat: - `_looks_like_resolved_media_ref(value) or os.path.exists(value)` across `_pick_usable_media_source`, `_normalize_onebot_media_data`, `_resolve_onebot_file_reference`. Extract this into a single helper that returns either a normalized string or `None`: ```python def _resolve_local_or_url_candidate(value: str | None) -> str | None: if not isinstance(value, str): return None candidate = value.strip() if not candidate: return None if _looks_like_resolved_media_ref(candidate) or os.path.exists(candidate): return candidate return None ``` Then simplify callers, e.g.: ```python def _pick_usable_media_source(seg_data: dict[str, Any]) -> str: for key in ("url", "file", "path", "file_path"): resolved = _resolve_local_or_url_candidate(seg_data.get(key)) if resolved: return resolved return "" ``` and in the resolver: ```python candidate = file_ref or file_id resolved = _resolve_local_or_url_candidate(candidate) if resolved: normalized["file"] = resolved return normalized ``` This reduces branching and duplication without changing semantics. ### 3. De‑duplicate action list construction with templates The `actions.extend([...])` pattern is verbose and repeated per candidate. You can describe the patterns once and instantiate them: ```python # module-level, static _BASE_ACTION_TEMPLATES: list[tuple[str, dict[str, str]]] = [ ("get_file", {"file_id": "candidate"}), ("get_file", {"file": "candidate"}), ("get_image", {"file": "candidate"}), ("get_image", {"file_id": "candidate"}), ("get_private_file_url", {"file_id": "candidate"}), ] def _build_actions_for_candidate( candidate: str, message_type: str, group_id: str | int | None, seg_type: str | None, ) -> list[tuple[str, dict[str, Any]]]: actions: list[tuple[str, dict[str, Any]]] = [] if seg_type == "record": actions.append(("get_record", {"file": candidate})) for action, params in _BASE_ACTION_TEMPLATES: concrete = {k: (candidate if v == "candidate" else v) for k, v in params.items()} actions.append((action, concrete)) if str(message_type).lower() == "group" and group_id not in (None, ""): actions.append(("get_group_file_url", {"group_id": group_id, "file_id": candidate})) return actions ``` Then `_resolve_onebot_file_reference` (or `MediaResolver._resolve_file_reference`) becomes clearer: ```python actions: list[tuple[str, dict[str, Any]]] = [] for candidate in candidates: actions.extend(_build_actions_for_candidate(candidate, message_type, group_id, seg_type)) ``` This removes duplicated action construction logic and makes it obvious what variations exist. --- These changes keep your new media resolution behavior intact, but: - Narrow the public API to `MediaResolver.normalize(...)` - Centralize repeated checks - Make candidate/action handling more declarative and easier to scan. </issue_to_address>