Command Line Tool for Posting to Termbin

pycat.py

#!/usr/bin/env python3.10
import asyncio
import os.path
import sys
from asyncio import StreamReader, StreamWriter


class PyCat:
    def __init__(self, host: str, port: int, _loop: asyncio.AbstractEventLoop):
        self._loop = _loop
        self.reader: StreamReader
        self.writer: StreamWriter
        self.host = host
        self.port = port

    async def __ainit__(self):
        try:
            self.reader, self.writer = await asyncio.open_connection(self.host, self.port)
        except asyncio.exceptions.TimeoutError as err:
            print(f"[!] Error: {err}")

    async def write(self, data: bytes) -> None:
        self.writer.write(data)

    async def read(self) -> bytes:
        return await self.reader.read()

    async def run(self, data: (str, bytes)) -> str:
        assert type(data) in [bytes, str]
        if type(data) is not bytes:
            data = data.encode()
        await self.write(data)
        ret = await self.read()
        return ret.decode()

    def read_file(self, filename) -> bytes:
        with open(filename, 'rb') as f:
            return f.read()

    async def pycat(self, data: any = None) -> str:
        """
        Simple send and read. Example: post to termbin.
        :param data: data or file path
        :return: reply
        """
        await self.__ainit__()
        if os.path.exists(data.strip("\r\n")):
            data = self.read_file(data)
            return await self.run(data)
        else:
            if len(data):
                return await self.run(data)


async def main(_loop: asyncio.AbstractEventLoop) -> str:
    _input = None
    try:
        _input = sys.argv[1]
    except IndexError:
        _input = sys.stdin.read()
    finally:
        if _input:
            pycat = PyCat('termbin.com', 9999, _loop=_loop)
            return await pycat.pycat(_input)


if __name__ == '__main__':
    loop = asyncio.new_event_loop()
    print(loop.run_until_complete(main(loop)))