Downloading a Dropbox shared folder without paying for storage

#!/usr/bin/env python3
"""
Resumable, file-by-file downloader for a Dropbox shared folder link.

Why this exists: Dropbox's "Download folder" makes one giant zip. If the
connection hiccups, you start over. This walks the shared folder, downloads
each file individually, and resumes any interrupted file from the exact byte
it stopped at (HTTP Range). Re-run it as many times as you like -- finished
files are skipped, partial files continue.

Requires only the Python standard library. No Dropbox account storage is used.

SETUP (one time, ~2 minutes):

  1. Go to https://www.dropbox.com/developers/apps  ->  "Create app"
  2. Choose:
       API:    Scoped access
       Access: Full Dropbox
       Name:   anything unique, e.g. "kaan-folder-dl"
  3. Open the app's "Permissions" tab and tick:
       files.metadata.read
       files.content.read
       sharing.read
     Click "Submit".  (Order matters: set permissions BEFORE making a token.)
  4. Back on the "Settings" tab, find "Generated access token" -> "Generate".
     Copy the token.

  The generated token expires after ~4 hours. That is fine -- if it dies
  mid-download, generate a new one and re-run; the script picks up where it
  left off.

USAGE:

  export DROPBOX_TOKEN='sl.xxxxxxxx...'
  python3 dropbox_folder_download.py       --url 'https://www.dropbox.com/scl/fo/.../...?rlkey=...'       --dest ~/Downloads/dungeondraft

  Useful flags:
      --list-only     print the file tree and total size, download nothing
      --workers 3     download N files at a time (default 1; keep it low)
"""

import argparse
import json
import os
import queue
import ssl
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request

API_LIST = "https://api.dropboxapi.com/2/files/list_folder"
API_LIST_CONT = "https://api.dropboxapi.com/2/files/list_folder/continue"
API_GET_FILE = "https://content.dropboxapi.com/2/sharing/get_shared_link_file"

CHUNK = 1024 * 256
MAX_ATTEMPTS = 12
SSL_CTX = ssl.create_default_context()

print_lock = threading.Lock()


def log(msg):
    with print_lock:
        print(msg, flush=True)


def human(n):
    for unit in ("B", "KB", "MB", "GB", "TB"):
        if n < 1024 or unit == "TB":
            return f"{n:.1f}{unit}" if unit != "B" else f"{n}B"
        n /= 1024


# ---------------------------------------------------------------- link parsing

class DbxError(Exception):
    def __init__(self, code, detail):
        super().__init__(f"{code}: {detail}")
        self.code, self.detail = code, detail


def normalize(url):
    """Return (base_link, subpath, full_link).

    A Dropbox folder link can point at a subfolder:
        https://www.dropbox.com/scl/fo/<id>/<hash>/Sub A/Sub B?rlkey=...
    The API wants the *root* shared link plus a path inside it. We split the
    URL after the <hash> segment and keep only the rlkey query param.
    """
    parts = urllib.parse.urlsplit(url)
    segs = [s for s in parts.path.split("/") if s]

    rlkey = urllib.parse.parse_qs(parts.query).get("rlkey", [None])[0]
    if not rlkey:
        sys.exit("ERROR: the URL has no ?rlkey=... -- copy the full share link.")

    # scl/fo/<id>/<hash>[/sub/folders...]
    if len(segs) >= 4 and segs[0] == "scl" and segs[1] == "fo":
        root_segs, sub_segs = segs[:4], segs[4:]
    elif len(segs) >= 2 and segs[0] == "sh":
        root_segs, sub_segs = segs[:3], segs[3:]
    else:
        root_segs, sub_segs = segs, []

    base = urllib.parse.urlunsplit(
        (parts.scheme, parts.netloc, "/" + "/".join(root_segs), f"rlkey={rlkey}", "")
    )
    full = urllib.parse.urlunsplit(
        (parts.scheme, parts.netloc, "/" + "/".join(segs), f"rlkey={rlkey}", "")
    )
    sub = "/" + "/".join(urllib.parse.unquote(s) for s in sub_segs) if sub_segs else ""
    return base, sub, full


# ------------------------------------------------------------------- transport

def request(url, token, api_arg, body=None, extra_headers=None, timeout=60):
    headers = {"Authorization": f"Bearer {token}"}
    if body is not None:
        headers["Content-Type"] = "application/json"
    if api_arg is not None:
        headers["Dropbox-API-Arg"] = json.dumps(api_arg, ensure_ascii=True)
    if extra_headers:
        headers.update(extra_headers)
    req = urllib.request.Request(url, data=body, headers=headers, method="POST")
    return urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX)


