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

# 用户头像 API 下发（与 TTS 一致）
AVATAR_PRESIGNED_EXPIRES_SEC = TTS_PRESIGNED_EXPIRES_SEC


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


def avatar_url_for_api_response(stored_url: str | None) -> str | None:
    """DB 中的头像 URL：本环境 COS 直链改为预签名下载 URL。"""
    if stored_url is None:
        return None
    s = str(stored_url).strip()
    if not s:
        return None
    key = extract_cos_object_key_if_owned(s)
    if not key:
        return s
    if not (
        (settings.tencent_cos_secret_id or "").strip()
        and (settings.tencent_cos_bucket or "").strip()
    ):
        return s
    from app.core.dependencies import get_object_storage

    storage = get_object_storage()
    try:
        return storage.get_url(key, expires=AVATAR_PRESIGNED_EXPIRES_SEC)
    except Exception as exc:
        logger.warning(
            "presign avatar url failed, keeping original: key={} err={}",
            key,
            exc,
        )
        return s


def best_effort_delete_cos_object_for_url(url: str | None) -> None:
    """本环境 COS 对象则尽力 delete（换头像 / 预设时清理）。"""
    key = extract_cos_object_key_if_owned(url)
    if not key:
        return
    if not (
        (settings.tencent_cos_secret_id or "").strip()
        and (settings.tencent_cos_bucket or "").strip()
    ):
        return
    from app.core.dependencies import get_object_storage

    try:
        get_object_storage().delete(key)
    except Exception as exc:
        logger.warning("delete cos avatar object failed: key={} err={}", key, exc)