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

"""Redis 客户端与会话/缓存能力：供应用生命周期、会话历史、任务追踪等使用。
配置从 app.core.config.settings 读取，禁止业务层散落 os.getenv。
"""

import json
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

import redis.asyncio as aioredis

from app.core.config import settings
from app.core.logging import get_logger
from app.core.runtime_constants import misc_defaults, redis_defaults

logger = get_logger(__name__)


class RedisService:
    """Redis 服务：连接管理、对话历史、通用缓存。"""

    def __init__(self) -> None:
        self.redis_url = settings.redis_url_resolved
        self._client: Optional[aioredis.Redis] = None
        self.session_ttl = misc_defaults.redis_session_ttl

    async def get_client(self) -> aioredis.Redis:
        """获取 Redis 客户端（延迟初始化）。"""
        if self._client is None:
            try:
                self._client = await aioredis.from_url(
                    self.redis_url,
                    encoding="utf-8",
                    decode_responses=True,
                    socket_timeout=redis_defaults.socket_timeout_seconds,
                    socket_connect_timeout=redis_defaults.socket_connect_timeout_seconds,
                    health_check_interval=redis_defaults.health_check_interval_seconds,
                    retry_on_timeout=True,
                )
                await self._client.ping()
                logger.info("Redis 连接成功")
                try:
                    from urllib.parse import urlparse

                    p = urlparse(self.redis_url)
                    logger.debug(
                        "Redis 连接 host={} port={}",
                        p.hostname or "",
                        p.port or "",
                    )
                except Exception:
                    logger.debug("Redis 已连接（URL 解析省略）")
            except Exception as e:
                logger.error("Redis 连接失败: {}", e)
                raise
        return self._client

    async def close(self) -> None:
        """关闭 Redis 连接。"""
        if self._client:
            await self._client.close()
            self._client = None

    def _conversation_key(self, conversation_id: str) -> str:
        return f"conversation:history:{conversation_id}"

    async def _key_type(self, client: aioredis.Redis, key: str) -> str:
        key_type = await client.type(key)
        if isinstance(key_type, bytes):
            return key_type.decode("utf-8")
        return str(key_type)

    async def _parse_history_items(self, raw_items: List[str]) -> List[Dict[str, Any]]:
        history: List[Dict[str, Any]] = []
        for raw in raw_items:
            try:
                parsed = json.loads(raw)
            except json.JSONDecodeError:
                logger.warning("跳过无效对话历史条目")
                continue
            if isinstance(parsed, dict):
                history.append(parsed)
        return history

    async def _migrate_string_history_to_list(
        self, client: aioredis.Redis, key: str, history: List[Dict[str, Any]]
    ) -> None:
        if not history:
            await client.delete(key)
            return
        pipe = client.pipeline(transaction=True)
        pipe.delete(key)
        for item in history:
            pipe.rpush(key, json.dumps(item, ensure_ascii=False))
        pipe.expire(key, self.session_ttl)
        await pipe.execute()

    async def get_conversation_history(
        self, conversation_id: str
    ) -> List[Dict[str, Any]]:
        try:
            client = await self.get_client()
            key = self._conversation_key(conversation_id)
            if not await client.exists(key):
                return []
            key_type = await self._key_type(client, key)
            if key_type == "list":
                raw_items = await client.lrange(key, 0, -1)
                return await self._parse_history_items(list(raw_items))
            if key_type == "string":
                data = await client.get(key)
                if not data:
                    return []
                legacy = json.loads(data)
                if not isinstance(legacy, list):
                    return []
                history = [x for x in legacy if isinstance(x, dict)]
                await self._migrate_string_history_to_list(client, key, history)
                return history
            logger.warning(
                "conversation history unexpected type={} key={}",
                key_type,
                key,
            )
            return []
        except Exception as e:
            logger.error("获取对话历史失败: {}", e)
            return []

    async def set_conversation_history(
        self, conversation_id: str, history: List[Dict[str, Any]]
    ) -> bool:
        """整表覆盖会话历史（用于从 DB 回填），应用 session_ttl。"""
        try:
            client = await self.get_client()
            key = self._conversation_key(conversation_id)
            pipe = client.pipeline(transaction=True)
            pipe.delete(key)
            for item in history:
                pipe.rpush(key, json.dumps(item, ensure_ascii=False))
            pipe.expire(key, self.session_ttl)
            await pipe.execute()
            return True
        except Exception as e:
            logger.error("写入对话历史失败: {}", e)
            return False

    async def add_message(
        self,
        conversation_id: str,
        role: str,
        content: str,
        message_type: str = "text",
        voice_session_id: str | None = None,
        timestamp: str | int | None = None,
        audio_duration_seconds: int | None = None,
    ) -> bool:
        try:
            client = await self.get_client()
            key = self._conversation_key(conversation_id)
            item = {
                "role": role,
                "content": content,
                "messageType": message_type,
                "timestamp": timestamp or datetime.now(timezone.utc).isoformat(),
            }
            if voice_session_id:
                item["voiceSessionId"] = voice_session_id
            if (
                audio_duration_seconds is not None
                and audio_duration_seconds > 0
                and message_type == "audio"
            ):
                item["durationSeconds"] = int(audio_duration_seconds)
            pipe = client.pipeline(transaction=True)
            pipe.rpush(key, json.dumps(item, ensure_ascii=False))
            pipe.expire(key, self.session_ttl)
            await pipe.execute()
            return True
        except Exception as e:
            logger.error("添加消息失败: {}", e)
            return False

    async def append_tts_audio_url_to_last_ai_message(
        self, conversation_id: str, url: str
    ) -> bool:
        """向最近一条 AI 消息的 ttsAudioUrls 追加 upload 返回的 canonical URL（非预签名）。客户端通过 GET /messages 等出口收到预签名 URL。"""
        if not url:
            return False
        try:
            client = await self.get_client()
            key = self._conversation_key(conversation_id)
            history = await self.get_conversation_history(conversation_id)
            target_index: int | None = None
            for i in range(len(history) - 1, -1, -1):
                if history[i].get("role") == "ai":
                    existing = history[i].get("ttsAudioUrls")
                    urls: List[str] = (
                        [x for x in existing if isinstance(x, str)]
                        if isinstance(existing, list)
                        else []
                    )
                    urls.append(url)
                    history[i]["ttsAudioUrls"] = urls
                    target_index = i
                    break
            if target_index is None:
                logger.warning(
                    "append_tts_audio_url: no ai message in history conversation_id={}",
                    conversation_id,
                )
                return False
            await client.lset(
                key,
                target_index,
                json.dumps(history[target_index], ensure_ascii=False),
            )
            await client.expire(key, self.session_ttl)
            return True
        except Exception as e:
            logger.error("append_tts_audio_url 失败: {}", e)
            return False

    async def clear_conversation_history(self, conversation_id: str) -> bool:
        try:
            client = await self.get_client()
            key = self._conversation_key(conversation_id)
            await client.delete(key)
            return True
        except Exception as e:
            logger.error("清除对话历史失败: {}", e)
            return False

    async def delete_keys_matching_pattern(self, pattern: str) -> int:
        """按 SCAN 批量删除 key，避免阻塞式 KEYS *。"""
        try:
            client = await self.get_client()
            batch: list[str] = []
            deleted = 0

            async for key in client.scan_iter(match=pattern):
                batch.append(key)
                if len(batch) >= 200:
                    deleted += int(await client.delete(*batch))
                    batch.clear()
            if batch:
                deleted += int(await client.delete(*batch))
            return deleted
        except Exception as e:
            logger.error("按 pattern 删除 Redis key 失败: {}", e)
            return 0

    async def extend_session_ttl(self, conversation_id: str) -> bool:
        try:
            client = await self.get_client()
            key = self._conversation_key(conversation_id)
            await client.expire(key, self.session_ttl)
            return True
        except Exception as e:
            logger.error("延长会话TTL失败: {}", e)
            return False

    async def set_cache(self, key: str, value: Any, ttl: Optional[int] = None) -> bool:
        if ttl is None or ttl <= 0:
            logger.error("设置缓存失败: TTL 必须为正整数 key={}", key)
            return False
        try:
            client = await self.get_client()
            data = (
                json.dumps(value, ensure_ascii=False)
                if not isinstance(value, str)
                else value
            )
            await client.setex(key, ttl, data)
            return True
        except Exception as e:
            logger.error("设置缓存失败: {}", e)
            return False

    async def get_cache(self, key: str) -> Optional[Any]:
        try:
            client = await self.get_client()
            data = await client.get(key)
            if data:
                try:
                    return json.loads(data)
                except json.JSONDecodeError:
                    return data
            return None
        except Exception as e:
            logger.error("获取缓存失败: {}", e)
            return None

    async def delete_cache(self, key: str) -> bool:
        try:
            client = await self.get_client()
            await client.delete(key)
            return True
        except Exception as e:
            logger.error("删除缓存失败: {}", e)
            return False

    def is_available(self) -> bool:
        return self._client is not None


# 全局单例，供 main 生命周期与各 feature 通过 get_redis_service 或直接引用使用
redis_service = RedisService()