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


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