matitos_news/utils/Ghost-Posts.ipynb

In [ ]:

import os
import time
import jwt
import requests
from datetime import datetime, timedelta, timezone

admin_api_url = "" # .env -> GHOST_ADMIN_API_URL
admin_api_key = "" # .env -> GHOST_ADMIN_API_KEY

def _create_jwt(admin_api_key):
    id_, secret = admin_api_key.split(':')
    iat = int(time.time())
    exp = iat + 5 * 60  # 5 minutes
    header = {'alg': 'HS256', 'kid': id_}
    payload = {
        'iat': iat,
        'exp': exp,
        'aud': '/v5/admin/'  # Adjust depending on your Ghost version
    }
    token = jwt.encode(payload, bytes.fromhex(secret), algorithm='HS256', headers=header)
    return token

# Get token
jwt_token = _create_jwt(os.getenv("GHOST_ADMIN_API_KEY"))

headers = {
    'Authorization': f'Ghost {jwt_token}',
    'Content-Type': 'application/json'
}

In [ ]:

DELETE_ALL_POSTS = False

if DELETE_ALL_POSTS:
    while (True):
        # GET /admin/posts/
        response = requests.get(os.path.join(admin_api_url, "posts"), headers=headers)
        dict_response = response.json()

        if (len(dict_response.get("posts")) == 0):
            break

        # Iterate posts
        for p in dict_response.get("posts"):
            # Post ID
            post_id = p.get("id")

            # DELETE /admin/posts/{id}/
            r = requests.delete(os.path.join(admin_api_url, "posts", "{}".format(post_id)), headers=headers)
            print("Post:", post_id, "Status:", r.status_code, r.text)

In [ ]:

PUBLISH_SAMPLE = False

def _create_ghost_post(jwt_token, admin_api_url, post_data):
    # Get Admin API URL
    admin_api_url = os.getenv("GHOST_ADMIN_API_URL")

    headers = {
        'Authorization': f'Ghost {jwt_token}',
        'Content-Type': 'application/json'
    }
    
    post_data = {"posts": [post_data]}

    response = requests.post(
        os.path.join(admin_api_url, "posts"),
        json=post_data,
        headers=headers,
        params={"source":"html"}
    )

    if response.status_code == 201:
        print("Ghost post published successfully")
        return response.json()
    else:
        print("Ghost - Failed to publish post: {} {}".format(response.status_code, response.text))
        return None

if (PUBLISH_SAMPLE):
    url_id = 150

    post_data = {
        # "slug": "hey-short",
        "title": "Hey there, sample title",
        "html": "<p>Hey there!</p>",
        # "feature_image": photo_url,
        # "feature_image_caption": "",
        "status": "published",
        "tags": ["#url-id-{}".format(url_id)]
    }

    # Publish post
    payload = _create_ghost_post(jwt_token, admin_api_url, post_data)

In [ ]:

# Filter by post title
post_title = "Funds raised for legal action over failure to stop grooming gangs"
# Filter by published date
iso_time = (datetime.now(timezone.utc) - timedelta(hours=48)).strftime('%Y-%m-%dT%H:%M:%S') + 'Z'
# Parameter for filter
params = {"filter": "title:'{}'+published_at:>{}".format(post_title, iso_time)}

# Filter by URL ID
url_id = 150
# Parameter for filter
params = {"filter": "tags:hash-url-id-{}".format(url_id)}

# Get posts using filter
response = requests.get(os.path.join(admin_api_url, "posts"), params=params, headers=headers)
dict_response = response.json()

len(dict_response.get("posts"))