| import gradio as gr |
| import pandas as pd |
| import requests |
| import zipfile |
| import io |
| import re |
| from datetime import datetime |
| from pathlib import Path |
| import tempfile |
| from concurrent.futures import ThreadPoolExecutor, as_completed |
|
|
| TEMP_DIR = Path(tempfile.gettempdir()) / "img_downloader" |
| TEMP_DIR.mkdir(exist_ok=True) |
|
|
| def parse_published_time(time_str): |
| if pd.isna(time_str): |
| return datetime.min |
| time_str = str(time_str).strip().lower() |
| now = datetime.now() |
| patterns = [ |
| (r'(\d+)\s*second', lambda m: now.replace(second=max(0, now.second - int(m.group(1))))), |
| (r'(\d+)\s*minute', lambda m: now.replace(minute=max(0, now.minute - int(m.group(1))))), |
| (r'(\d+)\s*hour', lambda m: now.replace(hour=max(0, now.hour - int(m.group(1))))), |
| (r'(\d+)\s*day', lambda m: now.replace(day=max(1, now.day - int(m.group(1))))), |
| (r'(\d+)\s*week', lambda m: now.replace(day=max(1, now.day - int(m.group(1))*7))), |
| (r'(\d+)\s*month', lambda m: now.replace(month=max(1,now.month - int(m.group(1))))), |
| (r'(\d+)\s*year', lambda m: now.replace(year=max(2000, now.year- int(m.group(1))))), |
| ] |
| for pattern, handler in patterns: |
| m = re.search(pattern, time_str) |
| if m: |
| try: |
| return handler(m) |
| except Exception: |
| return datetime.min |
| return datetime.min |
|
|
| def is_animated_url(url): |
| """ |
| YouTube ggpht URLs ending in '-rwa' serve animated content (WebM/GIF). |
| '-rw-nd-v1' or '-nd-v1' serve static JPEGs. |
| """ |
| return bool(re.search(r'-rwa$', url.strip())) |
|
|
| def detect_ext_from_bytes(data): |
| h = data[:16] |
| if h[:3] == b'\xff\xd8\xff': return ".jpg" |
| if h[:8] == b'\x89PNG\r\n\x1a\n': return ".png" |
| if h[:6] in (b'GIF87a', b'GIF89a'): return ".gif" |
| if h[:4] == b'RIFF' and h[8:12] == b'WEBP': return ".webp" |
| if h[4:8] == b'ftyp': return ".mp4" |
| if h[:4] in (b'\x1aE\xdf\xa3', b'\x1aE\xbf\xa3'): return ".webm" |
| return ".jpg" |
|
|
| def load_csv(file_obj): |
| if file_obj is None: |
| return gr.update(choices=[], value=None), gr.update(visible=False), "No file uploaded.", None |
|
|
| try: |
| df = pd.read_csv(file_obj.name) |
| except Exception as e: |
| return gr.update(choices=[], value=None), gr.update(visible=False), f"❌ Error reading CSV: {e}", None |
|
|
| image_cols = [c for c in df.columns if re.match(r'^images/\d+$', c)] |
| other_cols = [c for c in df.columns if c not in image_cols] |
| all_cols = image_cols + other_cols |
|
|
| n_rows = len(df) |
| n_img_c = len(image_cols) |
| total_possible = df[image_cols].notna().sum().sum() if image_cols else 0 |
|
|
| animated_cols = [] |
| for col in image_cols: |
| sample = df[col].dropna().head(5).tolist() |
| if any(is_animated_url(str(u)) for u in sample): |
| animated_cols.append(col) |
|
|
| note = "" |
| if animated_cols: |
| |
| note = f"\n⚡ **Animated columns detected:** {', '.join(animated_cols)} — will be preserved in their original animated format." |
|
|
| summary = ( |
| f"✅ **Loaded:** {n_rows} posts · " |
| f"{n_img_c} image columns · " |
| f"~{total_possible} media links found{note}" |
| ) |
|
|
| return ( |
| gr.update(choices=all_cols, value=image_cols, visible=True), |
| gr.update(visible=True), |
| summary, |
| df.to_json(orient="split") |
| ) |
|
|
| def download_images(selected_cols, df_json, progress=gr.Progress()): |
| if not selected_cols: |
| return None, "⚠️ Please select at least one column." |
| if not df_json: |
| return None, "⚠️ Please upload a CSV first." |
|
|
| df = pd.read_json(io.StringIO(df_json), orient="split") |
|
|
| if "publishedTime" in df.columns: |
| df["_sort_dt"] = df["publishedTime"].apply(parse_published_time) |
| df = df.sort_values("_sort_dt", ascending=True).reset_index(drop=True) |
|
|
| image_cols = [c for c in selected_cols if re.match(r'^images/\d+$', c)] |
| if not image_cols: |
| return None, "⚠️ No image columns selected (columns like images/0, images/1…)." |
|
|
| tasks = [] |
| for idx, row in df.iterrows(): |
| date_label = str(row.get("publishedTime", f"post_{idx}")).strip().replace(" ", "_").replace("/","_").replace(":","") |
| post_id = str(row.get("postId", f"post_{idx}"))[:20] |
| for col in image_cols: |
| url = row.get(col) |
| if pd.notna(url) and isinstance(url, str) and url.startswith("http"): |
| |
| final_url = url.strip() |
| tasks.append((date_label, post_id, col, final_url)) |
|
|
| if not tasks: |
| return None, "⚠️ No valid media URLs found in the selected columns." |
|
|
| def fetch(args): |
| i, date_label, post_id, col, url = args |
| col_safe = col.replace("/", "_") |
| try: |
| r = requests.get(url, timeout=15, headers={"User-Agent": "Mozilla/5.0"}) |
| r.raise_for_status() |
| ext = detect_ext_from_bytes(r.content) |
| filename = f"{i+1:04d}_{date_label}_{post_id}_{col_safe}{ext}" |
| return i, filename, r.content, None |
| except Exception as e: |
| filename = f"{i+1:04d}_{date_label}_{post_id}_{col_safe}.unknown" |
| return i, filename, None, str(e) |
|
|
| zip_path = TEMP_DIR / "media_download.zip" |
| results = {} |
| completed = 0 |
|
|
| with ThreadPoolExecutor(max_workers=16) as pool: |
| futures = {pool.submit(fetch, (i,) + t): i for i, t in enumerate(tasks)} |
| for future in as_completed(futures): |
| i, filename, content, err = future.result() |
| results[i] = (filename, content, err) |
| completed += 1 |
| progress(completed / len(tasks), desc=f"Downloaded {completed}/{len(tasks)}") |
|
|
| downloaded = 0 |
| failed = 0 |
| log_lines = [] |
|
|
| with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: |
| for i in range(len(tasks)): |
| filename, content, err = results[i] |
| if content: |
| zf.writestr(f"media/{filename}", content) |
| downloaded += 1 |
| log_lines.append(f"✅ {filename}") |
| else: |
| failed += 1 |
| log_lines.append(f"❌ {filename} — {err}") |
|
|
| overview = "\n".join([ |
| "MEDIA DOWNLOAD OVERVIEW", |
| "=" * 50, |
| f"Generated : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", |
| f"Total : {len(tasks)} attempted", |
| f"Downloaded: {downloaded}", |
| f"Failed : {failed}", |
| f"Columns : {', '.join(image_cols)}", |
| "", |
| "FILES (sorted oldest → newest post):", |
| "-" * 50, |
| ] + log_lines) |
| zf.writestr("overview.txt", overview) |
|
|
| status = f"✅ Done! **{downloaded}** files downloaded, {failed} failed. ZIP is ready." |
| return str(zip_path), status |
|
|
| |
| with gr.Blocks(title="YouTube Post Media Downloader", theme=gr.themes.Soft()) as demo: |
| df_state = gr.State(None) |
|
|
| |
| gr.Markdown( |
| """ |
| # 🖼️ YouTube Post Media Downloader |
| Upload a YouTube-posts-scraper CSV, choose which columns to download, and get a ZIP. |
| Animated columns (like images/0) are automatically saved in their original animated formats. |
| """ |
| ) |
|
|
| with gr.Row(): |
| with gr.Column(scale=1): |
| file_input = gr.File(label="📂 Upload CSV", file_types=[".csv"]) |
| status_box = gr.Markdown("Upload a CSV to begin.") |
| col_select = gr.CheckboxGroup(label="Select columns to download", visible=False) |
| dl_btn = gr.Button("⬇️ Download Media → ZIP", variant="primary", visible=False) |
|
|
| with gr.Column(scale=1): |
| dl_status = gr.Markdown("") |
| dl_file = gr.File(label="📦 Download ZIP", visible=True) |
|
|
| file_input.change( |
| load_csv, |
| inputs=[file_input], |
| outputs=[col_select, dl_btn, status_box, df_state] |
| ) |
|
|
| dl_btn.click( |
| download_images, |
| inputs=[col_select, df_state], |
| outputs=[dl_file, dl_status] |
| ) |
|
|
| demo.launch(server_name="0.0.0.0", server_port=7860) |
