Spaces:
Sleeping
Sleeping
| from fastapi import FastAPI | |
| import gspread | |
| from google.oauth2.service_account import Credentials | |
| from google.auth.exceptions import GoogleAuthError | |
| import os | |
| from typing import List, Dict | |
| app = FastAPI() | |
| def get_credentials(): | |
| """Get Google Sheets API credentials from environment variables.""" | |
| try: | |
| service_account_info = { | |
| "type": os.getenv("SERVICE_ACCOUNT_TYPE"), | |
| "project_id": os.getenv("PROJECT_ID"), | |
| "private_key_id": os.getenv("PRIVATE_KEY_ID"), | |
| "private_key": os.getenv("PRIVATE_KEY").replace('\\n', '\n'), | |
| "client_email": os.getenv("CLIENT_EMAIL"), | |
| "client_id": os.getenv("CLIENT_ID"), | |
| "auth_uri": os.getenv("AUTH_URI"), | |
| "token_uri": os.getenv("TOKEN_URI"), | |
| "auth_provider_x509_cert_url": os.getenv("AUTH_PROVIDER_X509_CERT_URL"), | |
| "client_x509_cert_url": os.getenv("CLIENT_X509_CERT_URL"), | |
| "universe_domain": os.getenv("UNIVERSE_DOMAIN") | |
| } | |
| scope = ['https://spreadsheets.google.com/feeds', 'https://www.googleapis.com/auth/drive'] | |
| creds = Credentials.from_service_account_info(service_account_info, scopes=scope) | |
| return creds | |
| except Exception as e: | |
| print(f"Error getting credentials: {e}") | |
| return None | |
| def get_productivity() -> Dict[str, List[int]]: | |
| """Endpoint to fetch productivity data from Google Sheets.""" | |
| try: | |
| creds = get_credentials() | |
| if creds is None: | |
| raise Exception("Failed to obtain credentials.") | |
| # Authorize the client | |
| client = gspread.authorize(creds) | |
| # Open the Google Sheet | |
| sheet = client.open_by_url('https://docs.google.com/spreadsheets/d/1EFf2lr4A10nt4RhIqxCD_fxe-l3sXH09II0TEkMmvhA/edit?gid=0#gid=0').worksheet('Sheet1') | |
| Yes_productivity = [0] * 11 # List to count 'yes' responses for scores 0-10 | |
| No_productivity = [0] * 11 # List to count 'no' responses for scores 0-10 | |
| # Get all values from the sheet | |
| values = sheet.get_all_values() | |
| # Iterate through each row in the sheet | |
| for row in values: | |
| last_column_value = None | |
| second_last_value = None | |
| # Iterate from the last cell to the first | |
| for cell in reversed(row): | |
| if cell: # Check if the cell is not empty | |
| if last_column_value is None: | |
| last_column_value = cell # Assign last non-empty value | |
| elif second_last_value is None: | |
| second_last_value = cell # Assign second last non-empty value | |
| break # Exit once both values are found | |
| # Check if we have valid last and second last values | |
| if last_column_value is not None and second_last_value is not None: | |
| try: | |
| last_column_value = int(last_column_value) # Convert to integer | |
| if 0 <= last_column_value <= 10: | |
| # Update the productivity lists based on second last value | |
| if second_last_value.lower() == 'yes': | |
| Yes_productivity[last_column_value] += 1 | |
| elif second_last_value.lower() == 'no': | |
| No_productivity[last_column_value] += 1 | |
| elif second_last_value.lower() == 'maybe': | |
| Yes_productivity[last_column_value] += 1 | |
| No_productivity[last_column_value] += 1 | |
| except ValueError: | |
| print(f"Invalid score '{last_column_value}' in row: {row}") | |
| # Return the results as JSON | |
| return { | |
| "Yes_Productivity": Yes_productivity, | |
| "No_Productivity": No_productivity | |
| } | |
| except GoogleAuthError as auth_error: | |
| return {"error": f"Authentication error: {auth_error}"} | |
| except Exception as e: | |
| return {"error": f"An error occurred: {e}"} | |