# `7MMM. ,MMF'`7MMM. ,MMF' `7MMM. ,MMF' `7MM # MMMb dPMM MMMb dPMM MMMb dPMM MM # M YM ,M MM M YM ,M MM M YM ,M MM ,pW"Wq. ,M""bMM ,pP"Ybd # M Mb M' MM M Mb M' MM M Mb M' MM 6W' `Wb ,AP MM 8I `" # M YM.P' MM M YM.P' MM mmmmm M YM.P' MM 8M M8 8MI MM `YMMMa. # M `YM' MM M `YM' MM M `YM' MM YA. ,A9 `Mb MM L. I8 # .JML. `' .JMML..JML. `' .JMML. .JML. `' .JMML.`Ybmd9' `Wbmd"MML.M9mmmP' # # (c) 2023 — licensed under Apache 2.0 — https://www.apache.org/licenses/LICENSE-2.0 # meta pic: https://img.icons8.com/stickers/344/block.png # meta developer: @mm_mods __version__ = "1.0.0" from hikka import loader, utils from telethon.tl.patched import Message import logging from datetime import datetime, timedelta logger = logging.getLogger(__name__) # noinspection PyCallingNonCallable @loader.tds class ASAPMod(loader.Module): """Advanced Sending Automatisation Program""" strings = { 'name': 'ASAP', 'args?!': '🟥 Arguments needed!', 'incorrecttime': '🟥 Incorrect time!\n{example}', 'taskslist': '📄 Tasks list:\n\n', 'task': 'Task {x}: send {text} to {address} ({at_time}).\n', 'tasklistempty': '🟨 Tasks list is empty!', 'taskadded': '🟩 Task added!\nNext execution: {at_time}, task number: {x}', 'taskremoved': '🟩 Task removed!', 'tasknotfound': '🟥 Task not found!', 'tasklistcleared': '🟩 Tasks list cleared!', 'taskpaused': '🟩 Task paused!', 'allpaused': '🟩 All tasks paused!', 'taskresumed': '🟩 Task resumed!', 'taskpretimeexecuted': '🟩 Task executed!\nNext execution: {at_time}', 'taskreplanned': '🟩 Task replanned!\nNext execution: {at_time}', 'defaultintervalset': '🟩 Default interval set!\n{example}', 'incorrectinterval': '🟥 Incorrect interval!', 'timeformatexample': '🟦 Time format example: 2h — 2 hours, ' '1d 2h 30m — 1 day, 2 hours and 30 minutes, ' '20s — 20 seconds.', 'tzset': '🟩 Timezone set!', 'incorrecttz': '🟥 Incorrect timezone!', 'safetyreasonerror': '🟨 For safety reasons, you can\'t set such short interval!', } strings_ru = { 'name': 'ASAP', '_cls_doc': 'Продвинутая программа автоматизации отправки сообщений', 'args?!': '🟥 Нужны аргументы!', 'incorrecttime': '🟥 Неверное время!\n{example}', 'taskslist': '📄 Список задач:\n\n', 'task': 'Задача {x}: отправить {text} на {address} ({at_time}).' '\n', 'tasklistempty': '🟨 Список задач пуст!', 'taskadded': '🟩 Задача добавлена!\nСледующее выполнение: {at_time}, номер задачи:

