Underground-Digital/Workflow-Engine

1from typing import Optional, Union2 3from core.app.entities.app_invoke_entities import InvokeFrom4from extensions.ext_database import db5from libs.infinite_scroll_pagination import InfiniteScrollPagination6from models.account import Account7from models.model import App, EndUser8from models.web import PinnedConversation9from services.conversation_service import ConversationService10 11 12class WebConversationService:13    @classmethod14    def pagination_by_last_id(15        cls,16        app_model: App,17        user: Optional[Union[Account, EndUser]],18        last_id: Optional[str],19        limit: int,20        invoke_from: InvokeFrom,21        pinned: Optional[bool] = None,22        sort_by="-updated_at",23    ) -> InfiniteScrollPagination:24        include_ids = None25        exclude_ids = None26        if pinned is not None:27            pinned_conversations = (28                db.session.query(PinnedConversation)29                .filter(30                    PinnedConversation.app_id == app_model.id,31                    PinnedConversation.created_by_role == ("account" if isinstance(user, Account) else "end_user"),32                    PinnedConversation.created_by == user.id,33                )34                .order_by(PinnedConversation.created_at.desc())35                .all()36            )37            pinned_conversation_ids = [pc.conversation_id for pc in pinned_conversations]38            if pinned:39                include_ids = pinned_conversation_ids40            else:41                exclude_ids = pinned_conversation_ids42 43        return ConversationService.pagination_by_last_id(44            app_model=app_model,45            user=user,46            last_id=last_id,47            limit=limit,48            invoke_from=invoke_from,49            include_ids=include_ids,50            exclude_ids=exclude_ids,51            sort_by=sort_by,52        )53 54    @classmethod55    def pin(cls, app_model: App, conversation_id: str, user: Optional[Union[Account, EndUser]]):56        pinned_conversation = (57            db.session.query(PinnedConversation)58            .filter(59                PinnedConversation.app_id == app_model.id,60                PinnedConversation.conversation_id == conversation_id,61                PinnedConversation.created_by_role == ("account" if isinstance(user, Account) else "end_user"),62                PinnedConversation.created_by == user.id,63            )64            .first()65        )66 67        if pinned_conversation:68            return69 70        conversation = ConversationService.get_conversation(71            app_model=app_model, conversation_id=conversation_id, user=user72        )73 74        pinned_conversation = PinnedConversation(75            app_id=app_model.id,76            conversation_id=conversation.id,77            created_by_role="account" if isinstance(user, Account) else "end_user",78            created_by=user.id,79        )80 81        db.session.add(pinned_conversation)82        db.session.commit()83 84    @classmethod85    def unpin(cls, app_model: App, conversation_id: str, user: Optional[Union[Account, EndUser]]):86        pinned_conversation = (87            db.session.query(PinnedConversation)88            .filter(89                PinnedConversation.app_id == app_model.id,90                PinnedConversation.conversation_id == conversation_id,91                PinnedConversation.created_by_role == ("account" if isinstance(user, Account) else "end_user"),92                PinnedConversation.created_by == user.id,93            )94            .first()95        )96 97        if not pinned_conversation:98            return99 100        db.session.delete(pinned_conversation)101        db.session.commit()102