A set of tools to improve Telegram experience. 
# this bot is inteded to transcribe all voice and video messages sent by user, using telethon
import json
import os
import asyncio
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv('API_ID')
api_hash = os.getenv('API_HASH')


import logging

logging.basicConfig(level=logging.INFO)

from commands import IngTranscribeCommand, IngGPTCommand

from control import ClientHandler, ClientFactory
import typer

app = typer.Typer()

if not os.path.exists('data/clients.json'):
    with open('data/clients.json', 'w') as f:
        json.dump({}, f)

with open('data/clients.json', 'r') as f:
    client_data = json.load(f)


@app.command()
def add_client(session: str):
    """
    Add a client to the client data based on user input for a session name. The commands will be standard.

    :param session: str The session name for the new client.
    :return: None
    """
    new_client = {
            "session_name": session,
            "commands": ["IngTranscribeCommand", "IngGPTCommand"]
        }
    client_data.append(new_client)
    # save json
    with open('data/clients.json', 'w') as f:
        json.dump(client_data, f, indent=4)

    print(f"Added new client with session name: {session}")

@app.command()
def delete_client():
    print("Here are the available clients:")
    for index, client in enumerate(client_data, start=1):
        print(f"{index}. {client['session_name']}")

    session = typer.prompt("Please enter the session name of the client you want to delete")
    client_data[:] = [client for client in client_data if client.get("session_name") != session]
    print(client_data)

    with open('data/clients.json', 'w') as f:
        json.dump(client_data, f, indent=4)
    print(f"Deleted client with session name: {session}")


@app.command()
def start_program():
    """
    Starts the main program after updating the client_data with added and/or deleted clients
    :return: None
    """
    loop = asyncio.get_event_loop()

    try:
        # Schedule the main coroutine and the do_nothing coroutine
        asyncio.ensure_future(main())
        asyncio.ensure_future(do_nothing())

        loop.run_forever()
    except KeyboardInterrupt:
        loop.stop()


async def main():
    client_handlers = {}
    for config in client_data:
        client = ClientFactory.create_client(config)
        await client.start()
    print("bot started")


async def do_nothing():
    while True:
        await asyncio.sleep(1)


if __name__ == "__main__":
    app()


app.py
44-5-53-6-k/ingenTelegam_ing
# commands/__init__.py

from commands.ing_transcribe import IngTranscribeCommand
from .ing_gpt import IngGPTCommand


commands/__init__.py
44-5-53-6-k/ingenTelegam_ing
from abc import ABC, abstractmethod

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from control import ClientHandler

class Command (ABC):
    command_name = ""
    aliases = []

    def __init__(self, event, client_handler: 'ClientHandler'):
        self.event = event
        self.client_handler = client_handler

    @abstractmethod
    async def execute(self):
        pass

    async def parse_args(self, args):
        pass

    @classmethod
    def get_identifiers(cls):
        return [cls.command_name] + cls.aliases

# class Command:
#     command_name = ""
#     aliases = []
#
#     def __init__(self, event, client):
#         self.event = event
#         self.client = client
#
#     async def execute(self):
#         pass
#
#     async def parse_args(self, args):
#         pass
#
#     @classmethod
#     def register_with_manager(cls, manager):
#         manager.register_command(cls.command_name, cls, cls.aliases)
commands/base.py
44-5-53-6-k/ingenTelegam_ing
from .base import Command


class IngGPTCommand(Command):
    command_name = "@ingGPT"
    aliases = ["@gpt"]

    async def execute(self):
        # Implementation
        ...

commands/ing_gpt.py
44-5-53-6-k/ingenTelegam_ing
from .ing_transcribe import IngTranscribeCommand
commands/ing_transcribe/__init__.py
44-5-53-6-k/ingenTelegam_ing
import logging

from commands.base import Command
import datetime
from telethon.tl.types import Message
from job_manager.voice_job import VoiceJob


