Initial commit - Telegram thumbnail uploader bot

.gitignore

@@ -0,0 +1,7 @@
+.env
+sessions/
+temp/
+__pycache__/
+*.pyc
+venv/
+.env.*

README.md

@@ -0,0 +1,266 @@
+# Telegram Thumbnail Upload Bot
+
+Bot ini untuk **upload video ke Telegram bersama thumbnail secara automatik**.  
+
+---
+
+## Keperluan
+
+Pastikan komputer sudah install:
+
+Disarankan guna Python **3.12 atau 3.13**
+
+Python 3.13  
+https://www.python.org/downloads/release/python-3130/
+
+Python 3.12  
+https://www.python.org/downloads/release/python-3120/
+
+**FFmpeg**  
+https://ffmpeg.org/download.html
+
+### Add FFmpeg to PATH (Windows)
+
+Selepas download dan extract FFmpeg, pastikan folder `bin` dimasukkan ke dalam **System PATH**.
+
+Contoh lokasi:
+
+```
+C:\ffmpeg\bin
+```
+
+Kalau folder ini tidak dimasukkan ke PATH, command `ffmpeg` dan `ffprobe` tidak akan dikenali oleh sistem.
+
+Lepas install, cuba test:
+
+```
+ffmpeg -version
+ffprobe -version
+```
+
+Kalau keluar version info maksudnya sudah OK.
+
+---
+
+## Setup Telegram API
+
+Pergi ke:
+
+https://my.telegram.org
+
+Login Telegram → masuk **API development tools**
+
+Ambil:
+
+```
+API_ID
+API_HASH
+```
+
+---
+
+## Buat Bot
+
+Pergi ke:
+
+https://t.me/BotFather
+
+Run command:
+
+```
+/newbot
+```
+
+Lepas siap BotFather akan bagi:
+
+```
+BOT_TOKEN
+```
+
+---
+
+## Setup `.env`
+
+Dalam folder bot ada file:
+
+```
+.env.example
+```
+
+Rename jadi:
+
+```
+.env
+```
+
+Lepas tu isi macam ni:
+
+```
+BOT_TOKEN=
+API_ID=
+API_HASH=
+```
+
+Contoh:
+
+```
+BOT_TOKEN=123456:ABCDEF
+API_ID=123456
+API_HASH=abcdef1234567890
+```
+
+Jangan share file `.env` ni dengan orang lain.
+
+---
+
+## Jalankan Bot
+
+Double click:
+
+```
+Start.bat
+```
+
+Script ni akan:
+
+- create virtual environment
+- install dependency
+- start bot
+
+Kalau berjaya akan nampak:
+
+```
+Bot started successfully
+```
+
+Selepas bot berjalan, buka Telegram dan cuba command berikut untuk test:
+
+```
+/send_video
+```
+
+atau
+
+```
+/batch
+```
+
+Bot akan menggunakan sample file yang sudah disediakan dalam folder:
+
+```
+single/
+batch/
+```
+
+## Test Bot (Sample Files)
+
+Repository ini sudah menyediakan **sample video dan thumbnail** supaya anda boleh terus test bot.
+
+---
+
+## Upload Video (Single)
+
+Letak video dalam folder:
+
+```
+single/
+```
+
+Contoh:
+
+```
+single/
+   sample_video.mp4
+   sample_video.jpg
+```
+
+Kemudian dalam Telegram hantar command:
+
+```
+/send_video
+```
+
+Bot akan upload semua video dalam folder tu.
+
+Pastikan **nama thumbnail sama dengan nama video**.
+
+Contoh:
+
+```
+sample_video.mp4
+sample_video.jpg
+```
+
+---
+
+## Upload Drama (Batch)
+
+Struktur folder:
+
+```
+batch/
+
+drama1/
+   sample_video.mp4
+   sample_video.jpg
+```
+
+Dalam Telegram run:
+
+```
+/batch
+```
+
+Bot akan upload semua video dalam setiap folder batch.
+
+---
+
+## Nota
+
+Kali pertama bot run dia akan minta:
+
+```
+Phone number
+OTP code
+```
+
+Ini untuk create **user session** supaya boleh upload video besar.
+
+Session akan disimpan dalam folder:
+
+```
+sessions/
+```
+
+---
+
+## Converter (Optional)
+
+Dalam folder `converter` ada tool kecil untuk convert image ke JPG kalau thumbnail format pelik macam:
+
+```
+AVIF
+WEBP
+HEIC
+PNG
+```
+
+Run saja:
+
+```
+converter/Start.bat
+```
+
+atau
+
+```
+python converter/converter.py
+```
+
+Letakkan image dalam folder `converter/` dan hasil convert akan disimpan dalam folder `output_images/`.
+
+---
+
+## Siap
+
+Lepas setup semua ni, bot dah boleh guna untuk upload video ke Telegram dengan senang.

