app.py

import io
import os
import sys
import time
import secrets
import asyncio 
import threading
import traceback
import gradio as gr

from io import BytesIO
from fusion import Fusion
from datetime import datetime
from telethon.tl.tlobject import TLObject
from telethon import TelegramClient, events, Button, types


API_ID = os.environ.get("API_ID")
API_HASH = os.environ.get("API_HASH")
BOT_TOKEN = os.environ.get("BOT_TOKEN")


client = TelegramClient('session_name', API_ID, API_HASH)


def utc_to_local(utc_datetime):
    now_timestamp = time.time()
    offset = datetime.fromtimestamp(now_timestamp) - datetime.utcfromtimestamp(
        now_timestamp
    )
    return utc_datetime + offset

def yaml_format(obj, indent=0, max_str_len=256, max_byte_len=64):
    result = []
    if isinstance(obj, TLObject):
        obj = obj.to_dict()

    if isinstance(obj, dict):
        if not obj:
            return "dict:"
        items = obj.items()
        has_items = len(items) > 1
        has_multiple_items = len(items) > 2
        result.append(obj.get("_", "dict") + (":" if has_items else ""))
        if has_multiple_items:
            result.append("\n")
            indent += 2
        for k, v in items:
            if k == "_" or v is None:
                continue
            formatted = yaml_format(v, indent)
            if not formatted.strip():
                continue
            result.append(" " * (indent if has_multiple_items else 1))
            result.append(f"{k}:")
            if not formatted[0].isspace():
                result.append(" ")
            result.append(f"{formatted}")
            result.append("\n")
        if has_items:
            result.pop()
        if has_multiple_items:
            indent -= 2
    elif isinstance(obj, str):
        result = repr(obj[:max_str_len])
        if len(obj) > max_str_len:
            result += "…"
        return result
    elif isinstance(obj, bytes):
        if all(0x20 <= c < 0x7F for c in obj):
            return repr(obj)
        return "<…>" if len(obj) > max_byte_len else " ".join(f"{b:02X}" for b in obj)
    elif isinstance(obj, datetime):
        return utc_to_local(obj).strftime("%Y-%m-%d %H:%M:%S")
    elif hasattr(obj, "__iter__"):
        result.append("\n")
        indent += 2
        for x in obj:
            result.append(f"{' ' * indent}- {yaml_format(x, indent + 2)}")
            result.append("\n")
        result.pop()
        indent -= 2
    else:
        return repr(obj)
    return "".join(result)

async def aexec(code, smessatatus):
    message = event = smessatatus
    p = lambda _x: print(yaml_format(_x))
    reply = await event.get_reply_message()
    exec("async def __aexec(message, event , reply, client, p, chat): "
            + "".join(f"\n {l}" for l in code.split("\n")))

    return await locals()["__aexec"](
        message, event, reply, message.client, p, message.chat_id
    )

@client.on(events.NewMessage(incoming=True, pattern="(/)?eval(?:\s|$)([\s\S]*)"))
async def evalution(event):
    if event.sender.id != 6034486765: return

    cmd = "".join(event.message.message.split(maxsplit=1)[1:])
    if not cmd:
        return await event.reply("`Give something to run! ...`")
    eval_ = await event.reply("`Evalution in process ...`")
    old_stderr = sys.stderr
    old_stdout = sys.stdout
    redirected_output = sys.stdout = io.StringIO()
    redirected_error = sys.stderr = io.StringIO()
    stdout, stderr, exc = None, None, None
    try:
        await aexec(cmd, event)
    except Exception:
        exc = traceback.format_exc()
    stdout = redirected_output.getvalue()
    stderr = redirected_error.getvalue()
    sys.stdout = old_stdout
    sys.stderr = old_stderr
    evaluation = ""
    if exc:
        evaluation = exc
    elif stderr:
        evaluation = stderr
    elif stdout:
        evaluation = stdout
    else:
        evaluation = "Success"
    final_output = (f"**•  Syntax : **\n```{cmd}``` \n\n**•  Output :**\n```{evaluation}``` \n")
    await eval_.edit(text=final_output)


get_duration = lambda x: int(len(Fusion.from_file(x))/1000.0)
states = {}

@client.on(events.NewMessage(pattern='/start'))
async def start_handler(event):
    await event.reply("Welcome to AudioFusion Bot! Send me an audio file, and I'll apply effects for you.")

