Source code for mi.actions.user

from __future__ import annotations

from typing import TYPE_CHECKING, List, Optional

from aiocache import Cache, cached
from mi.exception import NotExistRequiredData, NotExistRequiredParameters
from mi.framework.http import HTTPSession
from mi.framework.models.note import Note
from mi.framework.router import Route
from mi.utils import check_multi_arg, get_cache_key, key_builder, remove_dict_empty
from mi.wrapper.chat import ChatManager
from mi.wrapper.follow import FollowManager, FollowRequestManager
from mi.wrapper.models.note import RawNote
from mi.wrapper.models.user import RawUser
from mi.wrapper.note import NoteManager
from mi.wrapper.user import AdminUserManager

if TYPE_CHECKING:
    from mi.framework.models.user import User

__all__ = ['UserActions']


[docs]class UserActions: def __init__( self, user_id: Optional[str] = None, user: Optional[User] = None ): self.__user: User = user self.admin = AdminUserManager(user_id=user_id) self.note: NoteManager() self.follow: FollowManager = FollowManager(user_id=user_id) self.follow_request: FollowRequestManager = FollowRequestManager(user_id=user_id) self.chat: ChatManager = ChatManager(user_id=user_id) def _get_chat_instance(self, message_id: Optional[str] = None, user_id: Optional[str] = None) -> ChatManager: return ChatManager(user_id=self.__user.id, message_id=message_id)
[docs] @cached(ttl=10, namespace='get_user', key_builder=key_builder) async def get(self, user_id: Optional[str] = None, username: Optional[str] = None, host: Optional[str] = None) -> User: """ ユーザーのプロフィールを取得します。一度のみサーバーにアクセスしキャッシュをその後は使います。 fetch_userを使った場合はキャッシュが廃棄され再度サーバーにアクセスします。 Parameters ---------- user_id : str 取得したいユーザーのユーザーID username : str 取得したいユーザーのユーザー名 host : str, default=None 取得したいユーザーがいるインスタンスのhost Returns ------- User ユーザー情報 """ field = remove_dict_empty({"userId": user_id, "username": username, "host": host}) data = await HTTPSession.request(Route('POST', '/api/users/show'), json=field, auth=True, lower=True) return User(RawUser(data))
@get_cache_key async def fetch(self, user_id: Optional[str] = None, username: Optional[str] = None, host: Optional[str] = None, **kwargs) -> User: """ サーバーにアクセスし、ユーザーのプロフィールを取得します。基本的には get_userをお使いください。 Parameters ---------- user_id : str 取得したいユーザーのユーザーID username : str 取得したいユーザーのユーザー名 host : str, default=None 取得したいユーザーがいるインスタンスのhost Returns ------- User ユーザー情報 """ if not check_multi_arg(user_id, username): raise NotExistRequiredParameters("user_id, usernameどちらかは必須です") field = remove_dict_empty({"userId": user_id, "username": username, "host": host}) data = await HTTPSession.request(Route('POST', '/api/users/show'), json=field, auth=True, lower=True) old_cache = Cache(namespace='get_user') await old_cache.delete(kwargs['cache_key'].format('get_user')) return User(RawUser(data)) async def get_notes( self, user_id: Optional[str] = None, include_replies: bool = True, limit: int = 10, since_id: Optional[str] = None, until_id: Optional[str] = None, since_date: int = 0, until_date: int = 0, include_my_renotes: bool = True, with_files: bool = False, file_type: Optional[List[str]] = None, exclude_nsfw: bool = True ) -> List[Note]: user_id = user_id or self.__user.id data = { 'userId': user_id, 'includeReplies': include_replies, 'limit': limit, 'sinceId': since_id, 'untilId': until_id, 'sinceDate': since_date, 'untilDate': until_date, 'includeMyRenotes': include_my_renotes, 'withFiles': with_files, 'fileType': file_type, 'excludeNsfw': exclude_nsfw } res = await HTTPSession.request(Route('POST', '/api/users/notes'), json=data, auth=True, lower=True) return [Note(RawNote(i)) for i in res]
[docs] def get_mention(self, user: Optional[User] = None) -> str: """ Get mention name of user. Parameters ---------- user : Optional[User], default=None メンションを取得したいユーザーのオブジェクト Returns ------- str メンション """ user = user or self.__user if user is None: raise NotExistRequiredData('Required parameters: user') return f'@{user.name}@{user.host}' if user.instance else f'@{user.name}'
def get_follow(self, user_id: str) -> FollowManager: return FollowManager(user_id=user_id) def get_follow_request(self, user_id: str) -> FollowRequestManager: return FollowRequestManager(user_id=user_id)