Method execute

src/langbot/pkg/command/cmdmgr.py:76–120

执行命令

(
        self,
        command_text: str,
        full_command_text: str,
        query: pipeline_query.Query,
        session: provider_session.Session,
    )

            yield command_context.CommandReturn(error=command_errors.CommandNotFoundError(context.command))

    async def execute(
        self,
        command_text: str,
        full_command_text: str,
        query: pipeline_query.Query,
        session: provider_session.Session,
    ) -> typing.AsyncGenerator[command_context.CommandReturn, None]:
        """执行命令"""

        privilege = 1

        import sqlalchemy as _sa
        from ..entity.persistence.bot import BotAdmin as _BotAdmin

        _admins = await self.ap.persistence_mgr.execute_async(
            _sa.select(_BotAdmin).where(
                _BotAdmin.bot_uuid == (query.bot_uuid or ''),
                _BotAdmin.launcher_type == query.launcher_type.value,
                _BotAdmin.launcher_id == str(query.launcher_id),
            )
        )
        if _admins.first() is not None:
            privilege = 2

        ctx = command_context.ExecuteContext(
            query_id=query.query_id,
            session=session,
            command_text=command_text,
            full_command_text=full_command_text,
            command='',
            crt_command='',
            params=command_text.split(' '),
            crt_params=command_text.split(' '),
            privilege=privilege,
        )

        ctx.command = ctx.params[0]

        ctx.shift()

        # Get bound plugins from query
        bound_plugins = query.variables.get('_pipeline_bound_plugins', None)

        async for ret in self._execute(ctx, self.cmd_list, bound_plugins=bound_plugins):
            yield ret