328 lines
10 KiB
Python
328 lines
10 KiB
Python
|
"""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 update_contact(self, resource_name: str, updates: Dict) -> Dict:
|
||
|
|
"""Update an existing contact.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
resource_name: Contact resource name (e.g., "people/c1234567890")
|
||
|
|
updates: Dict with fields to update (given_name, family_name, email, phone, notes)
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Dict with success status or error
|
||
|
|
"""
|
||
|
|
if not self._ensure_service():
|
||
|
|
return {
|
||
|
|
"success": False,
|
||
|
|
"error": "Not authorized. Run: python bot_runner.py --setup-google",
|
||
|
|
}
|
||
|
|
|
||
|
|
try:
|
||
|
|
# Get current contact to obtain etag
|
||
|
|
current = (
|
||
|
|
self.service.people()
|
||
|
|
.get(resourceName=resource_name, personFields=PERSON_FIELDS)
|
||
|
|
.execute()
|
||
|
|
)
|
||
|
|
|
||
|
|
body: Dict[str, Any] = {"etag": current["etag"]}
|
||
|
|
update_fields = []
|
||
|
|
|
||
|
|
if "given_name" in updates or "family_name" in updates:
|
||
|
|
names = current.get("names", [{}])
|
||
|
|
name = names[0] if names else {}
|
||
|
|
body["names"] = [{
|
||
|
|
"givenName": updates.get("given_name", name.get("givenName", "")),
|
||
|
|
"familyName": updates.get("family_name", name.get("familyName", "")),
|
||
|
|
}]
|
||
|
|
update_fields.append("names")
|
||
|
|
|
||
|
|
if "email" in updates:
|
||
|
|
body["emailAddresses"] = [{"value": updates["email"]}]
|
||
|
|
update_fields.append("emailAddresses")
|
||
|
|
|
||
|
|
if "phone" in updates:
|
||
|
|
body["phoneNumbers"] = [{"value": updates["phone"]}]
|
||
|
|
update_fields.append("phoneNumbers")
|
||
|
|
|
||
|
|
if "notes" in updates:
|
||
|
|
body["biographies"] = [{"value": updates["notes"], "contentType": "TEXT_PLAIN"}]
|
||
|
|
update_fields.append("biographies")
|
||
|
|
|
||
|
|
if not update_fields:
|
||
|
|
return {"success": False, "error": "No valid fields to update"}
|
||
|
|
|
||
|
|
result = (
|
||
|
|
self.service.people()
|
||
|
|
.updateContact(
|
||
|
|
resourceName=resource_name,
|
||
|
|
body=body,
|
||
|
|
updatePersonFields=",".join(update_fields),
|
||
|
|
)
|
||
|
|
.execute()
|
||
|
|
)
|
||
|
|
|
||
|
|
return {
|
||
|
|
"success": True,
|
||
|
|
"resource_name": result.get("resourceName", resource_name),
|
||
|
|
"updated_fields": update_fields,
|
||
|
|
}
|
||
|
|
|
||
|
|
except HttpError as e:
|
||
|
|
return {"success": False, "error": str(e)}
|
||
|
|
|
||
|
|
def delete_contact(self, resource_name: str) -> Dict:
|
||
|
|
"""Delete a contact.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
resource_name: Contact resource name (e.g., "people/c1234567890")
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Dict with success status or error
|
||
|
|
"""
|
||
|
|
if not self._ensure_service():
|
||
|
|
return {
|
||
|
|
"success": False,
|
||
|
|
"error": "Not authorized. Run: python bot_runner.py --setup-google",
|
||
|
|
}
|
||
|
|
|
||
|
|
try:
|
||
|
|
self.service.people().deleteContact(
|
||
|
|
resourceName=resource_name,
|
||
|
|
).execute()
|
||
|
|
|
||
|
|
return {"success": True, "deleted": resource_name}
|
||
|
|
|
||
|
|
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)
|