def api_json(url, token, payload):
    body = json.dumps(payload).encode()
    for attempt in range(MAX_ATTEMPTS):
        try:
            with request(url, token, None, body=body) as r:
                return json.loads(r.read())
        except urllib.error.HTTPError as e:
            detail = e.read().decode(errors="replace")[:400]
            if e.code == 401:
                sys.exit("\nERROR 401: token is invalid or expired.\n"
                         "Generate a fresh one on your app's Settings tab and re-run.\n")
            if e.code == 429 or e.code >= 500:
                wait = int(e.headers.get("Retry-After", 2 ** attempt))
                log(f"  rate-limited/server error ({e.code}); waiting {wait}s")
                time.sleep(min(wait, 60))
                continue
            if e.code == 409:                    # path/not_found etc -- caller decides
                raise DbxError(e.code, detail)
            sys.exit(f"\nERROR {e.code} from Dropbox:\n{detail}\n")
        except (urllib.error.URLError, TimeoutError, OSError) as e:
            time.sleep(min(2 ** attempt, 30))
            if attempt == MAX_ATTEMPTS - 1:
                sys.exit(f"\nNetwork error talking to Dropbox: {e}\n")
    sys.exit("Gave up contacting Dropbox.")


# ------------------------------------------------------------------- listing

def list_one_folder(base, path, token):
    """List a single folder (one level). Dropbox forbids recursive=True on
    shared links, so we walk the tree ourselves."""
    args = {
        "path": path,
        "shared_link": {"url": base},
        "recursive": False,
        "include_non_downloadable_files": False,
    }
    page = api_json(API_LIST, token, args)
    entries = list(page.get("entries", []))
    while page.get("has_more"):
        page = api_json(API_LIST_CONT, token, {"cursor": page["cursor"]})
        entries.extend(page.get("entries", []))
    return entries


def resolve_start(base, sub, full, token):
    """Work out which (link, path) pair Dropbox actually accepts.

    Dropbox links of the form /scl/fo/<id>/<hash>/<subfolders> are inconsistent:
    sometimes <hash> already scopes the link to the subfolder (so the path must
    be ""), sometimes the link is the whole shared folder (so the path must be
    the subfolder). Try both, then fall back to descending by folder name.
    """
    attempts = [(base, sub), (full, ""), (base, "")]
    for link, path in attempts:
        if path == "" and link is base and sub:
            break                                # handled by the descent below
        try:
            list_one_folder(link, path, token)
            log(f"  resolved: path={path or '(link root)'}")
            return link, path
        except DbxError:
            continue

    # Descend from the shared-link root, matching folder names case-insensitively.
    log("  direct path rejected; walking down from the link root...")
    cur = ""
    for seg in [s for s in sub.split("/") if s]:
        entries = list_one_folder(base, cur, token)
        folders = [e for e in entries if e.get(".tag") == "folder"]
        hit = next((e for e in folders
                    if e["name"].strip().lower() == seg.strip().lower()), None)
        if not hit:
            names = "\n    ".join(e["name"] for e in entries) or "(empty)"
            sys.exit(f"\nCould not find a folder named '{seg}' inside "
                     f"'{cur or 'the shared link root'}'.\n"
                     f"  What's actually there:\n    {names}\n\n"
                     f"Pass --start-path with one of those names if the link "
                     f"already points inside the folder you want.\n")
        cur = hit.get("path_display") or f"{cur}/{hit['name']}"
        log(f"  found: {cur}")
    return base, cur


def list_files(base, sub, token):
    """Walk every folder under the shared link, breadth-first. Returns files."""
    files = []
    pending = [sub]          # "" means the shared-link root
    seen = set()

    while pending:
        folder = pending.pop(0)
        if folder in seen:
            continue
        seen.add(folder)

        for e in list_one_folder(base, folder, token):
            tag = e.get(".tag")
            # path_display is relative to the shared-link root, e.g.
            #   /3. DUNGEONDRAFT ASSETS/NEW ARTSTYLE/Walls/foo.dungeondraft_pack
            p = e.get("path_display") or (folder.rstrip("/") + "/" + e["name"])
            if tag == "file":
                files.append({"path": p, "name": e["name"], "size": e.get("size", 0)})
            elif tag == "folder":
                pending.append(p)

        log(f"  scanned {folder or '(root)'} -- {len(files)} files so far, "
            f"{len(pending)} folders queued")

    return files


# ------------------------------------------------------------------ downloading