class IngTranscribeCommand(Command):
    """
    This command transcribes a voice message or a reply to a message.
    """
    command_name = "@ingTranscribe"
    aliases = ["@"]

    def __init__(self, event, client_handler):
        super().__init__(event, client_handler)
        self.peer_id = self.event.message.peer_id

        self.format = "text"  # Default format
        self.summarize = False
        self.create_chapters = False
        self._edit_existing = True
        self._send_as_new = None
        self._prepend_message = True if not self.is_voice_message() else False

        self._message_to_prepend = f"[🐾](emoji/5460768917901285539) __{self.command_name}__\n\n"
        self._status_message = "[❤️](emoji/5321387857527447505) __Transcribing...__[✨](emoji/5278352839272309494)"

    @property
    def message_to_prepend(self):
        if self._prepend_message:
            return self._message_to_prepend
        else:
            return ""

    @property
    def edit_existing(self):
        return self._edit_existing if not self._send_as_new else False

    async def execute(self):
        if self.is_voice_message():
            await self.handle_voice_message()
        else:
            await self.handle_text_message()

    def is_voice_message(self):
        return hasattr(self.event.message, 'voice') and self.event.message.voice

    async def handle_voice_message(self):
        is_recent = lambda msg_date: (datetime.datetime.now(datetime.timezone.utc) - msg_date).total_seconds() <= 10

        if not is_recent(self.event.message.date):
            logging.info("Ignoring old message")
            return

        await self.client_handler.edit_message(self.peer_id, self.event.message.id,
                                               self._status_message)

        voice_job = VoiceJob(self.client_handler, self.event.message)
        result = await voice_job.process_job()
        plain_text = result.get_plain_text()

        await self.send_result(plain_text)

    def is_from_peer(self):
        return self.peer_id == self.event.message.from_id or self.event.message.from_id is None

    async def send_result(self, result):
        if self.is_voice_message():
            await self.client_handler.edit_message(self.peer_id, self.event.message.id,
                                                   self.message_to_prepend + result)
            return
        elif self.format == "text":
            if self.edit_existing and self.is_from_peer():
                await self.client_handler.edit_message(self.peer_id, self.event.message.reply_to_msg_id,
                                                       self.message_to_prepend + result)
            else:
                await self.client_handler.reply_message(self.peer_id, self.message_to_prepend + result, reply_to=self.event.message.reply_to_msg_id )
        elif self.format == "file":
            await self.client_handler.send_text_as_file(self.peer_id, result, "transcription.txt")
        elif self.format == "vtt":
            await self.client_handler.reply_message(self.peer_id, "VTT format is not supported yet, sending as file", reply_to=self.event.message.reply_to_msg_id)
            await self.client_handler.send_text_as_file(self.peer_id, result, "transcription.txt")

        await self.remove_command_message()

    async def set_status(self, status, message: Message, toReplace=False):
        if toReplace:
            await self.client_handler.edit_message(self.peer_id, self.event.message.id, status)
        else:
            await self.client_handler.edit_message(self.peer_id, self.event.message.id, status + "\n\n" + message.raw_text)

    async def remove_command_message(self):
        if self.is_voice_message():
            return
        await self.client_handler.delete_message(self.peer_id, self.event.message.id)
    async def handle_text_message(self):
        await self.set_status(self._status_message, self.event.message, toReplace=True)
        args = self.event.message.message.split()
        await self.parse_args(args)

        message_to_transcribe = await self.client_handler.get_message_by_id(self.peer_id,
                                                                            self.event.message.reply_to_msg_id)
        voice_job = VoiceJob(self.client_handler, message_to_transcribe)
        result = await voice_job.process_job()
        plain_text = result.get_plain_text()
        await self.send_result(plain_text)


    async def parse_args(self, args):

        for i, arg in enumerate(args):
            if arg in ("-f", "--format") and i + 1 < len(args):
                self.format = args[i + 1]
            elif arg in ("-s", "--summarize"):
                self.summarize = True
            elif arg in ("-c", "--chapters"):
                self.create_chapters = True
            elif arg in ("-n", "--new"):
                self._send_as_new = True
            elif arg in ("-e", "--existing"):
                self._edit_existing = True

commands/ing_transcribe/ing_transcribe.py
44-5-53-6-k/ingenTelegam_ing
import os
from io import BytesIO

from telethon import TelegramClient, events
from telethon.errors import MessageTooLongError, MediaCaptionTooLongError, MessageNotModifiedError, \
    MessageDeleteForbiddenError

from commands import IngTranscribeCommand, IngGPTCommand
from utils import CustomMarkdown
from functools import wraps

import logging

logging.basicConfig(level=logging.INFO)