Start.bat

@@ -0,0 +1,37 @@
+@echo off
+chcp 65001 >nul
+cd /d "%~dp0"
+
+echo ==============================
+echo Telegram Thumbnail Bot
+echo ==============================
+
+REM 
+where python >nul 2>nul
+if %errorlevel% neq 0 (
+    echo Python tidak dijumpai. Sila install Python dahulu.
+    pause
+    exit
+)
+
+REM 
+if not exist venv (
+    echo First time setup detected...
+    echo Creating virtual environment...
+    python -m venv venv
+
+    call venv\Scripts\activate.bat
+
+    echo Updating pip...
+    python -m pip install --upgrade pip setuptools wheel
+
+    echo Installing requirements...
+    pip install -r requirements.txt
+) else (
+    call venv\Scripts\activate.bat
+)
+
+echo Starting bot...
+python main.py
+
+pause

converter/Start.bat

@@ -0,0 +1,26 @@
+@echo off
+chcp 65001 >nul
+cd /d "%~dp0"
+
+echo ==============================
+echo Image Converter Tool
+echo ==============================
+echo.
+
+REM 
+cd ..
+
+REM 
+call venv\Scripts\activate.bat
+
+REM 
+cd converter
+
+echo Starting converter...
+echo.
+
+python converter.py
+
+echo.
+echo Conversion finished.
+pause

converter/converter.py

@@ -0,0 +1,53 @@
+import os
+import imageio.v3 as iio
+from PIL import Image
+
+input_dir = os.getcwd()
+output_dir = os.path.join(input_dir, "output_images")
+os.makedirs(output_dir, exist_ok=True)
+
+supported_extensions = {
+    ".avif", ".webp", ".png", ".heif", ".heic",
+    ".bmp", ".tiff"
+}
+
+files = [
+    f for f in os.listdir(input_dir)
+    if os.path.splitext(f)[1].lower() in supported_extensions
+]
+
+total = len(files)
+count = 0
+
+print(f"Found {total} convertible images\n")
+
+for filename in files:
+    filepath = os.path.join(input_dir, filename)
+    name, ext = os.path.splitext(filename)
+    ext = ext.lower()
+
+    count += 1
+
+    try:
+        if ext == ".avif":
+            img_array = iio.imread(filepath)
+            img = Image.fromarray(img_array).convert("RGB")
+        else:
+            img = Image.open(filepath).convert("RGB")
+
+        output_path = os.path.join(output_dir, name + ".jpg")
+
+        if os.path.exists(output_path):
+            print(f"[{count}/{total}] ⏭️ Skipped (exists): {name}.jpg")
+            continue
+
+        img.thumbnail((320, 320))
+
+        img.save(output_path, "JPEG", quality=90, subsampling=0)
+
+        print(f"[{count}/{total}] ✅ Converted: {filename} → {name}.jpg")
+
+    except Exception as e:
+        print(f"[{count}/{total}] ❌ Failed: {filename} ({e})")
+
+print("\nConversion finished.")