buttons = [
        [Button.inline('Slowed', b'slowed'), Button.inline('8D', b'8d')],
        [Button.inline('Reverb', b'reverb'), Button.inline('Reverse', b'reverse')],
        [Button.inline('Volume', b'volume'), Button.inline('Speedup', b'speedup')],
        [Button.inline('Preview', b'preview')],
        [Button.inline('Send', b'send')],
    ]

@client.on(events.NewMessage(pattern='/buttons'))
async def buttons_handler(event):
    user_id = event.sender_id
        
    # Download the audio file and store it in the user's state
    reply_message = await event.get_reply_message()
    if not reply_message or not reply_message.file:
        await event.reply("Please reply to an audio file.")
        return

    audio_file = BytesIO()
    await event.client.download_media(reply_message, audio_file)
    audio_file.seek(0)
    
    # Store the audio file in the user's state
    states[user_id] = {'audio': audio_file}

    await client.send_file(event.chat_id, file="image.jpg", caption="Preview the current modification:", buttons=buttons)
    

@client.on(events.CallbackQuery(pattern=b'(slowed|8d|reverb|reverse|trim|volume|speedup)'))
async def audio_effect_handler(event):
    user_id = event.sender_id
    if user_id not in states or not states[user_id]:
        await event.answer("No audio file found. Please use /buttons command to upload an audio file.")
        return

    # Retrieve the audio file from the user's state
    audio_file = states[user_id]['audio']

    query = event.pattern_match.group(1).decode("UTF-8")
    sound = Fusion.from_file(audio_file, format="mp3")
    if query == 'slowed':
        modified_sound = await Fusion.effectSlowed(sound)
    elif query == 'speedup':
        modified_sound = await Fusion.effectSlowed(sound, 1.1)
    elif query == '8d':
        modified_sound = await Fusion.effect8D(sound)
    elif query == 'reverb':
        modified_sound = await Fusion.effectReverb(sound)
    elif query == 'reverse':
        modified_sound = sound.reverse()
    else:
        return await event.answer("INvalid for now...")
        
    audio_file = BytesIO()
    audio = modified_sound.export(audio_file, format="mp3")
    audio.seek(0)
    # Update the user's state with the modified sound
    states[user_id]['audio'] = audio

    await event.answer("Effect applied. Click /send to receive the modified audio file.", alert=True)



@client.on(events.CallbackQuery(pattern=b'preview'))
async def preview_handler(event):
    user_id = event.sender_id
    if user_id in states and states[user_id]:
        # Send the current modification for preview
        output_file_name = f"{user_id}_preview"
        output_file = await Fusion.saveSound(states[user_id]['audio'], output_file_name)
        await event.edit("`Uploading...`", buttons=buttons)
        # Edit the message and send the audio file in the edited message
        await event.edit(file=output_file, text="`Preview the current modification:`", buttons=buttons)
        
        # Clean up - remove the saved preview audio file
        os.remove(output_file)
    else:
        await event.answer("No modified audio file found. Please apply an effect first.", alert=True)


@client.on(events.CallbackQuery(pattern=b'send'))
async def send_handler(event):      
    user_id = event.sender_id
    if user_id in states and states[user_id]:
        audio_file = states[user_id]['audio']

        audio_file.name = "AudioFusion.mp3"
        duration = int(len(Fusion.from_file(audio_file))/1000.0)
        audio_file.seek(0)
        await client.send_file(event.chat_id, file=audio_file)
        # Clean up - remove the user's state and the saved audio file
        del states[user_id]
        # os.remove(output_file)
        await event.delete()
    else:
        await event.answer("No modified audio file found. Please apply an effect first.")



def process_audio(input_file,
    effect_8d, pan_boundary, jump_percentage, time_l_to_r, volume_multiplier,
    effect_slowed, speed_multiplier,
    effect_reverb, room_size, damping, width, wet_level, dry_level
):
    # Load the sound file  
    sound = Fusion.loadSound(input_file)
    os.remove(os.path.abspath(input_file))
    effects_str = []
    
    # Apply effects based on user choices
    if effect_8d:
        sound = Fusion.effect8D(sound, pan_boundary, jump_percentage, time_l_to_r*1000, volume_multiplier)
        effects_str.append("8d")
    if effect_slowed:
        sound = Fusion.effectSlowed(sound, speed_multiplier)
        effects_str.append("Slowed")
    if effect_reverb:
        sound = Fusion.effectReverb(sound, room_size, damping, width, wet_level, dry_level, str(secrets.token_hex(5)))
        effects_str.append("Reverb")

    output_file = f"{input_file} {' + '.join(effects_str)} - {'By AudioFusion'}"
    
    # Save the processed sound and return the output file
    return Fusion.saveSound(sound, output_file)
    


