132 lines
4.6 KiB
Python
132 lines
4.6 KiB
Python
"""Client helpers for posting to Mastodon."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, 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
|
|
|
|
def get_status(self, status_id: str) -> Dict[str, Any]:
|
|
if not status_id:
|
|
raise MastodonError("status_id is required")
|
|
return self._request("GET", f"/api/v1/statuses/{status_id}")
|
|
|
|
def get_status_ancestors(self, status_id: str) -> List[Dict[str, Any]]:
|
|
if not status_id:
|
|
raise MastodonError("status_id is required")
|
|
data = self._request("GET", f"/api/v1/statuses/{status_id}/context")
|
|
ancestors = data.get("ancestors")
|
|
if isinstance(ancestors, list):
|
|
return ancestors
|
|
return []
|