life-echo/api/app/core/cos_url_keys.py

"""从 URL 解析当前环境腾讯云 COS object key（仅当 host 与配置一致时）。"""

from __future__ import annotations

from typing import Any
from urllib.parse import urlparse

from app.core.config import settings
from app.core.logging import get_logger
from app.ports.storage import ObjectStorage

logger = get_logger(__name__)

# 客户端再读 TTS / 拉取音频：预签名有效期（秒），与移动端会话长度匹配
TTS_PRESIGNED_EXPIRES_SEC = 86_400


def extract_cos_object_key_if_owned(url: str | None) -> str | None:
    """
    若 url 指向 settings 中配置的 COS 域名，返回去掉前导 / 的 object key。
    非 http(s)、或 host 不匹配时返回 None。
    """
    if not url:
        return None
    s = str(url).strip()
    if not s.startswith(("http://", "https://")):
        return None
    parsed = urlparse(s)
    host = (parsed.netloc or "").lower()
    if not host:
        return None

    candidates: list[str] = []
    bucket = (settings.tencent_cos_bucket or "").strip().lower()
    region = (settings.tencent_cos_region or "").strip().lower()
    if bucket and region:
        candidates.append(f"{bucket}.cos.{region}.myqcloud.com")
    base = (settings.tencent_cos_base_url or "").strip()
    if base:
        base_parsed = urlparse(base if "://" in base else f"https://{base}")
        bh = (base_parsed.netloc or "").lower()
        if bh:
            candidates.append(bh)

    if not candidates:
        return None

    matched = any(host == c for c in candidates if c)
    if not matched:
        return None

    key = (parsed.path or "").lstrip("/")
    return key or None


def collect_cos_keys_from_conversation_history(
    history: list[dict[str, Any]],
) -> set[str]:
    """从 Redis 会话历史中收集 AI 消息附带的 TTS 音频 COS object key。"""
    keys: set[str] = set()
    for msg in history:
        if msg.get("role") != "ai":
            continue
        raw = msg.get("ttsAudioUrls")
        if not isinstance(raw, list):
            continue
        for u in raw:
            if isinstance(u, str):
                k = extract_cos_object_key_if_owned(u)
                if k:
                    keys.add(k)
    return keys


def collect_cos_keys_from_tts_url_list(urls: list[str] | None) -> set[str]:
    if not urls:
        return set()
    keys: set[str] = set()
    for u in urls:
        if isinstance(u, str):
            k = extract_cos_object_key_if_owned(u)
            if k:
                keys.add(k)
    return keys


def presign_tts_urls_for_playback(
    urls: list[str],
    storage: ObjectStorage | None,
    *,
    expires: int = TTS_PRESIGNED_EXPIRES_SEC,
) -> list[str]:
    """
    将本环境 COS 直链替换为预签名下载 URL（私有桶下匿名 GET 会 AccessDenied）。

    目的与回忆录 `normalize_image_assets_for_api` 中对 `get_download_url` 的用法一致。
    非本环境 URL 或无法解析 key 时原样返回。
    """
    if not storage or not urls:
        return list(urls)
    out: list[str] = []
    for u in urls:
        if not isinstance(u, str):
            continue
        s = u.strip()
        if not s:
            continue
        key = extract_cos_object_key_if_owned(s)
        if key:
            try:
                out.append(storage.get_url(key, expires=expires))
            except Exception as exc:
                logger.warning(
                    "presign tts url failed, keeping original url: key={} err={}",
                    key,
                    exc,
                )
                out.append(s)
        else:
            out.append(s)
    return out