google-drive-upload.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "google-auth",
#     "google-auth-oauthlib",
#     "google-auth-httplib2",
#     "google-api-python-client",
# ]
# ///
#
# ============================================================================
# Google Drive File Uploader & Sharer
# ============================================================================
#
# A standalone script to upload and share files on Google Drive, supporting
# both personal drives and Shared Drives (formerly Team Drives).
#
# SETUP INSTRUCTIONS:
# -------------------
# 1. Go to https://console.cloud.google.com/
# 2. Create a new project (or select an existing one)
# 3. Enable the Google Drive API:
#    - Go to "APIs & Services" > "Library"
#    - Search for "Google Drive API" and enable it
# 4. Create OAuth 2.0 credentials:
#    - Go to "APIs & Services" > "Credentials"
#    - Click "Create Credentials" > "OAuth client ID"
#    - Choose "Desktop app" as the application type
#    - Download the JSON file and save it as "credentials.json"
#      in the same directory as this script
# 5. Configure the OAuth consent screen:
#    - Go to "APIs & Services" > "OAuth consent screen"
#    - Add your email as a test user (for "External" user type)
#
# FIRST RUN:
# ----------
# On first run, a browser window will open for OAuth authentication.
# After granting permissions, a token.json file is created for future use.
#
# UPLOAD USAGE:
# -------------
#   # Upload all PDFs from a directory:
#   ./google-drive-upload.py --folder-id YOUR_FOLDER_ID --source ./my-pdfs
#
#   # Upload specific file types:
#   ./google-drive-upload.py --folder-id YOUR_FOLDER_ID --source ./docs --pattern "*.docx"
#
#   # Upload a single file:
#   ./google-drive-upload.py --folder-id YOUR_FOLDER_ID --file ./report.pdf
#
# SHARING USAGE:
# --------------
#   # Upload and share with a user (writer access):
#   ./google-drive-upload.py --folder-id ID --file ./doc.pdf \
#       --share-user [email protected] --share-role writer
#
#   # Upload and share with multiple users:
#   ./google-drive-upload.py --folder-id ID --file ./doc.pdf \
#       --share-user [email protected] --share-user [email protected]
#
#   # Upload and share with entire domain (reader access):
#   ./google-drive-upload.py --folder-id ID --file ./doc.pdf \
#       --share-domain example.com --share-role reader
#
#   # Share existing file by ID (no upload):
#   ./google-drive-upload.py --share-file-id FILE_ID \
#       --share-user [email protected] --share-role writer
#
#   # Make file publicly accessible (anyone with link):
#   ./google-drive-upload.py --share-file-id FILE_ID --share-anyone
#
# SHARE ROLES:
# ------------
#   reader    - Can view
#   commenter - Can view and comment
#   writer    - Can edit
#
# FINDING YOUR FOLDER ID:
# -----------------------
# Open the folder in Google Drive in your browser. The URL will look like:
#   https://drive.google.com/drive/folders/1ABC123xyz...
# The folder ID is the part after "folders/": 1ABC123xyz...
#
# For Shared Drives, navigate to the folder and copy the ID from the URL.
#
# ============================================================================

import argparse
import mimetypes
import sys
from pathlib import Path

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
from googleapiclient.errors import HttpError
from googleapiclient.http import MediaFileUpload

# Full Drive access scope (required for Shared Drives)
# For more restrictive access, use: https://www.googleapis.com/auth/drive.file
SCOPES = ["https://www.googleapis.com/auth/drive"]


def get_credentials(credentials_path: Path, token_path: Path) -> Credentials:
    """Get or refresh Google Drive API credentials.

    Args:
        credentials_path: Path to the OAuth credentials JSON file
        token_path: Path to store/retrieve the user token

    Returns:
        Valid Google OAuth credentials
    """
    creds = None

    if not credentials_path.exists():
        print(f"Error: {credentials_path} not found.")
        print("Please download OAuth 2.0 credentials from Google Cloud Console")
        print("and save them in the specified location.")
        print("\nSee setup instructions at the top of this script.")
        sys.exit(1)

    # Load existing token
    if token_path.exists():
        creds = Credentials.from_authorized_user_file(str(token_path), SCOPES)

    # If no valid credentials, initiate OAuth flow
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            print("Refreshing expired credentials...")
            creds.refresh(Request())
        else:
            print("Opening browser for authentication...")
            flow = InstalledAppFlow.from_client_secrets_file(
                str(credentials_path), SCOPES
            )
            creds = flow.run_local_server(port=0)

        # Save credentials for future runs
        with open(token_path, "w") as token:
            token.write(creds.to_json())
        print(f"Credentials saved to {token_path}")

    return creds


def get_mimetype(file_path: Path) -> str:
    """Determine the MIME type of a file.

    Args:
        file_path: Path to the file

    Returns:
        MIME type string, defaults to 'application/octet-stream'
    """
    mime_type, _ = mimetypes.guess_type(str(file_path))
    return mime_type or "application/octet-stream"


