| import os |
| import pandas as pd |
| from google.auth.transport.requests import Request |
| from google.oauth2.credentials import Credentials |
| from google_auth_oauthlib.flow import InstalledAppFlow |
| from googleapiclient.discovery import build |
| from googleapiclient.errors import HttpError |
| from typing import List, Dict, Any, Optional |
|
|
| |
| SCOPES = ["https://www.googleapis.com/auth/spreadsheets"] |
|
|
| |
| SAMPLE_SPREADSHEET_ID = "147OAwrv_DAKJZM5fDrkslJa1pFP9T7HYlHEUKZyAFVo" |
| SAMPLE_RANGE_NAME = "Sheet1" |
|
|
| def get_credentials(): |
| """ |
| Authenticate and return Google Sheets API credentials. |
| """ |
| creds = None |
| if os.path.exists("token.json"): |
| creds = Credentials.from_authorized_user_file("token.json", SCOPES) |
| |
| if not creds or not creds.valid: |
| try: |
| if creds and creds.expired and creds.refresh_token: |
| creds.refresh(Request()) |
| else: |
| flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES) |
| creds = flow.run_local_server(port=0) |
| |
| with open("token.json", "w") as token: |
| token.write(creds.to_json()) |
| except Exception as e: |
| print(f"⚠️ Error while fetching credentials: {e}") |
| return None |
|
|
| return creds |
|
|
| def get_sheet_data(sheet_name: str, local_file: Optional[str] = None) -> List[Dict[str, Any]]: |
| """ |
| Fetch data from Google Sheets. |
| If API fails, fallback to loading data from the specified local file. |
| |
| :param sheet_name: Name of the sheet range in Google Sheets. |
| :param local_file: Path to the local Excel file for fallback. |
| :return: List of dictionary records with data. |
| """ |
| if SAMPLE_SPREADSHEET_ID is None: |
| print("⚠️ Google Spreadsheet ID is not set. Using local file instead...") |
| return load_local_file(local_file, sheet_name) |
|
|
| try: |
| creds = get_credentials() |
| if creds is None: |
| print("⚠️ Failed to get API credentials. Using local file instead...") |
| return load_local_file(local_file, sheet_name) |
|
|
| service = build("sheets", "v4", credentials=creds) |
| sheet = service.spreadsheets() |
| result = sheet.values().get(spreadsheetId=SAMPLE_SPREADSHEET_ID, range=sheet_name).execute() |
| values = result.get("values", []) |
|
|
| if not values: |
| print("⚠️ No data found in Google Sheets. Trying local file...") |
| return load_local_file(local_file, sheet_name) |
|
|
| df = pd.DataFrame(values[1:], columns=values[0]) |
| print("✅ Data loaded from Google Sheets.") |
| return df.to_dict(orient="records") |
|
|
| except HttpError as err: |
| print(f"⚠️ Google Sheets API Error: {err}. Falling back to local file...") |
| return load_local_file(local_file, sheet_name) |
|
|
| def load_local_file(local_file: Optional[str], sheet_name: str) -> List[Dict[str, Any]]: |
| """ |
| Load data from a local Excel file. |
| |
| :param local_file: Path to the local file. |
| :param sheet_name: Name of the sheet to read. |
| :return: List of dictionary records with data. |
| """ |
| if not local_file or not os.path.exists(local_file): |
| print(f"❌ Local file '{local_file}' not found or not specified.{sheet_name}") |
| return [] |
|
|
| try: |
| df = pd.ExcelFile(local_file, sheet_name=sheet_name) |
| print(f"✅ Data loaded from local file: {local_file} (Sheet: {sheet_name})") |
| return df.to_dict(orient="records") |
| except Exception as e: |
| print(f"❌ Error loading local file: {e}") |
| return [] |
|
|
| def main() -> List[Dict[str, Any]]: |
| """Main function to get data (API first, fallback to local).""" |
| local_file_path = "resume_sheet.xlsx" |
| return get_sheet_data(SAMPLE_RANGE_NAME, local_file=local_file_path) |
|
|
| if __name__ == "__main__": |
| print(main()) |
|
|