life-echo/api/app/features/conversation/tts_delivery.py

"""
对话 TTS 音频 URL 下发到客户端。

与回忆录章节图片一致：私有桶下不能把「直链」当公开可读 URL 使用，应对 COS object key
生成预签名下载地址后再交给 App（参见 `normalize_image_assets_for_api` 中的 `get_download_url`）。

持久化（DB / Redis）仍保存 upload 返回的 canonical URL，仅在 API 响应与 WS 实时下发时做 presign。
"""

from __future__ import annotations

from app.core.cos_url_keys import (
    TTS_PRESIGNED_EXPIRES_SEC,
    extract_cos_object_key_if_owned,
)
from app.core.logging import get_logger
from app.ports.storage import ObjectStorage

logger = get_logger(__name__)


def apply_presigned_tts_urls_to_messages(
    messages: list[dict],
    storage: ObjectStorage | None,
) -> None:
    """就地改写助手消息的 `ttsAudioUrls` 为预签名 URL；无 storage 时不变。"""
    if not storage:
        return
    for m in messages:
        tts = m.get("ttsAudioUrls")
        if not isinstance(tts, list) or not tts:
            continue
        out: list[str] = []
        for x in tts:
            if not isinstance(x, str):
                out.append("")
                continue
            s = x.strip()
            if not s:
                out.append("")
                continue
            key = extract_cos_object_key_if_owned(s)
            if key:
                try:
                    out.append(storage.get_url(key, expires=TTS_PRESIGNED_EXPIRES_SEC))
                except Exception as exc:
                    logger.warning(
                        "presign tts url failed, keeping original url: key={} err={}",
                        key,
                        exc,
                    )
                    out.append(s)
            else:
                out.append(s)
        m["ttsAudioUrls"] = out