mirror_maker
/
mipac
mirror of https://github.com/yupix/mipac
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

chore: axblackを使用してフォーマット

Browse Source
renovate/configure
yupix 2 years ago
parent 3ed1417b5c
commit 8eb34b321a
No known key found for this signature in database
GPG Key ID: 2FF705F5C56D9C06
37 changed files with 1014 additions and 487 deletions
Show Stats Download Patch File Download Diff File

6
mipac/__init__.py
Unescape View File

@ -1,11 +1,11 @@
__title__ = 'mipac'
__author__ = 'yupix'
__license__ = 'MIT'
__copyright__ = "Copyright 2022-present yupix"
__author_email__ = "yupi0982@outlook.jp"
__copyright__ = 'Copyright 2022-present yupix'
__author_email__ = 'yupi0982@outlook.jp'
__version__ = '0.0.1'
__path__ = __import__("pkgutil").extend_path(__path__, __name__)
__path__ = __import__('pkgutil').extend_path(__path__, __name__)
from .models import *
from .abc import *

178
mipac/actions/note.py
Unescape View File

@ -18,7 +18,13 @@ if TYPE_CHECKING:
class NoteActions:
def __init__(self, note_id: Optional[str] = None, *, session: HTTPClient, client: ClientActions):
def __init__(
self,
note_id: Optional[str] = None,
*,
session: HTTPClient,
client: ClientActions
):
self.__note_id: Optional[str] = note_id
self.favorite = FavoriteManager(note_id=note_id, session=session, client=client)
self.reaction = ReactionManager(note_id=note_id, session=session, client=client)
@ -47,23 +53,28 @@ class NoteActions:
note_id = note_id or self.__note_id
data = {'noteId': note_id, 'clipId': clip_id}
return bool(await self.__session.request(Route('POST', '/api/clips/add-note'), json=data, auth=True))
async def send(self,
content: Optional[str] = None,
visibility: str = "public",
visible_user_ids: Optional[List[str]] = None,
cw: Optional[str] = None,
local_only: bool = False,
extract_mentions: bool = True,
extract_hashtags: bool = True,
extract_emojis: bool = True,
reply_id: Optional[str] = None,
renote_id: Optional[str] = None,
channel_id: Optional[str] = None,
files: Optional[List[MiFile]] = None,
poll: Optional[Poll] = None
) -> Note:
return bool(
await self.__session.request(
Route('POST', '/api/clips/add-note'), json=data, auth=True
)
)
async def send(
self,
content: Optional[str] = None,
visibility: str = 'public',
visible_user_ids: Optional[List[str]] = None,
cw: Optional[str] = None,
local_only: bool = False,
extract_mentions: bool = True,
extract_hashtags: bool = True,
extract_emojis: bool = True,
reply_id: Optional[str] = None,
renote_id: Optional[str] = None,
channel_id: Optional[str] = None,
files: Optional[List[MiFile]] = None,
poll: Optional[Poll] = None,
) -> Note:
"""
ノートを投稿します。
@ -108,35 +119,41 @@ class NoteActions:
"""
field = {
"visibility": visibility,
"visibleUserIds": visible_user_ids,
"text": content,
"cw": cw,
"localOnly": local_only,
"noExtractMentions": not extract_mentions,
"noExtractHashtags": not extract_hashtags,
"noExtractEmojis": not extract_emojis,
"replyId": reply_id,
"renoteId": renote_id,
"channelId": channel_id
'visibility': visibility,
'visibleUserIds': visible_user_ids,
'text': content,
'cw': cw,
'localOnly': local_only,
'noExtractMentions': not extract_mentions,
'noExtractHashtags': not extract_hashtags,
'noExtractEmojis': not extract_emojis,
'replyId': reply_id,
'renoteId': renote_id,
'channelId': channel_id,
}
if not check_multi_arg(content, files, renote_id, poll):
raise ParameterError("ノートの送信にはcontent, file_ids, renote_id またはpollのいずれか1つが無くてはいけません")
raise ParameterError(
'ノートの送信にはcontent, file_ids, renote_id またはpollのいずれか1つが無くてはいけません'
)
if poll and type(Poll):
poll_data = remove_dict_empty({
'choices': poll.choices,
'multiple': poll.multiple,
'expiresAt': poll.expires_at,
'expiredAfter': poll.expired_after
})
field["poll"] = poll_data
poll_data = remove_dict_empty(
{
'choices': poll.choices,
'multiple': poll.multiple,
'expiresAt': poll.expires_at,
'expiredAfter': poll.expired_after,
}
)
field['poll'] = poll_data
# if files: # TODO: get_file_idsを直さないと使えない
# field['fileIds'] = await get_file_ids(files=files)
# field['fileIds'] = await get_file_ids(files=files)
field = remove_dict_empty(field)
res = await self.__session.request(Route('POST', '/api/notes/create'), json=field, auth=True, lower=True)
return Note(RawNote(res["created_note"]), client=self.__client)
res = await self.__session.request(
Route('POST', '/api/notes/create'), json=field, auth=True, lower=True
)
return Note(RawNote(res['created_note']), client=self.__client)
async def delete(self, note_id: Optional[str] = None) -> bool:
"""
@ -155,8 +172,10 @@ class NoteActions:
note_id = note_id or self.__note_id
data = {"noteId": note_id}
res = await self.__session.request(Route('POST', '/api/notes/delete'), json=data, auth=True)
data = {'noteId': note_id}
res = await self.__session.request(
Route('POST', '/api/notes/delete'), json=data, auth=True
)
return bool(res)
async def create_renote(self, note_id: Optional[str] = None) -> Note:
@ -177,18 +196,18 @@ class NoteActions:
return await self.send(renote_id=note_id)
async def create_quote(
self,
content: Optional[str] = None,
visibility: str = 'public',
visible_user_ids: Optional[List[str]] = None,
cw: Optional[str] = None,
local_only: bool = False,
extract_mentions: bool = True,
extract_hashtags: bool = True,
extract_emojis: bool = True,
file_ids: Optional[List[str]] = None,
poll: Optional[Poll] = None,
note_id: Optional[str] = None,
self,
content: Optional[str] = None,
visibility: str = 'public',
visible_user_ids: Optional[List[str]] = None,
cw: Optional[str] = None,
local_only: bool = False,
extract_mentions: bool = True,
extract_hashtags: bool = True,
extract_emojis: bool = True,
file_ids: Optional[List[str]] = None,
poll: Optional[Poll] = None,
note_id: Optional[str] = None,
) -> Note:
"""
Create a note quote.
@ -221,10 +240,18 @@ class NoteActions:
note_id = note_id or self.__note_id
return await self.send(content=content, visibility=visibility, visible_user_ids=visible_user_ids, cw=cw,
local_only=local_only, extract_mentions=extract_mentions,
extract_hashtags=extract_hashtags,
extract_emojis=extract_emojis, renote_id=note_id, poll=poll) # TODO: filesを受け取るように
return await self.send(
content=content,
visibility=visibility,
visible_user_ids=visible_user_ids,
cw=cw,
local_only=local_only,
extract_mentions=extract_mentions,
extract_hashtags=extract_hashtags,
extract_emojis=extract_emojis,
renote_id=note_id,
poll=poll,
) # TODO: filesを受け取るように
async def get_note(self, note_id: Optional[str] = None) -> Note:
"""
@ -241,15 +268,20 @@ class NoteActions:
取得したノートID
"""
note_id = note_id or self.__note_id
res = await self.__session.request(Route('POST', '/api/notes/show'), json={"noteId": note_id}, auth=True, lower=True)
res = await self.__session.request(
Route('POST', '/api/notes/show'),
json={'noteId': note_id},
auth=True,
lower=True,
)
return Note(RawNote(res), client=self.__client)
async def get_replies(
self,
since_id: Optional[str] = None,
until_id: Optional[str] = None,
limit: int = 10,
note_id: Optional[str] = None
self,
since_id: Optional[str] = None,
until_id: Optional[str] = None,
limit: int = 10,
note_id: Optional[str] = None,
) -> List[Note]:
"""
ノートに対する返信を取得します
@ -271,12 +303,22 @@ class NoteActions:
返信のリスト
"""
note_id = note_id or self.__note_id
res = await self.__session.request(Route('POST', '/api/notes/replies'),
json={"noteId": note_id, "sinceId": since_id, "untilId": until_id, "limit": limit},
auth=True, lower=True)
res = await self.__session.request(
Route('POST', '/api/notes/replies'),
json={
'noteId': note_id,
'sinceId': since_id,
'untilId': until_id,
'limit': limit,
},
auth=True,
lower=True,
)
return [Note(RawNote(i), client=self.__client) for i in res]
async def get_reaction(self, reaction: str, note_id: Optional[str] = None) -> List[NoteReaction]:
async def get_reaction(
self, reaction: str, note_id: Optional[str] = None
) -> List[NoteReaction]:
note_id = note_id or self.__note_id
# return await ReactionManager(note_id=note_id, client=self.__client).get_reaction(reaction)
# TODO: self.__client形式に置き換え

71
mipac/actions/user.py
Unescape View File

