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 @app.get("/productivity") 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/12UQzr7xy70-MvbKUuqM6YMUF-y2kY1rumX0vOj0hKXI/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}"}