0.1

LICENSE

@@ -0,0 +1,9 @@
+# Released under MIT License
+
+Copyright (c) 2022 Matheus Fillipe.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

README.md

@@ -0,0 +1,36 @@
+# Mailinator Public API
+
+Python wrapper for mailinator's public api. This library scrapes mailinator's websockets api and gives an abstraction layer for basic operations such as viewing the inbox, email's content and removing them.
+
+This means that this will allow you to to view and remove emails from public inboxes of (https://www.mailinator.com)[https://www.mailinator.com] without the need of a headless browser. Notice there is a limit for removing emails.
+
+
+## Usage
+
+I hope it is simple enough.
+```python
+from mailinatorapi import PublicInbox
+inbox = PublicInbox("carlos")
+print(inbox.web_url)
+# Iterate over the emails and print some info
+for email in inbox:
+    print(email.from_address)
+    print(email.from_name)
+    print(email.subject)
+    print(email.seconds_ago)
+    print(email.text)
+    print(email.html)
+    print([f"{link.text}: {link.url}" for link in email.links])
+
+# remove the last email
+email = inbox.get_lastest_email()
+print(email.from_address)
+ok, msg = email.remove()
+if ok:
+    print("Removal successful! ", msg)
+else:
+    print("Removal failed! ", msg)
+
+```
+
+You can access the raw json response for the email object with `email.json`.

build.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+[[ "$1" == "install" ]] && python3 setup.py install --user
+[[ "$1" == "update" ]] && (
+	rm -r dist
+	rm -r re_ircbot.egg-info
+	python3 setup.py sdist bdist_wheel
+	python3 -m twine upload dist/* --verbose
+) || echo "Use update or install"

mailinator.py

@@ -0,0 +1,179 @@
+import asyncio
+import json
+import random
+import string
+
+import requests
+import websockets
+
+ID_LEN = 29
+BASE_URL = "mailinator.com"
+MAILINATOR_WSS_URL = f"wss://{BASE_URL}/ws/fetchinbox?zone=public&query=%s"
+MAILINATOR_GET_EMAIL_URL = f"https://{BASE_URL}/fetch_email?msgid=%s&zone=public"
+MAILINATOR_WEB_URL = f"https://{BASE_URL}/v3/index.jsp?zone=public&query=%s"
+TIMEOUT = 2  # seconds
+
+
+def _generate_random_id():
+    return "".join(
+        random.choice(string.ascii_lowercase + string.digits) for _ in range(ID_LEN)
+    )
+
+
+class Link:
+    """Link."""
+
+    def __init__(self, link: str = "", text: str = ""):
+        """Basic html link abstraction.
+
+        :param link: Address of the link.
+        :type link: str
+        :param text: Text of the <a> tag
+        :type text: str
+        """
+        self.link = self.url = link
+        self.text = text
+
+    def __repr__(self):
+        return f'<a href="{self.link}">{self.text}</a>'
+
+
+class Email:
+    """Email."""
+
+    def __init__(self, msgid: str, jsessionid: str = None):
+        """Email object
+
+        :param msgid: message id that can be obtained from the Inbox object's email_info_list['id']
+        :type msgid: str
+        :param jsessionid: Optional. Jsessionid cookie.
+        :type jsessionid: str
+        """
+        self.msgid = msgid
+        self.jsessionid = jsessionid if jsessionid else _generate_random_id()
+        obj = self._fetch_email()["data"]
+        self.json = obj
+        self.from_address = obj.get("fromfull")
+        self.from_name = obj.get("from")
+        self.username = self.to = obj.get("to")
+        self.time = int(obj.get("time"))
+        self.headers = obj.get("headers")
+        self.subject = obj.get("subject")
+        self.ip = obj.get("ip")
+        self.seconds_ago = int(obj.get("seconds_ago"))
+
+        self.links = [Link(**link) for link in obj.get("clickablelinks") or []]
+        self.html = None
+        self.text = None
+        for part in obj.get("parts") or []:
+            if "text/html" in part["headers"]["content-type"]:
+                self.html = part.get("body")
+            if "text/plain" in part["headers"]["content-type"]:
+                self.text = part.get("body")
+
+    def _fetch_email(self):
+        """Fetches email data from mailinator"""
+        request = requests.get(MAILINATOR_GET_EMAIL_URL % self.msgid)
+        return request.json()
+
+    async def _remove_message(self):
+        """Coroutine to remove email from inbox"""
+        remove_msg = {"id": self.msgid, "cmd": "trash", "zone": "public"}
+        async with websockets.connect(
+            MAILINATOR_WSS_URL % self.username,
+            extra_headers={"Cookie": f"JSESSIONID={self.jsessionid}"},
+        ) as ws:
+            while True:
+                try:
+                    await asyncio.wait_for(ws.recv(), timeout=TIMEOUT)
+                except asyncio.TimeoutError:
+                    break
+            # The message has to be minified
+            await ws.send(json.dumps(remove_msg).replace(" ", ""))
+            try:
+                msg = await asyncio.wait_for(ws.recv(), timeout=TIMEOUT)
+            except asyncio.TimeoutError:
+                return
+            try:
+                obj = json.loads(msg)
+                if obj["channel"] in ["error", "status"]:
+                    return obj
+            except json.JSONDecodeError:
+                return
+
+    def remove(self) -> (bool, str):
+        """Removes email from inbox
+
+        :return: True if successful, False if not and message
+        :rtype: (bool, str)
+        """
+        response = asyncio.run(self._remove_message())
+        if response is None:
+            msg = "Failed to remove email"
+        else:
+            msg = response.get("msg") or "Failed to remove email"
+        return ("message deleted" in msg, msg)
+
+
+class PublicInbox:
+    """Creates a new public mailbox instance and fetch its emails"""
+
+    def __init__(self, username: str, message_fetch_timeout: float = TIMEOUT):
+        """Creates a new public mailbox instance and fetch its emails
+
+        :param username: The username of the public mailbox
+        :type username: str
+        :param message_fetch_timeout:
+        :type message_fetch_timeout: The timeout for fetching individual messages. You may want to increase this if you have a slow internet connection. If you call this too many times and have a good connection you can try to decrease this. Defaults to 2 seconds
+        """
+        self.username = username
+        self.address = f"{username}@{BASE_URL}"
+        self.web_url = MAILINATOR_WEB_URL % self.username
+        self.jsessionid = f"node01{_generate_random_id()}.node0"
+        self.email_info_list = []
+        self.fetch_emails()
+
+    async def _get_messages(self):
+        """Coroutine that fetches the messages from the public mailbox using the websockets stream"""
+        received = []
+        async with websockets.connect(
+            MAILINATOR_WSS_URL % self.username,
+            extra_headers={"Cookie": f"JSESSIONID={self.jsessionid}"},
+        ) as ws:
+            while True:
+                try:
+                    msg = await asyncio.wait_for(ws.recv(), timeout=TIMEOUT)
+                except asyncio.TimeoutError:
+                    break
+                try:
+                    obj = json.loads(msg)
+                    if "fromfull" in obj:
+                        received.append(obj)
+                except json.JSONDecodeError:
+                    continue
+        return received
+
+    def fetch_emails(self) -> list:
+        """Updates and Populates the email_info_list attribute with a list of the emails from the public mailbox
+
+        :return: A list of the emails information dicts from the public mailbox. These do not contain the actual body of the emails.
+        :rtype: list
+        """
+        self.email_info_list = asyncio.run(self._get_messages())
+        return self.email_info_list
+
+    def __iter__(self):
+        """Iterates over the emails in the public mailbox"""
+        for email in self.email_info_list:
+            yield Email(email["id"], self.jsessionid)
+
+    def get_lastest_email(self) -> Email:
+        """Returns the last email in the public mailbox
+
+        :return: The last email in the public mailbox
+        :rtype: Email
+        """
+        if self.email_info_list:
+            # TODO: This is a bit of a hack. We should probably just fetch the emails and sort them by time
+            # but it works
+            return Email(self.email_info_list[-1]["id"], self.jsessionid)

mailinatorapi/__init__.py

@@ -0,0 +1,3 @@
+from .mailinator import Email, Link, PublicInbox
+
+__all__ = ("PublicInbox", "Link", "Email")