MeowTg - A userbot software for telegram with plugins system.
class API:
    client = None
    logger = None
    plugins_loader = None

    me = None

    def __init__(self, client, logger, me):
        self.client = client
        self.logger = logger
        self.me = me

    def register_plugins_loader(self, plugins_loader):
        self.plugins_loader = plugins_loader

    def get_my_id(self):
        return self.me.id
meowtg/api.py
kotleni/meowtg
import json, os

class Config:
    path = 'config.json'
    pairs = {}

    def load(self):
        if not os.path.isfile(self.path):
            file = open(self.path, 'w')
            file.write("{}")
            file.close()

        file = open(self.path, 'r')
        self.pairs = json.loads(file.read())
        file.close()

    def save(self):
        json = json.dumps(self.pairs)

        file = open(self.path, 'w')
        file.write(json)
        file.close()

    def get(self, key):
        return self.pairs[key]

    def set(self, key, value):
        self.pairs[key] = value

meowtg/config.py
kotleni/meowtg
LOGGER_FORMATTER = '(%(levelname)s | %(asctime)s) -> %(message)s'
meowtg/const.py
kotleni/meowtg
import logging
from const import *

class CustomFormatter(logging.Formatter):
    grey = "\x1b[38;20m"
    yellow = "\x1b[33;20m"
    red = "\x1b[31;20m"
    bold_red = "\x1b[31;1m"
    reset = "\x1b[0m"

    FORMATS = {
        logging.DEBUG: grey + LOGGER_FORMATTER + reset,
        logging.INFO: grey + LOGGER_FORMATTER + reset,
        logging.WARNING: yellow + LOGGER_FORMATTER + reset,
        logging.ERROR: red + LOGGER_FORMATTER + reset,
        logging.CRITICAL: bold_red + LOGGER_FORMATTER + reset
    }

    def format(self, record):
        log_fmt = self.FORMATS.get(record.levelno)
        formatter = logging.Formatter(log_fmt, datefmt='%H:%M:%S')
        return formatter.format(record)
meowtg/custom_formatter.py
kotleni/meowtg
from config import Config
from wizard import Wizard
from api import API
from plugins_loader import PluginsLoader

from telethon import TelegramClient, events, types
from telethon.tl.functions.messages import SendReactionRequest
import logging, sys, os, datetime
from custom_formatter import CustomFormatter

from result_codes import *
from const import *

api = None

config = Config()
config.load()

wizard = Wizard()
if wizard.is_first_run():
    wizard.setup(config)
    config.save()

client = TelegramClient('tg', config.get('app_id'), config.get('app_hash'))
plugins_loader = None

logger = logging.getLogger('meowtg')
logger.setLevel(logging.INFO)
formatter = CustomFormatter()

stdout_handler = logging.StreamHandler(sys.stdout)
stdout_handler.setLevel(logging.DEBUG)
stdout_handler.setFormatter(formatter)

logger.addHandler(stdout_handler)

@client.on(events.NewMessage)
async def log_message(event):
    chat = await event.get_chat()
    sender = await event.get_sender()
    message = event.message
    message_text = event.text

    log_message = ""
    if isinstance(sender, types.Channel):
        return

    if isinstance(sender, types.User):
        log_message = f"{sender.first_name} ({sender.username}, {sender.id}) in {chat.username}: {message_text}"

    logger.info(log_message)
    
    prefix = "."
    if message_text.startswith(prefix) and len(message_text) > 3 and sender.id == api.get_my_id():
        command = message_text[len(prefix):]
        args = command.split(' ')
        result = await plugins_loader.on_command(event, args)

        if result == COMMAND_OK_MESSAGE_REMOVE:
            await client.delete_messages(chat, [message])
            return

        if result == None:
            result = "Unknown command"
        await client.edit_message(chat, message, f'{message_text}\n-------\n```{result}```')
    else:
        await plugins_loader.on_event(event)

async def main():
    global plugins_loader, api

    logger.info("MeowTG is starting...")

    await client.start()
    me = await client.get_me()
    api = API(client, logger, me)
    plugins_loader = PluginsLoader(api)
    api.register_plugins_loader(plugins_loader)

    logger.info("Loading plugins...")
    await plugins_loader.load_plugins()
    await client.run_until_disconnected()

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

meowtg/main.py
kotleni/meowtg
class PluginBase:
    api = None

    description = ""
    enabled = False

    def __init__(self, api) -> None:
        self.api = api

    async def load(self):
        pass

    async def on_event(self, event):
        pass

    async def on_command(self, event, args) -> str:
        return None
meowtg/plugin_base.py
kotleni/meowtg
import sys
import os
import importlib.util
from plugin_base import PluginBase

