rose-ash/blog/bp/blog/ghost/ghost_posts.py

"""
Ghost Admin API – post CRUD.

Uses the same JWT auth and httpx patterns as ghost_sync.py.
"""
from __future__ import annotations

import logging
import os

import httpx

from .ghost_admin_token import make_ghost_admin_jwt

log = logging.getLogger(__name__)

GHOST_ADMIN_API_URL = os.environ["GHOST_ADMIN_API_URL"]


def _auth_header() -> dict[str, str]:
    return {"Authorization": f"Ghost {make_ghost_admin_jwt()}"}


def _check(resp: httpx.Response) -> None:
    """Raise with the Ghost error body so callers see what went wrong."""
    if resp.is_success:
        return
    body = resp.text[:2000]
    log.error("Ghost API %s %s → %s: %s", resp.request.method, resp.request.url, resp.status_code, body)
    resp.raise_for_status()


async def get_post_for_edit(ghost_id: str, *, is_page: bool = False) -> dict | None:
    """Fetch a single post/page by Ghost ID, including lexical source."""
    resource = "pages" if is_page else "posts"
    url = (
        f"{GHOST_ADMIN_API_URL}/{resource}/{ghost_id}/"
        "?formats=lexical,html,mobiledoc&include=newsletters"
    )
    async with httpx.AsyncClient(timeout=30) as client:
        resp = await client.get(url, headers=_auth_header())
        if resp.status_code == 404:
            return None
        _check(resp)
    return resp.json()[resource][0]


async def create_post(
    title: str,
    lexical_json: str,
    status: str = "draft",
    feature_image: str | None = None,
    custom_excerpt: str | None = None,
    feature_image_caption: str | None = None,
) -> dict:
    """Create a new post in Ghost.  Returns the created post dict."""
    post_body: dict = {
        "title": title,
        "lexical": lexical_json,
        "mobiledoc": None,
        "status": status,
    }
    if feature_image:
        post_body["feature_image"] = feature_image
    if custom_excerpt:
        post_body["custom_excerpt"] = custom_excerpt
    if feature_image_caption is not None:
        post_body["feature_image_caption"] = feature_image_caption
    payload = {"posts": [post_body]}
    url = f"{GHOST_ADMIN_API_URL}/posts/"
    async with httpx.AsyncClient(timeout=30) as client:
        resp = await client.post(url, json=payload, headers=_auth_header())
        _check(resp)
    return resp.json()["posts"][0]


async def create_page(
    title: str,
    lexical_json: str,
    status: str = "draft",
    feature_image: str | None = None,
    custom_excerpt: str | None = None,
    feature_image_caption: str | None = None,
) -> dict:
    """Create a new page in Ghost (via /pages/ endpoint). Returns the created page dict."""
    page_body: dict = {
        "title": title,
        "lexical": lexical_json,
        "mobiledoc": None,
        "status": status,
    }
    if feature_image:
        page_body["feature_image"] = feature_image
    if custom_excerpt:
        page_body["custom_excerpt"] = custom_excerpt
    if feature_image_caption is not None:
        page_body["feature_image_caption"] = feature_image_caption
    payload = {"pages": [page_body]}
    url = f"{GHOST_ADMIN_API_URL}/pages/"
    async with httpx.AsyncClient(timeout=30) as client:
        resp = await client.post(url, json=payload, headers=_auth_header())
        _check(resp)
    return resp.json()["pages"][0]


async def update_post(
    ghost_id: str,
    lexical_json: str,
    title: str | None,
    updated_at: str,
    feature_image: str | None = None,
    custom_excerpt: str | None = None,
    feature_image_caption: str | None = None,
    status: str | None = None,
    newsletter_slug: str | None = None,
    email_segment: str | None = None,
    email_only: bool | None = None,
    is_page: bool = False,
) -> dict:
    """Update an existing Ghost post.  Returns the updated post dict.

    ``updated_at`` is Ghost's optimistic-locking token – pass the value
    you received from ``get_post_for_edit``.

    When ``newsletter_slug`` is set the publish request also triggers an
    email send via Ghost's query-parameter API:
    ``?newsletter={slug}&email_segment={segment}``.
    """
    post_body: dict = {
        "lexical": lexical_json,
        "mobiledoc": None,
        "updated_at": updated_at,
    }
    if title is not None:
        post_body["title"] = title
    if feature_image is not None:
        post_body["feature_image"] = feature_image or None
    if custom_excerpt is not None:
        post_body["custom_excerpt"] = custom_excerpt or None
    if feature_image_caption is not None:
        post_body["feature_image_caption"] = feature_image_caption
    if status is not None:
        post_body["status"] = status
    if email_only:
        post_body["email_only"] = True
    resource = "pages" if is_page else "posts"
    payload = {resource: [post_body]}

    url = f"{GHOST_ADMIN_API_URL}/{resource}/{ghost_id}/"
    if newsletter_slug:
        url += f"?newsletter={newsletter_slug}"
        if email_segment:
            url += f"&email_segment={email_segment}"
    async with httpx.AsyncClient(timeout=30) as client:
        resp = await client.put(url, json=payload, headers=_auth_header())
        _check(resp)
    return resp.json()[resource][0]


_SETTINGS_FIELDS = (
    "slug",
    "published_at",
    "featured",
    "visibility",
    "email_only",
    "custom_template",
    "meta_title",
    "meta_description",
    "canonical_url",
    "og_image",
    "og_title",
    "og_description",
    "twitter_image",
    "twitter_title",
    "twitter_description",
    "tags",
    "feature_image_alt",
)


async def update_post_settings(
    ghost_id: str,
    updated_at: str,
    is_page: bool = False,
    **kwargs,
) -> dict:
    """Update Ghost post/page settings (slug, tags, SEO, social, etc.).

    Only non-None keyword args are included in the PUT payload.
    Accepts any key from ``_SETTINGS_FIELDS``.
    """
    resource = "pages" if is_page else "posts"
    post_body: dict = {"updated_at": updated_at}
    for key in _SETTINGS_FIELDS:
        val = kwargs.get(key)
        if val is not None:
            post_body[key] = val

    payload = {resource: [post_body]}
    url = f"{GHOST_ADMIN_API_URL}/{resource}/{ghost_id}/"
    async with httpx.AsyncClient(timeout=30) as client:
        resp = await client.put(url, json=payload, headers=_auth_header())
        _check(resp)
    return resp.json()[resource][0]