""" Video Bulk Downloader - Single File Self-Hosted Solution Run: pip install flask requests && python app.py """ import os import re import io import zipfile import requests from flask import Flask, render_template_string, request, jsonify, send_file from concurrent.futures import ThreadPoolExecutor import threading import time import uuid app = Flask(__name__) # Track download progress jobs = {} jobs_lock = threading.Lock() HTML_TEMPLATE = """ Video Downloader

Video Downloader

Paste video links below, one per line

0 / 0 completed 0%
""" def extract_id(url): """Extract video ID from RedGifs URL""" patterns = [ r'redgifs\.com/watch/([a-zA-Z]+)', r'redgifs\.com/ifr/([a-zA-Z]+)', r'^([a-zA-Z]+)$' ] for pattern in patterns: match = re.search(pattern, url.strip()) if match: return match.group(1).lower() return None def get_video_url(video_id): """Get direct video URL from RedGifs API""" headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Accept': 'application/json', } token_resp = requests.get('https://api.redgifs.com/v2/auth/temporary', headers=headers) token = token_resp.json().get('token') if not token: raise Exception("Failed to get auth token") headers['Authorization'] = f'Bearer {token}' api_url = f'https://api.redgifs.com/v2/gifs/{video_id}' resp = requests.get(api_url, headers=headers) if resp.status_code != 200: raise Exception(f"API error: {resp.status_code}") data = resp.json() urls = data.get('gif', {}).get('urls', {}) return urls.get('hd') or urls.get('sd') def download_video(url, job_id, index): """Download a single video and store in memory""" video_id = extract_id(url) with jobs_lock: jobs[job_id]['items'][index] = { 'url': url, 'status': 'pending', 'message': 'Processing...' } if not video_id: with jobs_lock: jobs[job_id]['items'][index] = { 'url': url, 'status': 'error', 'message': 'Invalid URL format' } return try: video_url = get_video_url(video_id) if not video_url: raise Exception("No video URL found") resp = requests.get(video_url) resp.raise_for_status() with jobs_lock: jobs[job_id]['videos'][video_id] = resp.content jobs[job_id]['items'][index] = { 'url': url, 'status': 'success', 'message': f'Ready: {video_id}.mp4' } except Exception as e: with jobs_lock: jobs[job_id]['items'][index] = { 'url': url, 'status': 'error', 'message': str(e) } @app.route('/') def index(): return render_template_string(HTML_TEMPLATE) @app.route('/download', methods=['POST']) def download(): data = request.json links = data.get('links', []) job_id = str(uuid.uuid4()) with jobs_lock: jobs[job_id] = { 'items': [{ 'url': link, 'status': 'pending', 'message': 'Queued...' } for link in links], 'videos': {}, 'complete': False } def run_downloads(): with ThreadPoolExecutor(max_workers=3) as executor: futures = [] for i, link in enumerate(links): futures.append(executor.submit(download_video, link, job_id, i)) for f in futures: f.result() with jobs_lock: jobs[job_id]['complete'] = True threading.Thread(target=run_downloads, daemon=True).start() return jsonify({'job_id': job_id}) @app.route('/status/') def status(job_id): with jobs_lock: job = jobs.get(job_id, {'items': [], 'complete': True, 'videos': {}}) success_count = len(job.get('videos', {})) return jsonify({ 'items': job['items'], 'complete': job['complete'], 'success_count': success_count }) @app.route('/get-zip/') def get_zip(job_id): with jobs_lock: job = jobs.get(job_id) if not job or not job['videos']: return "No videos to download", 404 zip_buffer = io.BytesIO() with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zf: for video_id, content in job['videos'].items(): zf.writestr(f'{video_id}.mp4', content) # Clean up job after download del jobs[job_id] zip_buffer.seek(0) return send_file( zip_buffer, mimetype='application/zip', as_attachment=True, download_name='redgifs_download.zip' ) if __name__ == '__main__': port = int(os.environ.get('PORT', 7860)) print(f"\n Video Downloader running at http://localhost:{port}\n") app.run(host='0.0.0.0', port=port, debug=False)