direct2drive colab

direct2drive-colab.py

# Install
!pip install -q ipywidgets requests
from google.colab import output
output.enable_custom_widget_manager()

# Mount
from google.colab import drive
drive.mount('/content/drive')

# Imports
import os, subprocess, threading, queue, uuid, time, re
import requests
import ipywidgets as widgets
from IPython.display import display

# Paths
COLAB_ROOT = "/content/drive/MyDrive/Downloads"
DRIVE_BASE = "/content/drive/MyDrive/Downloads"
os.makedirs(DRIVE_BASE, exist_ok=True)

# aria2
!apt-get -y install aria2

# Filename
def get_filename(url, fallback):
    try:
        r = requests.head(url, allow_redirects=True, timeout=10)
        cd = r.headers.get("content-disposition")
        if cd and "filename=" in cd:
            return cd.split("filename=")[-1].strip('"')
    except:
        pass
    name = url.split("/")[-1].split("?")[0]
    return name if name else fallback

# Task
class DownloadTask:
    def __init__(self, url):
        self.id = str(uuid.uuid4())[:8]
        self.url = url
        self.status = "waiting"
        self.progress = 0

task_queue = queue.Queue()
active_tasks = []

# Download
def download_with_aria2(url, filename, task):
    path = os.path.join(COLAB_ROOT, filename)
    cmd = [
        "aria2c","-x","16","-s","16",
        "--summary-interval=1",
        "--console-log-level=notice",
        "--allow-overwrite=true",
        "-o", filename,"-d", COLAB_ROOT,url
    ]
    process = subprocess.Popen(cmd,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,universal_newlines=True)
    for line in process.stdout:
        match = re.search(r'\((\d+)%\)', line)
        if match:
            task.progress = int(match.group(1))
            task.status = "downloading"
    process.wait()
    if process.returncode == 0:
        task.progress = 100
        task.status = "done"
        return path
    task.status = "failed"
    return None

# Worker
def worker():
    while True:
        task = task_queue.get()
        if task is None: break
        task.status = "starting"
        task.progress = 0
        try:
            filename = get_filename(task.url, "file")
            if "." not in filename: filename += ".bin"
            download_with_aria2(task.url, filename, task)
        except:
            task.status = "failed"
        task_queue.task_done()

for _ in range(4):
    threading.Thread(target=worker, daemon=True).start()

# UI
url_input = widgets.Textarea(
    placeholder="Paste links (one per line)",
    layout=widgets.Layout(width='100%', height='120px')
)

start_btn = widgets.Button(description="Start", button_style='success')
ui_box = widgets.VBox()

def build_ui():
    items = []
    for t in active_tasks:
        bar = widgets.IntProgress(
            value=t.progress,
            min=0,
            max=100,
            description=t.status,
            layout=widgets.Layout(width='70%')
        )
        label = widgets.Label(t.url)
        items.append(widgets.HBox([label, bar]))
    ui_box.children = items

def ui_updater():
    while True:
        time.sleep(1)
        build_ui()

threading.Thread(target=ui_updater, daemon=True).start()

def start_click(b):
    links = url_input.value.strip().split("\n")
    for l in links:
        if l.strip():
            t = DownloadTask(l.strip())
            active_tasks.append(t)
            task_queue.put(t)
    build_ui()

start_btn.on_click(start_click)

display(url_input, start_btn, ui_box)