def upload_file_with_id(
    service, file_path: Path, folder_id: str, update_existing: bool = True
) -> str | None:
    """Upload a file to the specified Google Drive folder.

    Args:
        service: Google Drive API service instance
        file_path: Path to the file to upload
        folder_id: Target folder ID in Google Drive
        update_existing: If True, update existing files; if False, skip them

    Returns:
        File ID if upload succeeded, None otherwise
    """
    try:
        # Check if file already exists in the folder
        query = f"name='{file_path.name}' and '{folder_id}' in parents and trashed=false"
        results = service.files().list(
            q=query,
            fields="files(id, name)",
            spaces="drive",
            includeItemsFromAllDrives=True,
            supportsAllDrives=True,
        ).execute()

        existing_files = results.get("files", [])
        mime_type = get_mimetype(file_path)
        media = MediaFileUpload(str(file_path), mimetype=mime_type, resumable=True)

        if existing_files:
            file_id = existing_files[0]["id"]
            if not update_existing:
                print(f"  Skipped (exists): {file_path.name} (ID: {file_id})")
                return file_id

            # Update existing file
            updated = service.files().update(
                fileId=file_id,
                media_body=media,
                fields="id, name",
                supportsAllDrives=True,
            ).execute()
            result_id = updated.get("id")
            print(f"  Updated: {file_path.name} (ID: {result_id})")
            return result_id
        else:
            # Create new file
            file_metadata = {
                "name": file_path.name,
                "parents": [folder_id],
            }
            created = service.files().create(
                body=file_metadata,
                media_body=media,
                fields="id, name",
                supportsAllDrives=True,
            ).execute()
            result_id = created.get("id")
            print(f"  Uploaded: {file_path.name} (ID: {result_id})")
            return result_id

    except HttpError as error:
        print(f"  Error uploading {file_path.name}: {error}")
        return None


def share_file(
    service,
    file_id: str,
    users: list[str] | None = None,
    domain: str | None = None,
    anyone: bool = False,
    role: str = "reader",
    send_notification: bool = True,
) -> list[str]:
    """Share a file with users, a domain, or make it publicly accessible.

    Args:
        service: Google Drive API service instance
        file_id: ID of the file to share
        users: List of email addresses to share with
        domain: Domain to share with (e.g., "example.com")
        anyone: If True, make the file accessible to anyone with the link
        role: Permission role - "reader", "commenter", or "writer"
        send_notification: Whether to send email notifications to users

    Returns:
        List of permission IDs that were created
    """
    permission_ids = []

    def batch_callback(request_id, response, exception):
        if exception:
            print(f"  Error sharing ({request_id}): {exception}")
        else:
            perm_id = response.get("id")
            permission_ids.append(perm_id)
            print(f"  Shared ({request_id}): permission ID {perm_id}")

    batch = service.new_batch_http_request(callback=batch_callback)

    # Share with specific users
    if users:
        for email in users:
            user_permission = {
                "type": "user",
                "role": role,
                "emailAddress": email,
            }
            batch.add(
                service.permissions().create(
                    fileId=file_id,
                    body=user_permission,
                    fields="id",
                    sendNotificationEmail=send_notification,
                    supportsAllDrives=True,
                ),
                request_id=f"user:{email}",
            )

    # Share with a domain
    if domain:
        domain_permission = {
            "type": "domain",
            "role": role,
            "domain": domain,
        }
        batch.add(
            service.permissions().create(
                fileId=file_id,
                body=domain_permission,
                fields="id",
                supportsAllDrives=True,
            ),
            request_id=f"domain:{domain}",
        )

    # Make publicly accessible (anyone with the link)
    if anyone:
        anyone_permission = {
            "type": "anyone",
            "role": role,
        }
        batch.add(
            service.permissions().create(
                fileId=file_id,
                body=anyone_permission,
                fields="id",
                supportsAllDrives=True,
            ),
            request_id="anyone",
        )

    if users or domain or anyone:
        batch.execute()

    return permission_ids


def verify_folder(service, folder_id: str) -> str | None:
    """Verify that a folder exists and is accessible.

    Args:
        service: Google Drive API service instance
        folder_id: Folder ID to verify

    Returns:
        Folder name if accessible, None otherwise
    """
    try:
        folder = service.files().get(
            fileId=folder_id,
            fields="id, name",
            supportsAllDrives=True,
        ).execute()
        return folder.get("name")
    except HttpError as error:
        if error.resp.status == 404:
            print(f"Error: Folder '{folder_id}' not found or not accessible.")
            print("\nPossible causes:")
            print("  - The folder ID is incorrect")
            print("  - The folder isn't shared with your Google account")
            print("  - You need to request access to the folder")
        else:
            print(f"Error accessing folder: {error}")
        return None


