import os import re import tempfile from urllib.parse import urlparse, unquote import requests import gradio as gr from huggingface_hub import HfApi # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- HF_TOKEN = os.environ.get("HF_TOKEN") ALLOWED_EXTENSIONS = { "png", "jpg", "jpeg", "jfif", "exe", "tar", "rar", "zip", "7z", "gz", "webp", "webm", "txt", "py", "js", } MAX_BYTES = 2 * 1024 * 1024 * 1024 # 2 GB safety cap per file # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def slugify(value: str) -> str: """Turn an arbitrary string into a valid HF repo name component.""" value = value.strip().lower() value = re.sub(r"[^a-z0-9._-]+", "-", value) value = re.sub(r"-{2,}", "-", value).strip("-._") return value or "dataset" def domain_from_url(url: str) -> str: """example.com/path -> example-com""" netloc = urlparse(url).netloc netloc = netloc.split("@")[-1].split(":")[0] # drop creds / port if netloc.startswith("www."): netloc = netloc[4:] return slugify(netloc) def filename_from_url(url: str, index: int) -> str: path = urlparse(url).path name = unquote(os.path.basename(path)) if not name: name = f"file_{index}" return name def extension_of(name: str) -> str: return name.rsplit(".", 1)[-1].lower() if "." in name else "" def parse_urls(raw: str): urls = [] for line in (raw or "").splitlines(): line = line.strip() if line and not line.startswith("#"): urls.append(line) return urls # --------------------------------------------------------------------------- # Core logic # --------------------------------------------------------------------------- def push_to_hub(urls_text, dataset_name, progress=gr.Progress()): log = [] def emit(msg): log.append(msg) return "\n".join(log) if not HF_TOKEN: yield emit("❌ HF_TOKEN secret is not set. Add it in your Space settings (Settings → Variables and secrets → New secret → name `HF_TOKEN`).") return urls = parse_urls(urls_text) if len(urls) < 3: yield emit(f"❌ Please paste at least 3 direct download links (one per line). You provided {len(urls)}.") return # Validate extensions up front invalid = [] for u in urls: ext = extension_of(filename_from_url(u, 0)) if ext not in ALLOWED_EXTENSIONS: invalid.append(f"{u} → '.{ext or '?'}'") if invalid: allowed = ", ".join(sorted(ALLOWED_EXTENSIONS)) yield emit("❌ These links have unsupported file types:\n " + "\n ".join(invalid) + f"\n\nAllowed: {allowed}") return api = HfApi(token=HF_TOKEN) # Who am I try: yield emit("🔑 Authenticating with Hugging Face…") me = api.whoami() username = me["name"] yield emit(f"✅ Logged in as **{username}**") except Exception as e: yield emit(f"❌ Could not authenticate with the provided HF_TOKEN: {e}") return # Resolve dataset name name = slugify(dataset_name) if dataset_name and dataset_name.strip() else "" if not name: name = domain_from_url(urls[0]) yield emit(f"ℹ️ No dataset name given — using domain of first link → **{name}**") repo_id = f"{username}/{name}" # Create repo if it does not exist try: exists = api.repo_exists(repo_id=repo_id, repo_type="dataset") if exists: yield emit(f"📦 Found existing dataset **{repo_id}** — files will be added to it.") else: yield emit(f"📦 No dataset named **{repo_id}** found — creating it…") api.create_repo(repo_id=repo_id, repo_type="dataset", exist_ok=True) yield emit(f"✅ Created dataset **{repo_id}**") except Exception as e: yield emit(f"❌ Failed while checking/creating the dataset: {e}") return # Download + upload each file uploaded = [] seen = {} for i, url in enumerate(urls, start=1): progress((i - 1) / len(urls), desc=f"File {i}/{len(urls)}") fname = filename_from_url(url, i) # Avoid name collisions in the repo if fname in seen: seen[fname] += 1 stem, dot, ext = fname.rpartition(".") fname = f"{stem}_{seen[fname]}{dot}{ext}" if dot else f"{fname}_{seen[fname]}" else: seen[fname] = 0 try: yield emit(f"\n⬇️ [{i}/{len(urls)}] Downloading {url}") headers = {"User-Agent": "Mozilla/5.0 (compatible; HF-Dataset-Pusher/1.0)"} with requests.get(url, stream=True, timeout=120, headers=headers) as r: r.raise_for_status() total = 0 with tempfile.NamedTemporaryFile(delete=False) as tmp: tmp_path = tmp.name for chunk in r.iter_content(chunk_size=1024 * 1024): if not chunk: continue total += len(chunk) if total > MAX_BYTES: raise ValueError("file exceeds 2 GB cap") tmp.write(chunk) size_mb = total / (1024 * 1024) yield emit(f" ↳ downloaded {size_mb:.2f} MB as '{fname}'") yield emit(f"⬆️ Uploading '{fname}' to {repo_id}…") api.upload_file( path_or_fileobj=tmp_path, path_in_repo=fname, repo_id=repo_id, repo_type="dataset", commit_message=f"Add {fname}", ) uploaded.append(fname) yield emit(f"✅ Uploaded '{fname}'") except Exception as e: yield emit(f"❌ Failed on {url}: {e}") finally: try: os.remove(tmp_path) except Exception: pass progress(1.0, desc="Done") dataset_url = f"https://huggingface.co/datasets/{repo_id}" if uploaded: yield emit( f"\n🎉 Done — pushed {len(uploaded)}/{len(urls)} file(s) to the Hub.\n" f"🔗 {dataset_url}" ) else: yield emit("\n⚠️ No files were uploaded. Check the errors above.") # --------------------------------------------------------------------------- # UI # --------------------------------------------------------------------------- with gr.Blocks(title="Links → HF Dataset", theme=gr.themes.Soft(primary_hue="amber")) as demo: gr.Markdown( """ # 🤗 Links → Hugging Face Dataset Paste **direct download links** (one per line, at least 3) and they'll be pushed to the Hub as a dataset. - Provide a **dataset name**, or leave it blank to auto-name it after the **domain of the first link**. - If a dataset with that name already exists on your account, the files are added to it. Otherwise it's created. - Requires a write **`HF_TOKEN`** set as a Space secret. """ ) with gr.Row(): with gr.Column(scale=3): urls_box = gr.Textbox( label="Direct download links (one per line, 3+)", placeholder="https://example.com/data1.zip\nhttps://example.com/image.png\nhttps://example.com/notes.txt", lines=8, elem_id="urls", ) with gr.Column(scale=2): name_box = gr.Textbox( label="Dataset name (optional)", placeholder="leave blank → uses domain of first link", lines=1, elem_id="dataset-name", ) run_btn = gr.Button("🚀 Push to Hub", variant="primary") output = gr.Textbox(label="Status", lines=16, elem_id="status", show_copy_button=True) run_btn.click(fn=push_to_hub, inputs=[urls_box, name_box], outputs=output) if __name__ == "__main__": demo.queue().launch( server_name="0.0.0.0", server_port=int(os.environ.get("PORT", 7860)), ssr_mode=False, )