Spaces:
Sleeping
Sleeping
import os | |
import logging | |
from google.oauth2.credentials import Credentials | |
from google_auth_oauthlib.flow import InstalledAppFlow | |
from google.auth.transport.requests import Request | |
from googleapiclient.discovery import build | |
from datetime import datetime | |
# Setup logging | |
logging.basicConfig( | |
level=logging.INFO, | |
format='%(asctime)s - %(levelname)s - %(message)s', | |
handlers=[ | |
logging.StreamHandler(), | |
logging.FileHandler(f'drive_structure_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log') | |
] | |
) | |
def authenticate_google(): | |
"""Authenticate with Google Drive""" | |
# Update scopes to match what's being returned | |
SCOPES = [ | |
'https://www.googleapis.com/auth/drive.readonly', | |
'https://www.googleapis.com/auth/drive.file', | |
'https://www.googleapis.com/auth/drive.install', | |
'https://www.googleapis.com/auth/userinfo.email', | |
'https://www.googleapis.com/auth/userinfo.profile', | |
'https://www.googleapis.com/auth/gmail.readonly', | |
'openid' | |
] | |
creds = None | |
# Delete existing token to force new authentication | |
if os.path.exists('token.json'): | |
os.remove('token.json') | |
# Client configuration | |
CLIENT_CONFIG = { | |
"installed": { | |
"client_id": "483287191355-udtleajik8ko1o2n03fqmimuu47n3hba.apps.googleusercontent.com", | |
"client_secret": "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", | |
"auth_uri": "https://accounts.google.com/o/oauth2/auth", | |
"token_uri": "https://oauth2.googleapis.com/token", | |
"auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs", | |
"redirect_uris": ["http://localhost:8080/"] | |
} | |
} | |
try: | |
flow = InstalledAppFlow.from_client_config( | |
CLIENT_CONFIG, | |
SCOPES | |
) | |
creds = flow.run_local_server(port=8080) | |
# Save new credentials | |
with open('token.json', 'w') as token: | |
token.write(creds.to_json()) | |
return creds | |
except Exception as e: | |
logging.error(f"Authentication error: {str(e)}") | |
raise | |
def list_drive_contents(): | |
"""List all folders and files in Google Drive""" | |
try: | |
# Authenticate and build service | |
creds = authenticate_google() | |
service = build('drive', 'v3', credentials=creds) | |
def get_files_in_folder(folder_id='root', prefix=''): | |
"""Recursively list contents of folders""" | |
query = f"'{folder_id}' in parents and trashed=false" | |
try: | |
results = service.files().list( | |
q=query, | |
pageSize=1000, | |
fields="nextPageToken, files(id, name, mimeType)" | |
).execute() | |
items = results.get('files', []) | |
for item in items: | |
if item['mimeType'] == 'application/vnd.google-apps.folder': | |
# It's a folder | |
logging.info(f"{prefix}π {item['name']}/") | |
# Recursively list contents of this folder | |
get_files_in_folder(item['id'], prefix + " ") | |
else: | |
# It's a file | |
if item['name'].endswith('.csv'): | |
logging.info(f"{prefix}π {item['name']} (CSV)") | |
else: | |
logging.info(f"{prefix}π {item['name']}") | |
except Exception as e: | |
logging.error(f"Error listing contents of folder: {str(e)}") | |
# Start listing from root | |
logging.info("\nGoogle Drive Structure:") | |
logging.info("======================") | |
get_files_in_folder() | |
logging.info("======================") | |
except Exception as e: | |
logging.error(f"Error accessing Drive: {str(e)}") | |
if __name__ == "__main__": | |
list_drive_contents() | |