Spaces:
Sleeping
Sleeping
File size: 4,557 Bytes
0d37119 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | """Tool functions exposed to Gemini. Uses real Google APIs when authenticated, mock data otherwise."""
import json
from pathlib import Path
from google_auth import is_authenticated
from google_services import fetch_emails, fetch_events, fetch_doc_content, list_recent_docs, search_project_signals
from mock_data import MOCK_EMAILS, MOCK_EVENTS, MOCK_SEARCH
def _load_project_keywords(project_id: str) -> list[str]:
"""Load keywords for a project from config.json."""
config_path = Path(__file__).parent / "config.json"
with open(config_path) as f:
config = json.load(f)
for p in config.get("projects", []):
if p["id"] == project_id:
return p.get("keywords", [])
return []
def read_emails(project_id: str) -> str:
"""Read recent emails for a project. Uses Gmail API if authenticated, mock data otherwise."""
if is_authenticated():
keywords = _load_project_keywords(project_id)
emails = fetch_emails(project_id, keywords=keywords)
if not emails:
return f"No emails found for project {project_id}."
lines = []
for e in emails:
days = e.get("days_since_reply")
silence = f"{days} days ago" if days is not None else "unknown"
lines.append(
f"From: {e['from']}\n"
f"Subject: {e['subject']}\n"
f"Body: {e['body']}\n"
f"Last reply: {silence}"
)
return "\n---\n".join(lines)
# Fallback to mock data
emails = MOCK_EMAILS.get(project_id, [])
if not emails:
return f"No emails found for project {project_id}."
lines = []
for e in emails:
silence = f"{e['days_since_reply']} days ago" if e["days_since_reply"] is not None else "N/A (newsletter)"
lines.append(
f"From: {e['from']}\n"
f"Subject: {e['subject']}\n"
f"Body: {e['body']}\n"
f"Last reply: {silence}"
)
return "\n---\n".join(lines)
def get_events(project_id: str) -> str:
"""Get upcoming calendar events for a project. Uses Calendar API if authenticated, mock data otherwise."""
if is_authenticated():
keywords = _load_project_keywords(project_id)
events = fetch_events(project_id, keywords=keywords)
if not events:
return f"No upcoming events for project {project_id}."
lines = []
for ev in events:
prep = "YES" if ev["prep_block"] else "NO"
lines.append(f"{ev['title']} at {ev['time']} (prep block: {prep})")
return "\n".join(lines)
# Fallback to mock data
events = MOCK_EVENTS.get(project_id, [])
if not events:
return f"No upcoming events for project {project_id}."
lines = []
for ev in events:
prep = "YES" if ev["prep_block"] else "NO"
lines.append(f"{ev['title']} at {ev['time']} (prep block: {prep})")
return "\n".join(lines)
def search_web(project_id: str) -> str:
"""Search for recent external signals relevant to a project. Uses Gemini + Google Search grounding when possible."""
keywords = _load_project_keywords(project_id)
# Load project name from config
config_path = Path(__file__).parent / "config.json"
with open(config_path) as f:
config = json.load(f)
project_name = project_id
for p in config.get("projects", []):
if p["id"] == project_id:
project_name = p.get("name", project_id)
break
result = search_project_signals(project_name, keywords=keywords)
if result:
return result
# Fallback to mock data
return MOCK_SEARCH.get(project_id, "No relevant external signals found.")
def read_doc(doc_id: str) -> str:
"""Read a Google Doc by its document ID. Returns title and content."""
if not is_authenticated():
return "Not connected to Google. Please authenticate first."
result = fetch_doc_content(doc_id)
if not result["content"]:
return f"Could not read document {doc_id}."
return f"Title: {result['title']}\n\nContent:\n{result['content']}"
def list_docs() -> str:
"""List recently modified Google Docs."""
if not is_authenticated():
return "Not connected to Google. Please authenticate first."
docs = list_recent_docs(max_results=10)
if not docs:
return "No recent Google Docs found."
lines = []
for d in docs:
lines.append(f"- {d['title']} (ID: {d['id']}, modified: {d['modified_time']})")
return "\n".join(lines)
|