feat(qqofficial): chunked upload for large media and text fallback

[TheRainstorm]

Problem

QQ v2 inline uploads (file_data base64 in POST /v2/groups/{group_openid}/files) are capped around 10MB. Larger local files (e.g. an 18MB xlsx produced by a long task) fail with:

[botpy] HTTP 413 Request Entity Too Large

and the bot ends up sending nothing (the subsequent msg_type=2 payload with empty markdown content fails with 40034011 无效 markdown content).

Fix

1. Chunked upload for large local files

When file_source is a local file > 10MB, use the official chunked flow instead of base64:

1. POST .../upload_prepare with md5/sha1/md5_10m → upload_id + block_size + presigned COS part URLs
2. PUT each part to its presigned URL (concurrency follows server upload_config, capped at 4; each part retried)
3. POST .../upload_part_finish per part (retries on biz_code 40093001)
4. POST .../files with upload_id to merge → file_info

Daily cumulative quota (biz_code 40093002) surfaces a friendly message instead of a raw error. C2C (/v2/users/{openid}) and group (/v2/groups/{group_openid}) endpoints are both supported.

2. Plain-text fallback so users never get silence

• If a markdown or media send fails (e.g. 40034011 invalid markdown content), retry as msg_type=0 content=.
• If a media-only message fails with no text attached, send a short text explanation instead of nothing.
• Media upload failures raise QQMediaUploadError and degrade the payload to plain text.

Testing

• Mocked end-to-end test of the chunked driver on a 17MB file: parts reconstruct the source byte-for-byte, md5/block_size match, upload_part_finish 40093001 retry works, daily-limit error surfaces correctly.
• Fallback unit tests for 40034011 → content-only, media-only failure → explanation text.
• ruff check + ruff format clean.

Summary by Sourcery

Implement chunked upload for large QQ media files and introduce robust text fallbacks when media or markdown message delivery fails.

New Features:

• Support QQ official chunked upload flow for large local media files exceeding the inline upload size limit.
• Provide automatic plain-text fallback when markdown or media messages fail to send so users still receive a response.

Enhancements:

• Introduce dedicated QQ media and API error types and stricter response validation to surface clearer upload failures and quotas.
• Add centralized QQ API helper and chunk-part upload orchestration with concurrency control and retries for more reliable media delivery.

[sourcery-ai]

Hey - I've found 2 issues, and left some high level feedback:

• The new QQ chunked upload helpers (_qq_api_request, _chunked_upload_media, _upload_one_part, etc.) reach into self.bot.api._http internals and construct Route URLs directly; consider introducing a thin wrapper or extending the existing botpy API for these endpoints so you don’t depend on private attributes that may change.
• The plain-text fallback logic builds similar explanation strings and newline handling in both _degrade_media_payload_to_text and _send_with_markdown_fallback; consider extracting a small helper to generate the fallback content and normalize msg_type/content so future changes to the message format are centralized.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The new QQ chunked upload helpers (_qq_api_request, _chunked_upload_media, _upload_one_part, etc.) reach into self.bot.api._http internals and construct Route URLs directly; consider introducing a thin wrapper or extending the existing botpy API for these endpoints so you don’t depend on private attributes that may change.
- The plain-text fallback logic builds similar explanation strings and newline handling in both _degrade_media_payload_to_text and _send_with_markdown_fallback; consider extracting a small helper to generate the fallback content and normalize msg_type/content so future changes to the message format are centralized.

## Individual Comments

