#!/usr/bin/env python3 """Yandex.Disk grabber: helper program. Copies ALL your Yandex.Disk files to any folder (internal disk, external drive, network share). 1. Open https://yandex.rit-m.ru, sign in with Yandex and press "Программа-загрузчик": the page shows a one-time code. 2. Run: python3 grab.py --code ABCD2345 --to "/Volumes/MyDrive/yandex" (Windows: py grab.py --code ABCD2345 --to "E:\\yandex") What it does: lists every file, downloads them in parallel, skips files that are already complete, resumes broken downloads and checks every file against the md5 that Yandex reports. Needs only Python 3.8+ (no packages). Re-run the same command at any time: it continues where it stopped. Read-only for your Disk: nothing is changed or deleted there. """ import argparse import hashlib import json import os import re import sys import threading import time import urllib.error import urllib.parse import urllib.request from concurrent.futures import ThreadPoolExecutor, as_completed DEFAULT_SERVER = 'https://yandex.rit-m.ru' UA = 'yandex-disk-grabber/1.0' lock = threading.Lock() stats = dict(done=0, skipped=0, failed=0, bytes=0) failed = [] direct = {'ok': True} def api(server, token, path, body=None, timeout=60): req = urllib.request.Request(server + path, data=json.dumps(body).encode() if body is not None else None, headers={'User-Agent': UA, 'Content-Type': 'application/json', **({'Authorization': 'Bearer ' + token} if token else {})}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read()) def human(n): n = float(n) for u in ('Б', 'КБ', 'МБ', 'ГБ', 'ТБ'): if n < 1024 or u == 'ТБ': return f'{n:.0f} {u}' if u == 'Б' else f'{n:.1f} {u}' n /= 1024 def safe_part(p): p = re.sub(r'[<>:"|?*\x00-\x1f]', '_', p).strip().rstrip('. ') return p if p not in ('', '.', '..') else '_' def local_path(root, disk_path): parts = [safe_part(x) for x in disk_path[len('disk:/'):].split('/') if x] return os.path.join(root, *parts) def md5_of(path): h = hashlib.md5() with open(path, 'rb') as f: for chunk in iter(lambda: f.read(1 << 20), b''): h.update(chunk) return h.hexdigest() def fetch_one(server, token, item, root, tries=6): dst = local_path(root, item['path']) size = int(item.get('size') or 0) if os.path.isfile(dst) and os.path.getsize(dst) == size: with lock: stats['skipped'] += 1 return os.makedirs(os.path.dirname(dst), exist_ok=True) part = dst + '.part' err = '' for attempt in range(1, tries + 1): try: have = os.path.getsize(part) if os.path.isfile(part) else 0 if have > size: os.remove(part) have = 0 if have < size or size == 0: rng = {'Range': f'bytes={have}-'} if have else {} r = None if direct['ok']: # straight from Yandex storage: faster than through our server (abroad, throttled) try: href = api(server, token, '/api/dl-link?' + urllib.parse.urlencode({'path': item['path']}))['href'] r = urllib.request.urlopen(urllib.request.Request(href, headers={'User-Agent': UA, **rng}), timeout=120) except urllib.error.HTTPError as e: if e.url.startswith(server): # our own server refused (session, access): handled below raise raise IOError(f'Яндекс ответил {e.code}') # the storage itself failed for this file: retry, keep direct mode except Exception: # noqa: BLE001 - cannot reach Yandex storage from here: use our server direct['ok'] = False if r is None: r = urllib.request.urlopen(urllib.request.Request(server + '/api/dl?' + urllib.parse.urlencode({'path': item['path']}), headers={'User-Agent': UA, 'Authorization': 'Bearer ' + token, **rng}), timeout=120) with r: if have and r.status != 206: have = 0 with open(part, 'ab' if have else 'wb') as f: while True: chunk = r.read(1 << 20) if not chunk: break f.write(chunk) with lock: stats['bytes'] += len(chunk) if os.path.getsize(part) != size: raise IOError(f'size {os.path.getsize(part)} != {size}') if item.get('md5') and md5_of(part) != item['md5']: os.remove(part) raise IOError('md5 mismatch') os.replace(part, dst) with lock: stats['done'] += 1 return except urllib.error.HTTPError as e: if e.code == 402: raise SystemExit('Нужен доступ: пробный лимит закончился. Оплатите 500 ₽ или введите код на сайте, затем запустите команду снова.') if e.code in (401, 403): raise SystemExit('Сессия истекла или код неверный: получите новый код на сайте.') err = f'HTTP {e.code}' except Exception as e: # noqa: BLE001 err = str(e)[:120] time.sleep(min(30, 2 ** attempt)) with lock: stats['failed'] += 1 failed.append((item['path'], err)) def main(): ap = argparse.ArgumentParser(description='Скачать все файлы Яндекс.Диска в выбранную папку') ap.add_argument('--code', required=True, help='одноразовый код со страницы сайта') ap.add_argument('--to', default='yandex-disk', help='куда сохранять (папка на компьютере, внешнем диске, сетевой папке)') ap.add_argument('--server', default=DEFAULT_SERVER) ap.add_argument('--threads', type=int, default=4, help='параллельных загрузок (1-4)') ap.add_argument('--only', default='', help='только папка или файл, например "Фото/2020"') ap.add_argument('--tries', type=int, default=6, help='сколько раз повторять неудавшийся файл (пауза растёт до 30 с)') ap.add_argument('--list', action='store_true', help='только показать, сколько и чего будет скачано') a = ap.parse_args() server = a.server.rstrip('/') try: ex = api(server, None, '/api/cli/exchange', {'code': a.code}) except urllib.error.HTTPError: sys.exit('Код неверный или истёк (он действует 15 минут и подходит один раз). Получите новый на сайте.') token = ex['token'] me = api(server, token, '/api/me') print(f"Вход выполнен: {me.get('name') or me.get('login')}. На Диске {human(me['used'])} из {human(me['total'])}.") items, offset = [], 0 prefix = ('disk:/' + a.only.strip('/')) if a.only else '' while True: page = api(server, token, f'/api/files?offset={offset}&limit=1000') items += [i for i in page['items'] if not prefix or i['path'] == prefix or i['path'].startswith(prefix + '/')] offset += page['count'] print(f'\rСписок файлов: {offset} просмотрено, подходит {len(items)}', end='', flush=True) if page['count'] < 1000: break total = sum(int(i.get('size') or 0) for i in items) print(f'\nК загрузке: {len(items)} файлов, {human(total)}.') if a.list or not items: return root = os.path.abspath(a.to) os.makedirs(root, exist_ok=True) free = os.statvfs(root) if hasattr(os, 'statvfs') else None if free and free.f_bavail * free.f_frsize < total: print(f'ВНИМАНИЕ: свободно {human(free.f_bavail * free.f_frsize)}, нужно {human(total)}. Скачивание продолжится, пока есть место.') print(f'Папка назначения: {root}') t0 = time.time() stop = threading.Event() def progress(): while not stop.wait(1.0): with lock: d, s, f, b = stats['done'], stats['skipped'], stats['failed'], stats['bytes'] sp = b / max(1, time.time() - t0) print(f'\r{d + s}/{len(items)} файлов (пропущено готовых {s}, ошибок {f}) · {human(b)} · {human(sp)}/с ', end='', flush=True) threading.Thread(target=progress, daemon=True).start() with ThreadPoolExecutor(max_workers=max(1, min(a.threads, 4))) as pool: futs = [pool.submit(fetch_one, server, token, i, root, a.tries) for i in items] try: for _ in as_completed(futs): pass except KeyboardInterrupt: print('\nОстановлено. Запустите ту же команду снова: загрузка продолжится.') for f in futs: f.cancel() os._exit(130) stop.set() print(f"\nГотово: скачано {stats['done']}, уже было {stats['skipped']}, с ошибками {stats['failed']}. {human(stats['bytes'])} за {time.time() - t0:.0f} с.") if failed: rep = os.path.join(root, '_ошибки_загрузки.txt') with open(rep, 'w', encoding='utf-8') as f: f.write('\n'.join(f'{p}\t{e}' for p, e in failed)) print(f'Не скачалось {len(failed)} файлов, список: {rep}. Запустите команду ещё раз.') sys.exit(1) if __name__ == '__main__': main()