main.py

@@ -0,0 +1,231 @@
+import os
+import time
+import logging
+import subprocess
+import json
+from dotenv import load_dotenv
+from pyrogram import Client, filters
+from pyrogram.errors import FloodWait
+import asyncio
+
+from utils import (
+    get_local_videos,
+    process_local_thumbnail,
+    get_batch_directories,
+    get_batch_videos
+)
+
+load_dotenv()
+
+BOT_TOKEN = os.getenv("BOT_TOKEN")
+API_ID = int(os.getenv("API_ID"))
+API_HASH = os.getenv("API_HASH")
+
+if not BOT_TOKEN or not API_ID or not API_HASH:
+    raise ValueError("Please configure BOT_TOKEN, API_ID and API_HASH in .env")
+
+USER_SESSION = "user_session"
+
+SESSION_DIR = os.path.join(os.path.dirname(__file__), "sessions")
+os.makedirs(SESSION_DIR, exist_ok=True)
+
+TEMP_DIR = os.path.join(os.path.dirname(__file__), "temp")
+os.makedirs(TEMP_DIR, exist_ok=True)
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s"
+)
+
+logger = logging.getLogger(__name__)
+
+Bot = Client(
+    os.path.join(SESSION_DIR, "Thumb-Bot"),
+    bot_token=BOT_TOKEN,
+    api_id=API_ID,
+    api_hash=API_HASH
+)
+
+User = Client(
+    os.path.join(SESSION_DIR, USER_SESSION),
+    api_id=API_ID,
+    api_hash=API_HASH
+)
+
+
+def remove_extension(name: str) -> str:
+    return os.path.splitext(name)[0]
+
+
+def get_video_dimensions(path: str):
+    cmd = [
+        "ffprobe", "-v", "error",
+        "-select_streams", "v:0",
+        "-show_entries", "stream=width,height",
+        "-of", "json",
+        path
+    ]
+
+    try:
+        result = subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            check=False
+        )
+
+        if not result.stdout:
+            return 1280, 720
+
+        data = json.loads(result.stdout)
+        streams = data.get("streams")
+
+        if not streams:
+            return 1280, 720
+
+        width = streams[0].get("width", 1280)
+        height = streams[0].get("height", 720)
+
+        return int(width), int(height)
+
+    except Exception:
+        return 1280, 720
+
+
+def get_video_duration(path: str):
+    cmd = [
+        "ffprobe", "-v", "error",
+        "-show_entries", "format=duration",
+        "-of", "default=noprint_wrappers=1:nokey=1",
+        path
+    ]
+
+    try:
+        result = subprocess.run(cmd, capture_output=True, text=True)
+        return int(float(result.stdout.strip()))
+    except:
+        return 0
+
+async def send_video_safe(
+    client,
+    chat_id,
+    video_path,
+    caption=None,
+    thumb=None
+):
+    width, height = get_video_dimensions(video_path)
+    duration = get_video_duration(video_path)
+
+    while True:
+        try:
+            await client.send_video(
+                chat_id=chat_id,
+                video=video_path,
+                caption=caption,
+                thumb=thumb,
+                width=width,
+                height=height,
+                duration=duration,
+                supports_streaming=True
+            )
+            break
+
+        except FloodWait as e:
+            logger.warning(
+                f"FloodWait while uploading {os.path.basename(video_path)}. "
+                f"Sleeping {e.value + 1} seconds..."
+            )
+            await asyncio.sleep(e.value + 1)
+
+@Bot.on_message(filters.command("start"))
+async def start(_, m):
+    await m.reply(
+        "Video uploader bot ready.\n\n"
+        "Commands:\n"
+        "/send_video - Upload videos from single folder\n"
+        "/batch - Upload videos from batch folders"
+    )
+
+
+@Bot.on_message(filters.command("send_video"))
+async def send_local(_, m):
+
+    status = await m.reply("Scanning local videos...")
+
+    videos, _ = get_local_videos()
+
+    if not videos:
+        await status.edit("No videos found.")
+        return
+
+    for v in videos:
+
+        msg = await m.reply(f"Uploading {v['filename']}...")
+
+        thumb = None
+        if v["thumb_path"]:
+            thumb = process_local_thumbnail(v["thumb_path"])
+
+        await send_video_safe(
+            User,
+            m.chat.id,
+            v["video_path"],
+            caption=remove_extension(v["filename"]),
+            thumb=thumb
+        )
+
+        await msg.delete()
+
+        if thumb and os.path.exists(thumb):
+            os.remove(thumb)
+
+    await status.edit("All videos uploaded.")
+
+
+@Bot.on_message(filters.command("batch"))
+async def batch(_, m):
+
+    status = await m.reply("Scanning batch folders...")
+
+    batches = get_batch_directories()
+
+    if not batches:
+        await status.edit("No batch folders found.")
+        return
+
+    for batch_dir in batches:
+
+        videos = get_batch_videos(batch_dir)
+
+        if not videos:
+            continue
+
+        thumb = None
+
+        if videos[0]["thumb_path"]:
+            thumb = process_local_thumbnail(videos[0]["thumb_path"])
+
+        for v in videos:
+
+            msg = await m.reply(f"Uploading {v['filename']}...")
+
+            await send_video_safe(
+                User,
+                m.chat.id,
+                v["video_path"],
+                caption=remove_extension(v["filename"]),
+                thumb=thumb
+            )
+
+            await msg.delete()
+
+        if thumb and os.path.exists(thumb):
+            os.remove(thumb)
+
+    await status.edit("Batch upload complete.")
+
+
+print("Bot started successfully")
+
+with User:
+    Bot.run()

