"""Micro.blog client — Micropub posting for configured blogs. Micro.blog implements the Micropub standard for posting. FeedEcho connects with an app token (created at https://micro.blog/account/apps — tokens can post on the user's behalf but nothing else). One token may be able to post to several microblogs; GET /micropub?q=config lists them as "destination" entries whose "uid" is the blog URL. FeedEcho stores one account row per blog and targets it on each post with the mp-destination parameter. Posting is form-encoded (not JSON) because Micro.blog's photo handling follows the classic form flow: pass the feed item's image URL directly as photo=, with mp-photo-alt supplying alt text. No blob upload is needed on our side. """ from __future__ import annotations import logging import httpx from feed_parser import SSRFError, pinned_request from utils import DEFAULT_REQUEST_TIMEOUT, DestinationAuthError, DestinationError, json_error_detail logger = logging.getLogger(__name__) MICROPUB_ENDPOINT = "https://micro.blog/micropub" REQUEST_TIMEOUT = DEFAULT_REQUEST_TIMEOUT TOKEN_DISPLAY_MAX = 8 class MicroblogError(DestinationError): """Base error for Micro.blog API interactions.""" class MicroblogAuthError(MicroblogError, DestinationAuthError): """Token rejected, expired, or revoked.""" def _error_detail(response) -> str: """Extract a Micropub error message from a JSON error body.""" return json_error_detail(response, "error_description", "error") def _bearer(token: str) -> str: return f"Bearer {token.strip()}" def _is_auth_rejection(status_code: int) -> bool: """Whether the HTTP status means the token was rejected. Deliberately status-code-only: string-matching error bodies for "unauthorized"/"forbidden" on other statuses would misclassify transient server errors as permanent auth failures. """ return status_code in (401, 403) def fetch_config(token: str) -> dict: """Query GET /micropub?q=config with the given app token. Returns the parsed JSON config. Raises MicroblogAuthError when the token is rejected and MicroblogError for anything else (network, non-JSON body, unexpected status). """ try: resp = pinned_request( "GET", MICROPUB_ENDPOINT, timeout=REQUEST_TIMEOUT, params={"q": "config"}, headers={"Authorization": _bearer(token)}, follow_redirects=True, ) except (httpx.HTTPError, SSRFError) as e: raise MicroblogError(f"Could not reach micro.blog: {e}") from e detail = _error_detail(resp) if _is_auth_rejection(resp.status_code): raise MicroblogAuthError( "Micro.blog rejected the app token. Check the token and try again." ) if resp.status_code != 200: raise MicroblogError( f"Micro.blog config query failed (HTTP {resp.status_code})" + (f": {detail}" if detail else "") ) try: config = resp.json() except ValueError as e: raise MicroblogError("Micro.blog returned a non-JSON config response") from e if not isinstance(config, dict): raise MicroblogError("Micro.blog config response was not an object") return config def list_destinations(token: str) -> list[dict]: """Return the blogs a token can post to, as [{uid, name}]. The q=config "destination" key holds a list of blog objects; "uid" is the canonical blog URL used as mp-destination when posting, "name" the display name. Malformed entries are skipped rather than failing the whole connect. """ config = fetch_config(token) destinations = config.get("destination") if not isinstance(destinations, list): raise MicroblogError( "Micro.blog config listed no blogs for this token." ) blogs: list[dict] = [] seen: set[str] = set() for entry in destinations: if not isinstance(entry, dict): continue uid = entry.get("uid") if not isinstance(uid, str) or not uid.strip(): continue uid = uid.strip() if uid in seen: continue seen.add(uid) name = entry.get("name") if not isinstance(name, str) or not name.strip(): name = uid blogs.append({"uid": uid, "name": name.strip()[:200]}) if not blogs: raise MicroblogError( "Micro.blog config listed no usable blogs for this token." ) return blogs def create_post( token: str, content: str, destination: str = "", photo_url: str = "", photo_alt: str = "", name: str = "", ) -> dict: """Create an h=entry post via form-encoded Micropub. destination is the blog uid from list_destinations (mp-destination); when empty, Micro.blog posts to the account's default blog. photo_url is passed through as photo= so Micro.blog fetches and hosts the image itself; photo_alt becomes mp-photo-alt. name gives the post a title (blank for a normal microblog entry). Returns the parsed Location/JSON response. Micro.blog answers 201 or 202 with a Location header on success; both are accepted. """ if not content.strip(): raise MicroblogError("Cannot post empty content to Micro.blog") data: dict[str, str] = {"h": "entry", "content": content} if destination: data["mp-destination"] = destination if photo_url: data["photo"] = photo_url if photo_alt: data["mp-photo-alt"] = photo_alt if name: data["name"] = name try: resp = pinned_request( "POST", MICROPUB_ENDPOINT, timeout=REQUEST_TIMEOUT, data=data, headers={"Authorization": _bearer(token)}, follow_redirects=True, ) except (httpx.HTTPError, SSRFError) as e: raise MicroblogError(f"Could not reach micro.blog: {e}") from e detail = _error_detail(resp) if _is_auth_rejection(resp.status_code): raise MicroblogAuthError( "Micro.blog rejected the app token. Reconnect the account." ) if resp.status_code not in (200, 201, 202): raise MicroblogError( f"Micro.blog post failed (HTTP {resp.status_code})" + (f": {detail}" if detail else "") ) result: dict = {} location = resp.headers.get("Location", "") if location: result["location"] = location try: body = resp.json() if isinstance(body, dict): result.update(body) except ValueError: pass return result def test_connection(token: str) -> tuple[bool, str]: """Verify a token still works by querying config. For the Test button.""" try: blogs = list_destinations(token) except MicroblogAuthError as e: return False, str(e) except MicroblogError as e: return False, str(e) names = ", ".join(b["name"] for b in blogs) return True, f"Token OK — can post to: {names}"