@ -20,10 +20,7 @@ if TYPE_CHECKING:
class UserActions:
def __init__(
self,
session: HTTPClient,
client: ClientActions,
user: Optional[User] = None
self, session: HTTPClient, client: ClientActions, user: Optional[User] = None
):
self.__session: HTTPClient = session
self.__user: User = user
@ -31,7 +28,12 @@ class UserActions:
self.note: NoteManager(session=session, client=client)
@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:
async def get(
self,
user_id: Optional[str] = None,
username: Optional[str] = None,
host: Optional[str] = None,
) -> User:
"""
ユーザーのプロフィールを取得します。一度のみサーバーにアクセスしキャッシュをその後は使います。
fetch_userを使った場合はキャッシュが廃棄され再度サーバーにアクセスします。
@ -51,13 +53,22 @@ class UserActions:
ユーザー情報
"""
field = remove_dict_empty({"userId": user_id, "username": username, "host": host})
data = await self.__session.request(Route('POST', '/api/users/show'), json=field, auth=True, lower=True)
field = remove_dict_empty(
{'userId': user_id, 'username': username, 'host': host}
)
data = await self.__session.request(
Route('POST', '/api/users/show'), json=field, auth=True, lower=True
)
return User(RawUser(data), client=self.__client)
@get_cache_key
async def fetch(self, user_id: Optional[str] = None, username: Optional[str] = None,
host: Optional[str] = None, **kwargs) -> User:
async def fetch(
self,
user_id: Optional[str] = None,
username: Optional[str] = None,
host: Optional[str] = None,
**kwargs,
) -> User:
"""
サーバーにアクセスし、ユーザーのプロフィールを取得します。基本的には get_userをお使いください。
@ -76,27 +87,31 @@ class UserActions:
ユーザー情報
"""
if not check_multi_arg(user_id, username):
raise ParameterError("user_id, usernameどちらかは必須です")
field = remove_dict_empty({"userId": user_id, "username": username, "host": host})
data = await self.__session.request(Route('POST', '/api/users/show'), json=field, auth=True, lower=True)
raise ParameterError('user_id, usernameどちらかは必須です')
field = remove_dict_empty(
{'userId': user_id, 'username': username, 'host': host}
)
data = await self.__session.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), client=self.__client)
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
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 = {
@ -110,9 +125,11 @@ class UserActions:
'includeMyRenotes': include_my_renotes,
'withFiles': with_files,
'fileType': file_type,
'excludeNsfw': exclude_nsfw
'excludeNsfw': exclude_nsfw,
}
res = await self.__session.request(Route('POST', '/api/users/notes'), json=data, auth=True, lower=True)
res = await self.__session.request(
Route('POST', '/api/users/notes'), json=data, auth=True, lower=True
)
return [Note(RawNote(i), client=self.__client) for i in res]
def get_mention(self, user: Optional[User] = None) -> str:

1
mipac/client.py
Unescape View File

@ -14,4 +14,3 @@ class Client:
async def close_session(self):
await self.session.close_session()

18
mipac/core/models/chart.py
Unescape View File

@ -1,8 +1,18 @@
__all__ = ('RawActiveUsersChart', 'RawDriveRemoteChart', 'RawDriveLocalChart','RawDriveChart')
__all__ = (
'RawActiveUsersChart',
'RawDriveRemoteChart',
'RawDriveLocalChart',
'RawDriveChart',
)
from typing import List
from mipac.types import ActiveUsersChartPayload, DriveChartPayload, DriveLocalChartPayload, DriveRemoteChartPayload
from mipac.types import (
ActiveUsersChartPayload,
DriveChartPayload,
DriveLocalChartPayload,
DriveRemoteChartPayload,
)
class RawActiveUsersChart:
@ -37,7 +47,7 @@ class RawDriveLocalChart:
'inc_count',
'inc_size',
'dec_count',
'dec_size'
'dec_size',
)
def __init__(self, data: DriveLocalChartPayload):
@ -56,7 +66,7 @@ class RawDriveRemoteChart:
'inc_count',
'inc_size',
'dec_count',
'dec_size'
'dec_size',
)
def __init__(self, data: DriveRemoteChartPayload):

39
mipac/core/models/chat.py
Unescape View File

