matitos_news/utils/Ghost-Posts.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import time\n",
    "import jwt\n",
    "import requests\n",
    "from datetime import datetime, timedelta, timezone\n",
    "\n",
    "admin_api_url = \"\" # .env -> GHOST_ADMIN_API_URL\n",
    "admin_api_key = \"\" # .env -> GHOST_ADMIN_API_KEY\n",
    "\n",
    "def _create_jwt(admin_api_key):\n",
    "    id_, secret = admin_api_key.split(':')\n",
    "    iat = int(time.time())\n",
    "    exp = iat + 5 * 60  # 5 minutes\n",
    "    header = {'alg': 'HS256', 'kid': id_}\n",
    "    payload = {\n",
    "        'iat': iat,\n",
    "        'exp': exp,\n",
    "        'aud': '/v5/admin/'  # Adjust depending on your Ghost version\n",
    "    }\n",
    "    token = jwt.encode(payload, bytes.fromhex(secret), algorithm='HS256', headers=header)\n",
    "    return token\n",
    "\n",
    "# Get token\n",
    "jwt_token = _create_jwt(os.getenv(\"GHOST_ADMIN_API_KEY\"))\n",
    "\n",
    "headers = {\n",
    "    'Authorization': f'Ghost {jwt_token}',\n",
    "    'Content-Type': 'application/json'\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "DELETE_ALL_POSTS = False\n",
    "\n",
    "if DELETE_ALL_POSTS:\n",
    "    while (True):\n",
    "        # GET /admin/posts/\n",
    "        response = requests.get(os.path.join(admin_api_url, \"posts\"), headers=headers)\n",
    "        dict_response = response.json()\n",
    "\n",
    "        if (len(dict_response.get(\"posts\")) == 0):\n",
    "            break\n",
    "\n",
    "        # Iterate posts\n",
    "        for p in dict_response.get(\"posts\"):\n",
    "            # Post ID\n",
    "            post_id = p.get(\"id\")\n",
    "\n",
    "            # DELETE /admin/posts/{id}/\n",
    "            r = requests.delete(os.path.join(admin_api_url, \"posts\", \"{}\".format(post_id)), headers=headers)\n",
    "            print(\"Post:\", post_id, \"Status:\", r.status_code, r.text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "PUBLISH_SAMPLE = False\n",
    "\n",
    "def _create_ghost_post(jwt_token, admin_api_url, post_data):\n",
    "    # Get Admin API URL\n",
    "    admin_api_url = os.getenv(\"GHOST_ADMIN_API_URL\")\n",
    "\n",
    "    headers = {\n",
    "        'Authorization': f'Ghost {jwt_token}',\n",
    "        'Content-Type': 'application/json'\n",
    "    }\n",
    "    \n",
    "    post_data = {\"posts\": [post_data]}\n",
    "\n",
    "    response = requests.post(\n",
    "        os.path.join(admin_api_url, \"posts\"),\n",
    "        json=post_data,\n",
    "        headers=headers,\n",
    "        params={\"source\":\"html\"}\n",
    "    )\n",
    "\n",
    "    if response.status_code == 201:\n",
    "        print(\"Ghost post published successfully\")\n",
    "        return response.json()\n",
    "    else:\n",
    "        print(\"Ghost - Failed to publish post: {} {}\".format(response.status_code, response.text))\n",
    "        return None\n",
    "\n",
    "if (PUBLISH_SAMPLE):\n",
    "    url_id = 150\n",
    "\n",
    "    post_data = {\n",
    "        # \"slug\": \"hey-short\",\n",
    "        \"title\": \"Hey there, sample title\",\n",
    "        \"html\": \"<p>Hey there!</p>\",\n",
    "        # \"feature_image\": photo_url,\n",
    "        # \"feature_image_caption\": \"\",\n",
    "        \"status\": \"published\",\n",
    "        \"tags\": [\"#url-id-{}\".format(url_id)]\n",
    "    }\n",
    "\n",
    "    # Publish post\n",
    "    payload = _create_ghost_post(jwt_token, admin_api_url, post_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Filter by post title\n",
    "post_title = \"Funds raised for legal action over failure to stop grooming gangs\"\n",
    "# Filter by published date\n",
    "iso_time = (datetime.now(timezone.utc) - timedelta(hours=48)).strftime('%Y-%m-%dT%H:%M:%S') + 'Z'\n",
    "# Parameter for filter\n",
    "params = {\"filter\": \"title:'{}'+published_at:>{}\".format(post_title, iso_time)}\n",
    "\n",
    "# Filter by URL ID\n",
    "url_id = 150\n",
    "# Parameter for filter\n",
    "params = {\"filter\": \"tags:hash-url-id-{}\".format(url_id)}\n",
    "\n",
    "# Get posts using filter\n",
    "response = requests.get(os.path.join(admin_api_url, \"posts\"), params=params, headers=headers)\n",
    "dict_response = response.json()\n",
    "\n",
    "len(dict_response.get(\"posts\"))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "matitos_urls",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}