mrkrk/melisa
1
0
Fork 0
mirror of https://github.com/MelisaDev/melisa.git synced 2024-09-22 19:22:01 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions

uhm, something with types and members i dont remember

Browse source
This commit is contained in:
grey-cat-1908 2022-05-25 10:02:34 +03:00
parent 1ea77f8c11
commit 1fb0615b99
11 changed files with 364 additions and 139 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
melisa/__init__.py
View file

@ -1,7 +1,9 @@
# Copyright MelisaDev 2022 - Present
# Full MIT License can be found in `LICENSE.txt` at the project root.
from .client import *
from .client import (
Client, Bot
)
from .models import *
from .exceptions import *

7
melisa/models/app/cache.py → melisa/cache.py
View file

@ -6,9 +6,10 @@ from __future__ import annotations
from enum import Enum
from typing import List, Dict, Optional, Any, Union
from melisa.utils.types import UNDEFINED
from melisa.models.guild import Guild, ChannelType, UnavailableGuild, Channel
from melisa.utils.snowflake import Snowflake
from .utils.types import UNDEFINED
from .models.guild.guild import Guild, UnavailableGuild
from .models.guild.channel import ChannelType, Channel
from .utils.snowflake import Snowflake
class AutoCacheModels(Enum):

2
melisa/client.py
View file

@ -8,7 +8,7 @@ import sys
import traceback
from typing import Dict, List, Union, Any, Iterable, Optional, Callable
from .models.app.cache import CacheManager
from .cache import CacheManager
from .rest import RESTApp
from .core.gateway import GatewayBotInfo
from .models.guild.channel import Channel, ChannelType

24
melisa/models/guild/__init__.py
View file

@ -1,8 +1,28 @@
# Copyright MelisaDev 2022 - Present
# Full MIT License can be found in `LICENSE.txt` at the project root.
from .guild import *
from .channel import *
from .guild import (
DefaultMessageNotificationLevel,
ExplicitContentFilterLevel,
MFALevel,
VerificationLevel,
GuildNSFWLevel,
PremiumTier,
SystemChannelFlags,
Guild,
UnavailableGuild,
)
from .channel import (
ChannelType,
VideoQualityModes,
Channel,
MessageableChannel,
NoneTypedChannel,
TextChannel,
Thread,
ThreadsList,
_choose_channel_type
)
from .thread import *
from .webhook import *
from .emoji import *

95
melisa/models/guild/channel.py
View file

