#!/usr/bin/env python3 """ FileArk Manual Cloud Transfer Google Drive <-> Microsoft OneDrive, through a local staging directory. Copyright (c) 2026 FileArk SPDX-License-Identifier: MIT This program is intentionally copy-only. It contains no source-delete operation. See https://www.fileark.net/en/blog/manual-cloud-transfer-python-script """ from __future__ import annotations import argparse import hashlib import json import logging import mimetypes import os import shutil import sys import tempfile import time from abc import ABC, abstractmethod from dataclasses import asdict, dataclass from pathlib import Path, PurePosixPath from typing import Any, Iterable, Iterator from urllib.parse import quote VERSION = "1.0.0" MIB = 1024 * 1024 GIB = 1024 * MIB GOOGLE_FOLDER = "application/vnd.google-apps.folder" GOOGLE_EXPORTS = { "application/vnd.google-apps.document": ( "application/vnd.openxmlformats-officedocument.wordprocessingml.document", ".docx", ), "application/vnd.google-apps.spreadsheet": ( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ".xlsx", ), "application/vnd.google-apps.presentation": ( "application/vnd.openxmlformats-officedocument.presentationml.presentation", ".pptx", ), "application/vnd.google-apps.drawing": ("image/png", ".png"), "application/vnd.google-apps.script": ( "application/vnd.google-apps.script+json", ".json", ), } LOG = logging.getLogger("fileark-transfer") @dataclass(frozen=True) class RemoteFile: id: str path: str size: int mime_type: str provider_hash: str | None = None export_mime_type: str | None = None @dataclass(frozen=True) class UploadedFile: id: str path: str size: int provider_hash: str | None = None @dataclass(frozen=True) class Quota: total: int | None used: int | None remaining: int | None @dataclass(frozen=True) class TransferConfig: source: str destination: str destination_root: str staging_dir: Path state_file: Path batch_bytes: int batch_files: int local_reserve_bytes: int destination_reserve_bytes: int chunk_bytes: int verification: str dry_run: bool def human_bytes(value: int | None) -> str: if value is None: return "unknown" amount = float(value) for unit in ("B", "KiB", "MiB", "GiB", "TiB", "PiB"): if amount < 1024 or unit == "PiB": return f"{amount:.1f} {unit}" amount /= 1024 return f"{amount:.1f} PiB" def sha256_file(path: Path, chunk_size: int = 8 * MIB) -> str: digest = hashlib.sha256() with path.open("rb") as handle: while chunk := handle.read(chunk_size): digest.update(chunk) return digest.hexdigest() def safe_segment(value: str) -> str: """Make one cloud name safe for a local POSIX/Windows staging path.""" cleaned = "".join( "_" if ord(character) < 32 or character in '<>:"|?*' else character for character in value.replace("/", "／").replace("\\", "＼").replace("\x00", "") ) cleaned = cleaned.rstrip(" .") reserved = {"CON", "PRN", "AUX", "NUL"} reserved.update(f"COM{number}" for number in range(1, 10)) reserved.update(f"LPT{number}" for number in range(1, 10)) if cleaned.upper() in reserved: cleaned = f"_{cleaned}" return cleaned or "_unnamed" def normalized_relative_path(value: str) -> str: parts = [safe_segment(part) for part in PurePosixPath(value).parts if part not in ("", ".")] if not parts or ".." in parts: raise ValueError(f"Unsafe relative path: {value!r}") return "/".join(parts) def destination_path(root: str, relative_path: str) -> str: root = root.strip("/") relative_path = relative_path.strip("/") return f"{root}/{relative_path}" if root else relative_path def ensure_local_capacity(directory: Path, required: int, reserve: int) -> None: directory.mkdir(parents=True, exist_ok=True) free = shutil.disk_usage(directory).free needed = required + reserve if free < needed: raise RuntimeError( f"Local staging disk is too small: {human_bytes(free)} free, " f"{human_bytes(needed)} required including reserve." ) def ensure_destination_capacity(quota: Quota, required: int, reserve: int) -> None: if quota.remaining is None: LOG.warning( "Destination did not report a finite quota. Capacity will be checked " "again by the provider during each upload." ) return needed = required + reserve if quota.remaining < needed: raise RuntimeError( f"Destination is too small: {human_bytes(quota.remaining)} free, " f"{human_bytes(needed)} required including reserve." ) def batches( files: Iterable[RemoteFile], max_bytes: int, max_files: int, ) -> Iterator[list[RemoteFile]]: batch: list[RemoteFile] = [] batch_size = 0 for item in files: planned_size = max(item.size, 1) if batch and (len(batch) >= max_files or batch_size + planned_size > max_bytes): yield batch batch = [] batch_size = 0 batch.append(item) batch_size += planned_size if batch: yield batch def load_state(path: Path, config: TransferConfig) -> dict[str, Any]: if not path.exists(): return { "version": 1, "source": config.source, "destination": config.destination, "destination_root": config.destination_root, "completed": {}, } state = json.loads(path.read_text(encoding="utf-8")) expected = (config.source, config.destination, config.destination_root) actual = ( state.get("source"), state.get("destination"), state.get("destination_root"), ) if actual != expected: raise RuntimeError( "State file belongs to a different transfer. Choose another " "--state-file or restore the original arguments." ) state.setdefault("completed", {}) return state def save_state(path: Path, state: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(f"{path.suffix}.tmp") temporary.write_text(json.dumps(state, indent=2, sort_keys=True), encoding="utf-8") temporary.replace(path) class CloudProvider(ABC): name: str def __init__(self, config: TransferConfig) -> None: self.config = config @abstractmethod def inventory(self) -> list[RemoteFile]: raise NotImplementedError @abstractmethod def quota(self) -> Quota: raise NotImplementedError @abstractmethod def download(self, item: RemoteFile, target: Path) -> None: raise NotImplementedError @abstractmethod def upload(self, source: Path, relative_path: str) -> UploadedFile: raise NotImplementedError @abstractmethod def redownload(self, uploaded: UploadedFile, target: Path) -> None: raise NotImplementedError class GoogleDriveProvider(CloudProvider): name = "google" scopes = ["https://www.googleapis.com/auth/drive"] def __init__( self, config: TransferConfig, client_secret_file: Path, token_file: Path, ) -> None: super().__init__(config) try: from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build except ImportError as error: raise RuntimeError( "Google dependencies are missing. Install the packages shown in " "the guide before running a real transfer." ) from error credentials = None if token_file.exists(): credentials = Credentials.from_authorized_user_file( str(token_file), self.scopes, ) if credentials and credentials.expired and credentials.refresh_token: credentials.refresh(Request()) if not credentials or not credentials.valid: if not client_secret_file.exists(): raise RuntimeError( f"Google OAuth desktop client JSON not found: {client_secret_file}" ) flow = InstalledAppFlow.from_client_secrets_file( str(client_secret_file), self.scopes, ) credentials = flow.run_local_server(port=0) token_file.parent.mkdir(parents=True, exist_ok=True) token_file.write_text(credentials.to_json(), encoding="utf-8") self.drive = build( "drive", "v3", credentials=credentials, cache_discovery=False, ) self._folder_cache: dict[str, str] = {"": "root"} @staticmethod def _query_literal(value: str) -> str: return value.replace("\\", "\\\\").replace("'", "\\'") def _children(self, folder_id: str) -> Iterator[dict[str, Any]]: page_token = None while True: response = ( self.drive.files() .list( q=f"'{folder_id}' in parents and trashed = false", spaces="drive", pageSize=1000, pageToken=page_token, fields=( "nextPageToken,files(" "id,name,mimeType,size,md5Checksum,modifiedTime," "capabilities(canDownload))" ), ) .execute(num_retries=5) ) yield from response.get("files", []) page_token = response.get("nextPageToken") if not page_token: return def inventory(self) -> list[RemoteFile]: found: list[RemoteFile] = [] seen_paths: set[str] = set() def walk(folder_id: str, parent_path: str) -> None: for raw in self._children(folder_id): path = normalized_relative_path( f"{parent_path}/{raw['name']}" if parent_path else raw["name"] ) if raw["mimeType"] == GOOGLE_FOLDER: walk(raw["id"], path) continue if not raw.get("capabilities", {}).get("canDownload", True): LOG.warning("Skipping non-downloadable Google Drive item: %s", path) continue export = GOOGLE_EXPORTS.get(raw["mimeType"]) if raw["mimeType"].startswith("application/vnd.google-apps.") and not export: LOG.warning( "Skipping unsupported Google-native item %s (%s)", path, raw["mimeType"], ) continue export_mime = None if export: export_mime, extension = export if not path.lower().endswith(extension): path += extension if path in seen_paths: stem, suffix = os.path.splitext(path) path = f"{stem}__{raw['id'][:8]}{suffix}" seen_paths.add(path) found.append( RemoteFile( id=raw["id"], path=path, size=int(raw.get("size") or 0), mime_type=raw["mimeType"], provider_hash=raw.get("md5Checksum"), export_mime_type=export_mime, ) ) walk("root", "") return sorted(found, key=lambda item: item.path.casefold()) def quota(self) -> Quota: raw = ( self.drive.about() .get(fields="storageQuota(limit,usage)") .execute(num_retries=5)["storageQuota"] ) total = int(raw["limit"]) if raw.get("limit") else None used = int(raw["usage"]) if raw.get("usage") else None remaining = max(total - used, 0) if total is not None and used is not None else None return Quota(total=total, used=used, remaining=remaining) def download(self, item: RemoteFile, target: Path) -> None: from googleapiclient.http import MediaIoBaseDownload target.parent.mkdir(parents=True, exist_ok=True) request = ( self.drive.files().export_media( fileId=item.id, mimeType=item.export_mime_type, ) if item.export_mime_type else self.drive.files().get_media(fileId=item.id) ) with target.open("wb") as handle: downloader = MediaIoBaseDownload(handle, request, chunksize=self.config.chunk_bytes) done = False while not done: status, done = downloader.next_chunk(num_retries=5) if status: LOG.debug("Google download %s: %.1f%%", item.path, status.progress() * 100) def _find_folder(self, parent_id: str, name: str) -> str | None: escaped = self._query_literal(name) response = ( self.drive.files() .list( q=( f"'{parent_id}' in parents and name = '{escaped}' and " f"mimeType = '{GOOGLE_FOLDER}' and trashed = false" ), spaces="drive", pageSize=2, fields="files(id,name)", ) .execute(num_retries=5) ) files = response.get("files", []) if len(files) > 1: raise RuntimeError( f"Google Drive has duplicate destination folders named {name!r}; " "choose a unique --destination-root." ) return files[0]["id"] if files else None def _ensure_folder_path(self, folder_path: str) -> str: folder_path = folder_path.strip("/") if folder_path in self._folder_cache: return self._folder_cache[folder_path] current_id = "root" current_path = "" for name in PurePosixPath(folder_path).parts: current_path = f"{current_path}/{name}".strip("/") if current_path in self._folder_cache: current_id = self._folder_cache[current_path] continue existing = self._find_folder(current_id, name) if existing: current_id = existing else: created = ( self.drive.files() .create( body={ "name": name, "mimeType": GOOGLE_FOLDER, "parents": [current_id], }, fields="id", ) .execute(num_retries=5) ) current_id = created["id"] self._folder_cache[current_path] = current_id return current_id def upload(self, source: Path, relative_path: str) -> UploadedFile: from googleapiclient.http import MediaFileUpload full_path = destination_path(self.config.destination_root, relative_path) parent_path = str(PurePosixPath(full_path).parent) parent_path = "" if parent_path == "." else parent_path name = PurePosixPath(full_path).name parent_id = self._ensure_folder_path(parent_path) escaped = self._query_literal(name) conflict = ( self.drive.files() .list( q=f"'{parent_id}' in parents and name = '{escaped}' and trashed = false", spaces="drive", pageSize=1, fields="files(id)", ) .execute(num_retries=5) .get("files", []) ) if conflict: raise RuntimeError( f"Destination already contains {full_path!r}. " "The script will not overwrite it." ) media = MediaFileUpload( str(source), mimetype=mimetypes.guess_type(name)[0] or "application/octet-stream", chunksize=self.config.chunk_bytes, resumable=True, ) request = self.drive.files().create( body={"name": name, "parents": [parent_id]}, media_body=media, fields="id,name,size,md5Checksum", ) response = None while response is None: status, response = request.next_chunk(num_retries=5) if status: LOG.debug("Google upload %s: %.1f%%", full_path, status.progress() * 100) return UploadedFile( id=response["id"], path=full_path, size=int(response.get("size") or source.stat().st_size), provider_hash=response.get("md5Checksum"), ) def redownload(self, uploaded: UploadedFile, target: Path) -> None: from googleapiclient.http import MediaIoBaseDownload target.parent.mkdir(parents=True, exist_ok=True) request = self.drive.files().get_media(fileId=uploaded.id) with target.open("wb") as handle: downloader = MediaIoBaseDownload(handle, request, chunksize=self.config.chunk_bytes) done = False while not done: _, done = downloader.next_chunk(num_retries=5) class OneDriveProvider(CloudProvider): name = "onedrive" graph = "https://graph.microsoft.com/v1.0" scopes = ["Files.ReadWrite", "offline_access"] def __init__( self, config: TransferConfig, client_id: str, tenant: str, token_cache_file: Path, ) -> None: super().__init__(config) try: import msal import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry except ImportError as error: raise RuntimeError( "Microsoft dependencies are missing. Install the packages shown " "in the guide before running a real transfer." ) from error if not client_id: raise RuntimeError("--microsoft-client-id is required for OneDrive.") cache = msal.SerializableTokenCache() if token_cache_file.exists(): cache.deserialize(token_cache_file.read_text(encoding="utf-8")) app = msal.PublicClientApplication( client_id=client_id, authority=f"https://login.microsoftonline.com/{tenant}", token_cache=cache, ) accounts = app.get_accounts() result = app.acquire_token_silent(self.scopes, account=accounts[0]) if accounts else None if not result: flow = app.initiate_device_flow(scopes=self.scopes) if "user_code" not in flow: raise RuntimeError(f"Unable to start Microsoft device login: {flow}") print(flow["message"], file=sys.stderr) result = app.acquire_token_by_device_flow(flow) if "access_token" not in result: raise RuntimeError( "Microsoft login failed: " f"{result.get('error_description') or result.get('error')}" ) if cache.has_state_changed: token_cache_file.parent.mkdir(parents=True, exist_ok=True) token_cache_file.write_text(cache.serialize(), encoding="utf-8") self.requests = requests self.session = requests.Session() retry = Retry( total=6, backoff_factor=1, status_forcelist=(429, 500, 502, 503, 504), allowed_methods=frozenset(("GET", "POST", "PUT")), respect_retry_after_header=True, ) self.session.mount("https://", HTTPAdapter(max_retries=retry)) self.session.headers.update( { "Authorization": f"Bearer {result['access_token']}", "Accept": "application/json", } ) self._known_folders: set[str] = {""} def _request( self, method: str, endpoint: str, *, expected: tuple[int, ...] = (200,), **kwargs: Any, ) -> Any: url = endpoint if endpoint.startswith("https://") else f"{self.graph}{endpoint}" response = self.session.request(method, url, timeout=120, **kwargs) if response.status_code not in expected: detail = response.text[:1000] raise RuntimeError( f"Microsoft Graph {method} {endpoint} returned " f"{response.status_code}: {detail}" ) if response.status_code == 204 or not response.content: return None return response.json() def _children(self, item_id: str | None) -> Iterator[dict[str, Any]]: endpoint = ( "/me/drive/root/children" if item_id is None else f"/me/drive/items/{quote(item_id, safe='')}/children" ) params = { "$top": "200", "$select": ( "id,name,size,folder,file,package," "parentReference,@microsoft.graph.downloadUrl" ), } while endpoint: response = self._request("GET", endpoint, params=params) params = None yield from response.get("value", []) endpoint = response.get("@odata.nextLink") def inventory(self) -> list[RemoteFile]: found: list[RemoteFile] = [] seen_paths: set[str] = set() def walk(item_id: str | None, parent_path: str) -> None: for raw in self._children(item_id): path = normalized_relative_path( f"{parent_path}/{raw['name']}" if parent_path else raw["name"] ) if raw.get("folder") is not None: walk(raw["id"], path) continue if raw.get("file") is None: LOG.warning("Skipping unsupported OneDrive package: %s", path) continue if path in seen_paths: stem, suffix = os.path.splitext(path) path = f"{stem}__{raw['id'][:8]}{suffix}" seen_paths.add(path) hashes = raw.get("file", {}).get("hashes", {}) provider_hash = hashes.get("sha1Hash") or hashes.get("quickXorHash") found.append( RemoteFile( id=raw["id"], path=path, size=int(raw.get("size") or 0), mime_type=raw.get("file", {}).get("mimeType") or "application/octet-stream", provider_hash=provider_hash, ) ) walk(None, "") return sorted(found, key=lambda item: item.path.casefold()) def quota(self) -> Quota: raw = self._request("GET", "/me/drive", params={"$select": "quota"})["quota"] return Quota( total=int(raw["total"]) if raw.get("total") is not None else None, used=int(raw["used"]) if raw.get("used") is not None else None, remaining=( int(raw["remaining"]) if raw.get("remaining") is not None else None ), ) def download(self, item: RemoteFile, target: Path) -> None: target.parent.mkdir(parents=True, exist_ok=True) response = self.session.get( f"{self.graph}/me/drive/items/{quote(item.id, safe='')}/content", stream=True, timeout=300, ) response.raise_for_status() with target.open("wb") as handle: for chunk in response.iter_content(chunk_size=self.config.chunk_bytes): if chunk: handle.write(chunk) def _get_path(self, folder_path: str) -> dict[str, Any] | None: if not folder_path: return {"id": "root"} encoded = quote(folder_path.strip("/"), safe="/") response = self.session.get( f"{self.graph}/me/drive/root:/{encoded}", params={"$select": "id,name,folder"}, timeout=120, ) if response.status_code == 404: return None if response.status_code != 200: raise RuntimeError( f"Microsoft Graph could not inspect {folder_path!r}: " f"{response.status_code} {response.text[:500]}" ) return response.json() def _ensure_folder_path(self, folder_path: str) -> None: folder_path = folder_path.strip("/") if folder_path in self._known_folders: return current = "" for name in PurePosixPath(folder_path).parts: parent = current current = f"{current}/{name}".strip("/") if current in self._known_folders: continue existing = self._get_path(current) if existing: if existing.get("folder") is None: raise RuntimeError( f"OneDrive destination path {current!r} is a file, not a folder." ) self._known_folders.add(current) continue endpoint = ( "/me/drive/root/children" if not parent else f"/me/drive/root:/{quote(parent, safe='/')}:/children" ) self._request( "POST", endpoint, expected=(201,), json={ "name": name, "folder": {}, "@microsoft.graph.conflictBehavior": "fail", }, ) self._known_folders.add(current) def upload(self, source: Path, relative_path: str) -> UploadedFile: full_path = destination_path(self.config.destination_root, relative_path) parent = str(PurePosixPath(full_path).parent) self._ensure_folder_path("" if parent == "." else parent) encoded = quote(full_path, safe="/") total = source.stat().st_size if total == 0: response = self.session.put( f"{self.graph}/me/drive/root:/{encoded}:/content", data=b"", headers={"Content-Type": "application/octet-stream"}, timeout=120, ) if response.status_code not in (200, 201): raise RuntimeError( f"OneDrive zero-byte upload failed for {full_path!r}: " f"{response.status_code} {response.text[:1000]}" ) uploaded = response.json() return UploadedFile( id=uploaded["id"], path=full_path, size=int(uploaded.get("size") or 0), ) session = self._request( "POST", f"/me/drive/root:/{encoded}:/createUploadSession", expected=(200,), json={ "item": { "@microsoft.graph.conflictBehavior": "fail", "name": PurePosixPath(full_path).name, } }, ) upload_url = session["uploadUrl"] uploaded: dict[str, Any] | None = None with source.open("rb") as handle: start = 0 while start < total: data = handle.read(self.config.chunk_bytes) end = start + len(data) - 1 for attempt in range(1, 7): response = self.requests.put( upload_url, data=data, headers={ "Content-Length": str(len(data)), "Content-Range": f"bytes {start}-{end}/{total}", }, timeout=300, ) if response.status_code in (200, 201): uploaded = response.json() break if response.status_code == 202: break if response.status_code in (429, 500, 502, 503, 504) and attempt < 6: time.sleep(int(response.headers.get("Retry-After", 2**attempt))) continue raise RuntimeError( f"OneDrive upload failed for {full_path!r}: " f"{response.status_code} {response.text[:1000]}" ) start = end + 1 LOG.debug( "OneDrive upload %s: %.1f%%", full_path, (start / total * 100) if total else 100, ) if uploaded is None: uploaded = self._request( "GET", f"/me/drive/root:/{encoded}", params={"$select": "id,name,size,file"}, ) hashes = uploaded.get("file", {}).get("hashes", {}) return UploadedFile( id=uploaded["id"], path=full_path, size=int(uploaded.get("size") or total), provider_hash=hashes.get("sha1Hash") or hashes.get("quickXorHash"), ) def redownload(self, uploaded: UploadedFile, target: Path) -> None: target.parent.mkdir(parents=True, exist_ok=True) response = self.session.get( f"{self.graph}/me/drive/items/{quote(uploaded.id, safe='')}/content", stream=True, timeout=300, ) response.raise_for_status() with target.open("wb") as handle: for chunk in response.iter_content(chunk_size=self.config.chunk_bytes): if chunk: handle.write(chunk) class TransferEngine: def __init__( self, source: CloudProvider, destination: CloudProvider, config: TransferConfig, ) -> None: self.source = source self.destination = destination self.config = config self.state = load_state(config.state_file, config) def run(self) -> None: LOG.info("Safety mode: COPY ONLY. Source deletion is not implemented.") LOG.info("Scanning %s...", self.source.name) inventory = self.source.inventory() completed = self.state["completed"] for item in inventory: prior = completed.get(item.path) if prior and ( prior.get("source_id") != item.id or int(prior.get("source_size", prior.get("bytes", -1))) != item.size ): raise RuntimeError( f"Source item {item.path!r} changed after it was recorded in the " "checkpoint. Use a new --state-file and destination root to " "avoid silently mixing transfer versions." ) pending = [item for item in inventory if item.path not in completed] known_bytes = sum(item.size for item in pending) unknown_count = sum(1 for item in pending if item.size == 0) LOG.info( "Found %d files (%s known size); %d already completed; %d export/empty files.", len(inventory), human_bytes(sum(item.size for item in inventory)), len(completed), unknown_count, ) quota = self.destination.quota() LOG.info( "Destination quota: %s free of %s.", human_bytes(quota.remaining), human_bytes(quota.total), ) ensure_destination_capacity( quota, known_bytes, self.config.destination_reserve_bytes, ) ensure_local_capacity( self.config.staging_dir, min(known_bytes, self.config.batch_bytes), self.config.local_reserve_bytes, ) if self.config.dry_run: LOG.info("Dry run complete. No files were downloaded or uploaded.") return if not pending: LOG.info("Nothing to transfer.") return for batch_number, batch in enumerate( batches(pending, self.config.batch_bytes, self.config.batch_files), start=1, ): planned = sum(item.size for item in batch) LOG.info( "Batch %d: %d files, %s known size.", batch_number, len(batch), human_bytes(planned), ) ensure_local_capacity( self.config.staging_dir, planned, self.config.local_reserve_bytes, ) ensure_destination_capacity( self.destination.quota(), planned, self.config.destination_reserve_bytes, ) for index, item in enumerate(batch, start=1): self._transfer_one(item, batch_number, index, len(batch)) LOG.info( "Transfer complete: %d files recorded in %s.", len(self.state["completed"]), self.config.state_file, ) LOG.info("Source files were not deleted.") def _transfer_one( self, item: RemoteFile, batch_number: int, index: int, batch_count: int, ) -> None: relative = normalized_relative_path(item.path) local = self.config.staging_dir / Path(*PurePosixPath(relative).parts) verification_copy = local.with_name(f".{local.name}.destination-check") LOG.info( "[batch %d %d/%d] Downloading %s (%s).", batch_number, index, batch_count, relative, human_bytes(item.size), ) ensure_local_capacity( self.config.staging_dir, item.size, self.config.local_reserve_bytes, ) self.source.download(item, local) actual_size = local.stat().st_size local_hash = sha256_file(local) if item.size and actual_size != item.size: raise RuntimeError( f"Source size mismatch for {relative!r}: expected {item.size}, " f"downloaded {actual_size}." ) ensure_destination_capacity( self.destination.quota(), actual_size, self.config.destination_reserve_bytes, ) LOG.info( "[batch %d %d/%d] Uploading %s (%s).", batch_number, index, batch_count, relative, human_bytes(actual_size), ) uploaded = self.destination.upload(local, relative) if uploaded.size != actual_size: raise RuntimeError( f"Destination size mismatch for {relative!r}: uploaded metadata " f"reports {uploaded.size}, local file is {actual_size}." ) verification_hash = None if self.config.verification == "redownload": ensure_local_capacity( self.config.staging_dir, actual_size, self.config.local_reserve_bytes, ) LOG.info( "[batch %d %d/%d] Re-downloading destination for SHA-256 verification.", batch_number, index, batch_count, ) self.destination.redownload(uploaded, verification_copy) verification_hash = sha256_file(verification_copy) if verification_hash != local_hash: raise RuntimeError( f"SHA-256 mismatch after destination re-download: {relative!r}." ) self.state["completed"][relative] = { "source_id": item.id, "source_size": item.size, "destination_id": uploaded.id, "destination_path": uploaded.path, "bytes": actual_size, "sha256": local_hash, "destination_sha256": verification_hash, "completed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), } save_state(self.config.state_file, self.state) verification_copy.unlink(missing_ok=True) local.unlink(missing_ok=True) self._prune_empty_parents(local.parent) LOG.info( "[batch %d %d/%d] VERIFIED %s.", batch_number, index, batch_count, relative, ) def _prune_empty_parents(self, start: Path) -> None: staging = self.config.staging_dir.resolve() current = start while current.exists() and current.resolve() != staging: try: current.rmdir() except OSError: break current = current.parent def run_demo() -> None: sample = [ RemoteFile("g_01", "Documents/Board packet.pdf", 18 * MIB, "application/pdf"), RemoteFile("g_02", "Finance/Forecast.xlsx", 7 * MIB, "application/vnd.ms-excel"), RemoteFile("g_03", "Video/Launch-film.mp4", 2_650 * MIB, "video/mp4"), RemoteFile("g_04", "Notes/Planning.docx", 2 * MIB, "application/msword"), ] print("FileArk Manual Cloud Transfer 1.0.0 — DEMO (no network calls)") print("Safety mode: COPY ONLY. Source deletion is not implemented.") print("Source: Google Drive Destination: OneDrive") print(f"Inventory: {len(sample):,} files, {human_bytes(sum(x.size for x in sample))}") print("Destination: 112.4 GiB free; reserve: 10.0 GiB PASS") print("Local staging: 76.8 GiB free; reserve: 10.0 GiB PASS") for number, item in enumerate(sample[:3], start=1): print(f"[batch 1 {number}/3] download {item.path:<34} {human_bytes(item.size):>10}") print(f"[batch 1 {number}/3] upload {item.path:<34} resumable") print(f"[batch 1 {number}/3] verify size + destination SHA-256 VERIFIED") print("Checkpoint: 3 files safely recorded in .fileark-transfer-state.json") print("Source files deleted: 0") def parser() -> argparse.ArgumentParser: result = argparse.ArgumentParser( description=( "Copy files between Google Drive and OneDrive through bounded local " "batches. Never deletes source files." ), formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) result.add_argument("--version", action="version", version=f"%(prog)s {VERSION}") result.add_argument("--source", choices=("google", "onedrive")) result.add_argument("--destination", choices=("google", "onedrive")) result.add_argument( "--destination-root", default="FileArk Manual Transfer", help="Folder created in the destination cloud.", ) result.add_argument( "--staging-dir", type=Path, default=Path("./.fileark-staging"), ) result.add_argument( "--state-file", type=Path, default=Path("./.fileark-transfer-state.json"), ) result.add_argument("--batch-gib", type=float, default=10.0) result.add_argument("--batch-files", type=int, default=250) result.add_argument("--local-reserve-gib", type=float, default=10.0) result.add_argument("--destination-reserve-gib", type=float, default=5.0) result.add_argument( "--chunk-mib", type=float, default=10.0, help="Must be a multiple of both Google 256 KiB and OneDrive 320 KiB.", ) result.add_argument( "--verification", choices=("metadata", "redownload"), default="redownload", help="Redownload performs an exact destination SHA-256 comparison.", ) result.add_argument( "--google-client-secret", type=Path, default=Path(os.getenv("GOOGLE_OAUTH_CLIENT_SECRET_FILE", "client_secret.json")), ) result.add_argument( "--google-token-file", type=Path, default=Path(os.getenv("GOOGLE_TOKEN_FILE", ".fileark-google-token.json")), ) result.add_argument( "--microsoft-client-id", default=os.getenv("MICROSOFT_CLIENT_ID", ""), ) result.add_argument( "--microsoft-tenant", default=os.getenv("MICROSOFT_TENANT", "common"), ) result.add_argument( "--microsoft-token-cache", type=Path, default=Path(os.getenv("MICROSOFT_TOKEN_CACHE", ".fileark-ms-token-cache.json")), ) result.add_argument("--dry-run", action="store_true") result.add_argument("--demo", action="store_true") result.add_argument("--verbose", action="store_true") return result def validate_arguments(args: argparse.Namespace) -> TransferConfig: if not args.source or not args.destination: raise ValueError("--source and --destination are required unless --demo is used.") if args.source == args.destination: raise ValueError("Source and destination must be different providers.") if args.batch_gib <= 0 or args.batch_files <= 0: raise ValueError("Batch size and file count must be positive.") chunk_bytes = int(args.chunk_mib * MIB) if chunk_bytes <= 0 or chunk_bytes % (256 * 1024) or chunk_bytes % (320 * 1024): raise ValueError("--chunk-mib must be a multiple of 1.25 MiB.") return TransferConfig( source=args.source, destination=args.destination, destination_root=args.destination_root, staging_dir=args.staging_dir.expanduser().resolve(), state_file=args.state_file.expanduser().resolve(), batch_bytes=int(args.batch_gib * GIB), batch_files=args.batch_files, local_reserve_bytes=int(args.local_reserve_gib * GIB), destination_reserve_bytes=int(args.destination_reserve_gib * GIB), chunk_bytes=chunk_bytes, verification=args.verification, dry_run=args.dry_run, ) def make_provider( name: str, config: TransferConfig, args: argparse.Namespace, ) -> CloudProvider: if name == "google": return GoogleDriveProvider( config, args.google_client_secret.expanduser().resolve(), args.google_token_file.expanduser().resolve(), ) return OneDriveProvider( config, args.microsoft_client_id, args.microsoft_tenant, args.microsoft_token_cache.expanduser().resolve(), ) def main() -> int: args = parser().parse_args() logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)-7s %(message)s", datefmt="%H:%M:%S", ) if args.demo: run_demo() return 0 try: config = validate_arguments(args) LOG.info("Configuration: %s", json.dumps(asdict(config), default=str)) source = make_provider(config.source, config, args) destination = make_provider(config.destination, config, args) TransferEngine(source, destination, config).run() return 0 except (KeyboardInterrupt, EOFError): LOG.error("Interrupted. Re-run the same command to continue from the state file.") return 130 except Exception as error: # noqa: BLE001 - command-line error boundary LOG.error("%s", error) if args.verbose: LOG.exception("Detailed failure") return 1 if __name__ == "__main__": raise SystemExit(main())