{'
                     'x}

', 'taskremoved': '🟩 Задача удалена!', 'tasknotfound': '🟩 Задача не найдена!', 'tasklistcleared': '🟩 Список задач очищен!', 'taskpaused': '🟩 Задача приостановлена!', 'allpaused': '🟩 Все задачи приостановлены!', 'taskresumed': '🟩 Задача возобновлена!', 'taskpretimeexecuted': '🟩 Задача выполнена!\nСледующее выполнение: {at_time}', 'taskreplanned': '🟩 Задача перепланирована!\nСледующее выполнение: {at_time}', 'defaultintervalset': '🟩 Интервал по умолчанию установлен!', 'incorrectinterval': '🟥 Неверный интервал!\n{example}', 'timeformatexample': '🟦 Пример формата времени: 2h — 2 часа, ' '1d 2h 30m — 1 день, 2 часа и 30 минут, ' '20s — 20 секунд.', 'tzset': '🟩 Часовой пояс установлен!', 'incorrecttz': '🟥 Неверный часовой пояс!', 'safetyreasonerror': '🟨 По соображениям безопасности вы не можете установить такой короткий интервал!', } strings_de = { 'name': 'ASAP', '_cls_doc': 'Fortgeschrittenes Programm zur Automatisierung des Nachrichtenversands', 'args?!': '🟥 Argumente benötigt!', 'incorrecttime': '🟥 Falsche Zeit!\n{example}', 'taskslist': '📄 Aufgabenliste:\n\n', 'task': 'Aufgabe {x}: senden {text} zu {address} ({at_time}).\n', 'tasklistempty': '🟨 Aufgabenliste ist leer!', 'taskadded': '🟩 Aufgabe hinzugefügt!\nNächste Ausführung: {at_time}, ' 'Aufgabennummer: {x}', 'taskremoved': '🟩 Aufgabe entfernt!', 'tasknotfound': '🟩 Aufgabe nicht gefunden!', 'tasklistcleared': '🟩 Aufgabenliste gelöscht!', 'taskpaused': '🟩 Aufgabe pausiert!', 'allpaused': '🟩 Alle Aufgaben pausiert!', 'taskresumed': '🟩 Aufgabe fortgesetzt!', 'taskpretimeexecuted': '🟩 Aufgabe ausgeführt!\nNächste Ausführung: {at_time}', 'taskreplanned': '🟩 Aufgabe neu geplant!\nNächste Ausführung: {at_time}', 'defaultintervalset': '🟩 Standardintervall gesetzt!', 'incorrectinterval': '🟥 Falsches Intervall!\n{example}', 'timeformatexample': '🟦 Beispiel für das Zeitformat: 2h — 2 Stunden, ' '1d 2h 30m — 1 Tag, 2 Stunden und 30 Minuten, ' '20s — 20 Sekunden.', 'tzset': '🟩 Zeitzone gesetzt!', 'incorrecttz': '🟥 Falsche Zeitzone!', 'safetyreasonerror': '🟨 Aus Sicherheitsgründen können Sie kein so kurzes Intervall festlegen!', } async def client_ready(self): if not self.get('default_interval'): self.set('default_interval', '2h') if not self.get('tasks'): self.set('tasks', {}) if not self.get('tz'): self.set('tz', 0) @loader.loop(interval=1, autostart=True) async def _run_tasks(self): tasks = self.get('tasks') if not tasks: return for k, task in tasks.items(): if datetime.strptime(task['next_execution'], '%d.%m.%Y %H:%M:%S') <= datetime.now(): await self._run_task(task, k) @staticmethod def validate_time(time: str) -> bool: """ Validates the time. :param time: time to validate. :return: if time is valid or not. """ if not time: return False args = time.split() for arg in args: if not arg[:-1].isdigit(): return False if arg[-1] not in ['s', 'm', 'h', 'd']: return False return True @staticmethod def convert_time(time: str) -> timedelta: """ Converts time to timedelta. :param time: time to convert. :return: timedelta. """ args = time.split() res = timedelta() for arg in args: try: if arg[-1] == 's': res += timedelta(seconds=int(arg[:-1])) elif arg[-1] == 'm': res += timedelta(minutes=int(arg[:-1])) elif arg[-1] == 'h': res += timedelta(hours=int(arg[:-1])) elif arg[-1] == 'd': res += timedelta(days=int(arg[:-1])) except ValueError: pass return res @loader.command( 'ataskadd', ru_doc='Добавить задачу в список задач.\nИспользование: .ataskadd <текст>\n[время]\n[адресат]', de_doc='Fügen Sie eine Aufgabe zur Aufgabenliste hinzu.\nVerwendung: .ataskadd \n[Zeit]\n[Empfänger]', ) async def ataskaddcmd(self, m: Message): """Add a task to the list of tasks. Usage: .ataskadd [time] [target]""" args = utils.get_args_raw(m) if not args: return await utils.answer(m, self.strings('args?!')) args = args.split('\n') if len(args) == 2: interval = args[1] address = m.chat_id elif len(args) == 3: interval = args[1] address = args[2] else: interval = self.get('default_interval') address = m.chat_id if not self.validate_time(interval): return await utils.answer(m, self.strings('incorrecttime')) if self.convert_time(interval) < timedelta(seconds=10): return await utils.answer(m, self.strings('safetyreasonerror')) task = { 'text': args[0], 'interval': interval, 'address': address, 'paused': False, 'next_execution': (datetime.now() + self.convert_time(interval)).strftime('%d.%m.%Y %H:%M:%S'), } tasks = self.get('tasks') x = len(tasks) + 1 tasks[str(x)] = task self.set('tasks', tasks) await utils.answer( m, self.strings('taskadded').format( at_time=task['next_execution'], x=x ) ) @loader.command( 'ataskremove', ru_doc='Удалить задачу из списка задач.\nИспользование: .ataskremove <номер>', de_doc='Entfernen Sie eine Aufgabe aus der Aufgabenliste.\nVerwendung: .ataskremove ', ) async def ataskremovecmd(self, m: Message): """Remove a task from the list of tasks. Usage: .ataskremove """ args = utils.get_args_raw(m) if not args: return await utils.answer(m, self.strings('args?!')) try: x = int(args) except ValueError: return await utils.answer(m, self.strings('args?!')) x = str(x) tasks = self.get('tasks') if x not in tasks: return await utils.answer(m, self.strings('tasknotfound')) del tasks[x] self.set('tasks', tasks) await utils.answer(m, self.strings('taskremoved')) @loader.command( 'atasklist', ru_doc='Показать список задач.', de_doc='Zeigt die Liste der Aufgaben an.', ) async def atasklistcmd(self, m: Message): """Show the list of tasks.""" tasks = self.get('tasks') if not tasks: return await utils.answer(m, self.strings('tasklistempty')) res = self.strings('taskslist') for k, task in tasks.items(): res += self.strings('task').format( x=k if not task['paused'] else f'~~{k}~~

',
                text=task['text'],
                address=task['address'],
                at_time=datetime.strptime(task['next_execution'], '%d.%m.%Y %H:%M:%S') + timedelta(hours=self.get('tz'))
            )

        await utils.answer(m, res)

    @loader.command(
        'ataskclear',
        ru_doc='Очистить список задач.',
        de_doc='Löschen Sie die Liste der Aufgaben.',
    )
    async def ataskclearcmd(self, m: Message):
        """Clear the list of tasks."""
        self.set('tasks', {})
        await utils.answer(m, self.strings('tasklistcleared'))

    @loader.command(
        'ataskpause',
        ru_doc='Приостановить задачу.\nИспользование: .ataskpause <номер>\nБез аргументов — приостановить все задачи.',
        de_doc='Aufgabe pausieren.\nVerwendung: .ataskpause \nOhne Argumente - alle Aufgaben pausieren.',
    )
    async def ataskpausecmd(self, m: Message):
        """Pause a task.
        Usage: .ataskpause 
        If no arguments — pause all tasks."""
        args = utils.get_args_raw(m)

        if not args:
            tasks = self.get('tasks')

            for k, task in tasks.items():
                task['paused'] = True
                tasks[k] = task

            self.set('tasks', tasks)

            return await utils.answer(m, self.strings('taskpaused'))

        try:
            x = int(args)
        except ValueError:
            return await utils.answer(m, self.strings('args?!'))

        x = str(x)

        tasks = self.get('tasks')

        if x not in tasks:
            return await utils.answer(m, self.strings('tasknotfound'))

        tasks[x]['paused'] = True
        self.set('tasks', tasks)

        await utils.answer(m, self.strings('taskpaused'))

    @loader.command(
        'ataskresume',
        ru_doc='Возобновить задачу.\nИспользование: .ataskresume <номер>',
        de_doc='Aufgabe fortsetzen.\nVerwendung: .ataskresume ',
    )
    async def ataskresumecmd(self, m: Message):
        """Resume a task.
        Usage: .ataskresume """
        args = utils.get_args_raw(m)

        if not args:
            return await utils.answer(m, self.strings('args?!'))

        try:
            x = int(args)
        except ValueError:
            return await utils.answer(m, self.strings('args?!'))

        tasks = self.get('tasks')

        if x > len(tasks):
            return await utils.answer(m, self.strings('tasknotfound'))

        tasks[x - 1]['paused'] = False
        self.set('tasks', tasks)

        await utils.answer(m, self.strings('taskresumed'))

    @loader.command(
        'ataskexec',
        ru_doc='Выполнить задачу сейчас и перепланировать согласно интервалу.\nИспользование: .ataskexec <номер>',
        de_doc='Führen Sie die Aufgabe jetzt aus und planen Sie sie entsprechend dem Intervall neu.\nVerwendung: '
               '.ataskexec ',
    )
    async def ataskexeccmd(self, m: Message):
        """Execute a task right now and replan according to the interval.
        Usage: .ataskexec """
        args = utils.get_args_raw(m)

        if not args:
            return await utils.answer(m, self.strings('args?!'))

        try:
            x = int(args)
        except ValueError:
            return await utils.answer(m, self.strings('args?!'))

        x = str(x)

        tasks = self.get('tasks')

        if x not in tasks:
            return await utils.answer(m, self.strings('tasknotfound'))

        task = tasks[x]

        if task['paused']:
            return await utils.answer(m, self.strings('taskpaused'))

        task['next_execution'] = datetime.now() + self.convert_time(task['interval'])
        self.set('tasks', tasks)

        await utils.answer(
            m, self.strings('taskpretimeexecuted').format(
                at_time=datetime.strptime(task['next_execution'], '%d.%m.%Y %H:%M:%S') + timedelta(hours=self.get('tz'))
            )
        )

        await self._run_task(task, x)

    @loader.command(
        'ataskreplan',
        ru_doc='Изменить интервал задачи.\nИспользование: .ataskreplan <номер> <время>',
        de_doc='Ändern Sie das Intervall der Aufgabe.\nVerwendung: .ataskreplan  ',
    )
    async def ataskreplancmd(self, m: Message):
        """Change the interval of the task.
        Usage: .ataskreplan """
        args = utils.get_args_raw(m)

        if not args:
            return await utils.answer(m, self.strings('args?!'))

        args = args.split(maxsplit=1)
        x, time = args[0], args[1]

        tasks = self.get('tasks')

        if x not in tasks:
            return await utils.answer(m, self.strings('tasknotfound'))

        task = tasks[x]

        if not self.validate_time(time):
            return await utils.answer(m, self.strings('incorrecttime'))

        task['interval'] = time

        if task['paused']:
            return await utils.answer(m, self.strings('taskpaused'))

        task['next_execution'] = (datetime.now() + self.convert_time(task['interval'])).strftime('%d.%m.%Y %H:%M:%S')

        tasks[x] = task
        self.set('tasks', tasks)

        await utils.answer(
            m, self.strings('taskreplanned').format(
                at_time=datetime.strptime(task['next_execution'], '%d.%m.%Y %H:%M:%S') + timedelta(hours=self.get('tz'))
            )
        )

    @loader.command(
        'adeftint',
        ru_doc='Установить интервал по умолчанию.\nИспользование: .adeftint <время>',
        de_doc='Setzen Sie das Standardintervall.\nVerwendung: .adeftint ',
    )
    async def adeftintcmd(self, m: Message):
        """Set the default interval.
        Usage: .ataskdeftint """
        args = utils.get_args_raw(m)

        if not args:
            return await utils.answer(m, self.strings('args?!'))

        if not self.validate_time(args):
            return await utils.answer(m, self.strings('incorrectinterval'))

        self.set('default_interval', args)
        await utils.answer(
            m, self.strings('defaultintervalset').format(
                example=self.strings('timeformatexample')
            )
        )

    @loader.command(
        'atz',
        ru_doc='Явно установить часовой пояс, если время отображается неверно.\nИспользование: .atz <часовой пояс>',
        de_doc='Legen Sie die Zeitzone explizit fest, wenn die Zeit falsch angezeigt wird.\nVerwendung: '
               '.atz ',
    )
    async def atzcmd(self, m: Message):
        """Set the timezone explicitly if time displayed wrong.
        Usage: .atz """
        args = utils.get_args_raw(m)

        if not args:
            return await utils.answer(m, self.strings('args?!'))

        try:
            tz = int(args)
        except ValueError:
            try:
                tz = float(args)
            except ValueError:
                return await utils.answer(m, self.strings('incorrecttz'))

        if tz < -12 or tz > 12:
            return await utils.answer(m, self.strings('incorrecttz'))

        self.set('tz', tz)
        await utils.answer(m, self.strings('tzset'))

    async def _run_task(self, task: dict, index: str):
        if task['paused']:
            return

        await self.client.send_message(task['address'], task['text'])

        tasks = self.get('tasks')
        this = tasks[index]

        this['next_execution'] = (datetime.now() + self.convert_time(this['interval'])).strftime('%d.%m.%Y %H:%M:%S')

        tasks[index] = this
        self.set('tasks', tasks)