""" Google Sheets + Google Drive integration for Job Automation Agent. Writes all assessed jobs to: Sheet: https://docs.google.com/spreadsheets/d/SHEET_ID Drive: "Job Automation Agent — Resumes" folder Each run APPENDS rows (never overwrites), so you keep a history of all searches. Authentication: Option A (Service Account — for servers/automation): → Place google_credentials.json (service account key) in project root → Share the Google Sheet with the service account email Option B (OAuth — for personal use, easiest): → Run setup_google.py once for browser login → Token cached in google_token.json See setup_google.py for step-by-step instructions. """ import os import json import logging import re from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Tuple logger = logging.getLogger(__name__) # ── Column definitions ──────────────────────────────────────────────────────── HEADERS = [ "Batch Date", "Rank", "Job Title", "Company", "Location", "Platform", "Salary Range", "Experience", "Relevance Score", "ATS Before (%)", # JD match on original resume "ATS After (%)", # JD match on tailored resume "ATS Improvement", # After - Before "Resume Quality", # Resume-ATS structural quality (independent of JD) "Priority", "Matching Skills", "Missing Skills", "AI Recommendation", "Apply Link", "Resume File", # Filename of tailored resume "Resume Folder", # Local folder path — open in File Explorer "Application Status", "Date Applied", "Notes", ] STATUS_OPTIONS = ["Not Applied", "Applied", "Shortlisted", "Interview", "Offer", "Rejected"] # Row colors based on score def _score_color(score: int): if score >= 8: return {"red": 0.88, "green": 0.98, "blue": 0.88} # light green if score >= 6: return {"red": 1.00, "green": 0.97, "blue": 0.85} # light yellow if score >= 4: return {"red": 1.00, "green": 0.93, "blue": 0.88} # light orange return {"red": 1.00, "green": 0.90, "blue": 0.90} # light red def _get_client(): """Return authenticated gspread client (service account or OAuth).""" import gspread from google.oauth2.service_account import Credentials as SACredentials from google.oauth2.credentials import Credentials as OAuthCredentials from google_auth_oauthlib.flow import InstalledAppFlow from google.auth.transport.requests import Request SCOPES = [ "https://www.googleapis.com/auth/spreadsheets", "https://www.googleapis.com/auth/drive", ] sa_file = Path("google_credentials.json") token_file = Path("google_token.json") oauth_file = Path("google_oauth_client.json") # ── Option A: Service Account ── if sa_file.exists(): try: creds = SACredentials.from_service_account_file(str(sa_file), scopes=SCOPES) return gspread.authorize(creds) except Exception as e: logger.warning(f"Service account auth failed: {e}") # ── Option B: OAuth2 ── creds = None if token_file.exists(): try: creds = OAuthCredentials.from_authorized_user_file(str(token_file), SCOPES) except Exception: pass if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) elif oauth_file.exists(): flow = InstalledAppFlow.from_client_secrets_file(str(oauth_file), SCOPES) creds = flow.run_local_server(port=0) else: raise FileNotFoundError( "\n\n❌ Google credentials not found!\n" "Run: python setup_google.py\n" "for step-by-step setup instructions.\n" ) # Save token for next run token_file.write_text(creds.to_json()) return gspread.authorize(creds) def _get_or_create_drive_folder(drive_service, folder_name: str) -> str: """Get Google Drive folder ID, creating it if needed.""" query = ( f"name='{folder_name}' and mimeType='application/vnd.google-apps.folder' " f"and trashed=false" ) results = drive_service.files().list(q=query, fields="files(id, name)").execute() files = results.get("files", []) if files: return files[0]["id"] # Create folder meta = { "name": folder_name, "mimeType": "application/vnd.google-apps.folder", } folder = drive_service.files().create(body=meta, fields="id").execute() return folder["id"] def upload_resume_to_drive( file_path: str, folder_name: str = "Job Automation Agent — Resumes", ) -> str: """ Upload a DOCX resume to Google Drive and return a public shareable link. Returns empty string if Drive upload is unavailable. """ try: from googleapiclient.discovery import build from googleapiclient.http import MediaFileUpload import gspread client = _get_client() creds = client.auth drive_service = build("drive", "v3", credentials=creds) folder_id = _get_or_create_drive_folder(drive_service, folder_name) fname = Path(file_path).name media = MediaFileUpload( file_path, mimetype="application/vnd.openxmlformats-officedocument.wordprocessingml.document", ) file_meta = { "name": fname, "parents": [folder_id], } uploaded = drive_service.files().create( body=file_meta, media_body=media, fields="id" ).execute() file_id = uploaded["id"] # Make publicly accessible (anyone with link can view) drive_service.permissions().create( fileId=file_id, body={"type": "anyone", "role": "reader"}, ).execute() return f"https://drive.google.com/file/d/{file_id}/view?usp=sharing" except Exception as e: logger.warning(f"Drive upload failed for {file_path}: {e}") return "" def write_jobs_to_sheet( jobs: List[Dict], sheet_id: str, tab_name: str = "Job Applications", batch_label: str = "", ) -> bool: """ Write assessed jobs to Google Sheet. - Creates the tab if it doesn't exist - Adds header row if sheet is empty - APPENDS new rows (never overwrites existing data) - Returns True on success """ try: client = _get_client() sh = client.open_by_key(sheet_id) # Get or create tab try: ws = sh.worksheet(tab_name) except Exception: ws = sh.add_worksheet(title=tab_name, rows=2000, cols=len(HEADERS)) # Check if header row exists existing = ws.get_all_values() has_headers = bool(existing and existing[0] and existing[0][0] == "Batch Date") if not existing: # Empty sheet — write header row ws.append_row(HEADERS, value_input_option="USER_ENTERED") _format_header(ws) first_data_row = 2 elif not has_headers: # Data exists but no header — insert header at row 1 ws.insert_row(HEADERS, index=1, value_input_option="USER_ENTERED") _format_header(ws) first_data_row = len(existing) + 2 # existing rows shifted down by 1 else: first_data_row = len(existing) + 1 if not batch_label: batch_label = datetime.now().strftime("%Y-%m-%d %H:%M") # Build rows rows = [] for rank, job in enumerate(jobs, 1): job_url = job.get("url", "") resume_path = job.get("resume_path", "") resume_link = job.get("drive_resume_link", "") # Format job URL as clickable hyperlink formula apply_cell = f'=HYPERLINK("{job_url}","Apply →")' if job_url else "" # Resume file and folder cells resume_file_cell = "" resume_folder_cell = "" if resume_path and os.path.exists(resume_path): p = Path(resume_path).resolve() resume_file_cell = p.name # just filename resume_folder_cell = str(p.parent) # full Windows path to folder score = job.get("relevance_score", 0) before = job.get("ats_score_before") # None if not set after = job.get("ats_score_after") # None if not set improv = job.get("ats_improvement", 0) or 0 quality = job.get("resume_quality_score") def _pct(val): """Format an ATS percentage value for the sheet.""" if val is None or val == "": return "—" return f"{int(val)}%" improv_str = f"+{improv}pp" if improv > 0 else (f"{improv}pp" if improv < 0 else "—") row = [ batch_label, rank, job.get("title", ""), job.get("company", ""), job.get("location", ""), job.get("platform", ""), job.get("salary", "Not specified"), job.get("experience_required", ""), f"{score}/10", _pct(before), _pct(after), improv_str, _pct(quality), job.get("application_priority", ""), job.get("matching_skills", ""), job.get("missing_skills", ""), (job.get("recommendation", "") or "")[:200], apply_cell, resume_file_cell, resume_folder_cell, "Not Applied", "", "", ] rows.append(row) if rows: # Append all rows at once ws.append_rows(rows, value_input_option="USER_ENTERED") logger.info(f"Wrote {len(rows)} jobs to Google Sheet tab '{tab_name}'") # Apply conditional formatting for score column (col I = index 9) _apply_score_formatting(ws, first_data_row, first_data_row + len(rows) - 1, sheet_id) return True except FileNotFoundError as e: print(str(e)) return False except Exception as e: logger.error(f"Google Sheet write failed: {e}") print(f"⚠ Google Sheet update failed: {e}") return False def _format_header(ws): """Bold + freeze header row, auto-resize columns.""" try: from gspread.utils import rowcol_to_a1 ws.format("A1:W1", { "textFormat": {"bold": True, "foregroundColor": {"red": 1, "green": 1, "blue": 1}}, "backgroundColor": {"red": 0.086, "green": 0.282, "blue": 0.745}, "horizontalAlignment": "CENTER", }) ws.freeze(rows=1) # Set column widths via batchUpdate try: sh = ws.spreadsheet sh.batch_update({ "requests": [ {"updateDimensionProperties": { "range": {"sheetId": ws.id, "dimension": "COLUMNS", "startIndex": 0, "endIndex": len(HEADERS)}, "properties": {"pixelSize": 150}, "fields": "pixelSize", }}, # Wider columns {"updateDimensionProperties": { "range": {"sheetId": ws.id, "dimension": "COLUMNS", "startIndex": 2, "endIndex": 3}, # Job Title "properties": {"pixelSize": 260}, "fields": "pixelSize", }}, ] }) except Exception: pass except Exception as e: logger.warning(f"Header formatting failed: {e}") def _apply_score_formatting(ws, start_row: int, end_row: int, sheet_id: str): """Color rows based on relevance score column (col I).""" pass # gspread conditional formatting is complex; skipped for now def get_sheet_url(sheet_id: str) -> str: return f"https://docs.google.com/spreadsheets/d/{sheet_id}/edit"