ajarbot/google_tools/people_client.py

"""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)