aminoacid.abc

  1from __future__ import annotations
  2
  3from abc import ABC
  4from base64 import b64encode, urlsafe_b64decode
  5from importlib.util import find_spec
  6from os import PathLike, path
  7from typing import TYPE_CHECKING, BinaryIO, Callable, Dict, List, Optional, Union
  8
  9from .util import str_to_ts
 10
 11_ORJSON = find_spec("orjson")
 12if _ORJSON:
 13    import orjson as json
 14else:
 15    import json
 16
 17
 18if TYPE_CHECKING:
 19    from . import Bot
 20
 21
 22class AminoBaseClass(ABC):
 23    """This is the base class for all other classes defined by this library, except for clients and exceptions."""
 24
 25    def __init__(self) -> None:
 26        super().__init__()
 27
 28    def __repr__(self) -> str:
 29        try:  #! pdoc complained that vars(self) is a function (which it is NOT) so there's this try/except here to catch that
 30            _temp = [
 31                f"{attr}={value!r}, "
 32                for attr, value in vars(self).items()
 33                if value
 34                and not isinstance(
 35                    value,
 36                    (Callable, type(self.client))
 37                    if "client" in dir(self)
 38                    else Callable,
 39                )
 40            ]
 41            return f"{type(self).__name__}({''.join(_temp).rstrip(', ')})"
 42        except AttributeError:
 43            return f"{type(self).__name__}()"
 44
 45
 46class MessageAble(AminoBaseClass):
 47    def __init__(self, bot: Bot) -> None:
 48        self._bot = bot
 49        super().__init__()
 50
 51    async def send(self, content: str, **kwargs) -> Message:
 52        """Send a Message to the messageable object
 53
 54        Parameters
 55        ----------
 56        content : str
 57            Content of the message, cannot be empty, must be within 2000 characters
 58        embed : Embed, optional
 59            The embed to send alongside the message, by default ...
 60
 61        Returns
 62        -------
 63        Message
 64            Message object of the sent message
 65        """
 66        if isinstance(self, Message):
 67            return await self.client.send_message(
 68                content=content, threadId=self.threadId, ndcId=self.ndcId, **kwargs
 69            )
 70        if isinstance(self, Context):
 71            return await self.client.send_message(
 72                content=content,
 73                threadId=self.message.threadId,
 74                ndcId=self.message.ndcId,
 75                **kwargs,
 76            )
 77        if isinstance(self, Thread):
 78            return await self.client.send_message(
 79                content=content, threadId=self.id, ndcId=self.ndcId, **kwargs
 80            )
 81        if isinstance(self, User):
 82            return await self.client.message_user(
 83                content=content,
 84                userId=self.id,
 85                ndcId=self.ndcId if "ndcId" in dir(self) else 0,
 86                **kwargs,
 87            )
 88
 89
 90class Context(MessageAble):
 91    """Context of a message, this is passed into `aminoacid.util.UserCommand`s as the first positional argument"""
 92
 93    def __init__(
 94        self,
 95        client: Bot,
 96        message: Message,
 97    ) -> None:
 98        """Initialises the Context
 99
100        Parameters
101        ----------
102        client : Bot
103            The currently running client, used for replying, deleting, etc.
104        message : Message
105            The message the context describes
106        """
107        self.client = client
108        self.message = message
109        self.thread = message.thread
110        self.author = message.author
111
112        super().__init__(bot=client)
113
114    async def reply(self, content: str, **kwargs):
115        """Replies to the message described by the context (`self.message`)
116
117        Parameters
118        ----------
119        content : str
120            Content of the message to reply with
121        """
122        await super().send(content, replyMessageId=self.message.id, **kwargs)
123
124
125class User(MessageAble):
126    nickname: str
127    icon: str
128    content: str
129    id: str
130    status: int
131    moodSticker: dict
132    itemsCount: int
133    consecutiveCheckInDays: int
134    modifiedTime: str
135    followingStatus: int
136    onlineStatus: int
137    accountMembershipStatus: int
138    isGlobal: bool = True
139    avatarFrameId: str
140    fanClubList: list = []
141    reputation: int = 0
142    postsCount: int = 0
143    avatarFrame: Frame
144    membersCount: int
145    mediaList: list
146    isNicknameVerified: bool = False
147    visitorsCount: int = 0
148    mood: None
149    level: int = 0
150    notificationSubscriptionStatus: int = 0
151    settings: dict
152    pushEnabled: bool
153    membershipStatus: int
154    joinedCount: int
155    role: int = 0
156    commentsCount: int = 0
157    aminoId: str
158    createdTime: str
159    extensions: dict
160    visitPrivacy: int = 1
161    storiesCount: int = 0
162
163    client: Bot
164
165    def __init__(self, data: dict = {}, **kwargs) -> None:
166        """Initialises a new `User` object, calls `from_dict()` with a combination of the kwargs and the supplied data
167
168        Parameters
169        ----------
170        data : dict, optional
171            The data to initialise the `User` with, by default {}
172        """
173        self.from_dict({**data, **kwargs})
174        super().__init__(bot=self.client)
175
176    def from_dict(self, data: dict):
177        """Create a new `User` object from a dict
178
179        Parameters
180        ----------
181        data : dict
182            The dict to create the new `User` object from
183        """
184        self.nickname = data.pop("nickname", "")
185        self.content = data.pop("content", "")
186        self.icon = data.pop("icon", "")
187        self.id = data.pop("uid", "")
188        self.client = data.pop("client")
189
190    async def send(self, content: str, **kwargs) -> Message:
191        return await super().send(content, **kwargs)
192
193    async def get(self) -> User:
194        """Get the complete `User` object, used when a `User` is received partially by the socket to get the missing information."""
195        return await self.client.fetch_user(self.id)
196
197
198class Member(User):
199    ndcId: int
200
201    def __init__(self, data: dict = {}, **kwargs) -> None:
202        """Initialises a new `Member` object, calls `from_dict()` with a combination of the kwargs and the supplied data
203
204        Parameters
205        ----------
206        data : dict, optional
207            The data to initialise the `Member` with, by default {}
208        """
209        self.from_dict({**data, **kwargs})
210        super().__init__(data, **kwargs)
211
212    def from_dict(self, data: dict):
213        """Create a new `Member` object from a dict
214
215        Parameters
216        ----------
217        data : dict
218            The dict to create the new `Member` object from
219        """
220        self.ndcId = data.pop("ndcId")
221        return super().from_dict(data)
222
223    async def fetch_blogs(self) -> List[Blog]:
224        """Returns a list of blogs the user made
225
226        Returns
227        -------
228        List[Blog]
229            List of blogs
230        """
231        return await self.client.fetch_blogs(self.ndcId, userId=self.id)
232
233
234class Message(AminoBaseClass):
235    id: str
236    type: int
237    content: str
238    author: Union[User, Member]
239    threadId: str
240    ndcId: Optional[int] = None
241    isHidden: bool
242    includedInSummary: bool
243    createdTime: int
244    extensions: dict
245    alertOption: int
246    membershipStatus: int
247    chatBubbleId: str
248    chatBubbleVersion: int
249    clientRefId: int
250    mediaType: int
251
252    client: Bot
253
254    def __init__(self, data: dict = {}, **kwargs) -> None:
255        """Initialises a new `Message` object, calls `from_dict()` with a combination of the kwargs and the supplied data
256
257        Parameters
258        ----------
259        data : dict, optional
260            The data to initialise the `Message` with, by default {}
261        """
262        self.from_dict({**data, **kwargs})
263        super().__init__()
264        self.startswith = self.content.startswith
265
266    def from_dict(self, data: dict):
267        """Create a new `Message` object from a dict
268
269        Parameters
270        ----------
271        data : dict
272            The dict to create the new `Message` object from
273        """
274        self.client = data.pop("client")
275
276        self.nickname = data.pop("nickname", "")
277        self.content = data.pop("content", "")
278        self.id = data.pop("messageId", "")
279        self.threadId = data.pop("threadId", "")
280        self.ndcId = data.pop("ndcId", 0)
281        self.thread = Thread(id=self.threadId, ndcId=self.ndcId, client=self.client)
282        if self.ndcId:
283            self.author = Member(
284                data.pop("author", {}), client=self.client, ndcId=self.ndcId
285            )
286        else:
287            self.author = User(data.pop("author", {}), client=self.client)
288        self.type = data.pop("type", None)
289
290        self.createdTime = str_to_ts(data.pop("createdTime", ""))
291        self.alertOption = data.pop("alertOption", None)
292        self.chatBubbleId = data.pop("chatBubbleId", "")
293        self.clientRefId = data.pop("clientRefId", 0)
294        self.extensions = data.pop("extensions", {})
295        self.isHidden = data.pop("isHidden", False)
296        self.membershipStatus = data.pop("membershipStatus", 0)
297        self.includedInSummary = data.pop("includedInSummary", True)
298        self.mediaType = data.pop("mediaType", 0)
299
300    async def get(self):
301        """Get the complete `Message` object, used when a `Message` is received partially by the socket to get the missing information."""
302        await self.client.fetch_message(
303            messageId=self.id, threadId=self.thread.id, ndcId=self.ndcId
304        )
305
306
307class Thread(MessageAble):
308    id: str
309    content: str
310    author: User
311    ndcId: Optional[int] = None
312    title: str
313
314    # TODO: FINISH THIS
315
316    def __init__(self, data: dict = {}, **kwargs) -> None:
317        """Initialises a new `Thread` object, calls `from_dict()` with a combination of the kwargs and the supplied data
318
319        Parameters
320        ----------
321        data : dict, optional
322            The data to initialise the `Thread` with, by default {}
323        """
324        self.from_dict({**data, **kwargs})
325        super().__init__(bot=self.client)
326
327    def from_dict(self, data: dict):
328        """Create a new `Thread` object from a dict
329
330        Parameters
331        ----------
332        data : dict
333            The dict to create the new `Thread` object from
334        """
335        self.client = data.pop("client")
336
337        self.title = data.pop("title", "")
338        self.content = data.pop("content", "")
339        self.author = data.pop("author", "")
340        self.id = data.pop("threadId", "") or data.pop("id", "")
341        self.ndcId = data.pop("ndcId", "")
342
343    async def send(self, content: str, **kwargs) -> Message:
344        return await super().send(content, **kwargs)
345
346    async def get(self):
347        """Get the complete `Thread` object, used when a `Thread` is received partially by the socket to get the missing information."""
348        self.from_dict(
349            (await self._bot.get_thread(ndcId=self.ndcId, threadId=self.id)).__dict__
350        )
351
352
353class Community(AminoBaseClass):
354    userAddedTopicList: Optional[List]
355    agent: Member
356    listedStatus: int
357    probationStatus: int
358    themePack: dict
359    membersCount: int
360    primaryLanguage: str
361    communityHeat: int
362    strategyIngo: str
363    tagline: str
364    joinType: int
365    status: int
366    launchPage: dict
367    modifiedTime: int
368    ndcId: str
369    activeInfo: Optional[Dict]
370    link: str
371    icon: str
372    endpoint: str
373    name: str
374    templateId: int
375    createdTime: int
376    promotionalMediaList: Optional[List]
377
378    # TODO: Implement this!!
379
380    def __init__(self, data: dict = {}, **kwargs) -> None:
381        """Initialises a new `Community` object, calls `from_dict()` with a combination of the kwargs and the supplied data
382
383        Parameters
384        ----------
385        data : dict, optional
386            The data to initialise the `Community` with, by default {}
387        """
388        self.from_dict({**data, **kwargs})
389        super().__init__()
390
391    def from_dict(self, data: dict):
392        """Create a new `Community` object from a dict
393
394        Parameters
395        ----------
396        data : dict
397            The dict to create the new `Community` object from
398        """
399        self.client = data.pop("client")
400
401        self.name = data.pop("name", "")
402        self.id = data.pop("ndcId", "")
403
404
405class Blog(AminoBaseClass):
406    createdTime: int
407    mediaList: list
408    type: int
409    status: int
410    modifiedTime: int
411    id: str
412    ndcId: str
413    title: str
414    content: str
415    author: Member
416
417    client: Bot
418
419    def __init__(self, data: dict = {}, **kwargs) -> None:
420        """Initialises a new `Blog` object, calls `from_dict()` with a combination of the kwargs and the supplied data
421
422        Parameters
423        ----------
424        data : dict, optional
425            The data to initialise the `Blog` with, by default {}
426        """
427        self.from_dict({**data, **kwargs})
428        super().__init__()
429
430    def from_dict(self, data: dict):
431        """Create a new `Blog` object from a dict
432
433        Parameters
434        ----------
435        data : dict
436            The dict to create the new `Blog` object from
437        """
438        self.client = data.pop("client")
439
440        self.title = data.pop("title", "")
441        self.id = data.pop("blogId", "") or data.pop("id", "")
442        self.ndcId = data.pop("ndcId", "")
443        self.author = Member(
444            data.pop("author", {}), client=self.client, ndcId=self.ndcId
445        )
446        self.content = data.pop("content", "")
447        self.createdTime = str_to_ts(data.pop("createdTime", ""))
448        self.modifiedTime = str_to_ts(data.pop("modifiedTime", ""))
449        self.mediaList = data.pop("mediaList", [])
450        self.status = data.pop("status", "")
451        self.type = data.pop("type")
452
453    async def tip(self, amount: int):
454        """Send coins to a blog, shortcut to `aminoacid.client.ApiClient.tip_blog()`
455
456        Parameters
457        ----------
458        amount : int
459            amount of coins to send
460        """
461        await self.client.tip_blog(self.ndcId, self.id, amount)
462
463
464class Embed(AminoBaseClass):
465    def __init__(
466        self,
467        objectId: str,
468        objectType: int,
469        link: str,
470        title: str,
471        content: str,
472        mediaList: list,
473    ) -> None:
474
475        self.id = objectId
476        self.type = objectType
477        self.link = link
478        self.title = title
479        self.content = content
480        self.mediaList = mediaList
481
482        super().__init__()
483
484    def __dict__(self):
485        return {
486            "objectId": self.id,
487            "objectType": self.type,
488            "link": self.link,
489            "title": self.title,
490            "content": self.content,
491            "mediaList": self.mediaList,
492        }
493
494
495class linkSnippet(AminoBaseClass):
496    def __init__(self, link: str, image: Union[BinaryIO, PathLike]) -> None:
497        """Initialises a new Link Snippet object to use for sending them in chat
498
499        Parameters
500        ----------
501        link : str
502            link of the snipped
503        image : Union[BinaryIO, PathLike]
504            Either a Path of the image or an IO representation of the image
505        """
506        if isinstance(image, PathLike):
507            if path.exists(image):
508                with open(image, "rb") as f:
509                    self.image = b64encode(f.read())
510        else:
511            self.image = b64encode(image.read())
512        self.link = link
513        super().__init__()
514
515    async def prepare(self, client: Bot) -> dict:
516        """Prepare the link snippet by uploading `self.image` to amino servers via `aminoacid.Bot.upload_image()`
517
518        Parameters
519        ----------
520        client : Bot
521            Bot object to upload the image
522
523        Returns
524        -------
525        dict
526            Dict containing all the information of the linkSnippet object
527        """
528        return {
529            "link": self.link,
530            "mediaType": 100,
531            "mediaUploadValue": client.upload_image(self.image),
532            "mediaUploadValueContentType": "image/png",
533        }
534
535
536class Frame(AminoBaseClass):
537    # TODO: Add Frame class for OOP
538    ...
539
540
541class Bubble(AminoBaseClass):
542    # TODO: Same as Frame
543    ...
544
545
546class Notification(AminoBaseClass):
547    payload: dict
548    timestamp: int
549    messageType: int
550    ndcId: str
551    threadId: str
552    isHidden: bool
553    id: str
554    type: int
555
556    def __init__(self, data: dict) -> None:
557        self.from_dict(data)
558        super().__init__()
559
560    def from_dict(self, data: dict):
561        """Create a new `Notification` object from a dict
562
563        Parameters
564        ----------
565        data : dict
566            The dict to create the new `Notification` object from
567        """
568
569        self.payload = data.pop("payload", {})
570
571        self.timestamp = self.payload.get("ts", "")
572        self.threadId = self.payload.get("tid", "")
573        self.isHiddem = self.payload.get("isHidden", "")
574        self.id = self.payload.get("id", "")
575        self.ndcId = self.payload.get("ndcId", "")
576        self.messageType = self.payload.get("msgType", 0)
577        self.type = self.payload.get("notifType")
578
579        self.data = data
580
581        if self.timestamp:
582            self.timestamp = str_to_ts(self.timestamp)
583
584
585class Session(AminoBaseClass):
586    def __init__(self, session: str) -> None:
587        self.sid = f"sid={session}"
588        self.secret = self.deserialize_session(session)
589
590        self.uid: str = self.secret["2"]
591        self.ip: str = self.secret["4"]
592        self.created: int = self.secret["5"]
593        self.clientType: int = self.secret["6"]
594
595        super().__init__()
596
597    @staticmethod
598    def deserialize_session(session: str) -> dict:
599        """Takes all the info from a given session
600
601        Parameters
602        ----------
603        session : str
604            the session to deserialize
605
606        Returns
607        -------
608        dict
609            dictionary containing all of the information
610        """
611        return json.loads(
612            urlsafe_b64decode(session + "=" * (4 - len(session) % 4))[1:-20]
613        )