# -*- coding: utf-8 -*- """ 数据库状态 """ from datetime import datetime, timedelta from random import choices from typing import Dict, List, Tuple from pydantic import TypeAdapter from pydantic_ai import ModelMessage, ModelMessagesTypeAdapter from pydantic_ai._uuid import uuid7 import reflex as rx from sqlalchemy import desc from sqlmodel import Field, JSON, SQLModel, select, update from application.domain_models import Conversation, Dialog, Thought class CaptchaRecord(SQLModel, table=True): """ 验证码记录 邮箱、验证码和失效时间联合作为主键 """ email: str = Field(..., primary_key=True, description="邮箱") captcha: 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): """ 用户记录 """ id: str = Field( default_factory=lambda: str(uuid7()), primary_key=True, description="用户唯一标识", ) email: str = Field(..., index=True, description="邮箱") class ConversationRecord(SQLModel, table=True): """ 会话记录 """ id: str = Field( default_factory=lambda: str(uuid7()), primary_key=True, description="会话记录唯一标识", ) user_id: str = Field(..., index=True, description="用户唯一标识") description: str = Field(..., description="会话描述") is_deleted: bool = Field( default=False, index=True, description="会话已删除:True 表示已删除,False 表示未删除", ) created_at: datetime = Field( default_factory=datetime.now, description="会话创建时间" ) class ThoughtRecord(SQLModel): """ 思考记录(不创建数据表) """ id: str = Field( default_factory=lambda: str(uuid7()), primary_key=True, description="思考唯一标识", ) type: str = Field(description="思考类型") content: str = Field(default="", description="思考内容") class DialogRecord(SQLModel, table=True): """ 对话记录 """ id: str = Field( default_factory=lambda: str(uuid7()), primary_key=True, description="对话唯一标识", ) conversation_id: str = Field(..., index=True, description="会话唯一标识") user_prompt: str = Field(..., description="用户提示词") thoughts: List[ThoughtRecord] = Field( default_factory=List, sa_type=JSON, description="思考列表" ) result_output: str = Field(default="", description="结果输出") # 运行结果表 class RunResultRecord(SQLModel, table=True): conversation_id: str = Field(primary_key=True, description="会话唯一标识") dialog_id: str = Field(primary_key=True, description="对话唯一标识") new_messages: str = Field(description="新增消息") # 思考领域模型类型适配器 ThoughtTypeAdapter = TypeAdapter(List[Thought]) 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.captcha async def verify_captcha(self, email: str, captcha: str) -> bool: """ 核验验证码 :param email: 邮箱 :param captcha: 验证码 :return: 是否核验成功,True 表示核验成功,False 表示核验失败(根据邮箱和验证码未查询到有效且未核验的记录,或失效) """ async with rx.asession() as session: result = await session.exec( select(CaptchaRecord).where( CaptchaRecord.email == email, CaptchaRecord.captcha == captcha, 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: 会话字典 """ records: Dict[str, Conversation] = {} async with rx.asession() as session: result = await session.exec( select(ConversationRecord, DialogRecord) .outerjoin(DialogRecord, ConversationRecord.id == DialogRecord.conversation_id) # type: ignore .where( ConversationRecord.user_id == user_id, ConversationRecord.is_deleted == False, ) .order_by(ConversationRecord.id, DialogRecord.id) ) for conversation_record, dialog_record in result.all(): record = records.setdefault( conversation_record.id, Conversation( id=conversation_record.id, description=conversation_record.description, created_at=conversation_record.created_at, ), ) if not dialog_record: continue record.dialogs.append( Dialog( id=dialog_record.id, user_prompt=dialog_record.user_prompt, thoughts=ThoughtTypeAdapter.validate_python( dialog_record.thoughts ), result_output=dialog_record.result_output, ) ) return records async def create_conversations_record( self, user_id: str, description: str = "新会话" ) -> Dict[str, Conversation]: """ 创建会话记录 :param user_id: 用户唯一标识 :param description: 会话描述,默认为"新会话" :return: 键为会话唯一标识,值为会话实例的会话字典 """ async with rx.asession() as session: record = ConversationRecord(user_id=user_id, description=description) session.add(record) await session.commit() await session.refresh(record) return { record.id: Conversation( id=record.id, description=record.description, created_at=record.created_at, ) } async def delete_conversations_record(self, conversation_id: str) -> None: """ 逻辑删除会话记录(将会话已删除设置为 True) :param conversation_id: 指定会话唯一标识 :return: None """ async with rx.asession() as session: record = await session.get(ConversationRecord, conversation_id) if not record: return record.is_deleted = True await session.commit() async def create_dialogs_record(self, conversation_id: str, question: str) -> str: """ 创建对话记录 :param conversation_id: 会话唯一标识 :param question: 问题 :return: 创建对话记录的唯一标识 """ async with rx.asession() as session: record = DialogRecord( conversation_id=conversation_id, user_prompt=question, ) session.add(record) await session.commit() await session.refresh(record) return record.id async def update_dialog_record( self, dialog_id: str, thoughts: List[Thought], result_output: str, ) -> None: """ 更新对话记录 :param dialog_id: 对话唯一标识 :param thoughts: 思考列表 :param result_output: 结果输出 :return: None """ async with rx.asession() as session: record = await session.get(DialogRecord, dialog_id) if not record: return record.thoughts = ThoughtTypeAdapter.dump_python(thoughts) record.result_output = result_output await session.commit() async def create_run_result_record( self, conversation_id: str, dialog_id: str, new_messages: List[ModelMessage], ) -> None: """ 创建运行结果记录 :param conversation_id: 会话唯一标识 :param dialog_id: 对话唯一标识 :param new_messages: 新增消息 :return: None """ async with rx.asession() as session: session.add( RunResultRecord( conversation_id=conversation_id, dialog_id=dialog_id, new_messages=ModelMessagesTypeAdapter.dump_json( new_messages ).decode( "utf-8" ), # 序列化为 JSON 字符串 ) ) await session.commit() async def retrieve_message_history( self, conversation_id: str ) -> List[ModelMessage]: """ 获取消息历史 :param conversation_id: 会话唯一标识 :return: 消息历史 """ records: List[ModelMessage] = [] async with rx.asession() as session: result = await session.exec( select(RunResultRecord) .where(RunResultRecord.conversation_id == conversation_id) .order_by(desc(RunResultRecord.dialog_id)) ) for record in result.all(): records.extend( ModelMessagesTypeAdapter.validate_json(record.new_messages) ) return records