### Comment 1
<location path="astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py" line_range="1098-1107" />
<code_context>
+        total_parts: int,
+    ) -> None:
+        """PUT one part to its presigned URL, then acknowledge via part_finish."""
+        part_index = int(part.get("part_index") or part.get("index") or 0)
+        presigned_url = str(part.get("presigned_url") or part.get("url") or "")
+        if not presigned_url:
+            raise QQMediaUploadError(
+                f"upload_prepare 分片缺少 presigned_url: {str(part)[:200]}"
+            )
+        part_block_size = int(part.get("block_size") or block_size)
+        offset = (part_index - 1) * block_size
+        length = min(part_block_size, file_size - offset)
+
+        data = await asyncio.get_running_loop().run_in_executor(
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Validate `part_index` before computing offsets to avoid confusing errors on malformed responses.

If `part_index` is missing or <= 0, `offset` becomes <= 0 and `_read_file_chunk` will likely fail with an opaque `OSError`. Consider explicitly enforcing `part_index >= 1` and raising a `QQMediaUploadError` when the server response is malformed, including a truncated `part` payload in the error to aid diagnosis.

```suggestion
        """PUT one part to its presigned URL, then acknowledge via part_finish."""
        part_index_raw = part.get("part_index") or part.get("index")
        try:
            part_index = int(part_index_raw)
        except (TypeError, ValueError):
            raise QQMediaUploadError(
                f"upload_prepare 分片返回非法 part_index: {part_index_raw!r}, part: {str(part)[:200]}"
            )
        if part_index <= 0:
            raise QQMediaUploadError(
                f"upload_prepare 分片返回非法 part_index={part_index}, part: {str(part)[:200]}"
            )

        presigned_url = str(part.get("presigned_url") or part.get("url") or "")
        if not presigned_url:
            raise QQMediaUploadError(
                f"upload_prepare 分片缺少 presigned_url: {str(part)[:200]}"
            )
        part_block_size = int(part.get("block_size") or block_size)
        offset = (part_index - 1) * block_size
        length = min(part_block_size, file_size - offset)
```
</issue_to_address>

### Comment 2
<location path="astrbot/core/platform/sources/qqofficial/qqofficial_message_event.py" line_range="659" />
<code_context>
+                payload["content"] = content + "\n"
+        return plain_text
+
     async def _send_with_markdown_fallback(
         self,
         send_func,
</code_context>
<issue_to_address>
**issue (complexity):** Consider centralizing the media/markdown-to-text degradation logic and encapsulating shared chunked upload state into a session object to streamline the upload and fallback flow.

The added functionality is valuable, but there are two areas where you can significantly reduce complexity without changing behavior:

### 1. Centralize payload degradation logic

Right now, `_degrade_media_payload_to_text` and `_send_with_markdown_fallback` both implement slightly different degradation paths (markdown→text, media→text, explanation when no plain text). You can reuse `_degrade_media_payload_to_text` inside `_send_with_markdown_fallback` and remove duplicated transformations.

For example, replace the two inline degradation blocks in `_send_with_markdown_fallback` with calls to the helper:

```python
    async def _send_with_markdown_fallback(
        self,
        send_func,
        payload: dict,
        plain_text: str,
        stream: dict | None = None,
    ):
        try:
            return await send_func(payload)
        except _QQOFFICIAL_SEND_API_ERRORS as err:
            # ... existing proactive send fallback ...

            # 纯文本兜底：markdown 或媒体消息发送失败时降级为 msg_type=0 纯文本。
            if plain_text and (payload.get("markdown") or payload.get("media")):
                plain_text = self._degrade_media_payload_to_text(
                    payload, plain_text, err, stream
                )
                try:
                    ret = await send_func(payload)
                    logger.info("[QQOfficial] 主动发送接口（纯文本）发送成功。")
                    return ret
                except _QQOFFICIAL_SEND_API_ERRORS as content_err:
                    err = content_err

            # 媒体消息发送失败且无文本时，发送失败说明，避免用户收不到任何回复。
            if not plain_text and (
                payload.get("media") or payload.get("msg_type") == 7
            ):
                plain_text = self._degrade_media_payload_to_text(
                    payload, plain_text, err, stream
                )
                try:
                    ret = await send_func(payload)
                    logger.info("[QQOfficial] 媒体发送失败，已发送文本说明。")
                    return ret
                except _QQOFFICIAL_SEND_API_ERRORS as content_err:
                    err = content_err

            # ... existing newline fix and re-raise ...
```

This keeps the rules for:
- stripping `markdown`/`media`
- setting `msg_type`
- newline handling with `stream`
- default explanation text when `plain_text` is empty

in a single place, reducing cognitive load for future changes to degradation behavior.

### 2. Reduce parameter threading in chunked upload helpers

The chunked upload helpers `_chunked_upload_media`, `_upload_one_part`, `_part_finish_with_retry` currently pass many separate parameters, which makes the flow harder to follow and update.

Introduce a small session object to carry shared parameters and use it across helpers:

```python
from dataclasses import dataclass

@dataclass
class _ChunkUploadSession:
    base: str
    receiver: dict
    upload_id: str
    block_size: int
    file_source: str
    file_size: int
    retry_timeout: float
    total_parts: int
```

Then simplify helper signatures:

```python
    async def _chunked_upload_media(
        self,
        file_source: str,
        file_type: int,
        file_name: str,
        openid: str | None = None,
        group_openid: str | None = None,
    ) -> Media:
        # ... existing code up to upload_id/parts/concurrency/retry_timeout ...

        session = _ChunkUploadSession(
            base=base,
            receiver=receiver_field,
            upload_id=upload_id,
            block_size=block_size,
            file_source=file_source,
            file_size=file_size,
            retry_timeout=retry_timeout,
            total_parts=len(parts),
        )

        sem = asyncio.Semaphore(max(1, min(concurrency, 4)))

        async def upload_part(part: dict) -> None:
            async with sem:
                await self._upload_one_part(session, part)

        await asyncio.gather(*(upload_part(p) for p in parts))
        # ... complete step unchanged ...


    async def _upload_one_part(
        self,
        session: _ChunkUploadSession,
        part: dict,
    ) -> None:
        part_index = int(part.get("part_index") or part.get("index") or 0)
        presigned_url = str(part.get("presigned_url") or part.get("url") or "")
        part_block_size = int(part.get("block_size") or session.block_size)

        offset = (part_index - 1) * session.block_size
        length = min(part_block_size, session.file_size - offset)

        data = await asyncio.get_running_loop().run_in_executor(
            None, _read_file_chunk, session.file_source, offset, length
        )
        md5_hex = hashlib.md5(data).hexdigest()

        await self._put_part_with_retry(
            presigned_url, data, part_index, session.total_parts
        )
        await self._part_finish_with_retry(
            session, part_index, length, md5_hex
        )


    async def _part_finish_with_retry(
        self,
        session: _ChunkUploadSession,
        part_index: int,
        block_size: int,
        md5: str,
    ) -> None:
        body = {
            "upload_id": session.upload_id,
            "part_index": part_index,
            "block_size": block_size,
            "md5": md5,
            **session.receiver,
        }
        # ... existing retry loop using session.base and session.retry_timeout ...
```

This preserves behavior but:
- Makes each helper’s responsibility clearer.
- Reduces long parameter lists and coupling.
- Makes it easier to adjust protocol details (e.g., retry_timeout, block_size) in one place.

If desired, you can later move `_ChunkUploadSession` and chunked-upload-specific constants (`_QQOFFICIAL_*`) into a dedicated uploader module/class without changing the public `upload_group_and_c2c_media` API.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[TheRainstorm]

Addressed the review feedback:

1. part_index validation (bug_risk): _upload_one_part now parses part_index defensively and raises a clear QQMediaUploadError (with the truncated part payload) when it is missing, non-numeric, or <= 0, before any offset math.
2. Centralized degradation: _send_with_markdown_fallback now reuses _degrade_media_payload_to_text for both fallback paths, so stripping markdown/media, setting msg_type/content, stream newline handling, and the default explanation text live in one place.
3. Chunked session object: introduced _ChunkUploadSession to carry shared state and slimmed _upload_one_part/_part_finish_with_retry signatures.
4. botpy internals: added a docstring note explaining that the raw botpy HTTP session is used because botpy only surfaces the API error message and drops the platform error code needed for 40093001/40093002 handling; this matches the existing self.bot.api._http.request pattern already used in this file.

Re-ran the mocked end-to-end tests (17MB file: parts reconstruct byte-for-byte, part_finish 40093001 retry, daily-limit message, malformed part_index) and the fallback tests — all pass; ruff check + ruff format clean.