prasathmani/upload-to-google-drive.py

## upload-to-google-drive.py
import json
import os
import requests

# ============================================================
# Google Drive configuration
# ============================================================

CLIENT_ID = "YOUR_CLIENT_ID"
CLIENT_SECRET = "YOUR_CLIENT_SECRET"
REFRESH_TOKEN = "YOUR_REFRESH_TOKEN"

# Google Drive folder ID
FOLDER_ID = "YOUR_GOOGLE_DRIVE_FOLDER_ID"

# Backup file to upload
UPLOAD_FILE = "/path/to/your/backup.zip"

# ============================================================
# Google OAuth / Drive URLs
# ============================================================

TOKEN_URL = "https://oauth2.googleapis.com/token"

UPLOAD_URL = (
    "https://www.googleapis.com/upload/drive/v3/files"
    "?uploadType=multipart"
)


# ============================================================
# Get a new access token using the refresh token
# ============================================================

def get_access_token():
    data = {
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "refresh_token": REFRESH_TOKEN,
        "grant_type": "refresh_token",
    }

    response = requests.post(TOKEN_URL, data=data)

    if response.status_code != 200:
        raise Exception(
            f"Failed to get access token: {response.text}"
        )

    token_data = response.json()

    access_token = token_data.get("access_token")

    if not access_token:
        raise Exception(
            f"No access token returned: {token_data}"
        )

    return access_token


# ============================================================
# Upload file to Google Drive
# ============================================================

def upload_to_drive(access_token, file_path, folder_id):

    if not os.path.isfile(file_path):
        raise FileNotFoundError(
            f"Backup file does not exist: {file_path}"
        )

    file_name = os.path.basename(file_path)

    metadata = {
        "name": file_name,
        "parents": [folder_id],
    }

    headers = {
        "Authorization": f"Bearer {access_token}",
    }

    # Keep the file open while requests.post() is running.
    # This fixes the "read of closed file" error from the original Gist.
    with open(file_path, "rb") as backup_file:

        files = {
            "data": (
                "metadata",
                json.dumps(metadata),
                "application/json; charset=UTF-8",
            ),
            "file": (
                file_name,
                backup_file,
                "application/zip",
            ),
        }

        response = requests.post(
            UPLOAD_URL,
            headers=headers,
            files=files,
            timeout=3600,
        )

    if response.status_code not in (200, 201):
        raise Exception(
            f"Google Drive upload failed "
            f"(HTTP {response.status_code}):\n"
            f"{response.text}"
        )

    result = response.json()

    print("Upload successful!")
    print(f"File name: {result.get('name')}")
    print(f"Google Drive file ID: {result.get('id')}")


# ============================================================
# Main
# ============================================================

def main():

    print("Starting Google Drive backup upload...")

    try:

        print("Getting Google access token...")
        access_token = get_access_token()

        print(f"Uploading: {UPLOAD_FILE}")

        upload_to_drive(
            access_token,
            UPLOAD_FILE,
            FOLDER_ID,
        )

        print("Backup upload completed successfully.")

    except Exception as e:

        print(f"Error: {e}")
        raise


# ============================================================
# Run
# ============================================================

if __name__ == "__main__":
    main()
	import json
	import os
	import requests

	# ============================================================
	# Google Drive configuration
	# ============================================================

	CLIENT_ID = "YOUR_CLIENT_ID"
	CLIENT_SECRET = "YOUR_CLIENT_SECRET"
	REFRESH_TOKEN = "YOUR_REFRESH_TOKEN"

	# Google Drive folder ID
	FOLDER_ID = "YOUR_GOOGLE_DRIVE_FOLDER_ID"

	# Backup file to upload
	UPLOAD_FILE = "/path/to/your/backup.zip"

	# ============================================================
	# Google OAuth / Drive URLs
	# ============================================================

	TOKEN_URL = "https://oauth2.googleapis.com/token"

	UPLOAD_URL = (
	"https://www.googleapis.com/upload/drive/v3/files"
	"?uploadType=multipart"
	)


	# ============================================================
	# Get a new access token using the refresh token
	# ============================================================

	def get_access_token():
	data = {
	"client_id": CLIENT_ID,
	"client_secret": CLIENT_SECRET,
	"refresh_token": REFRESH_TOKEN,
	"grant_type": "refresh_token",
	}

	response = requests.post(TOKEN_URL, data=data)

	if response.status_code != 200:
	raise Exception(
	f"Failed to get access token: {response.text}"
	)

	token_data = response.json()

	access_token = token_data.get("access_token")

	if not access_token:
	raise Exception(
	f"No access token returned: {token_data}"
	)

	return access_token


	# ============================================================
	# Upload file to Google Drive
	# ============================================================

	def upload_to_drive(access_token, file_path, folder_id):

	if not os.path.isfile(file_path):
	raise FileNotFoundError(
	f"Backup file does not exist: {file_path}"
	)

	file_name = os.path.basename(file_path)

	metadata = {
	"name": file_name,
	"parents": [folder_id],
	}

	headers = {
	"Authorization": f"Bearer {access_token}",
	}

	# Keep the file open while requests.post() is running.
	# This fixes the "read of closed file" error from the original Gist.
	with open(file_path, "rb") as backup_file:

	files = {
	"data": (
	"metadata",
	json.dumps(metadata),
	"application/json; charset=UTF-8",
	),
	"file": (
	file_name,
	backup_file,
	"application/zip",
	),
	}

	response = requests.post(
	UPLOAD_URL,
	headers=headers,
	files=files,
	timeout=3600,
	)

	if response.status_code not in (200, 201):
	raise Exception(
	f"Google Drive upload failed "
	f"(HTTP {response.status_code}):\n"
	f"{response.text}"
	)

	result = response.json()

	print("Upload successful!")
	print(f"File name: {result.get('name')}")
	print(f"Google Drive file ID: {result.get('id')}")


	# ============================================================
	# Main
	# ============================================================

	def main():

	print("Starting Google Drive backup upload...")

	try:

	print("Getting Google access token...")
	access_token = get_access_token()

	print(f"Uploading: {UPLOAD_FILE}")

	upload_to_drive(
	access_token,
	UPLOAD_FILE,
	FOLDER_ID,
	)

	print("Backup upload completed successfully.")

	except Exception as e:

	print(f"Error: {e}")
	raise


	# ============================================================
	# Run
	# ============================================================

	if __name__ == "__main__":
	main()
No results found