Spaces:
Sleeping
Sleeping
File size: 4,477 Bytes
8df5b41 6f75e90 8df5b41 f363600 8df5b41 f363600 8df5b41 6f75e90 8df5b41 6f75e90 8df5b41 6f75e90 8df5b41 5448352 6f75e90 f363600 6f75e90 f363600 6f75e90 5448352 6f75e90 8df5b41 5448352 8df5b41 5448352 8df5b41 6f75e90 5448352 8df5b41 f363600 6f75e90 f363600 6f75e90 5448352 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 |
import streamlit as st
from google_auth_oauthlib.flow import Flow
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
import os
import random
from io import BytesIO
from PIL import Image
# Constants
SCOPES = ['https://www.googleapis.com/auth/drive.readonly']
CLIENT_SECRETS_FILE = 'client_secrets.json'
def create_flow():
return Flow.from_client_secrets_file(
CLIENT_SECRETS_FILE,
scopes=SCOPES,
redirect_uri=REDIRECT_URI
)
def list_folders(service):
results = service.files().list(
q="mimeType='application/vnd.google-apps.folder' and trashed=false",
spaces='drive',
fields='nextPageToken, files(id, name)',
).execute()
items = results.get('files', [])
folders = {item['id']: item['name'] for item in items}
return folders
def list_images_in_folder(service, folder_id):
query = f"parents='{folder_id}' and mimeType contains 'image/' and trashed=false"
results = service.files().list(
q=query,
spaces='drive',
fields='nextPageToken, files(id, name, mimeType)',
).execute()
items = results.get('files', [])
return items
def download_image(service, file_id):
request = service.files().get_media(fileId=file_id)
fh = BytesIO()
downloader = MediaIoBaseDownload(fh, request)
done = False
while not done:
status, done = downloader.next_chunk()
fh.seek(0)
return fh.read()
def creds_to_dict(creds):
"""Helper function to convert credentials object to a dictionary."""
return {
'token': creds.token,
'refresh_token': creds.refresh_token,
'token_uri': creds.token_uri,
'client_id': creds.client_id,
'client_secret': creds.client_secret,
'scopes': creds.scopes
}
def main():
st.title("Google Drive Image Viewer")
if 'credentials' not in st.session_state or st.session_state['credentials'] is None:
flow = create_flow()
authorization_url, _ = flow.authorization_url(
access_type='offline',
include_granted_scopes='true',
prompt='consent'
)
st.write("Please [authorize]({}) to continue.".format(authorization_url))
else:
creds = Credentials(**st.session_state['credentials'])
service = build('drive', 'v3', credentials=creds)
# List folders
folders = list_folders(service)
if folders:
folder_list = list(folders.items())
folder_name = [name for _, name in folder_list]
folder_id_list = [id for id, _ in folder_list]
selected_index = st.selectbox("Select a folder", range(len(folder_name)), format_func=lambda x: folder_name[x])
folder_id = folder_id_list[selected_index]
# Get images from the selected folder
images = list_images_in_folder(service, folder_id)
if images:
# Shuffle images
random.shuffle(images)
# Get the current image index
index = st.session_state.get('image_index', 0)
image = images[index]
img_data = download_image(service, image['id'])
image_pil = Image.open(BytesIO(img_data))
st.image(image_pil, caption=image['name'])
if st.button("Next Image"):
st.session_state['image_index'] = (index + 1) % len(images)
st.experimental_rerun()
else:
st.write("No images found in this folder.")
else:
st.write("No folders found in your Drive.")
if __name__ == '__main__':
from googleapiclient.http import MediaIoBaseDownload
# Set REDIRECT_URI
if 'REDIRECT_URI' not in st.session_state:
# Get the REDIRECT_URI from secrets
REDIRECT_URI = st.secrets["REDIRECT_URI"]
st.session_state['REDIRECT_URI'] = REDIRECT_URI
else:
REDIRECT_URI = st.session_state['REDIRECT_URI']
if 'credentials' not in st.session_state:
st.session_state['credentials'] = None
# Get query parameters
query_params = st.query_params
if 'code' in query_params:
code = query_params['code'][0]
flow = create_flow()
flow.fetch_token(code=code)
creds = flow.credentials
st.session_state['credentials'] = creds_to_dict(creds)
st.experimental_rerun()
else:
main() |