"""Client helpers for posting to Mastodon.""" from __future__ import annotations from typing import Any, Dict, Optional, Sequence import requests class MastodonError(RuntimeError): """Raised when the Mastodon API indicates a failure.""" class MastodonClient: """Minimal Mastodon API wrapper for uploading media and posting statuses.""" def __init__(self, base_url: str, access_token: str) -> None: if not base_url: raise ValueError("base_url is required") if not access_token: raise ValueError("access_token is required") self.base_url = base_url.rstrip("/") self.session = requests.Session() self.session.headers.update( { "Authorization": f"Bearer {access_token}", "Accept": "application/json", "User-Agent": "ImmichAltTextHelper/0.1 (+https://photos.4angle.com)", } ) self._account_id: Optional[str] = None def _raise_for_error(self, response: requests.Response) -> None: try: payload = response.json() except Exception: # pragma: no cover - fallback path payload = response.text message = f"{response.status_code} {response.reason}: {payload}" raise MastodonError(message) def _request(self, method: str, path: str, **kwargs: Any) -> Dict[str, Any]: url = f"{self.base_url}{path}" try: response = self.session.request(method, url, timeout=30, **kwargs) except requests.RequestException as exc: # pragma: no cover - network failure raise MastodonError(str(exc)) from exc if not response.ok: self._raise_for_error(response) if response.content: return response.json() return {} def upload_media( self, filename: str, data: bytes, mime_type: str, alt_text: str ) -> str: files = { "file": ( filename or "photo.jpg", data, mime_type or "application/octet-stream", ) } form = {} if alt_text: form["description"] = alt_text payload = self._request("POST", "/api/v2/media", files=files, data=form) media_id = payload.get("id") if not media_id: raise MastodonError("Mastodon response missing media id") return media_id def create_status( self, text: str, media_ids: Sequence[str], in_reply_to_id: Optional[str] = None ) -> Dict[str, str]: if not text.strip(): raise MastodonError("Post text cannot be empty") payload = { "status": text, "language": "en", "media_ids": list(media_ids), } if in_reply_to_id: payload["in_reply_to_id"] = in_reply_to_id data = self._request("POST", "/api/v1/statuses", json=payload) status_id = data.get("id") status_url = data.get("url") if not status_id: raise MastodonError("Mastodon response missing status id") return {"id": status_id, "url": status_url} def _get_account_id(self) -> str: if not self._account_id: data = self._request("GET", "/api/v1/accounts/verify_credentials") account_id = data.get("id") if not account_id: raise MastodonError("Unable to determine account id") self._account_id = str(account_id) return self._account_id def get_latest_status(self) -> Optional[Dict[str, Any]]: account_id = self._get_account_id() params = {"limit": 1, "exclude_reblogs": True} items = self._request( "GET", f"/api/v1/accounts/{account_id}/statuses", params=params ) if isinstance(items, list) and items: status = items[0] return { "id": status.get("id"), "content": status.get("content"), "created_at": status.get("created_at"), "url": status.get("url"), } return None