requirements.txt

@@ -0,0 +1,8 @@
+pyrofork>=2.3.69
+TgCrypto-pyrofork==1.2.8
+humanize==4.15.0
+Pillow>=12.1.0
+python-dotenv==1.2.1
+pillow-heif
+pillow-avif-plugin
+imageio

utils.py

@@ -0,0 +1,212 @@
+import os
+import re
+import time
+import logging
+from pathlib import Path
+from PIL import Image
+
+logger = logging.getLogger(__name__)
+
+BASE_DIR = os.path.dirname(__file__)
+SINGLE_DIR = os.path.join(BASE_DIR, "single")
+BATCH_DIR = os.path.join(BASE_DIR, "batch")
+
+TEMP_DIR = os.path.join(BASE_DIR, "temp")
+os.makedirs(TEMP_DIR, exist_ok=True)
+
+VIDEO_FORMATS = (
+    ".mp4", ".mkv", ".webm", ".avi", ".mov",
+    ".m4v", ".flv", ".wmv", ".ts",
+    ".mpg", ".mpeg", ".3gp", ".mp4v", ".vob"
+)
+
+THUMB_FORMATS = (".jpg", ".jpeg", ".png")
+
+def find_matching_thumbnail(video_path):
+    video_name = Path(video_path).stem.lower()
+
+    # Exact match
+    for ext in THUMB_FORMATS:
+        exact_match = os.path.join(SINGLE_DIR, f"{video_name}{ext}")
+        if os.path.exists(exact_match):
+            logger.info(f"Exact thumbnail found: {exact_match}")
+            return exact_match
+
+    video_keywords = video_name.split(".")
+
+    for file in os.listdir(SINGLE_DIR):
+        if file.lower().endswith(THUMB_FORMATS):
+
+            thumb_name = Path(file).stem.lower()
+
+            if (
+                thumb_name in video_name
+                or video_name in thumb_name
+                or any(keyword in thumb_name for keyword in video_keywords)
+            ):
+                return os.path.join(SINGLE_DIR, file)
+
+    return None
+
+
+def get_local_videos():
+    video_files = []
+
+    logger.info(f"Scanning single folder: {SINGLE_DIR}")
+
+    videos = [
+        f for f in os.listdir(SINGLE_DIR)
+        if f.lower().endswith(VIDEO_FORMATS)
+    ]
+
+    videos.sort()
+
+    for video in videos:
+
+        video_path = os.path.join(SINGLE_DIR, video)
+
+        thumb_path = find_matching_thumbnail(video_path)
+
+        video_files.append({
+            "video_path": video_path,
+            "thumb_path": thumb_path,
+            "filename": video,
+            "size": os.path.getsize(video_path),
+            "has_thumb": thumb_path is not None
+        })
+
+        logger.info(
+            f"Found video: {video} {'with thumbnail' if thumb_path else 'no thumbnail'}"
+        )
+
+    return video_files, {}
+
+def process_local_thumbnail(thumb_path):
+
+    try:
+        img = Image.open(thumb_path).convert("RGB")
+
+        width, height = img.size
+
+        if width > height:
+            new_width = 320
+            new_height = int(320 * height / width)
+        else:
+            new_height = 320
+            new_width = int(320 * width / height)
+
+        img = img.resize((new_width, new_height), Image.Resampling.LANCZOS)
+
+        output_path = os.path.join(
+            TEMP_DIR,
+            f"thumb_{int(time.time())}.jpg"
+        )
+
+        img.save(output_path, "JPEG", quality=95)
+
+        return output_path
+
+    except Exception as e:
+        logger.error(f"Thumbnail processing error: {e}")
+        return None
+
+def find_batch_thumbnail(batch_dir):
+
+    for file in os.listdir(batch_dir):
+        if file.lower().endswith(THUMB_FORMATS):
+            return os.path.join(batch_dir, file)
+
+    return None
+
+
+def extract_episode_number(filename):
+
+    patterns = [
+        r'[Ss](\d+)[Ee](\d+)',  
+        r'[Ee][Pp]?\.?\s*(\d+)', 
+        r'E(\d+)',               
+        r'(\d+)'               
+    ]
+
+    filename = filename.lower()
+
+    for pattern in patterns:
+
+        match = re.search(pattern, filename)
+
+        if match:
+            try:
+                if len(match.groups()) == 2:
+                    return int(match.group(2))
+
+                return int(match.group(1))
+
+            except:
+                continue
+
+    return 0
+
+
+def get_batch_videos(batch_dir):
+
+    video_files = []
+
+    thumb_path = find_batch_thumbnail(batch_dir)
+
+    videos = []
+
+    for f in os.listdir(batch_dir):
+
+        if f.lower().endswith(VIDEO_FORMATS):
+
+            ep = extract_episode_number(f)
+
+            videos.append((f, ep))
+
+    videos.sort(key=lambda x: x[1])
+
+    for video, ep_num in videos:
+
+        video_path = os.path.join(batch_dir, video)
+
+        video_files.append({
+            "video_path": video_path,
+            "thumb_path": thumb_path,
+            "filename": video,
+            "episode": ep_num,
+            "size": os.path.getsize(video_path)
+        })
+
+        logger.info(f"Found video {video} episode {ep_num}")
+
+    return video_files
+
+def natural_sort_key(s):
+    return [
+        int(text) if text.isdigit() else text.lower()
+        for text in re.split(r'([0-9]+)', s)
+    ]
+
+def get_batch_directories():
+
+    batch_dirs = []
+
+    if not os.path.exists(BATCH_DIR):
+        return batch_dirs
+
+    has_root_videos = False
+
+    for item in sorted(os.listdir(BATCH_DIR), key=natural_sort_key):
+
+        full = os.path.join(BATCH_DIR, item)
+
+        if os.path.isdir(full):
+            batch_dirs.append(full)
+
+        elif item.lower().endswith(VIDEO_FORMATS):
+            has_root_videos = True
+
+    if has_root_videos:
+        batch_dirs.insert(0, BATCH_DIR)
+
+    return batch_dirs