from datetime import datetime, timezone from typing import Optional, Union from sqlalchemy import asc, desc, or_ from core.app.entities.app_invoke_entities import InvokeFrom from core.llm_generator.llm_generator import LLMGenerator from extensions.ext_database import db from libs.infinite_scroll_pagination import InfiniteScrollPagination from models.account import Account from models.model import App, Conversation, EndUser, Message from services.errors.conversation import ConversationNotExistsError, LastConversationNotExistsError from services.errors.message import MessageNotExistsError class ConversationService: @classmethod def pagination_by_last_id( cls, app_model: App, user: Optional[Union[Account, EndUser]], last_id: Optional[str], limit: int, invoke_from: InvokeFrom, include_ids: Optional[list] = None, exclude_ids: Optional[list] = None, sort_by: str = "-updated_at", ) -> InfiniteScrollPagination: if not user: return InfiniteScrollPagination(data=[], limit=limit, has_more=False) base_query = db.session.query(Conversation).filter( Conversation.is_deleted == False, Conversation.app_id == app_model.id, Conversation.from_source == ("api" if isinstance(user, EndUser) else "console"), Conversation.from_end_user_id == (user.id if isinstance(user, EndUser) else None), Conversation.from_account_id == (user.id if isinstance(user, Account) else None), or_(Conversation.invoke_from.is_(None), Conversation.invoke_from == invoke_from.value), ) if include_ids is not None: base_query = base_query.filter(Conversation.id.in_(include_ids)) if exclude_ids is not None: base_query = base_query.filter(~Conversation.id.in_(exclude_ids)) # define sort fields and directions sort_field, sort_direction = cls._get_sort_params(sort_by) if last_id: last_conversation = base_query.filter(Conversation.id == last_id).first() if not last_conversation: raise LastConversationNotExistsError() # build filters based on sorting filter_condition = cls._build_filter_condition(sort_field, sort_direction, last_conversation) base_query = base_query.filter(filter_condition) base_query = base_query.order_by(sort_direction(getattr(Conversation, sort_field))) conversations = base_query.limit(limit).all() has_more = False if len(conversations) == limit: current_page_last_conversation = conversations[-1] rest_filter_condition = cls._build_filter_condition( sort_field, sort_direction, current_page_last_conversation, is_next_page=True ) rest_count = base_query.filter(rest_filter_condition).count() if rest_count > 0: has_more = True return InfiniteScrollPagination(data=conversations, limit=limit, has_more=has_more) @classmethod def _get_sort_params(cls, sort_by: str) -> tuple[str, callable]: if sort_by.startswith("-"): return sort_by[1:], desc return sort_by, asc @classmethod def _build_filter_condition( cls, sort_field: str, sort_direction: callable, reference_conversation: Conversation, is_next_page: bool = False ): field_value = getattr(reference_conversation, sort_field) if (sort_direction == desc and not is_next_page) or (sort_direction == asc and is_next_page): return getattr(Conversation, sort_field) < field_value else: return getattr(Conversation, sort_field) > field_value @classmethod def rename( cls, app_model: App, conversation_id: str, user: Optional[Union[Account, EndUser]], name: str, auto_generate: bool, ): conversation = cls.get_conversation(app_model, conversation_id, user) if auto_generate: return cls.auto_generate_name(app_model, conversation) else: conversation.name = name conversation.updated_at = datetime.now(timezone.utc).replace(tzinfo=None) db.session.commit() return conversation @classmethod def auto_generate_name(cls, app_model: App, conversation: Conversation): # get conversation first message message = ( db.session.query(Message) .filter(Message.app_id == app_model.id, Message.conversation_id == conversation.id) .order_by(Message.created_at.asc()) .first() ) if not message: raise MessageNotExistsError() # generate conversation name try: name = LLMGenerator.generate_conversation_name( app_model.tenant_id, message.query, conversation.id, app_model.id ) conversation.name = name except: pass db.session.commit() return conversation @classmethod def get_conversation(cls, app_model: App, conversation_id: str, user: Optional[Union[Account, EndUser]]): conversation = ( db.session.query(Conversation) .filter( Conversation.id == conversation_id, Conversation.app_id == app_model.id, Conversation.from_source == ("api" if isinstance(user, EndUser) else "console"), Conversation.from_end_user_id == (user.id if isinstance(user, EndUser) else None), Conversation.from_account_id == (user.id if isinstance(user, Account) else None), Conversation.is_deleted == False, ) .first() ) if not conversation: raise ConversationNotExistsError() return conversation @classmethod def delete(cls, app_model: App, conversation_id: str, user: Optional[Union[Account, EndUser]]): conversation = cls.get_conversation(app_model, conversation_id, user) conversation.is_deleted = True db.session.commit()