def download(f, base, dest_root, sub, token):
    rel = f["path"]
    # Strip the subfolder prefix so we don't recreate the whole tree above it.
    trimmed = rel[len(sub):] if sub and rel.lower().startswith(sub.lower()) else rel
    out = os.path.join(dest_root, trimmed.lstrip("/"))
    part = out + ".part"
    os.makedirs(os.path.dirname(out) or ".", exist_ok=True)

    expected = f["size"]
    if os.path.exists(out) and os.path.getsize(out) == expected:
        log(f"  skip (done)  {trimmed}")
        return "skipped", expected

    for attempt in range(MAX_ATTEMPTS):
        have = os.path.getsize(part) if os.path.exists(part) else 0
        if have > expected:                      # corrupt leftover
            os.remove(part)
            have = 0
        if have == expected and expected > 0:
            os.replace(part, out)
            log(f"  ok           {trimmed}  ({human(expected)})")
            return "done", expected

        headers = {"Range": f"bytes={have}-"} if have else {}
        arg = {"url": base, "path": rel}
        try:
            with request(API_GET_FILE, token, arg, extra_headers=headers, timeout=120) as r:
                # If the server ignored our Range, restart the file cleanly.
                mode = "ab" if (have and r.status == 206) else "wb"
                if mode == "wb":
                    have = 0
                with open(part, mode) as fh:
                    while True:
                        buf = r.read(CHUNK)
                        if not buf:
                            break
                        fh.write(buf)
                        have += len(buf)
            if have >= expected or expected == 0:
                os.replace(part, out)
                log(f"  ok           {trimmed}  ({human(max(have, expected))})")
                return "done", have
            log(f"  short read   {trimmed} ({human(have)}/{human(expected)}), resuming")
        except urllib.error.HTTPError as e:
            if e.code == 401:
                sys.exit("\nERROR 401: token expired mid-download.\n"
                         "Generate a new token and re-run -- progress is kept.\n")
            if e.code == 416:                    # range past end: file is complete
                os.replace(part, out)
                return "done", expected
            if e.code == 429 or e.code >= 500:
                wait = int(e.headers.get("Retry-After", 2 ** attempt))
                log(f"  retry in {wait}s ({e.code})  {trimmed}")
                time.sleep(min(wait, 60))
                continue
            log(f"  FAILED {e.code}  {trimmed}: {e.read().decode(errors='replace')[:200]}")
            return "failed", 0
        except (urllib.error.URLError, TimeoutError, OSError, ssl.SSLError) as e:
            wait = min(2 ** attempt, 30)
            log(f"  interrupted ({type(e).__name__}); resuming in {wait}s  {trimmed}")
            time.sleep(wait)

    log(f"  FAILED after {MAX_ATTEMPTS} attempts  {trimmed}")
    return "failed", 0


# ------------------------------------------------------------------------ main

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--url", required=True, help="Dropbox shared folder link")
    ap.add_argument("--dest", required=True, help="local folder to download into")
    ap.add_argument("--token", default=os.environ.get("DROPBOX_TOKEN"))
    ap.add_argument("--list-only", action="store_true")
    ap.add_argument("--workers", type=int, default=1)
    ap.add_argument("--start-path", default=None,
                    help="override the folder inside the link to start from, "
                         "e.g. '' for the link root")
    a = ap.parse_args()

    if not a.token:
        sys.exit("No token. Set DROPBOX_TOKEN or pass --token. See the header of this file.")

    base, sub, full = normalize(a.url)
    dest = os.path.abspath(os.path.expanduser(a.dest))
    os.makedirs(dest, exist_ok=True)

    log(f"Shared link : {base}")
    log(f"Subfolder   : {sub or '(root)'}")
    log(f"Destination : {dest}\nResolving folder...")

    if a.start_path is not None:
        base, sub = base, a.start_path
    else:
        base, sub = resolve_start(base, sub, full, a.token)

    log("Listing files...")
    files = list_files(base, sub, a.token)
    if not files:
        sys.exit("No files found. Check the link is a folder link and still valid.")

    total = sum(f["size"] for f in files)
    log(f"{len(files)} files, {human(total)} total.\n")

    if a.list_only:
        for f in sorted(files, key=lambda x: x["path"]):
            log(f"  {human(f['size']):>9}  {f['path']}")
        return

    already = 0
    for f in files:
        rel = f["path"]
        trimmed = rel[len(sub):] if sub and rel.lower().startswith(sub.lower()) else rel
        out = os.path.join(dest, trimmed.lstrip("/"))
        if os.path.exists(out) and os.path.getsize(out) == f["size"]:
            already += f["size"]
    if already:
        log(f"{human(already)} already on disk; will skip those.\n")

    q = queue.Queue()
    for f in sorted(files, key=lambda x: x["path"]):
        q.put(f)

    stats = {"done": 0, "skipped": 0, "failed": 0, "bytes": 0}
    stats_lock = threading.Lock()

    def worker():
        while True:
            try:
                f = q.get_nowait()
            except queue.Empty:
                return
            try:
                status, n = download(f, base, dest, sub, a.token)
            except Exception as e:                      # never kill the run
                log(f"  ERROR {f['path']}: {e}")
                status, n = "failed", 0
            with stats_lock:
                stats[status] += 1
                stats["bytes"] += n
                pct = 100 * stats["bytes"] / total if total else 100
                if (stats["done"] + stats["skipped"] + stats["failed"]) % 10 == 0:
                    log(f"  -- {pct:.1f}% ({human(stats['bytes'])} / {human(total)})")
            q.task_done()

    threads = [threading.Thread(target=worker, daemon=True)
               for _ in range(max(1, a.workers))]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    log(f"\nFinished. {stats['done']} downloaded, {stats['skipped']} already had, "
        f"{stats['failed']} failed.")
    if stats["failed"]:
        log("Re-run the same command to retry the failures; completed files are skipped.")


if __name__ == "__main__":
    main()