Lets use Python to automatically tweet recent posts from wordpress
For discovery purposes, I chose to build a blog with headless wordpress, React, GraphQL/Apollo. I would like to find a way to automate these posts across all platforms as much as possible.
I think we could come up with a solution to get posts from the wordpress API and and send a simple tweet about the new article. After thinking about this, I have realized that we will need to store the ID of the post that is selected to prevent retweets of the same post.
I will walk you through the first messy version of automating this task and maybe we can clean it up a bit along the way.
First things first, we have Twython as a dependency to interact with twitter's API.
Now lets create a class that interact with Wordpress. For now, we will grab all posts and store in a json file.
Wordpress.py
class WPGateway:
def __init__(self):
self.url = os.environ.get("WP_POSTS")
def get_posts(self):
response = requests.get(self.url, verify=False)
wp_content = response.json()
return wp_content
def get_image(self, img_data):
url = [x["href"] for x in img_data]
r = requests.get(url[0])
if r.ok:
data = json.loads(r.content.decode('UTF-8'))
image_url = data['guid']['rendered']
return image_url
Tweeter.py
import json
import csv
import logging
import requests
import urllib3
urllib3.disable_warnings()
logger = logging.getLogger(__name__)
wp = WPGateway()
if __name__ == "__main__":
logging.basicConfig(
format="%(levelname)s: %(message)s",
level=logging.INFO, stream=sys.stdout
)
with open("posts.json", "w") as write_file:
data = wp.get_posts()
json.dump(data, write_file, indent=4)
So this is our main script which calls get_posts() which will populate a json file with the given response from the rest api.
The schema of a wordpress response is consistent by default unless there have been custom modifications to the api with custom post types, taxonomies or so on. So parsing this Json should be similiar to your experience if not the same.
[
{
"id": 15,
"date": "2019-10-13T14:48:17",
"date_gmt": "2019-10-13T14:48:17",
"guid": {
"rendered": "https://{domain}.com/?p=15"
},
"modified": "2019-10-13T17:13:35",
"modified_gmt": "2019-10-13T17:13:35",
"slug": "{title}",
"status": "publish",
"type": "post",
"link": "https://{domain}.com/{title}/",
"title": {
"rendered": "{title}"
},
"content": {
"rendered": "\n<p>Main content</p>,
"protected": false
},
"excerpt": {
"rendered": "<p>Excerpt Content.</p>\n",
"protected": false
},
"author": 2,
"featured_media": 16,
"comment_status": "open",
"ping_status": "open",
"sticky": false,
"template": "",
"format": "standard",
"meta": [],
"categories": [
2
],
"tags": [
16
],
"acf": [],
"_links": {
"self": [
{
"href": "https://{domain}.com/wp-json/wp/v2/posts/{id}"
}
],
"collection": [
{
"href": "https://{domain}.com/wp-json/wp/v2/posts"
}
],
"about": [
{
"href": "https://{domain}.com/wp-json/wp/v2/types/post"
}
],
"author": [
{
"embeddable": true,
"href": "https://{domain}.com/wp-json/wp/v2/users/{amount}"
}
],
"replies": [
{
"embeddable": true,
"href": "https://{domain}.com/wp-json/wp/v2/comments?post={id}"
}
],
"version-history": [
{
"count": 38,
"href": "https://{domain}.com/wp-json/wp/v2/posts/{id}/revisions"
}
],
"predecessor-version": [
{
"id": 60,
"href": "https://{domain}.com/wp-json/wp/v2/posts/{id}/revisions/60"
}
],
"wp:featuredmedia": [
{
"embeddable": true,
"href": "https://{domain}.com/wp-json/wp/v2/media/16"
}
],
"wp:attachment": [
{
"href": "https://{domain}.com/wp-json/wp/v2/media?parent={id}"
}
],
"wp:term": [
{
"taxonomy": "category",
"embeddable": true,
"href": "https://{domain}.com/wp-json/wp/v2/categories?post={id}"
},
{
"taxonomy": "post_tag",
"embeddable": true,
"href": "https://{domain}.com/wp-json/wp/v2/tags?post={id}"
}
],
"curies": [
{
"name": "wp",
"href": "https://api.w.org/{rel}",
"templated": true
}
]
},
"_embedded": {
"author": [
{
"id": 2,
"name": "Your name",
"url": "",
"description": "",
"link": "https://{domain}.com/author/admin/",
"slug": "admin",
"avatar_urls": {
"24": "https://secure.gravatar.com/avatar/57323407ed9d034148c1489a370970cd?s=24&d=mm&r=g",
"48": "https://secure.gravatar.com/avatar/57323407ed9d034148c1489a370970cd?s=48&d=mm&r=g",
"96": "https://secure.gravatar.com/avatar/57323407ed9d034148c1489a370970cd?s=96&d=mm&r=g"
},
"acf": [],
"_links": {
"self": [
{
"href": "https://{domain}.com/wp-json/wp/v2/users/2"
}
],
"collection": [
{
"href": "https://{domain}.com/wp-json/wp/v2/users"
}
]
}
}
],
"wp:featuredmedia": [
{
"id": 16,
"date": "2019-10-13T14:46:52",
"slug": "slug",
"type": "attachment",
"link": "https://{domain}.com/title",
"title": {
"rendered": "title"
},
"author": 2,
"acf": [],
"caption": {
"rendered": "<p>title</p>\n"
},
"alt_text": "slack with django",
"media_type": "image",
"mime_type": "image/png",
"media_details": {
"width": 512,
"height": 256,
"file": "2019/10/postimage.png",
"sizes": {
"thumbnail": {
"file": "postimage-150x150.png",
"width": 150,
"height": 150,
"mime_type": "image/png",
"source_url": "https://{domain}.com/wp-content/uploads/2019/10/slackdjango-150x150.png"
},
"medium": {
"file": "postimage-300x150.png",
"width": 300,
"height": 150,
"mime_type": "image/png",
"source_url": "https://{domain}.com/wp-content/uploads/2019/10/slackdjango-300x150.png"
},
"full": {
"file": "postimage.png",
"width": 512,
"height": 256,
"mime_type": "image/png",
"source_url": "https://teaglebuilt.com/wp-content/uploads/2019/10/slackdjango.png"
}
},
"image_meta": {
"aperture": "0",
"credit": "",
"camera": "",
"caption": "",
"created_timestamp": "0",
"copyright": "",
"focal_length": "0",
"iso": "0",
"shutter_speed": "0",
"title": "",
"orientation": "0",
"keywords": []
}
},
"source_url": "https://{domain}.com/wp-content/uploads/2019/10/slackdjango.png",
"_links": {
"self": [
{
"href": "https://{domain}.com/wp-json/wp/v2/media/16"
}
],
"collection": [
{
"href": "https://{domain}.com/wp-json/wp/v2/media"
}
],
"about": [
{
"href": "https://{domain}.com/wp-json/wp/v2/types/attachment"
}
],
"author": [
{
"embeddable": true,
"href": "https://{domain}.com/wp-json/wp/v2/users/2"
}
],
"replies": [
{
"embeddable": true,
"href": "https://{domain}.com/wp-json/wp/v2/comments?post=16"
}
]
}
}
],
"wp:term": [
[
{
"id": 2,
"link": "https://{domain}.com/category/back_end/",
"name": "Back End",
"slug": "back_end",
"taxonomy": "category",
"acf": [],
"_links": {
"self": [
{
"href": "https://teaglebuilt.com/wp-json/wp/v2/categories/2"
}
],
"collection": [
{
"href": "https://{domain}.com/wp-json/wp/v2/categories"
}
],
"about": [
{
"href": "https://{domain}t.com/wp-json/wp/v2/taxonomies/category"
}
],
"wp:post_type": [
{
"href": "https://{domain}.com/wp-json/wp/v2/posts?categories=2"
},
{
"href": "https://{domain}t.com/wp-json/wp/v2/projects?categories=2"
}
],
"curies": [
{
"name": "wp",
"href": "https://api.w.org/{rel}",
"templated": true
}
]
}
}
],
[
{
"id": 16,
"link": "https://{domain}.com/tag/python/",
"name": "Python",
"slug": "python",
"taxonomy": "post_tag",
"_links": {
"self": [
{
"href": "https://{domain}.com/wp-json/wp/v2/tags/16"
}
],
"collection": [
{
"href": "https://{domain}.com/wp-json/wp/v2/tags"
}
],
"about": [
{
"href": "https://{domain}.com/wp-json/wp/v2/taxonomies/post_tag"
}
],
"wp:post_type": [
{
"href": "https://{domain}.com/wp-json/wp/v2/posts?tags=16"
},
{
"href": "https://{domain}.com/wp-json/wp/v2/projects?tags=16"
}
],
"curies": [
{
"name": "wp",
"href": "https://api.w.org/{rel}",
"templated": true
}
]
}
}
]
]
}
}
]
So this file will always hold our collection of data from wordpress, great!. We need to grab the "ID" of every post and IF it has not yet been TWEETED, then we will do so.
One quick and simple solution is to add the id of the post to a csv file that has been selected to be tweeted. This way we can keep a history collection of posts by the id.
id_records.txt
1,2,3
So at this point we opened the json file with the context manager and stored our response schema. So lets grab the Id's of already posted content in the txt file.
def record_check():
with open("social/id_record.txt", "r") as id_file:
reader = csv.reader(id_file)
for row in reader:
print(row)
return row
['1', '2', '3']
We will filter the posts by the ids and return an eligible list of new posts for possible tweediness.
def eligible_posts(records):
post = {}
eligible = []
read_json = open("social/wp_posts.json", "r")
data = json.load(read_json)
for index in range(len(data)):
if str(data[index]["id"]) not in records:
wp_img = wp.get_image(data[index]["_links"]["wp:featuredmedia"])
post["id"] = data[index]["id"]
post['title'] = data[index]['title']['rendered']
post['image'] = wp_img
eligible.append(post)
return eligible
Here we parse the json filter and format the posts for publishing. This function will return a list of objects where the posts have not been sent to twitter. Inside this object we call a function called get_image. This is important because you have to download the image to send to twitter. I was suprised you could not just send the source url. This method has been added to the wordpress class.
Here is the complete class as of now.
wordpress.py
class WPGateway:
def __init__(self):
self.url = os.environ.get("WP_POSTS")
def get_posts(self):
response = requests.get(self.url, verify=False)
wp_content = response.json()
return wp_content
def get_image(self, img_data):
url = [x["href"] for x in img_data]
r = requests.get(url[0])
if r.ok:
data = json.loads(r.content.decode('UTF-8'))
image_url = data['guid']['rendered']
return image_url
So now we have the image uri. At this point i thought I could just send the image_url with the post object like so...
[{'id': 15, 'title': 'Post title', 'image': 'https://{domain}.com/wp-content/uploads/2019/10/image.png'}]]
INFO: Uploading image
['6', '7', '8', 12] records
This is the list of eligible posts. This is tedious, but we have to download the image to upload to twitter, then delete the image from your path.
def upload_image(url):
r = requests.get(url, stream=True)
with open('social/img.png', 'wb') as out_file:
shutil.copyfileobj(r.raw, out_file)
photo = open('social/img.png', 'rb')
media = twitter.upload_media(media=photo)
logger.info("Uploading image")
return media
Now I will stop right here and setup your credentials for authenticating to twitter.
twitter = Twython(os.getenv("CONSUMER_KEY"),
os.getenv("CONSUMER_SECRET"),
os.getenv("ACCESS_TOKEN"),
os.getenv("ACCESS_TOKEN_SECRET")
)
Im going to go ahead and post the complete tweeter.py script and well continue to break it down.
from twython import Twython, TwythonError
from wordpress import WPGateway
import json
import csv
import shutil
import os, sys
import logging
import requests
import urllib3
urllib3.disable_warnings()
logger = logging.getLogger(__name__)
wp = WPGateway()
twitter = Twython(os.getenv("CONSUMER_KEY"),
os.getenv("CONSUMER_SECRET"),
os.getenv("ACCESS_TOKEN"),
os.getenv("ACCESS_TOKEN_SECRET")
)
def record_check():
with open("id_record.txt", "r") as id_file:
reader = csv.reader(id_file)
for row in reader:
print(row)
return row
def update_records(records, post):
records.append(post["id"])
print(records, "records")
with open("id_record.txt", "w", newline="") as id_file:
id_writer = csv.writer(id_file)
id_writer.writerow(records)
def eligible_posts(records):
post = {}
eligible = []
read_json = open("wp_posts.json", "r")
data = json.load(read_json)
for index in range(len(data)):
if str(data[index]["id"]) not in records:
wp_img = wp.get_image(data[index]["_links"]["wp:featuredmedia"])
post["id"] = data[index]["id"]
post['title'] = data[index]['title']['rendered']
post['image'] = wp_img
eligible.append(post)
return eligible
def upload_image(url):
r = requests.get(url, stream=True)
with open('img.png', 'wb') as out_file:
shutil.copyfileobj(r.raw, out_file)
photo = open('img.png', 'rb')
media = twitter.upload_media(media=photo)
logger.info("Uploading image")
return media
if __name__ == "__main__":
logging.basicConfig(
format="%(levelname)s: %(message)s",
level=logging.INFO, stream=sys.stdout
)
with open("wp_posts.json", "w") as write_file:
data = wp.get_posts()
json.dump(data, write_file, indent=4)
records = record_check()
new_posts = eligible_posts(records)
print(new_posts)
if len(new_posts) >= 1:
uploaded_media = upload_image(new_posts[0]["image"])
twitter.update_status(
status=f"Visit https://www.{domain}.com, to see recent posts like {new_posts[0]['title']}.",
media_ids=[uploaded_media['media_id']]
)
update_records(records, new_posts[0])
os.remove('img.png')
else:
logger.info("There are no new posts")
You can see the additional handlers for updating records -> the txt file.
Finally remove the image from your path. I wonder if you could use Python's tempdir and still capture the image for upload?
There is plenty of room to simplify this code. Even the idea of recording ID's to a txt file is sloppy. In the next post, lets consolidate this code and add a github action to trigger this script periodically.
Oldest comments (0)