# -*- coding: utf-8 -*- """ 数据库状态 """ from datetime import datetime, timedelta from random import choices from typing import Any from pydantic_ai import ModelMessage, ModelMessagesTypeAdapter, RunUsage from pydantic_ai._uuid import uuid7 import reflex as rx from sqlmodel import Field, JSON, SQLModel, select, update from application.states.models import ( Conversation, MessageType, Message, usage_dump_python, ) class CaptchaRecord(SQLModel, table=True, table_name="captcha_code"): """ 验证码记录 """ email: str = Field(..., primary_key=True, description="邮箱") code: str = Field( default_factory=lambda: "".join(choices("0123456789", k=6)), primary_key=True, description="验证码", ) is_valid: bool = Field( default=True, index=True, description="验证码有效:True 表示有效,False 表示无效", ) is_verified: bool = Field( default=False, index=True, description="验证码已核验:True 表示已核验,False 表示未核验", ) expired_at: datetime = Field( default_factory=lambda: datetime.now() + timedelta(minutes=30), primary_key=True, description="过期时间", ) class UserRecord(SQLModel, table=True, table_name="user"): """ 用户记录 """ id: str = Field( default_factory=lambda: str(uuid7()), primary_key=True, description="用户唯一标识", ) email: str = Field(..., index=True, description="邮箱") class ConversationRecord(SQLModel, table=True, table_name="conversation"): """ 会话记录 """ id: str = Field( default_factory=lambda: str(uuid7()), primary_key=True, description="会话唯一标识", ) user_id: str = Field(..., index=True, description="用户唯一标识") description: str = Field(default="新会话", description="会话描述") usage: dict[str, Any] = Field( default_factory=dict, sa_type=JSON, description="会话使用量" ) is_deleted: bool = Field( default=False, index=True, description="会话已删除:True 表示已删除,False 表示未删除", ) created_at: datetime = Field( default_factory=datetime.now, description="会话创建时间" ) class MessageRecord(SQLModel, table=True, table_name="message"): """ 消息记录 """ id: str = Field( default_factory=lambda: str(uuid7()), primary_key=True, description="消息唯一标识", ) conversation_id: str = Field(..., index=True, description="会话唯一标识") type: MessageType = Field(..., description="消息类型") title: str = Field(default="", description="消息标题") content: str = Field(default="", description="消息内容") class RunRecord(SQLModel, table=True, table_name="run"): """ 运行记录 """ id: str = Field( default_factory=lambda: str(uuid7()), primary_key=True, description="运行唯一标识", ) conversation_id: str = Field(..., index=True, description="会话唯一标识") new_messages: str = Field( default_factory=list, sa_type=JSON, description="新增消息" ) class DatabaseState(rx.State): """ 数据库状态 """ async def create_captcha_record(self, email: str) -> str: """ 创建验证记录 :param email: 邮箱 :return: 验证码 """ async with rx.asession() as session: # 先将该邮箱有效、未核验的验证码记录设置为无效 await session.exec( update(CaptchaRecord) .where( CaptchaRecord.email == email, # type: ignore CaptchaRecord.is_valid == True, # type: ignore CaptchaRecord.is_verified == False, # type: ignore ) .values(is_valid=False) ) await session.flush() # 创建验证记录 record = CaptchaRecord( email=email, ) session.add(record) await session.commit() await session.refresh(record) return record.code async def verify_captcha_code(self, email: str, captcha_code: str) -> bool: """ 核验验证码 :param email: 邮箱 :param code: 验证码 :return: 核验是否成功,True 表示核验成功,False 表示核验失败 """ async with rx.asession() as session: result = await session.exec( select(CaptchaRecord).where( CaptchaRecord.email == email, CaptchaRecord.code == captcha_code, CaptchaRecord.is_valid == True, CaptchaRecord.is_verified == False, CaptchaRecord.expired_at > datetime.now(), ) ) record = result.first() if not record: return False record.is_verified = True await session.commit() await session.refresh(record) return True async def create_user_record(self, email: str) -> str: """ 创建用户记录 :param email: 邮箱 :return: 用户唯一标识 """ async with rx.asession() as session: result = await session.exec( select(UserRecord).where(UserRecord.email == email) ) record = result.first() # 若用户记录已存在则返回用户唯一标识,否则先创建用户记录再返回用户唯一标识 if record: return record.id record = UserRecord(email=email) session.add(record) await session.commit() await session.refresh(record) return record.id async def get_conversations(self, user_id: str) -> dict[str, Conversation]: """ 获取当前用户的会话字典 :param user_id: 用户唯一标识 :return: 当前用户的会话字典 """ conversations: dict[str, Conversation] = {} async with rx.asession() as session: result = await session.exec( select(ConversationRecord, MessageRecord) .outerjoin(MessageRecord, MessageRecord.conversation_id == ConversationRecord.id) # type: ignore .where( ConversationRecord.user_id == user_id, ConversationRecord.is_deleted == False, ) .order_by(ConversationRecord.id, MessageRecord.id) ) for ( conversation_record, message_record, ) in result.all(): conversation = conversations.setdefault( conversation_record.id, Conversation( id=conversation_record.id, description=conversation_record.description, usage=conversation_record.usage, messages={}, created_at=conversation_record.created_at, ), ) if not message_record: continue conversation.messages.setdefault( message_record.id, Message( id=message_record.id, type=message_record.type, title=message_record.title, content=message_record.content, ), ) return conversations async def create_conversation_record(self, user_id: str) -> dict[str, Conversation]: """ 创建会话记录 :param user_id: 用户唯一标识 :return: 会话实例 """ async with rx.asession() as session: record = ConversationRecord(user_id=user_id) session.add(record) await session.commit() await session.refresh(record) return { record.id: Conversation( id=record.id, description=record.description, usage=record.usage, messages={}, # 新会话默认消息字典为空 created_at=record.created_at, ) } async def update_conversation_record( self, conversation_id: str, description: str | None = None, is_deleted: bool | None = None, usage: RunUsage | None = None, ) -> None: """ 更新会话记录 :param conversation_id: 指定会话唯一标识 :param description: 会话描述 :param is_deleted: 会话已删除 :param usage: 使用量 :return: None """ async with rx.asession() as session: record = await session.get(ConversationRecord, conversation_id) if not record: return if isinstance(description, str): record.description = description if isinstance(is_deleted, bool): record.is_deleted = is_deleted if isinstance(usage, RunUsage): record.usage = usage_dump_python(usage) await session.commit() async def create_message_record( self, conversation_id: str, message: Message, ) -> dict[str, Message]: """ 创建消息记录 :param conversation_id: 会话唯一标识 :param message: 消息实例 :return: 消息实例 """ async with rx.asession() as session: record = MessageRecord( id=message.id, conversation_id=conversation_id, type=message.type, title=message.title, content=message.content, ) session.add(record) await session.commit() await session.refresh(record) return {message.id: message} async def get_message_history(self, conversation_id: str) -> list[ModelMessage]: """ 获取消息历史列表 :param conversation_id: 会话唯一标识 :return: 消息历史 """ message_history: list[ModelMessage] = [] async with rx.asession() as session: result = await session.exec( select(RunRecord) .where(RunRecord.conversation_id == conversation_id) .order_by(RunRecord.id) ) for run in result.all(): message_history.extend( ModelMessagesTypeAdapter.validate_json( run.new_messages ) # 将 run.new_messages 由 JSON 字符串反序列化为 List[ModelMessage] ) return message_history async def create_run_record( self, conversation_id: str, new_messages: list[ModelMessage], ) -> None: """ 创建运行记录 :param conversation_id: 会话唯一标识 :param new_messages: 新增消息 :return: None """ async with rx.asession() as session: session.add( RunRecord( conversation_id=conversation_id, new_messages=ModelMessagesTypeAdapter.dump_json( new_messages ).decode(), # 将 messages 由 List[ModelMessage] 序列化为 JSON 字符串 ) ) await session.commit()