app/core/session_manager.py

# app/core/session_manager.py

import time
import uuid  # 导入 uuid 模块用于生成会话 ID
import logging
from typing import Dict, Any, List, Optional
from gemini_webapi import GeminiClient, ChatSession # 导入 ChatSession 类型

logger = logging.getLogger(__name__)

# 每个认证令牌的最大会话数
MAX_SESSIONS_PER_TOKEN = 5

# 会话数据结构
class Session:
    def __init__(self, thread_id: str, name: Optional[str] = None):
        self.thread_id = thread_id
        self.name = name or f"会话-{time.strftime('%m%d')}-{str(uuid.uuid4())[:4]}"
        self.created_at = time.time()
        self.last_active = time.time()
        self._chat_instance: Optional[ChatSession] = None  # 存储 Gemini Chat 实例

# 全局变量，用于存储每个认证令牌的聊天会话
chat_sessions: Dict[str, Dict[str, Session]] = {}

def cleanup_expired_sessions(auth_token: str):
    """清理过期会话（24小时未活动）和超出限制的会话"""
    now = time.time()
    sessions = chat_sessions.get(auth_token, {})
    
    # 清理超过24小时的会话
    expired_threads = [
        tid for tid, session in sessions.items()
        if now - session.last_active > 86400
    ]
    for tid in expired_threads:
        logger.info(f"清理过期会话: {tid}")
        del sessions[tid]
    
    # 清理超出数量限制的旧会话
    if len(sessions) > MAX_SESSIONS_PER_TOKEN:
        # 按照最后活跃时间排序，删除最旧的会话
        oldest_sessions = sorted(sessions.items(), key=lambda x: x[1].last_active)
        for i in range(len(sessions) - MAX_SESSIONS_PER_TOKEN):
            tid_to_delete = oldest_sessions[i][0]
            logger.info(f"清理超出限制的会话: {tid_to_delete}")
            del sessions[tid_to_delete]
    
    chat_sessions[auth_token] = sessions # 确保更新全局字典
    return sessions

def get_or_create_session(auth_token: str, thread_id: Optional[str], client: GeminiClient) -> Session:
    """
    获取或创建一个新的聊天会话。
    如果 thread_id 为 None，则创建一个新会话。
    如果会话存在但其 _chat_instance 为 None，则重新初始化 _chat_instance。
    """
    sessions = chat_sessions.get(auth_token, {})
    if not sessions:
        sessions = {}
        chat_sessions[auth_token] = sessions

    session = None
    if thread_id:
        session = sessions.get(thread_id)

    if not session:
        thread_id = str(uuid.uuid4())
        session = Session(thread_id)
        sessions[thread_id] = session
        logger.info(f"认证令牌 {auth_token} 生成新会话: {thread_id}")
    
    # 确保 chat 实例已初始化
    if session._chat_instance is None: # 只有当 _chat_instance 为 None 时才创建
        logger.info(f"认证令牌 {auth_token} 会话 {session.thread_id} 聊天实例未初始化，正在创建...")
        session._chat_instance = client.start_chat() # 使用 client 启动聊天
        logger.info(f"认证令牌 {auth_token} 会话 {session.thread_id} 聊天实例初始化成功。")
    else:
        logger.info(f"认证令牌 {auth_token} 会话 {session.thread_id} 复用现有聊天实例。")
    
    session.last_active = time.time() # 更新最后活跃时间
    return session

def get_session(auth_token: str, thread_id: str) -> Optional[Session]:
    """根据认证令牌和会话ID获取会话。"""
    sessions = chat_sessions.get(auth_token)
    if sessions:
        session = sessions.get(thread_id)
        if session:
            session.last_active = time.time() # 更新最后活跃时间
        return session
    return None

def delete_session_by_id(auth_token: str, thread_id: str) -> bool:
    """根据认证令牌和会话ID删除会话。"""
    sessions = chat_sessions.get(auth_token)
    if sessions and thread_id in sessions:
        del sessions[thread_id]
        logger.info(f"认证令牌 {auth_token} 会话 {thread_id} 已删除。")
        return True
    return False

def list_all_sessions(auth_token: str) -> List[Dict[str, Any]]:
    """列出指定认证令牌下的所有会话。"""
    sessions = cleanup_expired_sessions(auth_token) # 清理并获取最新会话列表
    return [
        {
            "thread_id": tid,
            "name": session.name,
            "created_at": session.created_at,
            "last_active": session.last_active
        } for tid, session in sessions.items()
    ]