172 lines
7.4 KiB
Python
172 lines
7.4 KiB
Python
import asyncio
|
|
import os
|
|
from dotenv import load_dotenv
|
|
import aiohttp
|
|
from telegram import Bot
|
|
from telegram.error import TelegramError
|
|
from telegram.ext import Application
|
|
|
|
async def download_single_pack(bot: Bot, session: aiohttp.ClientSession, sticker_url: str, allowed_types: list):
|
|
"""
|
|
Asynchronously downloads sticker files based on user-selected types or an 'auto' mode.
|
|
"""
|
|
master_download_dir = "downloads"
|
|
try:
|
|
if 't.me/addstickers/' not in sticker_url:
|
|
print(f"--> Invalid URL format: {sticker_url}. Skipping.")
|
|
return
|
|
|
|
pack_name = sticker_url.split('/')[-1]
|
|
print(f"\n--- Processing pack: {pack_name} ---")
|
|
|
|
pack_path = os.path.join(master_download_dir, pack_name)
|
|
os.makedirs(pack_path, exist_ok=True)
|
|
print(f"Saving to directory: ./{pack_path}/")
|
|
|
|
sticker_set = await bot.get_sticker_set(pack_name)
|
|
total_stickers = len(sticker_set.stickers)
|
|
|
|
mode_text = "auto" if "auto" in allowed_types else ', '.join(allowed_types)
|
|
print(f"Found {total_stickers} stickers. Mode: {mode_text}")
|
|
|
|
download_count = 0
|
|
skipped_count = 0
|
|
for i, sticker in enumerate(sticker_set.stickers):
|
|
|
|
# GEWIJZIGD: Hoofdlogica gesplitst voor 'auto' en specifieke modi
|
|
if 'auto' in allowed_types:
|
|
# --- AUTO MODE LOGICA ---
|
|
file_to_download = None
|
|
file_name = None
|
|
|
|
if sticker.is_video:
|
|
# Voor video-stickers, pak de originele .webm
|
|
file_to_download = await bot.get_file(sticker.file_id)
|
|
file_name = f"{i+1:03d}.webm"
|
|
elif sticker.is_animated:
|
|
# Voor geanimeerde stickers (.tgs), pak de .png thumbnail
|
|
if sticker.thumbnail:
|
|
file_to_download = await bot.get_file(sticker.thumbnail.file_id)
|
|
file_name = f"{i+1:03d}_thumb.png"
|
|
else:
|
|
# Voor statische stickers, pak de originele .webp
|
|
file_to_download = await bot.get_file(sticker.file_id)
|
|
file_name = f"{i+1:03d}.webp"
|
|
|
|
if file_to_download and file_name:
|
|
file_path = os.path.join(pack_path, file_name)
|
|
async with session.get(file_to_download.file_path) as response:
|
|
if response.status == 200:
|
|
content = await response.read()
|
|
with open(file_path, 'wb') as f: f.write(content)
|
|
download_count += 1
|
|
else:
|
|
skipped_count += 1
|
|
|
|
else:
|
|
# --- SPECIFIEKE TYPES MODE LOGICA (DE OUDE CODE) ---
|
|
was_downloaded_this_run = False
|
|
|
|
# 1. Check voor primaire sticker (webp, tgs, webm)
|
|
primary_extension = ".webp"
|
|
if sticker.is_animated: primary_extension = ".tgs"
|
|
elif sticker.is_video: primary_extension = ".webm"
|
|
|
|
if primary_extension.strip('.') in allowed_types:
|
|
file = await bot.get_file(sticker.file_id)
|
|
file_path = os.path.join(pack_path, f"{i+1:03d}{primary_extension}")
|
|
async with session.get(file.file_path) as response:
|
|
if response.status == 200:
|
|
with open(file_path, 'wb') as f: f.write(await response.read())
|
|
was_downloaded_this_run = True
|
|
|
|
# 2. Check voor PNG thumbnail
|
|
if 'png' in allowed_types and sticker.thumbnail:
|
|
thumb = await bot.get_file(sticker.thumbnail.file_id)
|
|
thumb_path = os.path.join(pack_path, f"{i+1:03d}_thumb.png")
|
|
async with session.get(thumb.file_path) as response:
|
|
if response.status == 200:
|
|
with open(thumb_path, 'wb') as f: f.write(await response.read())
|
|
was_downloaded_this_run = True
|
|
|
|
if was_downloaded_this_run:
|
|
download_count += 1
|
|
else:
|
|
skipped_count += 1
|
|
|
|
print(f"\rProcessing... (sticker {i+1}/{total_stickers}, downloaded: {download_count}, skipped: {skipped_count})", end="", flush=True)
|
|
|
|
print(f"\n✅ Download complete for pack: {pack_name}. Total files downloaded: {download_count}")
|
|
|
|
except TelegramError as e:
|
|
print(f"\nError for pack '{pack_name}': {e}")
|
|
except aiohttp.ClientError as e:
|
|
print(f"\nNetwork error while downloading for '{pack_name}': {e}")
|
|
except Exception as e:
|
|
print(f"\nAn unexpected error occurred with pack '{pack_name}': {e}")
|
|
|
|
async def main():
|
|
""" Main function to run the downloader. """
|
|
print("--- Telegram Sticker Pack Downloader ---")
|
|
|
|
load_dotenv()
|
|
bot_token = os.getenv("TELEGRAM_BOT_TOKEN")
|
|
|
|
if not bot_token:
|
|
print("\n[ERROR] Telegram Bot Token niet gevonden!")
|
|
print("Maak een '.env' bestand aan en voeg 'TELEGRAM_BOT_TOKEN=jouw_token' toe.")
|
|
return
|
|
|
|
application = Application.builder().token(bot_token).build()
|
|
bot = application.bot
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
try:
|
|
bot_user = await bot.get_me()
|
|
print(f"Bot '{bot_user.first_name}' initialized successfully.")
|
|
except Exception as e:
|
|
print(f"Error initializing bot. Is je token in .env correct? Details: {e}")
|
|
return
|
|
|
|
# GEWIJZIGD: 'auto' toegevoegd als optie
|
|
print("\nAvailable file types: webp, tgs, webm, png")
|
|
type_input = input("Enter desired file types (comma-separated), 'all', or 'auto': ").strip().lower()
|
|
|
|
if type_input == 'auto':
|
|
allowed_types = ['auto']
|
|
print("--> Auto mode selected. Choosing best format per sticker.")
|
|
elif not type_input or type_input == 'all':
|
|
allowed_types = ['webp', 'tgs', 'webm', 'png']
|
|
print("--> Downloading all available types.")
|
|
else:
|
|
allowed_types = [t.strip() for t in type_input.split(',')]
|
|
print(f"--> Selected types: {', '.join(allowed_types)}")
|
|
|
|
print("\nChoose an option:")
|
|
print("1: Download a single sticker pack from a URL")
|
|
print("2: Download multiple packs from a 'urls.txt' file")
|
|
|
|
choice = input("Enter your choice (1 or 2): ").strip()
|
|
|
|
if choice == '1':
|
|
sticker_url = input("Enter the Sticker Pack URL: ").strip()
|
|
if sticker_url:
|
|
await download_single_pack(bot, session, sticker_url, allowed_types)
|
|
elif choice == '2':
|
|
try:
|
|
with open('urls.txt', 'r') as f: urls = [line.strip() for line in f if line.strip()]
|
|
if not urls: print("'urls.txt' is empty."); return
|
|
print(f"\nFound {len(urls)} URLs. Starting batch download...")
|
|
for url in urls:
|
|
await download_single_pack(bot, session, url, allowed_types)
|
|
except FileNotFoundError:
|
|
print("\nError: 'urls.txt' not found.")
|
|
except Exception as e:
|
|
print(f"An error occurred while reading the file: {e}")
|
|
else:
|
|
print("Invalid choice.")
|
|
|
|
print("\n--- Script finished ---")
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(main()) |