def main():
    parser = argparse.ArgumentParser(
        description="Upload and share files on Google Drive",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Upload files
  %(prog)s --folder-id ABC123 --source ./pdfs
  %(prog)s --folder-id ABC123 --file ./report.pdf

  # Upload and share with users
  %(prog)s --folder-id ABC123 --file ./doc.pdf --share-user [email protected]

  # Upload and share with domain
  %(prog)s --folder-id ABC123 --file ./doc.pdf --share-domain example.com --share-role reader

  # Share existing file (no upload)
  %(prog)s --share-file-id FILE_ID --share-user [email protected] --share-role writer

  # Make file public
  %(prog)s --share-file-id FILE_ID --share-anyone
        """,
    )

    # Upload options
    upload_group = parser.add_argument_group("Upload options")
    upload_group.add_argument(
        "--folder-id",
        help="Target Google Drive folder ID (from the folder URL)",
    )
    upload_group.add_argument(
        "--source",
        type=Path,
        help="Directory containing files to upload",
    )
    upload_group.add_argument(
        "--file",
        type=Path,
        help="Single file to upload",
    )

    upload_group.add_argument(
        "--pattern",
        default="*.pdf",
        help="Glob pattern for files to upload (default: *.pdf)",
    )
    upload_group.add_argument(
        "--no-update",
        action="store_true",
        help="Skip existing files instead of updating them",
    )

    # Sharing options
    share_group = parser.add_argument_group("Sharing options")
    share_group.add_argument(
        "--share-file-id",
        help="Share an existing file by ID (no upload, just share)",
    )
    share_group.add_argument(
        "--share-user",
        action="append",
        dest="share_users",
        metavar="EMAIL",
        help="Share with user email (can be specified multiple times)",
    )
    share_group.add_argument(
        "--share-domain",
        metavar="DOMAIN",
        help="Share with entire domain (e.g., example.com)",
    )
    share_group.add_argument(
        "--share-anyone",
        action="store_true",
        help="Make file accessible to anyone with the link",
    )
    share_group.add_argument(
        "--share-role",
        choices=["reader", "commenter", "writer"],
        default="reader",
        help="Permission role for sharing (default: reader)",
    )
    share_group.add_argument(
        "--no-notification",
        action="store_true",
        help="Don't send email notifications when sharing with users",
    )

    # Authentication options
    auth_group = parser.add_argument_group("Authentication options")
    auth_group.add_argument(
        "--credentials",
        type=Path,
        default=Path("credentials.json"),
        help="Path to OAuth credentials file (default: credentials.json)",
    )
    auth_group.add_argument(
        "--token",
        type=Path,
        default=Path("token.json"),
        help="Path to store/retrieve user token (default: token.json)",
    )

    args = parser.parse_args()

    # Determine mode: share-only or upload (optionally with sharing)
    share_only = args.share_file_id is not None
    has_sharing = args.share_users or args.share_domain or args.share_anyone

    # Validate arguments
    if share_only:
        if args.folder_id or args.source or args.file:
            parser.error("--share-file-id cannot be used with upload options")
        if not has_sharing:
            parser.error("--share-file-id requires at least one sharing option")
    else:
        if not args.folder_id:
            parser.error("--folder-id is required for uploading")
        if not args.source and not args.file:
            parser.error("--source or --file is required for uploading")

    # Authenticate and build service
    creds = get_credentials(args.credentials, args.token)
    service = build("drive", "v3", credentials=creds)

    # Mode: Share existing file only
    if share_only:
        print(f"Sharing file ID: {args.share_file_id}")
        share_file(
            service,
            args.share_file_id,
            users=args.share_users,
            domain=args.share_domain,
            anyone=args.share_anyone,
            role=args.share_role,
            send_notification=not args.no_notification,
        )
        print("\nDone!")
        return

    # Mode: Upload files (optionally with sharing)
    # Collect files to upload
    if args.file:
        if not args.file.exists():
            print(f"Error: File not found: {args.file}")
            sys.exit(1)
        files_to_upload = [args.file]
    else:
        if not args.source.exists():
            print(f"Error: Directory not found: {args.source}")
            sys.exit(1)
        files_to_upload = list(args.source.glob(args.pattern))
        if not files_to_upload:
            print(f"No files matching '{args.pattern}' found in {args.source}")
            sys.exit(1)

    # Verify folder access
    folder_name = verify_folder(service, args.folder_id)
    if not folder_name:
        sys.exit(1)

    print(f"\nTarget folder: {folder_name}")
    print(f"Files to upload: {len(files_to_upload)}")
    for f in files_to_upload:
        print(f"  - {f.name}")

    # Upload files and collect IDs for sharing
    print("\nUploading...")
    success_count = 0
    uploaded_file_ids = []

    for file_path in files_to_upload:
        file_id = upload_file_with_id(service, file_path, args.folder_id, not args.no_update)
        if file_id:
            success_count += 1
            uploaded_file_ids.append((file_path.name, file_id))

    print(f"\nUploaded: {success_count}/{len(files_to_upload)} files")

    # Share uploaded files if sharing options were provided
    if has_sharing and uploaded_file_ids:
        print("\nSharing uploaded files...")
        for file_name, file_id in uploaded_file_ids:
            print(f"\n{file_name}:")
            share_file(
                service,
                file_id,
                users=args.share_users,
                domain=args.share_domain,
                anyone=args.share_anyone,
                role=args.share_role,
                send_notification=not args.no_notification,
            )

    print("\nDone!")

    if success_count < len(files_to_upload):
        sys.exit(1)


if __name__ == "__main__":
    main()