def emoji_parser(func):
    @wraps(func)
    async def wrapper(self, *args, **kwargs):
        parse_mode = kwargs.get('parse_mode', None) or self.client.parse_mode
        kwargs['parse_mode'] = parse_mode
        return await func(self, *args, **kwargs)

    return wrapper


class CommandFactory:
    """
        Factory class responsible for creating command instances based on the provided command name.

        Methods:
        --------
        create_command(command_name, event, client, command_manager) -> Command:
            Returns a command instance for the given command name, or None if no matching command is found.
    """

    @staticmethod
    def create_command(command_name: str, event, client_handler: 'ClientHandler', command_manager: 'CommandManager'):
        command_class = command_manager.get_command(command_name)
        if command_class:
            return command_class(event, client_handler)
        return None


class EventAdapter:
    """
        Adapter class that provides a consistent interface for working with events.

        Attributes:
        -----------
        event: Event
            The original event object being wrapped.

        Methods:
        --------
        is_voice() -> bool:
            Determines if the event contains a voice message.

        get_message_text() -> str:
            Retrieves the text message from the event.
    """

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

    def is_voice(self):
        return self.event.message.voice

    def get_message_text(self):
        return self.event.message.message


class CommandManager:
    """
        Manager class that handles the registration and retrieval of command classes.

        Attributes:
        -----------
        _commands: dict
            Dictionary mapping command names to their corresponding classes.

        Methods:
        --------
        register_command(command_class):
            Registers a command class using its identifiers.

        register_commands(command_classes):
            Registers multiple command classes.

        get_command(command_name) -> CommandClass:
            Retrieves the command class for a given command name.
        """

    def __init__(self, commands=None):
        self._commands = {}
        if commands:
            self.register_commands(commands)

    def register_command(self, command_class):
        for identifier in command_class.get_identifiers():
            if identifier in self._commands:
                raise ValueError(f"Command with identifier {identifier} already registered!")
            self._commands[identifier] = command_class

    def register_commands(self, command_classes):
        for cmd_class in command_classes:
            self.register_command(cmd_class)

    def get_command(self, command_name):
        return self._commands.get(command_name)


class EventHandler:
    """
        Handler class that processes incoming events and executes the appropriate commands.

        Attributes:
        -----------
        client_handler: ClientHandler
            The client handler instance for sending messages.

        command_manager: CommandManager
            The command manager instance for retrieving command classes.

        Methods:
        --------
        handle(event):
            Handles the provided event by executing the relevant command.
    """

    def __init__(self, client_handler: 'ClientHandler'):
        self.client_handler = client_handler
        self.command_manager = CommandManager(client_handler.command_classes)

    async def handle(self, event):
        event_adapter = EventAdapter(event)
        command_name = EventHandler.get_command_name(event_adapter)
        if not command_name:
            return

        command = CommandFactory.create_command(command_name, event, self.client_handler, self.command_manager)
        if command:
            await command.execute()

    @staticmethod
    def get_command_name(event_adapter):
        if not isinstance(event_adapter, EventAdapter):
            raise TypeError(f"Expected an instance of EventAdapter, but got {type(event_adapter).__name__}.")

        if event_adapter.is_voice():
            return IngTranscribeCommand.command_name
        else:
            command_text = event_adapter.get_message_text()
            if not command_text:
                return None
            return command_text.split()[0]


