HeXi/hexi/plugins/nonebot_plugin_video_analysis/handlers/sender.py

"""媒体多级发送 — temp 本地文件优先，失败逐级降级

发送策略（2026-08-22 用户需求）:
1. temp 本地文件直接发送（最快，不经 S3）
2. 失败 → 上传本地 S3，用预签名链接发送
3. 再失败 → 回退 temp 本地文件再发一次

temp 下的文件发送成功后也不清理（用户手动处理 data/temp）。
"""

from dataclasses import dataclass
from pathlib import Path
from typing import Optional, Union

from nonebot import logger
from nonebot_plugin_alconna import UniMessage

from ..services.storage.s3 import upload_with_plan


@dataclass
class PendingMedia:
    """待发送媒体：本地文件 + 上传元数据（发送前不做任何上传/清理）"""

    files: list[Path]
    image_post: bool = False
    is_private: bool = False
    plan: Optional[str] = None
    title: str = ""


def _as_paths(file_paths: Union[Path, list[Path]]) -> list[Path]:
    if isinstance(file_paths, list):
        return [Path(p) for p in file_paths]
    return [Path(file_paths)]


def _build_local_msg(files: list[Path], image_post: bool) -> UniMessage:
    """本地文件版消息（mp4 → 视频，其余 → 图片）"""
    msg = UniMessage()
    for fp in files:
        if fp.suffix.lower() == ".mp4":
            msg.video(path=fp)
        else:
            msg.image(path=fp)
    return msg


def _build_s3_msg(
    media: PendingMedia,
) -> tuple[UniMessage, Optional[str]]:
    """上传本地 S3 并构建链接版消息，返回 (message, public_url)"""
    msg = UniMessage()
    public_url = None
    for fp in media.files:
        local_url, pub = upload_with_plan(
            fp,
            plan=media.plan,
            is_private=media.is_private,
            title=media.title,
            image_post=media.image_post,
        )
        if not local_url:
            raise RuntimeError(f"上传本地 S3 失败: {fp}")
        if pub:
            public_url = pub
        if fp.suffix.lower() == ".mp4":
            msg.video(url=local_url)
        else:
            msg.image(url=local_url)
    return msg, public_url


async def send_pending_media(media: PendingMedia) -> tuple[bool, Optional[str]]:
    """多级发送，返回 (是否成功, public_url)

    public_url 仅在走 S3 链接发送成功时返回（调用方决定是否发文字）。
    temp 文件发送成功后保留（用户手动清理 data/temp）。
    """
    if not media.files:
        return False, None

    # ── 1. temp 本地文件直接发送 ──────────────────────────────
    try:
        await _build_local_msg(media.files, media.image_post).send()
        logger.info("媒体发送成功（temp 本地文件直达）")
        return True, None
    except Exception as e:
        logger.warning(f"temp 本地文件发送失败，切换本地 S3 链接: {e}")

    # ── 2. 上传本地 S3 → 链接发送 ─────────────────────────────
    try:
        s3_msg, public_url = _build_s3_msg(media)
        await s3_msg.send()
        logger.info("媒体发送成功（本地 S3 链接）")
        return True, public_url
    except Exception as e:
        logger.warning(f"本地 S3 链接发送失败，回退 temp 本地文件: {e}")

    # ── 3. 回退：temp 本地文件再发一次 ────────────────────────
    try:
        await _build_local_msg(media.files, media.image_post).send()
        logger.info("媒体发送成功（回退 temp 本地文件）")
        return True, None
    except Exception as e:
        logger.exception(f"回退发送失败: {e}")
        return False, None