class PluginsLoader:
    api = None
    folder_path = "plugins"
    plugins = []

    def __init__(self, api):
        self.api = api

    def get_plugins_files(self):
        return [file for file in os.listdir(self.folder_path) if file.endswith('.py')]

    async def load_plugin(self, path):
        module_name = path.replace('.py', '')
        module_path = f'{self.folder_path}/{module_name}.py'

        try:
            # Use the built-in importlib library to dynamically import the plugin module
            spec = importlib.util.spec_from_file_location(module_name, module_path)
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)

            # Find and create an instance of the plugin class
            for name, obj in vars(module).items():
                if isinstance(obj, type) and issubclass(obj, PluginBase) and obj is not PluginBase:
                    instance = obj(self.api)
                    if instance.enabled:
                        await instance.load()
                        self.plugins.append(instance)

                        self.api.logger.info(f'Plugin {path} loaded.')
                    else:
                        self.api.logger.warn(f'Plugin {path} not loaded. (DISABLED)')
                    break
        except Exception as e:
            self.api.logger.error(f"Error loading {path} plugin: {str(e)}")

    async def load_plugins(self):
        files = self.get_plugins_files()
        for path in files:
            await self.load_plugin(path)

    async def on_event(self, event):
        for plugin in self.plugins:
            await plugin.on_event(event)

    async def on_command(self, event, args) -> str:
        for plugin in self.plugins:
            output = await plugin.on_command(event, args)
            if output != None:
                return output

        return None
meowtg/plugins_loader.py
kotleni/meowtg
COMMAND_OK_MESSAGE_REMOVE = "$OK_MESSAGE_REMOVED"
meowtg/result_codes.py
kotleni/meowtg
import os
from config import Config

class Wizard:
    def is_first_run(self):
        return not os.path.isfile('tg.session')

    def setup(self, config):
        print("-------")
        print("For first run you need configure bot.")
        print("To get telegram app id and hash visit: https://my.telegram.org")
        print("-------\n")

        app_id = input("app_id = ")
        app_hash = input("app_hash = ")

        config.set('app_id', app_id)
        config.set('app_hash', app_hash)
meowtg/wizard.py
kotleni/meowtg
from plugin_base import PluginBase
from telethon import TelegramClient, events, types
from telethon.tl.functions.messages import SendReactionRequest
import os

class AssetsGrab(PluginBase):
    description = "Automatically download all media from chats"
    enabled = False

    assets_path = 'assets/'

    SIZE_LIMIT_KB = 32 * 1024

    def __init__(self, api) -> None:
        super().__init__(api)
    
    async def load(self):
        if not os.path.isdir(self.assets_path):
            os.mkdir(self.assets_path)

    async def on_event(self, event):
        chat = await event.get_chat()
        sender = await event.get_sender()
        message = event.message
        message_text = event.text

        if isinstance(sender, types.Channel):
            return
        
        media = message.media

        if media is None:
            return
        
        if message.file is None:
            return
        
        size = message.file.size
        size_kb = size / 1024

        if size_kb > self.SIZE_LIMIT_KB:
            self.api.logger.warn(f'File size is {size_kb} KB (more than {self.SIZE_LIMIT_KB}), skipping')
            return
        
        # download image
        await self.api.client.download_media(message, self.assets_path)
plugins/assets_grab.py
kotleni/meowtg
from plugin_base import PluginBase
from telethon import TelegramClient, events, types
from telethon.tl.functions.messages import SendReactionRequest
from random import randint

class ReactTarget:
    id = None
    emoji = None

    def __init__(self, id, emoji) -> None:
        self.id = id
        self.emoji = emoji

class AutoReact(PluginBase):
    description = "Automatically react to messages from certain users (config in py code)"
    enabled = False

    react_targets = [ReactTarget(802764912, '💩'), ReactTarget(971778574, '🔥')]

    def __init__(self, api) -> None:
        super().__init__(api)
    
    async def load(self):
        pass

    async def on_event(self, event):
        chat = await event.get_chat()
        sender = await event.get_sender()
        message = event.message
        message_text = event.text

        if isinstance(sender, types.Channel):
            return
        
        for target in self.react_targets:
            if sender.id == target.id and randint(0, 100) < 5:
                await self.api.client(SendReactionRequest(
                    peer = chat,
                    msg_id = message.id,
                    reaction = [types.ReactionEmoji(
                        emoticon=target.emoji
                    )]
                ))
                break
plugins/auto_react.py
kotleni/meowtg
from plugin_base import PluginBase
from telethon import TelegramClient, events, types
from telethon.tl.functions.messages import SendReactionRequest
import requests
from result_codes import *

class CatPicApi:
    def get_random_cat_pic_url(self):
        return requests.get('https://api.thecatapi.com/v1/images/search').json()[0]['url']
    
    def get_random_cat_pic(self):
        return requests.get(self.get_random_cat_pic_url()).content