class ClientHandler:
    """
        Handler class that encapsulates client-specific configurations and event handling logic.

        Attributes:
        -----------
        client: Client
            The client instance being managed.

        command_manager: CommandManager
            The command manager instance for the client.

        event_handler: EventHandler
            The event handler for processing client events.

        Methods:
        --------
        handle_event(event):
            Processes the given event using the client's event handler.

        download_voice(message, filepath, callback) -> str or None:
            Downloads the voice message from the given message and saves it to the specified filepath.
    """

    def __init__(self, client, command_classes):
        self.client = client
        client.parse_mode = CustomMarkdown()
        self.command_classes = command_classes
        self.event_handler = EventHandler(self)

    async def _register_event_handlers(self):
        @self.client.on(events.NewMessage(incoming=False))
        async def event_listener(event):
            await self.handle_event(event)

        logging.info("ClientHandler initialized.")

    async def start(self):
        await self._register_event_handlers()
        await self.client.start()

    async def handle_event(self, event):
        await self.event_handler.handle(event)

    async def download_media(self, message, file, progress_callback=None) -> str or None:
        return await self.client.download_media(message, file=file, progress_callback=progress_callback)

    async def send_text_as_file(self, peer_id, text, filename, reply_to=None):
        text_bytes = text.encode('utf-8')
        buffer = BytesIO(text_bytes)
        buffer.name = filename  # Add a name attribute to the BytesIO object
        buffer.seek(0)
        if reply_to:
            await self.client.send_file(peer_id, buffer, reply_to=reply_to)
        else:
            await self.client.send_file(peer_id, buffer)

    async def delete_message(self, peer_id, message_id):
        try:
            await self.client.delete_messages(peer_id, message_id)
        except MessageDeleteForbiddenError:
            logging.warning("Message delete forbidden")

    @emoji_parser
    async def edit_message(self, peer_id, message_id, new_text, parse_mode=None, link_preview=None, file=None,
                           force_document=None):
        try:
            await self.client.edit_message(peer_id, message_id, new_text, link_preview=link_preview, file=file,
                                           force_document=force_document, parse_mode=parse_mode)
        except MediaCaptionTooLongError:
            await self.send_text_as_file(peer_id, new_text, "transcription.txt")
        except MessageNotModifiedError:
            logging.warning("Message not modified")

    @emoji_parser
    async def send_message(self, peer_id, message, parse_mode=None, link_preview=None, file=None, force_document=None):
        parse_mode = parse_mode or self.client.parse_mode

        if len(message) > 4096:
            await self.send_text_as_file(peer_id, message, "transcription.txt")
        else:
            await self.client.send_message(peer_id, message, parse_mode=parse_mode, link_preview=link_preview,
                                           file=file, force_document=force_document)

    @emoji_parser
    async def reply_message(self, peer_id, message, reply_to, parse_mode=None, link_preview=None, file=None,
                            force_document=None):
        if len(message) > 4096:
            await self.send_text_as_file(peer_id, message, "transcription.txt", reply_to=reply_to)
        else:
            await self.client.send_message(peer_id, message, reply_to=reply_to, parse_mode=parse_mode,
                                           link_preview=link_preview, file=file, force_document=force_document)

    async def get_message_by_id(self, peer_id, ids):
        return await self.client.get_messages(peer_id, ids=ids)

    async def download_voice(self, message, filepath, callback) -> str or None:
        if message.voice:
            filename = await self.client.download_media(message, file=filepath,
                                                        progress_callback=callback)

            return filename if filename else None


class ClientFactory():
    @staticmethod
    def create_client(client_data) -> 'ClientHandler':
        session_location_prefix = 'data/sessions/'
        api_id = os.getenv("API_ID")
        api_hash = os.getenv("API_HASH")
        session_name = client_data.get("session_name")
        client = TelegramClient(session_location_prefix + session_name + ".session", api_id, api_hash)

        command_objects = [eval(command) for command in client_data['commands']]
        handler = ClientHandler(client, command_objects)

        return handler

control.py
44-5-53-6-k/ingenTelegam_ing
import os
import uuid
from abc import ABC


class BaseJob(ABC):
    def __init__(self, client_handler, message):
        self._id = str(uuid.uuid4())
        self._client_handler = client_handler
        self._message = message

    def _create_job_directory(self) -> str:
        job_directory = "jobs/" + self._id
        if not os.path.exists(job_directory):
            os.makedirs(job_directory)

        return job_directory


job_manager/job.py
44-5-53-6-k/ingenTelegam_ing
class TranscriptionResult:
    def __init__(self, verbose_answer):
        self._result = {
            "segments": []
        }
        self.duration = verbose_answer["duration"]
        self.verbose_answer = verbose_answer
        for segment in verbose_answer["segments"]:
            self._result["segments"].append({
                "start": segment["start"],
                "text": segment["text"]
            })

    def add_verbose_answer(self, verbose_answer):
        for segment in verbose_answer["segments"]:
            self._result["segments"].append({
                "start": segment["start"] + self.duration,
                "text": segment["text"]
            })

        self.duration += verbose_answer["duration"]

    def get_plain_text(self):
        text = ""
        for segment in self._result["segments"]:
            text += segment["text"]

        text = text[1:] if text[0] == " " else text

        return text

    def get_json(self):
        return self._result

