89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
"""Client helpers for posting to Mastodon."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import List, 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)",
|
|
}
|
|
)
|
|
|
|
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 upload_media(
|
|
self, filename: str, data: bytes, mime_type: str, alt_text: str
|
|
) -> str:
|
|
url = f"{self.base_url}/api/v2/media"
|
|
files = {
|
|
"file": (
|
|
filename or "photo.jpg",
|
|
data,
|
|
mime_type or "application/octet-stream",
|
|
)
|
|
}
|
|
form = {}
|
|
if alt_text:
|
|
form["description"] = alt_text
|
|
try:
|
|
response = self.session.post(url, files=files, data=form, timeout=30)
|
|
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)
|
|
payload = response.json()
|
|
|
|
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]) -> str:
|
|
if not text.strip():
|
|
raise MastodonError("Post text cannot be empty")
|
|
url = f"{self.base_url}/api/v1/statuses"
|
|
payload = {
|
|
"status": text,
|
|
"language": "en",
|
|
"media_ids": list(media_ids),
|
|
}
|
|
try:
|
|
response = self.session.post(url, json=payload, timeout=30)
|
|
except requests.RequestException as exc: # pragma: no cover
|
|
raise MastodonError(str(exc)) from exc
|
|
if not response.ok:
|
|
self._raise_for_error(response)
|
|
data = response.json()
|
|
|
|
status_id = data.get("id")
|
|
if not status_id:
|
|
raise MastodonError("Mastodon response missing status id")
|
|
return status_id
|