@ -31,8 +31,19 @@ class RawChat:
reads : List
"""
__slots__ = ('id', 'created_at', 'content', 'user_id', 'author', 'recipient_id', 'recipient', 'group_id', 'file_id',
'is_read', 'reads')
__slots__ = (
'id',
'created_at',
'content',
'user_id',
'author',
'recipient_id',
'recipient',
'group_id',
'file_id',
'is_read',
'reads',
)
def __init__(self, data: ChatPayload):
"""
@ -41,14 +52,16 @@ class RawChat:
----------
data: ChatPayload
"""
self.id: str = data["id"]
self.created_at: datetime = datetime.strptime(data["created_at"], '%Y-%m-%dT%H:%M:%S.%fZ')
self.content: str = data["text"]
self.user_id: str = data["user_id"]
self.author: RawUser = RawUser(data["user"])
self.recipient_id: str = data["recipient_id"]
self.recipient: str = data["recipient"]
self.group_id: str = data["group_id"]
self.file_id: str = data["file_id"]
self.is_read: bool = bool(data["is_read"])
self.reads: List = data["reads"]
self.id: str = data['id']
self.created_at: datetime = datetime.strptime(
data['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ'
)
self.content: str = data['text']
self.user_id: str = data['user_id']
self.author: RawUser = RawUser(data['user'])
self.recipient_id: str = data['recipient_id']
self.recipient: str = data['recipient']
self.group_id: str = data['group_id']
self.file_id: str = data['file_id']
self.is_read: bool = bool(data['is_read'])
self.reads: List = data['reads']

34
mipac/core/models/drive.py
Unescape View File

@ -46,7 +46,9 @@ class RawFolder:
def __init__(self, data: FolderPayload):
self.id: str = data['id']
self.created_at: datetime = datetime.strptime(data["created_at"], '%Y-%m-%dT%H:%M:%S.%fZ')
self.created_at: datetime = datetime.strptime(
data['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ'
)
self.name: str = data['name']
self.folders_count: Optional[int] = data.get('folders_count', 0)
self.parent_id: str = data['parent_id']
@ -93,24 +95,44 @@ class RawFile:
"""
__slots__ = (
'id', 'created_at', 'name', 'type', 'md5', 'size', 'is_sensitive', 'blurhash', 'properties', 'url', 'thumbnail_url',
'comment', 'folder_id', 'folder', 'user_id', 'user'
'id',
'created_at',
'name',
'type',
'md5',
'size',
'is_sensitive',
'blurhash',
'properties',
'url',
'thumbnail_url',
'comment',
'folder_id',
'folder',
'user_id',
'user',
)
def __init__(self, data: FilePayload):
self.id: str = data['id']
self.created_at: datetime = datetime.strptime(data["created_at"], '%Y-%m-%dT%H:%M:%S.%fZ')
self.created_at: datetime = datetime.strptime(
data['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ'
)
self.name: str = data['name']
self.type: str = data['type']
self.md5: str = data['md5']
self.size: int = data['size']
self.is_sensitive: bool = data['is_sensitive']
self.blurhash: str = data['blurhash']
self.properties: Optional[RawProperties] = RawProperties(data['properties']) if len(data.get('properties')) else None
self.properties: Optional[RawProperties] = RawProperties(
data['properties']
) if len(data.get('properties')) else None
self.url: str = data['url']
self.thumbnail_url: str = data['thumbnail_url']
self.comment: str = data['comment']
self.folder_id: str = data['folder_id']
self.folder: Optional[RawFolder] = RawFolder(data['folder']) if data.get('folder') else None
self.folder: Optional[RawFolder] = RawFolder(data['folder']) if data.get(
'folder'
) else None
self.user_id: str = data['user_id']
self.user: Dict[str, Any] = data['user']

10
mipac/core/models/instance.py
Unescape View File

@ -18,7 +18,15 @@ class RawInstance:
theme_color : Optional[str], default=None
"""
__slots__ = ('host', 'name', 'software_name', 'software_version', 'icon_url', 'favicon_url', 'theme_color')
__slots__ = (
'host',
'name',
'software_name',
'software_version',
'icon_url',
'favicon_url',
'theme_color',
)
def __init__(self, data: InstancePayload):
self.host: Optional[str] = data.get('host')

175
mipac/core/models/note.py
Unescape View File

@ -32,29 +32,51 @@ class RawRenote:
poll Optional[RawPoll]
"""
__slots__ = ('id', 'created_at', 'user_id', 'user', 'content', 'cw', 'visibility', 'renote_count', 'replies_count',
'replies_count', 'reactions', 'emojis', 'file_ids', 'files', 'reply_id', 'renote_id', 'uri', 'poll')
__slots__ = (
'id',
'created_at',
'user_id',
'user',
'content',
'cw',
'visibility',
'renote_count',
'replies_count',
'replies_count',
'reactions',
'emojis',
'file_ids',
'files',
'reply_id',
'renote_id',
'uri',
'poll',
)
def __init__(self, data: RenotePayload):
self.id: str = data["id"]
self.created_at: datetime = datetime.strptime(data["created_at"], '%Y-%m-%dT%H:%M:%S.%fZ')
self.user_id: str = data["user_id"]
self.id: str = data['id']
self.created_at: datetime = datetime.strptime(
data['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ'
)
self.user_id: str = data['user_id']
self.user: RawUser = RawUser(data['user'])
self.content: Optional[str] = data.get("text", None)
self.cw: Optional[str] = data["cw"]
self.visibility: str = data["visibility"]
self.renote_count: int = data["renote_count"]
self.replies_count: int = data["replies_count"]
self.reactions = data["reactions"] # TODO:型探す
self.emojis = data["emojis"] # TODO:型探す
self.file_ids: List[str] = data["file_ids"]
self.files = data["files"]
self.reply_id = data["reply_id"]
self.files = data["files"]
self.reply_id = data["reply_id"]
self.renote_id = data["renote_id"]
self.uri = data.get("uri")
self.poll: Optional[RawPoll] = RawPoll(data["poll"]) if data.get("poll") else None
self.content: Optional[str] = data.get('text', None)
self.cw: Optional[str] = data['cw']
self.visibility: str = data['visibility']
self.renote_count: int = data['renote_count']
self.replies_count: int = data['replies_count']
self.reactions = data['reactions'] # TODO:型探す
self.emojis = data['emojis'] # TODO:型探す
self.file_ids: List[str] = data['file_ids']
self.files = data['files']
self.reply_id = data['reply_id']
self.files = data['files']
self.reply_id = data['reply_id']
self.renote_id = data['renote_id']
self.uri = data.get('uri')
self.poll: Optional[RawPoll] = RawPoll(data['poll']) if data.get(
'poll'
) else None
class RawReaction:
@ -74,12 +96,17 @@ class RawReaction:
def __init__(self, data: ReactionPayload):
self.id: Optional[str] = data.get('id')
self.created_at: Optional[str] = datetime.strptime(data["created_at"], '%Y-%m-%dT%H:%M:%S.%fZ') if data.get(
'created_at') else None
self.created_at: Optional[str] = datetime.strptime(
data['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ'
) if data.get('created_at') else None
self.type: Optional[str] = data.get('type')
self.is_read: bool = bool(data.get('is_read'))
self.user: Optional[RawUser] = RawUser(data['user']) if data.get('user') else None
self.note: Optional[RawNote] = RawNote(data['note']) if data.get('note') else None
self.user: Optional[RawUser] = RawUser(data['user']) if data.get(
'user'
) else None
self.note: Optional[RawNote] = RawNote(data['note']) if data.get(
'note'
) else None
self.reaction: str = data['reaction']
@ -118,37 +145,75 @@ class RawNote:
"""
__slots__ = (
'id', 'created_at', 'user_id', 'author', 'content', 'cw', 'renote', 'visibility', 'renote_count', 'replies_count',
'reactions', 'emojis', 'file_ids', 'files', 'reply_id', 'renote_id', 'uri', 'poll', 'visible_user_ids',
'via_mobile', 'local_only', 'extract_mentions', 'extract_hashtags', 'extract_emojis', 'preview', 'media_ids',
'field', 'tags', 'channel_id')
'id',
'created_at',
'user_id',
'author',
'content',
'cw',
'renote',
'visibility',
'renote_count',
'replies_count',
'reactions',
'emojis',
'file_ids',
'files',
'reply_id',
'renote_id',
'uri',
'poll',
'visible_user_ids',
'via_mobile',
'local_only',
'extract_mentions',
'extract_hashtags',
'extract_emojis',
'preview',
'media_ids',
'field',
'tags',
'channel_id',
)
def __init__(self, data: NotePayload):
self.id: str = data["id"]
self.created_at: datetime = datetime.strptime(data["created_at"], '%Y-%m-%dT%H:%M:%S.%fZ')
self.user_id: str = data["user_id"]
self.id: str = data['id']
self.created_at: datetime = datetime.strptime(
data['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ'
)
self.user_id: str = data['user_id']
self.author: RawUser = RawUser(data['user'])
self.content: Optional[str] = data.get("text")
self.cw: Optional[str] = data.get("cw")
self.renote: Optional[RawRenote] = RawRenote(data['renote']) if data.get('renote') else None
self.visibility: Optional[str] = data.get("visibility") # This may be an optional
self.renote_count: Optional[int] = data.get("renote_count") # TODO: Optionalかどうか
self.replies_count: Optional[int] = data.get("replies_count") # TODO: Optionalかどうか
self.reactions: Dict[str, Any] = data["reactions"]
self.emojis: List[RawEmoji] = [RawEmoji(i) for i in data["emojis"]]
self.file_ids: Optional[List[str]] = data["file_ids"]
self.files: List[RawFile] = [RawFile(upper_to_lower(i)) for i in data["files"]]
self.reply_id: Optional[str] = data["reply_id"]
self.renote_id: Optional[str] = data["renote_id"]
self.poll: Optional[RawPoll] = RawPoll(data["poll"]) if data.get("poll") else None
self.visible_user_ids: Optional[List[str]] = data.get("visible_user_ids", [])
self.via_mobile: bool = bool(data.get("via_mobile", False))
self.local_only: bool = bool(data.get("local_only", False))
self.extract_mentions: bool = bool(data.get("extract_mentions"))
self.extract_hashtags: bool = bool(data.get("extract_hashtags"))
self.extract_emojis: bool = bool(data.get("extract_emojis"))
self.preview: bool = bool(data.get("preview"))
self.media_ids: Optional[List[str]] = data.get("media_ids")
self.field: Optional[dict] = {}
self.tags: Optional[List[str]] = data.get("tags", [])
self.channel_id: Optional[str] = data.get("channel_id")
self.content: Optional[str] = data.get('text')
self.cw: Optional[str] = data.get('cw')
self.renote: Optional[RawRenote] = RawRenote(data['renote']) if data.get(
'renote'
) else None
self.visibility: Optional[str] = data.get(
'visibility'
) # This may be an optional
self.renote_count: Optional[int] = data.get(
'renote_count'
) # TODO: Optionalかどうか
self.replies_count: Optional[int] = data.get(
'replies_count'
) # TODO: Optionalかどうか
self.reactions: Dict[str, Any] = data['reactions']
self.emojis: List[RawEmoji] = [RawEmoji(i) for i in data['emojis']]
self.file_ids: Optional[List[str]] = data['file_ids']
self.files: List[RawFile] = [RawFile(upper_to_lower(i)) for i in data['files']]
self.reply_id: Optional[str] = data['reply_id']
self.renote_id: Optional[str] = data['renote_id']
self.poll: Optional[RawPoll] = RawPoll(data['poll']) if data.get(
'poll'
) else None
self.visible_user_ids: Optional[List[str]] = data.get('visible_user_ids', [])
self.via_mobile: bool = bool(data.get('via_mobile', False))
self.local_only: bool = bool(data.get('local_only', False))
self.extract_mentions: bool = bool(data.get('extract_mentions'))
self.extract_hashtags: bool = bool(data.get('extract_hashtags'))
self.extract_emojis: bool = bool(data.get('extract_emojis'))
self.preview: bool = bool(data.get('preview'))
self.media_ids: Optional[List[str]] = data.get('media_ids')
self.field: Optional[Dict[Any, Any]] = {}
self.tags: Optional[List[str]] = data.get('tags', [])
self.channel_id: Optional[str] = data.get('channel_id')

17
mipac/core/models/poll.py
Unescape View File

@ -21,9 +21,9 @@ class RawPollChoices:
__slots__ = ('text', 'votes', 'is_voted')
def __init__(self, data):
self.text: str = data["text"]
self.votes: int = data["votes"]
self.is_voted: bool = data["isVoted"]
self.text: str = data['text']
self.votes: int = data['votes']
self.is_voted: bool = data['isVoted']
class RawPoll:
@ -43,8 +43,9 @@ class RawPoll:
__slots__ = ('multiple', 'expires_at', 'choices', 'expired_after')
def __init__(self, data: PollPayload):
self.multiple: Optional[bool] = data.get("multiple")
self.expires_at: Optional[int] = data.get("expires_at")
self.choices: Optional[List[RawPollChoices]] = [RawPollChoices(i) for i in data['choices']] if data.get(
"choices") else None
self.expired_after: Optional[int] = data.get("expired_after")
self.multiple: Optional[bool] = data.get('multiple')
self.expires_at: Optional[int] = data.get('expires_at')
self.choices: Optional[List[RawPollChoices]] = [
RawPollChoices(i) for i in data['choices']
] if data.get('choices') else None
self.expired_after: Optional[int] = data.get('expired_after')

4
mipac/core/models/reaction.py
Unescape View File

@ -22,6 +22,8 @@ class RawNoteReaction:
def __init__(self, data: NoteReactionPayload):
self.id: str = data['id']
self.created_at: datetime = datetime.strptime(data["created_at"], '%Y-%m-%dT%H:%M:%S.%fZ')
self.created_at: datetime = datetime.strptime(
data['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ'
)
self.user: UserPayload = data['user']
self.reaction: str = data['type']

161
mipac/core/models/user.py
Unescape View File

@ -23,24 +23,37 @@ class RawUserDetails:
"""
__slots__ = (
'avatar_blurhash', 'avatar_color', 'banner_url', 'banner_blurhash', 'banner_color', 'two_factor_enabled',
'use_password_less_login', 'security_keys', 'has_pending_follow_request_from_you',
'has_pending_follow_request_to_you', 'public_reactions', 'lang'
'avatar_blurhash',
'avatar_color',
'banner_url',
'banner_blurhash',
'banner_color',
'two_factor_enabled',
'use_password_less_login',
'security_keys',
'has_pending_follow_request_from_you',
'has_pending_follow_request_to_you',
'public_reactions',
'lang',
)
def __init__(self, data):
self.avatar_blurhash: Optional[str] = data.get("avatar_blurhash")
self.avatar_color: Optional[str] = data.get("avatar_color")
self.banner_url = data.get("banner_url")
self.banner_blurhash = data.get("banner_blurhash")
self.banner_color = data.get("banner_color")
self.two_factor_enabled = data.get("two_factor_enabled", False)
self.use_password_less_login = data.get("use_password_less_login", False)
self.security_keys = data.get("security_keys", False)
self.has_pending_follow_request_from_you = data.get("has_pending_follow_request_from_you", False)
self.has_pending_follow_request_to_you = data.get("has_pending_follow_request_to_you", False)
self.public_reactions = data.get("public_reactions", False)
self.lang = data.get("lang")
self.avatar_blurhash: Optional[str] = data.get('avatar_blurhash')
self.avatar_color: Optional[str] = data.get('avatar_color')
self.banner_url = data.get('banner_url')
self.banner_blurhash = data.get('banner_blurhash')
self.banner_color = data.get('banner_color')
self.two_factor_enabled = data.get('two_factor_enabled', False)
self.use_password_less_login = data.get('use_password_less_login', False)
self.security_keys = data.get('security_keys', False)
self.has_pending_follow_request_from_you = data.get(
'has_pending_follow_request_from_you', False
)
self.has_pending_follow_request_to_you = data.get(
'has_pending_follow_request_to_you', False
)
self.public_reactions = data.get('public_reactions', False)
self.lang = data.get('lang')
class RawUser:
@ -90,50 +103,86 @@ class RawUser:
"""
__slots__ = (
'id', 'name', 'nickname', 'host', 'avatar_url', 'is_admin', 'is_moderator', 'is_bot', 'is_cat', 'is_lady', 'emojis',
'online_status', 'url', 'uri', 'created_at', 'updated_at', 'is_locked', 'is_silenced', 'is_suspended', 'description',
'location', 'birthday', 'fields', 'followers_count', 'following_count', 'notes_count', 'pinned_note_ids',
'pinned_notes', 'pinned_page_id', 'pinned_page', 'ff_visibility', 'is_following', 'is_follow', 'is_blocking',
'is_blocked', 'is_muted', 'details', 'instance'
'id',
'name',
'nickname',
'host',
'avatar_url',
'is_admin',
'is_moderator',
'is_bot',
'is_cat',
'is_lady',
'emojis',
'online_status',
'url',
'uri',
'created_at',
'updated_at',
'is_locked',
'is_silenced',
'is_suspended',
'description',
'location',
'birthday',
'fields',
'followers_count',
'following_count',
'notes_count',
'pinned_note_ids',
'pinned_notes',
'pinned_page_id',
'pinned_page',
'ff_visibility',
'is_following',
'is_follow',
'is_blocking',
'is_blocked',
'is_muted',
'details',
'instance',
)
def __init__(self, data: UserPayload):
self.id: str = data['user_id'] if data.get('user_id') else data['id']
self.name: str = data["username"]
self.nickname: Optional[str] = data.get("name")
self.host: Optional[str] = data.get("host")
self.avatar_url: Optional[str] = data.get("avatar_url")
self.is_admin: bool = bool(data.get("is_admin"))
self.is_moderator: bool = bool(data.get("is_moderator"))
self.is_bot: bool = bool(data.get("is_bot"))
self.is_cat: bool = bool(data.get("is_cat", False))
self.name: str = data['username']
self.nickname: Optional[str] = data.get('name')
self.host: Optional[str] = data.get('host')
self.avatar_url: Optional[str] = data.get('avatar_url')
self.is_admin: bool = bool(data.get('is_admin'))
self.is_moderator: bool = bool(data.get('is_moderator'))
self.is_bot: bool = bool(data.get('is_bot'))
self.is_cat: bool = bool(data.get('is_cat', False))
self.is_lady: bool = bool(data.get('is_lady', False))
self.emojis: Optional[List[str]] = data.get("emojis")
self.online_status = data.get("online_status", None)
self.url: Optional[str] = data.get("url")
self.uri: Optional[str] = data.get("uri")
self.created_at: Optional[datetime] = datetime.strptime(data["created_at"], '%Y-%m-%dT%H:%M:%S.%fZ') if data.get(
"created_at") else None
self.updated_at = data.get("updated_at")
self.is_locked = data.get("is_locked", False)
self.is_silenced = data.get("is_silenced", False)
self.is_suspended = data.get("is_suspended", False)
self.description = data.get("description")
self.location = data.get("location")
self.birthday = data.get("birthday")
self.fields = data.get("fields", [])
self.followers_count = data.get("followers_count", 0)
self.following_count = data.get("following_count", 0)
self.notes_count = data.get("notes_count", 0)
self.pinned_note_ids = data.get("pinned_note_ids", [])
self.pinned_notes = data.get("pinned_notes", [])
self.pinned_page_id = data.get("pinned_page_id")
self.pinned_page = data.get("pinned_page")
self.ff_visibility: str = data.get("ff_visibility", 'public')
self.is_following: bool = bool(data.get("is_following", False))
self.is_follow: bool = bool(data.get("is_follow", False))
self.is_blocking: bool = bool(data.get("is_blocking", False))
self.is_blocked: bool = bool(data.get("is_blocked", False))
self.is_muted: bool = bool(data.get("is_muted", False))
self.emojis: Optional[List[str]] = data.get('emojis')
self.online_status = data.get('online_status', None)
self.url: Optional[str] = data.get('url')
self.uri: Optional[str] = data.get('uri')
self.created_at: Optional[datetime] = datetime.strptime(
data['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ'
) if data.get('created_at') else None
self.updated_at = data.get('updated_at')
self.is_locked = data.get('is_locked', False)
self.is_silenced = data.get('is_silenced', False)
self.is_suspended = data.get('is_suspended', False)
self.description = data.get('description')
self.location = data.get('location')
self.birthday = data.get('birthday')
self.fields = data.get('fields', [])
self.followers_count = data.get('followers_count', 0)
self.following_count = data.get('following_count', 0)
self.notes_count = data.get('notes_count', 0)
self.pinned_note_ids = data.get('pinned_note_ids', [])
self.pinned_notes = data.get('pinned_notes', [])
self.pinned_page_id = data.get('pinned_page_id')
self.pinned_page = data.get('pinned_page')
self.ff_visibility: str = data.get('ff_visibility', 'public')
self.is_following: bool = bool(data.get('is_following', False))
self.is_follow: bool = bool(data.get('is_follow', False))
self.is_blocking: bool = bool(data.get('is_blocking', False))
self.is_blocked: bool = bool(data.get('is_blocked', False))
self.is_muted: bool = bool(data.get('is_muted', False))
self.details: RawUserDetails = RawUserDetails(data)
self.instance: Optional[RawInstance] = RawInstance(data["instance"]) if data.get("instance") else None
self.instance: Optional[RawInstance] = RawInstance(
data['instance']
) if data.get('instance') else None

8
mipac/http.py
Unescape View File

@ -43,7 +43,9 @@ class Route:
class HTTPClient:
def __init__(self, url: str, token: str) -> None:
user_agent = 'Misskey Bot (https://github.com/yupix/Mi.py {0}) Python/{1[0]}.{1[1]} aiohttp/{2}'
self.user_agent = user_agent.format(__version__, sys.version_info, aiohttp.__version__)
self.user_agent = user_agent.format(
__version__, sys.version_info, aiohttp.__version__
)
self.__session: aiohttp.ClientSession = MISSING
self.__url: str = url
self.__token: str = token
@ -69,7 +71,9 @@ class HTTPClient:
if kwargs.get(i):
kwargs[i] = remove_dict_empty(kwargs[i])
async with self.__session.request(route.method, self.__url + route.path, **kwargs) as res:
async with self.__session.request(
route.method, self.__url + route.path, **kwargs
) as res:
data = await json_or_text(res)
if is_lower:
if isinstance(data, list):

17
mipac/manager/ad.py
Unescape View File

@ -13,7 +13,16 @@ class AdminAdvertisingManager(AbstractManager):
self.__session: HTTPClient = session
self.__client: ClientActions = client
async def create(self, url: str, memo: str, place: str, priority: str, ratio: str, expires_at: int, image_url: str):
async def create(
self,
url: str,
memo: str,
place: str,
priority: str,
ratio: str,
expires_at: int,
image_url: str,
):
data = {
'url': url,
'memo': memo,
@ -21,6 +30,8 @@ class AdminAdvertisingManager(AbstractManager):
'priority': priority,
'ratio': ratio,
'expires_at': expires_at,
'image_url': image_url
'image_url': image_url,
}
return await self.__session.request(Route('POST', '/api/ad/create'), json=data, auth=True, lower=True)
return await self.__session.request(
Route('POST', '/api/ad/create'), json=data, auth=True, lower=True
)

36
mipac/manager/admin/user.py
Unescape View File

@ -11,7 +11,13 @@ if TYPE_CHECKING:
class AdminUserManager:
def __init__(self, user_id: Optional[str] = None, *, session: HTTPClient, client: ClientActions):
def __init__(
self,
user_id: Optional[str] = None,
*,
session: HTTPClient,
client: ClientActions
):
self.__user_id = user_id
self.__session: HTTPClient = session
self.__client: ClientActions = client
@ -34,9 +40,14 @@ class AdminUserManager:
"""
if config.is_ayuskey:
raise NotSupportedError("Ayuskeyではサポートされていません")
raise NotSupportedError('Ayuskeyではサポートされていません')
data = {'username': username, 'password': password}
res = await self.__session.request(Route('POST', '/api/admin/accounts/create'), json=data, auth=True, lower=True)
res = await self.__session.request(
Route('POST', '/api/admin/accounts/create'),
json=data,
auth=True,
lower=True,
)
return User(RawUser(res), client=self.__client)
async def delete_account(self, user_id: Optional[str] = None) -> bool:
@ -56,7 +67,12 @@ class AdminUserManager:
user_id = user_id or self.__user_id
data = {'userId': user_id}
res = await self.__session.request(Route('POST', '/api/admin/accounts/delete'), json=data, auth=True, lower=True)
res = await self.__session.request(
Route('POST', '/api/admin/accounts/delete'),
json=data,
auth=True,
lower=True,
)
return bool(res)
async def show_user(self, user_id: Optional[str] = None) -> User:
@ -75,7 +91,9 @@ class AdminUserManager:
user_id = user_id or self.__user_id
data = {'userId': user_id}
res = await self.__session.request(Route('GET', '/api/admin/show-user'), json=data, auth=True, lower=True)
res = await self.__session.request(
Route('GET', '/api/admin/show-user'), json=data, auth=True, lower=True
)
return User(RawUser(res), client=self.__client)
async def suspend(self, user_id: Optional[str] = None) -> bool:
@ -95,7 +113,9 @@ class AdminUserManager:
user_id = user_id or self.__user_id
data = {'userId': user_id}
res = await self.__session.request(Route('POST', '/api/admin/suspend-user'), json=data, auth=True, lower=True)
res = await self.__session.request(
Route('POST', '/api/admin/suspend-user'), json=data, auth=True, lower=True
)
return bool(res)
async def unsuspend(self, user_id: Optional[str] = None) -> bool:
@ -115,5 +135,7 @@ class AdminUserManager:
user_id = user_id or self.__user_id
data = {'userId': user_id}
res = await self.__session.request(Route('POST', '/api/admin/unsuspend-user'), json=data, auth=True, lower=True)
res = await self.__session.request(
Route('POST', '/api/admin/unsuspend-user'), json=data, auth=True, lower=True
)
return bool(res)

28
mipac/manager/chart.py
Unescape View File

@ -16,20 +16,20 @@ class ChartManager(AbstractManager):
self.__session: HTTPClient = session
self.__client: ClientActions = client
async def get_active_user(self, span: str = 'day', limit: int = 30, offset: int = 0) -> RawActiveUsersChart:
data = {
'span': span,
'limit': limit,
'offset': offset
}
data = await self.__session.request(Route('POST', '/api/charts/active-users'), json=data, auth=True, lower=True)
async def get_active_user(
self, span: str = 'day', limit: int = 30, offset: int = 0
) -> RawActiveUsersChart:
data = {'span': span, 'limit': limit, 'offset': offset}
data = await self.__session.request(
Route('POST', '/api/charts/active-users'), json=data, auth=True, lower=True
)
return RawActiveUsersChart(data)
async def get_drive(self, span: str = 'day', limit: int = 30, offset: int = 0) -> RawDriveChart:
data = {
'span': span,
'limit': limit,
'offset': offset
}
data = await self.__session.request(Route('POST', '/api/charts/drive'), json=data, auth=True, lower=True)
async def get_drive(
self, span: str = 'day', limit: int = 30, offset: int = 0
) -> RawDriveChart:
data = {'span': span, 'limit': limit, 'offset': offset}
data = await self.__session.request(
Route('POST', '/api/charts/drive'), json=data, auth=True, lower=True
)
return RawDriveChart(data)

36
mipac/manager/chat.py
Unescape View File

@ -12,8 +12,13 @@ if TYPE_CHECKING:
class ChatManager:
def __init__(self, session: HTTPClient, client: ClientActions, user_id: Optional[str] = None,
message_id: Optional[str] = None):
def __init__(
self,
session: HTTPClient,
client: ClientActions,
user_id: Optional[str] = None,
message_id: Optional[str] = None,
):
self.__session: HTTPClient = session
self.__client: ClientActions = client
self.__user_id = user_id
@ -40,16 +45,18 @@ class ChatManager:
raise ParameterError('limit must be greater than 100')
args = {'limit': limit, 'group': group}
data = await self.__session.request(Route('POST', '/api/messaging/history'), json=args, auth=True)
data = await self.__session.request(
Route('POST', '/api/messaging/history'), json=args, auth=True
)
return [Chat(RawChat(d)) for d in data]
async def send(
self,
text: Optional[str] = None,
*,
file_id: Optional[str] = None,
user_id: Optional[str] = None,
group_id: Optional[str] = None
self,
text: Optional[str] = None,
*,
file_id: Optional[str] = None,
user_id: Optional[str] = None,
group_id: Optional[str] = None,
) -> Chat:
"""
Send chat.
@ -67,7 +74,12 @@ class ChatManager:
"""
user_id = user_id or self.__user_id
data = {'userId': user_id, 'groupId': group_id, 'text': text, 'fileId': file_id}
res = await self.__session.request(Route('POST', '/api/messaging/messages/create'), json=data, auth=True, lower=True)
res = await self.__session.request(
Route('POST', '/api/messaging/messages/create'),
json=data,
auth=True,
lower=True,
)
return Chat(RawChat(res))
async def delete(self, message_id: str) -> bool:
@ -87,5 +99,7 @@ class ChatManager:
message_id = message_id or self.__message_id
args = {'messageId': f'{message_id}'}
data = await self.__session.request(Route('POST', '/api/messaging/messages/delete'), json=args, auth=True)
data = await self.__session.request(
Route('POST', '/api/messaging/messages/delete'), json=args, auth=True
)
return bool(data)

16
mipac/manager/client.py
Unescape View File

@ -20,10 +20,18 @@ class ClientActions:
self.note: NoteManager = NoteManager(session=session, client=self)
self.chat: ChatManager = ChatManager(session=session, client=self)
self.user: UserManager = UserManager(session=session, client=self)
self.admin: AdminActions = AdminActions(session=session, client=self) # TODO: 作る
self.drive: DriveManager = DriveManager(session=session, client=self) # TODO: 作る
self.reaction: ReactionManager = ReactionManager(session=session, client=self) # TODO: 作る
self.chart: ChartManager = ChartManager(session=session, client=self) # TODO: 作る
self.admin: AdminActions = AdminActions(
session=session, client=self
) # TODO: 作る
self.drive: DriveManager = DriveManager(
session=session, client=self
) # TODO: 作る
self.reaction: ReactionManager = ReactionManager(
session=session, client=self
) # TODO: 作る
self.chart: ChartManager = ChartManager(
session=session, client=self
) # TODO: 作る
def _create_user_instance(self, user: User) -> UserManager:
return UserManager(session=self.__session, client=self, user=user)

112
mipac/manager/drive.py
Unescape View File

@ -16,7 +16,13 @@ __all__ = ['FolderManager', 'FileManager', 'DriveManager']
class FileManager:
def __init__(self, file_id: Optional[str] = None, *, session: HTTPClient, client: ClientActions):
def __init__(
self,
file_id: Optional[str] = None,
*,
session: HTTPClient,
client: ClientActions
):
self.__session: HTTPClient = session
self.__client: ClientActions = client
self.__file_id = file_id
@ -38,8 +44,13 @@ class FileManager:
ファイルの情報
"""
data = remove_dict_empty({"fileId": file_id, "url": url})
res = await self.__session.request(Route('POST', '/api/admin/drive/show-file'), json=data, auth=True, lower=True)
data = remove_dict_empty({'fileId': file_id, 'url': url})
res = await self.__session.request(
Route('POST', '/api/admin/drive/show-file'),
json=data,
auth=True,
lower=True,
)
return File(RawFile(res), client=self.__client)
async def remove_file(self, file_id: Optional[str] = None) -> bool:
@ -59,15 +70,20 @@ class FileManager:
file_id = file_id or self.__file_id
return bool(
await self.__session.request(Route('POST', '/api/drive/files/delete'), json={'fileId': file_id}, auth=True))
await self.__session.request(
Route('POST', '/api/drive/files/delete'),
json={'fileId': file_id},
auth=True,
)
)
async def get_files(
self,
limit: int = 10,
since_id: Optional[str] = None,
until_id: Optional[str] = None,
folder_id: Optional[str] = None,
file_type: Optional[str] = None
self,
limit: int = 10,
since_id: Optional[str] = None,
until_id: Optional[str] = None,
folder_id: Optional[str] = None,
file_type: Optional[str] = None,
) -> List[File]:
"""
ファイルを取得します
@ -88,14 +104,27 @@ class FileManager:
if limit > 100:
raise ParameterError('limit must be less than 100')
data = {'limit': limit, 'sinceId': since_id, 'untilId': until_id, 'folderId': folder_id, 'Type': file_type}
res = await self.__session.request(Route('POST', '/api/drive/files'), json=data, auth=True, lower=True)
data = {
'limit': limit,
'sinceId': since_id,
'untilId': until_id,
'folderId': folder_id,
'Type': file_type,
}
res = await self.__session.request(
Route('POST', '/api/drive/files'), json=data, auth=True, lower=True
)
return [File(RawFile(i), client=self.__client) for i in res]
class FolderManager(AbstractManager):
def __init__(self, folder_id: Optional[str] = None, *, session: HTTPClient, client: ClientActions):
def __init__(
self,
folder_id: Optional[str] = None,
*,
session: HTTPClient,
client: ClientActions
):
self.__folder_id = folder_id
self.__session: HTTPClient = session
self.__client: ClientActions = client
@ -120,7 +149,14 @@ class FolderManager(AbstractManager):
parent_id = parent_id or self.__folder_id
data = {'name': name, 'parent_id': parent_id}
return bool(await self.__session.request(Route('POST', '/api/drive/folders/create'), json=data, lower=True, auth=True))
return bool(
await self.__session.request(
Route('POST', '/api/drive/folders/create'),
json=data,
lower=True,
auth=True,
)
)
async def delete(self, folder_id: Optional[str] = None) -> bool:
"""
@ -137,10 +173,23 @@ class FolderManager(AbstractManager):
folder_id = folder_id or self.__folder_id
data = {'folderId': folder_id}
return bool(await self.__session.request(Route('POST', '/api/drive/folders/delete'), json=data, lower=True, auth=True))
return bool(
await self.__session.request(
Route('POST', '/api/drive/folders/delete'),
json=data,
lower=True,
auth=True,
)
)
async def get_files(self, limit: int = 10, since_id: Optional[str] = None, until_id: Optional[str] = None,
folder_id: Optional[str] = None, file_type: Optional[str] = None) -> List[File]:
async def get_files(
self,
limit: int = 10,
since_id: Optional[str] = None,
until_id: Optional[str] = None,
folder_id: Optional[str] = None,
file_type: Optional[str] = None,
) -> List[File]:
"""
ファイルを取得します
@ -161,8 +210,16 @@ class FolderManager(AbstractManager):
raise ParameterError('limit must be less than 100')
folder_id = folder_id or self.__folder_id
data = {'limit': limit, 'sinceId': since_id, 'untilId': until_id, 'folderId': folder_id, 'Type': file_type}
res = await self.__session.request(Route('POST', '/api/drive/files'), json=data, auth=True, lower=True)
data = {
'limit': limit,
'sinceId': since_id,
'untilId': until_id,
'folderId': folder_id,
'Type': file_type,
}
res = await self.__session.request(
Route('POST', '/api/drive/files'), json=data, auth=True, lower=True
)
return [File(RawFile(i), client=self.__client) for i in res]
@ -171,8 +228,13 @@ class DriveManager(AbstractManager):
self.__session: HTTPClient = session
self.__client: ClientActions = client
async def get_folders(self, limit: int = 100, since_id: Optional[str] = None, until_id: Optional[str] = None,
folder_id: Optional[str] = None) -> List[Folder]:
async def get_folders(
self,
limit: int = 100,
since_id: Optional[str] = None,
until_id: Optional[str] = None,
folder_id: Optional[str] = None,
) -> List[Folder]:
"""
フォルダーの一覧を取得します
@ -192,7 +254,9 @@ class DriveManager(AbstractManager):
'limit': limit,
'sinceId': since_id,
'untilId': until_id,
'folderId': folder_id
'folderId': folder_id,
}
data = await self.__session.request(Route('POST', '/api/drive/folders'), json=data, lower=True, auth=True)
data = await self.__session.request(
Route('POST', '/api/drive/folders'), json=data, lower=True, auth=True
)
return [Folder(RawFolder(i)) for i in data]

38
mipac/manager/emoji.py
Unescape View File

@ -12,19 +12,25 @@ if TYPE_CHECKING:
class AdminEmojiManager(AbstractManager):
def __init__(self, emoji_id: Optional[str] = None, *, session: HTTPClient, client: ClientActions):
def __init__(
self,
emoji_id: Optional[str] = None,
*,
session: HTTPClient,
client: ClientActions
):
self.emoji_id: Optional[str] = emoji_id
self.__session: HTTPClient = session
self.__client: ClientActions = client
async def add(
self,
file_id: Optional[str] = None,
*,
name: Optional[str] = None,
url: Optional[str] = None,
category: Optional[str] = None,
aliases: Optional[List[str]] = None
self,
file_id: Optional[str] = None,
*,
name: Optional[str] = None,
url: Optional[str] = None,
category: Optional[str] = None,
aliases: Optional[List[str]] = None
) -> bool:
if config.is_ayuskey: # TODO: どうにかする
data = {'name': name, 'url': url, 'category': category, 'aliases': aliases}
@ -33,7 +39,11 @@ class AdminEmojiManager(AbstractManager):
if not check_multi_arg(file_id, url):
raise NotExistRequiredData('required a file_id or url')
return bool(await self.__session.request(Route('POST', '/api/admin/emoji/add'), json=data, lower=True, auth=True))
return bool(
await self.__session.request(
Route('POST', '/api/admin/emoji/add'), json=data, lower=True, auth=True
)
)
async def remove(self, emoji_id: Optional[str] = None) -> bool:
emoji_id = emoji_id or self.emoji_id
@ -41,5 +51,11 @@ class AdminEmojiManager(AbstractManager):
if emoji_id is None:
raise NotExistRequiredData('idが不足しています')
return bool(await self.__session.request(Route('POST', '/api/admin/emoji/remove'), json={'id': emoji_id}, lower=True,
auth=True))
return bool(
await self.__session.request(
Route('POST', '/api/admin/emoji/remove'),
json={'id': emoji_id},
lower=True,
auth=True,
)
)

21
mipac/manager/favorite.py
Unescape View File

@ -10,8 +10,13 @@ if TYPE_CHECKING:
class FavoriteManager(AbstractManager):
def __init__(self, note_id: Optional[str] = None, *, session: HTTPClient, client: ClientActions):
def __init__(
self,
note_id: Optional[str] = None,
*,
session: HTTPClient,
client: ClientActions
):
self.__note_id = note_id
self.__session: HTTPClient = session
self.__client: ClientActions = client
@ -19,9 +24,17 @@ class FavoriteManager(AbstractManager):
async def add(self, note_id: Optional[str] = None) -> bool:
note_id = note_id or self.__note_id
data = {'noteId': note_id}
return bool(await self.__session.request(Route('POST', '/api/notes/favorites/create'), json=data, auth=True))
return bool(
await self.__session.request(
Route('POST', '/api/notes/favorites/create'), json=data, auth=True
)
)
async def remove(self, note_id: Optional[str] = None) -> bool:
note_id = note_id or self.__note_id
data = {'noteId': note_id}
return bool(await self.__session.request(Route('POST', '/api/notes/favorites/delete'), json=data, auth=True))
return bool(
await self.__session.request(
Route('POST', '/api/notes/favorites/delete'), json=data, auth=True
)
)

21
mipac/manager/file.py
Unescape View File

@ -5,6 +5,7 @@ from mipac.http import Route
# __all__ = ('MiFile', 'check_upload', 'get_file_ids')
class MiFile:
__slots__ = (
'path',
@ -13,17 +14,19 @@ class MiFile:
'folder_id',
'comment',
'is_sensitive',
'force'
'force',
)
def __init__(self, path: Optional[str] = None,
file_id: Optional[str] = None,
name: Optional[str] = None,
folder_id: Optional[str] = None,
comment: Optional[str] = None,
is_sensitive: bool = False,
force: bool = False
):
def __init__(
self,
path: Optional[str] = None,
file_id: Optional[str] = None,
name: Optional[str] = None,
folder_id: Optional[str] = None,
comment: Optional[str] = None,
is_sensitive: bool = False,
force: bool = False,
):
"""
Parameters
----------

53
mipac/manager/follow.py
Unescape View File

@ -13,8 +13,13 @@ if TYPE_CHECKING:
class FollowManager(AbstractManager):
def __init__(self, user_id: Optional[str] = None, *, session: HTTPClient, client: ClientActions):
def __init__(
self,
user_id: Optional[str] = None,
*,
session: HTTPClient,
client: ClientActions
):
self.__user_id: Optional[str] = user_id
self.__session: HTTPClient = session
self.__client: ClientActions = client
@ -33,10 +38,12 @@ class FollowManager(AbstractManager):
user_id = user_id or self.__user_id
data = {"userId": user_id}
res = await self.__session.request(Route('POST', '/api/following/create'), json=data, auth=True, lower=True)
if res.get("error"):
code = res["error"]["code"]
data = {'userId': user_id}
res = await self.__session.request(
Route('POST', '/api/following/create'), json=data, auth=True, lower=True
)
if res.get('error'):
code = res['error']['code']
status = False
else:
code = None
@ -55,13 +62,21 @@ class FollowManager(AbstractManager):
user_id = user_id or self.__user_id
data = {"userId": user_id}
res = await self.__session.request(Route('POST', '/api/following/delete'), json=data, auth=True)
data = {'userId': user_id}
res = await self.__session.request(
Route('POST', '/api/following/delete'), json=data, auth=True
)
return bool(res.status_code == 204 or 200)
class FollowRequestManager(AbstractManager):
def __init__(self, user_id: Optional[str] = None, *, session: HTTPClient, client: ClientActions):
def __init__(
self,
user_id: Optional[str] = None,
*,
session: HTTPClient,
client: ClientActions
):
self.__user_id: Optional[str] = user_id
self.__session: HTTPClient = session
self.__client: ClientActions = client
@ -71,8 +86,12 @@ class FollowRequestManager(AbstractManager):
未承認のフォローリクエストを取得します
"""
return [FollowRequest(i['follower']) for i in
await self.__session.request(Route('POST', '/api/following/requests/list'), auth=True, lower=True)]
return [
FollowRequest(i['follower'])
for i in await self.__session.request(
Route('POST', '/api/following/requests/list'), auth=True, lower=True
)
]
async def get_user(self, user_id: Optional[str] = None) -> User:
"""
@ -100,7 +119,11 @@ class FollowRequestManager(AbstractManager):
user_id = user_id or self.__user_id
data = {'userId': user_id}
return bool(await self.__session.request(Route('POST', '/api/following/requests/accept'), json=data, auth=True))
return bool(
await self.__session.request(
Route('POST', '/api/following/requests/accept'), json=data, auth=True
)
)
async def reject(self, user_id: Optional[str]) -> bool:
"""
@ -110,4 +133,8 @@ class FollowRequestManager(AbstractManager):
user_id = user_id or self.__user_id
data = {'userId': user_id}
return bool(await self.__session.request(Route('POST', '/api/following/requests/reject'), json=data, auth=True))
return bool(
await self.__session.request(
Route('POST', '/api/following/requests/reject'), json=data, auth=True
)
)

16
mipac/manager/moderator.py
Unescape View File

@ -9,7 +9,13 @@ if TYPE_CHECKING:
class AdminModeratorManager(AbstractManager):
def __init__(self, user_id: Optional[str] = None, *, session: HTTPClient, client: ClientActions):
def __init__(
self,
user_id: Optional[str] = None,
*,
session: HTTPClient,
client: ClientActions
):
self.__user_id: Optional[str] = user_id
self.__session: HTTPClient = session
self.__client: ClientActions = client
@ -31,7 +37,9 @@ class AdminModeratorManager(AbstractManager):
user_id = user_id or self.__user_id
data = {'userId': user_id}
res = await self.__session.request(Route('POST', '/api/moderators/add'), json=data, auth=True, lower=True)
res = await self.__session.request(
Route('POST', '/api/moderators/add'), json=data, auth=True, lower=True
)
return bool(res)
async def remove(self, user_id: Optional[str] = None) -> bool:
@ -50,5 +58,7 @@ class AdminModeratorManager(AbstractManager):
"""
user_id = user_id or self.__user_id
data = {'userId': user_id}
res = await self.__session.request(Route('POST', '/api/moderators/remove'), json=data, auth=True, lower=True)
res = await self.__session.request(
Route('POST', '/api/moderators/remove'), json=data, auth=True, lower=True
)
return bool(res)

33
mipac/manager/note.py
Unescape View File

@ -13,17 +13,34 @@ if TYPE_CHECKING:
class NoteManager(AbstractManager):
"""User behavior for notes"""
def __init__(self, note_id: Optional[str] = None, *, session: HTTPClient, client: ClientActions):
def __init__(
self,
note_id: Optional[str] = None,
*,
session: HTTPClient,
client: ClientActions
):
self.__session: HTTPClient = session
self.__client: ClientActions = client
self.__note_id = note_id
@property
def action(self) -> NoteActions:
return NoteActions(note_id=self.__note_id, session=self.__session, client=self.__client)
async def get(self, local: bool = True, reply: bool = False, renote: bool = True, with_files: bool = False,
poll: bool = True, limit: int = 10, since_id: Optional[str] = None, until_id: Optional[str] = None):
return NoteActions(
note_id=self.__note_id, session=self.__session, client=self.__client
)
async def get(
self,
local: bool = True,
reply: bool = False,
renote: bool = True,
with_files: bool = False,
poll: bool = True,
limit: int = 10,
since_id: Optional[str] = None,
until_id: Optional[str] = None,
):
data = {
'local': local,
'reply': reply,
@ -32,6 +49,8 @@ class NoteManager(AbstractManager):
'poll': poll,
'limit': limit,
'sinceId': since_id,
'untilId': until_id
'untilId': until_id,
}
await self.__session.request(Route('POST', '/api/notes'), json=data, auth=True, lower=True)
await self.__session.request(
Route('POST', '/api/notes'), json=data, auth=True, lower=True
)

18
mipac/manager/page.py
Unescape View File

@ -9,15 +9,17 @@ if TYPE_CHECKING:
class PagesManager(AbstractManager):
def __init__(self, *, session: HTTPClient, client: ClientActions):
self.__session: HTTPClient = session
self.__client: ClientActions = client
async def get_pages(self, limit: int = 100, since_id: Optional[int] = None, until_id: Optional[int] = None):
data = {
'limit': limit,
'since_id': since_id,
'until_id': until_id
}
res = await self.__session.request(Route('POST', '/api/i/pages'), json=data, auth=True)
async def get_pages(
self,
limit: int = 100,
since_id: Optional[int] = None,
until_id: Optional[int] = None,
):
data = {'limit': limit, 'since_id': since_id, 'until_id': until_id}
res = await self.__session.request(
Route('POST', '/api/i/pages'), json=data, auth=True
)

31
mipac/manager/reaction.py
Unescape View File

@ -15,8 +15,13 @@ if TYPE_CHECKING:
class ReactionManager(AbstractManager):
def __init__(self, note_id: Optional[str] = None, *, session: HTTPClient, client: ClientActions):
def __init__(
self,
note_id: Optional[str] = None,
*,
session: HTTPClient,
client: ClientActions
):
self.__note_id: Optional[str] = note_id
self.__session: HTTPClient = session
self.__client: ClientActions = client
@ -39,23 +44,31 @@ class ReactionManager(AbstractManager):
"""
note_id = note_id or self.__note_id
data = remove_dict_empty({"noteId": note_id, "reaction": reaction})
data = remove_dict_empty({'noteId': note_id, 'reaction': reaction})
route = Route('POST', '/api/notes/reactions/create')
return await self.__session.request(route, json=data, auth=True, lower=True)
async def remove(self, note_id: Optional[str] = None) -> bool:
note_id = note_id or self.__note_id
data = remove_dict_empty({"noteId": note_id})
data = remove_dict_empty({'noteId': note_id})
route = Route('POST', '/api/notes/reactions/delete')
return bool(await self.__session.request(route, json=data, auth=True, lower=True))
return bool(
await self.__session.request(route, json=data, auth=True, lower=True)
)
async def get_reaction(self, reaction: str, note_id: Optional[str] = None, *, limit: int = 11) -> List[NoteReaction]:
async def get_reaction(
self, reaction: str, note_id: Optional[str] = None, *, limit: int = 11
) -> List[NoteReaction]:
note_id = note_id or self.__note_id
data = remove_dict_empty({"noteId": note_id, 'limit': limit, 'type': reaction})
res = await self.__session.request(Route('POST', '/api/notes/reactions'), json=data, auth=True, lower=True)
data = remove_dict_empty({'noteId': note_id, 'limit': limit, 'type': reaction})
res = await self.__session.request(
Route('POST', '/api/notes/reactions'), json=data, auth=True, lower=True
)
return [NoteReaction(RawNoteReaction(i)) for i in res]
async def get_emoji_list(self) -> List[Emoji]:
data = await self.__session.request(Route('GET', '/api/meta'), json={'detail': False}, auth=True)
data = await self.__session.request(
Route('GET', '/api/meta'), json={'detail': False}, auth=True
)
return [Emoji(RawEmoji(i)) for i in data['emojis']]

4
mipac/manager/user.py
Unescape View File

@ -11,7 +11,9 @@ if TYPE_CHECKING:
class UserManager(AbstractManager):
def __init__(self, user: Optional[User] = None, *, session: HTTPClient, client: ClientActions):
def __init__(
self, user: Optional[User] = None, *, session: HTTPClient, client: ClientActions
):
self.__session: HTTPClient = session
self.__client: ClientActions = client
self.user: Optional[User] = user

8
mipac/models/drive.py
Unescape View File

@ -1,8 +1,8 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Union
from typing import TYPE_CHECKING, Optional, Union
from mipac.core import RawUser
from mipac.core.models.user import RawUser
from mipac.core.models.drive import RawFile, RawFolder, RawProperties
from mipac.models.user import User
@ -17,7 +17,7 @@ class Properties:
self.__raw_data: RawProperties = raw_data
@property
def width(self) -> int:
def width(self) -> Optional[int]:
return self.__raw_data.width
@property
@ -25,7 +25,7 @@ class Properties:
return self.__raw_data.height
@property
def avg_color(self) -> Union[float, None]:
def avg_color(self) -> Optional[str]:
return self.__raw_data.avg_color

31
mipac/models/instance.py
Unescape View File

@ -28,8 +28,12 @@ class InstanceMeta:
self.disable_registration: bool = bool(data['disable_registration'])
self.disable_local_timeline: bool = bool(data['disable_local_timeline'])
self.disable_global_timeline: bool = bool(data['disable_global_timeline'])
self.drive_capacity_per_local_user_mb: int = data['drive_capacity_per_local_user_mb']
self.drive_capacity_per_remote_user_mb: int = data['drive_capacity_per_remote_user_mb']
self.drive_capacity_per_local_user_mb: int = data[
'drive_capacity_per_local_user_mb'
]
self.drive_capacity_per_remote_user_mb: int = data[
'drive_capacity_per_remote_user_mb'
]
self.email_required_for_signup: bool = bool(data['email_required_for_signup'])
self.enable_hcaptcha: bool = bool(data['enable_hcaptcha'])
self.enable_recaptcha: bool = bool(data['enable_recaptcha'])
@ -95,17 +99,18 @@ class Instance:
def theme_color(self):
return self.__raw_data.theme_color
def get_users(self,
limit: int = 10,
*,
offset: int = 0,
sort: Optional[str] = None,
state: str = 'all',
origin: str = 'local',
username: Optional[str] = None,
hostname: Optional[str] = None,
get_all: bool = False
) -> AsyncIterator[User]:
def get_users(
self,
limit: int = 10,
*,
offset: int = 0,
sort: Optional[str] = None,
state: str = 'all',
origin: str = 'local',
username: Optional[str] = None,
hostname: Optional[str] = None,
get_all: bool = False
) -> AsyncIterator[User]:
"""
Parameters

55
mipac/models/note.py
Unescape View File

@ -16,14 +16,24 @@ if TYPE_CHECKING:
from mipac.manager.client import ClientActions
from mipac.actions.note import NoteActions
__all__ = ('Note', 'Poll', 'Reaction', 'Follow', 'Header', 'File', 'Renote', 'NoteReaction')
__all__ = (
'Note',
'Poll',
'Reaction',
'Follow',
'Header',
'File',
'Renote',
'NoteReaction',
)
class Follow:
def __init__(self, data):
self.id: Optional[str] = data.get('id')
self.created_at: Optional[datetime] = datetime.strptime(data["created_at"], '%Y-%m-%dT%H:%M:%S.%fZ') if data.get(
"created_at") else None
self.created_at: Optional[datetime] = datetime.strptime(
data['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ'
) if data.get('created_at') else None
self.type: Optional[str] = data.get('type')
self.user: Optional[User] = data.get('user')
@ -64,8 +74,8 @@ class Follow:
class Header:
def __init__(self, data):
self.id = data.get("id")
self.type = data.get("type")
self.id = data.get('id')
self.type = data.get('type')
class Poll:
@ -210,11 +220,19 @@ class Reaction:
@property
def user(self) -> Optional[User]:
return User(self.__raw_data.user, client=self.__client) if self.__raw_data.user else None
return (
User(self.__raw_data.user, client=self.__client)
if self.__raw_data.user
else None
)
@property
def note(self) -> Optional[Note]:
return Note(self.__raw_data.note, client=self.__client) if self.__raw_data.note else None
return (
Note(self.__raw_data.note, client=self.__client)
if self.__raw_data.note
else None
)
@property
def reaction(self) -> str:
@ -339,7 +357,7 @@ class Note:
return self.__raw_data.media_ids
@property
def field(self) -> Optional[Dict]:
def field(self) -> Optional[Dict[Any, Any]]: # TODO: any
return self.__raw_data.field
@property
@ -362,15 +380,16 @@ class Note:
return self.__client._create_note_instance(self.id).action
async def reply(
self, content: Optional[str],
cw: Optional[str] = None,
extract_mentions: bool = True,
extract_hashtags: bool = True,
extract_emojis: bool = True,
renote_id: Optional[str] = None,
channel_id: Optional[str] = None,
file_ids=None,
poll: Optional[Poll] = None
self,
content: Optional[str],
cw: Optional[str] = None,
extract_mentions: bool = True,
extract_hashtags: bool = True,
extract_emojis: bool = True,
renote_id: Optional[str] = None,
channel_id: Optional[str] = None,
file_ids=None,
poll: Optional[Poll] = None,
) -> Note:
"""
ノートに対して返信を送信します
@ -411,5 +430,5 @@ class Note:
renote_id=renote_id,
channel_id=channel_id,
file_ids=file_ids,
poll=poll
poll=poll,
)

126
mipac/models/user.py
Unescape View File

@ -6,7 +6,12 @@ from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union
from mipac.core import RawUser
from mipac.models.emoji import Emoji
from mipac.models.instance import Instance
from mipac.types import ChannelPayload, FieldContentPayload, PinnedNotePayload, PinnedPagePayload
from mipac.types import (
ChannelPayload,
FieldContentPayload,
PinnedNotePayload,
PinnedPagePayload,
)
if TYPE_CHECKING:
from mipac.manager.client import ClientActions
@ -19,7 +24,9 @@ __all__ = ['User', 'FollowRequest', 'Followee']
class Followee:
def __init__(self, data, *, client: ClientActions):
self.id: str = data['id']
self.created_at: datetime = datetime.strptime(data["created_at"], '%Y-%m-%dT%H:%M:%S.%fZ')
self.created_at: datetime = datetime.strptime(
data['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ'
)
self.followee_id: str = data['followee_id']
self.follower_id: str = data['follower_id']
self.user: User = User(RawUser(data['follower']), client=client)
@ -51,70 +58,81 @@ class FollowRequest:
class Channel:
def __init__(self, data: ChannelPayload):
self.id: Optional[str] = data.get("id")
self.created_at: Optional[datetime] = datetime.strptime(data["created_at"], '%Y-%m-%dT%H:%M:%S.%fZ') if data.get(
"created_at") else None
self.last_noted_at: Optional[str] = data.get("last_noted_at")
self.name: Optional[str] = data.get("name")
self.description: Optional[str] = data.get("description")
self.banner_url: Optional[str] = data.get("banner_url")
self.notes_count: Optional[int] = data.get("notes_count")
self.users_count: Optional[int] = data.get("users_count")
self.is_following: Optional[bool] = data.get("is_following")
self.user_id: Optional[str] = data.get("user_id")
self.id: Optional[str] = data.get('id')
self.created_at: Optional[datetime] = datetime.strptime(
data['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ'
) if data.get('created_at') else None
self.last_noted_at: Optional[str] = data.get('last_noted_at')
self.name: Optional[str] = data.get('name')
self.description: Optional[str] = data.get('description')
self.banner_url: Optional[str] = data.get('banner_url')
self.notes_count: Optional[int] = data.get('notes_count')
self.users_count: Optional[int] = data.get('users_count')
self.is_following: Optional[bool] = data.get('is_following')
self.user_id: Optional[str] = data.get('user_id')
class PinnedNote:
def __init__(self, data: PinnedNotePayload, *, client: ClientActions):
self.id: Optional[str] = data.get("id")
self.created_at: Optional[datetime] = datetime.strptime(data["created_at"], '%Y-%m-%dT%H:%M:%S.%fZ') if data.get(
"created_at") else None
self.text: Optional[str] = data.get("text")
self.cw: Optional[str] = data.get("cw")
self.user_id: Optional[str] = data.get("user_id")
self.user: Optional[User] = User(RawUser(data['user']), client=client) if data.get('user') else None
self.reply_id: Optional[str] = data.get("reply_id")
self.reply: Optional[Dict[str, Any]] = data.get("reply")
self.renote: Optional[Dict[str, Any]] = data.get("renote")
self.via_mobile: Optional[bool] = data.get("via_mobile")
self.is_hidden: Optional[bool] = data.get("is_hidden")
self.visibility: Optional[bool] = bool(data["visibility"]) if data.get("visibility") else None
self.mentions: Optional[List[str]] = data.get("mentions")
self.visible_user_ids: Optional[List[str]] = data.get("visible_user_ids")
self.file_ids: Optional[List[str]] = data.get("file_ids")
self.files: Optional[List[str]] = data.get("files")
self.tags: Optional[List[str]] = data.get("tags")
self.poll: Optional[List[str]] = data.get("poll")
self.channel: Optional[Channel] = Channel(data["channel"]) if data.get("channel") else None
self.local_only: Optional[bool] = data.get("local_only")
self.emojis: Optional[List[Emoji]] = [Emoji(i) for i in data["emojis"]] if data.get("emojis") else None
self.reactions: Optional[Dict[str, Any]] = data.get("reactions")
self.renote_count: Optional[int] = data.get("renote_count")
self.replies_count: Optional[int] = data.get("replies_count")
self.uri: Optional[str] = data.get("uri")
self.url: Optional[str] = data.get("url")
self.my_reaction: Optional[Dict[str, Any]] = data.get("my_reaction")
self.id: Optional[str] = data.get('id')
self.created_at: Optional[datetime] = datetime.strptime(
data['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ'
) if data.get('created_at') else None
self.text: Optional[str] = data.get('text')
self.cw: Optional[str] = data.get('cw')
self.user_id: Optional[str] = data.get('user_id')
self.user: Optional[User] = User(
RawUser(data['user']), client=client
) if data.get('user') else None
self.reply_id: Optional[str] = data.get('reply_id')
self.reply: Optional[Dict[str, Any]] = data.get('reply')
self.renote: Optional[Dict[str, Any]] = data.get('renote')
self.via_mobile: Optional[bool] = data.get('via_mobile')
self.is_hidden: Optional[bool] = data.get('is_hidden')
self.visibility: Optional[bool] = bool(data['visibility']) if data.get(
'visibility'
) else None
self.mentions: Optional[List[str]] = data.get('mentions')
self.visible_user_ids: Optional[List[str]] = data.get('visible_user_ids')
self.file_ids: Optional[List[str]] = data.get('file_ids')
self.files: Optional[List[str]] = data.get('files')
self.tags: Optional[List[str]] = data.get('tags')
self.poll: Optional[List[str]] = data.get('poll')
self.channel: Optional[Channel] = Channel(data['channel']) if data.get(
'channel'
) else None
self.local_only: Optional[bool] = data.get('local_only')
self.emojis: Optional[List[Emoji]] = [
Emoji(i) for i in data['emojis']
] if data.get('emojis') else None
self.reactions: Optional[Dict[str, Any]] = data.get('reactions')
self.renote_count: Optional[int] = data.get('renote_count')
self.replies_count: Optional[int] = data.get('replies_count')
self.uri: Optional[str] = data.get('uri')
self.url: Optional[str] = data.get('url')
self.my_reaction: Optional[Dict[str, Any]] = data.get('my_reaction')
class PinnedPage:
def __init__(self, data: PinnedPagePayload):
self.id: Optional[str] = data.get("id")
self.created_at: Optional[datetime] = datetime.strptime(data["created_at"], '%Y-%m-%dT%H:%M:%S.%fZ') if data.get(
"created_at") else None
self.updated_at: Optional[str] = data.get("updated_at")
self.title: Optional[str] = data.get("title")
self.name: Optional[str] = data.get("name")
self.summary: Optional[str] = data.get("summary")
self.content: Optional[List] = data.get("content")
self.variables: Optional[List] = data.get("variables")
self.user_id: Optional[str] = data.get("user_id")
self.author: Optional[Dict[str, Any]] = data.get("author")
self.id: Optional[str] = data.get('id')
self.created_at: Optional[datetime] = datetime.strptime(
data['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ'
) if data.get('created_at') else None
self.updated_at: Optional[str] = data.get('updated_at', None)
self.title: Optional[str] = data.get('title')
self.name: Optional[str] = data.get('name')
self.summary: Optional[str] = data.get('summary')
self.content: Optional[List] = data.get('content')
self.variables: Optional[List] = data.get('variables')
self.user_id: Optional[str] = data.get('user_id')
self.author: Optional[Dict[str, Any]] = data.get('author')
class FieldContent:
def __init__(self, data: FieldContentPayload):
self.name: str = data["name"]
self.value: str = data["value"]
self.name: str = data['name']
self.value: str = data['value']
class User:

7
mipac/types/chart.py
Unescape View File

@ -1,4 +1,9 @@
__all__ = ('ActiveUsersChartPayload', 'DriveLocalChartPayload', 'DriveRemoteChartPayload', 'DriveChartPayload')
__all__ = (
'ActiveUsersChartPayload',
'DriveLocalChartPayload',
'DriveRemoteChartPayload',
'DriveChartPayload',
)
from typing import List, TypedDict, Union

8
mipac/types/instance.py
Unescape View File

@ -5,7 +5,13 @@ from typing import TYPE_CHECKING, List, Optional, TypedDict
if TYPE_CHECKING:
from mipac.types import EmojiPayload
__all__ = ('FeaturesPayload', 'MetaPayload', 'InstancePayload', 'OptionalInstance', 'OptionalMeta')
__all__ = (
'FeaturesPayload',
'MetaPayload',
'InstancePayload',
'OptionalInstance',
'OptionalMeta',
)
class FeaturesPayload:

11
mipac/types/note.py
Unescape View File

@ -4,13 +4,21 @@ from .drive import FilePayload
from .emoji import EmojiPayload
from .user import UserPayload
__all__ = ('NotePayload', 'GeoPayload', 'ReactionPayload', 'PollPayload', 'RenotePayload', 'OptionalReaction')
__all__ = (
'NotePayload',
'GeoPayload',
'ReactionPayload',
'PollPayload',
'RenotePayload',
'OptionalReaction',
)
class GeoPayload(TypedDict):
"""
衛星情報
"""
coordinates: Optional[List[Any]]
altitude: Optional[int]
accuracy: Optional[int]
@ -56,6 +64,7 @@ class _NoteOptional(TypedDict, total=False):
"""
ノートに必ず存在すると限らない物
"""
text: str
cw: str
geo: GeoPayload

9
mipac/types/user.py
Unescape View File

@ -4,7 +4,14 @@ from .drive import FilePayload
from .emoji import EmojiPayload
from .instance import InstancePayload
__all__ = ('ChannelPayload', 'FieldContentPayload', 'UserPayload', 'PinnedPagePayload', 'PinnedNotePayload', 'OptionalUser')
__all__ = (
'ChannelPayload',
'FieldContentPayload',
'UserPayload',
'PinnedPagePayload',
'PinnedNotePayload',
'OptionalUser',
)
class ChannelPayload(TypedDict):

44
mipac/util.py
Unescape View File

@ -20,10 +20,10 @@ __all__ = (
'remove_dict_empty',
'upper_to_lower',
'str_lower',
'bool_to_string'
'bool_to_string',
)
T = TypeVar("T")
T = TypeVar('T')
def deprecated_func(func):
@ -40,11 +40,11 @@ def get_cache_key(func):
async def decorator(self, *args, **kwargs):
ordered_kwargs = sorted(kwargs.items())
key = (
(func.__module__ or "")
+ '.{0}'
+ f'{self}'
+ str(args)
+ str(ordered_kwargs)
(func.__module__ or '')
+ '.{0}'
+ f'{self}'
+ str(args)
+ str(ordered_kwargs)
)
return await func(self, *args, **kwargs, cache_key=key)
@ -54,11 +54,11 @@ def get_cache_key(func):
def key_builder(func, cls, *args, **kwargs):
ordered_kwargs = sorted(kwargs.items())
key = (
(func.__module__ or "")
+ f'.{func.__name__}'
+ f'{cls}'
+ str(args)
+ str(ordered_kwargs)
(func.__module__ or '')
+ f'.{func.__name__}'
+ f'{cls}'
+ str(args)
+ str(ordered_kwargs)
)
return key
@ -149,8 +149,10 @@ def remove_dict_empty(data: Dict[str, Any]) -> Dict[str, Any]:
def upper_to_lower(
data: Dict[str, Any], field: Optional[Dict[str, Any]] = None, nest: bool = True,
replace_list: Optional[Dict[str, Any]] = None
data: Dict[str, Any],
field: Optional[Dict[str, Any]] = None,
nest: bool = True,
replace_list: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""
Parameters
@ -178,12 +180,12 @@ def upper_to_lower(
if field is None:
field = {}
for attr in data:
pattern = re.compile("[A-Z]")
pattern = re.compile('[A-Z]')
large = [i.group().lower() for i in pattern.finditer(attr)]
result = [None] * (len(large + pattern.split(attr)))
result[::2] = pattern.split(attr)
result[1::2] = ["_" + i.lower() for i in large]
default_key = "".join(result)
result[1::2] = ['_' + i.lower() for i in large]
default_key = ''.join(result)
if replace_list.get(attr):
default_key = default_key.replace(attr, replace_list.get(attr))
field[default_key] = data[attr]
@ -193,12 +195,12 @@ def upper_to_lower(
def str_lower(text: str):
pattern = re.compile("[A-Z]")
pattern = re.compile('[A-Z]')
large = [i.group().lower() for i in pattern.finditer(text)]
result = [None] * (len(large + pattern.split(text)))
result[::2] = pattern.split(text)
result[1::2] = ["_" + i.lower() for i in large]
return "".join(result)
result[1::2] = ['_' + i.lower() for i in large]
return ''.join(result)
def bool_to_string(boolean: bool) -> str:
@ -214,4 +216,4 @@ def bool_to_string(boolean: bool) -> str:
true or false: str
小文字になったbool文字列
"""
return "true" if boolean else "false"
return 'true' if boolean else 'false'

Write Preview
Loading…
Cancel
Save