"""Google People API client for managing contacts.""" from typing import Any, Dict, List, Optional from googleapiclient.discovery import build from googleapiclient.errors import HttpError from .oauth_manager import GoogleOAuthManager PERSON_FIELDS = "names,emailAddresses,phoneNumbers,biographies" class PeopleClient: """Client for Google People API operations.""" def __init__(self, oauth_manager: GoogleOAuthManager): """Initialize People client. Args: oauth_manager: Initialized OAuth manager with valid credentials """ self.oauth_manager = oauth_manager self.service = None self._initialize_service() def _initialize_service(self) -> bool: """Initialize People API service. Returns: True if initialized successfully, False otherwise """ credentials = self.oauth_manager.get_credentials() if not credentials: return False try: self.service = build("people", "v1", credentials=credentials) return True except Exception as e: print(f"[People] Failed to initialize service: {e}") return False def _ensure_service(self) -> bool: """Ensure the service is initialized.""" if not self.service: return self._initialize_service() return True def create_contact( self, given_name: str, family_name: str = "", email: str = "", phone: Optional[str] = None, notes: Optional[str] = None, ) -> Dict: """Create a new contact. Args: given_name: First name family_name: Last name email: Email address phone: Optional phone number notes: Optional notes/biography Returns: Dict with success status and contact details or error """ if not self._ensure_service(): return { "success": False, "error": "Not authorized. Run: python bot_runner.py --setup-google", } try: body: Dict[str, Any] = { "names": [{"givenName": given_name, "familyName": family_name}], } if email: body["emailAddresses"] = [{"value": email}] if phone: body["phoneNumbers"] = [{"value": phone}] if notes: body["biographies"] = [{"value": notes, "contentType": "TEXT_PLAIN"}] result = ( self.service.people() .createContact(body=body) .execute() ) return { "success": True, "resource_name": result.get("resourceName", ""), "name": f"{given_name} {family_name}".strip(), } except HttpError as e: return {"success": False, "error": str(e)} def list_contacts( self, max_results: int = 100, query: Optional[str] = None, ) -> Dict: """List or search contacts. Args: max_results: Maximum number of contacts to return (max: 1000) query: Optional search query string Returns: Dict with success status and contacts or error """ if not self._ensure_service(): return { "success": False, "error": "Not authorized. Run: python bot_runner.py --setup-google", } try: max_results = min(max_results, 1000) if query: # Use searchContacts for query-based search result = ( self.service.people() .searchContacts( query=query, readMask="names,emailAddresses,phoneNumbers,biographies", pageSize=min(max_results, 30), # searchContacts max is 30 ) .execute() ) people = [r["person"] for r in result.get("results", [])] else: # Use connections.list for full listing result = ( self.service.people() .connections() .list( resourceName="people/me", pageSize=max_results, personFields=PERSON_FIELDS, sortOrder="LAST_NAME_ASCENDING", ) .execute() ) people = result.get("connections", []) contacts = [self._format_contact(p) for p in people] return { "success": True, "contacts": contacts, "count": len(contacts), "summary": self._format_contacts_summary(contacts), } except HttpError as e: return {"success": False, "error": str(e)} def get_contact(self, resource_name: str) -> Dict: """Get a single contact by resource name. Args: resource_name: Contact resource name (e.g., "people/c1234567890") Returns: Dict with success status and contact details or error """ if not self._ensure_service(): return { "success": False, "error": "Not authorized. Run: python bot_runner.py --setup-google", } try: result = ( self.service.people() .get(resourceName=resource_name, personFields=PERSON_FIELDS) .execute() ) return { "success": True, "contact": self._format_contact(result), } except HttpError as e: return {"success": False, "error": str(e)} def _format_contact(self, person: Dict) -> Dict: """Format a person resource into a simple contact dict.""" names = person.get("names", []) emails = person.get("emailAddresses", []) phones = person.get("phoneNumbers", []) bios = person.get("biographies", []) name = names[0] if names else {} return { "resource_name": person.get("resourceName", ""), "given_name": name.get("givenName", ""), "family_name": name.get("familyName", ""), "display_name": name.get("displayName", ""), "email": emails[0].get("value", "") if emails else "", "phone": phones[0].get("value", "") if phones else "", "notes": bios[0].get("value", "") if bios else "", } def _format_contacts_summary(self, contacts: List[Dict]) -> str: """Format contacts into a readable summary.""" if not contacts: return "No contacts found." lines = [] for i, c in enumerate(contacts, 1): name = c.get("display_name") or f"{c.get('given_name', '')} {c.get('family_name', '')}".strip() parts = [f"{i}. {name or '(no name)'}"] if c.get("email"): parts.append(f" Email: {c['email']}") if c.get("phone"): parts.append(f" Phone: {c['phone']}") lines.append("\n".join(parts)) return "\n".join(lines)