"""
Google Sheets Integration Module
Full CRUD operations with formatting support
"""
from googleapiclient.discovery import build
from src.google_auth import get_credentials
def get_sheets_service():
creds = get_credentials()
return build("sheets", "v4", credentials=creds)
def get_drive_service():
creds = get_credentials()
return build("drive", "v3", credentials=creds)
def create_spreadsheet(title, sheet_names=None):
"""Create a new spreadsheet with optional multiple sheets/tabs"""
try:
service = get_sheets_service()
spreadsheet_body = {"properties": {"title": title}}
if sheet_names:
spreadsheet_body["sheets"] = [
{"properties": {"title": name}} for name in sheet_names
]
spreadsheet = (
service.spreadsheets()
.create(body=spreadsheet_body, fields="spreadsheetId,spreadsheetUrl,sheets")
.execute()
)
return {
"success": True,
"spreadsheetId": spreadsheet.get("spreadsheetId"),
"spreadsheetUrl": spreadsheet.get("spreadsheetUrl"),
"sheets": [
{
"sheetId": sheet["properties"]["sheetId"],
"title": sheet["properties"]["title"],
}
for sheet in spreadsheet.get("sheets", [])
],
}
except Exception as error:
return {"success": False, "error": str(error)}
def read_sheet_data(spreadsheet_id, range_name):
"""Read data from a Google Sheet range"""
try:
service = get_sheets_service()
result = (
service.spreadsheets()
.values()
.get(spreadsheetId=spreadsheet_id, range=range_name)
.execute()
)
values = result.get("values", [])
return {
"success": True,
"data": values,
"range": result.get("range", range_name),
}
except Exception as error:
return {"success": False, "error": str(error)}
def write_sheet_data(spreadsheet_id, range_name, values):
"""Write data to a Google Sheet range"""
try:
service = get_sheets_service()
result = (
service.spreadsheets()
.values()
.update(
spreadsheetId=spreadsheet_id,
range=range_name,
valueInputOption="USER_ENTERED",
body={"values": values},
)
.execute()
)
return {
"success": True,
"updatedCells": result.get("updatedCells", 0),
"updatedRange": result.get("updatedRange", range_name),
}
except Exception as error:
return {"success": False, "error": str(error)}
def append_sheet_data(spreadsheet_id, range_name, values):
"""Append data to a Google Sheet"""
try:
service = get_sheets_service()
result = (
service.spreadsheets()
.values()
.append(
spreadsheetId=spreadsheet_id,
range=range_name,
valueInputOption="USER_ENTERED",
insertDataOption="INSERT_ROWS",
body={"values": values},
)
.execute()
)
updates = result.get("updates", {})
return {
"success": True,
"updatedCells": updates.get("updatedCells", 0),
"updatedRange": updates.get("updatedRange", range_name),
}
except Exception as error:
return {"success": False, "error": str(error)}
def get_sheet_info(spreadsheet_id):
"""Get information about a spreadsheet"""
try:
service = get_sheets_service()
result = service.spreadsheets().get(spreadsheetId=spreadsheet_id).execute()
properties = result.get("properties", {})
sheets_info = []
for sheet_info in result.get("sheets", []):
sheet_props = sheet_info.get("properties", {})
grid_props = sheet_props.get("gridProperties", {})
sheets_info.append(
{
"title": sheet_props.get("title", ""),
"sheetId": sheet_props.get("sheetId", ""),
"rowCount": grid_props.get("rowCount", 0),
"columnCount": grid_props.get("columnCount", 0),
}
)
return {
"success": True,
"title": properties.get("title", ""),
"spreadsheetId": spreadsheet_id,
"sheets": sheets_info,
}
except Exception as error:
return {"success": False, "error": str(error)}
def create_sheet_tab(spreadsheet_id, tab_name):
"""Create a new tab/worksheet in an existing spreadsheet"""
try:
service = get_sheets_service()
request_body = {"requests": [{"addSheet": {"properties": {"title": tab_name}}}]}
response = (
service.spreadsheets()
.batchUpdate(spreadsheetId=spreadsheet_id, body=request_body)
.execute()
)
reply = response.get("replies", [{}])[0]
sheet_props = reply.get("addSheet", {}).get("properties", {})
return {
"success": True,
"sheetId": sheet_props.get("sheetId"),
"title": sheet_props.get("title"),
}
except Exception as error:
return {"success": False, "error": str(error)}
def list_spreadsheets(max_results=50):
"""List all spreadsheets in Google Drive"""
try:
drive_service = get_drive_service()
results = (
drive_service.files()
.list(
q="mimeType='application/vnd.google-apps.spreadsheet'",
spaces="drive",
fields="files(id, name, createdTime, modifiedTime, webViewLink)",
orderBy="modifiedTime desc",
pageSize=max_results,
)
.execute()
)
files = results.get("files", [])
return {
"success": True,
"spreadsheets": [
{
"id": f["id"],
"name": f["name"],
"createdTime": f.get("createdTime", ""),
"modifiedTime": f.get("modifiedTime", ""),
"webViewLink": f.get("webViewLink", ""),
}
for f in files
],
"total": len(files),
}
except Exception as error:
return {"success": False, "error": str(error)}
def format_cells(spreadsheet_id, range_name, format_options):
"""
Format cells in a spreadsheet
format_options can include:
- bold: bool
- italic: bool
- fontSize: int
- foregroundColor: {red, green, blue} (0-1 values)
- backgroundColor: {red, green, blue} (0-1 values)
- horizontalAlignment: 'LEFT', 'CENTER', 'RIGHT'
- verticalAlignment: 'TOP', 'MIDDLE', 'BOTTOM'
- wrapStrategy: 'OVERFLOW_CELL', 'CLIP', 'WRAP'
- borders: {top, bottom, left, right} with {style, color}
"""
try:
service = get_sheets_service()
# First get sheet info to resolve range to grid coordinates
sheet_info = (
service.spreadsheets()
.get(
spreadsheetId=spreadsheet_id, ranges=[range_name], includeGridData=False
)
.execute()
)
# Parse range to get sheet ID and grid range
sheets = sheet_info.get("sheets", [])
if not sheets:
return {"success": False, "error": "Sheet not found"}
sheet_id = sheets[0]["properties"]["sheetId"]
# Parse range string (e.g., "Sheet1!A1:B5" or "A1:B5")
grid_range = _parse_range(range_name, sheet_id)
# Build format request
cell_format = {}
# Text format
text_format = {}
if "bold" in format_options:
text_format["bold"] = format_options["bold"]
if "italic" in format_options:
text_format["italic"] = format_options["italic"]
if "fontSize" in format_options:
text_format["fontSize"] = format_options["fontSize"]
if "foregroundColor" in format_options:
text_format["foregroundColor"] = format_options["foregroundColor"]
if text_format:
cell_format["textFormat"] = text_format
# Background color
if "backgroundColor" in format_options:
cell_format["backgroundColor"] = format_options["backgroundColor"]
# Alignment
if "horizontalAlignment" in format_options:
cell_format["horizontalAlignment"] = format_options["horizontalAlignment"]
if "verticalAlignment" in format_options:
cell_format["verticalAlignment"] = format_options["verticalAlignment"]
# Wrap strategy
if "wrapStrategy" in format_options:
cell_format["wrapStrategy"] = format_options["wrapStrategy"]
# Build the request
requests = [
{
"repeatCell": {
"range": grid_range,
"cell": {"userEnteredFormat": cell_format},
"fields": "userEnteredFormat",
}
}
]
# Add borders if specified
if "borders" in format_options:
border_request = _build_border_request(
grid_range, format_options["borders"]
)
if border_request:
requests.append(border_request)
response = (
service.spreadsheets()
.batchUpdate(spreadsheetId=spreadsheet_id, body={"requests": requests})
.execute()
)
return {
"success": True,
"updatedRange": range_name,
"appliedFormat": format_options,
}
except Exception as error:
return {"success": False, "error": str(error)}
def _parse_range(range_str, sheet_id):
"""Parse a range string like 'Sheet1!A1:B5' into grid coordinates"""
# Remove sheet name if present
if "!" in range_str:
range_str = range_str.split("!")[1]
# Split into start and end cells
if ":" in range_str:
start, end = range_str.split(":")
else:
start = end = range_str
start_col, start_row = _parse_cell(start)
end_col, end_row = _parse_cell(end)
return {
"sheetId": sheet_id,
"startRowIndex": start_row - 1,
"endRowIndex": end_row,
"startColumnIndex": start_col,
"endColumnIndex": end_col + 1,
}
def _parse_cell(cell_str):
"""Parse a cell reference like 'A1' into (column_index, row_number)"""
col_str = ""
row_str = ""
for char in cell_str:
if char.isalpha():
col_str += char.upper()
else:
row_str += char
# Convert column letters to index (A=0, B=1, ..., AA=26, etc.)
col_index = 0
for char in col_str:
col_index = col_index * 26 + (ord(char) - ord("A") + 1)
col_index -= 1
row_num = int(row_str) if row_str else 1
return col_index, row_num
def _build_border_request(grid_range, borders):
"""Build a border update request"""
border_style = {"style": "SOLID", "color": {"red": 0, "green": 0, "blue": 0}}
update_borders = {}
for side in ["top", "bottom", "left", "right"]:
if side in borders:
side_config = borders[side]
if isinstance(side_config, bool) and side_config:
update_borders[side] = border_style
elif isinstance(side_config, dict):
update_borders[side] = {
"style": side_config.get("style", "SOLID"),
"color": side_config.get(
"color", {"red": 0, "green": 0, "blue": 0}
),
}
if update_borders:
return {"updateBorders": {"range": grid_range, **update_borders}}
return None
def add_formula(spreadsheet_id, cell, formula):
"""Add a formula to a specific cell"""
try:
service = get_sheets_service()
result = (
service.spreadsheets()
.values()
.update(
spreadsheetId=spreadsheet_id,
range=cell,
valueInputOption="USER_ENTERED",
body={"values": [[formula]]},
)
.execute()
)
return {"success": True, "updatedCell": cell, "formula": formula}
except Exception as error:
return {"success": False, "error": str(error)}
def share_spreadsheet(spreadsheet_id, email, role="reader"):
"""
Share a spreadsheet with a user
role: 'reader', 'writer', 'commenter'
"""
try:
drive_service = get_drive_service()
permission = {"type": "user", "role": role, "emailAddress": email}
result = (
drive_service.permissions()
.create(fileId=spreadsheet_id, body=permission, sendNotificationEmail=True)
.execute()
)
return {
"success": True,
"permissionId": result.get("id"),
"sharedWith": email,
"role": role,
}
except Exception as error:
return {"success": False, "error": str(error)}
# Helper function for creating job tracker
def create_job_tracker(title="Job Applications Tracker"):
"""Create a job tracker spreadsheet with predefined columns"""
try:
# Create spreadsheet
result = create_spreadsheet(title, ["Applications", "Interviews", "Offers"])
if not result["success"]:
return result
spreadsheet_id = result["spreadsheetId"]
# Add headers to Applications sheet
headers = [
[
"Company",
"Position",
"Location",
"Salary Range",
"Date Applied",
"Status",
"Contact",
"Email",
"Notes",
"Follow-up Date",
]
]
write_sheet_data(spreadsheet_id, "Applications!A1:J1", headers)
# Add headers to Interviews sheet
interview_headers = [
[
"Company",
"Position",
"Interview Date",
"Interview Type",
"Interviewer",
"Status",
"Notes",
"Next Steps",
]
]
write_sheet_data(spreadsheet_id, "Interviews!A1:H1", interview_headers)
# Add headers to Offers sheet
offer_headers = [
[
"Company",
"Position",
"Base Salary",
"Bonus",
"Benefits",
"Start Date",
"Deadline",
"Status",
"Notes",
]
]
write_sheet_data(spreadsheet_id, "Offers!A1:I1", offer_headers)
# Format headers (bold)
format_cells(
spreadsheet_id,
"Applications!A1:J1",
{"bold": True, "backgroundColor": {"red": 0.9, "green": 0.9, "blue": 0.9}},
)
format_cells(
spreadsheet_id,
"Interviews!A1:H1",
{"bold": True, "backgroundColor": {"red": 0.9, "green": 0.9, "blue": 0.9}},
)
format_cells(
spreadsheet_id,
"Offers!A1:I1",
{"bold": True, "backgroundColor": {"red": 0.9, "green": 0.9, "blue": 0.9}},
)
return {
"success": True,
"spreadsheetId": spreadsheet_id,
"spreadsheetUrl": result["spreadsheetUrl"],
"message": "Job tracker created with Applications, Interviews, and Offers sheets",
}
except Exception as error:
return {"success": False, "error": str(error)}