job_manager/transcription_result.py
44-5-53-6-k/ingenTelegam_ing
from .job import BaseJob
from utils import AudioHelper
import openai
from job_manager.transcription_result import TranscriptionResult

import logging
logger = logging.getLogger(__name__)



class VoiceJob(BaseJob):
    SIZE_LIMIT = 20

    def __init__(self, client_handler, message):
        super().__init__(client_handler, message)
        self._job_directory = self._create_job_directory()
        self._prompt = ""
        self._client_handler = client_handler

    def get_file_path(self):
        return self._job_directory + "/" + self._id + ".ogg"

    async def process_job(self) -> TranscriptionResult:
        downloaded_media_path = await self.manage_download()
        converted_media_path = await AudioHelper.convert_media_to_mp3(downloaded_media_path)

        size = await AudioHelper.get_size_mb(converted_media_path)

        result = None
        if size > self.SIZE_LIMIT:
            logger.info(f"Audio file is too large ({size} MB), splitting into smaller segments")
            audio_segments = AudioHelper.split_audio(converted_media_path)
            for audio_segment in audio_segments:
                verbose_answer = await openai.Audio.atranscribe("whisper-1", audio_segment, response_format="verbose_json",
                                                                prompt=self._prompt)
                if result is None:
                    result = TranscriptionResult(verbose_answer)
                else:
                    result.add_verbose_answer(verbose_answer)

            print(result.get_json())

        else:
            result = await self.transcribe()

        return result

    async def manage_download(self):
        if not self._message.voice:
            raise TypeError("Message is not a voice message")

        filename = self._id + ".ogg"
        client = self._client_handler

        return await client.download_media(self._message, file=self._job_directory + "/" + filename)

    async def transcribe(self) -> TranscriptionResult:

        with open(self.get_file_path(), 'rb') as file:
            verbose_answer = await openai.Audio.atranscribe("whisper-1", file, response_format="verbose_json",
                                                            prompt=self._prompt)

        result = TranscriptionResult(verbose_answer)

        return result

job_manager/voice_job.py
44-5-53-6-k/ingenTelegam_ing
import math
import os

from pydub import AudioSegment


class AudioHelper:
    @staticmethod
    async def convert_media_to_mp3(file_path):
        input_format = os.path.splitext(file_path)[1][1:]
        output_path = os.path.splitext(file_path)[0] + '.mp3'
        AudioSegment.from_file(file_path, input_format).export(output_path, format='mp3')
        return output_path

    @staticmethod
    async def get_size_mb(file_path):
        size_in_bytes = os.path.getsize(file_path)
        size_in_mb = size_in_bytes / (1024 * 1024)
        return size_in_mb

    @staticmethod
    def split_audio(file_path, max_size=2) -> list[AudioSegment]:
        audio = AudioSegment.from_mp3(file_path)

        duration = len(audio)
        fragment_duration = math.ceil(
            (max_size * 1000 * 1024 * 1024) / audio.frame_rate / audio.sample_width / audio.channels)
        fragments = []
        start = 0

        while start < duration:
            end = min(start + fragment_duration, duration)
            fragments.append(audio[start:end])
            start = end

        return fragments


from telethon.extensions import markdown
from telethon import types


class CustomMarkdown:
    @staticmethod
    def parse(text):
        text, entities = markdown.parse(text)
        for i, e in enumerate(entities):
            if isinstance(e, types.MessageEntityTextUrl):
                if e.url == 'spoiler':
                    entities[i] = types.MessageEntitySpoiler(e.offset, e.length)
                elif e.url.startswith('emoji/'):
                    entities[i] = types.MessageEntityCustomEmoji(e.offset, e.length, int(e.url.split('/')[1]))
        return text, entities

    @staticmethod
    def unparse(text, entities):
        for i, e in enumerate(entities or []):
            if isinstance(e, types.MessageEntityCustomEmoji):
                entities[i] = types.MessageEntityTextUrl(e.offset, e.length, f'emoji/{e.document_id}')
            if isinstance(e, types.MessageEntitySpoiler):
                entities[i] = types.MessageEntityTextUrl(e.offset, e.length, 'spoiler')
        return markdown.unparse(text, entities)

utils.py
44-5-53-6-k/ingenTelegam_ing