zoom_api.py

"""Zoom API integration with OAuth 2.0 Server-to-Server authentication."""

import base64
import logging
import os
import random
import tempfile
import time
from datetime import date
from typing import Optional

import requests

logger = logging.getLogger(__name__)

ZOOM_OAUTH_URL = "https://zoom.us/oauth/token"
ZOOM_API_BASE = "https://api.zoom.us/v2"


class ZoomAuthError(Exception):
    """Raised when OAuth authentication fails."""
    pass


class ZoomAPIError(Exception):
    """Raised when a Zoom API call fails after retries."""
    pass


class ZoomRateLimitError(ZoomAPIError):
    """Raised when rate limit is exhausted after all retries."""
    pass


class ZoomClient:
    """Handles OAuth 2.0 S2S authentication and API calls for a single Zoom account."""

    MAX_RETRIES = 3
    TOKEN_REFRESH_BUFFER = 100  # seconds before expiry to refresh

    def __init__(self, account_id: str, client_id: str, client_secret: str,
                 account_name: str = ""):
        self._account_id = account_id
        self._client_id = client_id
        self._client_secret = client_secret
        self._account_name = account_name
        self._access_token: Optional[str] = None
        self._token_expiry: float = 0
        self._session = requests.Session()

    def _get_basic_auth(self) -> str:
        """Return Base64-encoded 'client_id:client_secret' for Basic auth."""
        credentials = f"{self._client_id}:{self._client_secret}"
        return base64.b64encode(credentials.encode()).decode()

    def authenticate(self) -> str:
        """Obtain or return cached OAuth access token.

        Uses Server-to-Server OAuth with account_credentials grant type.
        Caches token until (expiry - TOKEN_REFRESH_BUFFER).
        Raises ZoomAuthError on failure.
        """
        if self._access_token and time.time() < self._token_expiry:
            return self._access_token

        logger.info("Authenticating account: %s", self._account_name)
        try:
            response = self._session.post(
                ZOOM_OAUTH_URL,
                headers={
                    "Authorization": f"Basic {self._get_basic_auth()}",
                    "Content-Type": "application/x-www-form-urlencoded",
                },
                data={
                    "grant_type": "account_credentials",
                    "account_id": self._account_id,
                },
                timeout=30,
            )
        except requests.RequestException as e:
            raise ZoomAuthError(
                f"Network error authenticating account '{self._account_name}': {e}"
            ) from e

        if response.status_code != 200:
            raise ZoomAuthError(
                f"Authentication failed for account '{self._account_name}': "
                f"HTTP {response.status_code} - {response.text}"
            )

        data = response.json()
        self._access_token = data["access_token"]
        expires_in = data.get("expires_in", 3600)
        self._token_expiry = time.time() + expires_in - self.TOKEN_REFRESH_BUFFER
        logger.debug("Token obtained for %s, expires in %ds",
                      self._account_name, expires_in)
        return self._access_token

    def _request_with_retry(self, method: str, url: str,
                            **kwargs) -> requests.Response:
        """Make an HTTP request with automatic retry on transient errors.

        Handles:
        - 401: Force re-authentication and retry
        - 429: Exponential backoff with jitter
        - 5xx / connection errors: Retry with backoff
        """
        last_exception = None
        for attempt in range(self.MAX_RETRIES + 1):
            try:
                token = self.authenticate()
                headers = kwargs.pop("headers", {})
                headers["Authorization"] = f"Bearer {token}"
                kwargs["headers"] = headers
                kwargs.setdefault("timeout", 30)

                response = self._session.request(method, url, **kwargs)

                if response.status_code == 401:
                    logger.warning("Got 401 for %s, forcing re-auth", url)
                    self._access_token = None
                    continue

                if response.status_code == 429:
                    retry_after = int(
                        response.headers.get("Retry-After", 2 ** attempt)
                    )
                    jitter = random.uniform(0, 1)
                    sleep_time = min(retry_after + jitter, 60)
                    logger.warning("Rate limited on %s. Sleeping %.1fs "
                                   "(attempt %d/%d)",
                                   url, sleep_time, attempt + 1,
                                   self.MAX_RETRIES + 1)
                    time.sleep(sleep_time)
                    continue

                if response.status_code >= 500:
                    logger.warning("Server error %d on %s (attempt %d/%d)",
                                   response.status_code, url, attempt + 1,
                                   self.MAX_RETRIES + 1)
                    time.sleep(2 ** attempt)
                    continue

                response.raise_for_status()
                return response

            except (requests.ConnectionError, requests.Timeout) as e:
                last_exception = e
                logger.warning("Connection error on %s (attempt %d/%d): %s",
                               url, attempt + 1, self.MAX_RETRIES + 1, e)
                time.sleep(2 ** attempt)
            except ZoomAuthError:
                raise
            except requests.HTTPError as e:
                raise ZoomAPIError(
                    f"HTTP error on {url}: {e}"
                ) from e

        if last_exception:
            raise ZoomAPIError(
                f"Failed after {self.MAX_RETRIES + 1} attempts on {url}: "
                f"{last_exception}"
            )
        raise ZoomAPIError(
            f"Failed after {self.MAX_RETRIES + 1} attempts on {url}"
        )

    def list_recordings(self, from_date: date,
                        to_date: date) -> list[dict]:
        """List all recordings between from_date and to_date.

        Handles pagination via next_page_token.
        Returns list of meeting dicts with recording_files arrays.
        """
        all_meetings = []
        next_page_token = ""

        while True:
            params = {
                "from": from_date.strftime("%Y-%m-%d"),
                "to": to_date.strftime("%Y-%m-%d"),
                "page_size": 300,
            }
            if next_page_token:
                params["next_page_token"] = next_page_token

            response = self._request_with_retry(
                "GET",
                f"{ZOOM_API_BASE}/users/me/recordings",
                params=params,
            )
            data = response.json()
            meetings = data.get("meetings", [])
            all_meetings.extend(meetings)

            next_page_token = data.get("next_page_token", "")
            if not next_page_token:
                break

        logger.info("Found %d meetings for %s between %s and %s",
                     len(all_meetings), self._account_name, from_date, to_date)
        return all_meetings

    def get_audio_recordings(self, from_date: date,
                             to_date: date) -> list[dict]:
        """Get only M4A audio recordings, flattened across all meetings.

        Returns list of dicts with: recording_id, meeting_id, meeting_topic,
        recording_start, download_url, file_size, file_type.
        """
        meetings = self.list_recordings(from_date, to_date)
        audio_files = []

        for meeting in meetings:
            topic = meeting.get("topic", "Untitled Meeting")
            meeting_id = meeting.get("uuid", meeting.get("id", ""))
            recording_start = meeting.get("start_time", "")

            for rec_file in meeting.get("recording_files", []):
                if rec_file.get("file_type", "").upper() == "M4A":
                    audio_files.append({
                        "recording_id": rec_file.get("id", ""),
                        "meeting_id": meeting_id,
                        "meeting_topic": topic,
                        "recording_start": recording_start,
                        "download_url": rec_file.get("download_url", ""),
                        "file_size": rec_file.get("file_size", 0),
                        "file_type": "M4A",
                    })

        logger.info("Found %d M4A files for %s", len(audio_files),
                     self._account_name)
        return audio_files

    def download_recording(self, download_url: str, dest_path: str,
                           expected_size: Optional[int] = None) -> str:
        """Stream-download a recording file to dest_path.

        Writes to a temp file first, then renames for atomicity.
        Returns final file path.
        Raises ValueError if the download URL is not HTTPS.
        """
        if not download_url.startswith("https://"):
            raise ValueError(
                f"Refusing to download over insecure URL: "
                f"{download_url[:60]}..."
            )
        dest = dest_path
        dir_path = os.path.dirname(dest)
        fd, tmp_path = tempfile.mkstemp(dir=dir_path, suffix=".tmp")

        try:
            token = self.authenticate()
            with self._session.get(
                download_url,
                headers={"Authorization": f"Bearer {token}"},
                stream=True,
                timeout=300,
            ) as response:
                response.raise_for_status()
                with os.fdopen(fd, "wb") as f:
                    for chunk in response.iter_content(chunk_size=8192):
                        if chunk:
                            f.write(chunk)

            # Check file size if expected
            actual_size = os.path.getsize(tmp_path)
            if expected_size and actual_size != expected_size:
                logger.warning(
                    "Size mismatch: expected %d, got %d for %s",
                    expected_size, actual_size, dest,
                )

            os.replace(tmp_path, dest)
            logger.info("Downloaded: %s (%s bytes)", dest, actual_size)
            return dest

        except Exception:
            # Clean up temp file on any failure
            try:
                os.close(fd)
            except OSError:
                pass
            try:
                os.unlink(tmp_path)
            except OSError:
                pass
            raise

    def test_connection(self) -> dict:
        """Test credentials by authenticating and calling GET /users/me.

        Returns user info dict on success.
        Raises ZoomAuthError or ZoomAPIError on failure.
        """
        self.authenticate()
        response = self._request_with_retry(
            "GET", f"{ZOOM_API_BASE}/users/me"
        )
        return response.json()

    def close(self):
        """Close the requests session."""
        self._session.close()