class CatPic(PluginBase):
    description = "Send random cat pic on .cat command"
    enabled = True
    
    cat_pic_api = CatPicApi()

    def __init__(self, api) -> None:
        super().__init__(api)
    
    async def load(self):
        pass

    async def on_command(self, event, args) -> str:
        chat = await event.get_chat()
        message = event.message

        if args[0] == "cat":
            try:
                catpic = self.cat_pic_api.get_random_cat_pic()
                # await self.api.client.delete_messages(chat, [message])
                await self.api.client.send_file(chat, catpic)
                return COMMAND_OK_MESSAGE_REMOVE
            except Exception as e:
                return "Exception: " + str(e)
plugins/cat_pic.py
kotleni/meowtg
from plugin_base import PluginBase
from telethon import TelegramClient, events, types
from telethon.tl.functions.messages import SendReactionRequest
import subprocess

class CommandLineExecutor:
    def execute(self, command):
        return subprocess.run(command, shell=True, capture_output=True)
    
class Cmd(PluginBase):
    description = "Execute shell commands on .cmd command"
    enabled = True
    
    executor = CommandLineExecutor()

    def __init__(self, api) -> None:
        super().__init__(api)
        
    async def on_command(self, event, args) -> str:
        if args[0] == "cmd" or args[0] == "shell" or args[0] == "exec":
            output = ""
            try:
                output = output + str(self.executor.execute(args[1]).stdout.decode('utf-8'))
                output = output + str(self.executor.execute(args[1]).stderr.decode('utf-8'))
            except Exception as e:
                output = e
            return output
plugins/cmd.py
kotleni/meowtg
from plugin_base import PluginBase
from telethon import TelegramClient, events, types
from telethon.tl.functions.messages import SendReactionRequest

class Eval(PluginBase):
    description = "Evaluate python code on .eval command"
    enabled = True
    
    def __init__(self, api) -> None:
        super().__init__(api)
    
    async def load(self):
        pass

    async def on_command(self, event, args) -> str:
        if args[0] == "eval":
            result = ''
            try:
                message_text = event.text
                code = message_text[6:]
                result = eval(args[1])
            except Exception as e:
                result = e
            return result
plugins/eval.py
kotleni/meowtg
from plugin_base import PluginBase
from telethon import TelegramClient, events, types
from telethon.tl.functions.messages import SendReactionRequest
import logging, os, datetime
from const import *

class FilesLogs(PluginBase):
    description = "Enable logging to files"
    enabled = False

    logs_folder = "logs"

    async def load(self):
        formatter = logging.Formatter(LOGGER_FORMATTER)

        current_date = datetime.datetime.now().strftime("%Y-%m-%d")
        log_file_path = os.path.join(self.logs_folder, f"{current_date}.log")
        if not os.path.isdir(self.logs_folder):
            os.mkdir(self.logs_folder)
        file_handler = logging.FileHandler(log_file_path, encoding='utf-8')
        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(formatter)

        self.api.logger.addHandler(file_handler)
plugins/files_logs.py
kotleni/meowtg
from plugin_base import PluginBase
from telethon import TelegramClient, events, types
from telethon.tl.functions.messages import SendReactionRequest

class Halt(PluginBase):
    description = "Halt bot on .halt command"
    enabled = True
    
    def __init__(self, api) -> None:
        super().__init__(api)
    
    async def load(self):
        pass

    async def on_command(self, event, args) -> str:
        if args[0] == "halt":
            await self.api.client.disconnect()
            exit(0)
plugins/halt.py
kotleni/meowtg
from plugin_base import PluginBase
from telethon import TelegramClient, events, types
from telethon.tl.functions.messages import SendReactionRequest

class Help(PluginBase):
    description = "Help commands: .plugins"
    enabled = True
    
    def __init__(self, api) -> None:
        super().__init__(api)
    
    async def load(self):
        pass

    async def on_command(self, event, args) -> str:
        if args[0] == "plugins":
            output = ""
            for plugin in self.api.plugins_loader.plugins:
                class_name = plugin.__class__.__name__
                output += f'{class_name} - {plugin.description}\n'
            return output
plugins/help.py
kotleni/meowtg
from plugin_base import PluginBase
from telethon import TelegramClient, events, types
from telethon.tl.functions.messages import SendReactionRequest

class MediaTimerSave(PluginBase):
    description = "Save and post to saved messages all media with timer"
    enabled = True
    
    def __init__(self, api) -> None:
        super().__init__(api)
    
    async def load(self):
        pass

    async def on_event(self, event):
        chat = await event.get_chat()
        sender = await event.get_sender()
        message = event.message
        message_text = event.text

        if isinstance(sender, types.Channel):
            return
        
        # check is image
        media = message.media
        if media is None:
            return
        if isinstance(media, types.MessageMediaWebPage):
            return
        # if not isinstance(media, types.MessageMediaPhoto):
        #     return
        if media.ttl_seconds is None:
            return
        
        # downlaod image
        path = await self.api.client.download_media(message, 'assets')
        
        # send pic to me
        await self.api.client.send_file(self.api.get_my_id(), path, caption=f'Saved from media with timer.\n{sender.first_name} (@{sender.username} {sender.id}) - {message_text}')
plugins/media_timer_save.py
kotleni/meowtg