before_text = """<div align="center">
<h1>AudioFusion</h1>
<i>Add a touch of uniqueness with various customizable effects like slowed and reverb.</i>
</div>
<hr>"""

after_text = """<hr>
PR in [github](https://github.com/MineisZarox/AudioFusion) repository beta branch are always welcome.

<h3>Todo</h3>

\# Acapella Extractor<br>
\# Karoke Maker<br>
\# Bass Booster<br>
\# Volume Booster<br>


<h3>Inspiration & Credits</h3>

- Special thanks to [Jiaaro](https://github.com/jiaaro) for pydub. AudioFusion is mainly wrapped around pydub

- My Soundscapes of Serenity - [Because](https://t.me/bcuzwhynot)
"""

with gr.Blocks(title="Audio Fusion") as iface:
    gr.Markdown(before_text)
    input_audio = gr.Audio(label="Upload your music file", type="filepath")
    
    # SLowed Effect and its arguments
    with gr.Tab("Slowed Effect"):
        speed_check = gr.Checkbox(label="Apply slowed effect")
        with gr.Column(visible=False) as se_col:
            speed = gr.Slider(label="Speed Multiplier", minimum=0.1, maximum=4, step=0.05, value=0.90)
    
    # Reverb Effect and its argument
    with gr.Tab("Reverb Effect"):
        reverb_check = gr.Checkbox(label="Apply reverb effect")
        with gr.Column(visible=False) as re_col:
            with gr.Row():
                room = gr.Slider(label="Room Size", minimum=0, maximum=1, step=0.01, value=0.8)
                damp = gr.Slider(label="Damping", minimum=0, maximum=1, step=0.05, value=1)
            width = gr.Slider(label="Width", minimum=0, maximum=1, step=0.05, value=0.5)
            with gr.Row():
                wet = gr.Slider(label="Wet Level", minimum=0, maximum=1, step=0.05, value=0.3)
                dry = gr.Slider(label="Dry Level", minimum=0, maximum=1, step=0.05, value=0.8)
    
    # 8d Effect and its arguments
    with gr.Tab("8d Effect"):
        dimension_check = gr.Checkbox(label="Apply 8D effect")
        with gr.Column(visible=False) as di_col:
            with gr.Row():
                pan = gr.Slider(label="Pan Boundary", minimum=0, maximum=100, value=90)
                jump = gr.Slider(label="Jump Percentage", minimum=1, maximum=100, value=5)
            with gr.Row():
                time = gr.Slider(label="Time L to R (s)", minimum=1, maximum=30, value=10)
                volx = gr.Slider(label="Volume Multiplier", minimum=1, maximum=20, value=6)

    # =====================================================
    def di_v(check):
        if check:
            return {di_col: gr.Column(visible=True)}
        else:
            return {di_col: gr.Column(visible=False)}
            
    def se_v(check):
        if check:
            return {se_col: gr.Column(visible=True)}
        else:
            return {se_col: gr.Column(visible=False)}
            
    def re_v(check):
        if check:
            return {re_col: gr.Column(visible=True)}
        else:
            return {re_col: gr.Column(visible=False)}
        

    dimension_check.change(di_v, inputs=[dimension_check], outputs=[di_col])
    speed_check.change(se_v, inputs=[speed_check], outputs=[se_col])
    reverb_check.change(re_v, inputs=[reverb_check], outputs=[re_col])
    # =====================================================

    with gr.Row():
        btnClear = gr.ClearButton(components=[dimension_check, speed_check, reverb_check])
        btnRun = gr.Button("Run", size="sm", variant="primary")
    
    inputs = [input_audio, dimension_check, pan, jump, time, volx, speed_check, speed, reverb_check, room, damp, width, wet, dry]
    output = [gr.Audio(label="Download processed music", type="filepath")]

    gr.Markdown(after_text)

    btnClear.add(components=output)
    btnRun.click(fn=process_audio, inputs=inputs, outputs=output, api_name="AudioFusion")

async def initiation():
    await client.send_message(-1001662130485, "**Hugging is Running.**", buttons=[(Button.url("Execal", "https://t.me/execal"),)],)

if __name__ == '__main__':
    client.start(bot_token=BOT_TOKEN)
    client.loop.run_until_complete(initiation())
    threading.Thread(target=iface.launch).start() #(share=False)
    print("Bot started succefully")
    client.run_until_disconnected()