今回の内容
これまでの記事では、WordPressにPythonから自動で投稿する方法として、記事の投稿方法や、記事に必要なメタデータの管理方法、サムネイルの作り方などについてみてきました。
今回は、実際にメタデータをまとめたCSVをもとに、WordPressに記事を投稿する一連のコードを作成します。
処理の流れ
基本的な流れは次のようになります。
1. 記事データマークダウンの読み込みとメタデータCSV作成
2. CSVを読み込んで1記事ずつ処理
1. メタデータの読み込み
2. サムネイルの作成と画像アップロード
3. 記事をマークダウンからHTMLに変換
4. 記事内部の画像のアップロード
5. 記事の投稿
必要なライブラリのインポートと変数の設定
import pandas as pd
import time
from PIL import Image, ImageDraw, ImageFont
from wordpress_xmlrpc import Client
import shutil
import os
import numpy as np
from dotenv import load_dotenv
import datetime
from datetime import timedelta
import pytz
import httplib2
import markdown
from IPython.display import HTML, display
from apiclient.discovery import build
from oauth2client import tools, file, client
import json
from bs4 import BeautifulSoup
from wordpress_xmlrpc import Client, WordPressPost, WordPressPage
from wordpress_xmlrpc.methods import media, posts
from wordpress_xmlrpc.methods.posts import NewPost,GetPost, EditPost
from wordpress_xmlrpc.methods.media import UploadFile,GetMediaItem
import re
import shutil
import glob
import sys
import requests
投稿システムの利用において、設定すべき変数は以下になります。
from dotenv import load_dotenv
load_dotenv() #.envの読み込み
# ローカル環境用の設定や処理をここに記述
post_data_path = 'wp_data.csv'
folder_base = '/workspaces/マイドライブ/blog/'
img_folder_base = '/workspaces/マイドライブ/blog/images/'
base = '/workspaces/マイドライブ/'
上記の例では、投稿のメタデータファイルはwp_data.csvに保存、投稿記事のマークダウンファイルは/workspaces/マイドライブ/blog/以下にカテゴリごとにフォルダを作成して保存。画像は、 ‘/workspaces/マイドライブ/blog/images/’以下にカテゴリごとに保存しています。
.envファイルを別途作成して、WordPressの認証情報を記述します。
wordpress_id = 'id'
wordpress_pw = 'pass'
投稿関連の設定は以下です。
priority = 'markdown'
update_post_data = True #メタファイルの更新を行うか
overwrite = False #画像の上書きをするか
# 日本標準時 (JST) のタイムゾーンを設定
japan_tz = pytz.timezone('Asia/Tokyo')
# 現在の日時を日本標準時で取得
now = datetime.now(japan_tz)
# 日時を指定して、時間を12:00:00に設定
date = now.replace(hour=12, minute=0, second=0, microsecond=0)
status = 'future' #"draft", "publish",'future
base_image_path = "base_image.png"
color1 = '#E36F24'#サムネイル作成時に強調する文字の色
base_url = "https://" #自分のドメイン
upload_type=1 #画像のアップデート方法0 アップロードしない upload_type:1 ファイルがない場合のみupload upload_type 2 すべて
マークダウンの読み込みとCSV作成
まずは、マークダウンの内容を読み込んで、CSVを作成するコードを作成します。
最初に、記事を保存しているフォルダにあるマークダウンファイルをすべて読み込む関数を作成します。
def get_md_files(base_path, folders):
markdown_files = []
for folder in folders:
folder_path = os.path.join(base_path, folder)
# 再帰的にフォルダー内のすべてのMarkdownファイルを検索
for filepath in glob.glob(os.path.join(folder_path, '**', '*.md'), recursive=True):
# 絶対パスに変換してリストに追加
markdown_files.append(os.path.abspath(filepath))
return markdown_files
get_md_files('blog/', folders=['category1','category2'])
上記のコードは、blog/category1とblog/category2にあるマークダウンファイルの一覧を取得する関数です。
続いて、マークダウンファイルを読み込んで、ファイル上部に記述されているメタデータを読み込む関数を作成します。ここでは、
nopostと書かれていた場合には、投稿しないデータとして、処理を終了、
[“title:”, “title_png:”, “update_post:”,”meta:”]
のいずれかが書かれていた場合には、メタデータが存在するとして、メタデータの解析。解析対象のメタデータは
[“title:”, “title_png:”, “update_post:”,”meta:”,”tage”,”colors”,”update_post”]としています。
必要に応じて、項目を減らしたり、削除したりしてください。
含まれない場合は、メタデータはなく、すべて投稿文章と判断
する関数になっています。
def parse_metadata(markdown_file_path):
with open(markdown_file_path, 'r', encoding='utf-8') as file:
lines = file.readlines()
# メタデータ部分と本文部分を分けるためのインデックスを見つける
metadata_end_index = 0
metadata_found = False
for i, line in enumerate(lines):
if "nopost" in line:
return False, "", "", "", "", "","", ""
if any(keyword in line for keyword in ["title:", "title_png:", "update_post:","meta:"]):
metadata_found = True
if line.strip() == "":
metadata_end_index = i
break
# メタデータが存在しない場合、すべてを本文として扱う
if not metadata_found:
text = "".join(lines).strip()
return True, "", "", "", "", "","", text
# メタデータ部分と本文部分を分ける
metadata_lines = lines[:metadata_end_index]
text_lines = lines[metadata_end_index + 1:] # 空行の後から本文
# メタデータを解析
metadata = {}
for line in metadata_lines:
if ':' not in line:
print(markdown_file_path)
key, value = line.split(":", 1)
metadata[key.strip()] = value.strip()
# 必要なメタデータを変数に格納
title = metadata.get("title", "").strip()
meta = metadata.get("meta", "").strip()
tags = metadata.get("tags", "").strip()
title_png = metadata.get("title_png", "").strip().replace('\\n', '\n')
colors = metadata.get("colors", "").strip()
update_post = metadata.get("update_post", "").strip()
# 本文を取得
text = "".join(text_lines).strip()
return True, title, title_png, colors, meta, tags, update_post, text
これらの関数を以下のように実行します。
if update_post_data:
df = pd.read_csv(post_data_path,index_col=0)
markdown_files = get_md_files(base_path=folder_base)
df = update_post_data(df, markdown_files,priority=priority)
df.to_csv(post_data_path)
df = pd.read_csv(post_data_path)
更新するかのupdate_post_dataに応じて、メタデータのファイルを更新し、読み込んでdfに格納します。
WordPress認証上の設定
以下のコードでWordPressの投稿に必要な認証情報を設定します。
# Set URL, ID, Password
wordpress_id = os.environ['wordpress_id']
wordpress_password = os.environ['wordpress_pw']
wp = Client(base_url+'xmlrpc.php', wordpress_id, wordpress_password)
記事の投稿
メタデータを取得する関数
def get_params(row,base_image_path = "base_image.png",folder_base = base_path + 'blog/',img_folder_base = base_path + 'blog/images/'):
row = row.fillna('')
color1 = '#E36F24'
title = row['title']
categories = row['categories'].split(', ')
folder = os.path.join(folder_base, *categories)
img_folder = os.path.join(img_folder_base, *categories)
title_png = row['title_png']
if title =='':
title = title_png
link_name = row['link_name'].lower()
colors = row['colors'].replace(' ','').split(',')
tags = row['tags'].split(', ')
meta_data = {'title':title, 'description':row['meta description'],'keywords':", ".join(categories + tags)}
path = os.path.join(folder ,link_name + '.md')
post_id = row['post_id']
if type(post_id) == np.float64:
post_id = int(post_id)
color_words = {}
for cw in colors:
color_words[cw] = color1
if row['update_post'] == 1 or row['update_post'] == '1' or row['update_post']=='True':
update_post = True
else:
update_post = False
return title, path, title_png, color_words, tags,categories, link_name, img_folder,post_id, meta_data, update_post
MarkdownをHTMLに変換する関数
def convert_to_html_and_display(text, extensions=None,configs={}):
if extensions is None:
extensions = ['fenced_code', 'tables',
'abbr','attr_list','def_list','footnotes',
'admonition','nl2br','sane_lists', 'toc','mdx_math','extra'
]
if configs=={}:
configs = {
'codehilite':{
'noclasses': True
},
'toc': {
'title': '目次',
'toc_depth':3, #どの階層まで表示するか
},
# 'mdx_math': {'enable_dollar_delimiter': True}
}
html = markdown.markdown(text, extensions=extensions, extension_configs=configs)
return html
画像投稿の関数
def get_media_data(wp, media_id):
response = wp.call(GetMediaItem(media_id))
metadata = response.metadata
return metadata
def wp_upload_image(wp, img_path, out_img_name=None, overwrite=True, upload_type=1):
'''
upload_type:0 アップロードしない
upload_type:1 ファイルがない場合のみupload
upload_type 2 すべて
'''
if upload_type == 0:
media_items = wp.call(media.GetMediaLibrary({'mime_type': 'image/png'}))
for item in media_items:
if item.metadata['file'] == out_img_name:
return item.id
return None
if out_img_name is None:
out_img_name = os.path.basename(img_path)
if os.path.exists(img_path):
# 既存の同名画像を削除
if overwrite:
media_items = wp.call(media.GetMediaLibrary({'mime_type': 'image/png'}))
for item in media_items:
if item.metadata['file'] == out_img_name:
if upload_type == 2:
check = input(f'{out_img_name}はすでに存在します。上書きしますか？（y/n/p）')
if check == 'y' or check == 'Y':
wp.call(posts.DeletePost(item.id))
print(f"{out_img_name} の既存ファイルを削除しました。ID: {item.id}")
elif check == 'p':
return item.id
else:
return item.id
with open(img_path, 'rb') as f:
binary = f.read()
data = {
"name": out_img_name,
"type": 'image/png', # 画像のMIMEタイプ
"bits": binary
}
response = wp.call(media.UploadFile(data))
media_id = response['id']
print(f"{out_img_name} のアップロードに成功しました。ID: {media_id}")
return media_id
else:
print(f"{out_img_name} が見つかりません")
return None
記事の投稿部分の関数
記事の画像の処理をまとめて行う
def copy_text_wp(wp,html,img_folder,upload_type=1):
# Beautiful SoupでHTMLを解析
soup = BeautifulSoup(html, 'html.parser')
media_ids = []
non_img = False
for img_tag in soup.find_all('img'):
# imgタグの親要素が<pre>または<code>タグでない場合にのみ処理を行う
if img_tag.find_parent(['pre', 'code']) is None:
if img_tag.get('src')[:4]!='http':
img_file = os.path.basename(img_tag['src'])
img_path = os.path.join(img_folder, img_file)
media_id = wp_upload_image(wp, img_path, overwrite=True,upload_type=upload_type)
if media_id is not None:
media_ids.append(media_id)
meta_data = get_media_data(wp, media_ids[-1])
sizes = meta_data['sizes']
if 'large' in sizes.keys():
new_file_name = sizes['large']['file']
else:
new_file_name = meta_data['file']
img_tag['src'] = 'https://○○.com/img/' + new_file_name
img_tag['class'] = f'wp-image-{media_id}'
else:
raise ValueError
# 新しいHTMLを表示
# display(HTML(str(soup)))
if non_img:
return None
# import pyperclip
# pyperclip.copy(str(soup))
return str(soup), media_ids
記事の投稿
def post_to_wp(wp, title, body, categories=[], tags=[], status = "draft",eye_catch_img_id=None, link_name=None,meta_data=None, date=None, comment_status=False):
#publish
# Post
post = WordPressPost()
post.title = title
post.content = body
post.post_status = status
post.comment_status=comment_status
if tags == ['']:
post.terms_names = {
"category": categories,
}
else:
post.terms_names = {
"category": categories,
"post_tag": tags,
}
if meta_data:
post.custom_fields = [
{'key': 'the_page_seo_title', 'value': meta_data['title']},
{'key': 'the_page_meta_description', 'value': meta_data['description']},
{'key': 'the_page_meta_keywords', 'value': meta_data['keywords']},
]
post.slug = link_name
# Set eye-catch image
if eye_catch_img_id:
post.thumbnail = eye_catch_img_id
# Post Time
# post.date = datetime.datetime.now() - datetime.timedelta(hours=9)
if date is not None:
post.date = date
post_id =wp.call(NewPost(post))
return post_id
記事の更新
def update_wp_post(wp, post_id, title=None, body=None, categories=[], tags=[], status=None, eye_catch_img_id=None, link_name=None, meta_data=None, date=None, comment_status=None):
# Get the existing post
post = wp.call(GetPost(post_id))
# Update title if provided
if title is not None:
post.title = title
# Update body if provided
if body is not None:
post.content = body
# Update status if provided
if status is not None:
post.post_status = status
# Update comment status if provided
if comment_status is not None:
post.comment_status = comment_status
# Update categories and tags
if tags == [''] or tags == [] and categories != []:
post.terms_names = {
"category": categories,
}
elif categories != []:
post.terms_names = {
"category": categories,
"post_tag": tags,
}
# Update meta data if provided
if meta_data:
post.custom_fields = [
{'key': 'the_page_seo_title', 'value': meta_data.get('title', '')},
{'key': 'the_page_meta_description', 'value': meta_data.get('description', '')},
{'key': 'the_page_meta_keywords', 'value': meta_data.get('keywords', '')},
]
# Update link name if provided
if link_name is not None:
post.slug = link_name
# Update eye-catch image if provided
if eye_catch_img_id is not None:
post.thumbnail = eye_catch_img_id
# Update post date if provided
if date is not None:
post.date = date
# Update the post
return wp.call(EditPost(post_id, post))
投稿の全体処理
for i in range(len(df)):
# メタデータの取得
title, path, title_png, color_words, tags,categories, link_name, img_folder, post_id, meta_data, update_post = get_params(df.iloc[i,:])
output_image_path = os.path.join(img_folder,f"{link_name}_thum.png")
if post_id == '' or post_id is None or post_id is np.nan or update_post:
#テキストとメタデータの分離
is_post,_,_, _, _, _, _, text = parse_metadata(path)
if is_post == False:
continue
#sサムネイルの作成とアップロード
if os.path.exists(output_image_path):
if overwrite is None:
c = input(output_image_path+'は既に存在します。上書きしますか？(Y/N)')
if c == 'y' or c=='Y':
add_text_to_image(base_image_path, title_png, output_image_path, color_words)
else:
if overwrite:
add_text_to_image(base_image_path, title_png, output_image_path, color_words)
upload_type = 2
else:
add_text_to_image(base_image_path, title_png, output_image_path, color_words)
media_id = wp_upload_image(wp,output_image_path,upload_type=upload_type)
html = convert_to_html_and_display(text)
html, media_ids = copy_text_wp(wp, html, img_folder)
if update_post and post_id!='':
post_status = update_wp_post(wp, post_id, title=title, body=html, categories=categories,tags=tags, eye_catch_img_id=media_id,meta_data=meta_data,status=status, link_name=link_name,date=date)
print(f'{link_name} {post_id}を更新しました')
assert post_status==True
df.at[i,'update_post'] = np.nan
del_update(path)
df.to_csv(post_data_path,index=False)
else:
post_id = post_to_wp(wp, title, html, categories=categories,tags=tags, eye_catch_img_id=media_id,meta_data=meta_data,status=status, link_name=link_name,date=date)
df.at[i,'post_id'] = int(post_id)
print(f'{link_name} {post_id}を投稿しました')
df.to_csv(post_data_path,index=False)
time.sleep(3)
かなりコードが長くなってしまいましたが、これらを実行することでPythonから自動でブログ記事を投稿するシステムができます。
コメント