app.py

"""
TikTok Downloader Bot - Gradio Web Interface
"""
import os
import sys
import asyncio
import tempfile
import logging
import threading
import yt_dlp
import gradio as gr
from pyrogram import Client, filters
from pyrogram.enums import ParseMode

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Import config
from config import API_ID, API_HASH, BOT_TOKEN

# Initialize Pyrogram bot (without phone_number - it's a bot account)
app = Client("TikTokDownloaderBot", api_id=API_ID, api_hash=API_HASH, bot_token=BOT_TOKEN)

# Bot status
bot_status = {"status": "Initializing...", "running": False}

# ============== TELEGRAM BOT HANDLERS ==============

@app.on_message(filters.command("start") & filters.private)
async def start_handler(client, message):
    await message.reply_text(
        "👋 **Welcome to TikTok Downloader Bot!**\n\n"
        "Send me a TikTok video link to download.\n\n"
        "**Commands:**\n"
        "/tt <url> - Download TikTok video\n"
        "/help - Show help",
        parse_mode=ParseMode.MARKDOWN
    )

@app.on_message(filters.command("help") & filters.private)
async def help_handler(client, message):
    await message.reply_text(
        "📖 **Help**\n\n"
        "**Usage:**\n"
        "1. Send a TikTok video link\n"
        "2. Wait for download\n"
        "3. Video will be sent to you\n\n"
        "**Supported:**\n"
        "- TikTok videos (with/without watermark)\n"
        "- Video metadata display",
        parse_mode=ParseMode.MARKDOWN
    )

@app.on_message(filters.command("tt") & filters.private)
async def download_handler(client, message):
    if len(message.command) < 2:
        await message.reply_text(
            "❌ **Please provide a TikTok URL**\n\n"
            "**Usage:** `/tt <tiktok_url>`",
            parse_mode=ParseMode.MARKDOWN
        )
        return
    
    url = message.command[1]
    status_msg = await message.reply_text("🔍 Searching...")
    
    try:
        # Get video info
        await status_msg.edit("📡 Fetching video info...")
        ydl_opts = {'quiet': True, 'extract_flat': False}
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            info = ydl.extract_info(url, download=False)
            title = info.get('title', 'Video')
            views = info.get('view_count', 0)
            duration = info.get('duration', 0)
        
        # Download
        await status_msg.edit("⬇️ Downloading video...")
        output_file = tempfile.NamedTemporaryFile(suffix='.mp4', delete=False)
        output_file.close()
        
        ydl_opts = {
            "format": "mp4",
            "outtmpl": output_file.name,
            "quiet": True,
        }
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            ydl.download([url])
        
        # Upload
        await status_msg.edit("⬆️ Uploading to Telegram...")
        duration_str = f"{int(duration//60)}:{int(duration%60):02d}" if duration else "N/A"
        
        caption = (
            f"🎵 **Title:** {title}\n"
            f"👁️ **Views:** {views:,}\n"
            f"⏱️ **Duration:** {duration_str}\n"
            f"🔗 [Watch on TikTok]({url})"
        )
        
        await message.reply_video(
            video=output_file.name,
            caption=caption,
            parse_mode=ParseMode.MARKDOWN
        )
        
        await status_msg.delete()
        os.remove(output_file.name)
        
    except Exception as e:
        logger.error(f"Download error: {e}")
        await status_msg.edit(f"❌ **Error:** `{str(e)}`", parse_mode=ParseMode.MARKDOWN)

# ============== GRADIO WEB INTERFACE ==============

def download_tiktok(url):
    """Download TikTok video using yt-dlp"""
    output_file = tempfile.NamedTemporaryFile(suffix='.mp4', delete=False)
    output_file.close()
    
    ydl_opts = {
        "format": "mp4",
        "outtmpl": output_file.name,
        "quiet": True,
    }
    
    try:
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            info = ydl.extract_info(url, download=True)
            return output_file.name, info.get('title', 'Video')
    except Exception as e:
        logger.error(f"Download error: {e}")
        return None, str(e)

def get_video_info(url):
    """Get video metadata without downloading"""
    ydl_opts = {'quiet': True, 'extract_flat': False}
    try:
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            info = ydl.extract_info(url, download=False)
            return {
                "title": info.get('title', 'N/A'),
                "views": f"{info.get('view_count', 0):,}",
                "duration": f"{int(info.get('duration', 0)//60)}:{int(info.get('duration', 0)%60):02d}",
            }
    except Exception as e:
        return {"error": str(e)}

def gradio_download(url):
    """Gradio interface for downloading TikTok videos"""
    if not url:
        yield None, "❌ Please enter a TikTok URL"
        return
    
    yield None, "🔍 Fetching video info..."
    info = get_video_info(url)
    
    if "error" in info:
        yield None, f"❌ Error: {info['error']}"
        return
    
    yield None, f"📥 Downloading: {info['title']}..."
    file_path, result = download_tiktok(url)
    
    if not file_path:
        yield None, f"❌ Download failed: {result}"
        return
    
    yield file_path, f"✅ Success! Downloaded: {info['title']}"
    
    try:
        os.remove(file_path)
    except:
        pass

# ============== START TELEGRAM BOT ==============

def run_pyrogram():
    """Run Pyrogram bot in separate thread"""
    global bot_status
    try:
        if API_ID and API_HASH and BOT_TOKEN:
            bot_status["status"] = "🤖 Telegram bot starting..."
            import asyncio
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            loop.run_until_complete(app.run())
        else:
            bot_status["status"] = "⚠️ Telegram credentials not configured"
    except Exception as e:
        bot_status["status"] = f"❌ Bot error: {e}"
        logger.error(f"Bot error: {e}")

# Start bot in background thread if credentials exist
if API_ID and API_HASH and BOT_TOKEN:
    bot_thread = threading.Thread(target=run_pyrogram, daemon=True)
    bot_thread.start()
    bot_status["status"] = "🤖 Telegram bot running"
    bot_status["running"] = True
else:
    bot_status["status"] = "⚠️ Telegram credentials not configured"

# ============== GRADIO APP ==============

with gr.Blocks(title="TikTok Downloader") as demo:
    gr.Markdown("# 🎵 TikTok Video Downloader")
    gr.Markdown("### Download TikTok videos directly to your device")
    
    with gr.Row():
        with gr.Column():
            url_input = gr.Textbox(label="TikTok URL", placeholder="Paste TikTok link here...")
            download_btn = gr.Button("⬇️ Download", variant="primary")
        with gr.Column():
            video_output = gr.Video(label="Downloaded Video")
            status_output = gr.Textbox(label="Status", lines=2)
    
    gr.Markdown("---")
    gr.Markdown(f"**Telegram Bot Status:** `{bot_status['status']}`")
    gr.Markdown("""
    ### 📌 How to use:
    1. Paste TikTok URL above
    2. Click Download
    3. Save the video
    """)
    
    download_btn.click(fn=gradio_download, inputs=url_input, outputs=[video_output, status_output])

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860)