@ -14,19 +14,18 @@ from typing import (
Union,
Dict,
overload,
TYPE_CHECKING,
TYPE_CHECKING
)
from ..message.file import File, create_form
from ..message.message import Message, AllowedMentions
from ...exceptions import EmbedFieldError
from ...models.message.embed import Embed
from ...utils import Snowflake, Timestamp
from ...utils.api_model import APIModelBase
from ...utils.types import APINullable, UNDEFINED
from ...utils.types import APINullable
from .thread import ThreadMember, ThreadMetadata
if TYPE_CHECKING:
from .thread import ThreadMember, ThreadMetadata
from ...models.message.embed import Embed
from ..message.file import File
from ..message.message import AllowedMentions, Message
def _choose_channel_type(data):
@ -381,32 +380,10 @@ class MessageableChannel(Channel):
An iterator of messages.
"""
# ToDo: Add check parameter
data = self._client.rest.get_channel_messages_history(self.id, limit, around=around, before=before, after=after)
if limit is None:
limit = 100
while limit > 0:
search_limit = min(limit, 100)
raw_messages = await self._http.get(
f"/channels/{self.id}/messages",
params={
"limit": search_limit,
"before": before,
"after": after,
"around": around,
},
)
if not raw_messages:
break
for message_data in raw_messages:
yield Message.from_dict(message_data)
before = raw_messages[-1]["id"]
limit -= search_limit
for i in data:
yield i
async def fetch_message(
self,
@ -434,11 +411,7 @@ class MessageableChannel(Channel):
Message object.
"""
message = await self._http.get(
f"/channels/{self.id}/messages/{message_id}",
)
return Message.from_dict(message)
return self._client.rest.fetch_message(self.id, message_id)
async def pins(self) -> AsyncIterator[Message]:
"""|coro|
@ -456,12 +429,10 @@ class MessageableChannel(Channel):
AsyncIterator of Message objects.
"""
messages = await self._http.get(
f"/channels/{self.id}/pins",
)
data = self._client.rest.fetch_channel_pins(self.id)
for message in messages:
yield Message.from_dict(message)
for i in data:
yield i
async def bulk_delete_messages(
self, messages: List[Snowflake], *, reason: Optional[str] = None
@ -565,45 +536,7 @@ class MessageableChannel(Channel):
Some of specified parameters is invalid.
"""
# ToDo: Add other parameters
# ToDo: add file checks
if embeds is None:
embeds = [embed.to_dict()] if embed is not None else []
if files is None:
files = [file] if file is not None else []
payload = {"content": str(content) if content is not None else None}
for _embed in embeds:
if embed.total_length() > 6000:
raise EmbedFieldError.characters_from_desc(
"Embed", embed.total_length(), 6000
)
payload["embeds"] = embeds
payload["tts"] = tts
# ToDo: add auto allowed_mentions from client
if allowed_mentions is not None:
payload["allowed_mentions"] = allowed_mentions.to_dict()
elif self._client.allowed_mentions is not None:
payload["allowed_mentions"] = self._client.allowed_mentions.to_dict()
content_type, data = create_form(payload, files)
message_data = Message.from_dict(
await self._http.post(
f"/channels/{self.id}/messages",
data=data,
headers={"Content-Type": content_type},
)
)
if delete_after:
await message_data.delete(delay=delete_after)
return message_data
return self._client.rest.create_message(self.id, content, tts=tts, embed=embed, embeds=embeds, file=file, files=files, allowed_mentions=allowed_mentions, delete_after=delete_after, _client_allowed_mentions=self._client.allowed_mentions)
async def purge(
self,

20
melisa/models/guild/guild.py
View file

@ -4,21 +4,25 @@
from __future__ import annotations
from dataclasses import dataclass
from enum import IntEnum, Enum
from typing import List, Any, Optional, overload, Dict
from enum import IntEnum
from typing import List, Any, Optional, overload, Dict, TYPE_CHECKING
from .channel import (
Channel,
ChannelType,
ThreadsList,
Thread,
_choose_channel_type,
)
from .member import GuildMember
from ...utils import Snowflake, Timestamp
from ...utils.api_model import APIModelBase
from ...utils.conversion import try_enum
from ...utils.types import APINullable
if TYPE_CHECKING:
from .channel import ChannelType, Channel
class DefaultMessageNotificationLevel(IntEnum):
"""Message notification level
@ -244,7 +248,7 @@ class Guild(APIModelBase):
Total number of members in this guild
voice_states:
States of members currently in voice channels; lacks the `guild_id` key
members: APINullable[:class:`typing.Any`]
members: APINullable[:class:`~melisa.models.guild.member.GuildMember`]
Users in the guild
channels: APINullable[Dict[:class:`~melisa.models.guild.channel.Channel`]]
Channels in the guild
@ -457,11 +461,15 @@ class Guild(APIModelBase):
self.owner = None if data.get("owner") is None else data["owner"]
self.large = None if self.member_count == 0 else self.member_count >= 250
self.voice_states = data.get("voice_states")
self.members = data.get("members")
self.presences = data.get("presences")
self.threads = {}
self.channels = {}
self.members = {}
for member in data.get("members", []):
member = GuildMember.from_dict(member)
self.members[member.user.id] = member
for thread in data.get("threads", []):
self.threads[thread["id"]] = Thread.from_dict(thread)

27
melisa/models/guild/member.py
View file

@ -53,6 +53,7 @@ class GuildMember(APIModelBase):
guild_id: List[:class:`~melisa.utils.snowflake.Snowflake`]
The id of the guild this member belongs to.
"""
user: APINullable[User] = None
nick: APINullable[str] = None
guild_avatar: APINullable[str] = None
@ -91,17 +92,31 @@ class GuildMember(APIModelBase):
self: GuildMember = super().__new__(cls)
self.user = User.from_dict(data['user']) if data.get('user') is not None else None
self.user = (
User.from_dict(data["user"]) if data.get("user") is not None else None
)
self.nick = data.get("nick")
self.guild_avatar = data.get('avatar')
self.role_ids = [Snowflake(x) for x in data['roles']]
self.joined_at = Timestamp.parse(data['joined_at']) if data.get('joined_at') is not None else None
self.premium_since = Timestamp.parse(data['premium_since']) if data.get('premium_since') is not None else None
self.guild_avatar = data.get("avatar")
self.role_ids = [Snowflake(x) for x in data["roles"]]
self.joined_at = (
Timestamp.parse(data["joined_at"])
if data.get("joined_at") is not None
else None
)
self.premium_since = (
Timestamp.parse(data["premium_since"])
if data.get("premium_since") is not None
else None
)
self.is_deaf = data.get("deaf")
self.is_mute = data.get("mute")
self.is_pending = data.get("pending")
self.permissions = data.get("permissions")
self.communication_disabled_until = Timestamp.parse(data['communication_disabled_until']) if data.get('communication_disabled_until') is not None else None
self.communication_disabled_until = (
Timestamp.parse(data["communication_disabled_until"])
if data.get("communication_disabled_until") is not None
else None
)
self.guild_id = data.get("guild_id")
return self

9
melisa/models/message/__init__.py
View file

@ -1,6 +1,13 @@
# Copyright MelisaDev 2022 - Present
# Full MIT License can be found in `LICENSE.txt` at the project root.
from .message import *
from .message import (
MessageType,
MessageActivityType,
MessageFlags,
AllowedMentions,
Message,
)
from .embed import *
from .colors import *
from .file import File

25
melisa/models/message/file.py
View file

@ -5,30 +5,7 @@ from __future__ import annotations
import io
import os
from typing import Union, Dict, Any, List, Tuple
from aiohttp import FormData, Payload
import melisa.utils.json as json
def create_form(payload: Dict[str, Any], files: List[File]):
"""
Creates an aiohttp payload from an array of File objects.
"""
form = FormData()
form.add_field("payload_json", json.dumps(payload))
for index, file in enumerate(files):
form.add_field(
"file",
file.filepath,
filename=file.filename,
content_type="application/octet-stream",
)
payload = form()
return payload.headers["Content-Type"], payload
from typing import Union
class File:

27
melisa/models/message/message.py
View file

@ -12,8 +12,11 @@ from .embed import Embed
from ...utils import Snowflake, Timestamp, try_enum, APIModelBase
from ...utils.types import APINullable, UNDEFINED
if TYPE_CHECKING:
from ..guild.channel import Thread, _choose_channel_type
# if TYPE_CHECKING:
# from . import Thread, _choose_channel_type
from ..guild.channel import Thread, _choose_channel_type
from ..guild.member import GuildMember
class MessageType(IntEnum):
@ -190,10 +193,8 @@ class Message(APIModelBase):
Object of guild where message was sent in
guild_id: :class:`~melisa.utils.types.snowflake.Snowflake`
Id of the guild the message was sent in
author: :class:`typing.Any`
The author of this message (not guaranteed to be a valid user, see below)
member: :class:`typing.Any`
Member properties for this message's author
author: :class:`~melisa.models.guild.member.GuildMember`
The author of this message.
content: :class:`str`
Contents of the message
timestamp: :class:`~melisa.utils.timestamp.Timestamp`
@ -251,8 +252,7 @@ class Message(APIModelBase):
id: APINullable[Snowflake] = None
channel_id: APINullable[Snowflake] = None
guild_id: APINullable[Snowflake] = None
author: APINullable[Dict] = None
member: APINullable[Dict] = None
author: APINullable[GuildMember] = None
content: APINullable[str] = None
timestamp: APINullable[Timestamp] = None
edited_timestamp: APINullable[Timestamp] = None
@ -291,13 +291,16 @@ class Message(APIModelBase):
"""
self: Message = super().__new__(cls)
_member = data.get("member")
_member.update({"user": data.get("author")})
self.id = data["id"]
self.channel_id = Snowflake(data["channel_id"])
self.guild_id = (
Snowflake(data["guild_id"]) if data.get("guild_id") is not None else None
)
self.author = data.get("author") # ToDo: User object
self.member = data.get("member")
self.author = _member
self.content = data.get("content", "")
self.timestamp = Timestamp.parse(data["timestamp"])
self.edited_timestamp = (
@ -337,7 +340,7 @@ class Message(APIModelBase):
)
self.interaction = data.get("interaction")
self.thread = (
Thread.from_dict(data["thread"]) if data.get("thread") is not None else None
Thread.from_dict(data['thread']) if data.get("thread") is not None else None
)
self.components = data.get("components")
self.sticker_items = data.get("sticker_items")
@ -363,8 +366,6 @@ class Message(APIModelBase):
@property
def channel(self):
print(self.channel_id)
print(self._client.cache._channel_symlinks)
if self.channel_id is not None:
return self._client.cache.get_guild_channel(self.channel_id)

263
melisa/rest.py
View file

@ -1,15 +1,39 @@
# Copyright MelisaDev 2022 - Present
# Full MIT License can be found in `LICENSE.txt` at the project root.
from typing import Union, Optional
from typing import Union, Optional, List, Dict, Any, AsyncIterator
from aiohttp import FormData
from .models.message import Embed, File, AllowedMentions, Message
from .exceptions import EmbedFieldError
from .core.http import HTTPClient
from .utils import json
from .utils.snowflake import Snowflake
from .models.guild.guild import Guild
from .models.user.user import User
from .models.guild.channel import _choose_channel_type, Channel
def create_form(payload: Dict[str, Any], files: List[File]):
"""
Creates an aiohttp payload from an array of File objects.
"""
form = FormData()
form.add_field("payload_json", json.dumps(payload))
for index, file in enumerate(files):
form.add_field(
"file",
file.filepath,
filename=file.filename,
content_type="application/octet-stream",
)
payload = form()
return payload.headers["Content-Type"], payload
class RESTApp:
"""
This instance may be used to send http requests to the Discord REST API.
@ -99,3 +123,240 @@ class RESTApp:
f"channels/{channel_id}/messages/{message_id}",
headers={"X-Audit-Log-Reason": reason},
)
async def create_message(
self,
channel_id: Union[Snowflake, str, int],
content: str = None,
*,
tts: bool = False,
embed: Embed = None,
embeds: List[Embed] = None,
file: File = None,
files: List[File] = None,
allowed_mentions: AllowedMentions = None,
delete_after: int = None,
_client_allowed_mentions: AllowedMentions = None
) -> Message:
"""|coro|
[**REST API**] Create message.
Sends a message to the destination with the content given.
The content must be a type that can convert to a string through str(content).
Parameters
----------
channel_id: Union[:class:`int`, :class:`str`, :class:`~.melisa.utils.snowflake.Snowflake`]
Id of channel where message should be sent
content: Optional[:class:`str`]
The content of the message to send.
tts: Optional[:class:`bool`]
Whether the message should be sent using text-to-speech.
embed: Optional[:class:`~melisa.models.message.embed.Embed`]
Embed
embeds: Optional[List[:class:`~melisa.models.message.embed.Embed`]]
List of embeds
file: Optional[:class:`~melisa.models.message.file.File`]
File
files: Optional[List[:class:`~melisa.models.message.file.File`]]
List of files
allowed_mentions: Optional[:class:`~melisa.models.message.message.AllowedMentions`]
Controls the mentions being processed in this message.
delete_after: Optional[:class:`int`]
Provided value must be an int.
if provided, deletes message after some seconds.
May raise ``ForbiddenError`` or ``NotFoundError``.
Raises
-------
HTTPException
The request to perform the action failed with other http exception.
ForbiddenError
You do not have the proper permissions to send the message.
BadRequestError
Some of specified parameters is invalid.
"""
# ToDo: Add other parameters
# ToDo: add file checks
if embeds is None:
embeds = [embed.to_dict()] if embed is not None else []
if files is None:
files = [file] if file is not None else []
payload = {"content": str(content) if content is not None else None}
for _embed in embeds:
if embed.total_length() > 6000:
raise EmbedFieldError.characters_from_desc(
"Embed", embed.total_length(), 6000
)
payload["embeds"] = embeds
payload["tts"] = tts
# ToDo: add auto allowed_mentions from client
if allowed_mentions is not None:
payload["allowed_mentions"] = allowed_mentions.to_dict()
elif _client_allowed_mentions is not None:
payload["allowed_mentions"] = _client_allowed_mentions.to_dict()
content_type, data = create_form(payload, files)
message_data = Message.from_dict(
await self._http.post(
f"/channels/{channel_id}/messages",
data=data,
headers={"Content-Type": content_type},
)
)
if delete_after:
await message_data.delete(delay=delete_after)
return message_data
async def get_channel_messages_history(
self,
channel_id: Union[Snowflake, str, int],
limit: int = 50,
*,
before: Optional[Snowflake] = None,
after: Optional[Snowflake] = None,
around: Optional[Snowflake] = None,
) -> AsyncIterator[Message]:
"""|coro|
[**REST API**] Fetch messages history.
Returns a list of messages in this channel.
Examples
---------
Flattening messages into a list: ::
messages = [message async for message in channel.history(limit=111)]
All parameters are optional.
Parameters
----------
channel_id: Union[:class:`int`, :class:`str`, :class:`~.melisa.utils.snowflake.Snowflake`]
Id of channel where messages should be fetched.
limit : Optional[:class:`~.melisa.utils.snowflake.Snowflake`]
Max number of messages to return (1-100).
around : Optional[:class:`~.melisa.utils.snowflake.Snowflake`]
Get messages around this message ID.
before : Optional[:class:`~.melisa.utils.snowflake.Snowflake`]
Get messages before this message ID.
after : Optional[:class:`~.melisa.utils.snowflake.Snowflake`]
Get messages after this message ID.
Raises
-------
HTTPException
The request to perform the action failed with other http exception.
ForbiddenError
You do not have proper permissions to do the actions required.
Returns
-------
AsyncIterator[:class:`~melisa.models.message.message.Message`]
An iterator of messages.
"""
# ToDo: Add check parameter
if limit is None:
limit = 100
while limit > 0:
search_limit = min(limit, 100)
raw_messages = await self._http.get(
f"/channels/{channel_id}/messages",
params={
"limit": search_limit,
"before": before,
"after": after,
"around": around,
},
)
if not raw_messages:
break
for message_data in raw_messages:
yield Message.from_dict(message_data)
before = raw_messages[-1]["id"]
limit -= search_limit
async def fetch_message(
self,
channel_id: Union[Snowflake, int, str],
message_id: Union[Snowflake, int, str],
) -> Message:
"""|coro|
[**REST API**] Fetch message.
Returns a specific message in the channel.
Parameters
----------
message_id : Union[:class:`~.melisa.utils.snowflake.Snowflake`]
Id of message to fetch.
channel_id: Union[:class:`int`, :class:`str`, :class:`~.melisa.utils.snowflake.Snowflake`]
Id of channel where message should be fetched.
Raises
-------
HTTPException
The request to perform the action failed with other http exception.
ForbiddenError
You do not have proper permissions to do the actions required.
Returns
-------
:class:`~melisa.models.message.message.Message`
Message object.
"""
message = await self._http.get(
f"/channels/{channel_id}/messages/{message_id}",
)
return Message.from_dict(message)
async def fetch_channel_pins(self, channel_id: Union[Snowflake, int, str]) -> AsyncIterator[Message]:
"""|coro|
Retrieves all messages that are currently pinned in the channel.
Parameters
----------
channel_id: Union[:class:`int`, :class:`str`, :class:`~.melisa.utils.snowflake.Snowflake`]
Id of channel where messages should be fetched.
Raises
-------
HTTPException
The request to perform the action failed with other http exception.
Returns
-------
AsyncIterator[:class:`~melisa.models.message.message.Message`]
AsyncIterator of Message objects.
"""
messages = await self._http.get(
f"/channels/{channel_id}/pins",
)
for message in messages:
yield Message.from_dict(message)