#!/usr/bin/env python3 """Sync one Drive folder's direct PDF children to Markdown; run one instance at a time.""" import argparse import hashlib import json import os from pathlib import Path import re import sys import tempfile import requests DRIVE_FILES = "https://www.googleapis.com/drive/v3/files" ENDPOINT = "https://api.parsebridge.com/v1/parse/document" SCOPES = ["https://www.googleapis.com/auth/drive.readonly"] MAX_BYTES = 50 * 1024 * 1024 def atomic_write(path, text): """Replace a file only after writing its complete contents; mode is private.""" with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=path.parent, delete=False) as handle: temporary = Path(handle.name) try: handle.write(text) handle.flush() os.fsync(handle.fileno()) except BaseException: temporary.unlink(missing_ok=True) raise try: temporary.replace(path) finally: temporary.unlink(missing_ok=True) def safe_id(value): if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_-]+", value): raise ValueError("Invalid Drive ID") return value def drive_session(client_path, token_path, authorize=False): """Obtain desktop OAuth credentials, then save the private refresh token locally.""" from google.auth.transport.requests import AuthorizedSession, Request from google.auth.exceptions import RefreshError from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow credentials = None if not authorize and token_path.exists(): try: credentials = Credentials.from_authorized_user_file(str(token_path), SCOPES) except (ValueError, OSError): raise ValueError("Invalid Google token file; rerun locally with --authorize") from None if authorize: credentials = InstalledAppFlow.from_client_secrets_file( str(client_path), SCOPES ).run_local_server(port=0, prompt="consent") if not credentials.refresh_token: raise ValueError("Google did not grant a refresh token; repeat consent with --authorize") elif not credentials or not credentials.valid: if credentials and credentials.expired and credentials.refresh_token: try: credentials.refresh(Request()) except RefreshError: raise ValueError("Google token refresh failed; rerun locally with --authorize") from None else: raise ValueError("No usable Google token; run locally with --authorize") token_path.parent.mkdir(parents=True, exist_ok=True) atomic_write(token_path, credentials.to_json()) return AuthorizedSession(credentials) def list_pdfs(drive, folder_id): """Collect every page before converting or allowing absent-file pruning.""" response = drive.get(f"{DRIVE_FILES}/{safe_id(folder_id)}", params={"fields": "id,mimeType,trashed,driveId", "supportsAllDrives": "true"}, timeout=(15, 60)) response.raise_for_status() folder = response.json() if (folder.get("mimeType") != "application/vnd.google-apps.folder" or folder.get("trashed") or folder.get("driveId")): raise ValueError("Source must be an accessible, non-trashed My Drive folder") params = { "q": f"'{safe_id(folder_id)}' in parents and mimeType='application/pdf' and trashed=false", "fields": "nextPageToken,incompleteSearch,files(id,name,md5Checksum,version,modifiedTime,size,capabilities(canDownload))", "pageSize": 1000, "corpora": "user", "spaces": "drive", } files = [] while True: response = drive.get(DRIVE_FILES, params=params, timeout=(15, 60)) response.raise_for_status() page = response.json() if page.get("incompleteSearch"): raise ValueError("Incomplete Drive listing; no files changed") files.extend(page.get("files", [])) if not page.get("nextPageToken"): return files params["pageToken"] = page["nextPageToken"] def download_pdf(drive, file_id): """Bound download memory to the API's file-size limit.""" with drive.get(f"{DRIVE_FILES}/{safe_id(file_id)}", params={"alt": "media"}, stream=True, timeout=(15, 120)) as response: response.raise_for_status() chunks = [] size = 0 for chunk in response.iter_content(chunk_size=1024 * 1024): size += len(chunk) if size > MAX_BYTES: raise ValueError("PDF exceeds 50 MB") chunks.append(chunk) return b"".join(chunks) def parse_pdf(session, api_key, name, data): """Send once; never automatically retry a potentially chargeable conversion.""" response = session.post( ENDPOINT, headers={"Authorization": f"Bearer {api_key}"}, files={"file": (name, data, "application/pdf")}, timeout=(15, 600), ) if response.status_code != 200: raise ValueError(f"Parsebridge HTTP {response.status_code}") result = response.json() if (not isinstance(result, dict) or result.get("success") is not True or not isinstance(result.get("content"), str)): raise ValueError("Invalid Parsebridge response") return result["content"] def sync_folder(drive, parse_session, folder_id, output_dir, api_key, prune=False): """Stable Drive-ID paths update in place; only successful conversions enter state.""" safe_id(folder_id) output_dir.mkdir(parents=True, exist_ok=True) state_path = output_dir / ".drive-state.json" state = json.loads(state_path.read_text()) if state_path.exists() else { "folder": folder_id, "files": {} } if state.get("folder") != folder_id or not isinstance(state.get("files"), dict): raise ValueError("Use a separate output directory for each Drive folder") files = list_pdfs(drive, folder_id) present = {safe_id(item["id"]) for item in files} failed = False for item in files: file_id = item["id"] try: target = output_dir / f"{file_id}.md" fingerprint = item.get("md5Checksum") or item.get("version") previous = state["files"].get(file_id, {}) entry = {"fingerprint": fingerprint, "sha256": previous.get("sha256"), "name": item["name"], "modifiedTime": item.get("modifiedTime")} if fingerprint and previous.get("fingerprint") == fingerprint and target.is_file(): if entry != previous: state["files"][file_id] = entry atomic_write(state_path, json.dumps(state, indent=2) + "\n") print(f"Skipped {file_id}") continue if item.get("capabilities", {}).get("canDownload") is False: raise ValueError("Google Drive disallows download of this file") if int(item.get("size", 0)) > MAX_BYTES: raise ValueError("PDF exceeds 50 MB") data = download_pdf(drive, file_id) if item.get("md5Checksum") and hashlib.md5(data).hexdigest() != item["md5Checksum"]: raise ValueError("Source changed during download; run again after it settles") digest = hashlib.sha256(data).hexdigest() if previous.get("sha256") != digest or not target.is_file(): content = parse_pdf(parse_session, api_key, item["name"], data) atomic_write(target, content) print(f"Converted {file_id}") entry["sha256"] = digest state["files"][file_id] = entry atomic_write(state_path, json.dumps(state, indent=2) + "\n") except Exception as error: failed = True detail = str(error) if isinstance(error, ValueError) and not isinstance(error, requests.RequestException) else type(error).__name__ print(f"Failed {file_id}: {detail}", file=sys.stderr) if prune: for file_id in list(state["files"]): if file_id not in present: (output_dir / f"{safe_id(file_id)}.md").unlink(missing_ok=True) del state["files"][file_id] atomic_write(state_path, json.dumps(state, indent=2) + "\n") return 1 if failed else 0 def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--folder-id", required=True) parser.add_argument("--output", required=True, type=Path) parser.add_argument("--credentials", type=Path, default=Path("credentials.json")) parser.add_argument("--token", type=Path, default=Path(".drive-token.json")) parser.add_argument("--authorize", action="store_true", help="Obtain a token interactively on a computer with a browser") parser.add_argument("--prune", action="store_true", help="Remove only previously managed, now absent PDFs") args = parser.parse_args() api_key = os.environ.get("PARSEBRIDGE_API_KEY") if not api_key: parser.error("Set PARSEBRIDGE_API_KEY") try: with drive_session(args.credentials, args.token, args.authorize) as drive, requests.Session() as session: return sync_folder(drive, session, args.folder_id, args.output, api_key, args.prune) except Exception as error: detail = str(error) if isinstance(error, ValueError) and not isinstance(error, requests.RequestException) else type(error).__name__ print(f"Stopped: {detail}", file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())