"""Work Google Workspace (ahmed.alrajeh@alrugaibfurniture.com) for the jarvis-tools service ā€” Gmail + Calendar + Sheets, over an OAuth refresh token carried in env so it runs headless and every device reaches it via the one connector. Env: GWS_REFRESH_TOKEN the work account's OAuth refresh token GWS_CLIENT_ID OAuth client id GWS_CLIENT_SECRET OAuth client secret GWS_EMAIL the account address (for display; default alrugaib) """ from __future__ import annotations import base64 import os from datetime import datetime, timedelta, timezone from email.message import EmailMessage _SCOPES = [ "https://www.googleapis.com/auth/gmail.send", "https://www.googleapis.com/auth/gmail.readonly", "https://www.googleapis.com/auth/gmail.modify", "https://www.googleapis.com/auth/calendar", "https://www.googleapis.com/auth/calendar.events", "https://www.googleapis.com/auth/spreadsheets", ] def configured() -> bool: return bool(os.environ.get("GWS_REFRESH_TOKEN") and os.environ.get("GWS_CLIENT_ID") and os.environ.get("GWS_CLIENT_SECRET")) def _creds(): from google.oauth2.credentials import Credentials return Credentials( token=None, refresh_token=os.environ["GWS_REFRESH_TOKEN"], client_id=os.environ["GWS_CLIENT_ID"], client_secret=os.environ["GWS_CLIENT_SECRET"], token_uri="https://oauth2.googleapis.com/token", scopes=_SCOPES, ) def _svc(api: str, ver: str): from googleapiclient.discovery import build return build(api, ver, credentials=_creds(), cache_discovery=False) # ---- Gmail ---------------------------------------------------------------- def gmail_send(to: str, subject: str, body: str, cc: str = "") -> str: to = str(to).strip() if not to: return "gmail_send failed: no recipient." msg = EmailMessage() msg["To"] = to if cc: msg["Cc"] = cc msg["Subject"] = subject msg.set_content(body) raw = base64.urlsafe_b64encode(msg.as_bytes()).decode() out = _svc("gmail", "v1").users().messages().send( userId="me", body={"raw": raw}).execute() return f"Sent '{subject}' to {to} (message id {out.get('id')})." def _gmail_body(payload) -> str: def walk(p): if p.get("mimeType") == "text/plain" and p.get("body", {}).get("data"): return base64.urlsafe_b64decode( p["body"]["data"]).decode("utf-8", "replace") for part in p.get("parts", []) or []: r = walk(part) if r: return r return "" return walk(payload) or "(no plain-text body)" def gmail_search(query: str, limit: int = 10) -> str: limit = max(1, min(int(limit or 10), 25)) svc = _svc("gmail", "v1") res = svc.users().messages().list( userId="me", q=str(query or ""), maxResults=limit).execute() msgs = res.get("messages", []) if not msgs: return "No matching messages." lines = [] for mref in msgs: m = svc.users().messages().get( userId="me", id=mref["id"], format="metadata", metadataHeaders=["From", "Subject", "Date"]).execute() h = {x["name"]: x["value"] for x in m.get("payload", {}).get("headers", [])} lines.append(f"[id {mref['id']}] {h.get('Date','')}\n" f" from: {h.get('From','')}\n subj: {h.get('Subject','')}") return "\n".join(lines) def gmail_read(message_id: str) -> str: m = _svc("gmail", "v1").users().messages().get( userId="me", id=str(message_id).strip(), format="full").execute() h = {x["name"]: x["value"] for x in m.get("payload", {}).get("headers", [])} body = _gmail_body(m.get("payload", {})) if len(body) > 5000: body = body[:5000] + "\nā€¦(truncated)" return (f"From: {h.get('From','')}\nSubject: {h.get('Subject','')}\n" f"Date: {h.get('Date','')}\n\n{body}") # ---- Calendar ------------------------------------------------------------- def calendar_list(days: int = 7) -> str: days = max(1, min(int(days or 7), 60)) now = datetime.now(timezone.utc) svc = _svc("calendar", "v3") res = svc.events().list( calendarId="primary", timeMin=now.isoformat(), timeMax=(now + timedelta(days=days)).isoformat(), singleEvents=True, orderBy="startTime", maxResults=50).execute() items = res.get("items", []) if not items: return f"No events in the next {days} day(s)." lines = [] for e in items: st = e.get("start", {}).get("dateTime") or e.get("start", {}).get("date") lines.append(f"- {st} {e.get('summary','(no title)')}" + (f" [{e['id']}]" if e.get("id") else "")) return f"{len(items)} event(s) in the next {days} day(s):\n" + "\n".join(lines) def calendar_create_event(title: str, start: str, end: str, attendees: str = "", description: str = "", add_meet: str = "") -> str: """start/end = ISO datetime (2026-07-15T14:00:00+03:00). attendees = comma-separated emails. add_meet='yes' attaches a Google Meet link.""" body = {"summary": title, "description": description, "start": {"dateTime": start}, "end": {"dateTime": end}} ats = [a.strip() for a in str(attendees).split(",") if a.strip()] if ats: body["attendees"] = [{"email": a} for a in ats] kw = {} if str(add_meet).lower() in ("yes", "true", "1"): import hashlib body["conferenceData"] = {"createRequest": { "requestId": hashlib.sha256( (title + start).encode()).hexdigest()[:16], "conferenceSolutionKey": {"type": "hangoutsMeet"}}} kw["conferenceDataVersion"] = 1 out = _svc("calendar", "v3").events().insert( calendarId="primary", body=body, sendUpdates="all", **kw).execute() meet = out.get("hangoutLink") return (f"Created '{title}' ({start} ā†’ {end})" + (f", {len(ats)} invited" if ats else "") + (f". Meet: {meet}" if meet else ".") + f" [{out.get('id')}]") # ---- Sheets --------------------------------------------------------------- def sheets_read(spreadsheet_id: str, a1_range: str) -> str: res = _svc("sheets", "v4").spreadsheets().values().get( spreadsheetId=str(spreadsheet_id).strip(), range=str(a1_range).strip()).execute() rows = res.get("values", []) if not rows: return "(no data in that range)" out = "\n".join(" | ".join(str(c) for c in r) for r in rows[:200]) return out[:5000] + (" ā€¦[truncated]" if len(out) > 5000 else "") def sheets_write(spreadsheet_id: str, a1_range: str, values_json: str) -> str: import json try: values = json.loads(values_json) except Exception as e: # noqa: BLE001 return f"values_json must be a JSON array of rows: {str(e)[:100]}" res = _svc("sheets", "v4").spreadsheets().values().update( spreadsheetId=str(spreadsheet_id).strip(), range=str(a1_range).strip(), valueInputOption="USER_ENTERED", body={"values": values}).execute() return f"Updated {res.get('updatedCells', 0)} cell(s) in {a1_range}."