LLMを活用したWordPressに記事を自動投稿するPythonスクリプト

Python

はじめに

ブログ運営の効率化をお考えですか？本記事では、最新のAI技術であるLarge Language Model（LLM）を活用し、Pythonを使ってWordPressに記事を自動投稿する革新的な方法をご紹介します。このスクリプトは、記事の下書き作成からカテゴリやタグの設定、さらにはサムネイル画像のアップロードまで、すべてを自動化します。LLMの力を借りることで、より intelligent で効率的な記事管理が可能になります。

環境設定

まずは、必要な環境設定を行います。以下のライブラリをインストールしてください。

pip install requests python-dotenv matplotlib japanize-matplotlib numpy tenacity litellm loguru

必要なライブラリのインポート

次に、必要なライブラリをインポートします。

import requests
import json
import os
import sys
from dotenv import load_dotenv
import matplotlib.pyplot as plt
import japanize_matplotlib
from datetime import datetime
import numpy as np
import re
from tenacity import retry, stop_after_attempt, wait_fixed
from litellm import completion
import logging
from loguru import logger

# 記事の下書きが保存されているフォルダのパスを設定
folder_path = r"articles_draft\Sample01"

ここで注目すべきは litellm ライブラリです。これはLLMとの通信を簡単に行うためのライブラリで、本スクリプトではGoogle GeminiというLLMを利用しています。

設定ファイルの読み込み

環境変数から設定を読み込みます。

# .envファイルから環境変数を読み込む
load_dotenv()

# 環境変数から設定を読み込む
AUTH_USER = os.getenv('AUTH_USER')
AUTH_PASS = os.getenv('AUTH_PASS')
BASE_URL = os.getenv('BASE_URL')
GEMINI_API_KEY = os.getenv('GEMINI_API_KEY')

# 投稿するURLの設定
END_POINT_URL = f"{BASE_URL}/wp-json/wp/v2/posts/"

ここで GEMINI_API_KEY を設定していることに注目してください。これは、LLM（Google Gemini）とやり取りするために必要なAPIキーです。

既存のカテゴリとタグの取得

WordPressの既存のカテゴリとタグを取得する関数を定義します。

def get_existing_categories_and_tags(wp_api_base, auth):
    logger.info("既存のカテゴリとタグを取得中...")
    categories_url = f"{wp_api_base}/categories"
    tags_url = f"{wp_api_base}/tags"

    categories = requests.get(categories_url, auth=auth, params={'per_page': 100}).json()
    category_map = {cat['name']: cat['id'] for cat in categories}

    all_tags = []
    page = 1
    per_page = 100
    while True:
        response = requests.get(tags_url, auth=auth, params={'per_page': per_page, 'page': page})
        if response.status_code != 200:
            logger.error(f"タグの取得中にエラーが発生しました。ステータスコード: {response.status_code}")
            break
        tags = response.json()
        if not tags:
            break
        all_tags.extend(tags)
        if len(tags) < per_page:
            break
        page += 1
        if len(all_tags) >= 300:
            break

    tag_map = {tag['name']: tag['id'] for tag in all_tags}

    logger.info(f"{len(category_map)}個のカテゴリと{len(tag_map)}個のタグを取得しました。")
    return (category_map, tag_map)

LLMを活用したカテゴリとタグの提案

ここからがLLMの真骨頂です。Google GeminiのAI APIを使用して、記事の内容に基づいてカテゴリとタグを提案する関数を定義します。

@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def suggest_categories_and_tags(content, existing_categories, existing_tags, gemini_api_key):
    logger.info("Gemini LLMを使用してカテゴリとタグを提案中...")
    os.environ['GEMINI_API_KEY'] = gemini_api_key
    prompt = f"""
    既存のカテゴリ: {', '.join(existing_categories)}
    既存のタグ: {', '.join(existing_tags)}

    以下の記事内容に対して、最適でシンプルで簡潔なカテゴリとタグを提案してください。
    既存のカテゴリとタグを参考にしつつ、必要に応じて新しいものも提案してください。
    カテゴリとタグの両方に名前とslugを含めてください。
    タグにはカテゴリを含めないで。

    記事内容:
    {content[:1000]}  # 長すぎる場合は最初の1000文字のみ使用

    回答は以下のJSONフォーマットで提供してください：
    {{
        "categories": [
            {{"name": "カテゴリ名1", "slug": "カテゴリslug1"}},
            {{"name": "カテゴリ名2", "slug": "カテゴリslug2"}},
            ...
        ],
        "tags": [
            {{"name": "タグ名1", "slug": "タグslug1"}},
            {{"name": "タグ名2", "slug": "タグslug2"}},
            ...
        ]
    }}
    """

    response = completion(
        model="gemini/gemini-1.5-pro-latest", 
        messages=[{"role": "user", "content": prompt}]
    )
    logger.info("LLMによるカテゴリとタグの提案が完了しました。")

    try:
        content = response.choices[0].message.content
        json_match = re.search(r'```json\s*(.*?)\s*```', content, re.DOTALL)
        if json_match:
            json_content = json_match.group(1)
        else:
            json_content = content

        suggestions = json.loads(json_content)
        return suggestions
    except json.JSONDecodeError as e:
        logger.error(f"JSONデコードエラー: {e}")
        logger.error(f"受け取った応答: {content}")
        raise
    except Exception as e:
        logger.error(f"予期せぬエラーが発生しました: {e}")
        logger.error(f"受け取った応答: {content}")
        raise

この関数では、LLM（Google Gemini）に記事の内容、既存のカテゴリ、タグを送信し、適切なカテゴリとタグの提案を受け取っています。LLMの高度な言語理解能力により、記事の内容に最適なカテゴリとタグを提案することができます。

LLMによる英語のスラッグ生成

次に、LLMを使って記事のタイトルから英語のスラッグを生成する関数を定義します。

@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def generate_english_slug(title, gemini_api_key):
    logger.info("Gemini LLMを使用して英語のslugを生成中...")
    os.environ['GEMINI_API_KEY'] = gemini_api_key
    prompt = f"""
    以下の日本語のタイトルを英語に翻訳し、WordPressのslugとして適切な形式に変換してください。
    slugは短く、簡潔で、URLに適した形式にしてください。

    日本語タイトル: {title}

    回答は以下のJSONフォーマットで提供してください：
    {{
        "slug": "英語のslug"
    }}
    """

    response = completion(
        model="gemini/gemini-1.5-pro-latest", 
        messages=[{"role": "user", "content": prompt}]
    )
    logger.info("LLMによる英語のslugの生成が完了しました。")

    try:
        content = response.choices[0].message.content
        json_match = re.search(r'```json\s*(.*?)\s*```', content, re.DOTALL)
        if json_match:
            json_content = json_match.group(1)
        else:
            json_content = content

        result = json.loads(json_content)
        return result['slug']
    except Exception as e:
        logger.error(f"英語のslug生成中にエラーが発生しました: {e}")
        logger.error(f"受け取った応答: {content}")
        raise

この関数では、LLMに日本語のタイトルを送信し、適切な英語のスラッグを生成しています。LLMの翻訳能力と言語理解能力により、SEOに最適化された英語のスラッグを生成することができます。

JSONファイルの保存

生成されたメタデータをJSONファイルとして保存する関数を定義します。

def save_json_to_file(data, filename, folder_path):
    file_path = os.path.join(folder_path, filename)
    with open(file_path, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    logger.info(f"JSONファイルを保存しました: {file_path}")

メイン処理

最後に、スクリプトのメイン処理部分を見ていきましょう。

# README.mdファイルを読み込む
readme_path = os.path.join(folder_path, 'README.md')
if not os.path.exists(readme_path):
    print(f"エラー: {readme_path} が見つかりません。")
    sys.exit(1)

with open(readme_path, 'r', encoding='utf-8') as file:
    content = file.read()

# タイトルを抽出（最初の#で始まる行）
title = next((line.strip('# ') for line in content.split('\n') if line.startswith('#')), "デフォルトタイトル")

# 既存のカテゴリとタグを取得
category_map, tag_map = get_existing_categories_and_tags(f"{BASE_URL}/wp-json/wp/v2", (AUTH_USER, AUTH_PASS))

# LLMを使用してカテゴリとタグを提案
suggestions = suggest_categories_and_tags(content, list(category_map.keys()), list(tag_map.keys()), GEMINI_API_KEY)

# LLMを使用してタイトルから英語のslugを生成
p_slug = generate_english_slug(title, GEMINI_API_KEY)

# カテゴリとタグの名前をIDに変換
category_ids = [category_map.get(cat['name']) for cat in suggestions['categories'] if cat['name'] in category_map]
tag_ids = [tag_map.get(tag['name']) for tag in suggestions['tags'] if tag['name'] in tag_map]

# 新しいカテゴリとタグを作成
for cat in suggestions['categories']:
    if cat['name'] not in category_map:
        response = requests.post(f"{BASE_URL}/wp-json/wp/v2/categories", json={'name': cat['name']}, auth=(AUTH_USER, AUTH_PASS))
        if response.status_code == 201:
            new_cat = response.json()
            category_ids.append(new_cat['id'])
            category_map[cat['name']] = new_cat['id']

for tag in suggestions['tags']:
    if tag['name'] not in tag_map:
        response = requests.post(f"{BASE_URL}/wp-json/wp/v2/tags", json={'name': tag['name']}, auth=(AUTH_USER, AUTH_PASS))
        if response.status_code == 201:
            new_tag = response.json()
            tag_ids.append(new_tag['id'])
            tag_map[tag['name']] = new_tag['id']

# メタデータをJSONファイルに保存
metadata = {
    "slug": p_slug,
    "categories": [{"name": cat['name'], "id": category_map.get(cat['name'])} for cat in suggestions['categories']],
    "tags": [{"name": tag['name'], "id": tag_map.get(tag['name'])} for tag in suggestions['tags']]
}
save_json_to_file(metadata, 'metadata.json', folder_path)

このメイン処理部分では、LLMを活用して以下の作業を自動化しています：

  1. 記事の内容に基づいたカテゴリとタグの提案
  2. 日本語タイトルからSEO最適化された英語のスラッグの生成

これらの処理により、人間の介入を最小限に抑えつつ、高品質な記事メタデータを生成することが可能になります。

まとめ

本記事では、LLM（Large Language Model）を活用してWordPressに記事を自動投稿するPythonスクリプトについて解説しました。このスクリプトの革新的な点は、AIの力を借りて以下のタスクを自動化している点です：

  1. 記事内容に基づいた最適なカテゴリとタグの提案
  2. 日本語タイトルからSEO最適化された英語スラッグの生成

これらの機能により、ブログ運営者は記事作成に集中し、メタデータの最適化をAIに任せることができます。

全体コード

import requests
import json
import os
import sys
from dotenv import load_dotenv
import matplotlib.pyplot as plt
import japanize_matplotlib
from datetime import datetime
import numpy as np
import re
from tenacity import retry, stop_after_attempt, wait_fixed
from litellm import completion
import logging
from loguru import logger

folder_path = r"articles_draft\Sample01"

# .envファイルから環境変数を読み込む
load_dotenv()

# 環境変数から設定を読み込む
AUTH_USER = os.getenv('AUTH_USER')
AUTH_PASS = os.getenv('AUTH_PASS')
BASE_URL = os.getenv('BASE_URL')
GEMINI_API_KEY = os.getenv('GEMINI_API_KEY')

# 投稿するURLの設定
END_POINT_URL = f"{BASE_URL}/wp-json/wp/v2/posts/"

def get_existing_categories_and_tags(wp_api_base, auth):
    logger.info("既存のカテゴリとタグを取得中...")
    categories_url = f"{wp_api_base}/categories"
    tags_url = f"{wp_api_base}/tags"

    categories = requests.get(categories_url, auth=auth, params={'per_page': 100}).json()
    category_map = {cat['name']: cat['id'] for cat in categories}

    all_tags = []
    page = 1
    per_page = 100
    while True:
        response = requests.get(tags_url, auth=auth, params={'per_page': per_page, 'page': page})
        if response.status_code != 200:
            logger.error(f"タグの取得中にエラーが発生しました。ステータスコード: {response.status_code}")
            break
        tags = response.json()
        if not tags:
            break
        all_tags.extend(tags)
        if len(tags) < per_page:
            break
        page += 1
        if len(all_tags) >= 300:
            break

    tag_map = {tag['name']: tag['id'] for tag in all_tags}

    logger.info(f"{len(category_map)}個のカテゴリと{len(tag_map)}個のタグを取得しました。")
    return (category_map, tag_map)

@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def suggest_categories_and_tags(content, existing_categories, existing_tags, gemini_api_key):
    logger.info("Geminiを使用してカテゴリとタグを提案中...")
    os.environ['GEMINI_API_KEY'] = gemini_api_key
    prompt = f"""
    既存のカテゴリ: {', '.join(existing_categories)}
    既存のタグ: {', '.join(existing_tags)}

    以下の記事内容に対して、最適でシンプルで簡潔なカテゴリとタグを提案してください。
    既存のカテゴリとタグを参考にしつつ、必要に応じて新しいものも提案してください。
    カテゴリとタグの両方に名前とslugを含めてください。
    タグにはカテゴリを含めないで。

    記事内容:
    {content[:1000]}  # 長すぎる場合は最初の1000文字のみ使用

    回答は以下のJSONフォーマットで提供してください：
    {{
        "categories": [
            {{"name": "カテゴリ名1", "slug": "カテゴリslug1"}},
            {{"name": "カテゴリ名2", "slug": "カテゴリslug2"}},
            ...
        ],
        "tags": [
            {{"name": "タグ名1", "slug": "タグslug1"}},
            {{"name": "タグ名2", "slug": "タグslug2"}},
            ...
        ]
    }}
    """

    response = completion(
        model="gemini/gemini-1.5-pro-latest", 
        messages=[{"role": "user", "content": prompt}]
    )
    logger.info("カテゴリとタグの提案が完了しました。")

    try:
        content = response.choices[0].message.content
        json_match = re.search(r'```json\s*(.*?)\s*```', content, re.DOTALL)
        if json_match:
            json_content = json_match.group(1)
        else:
            json_content = content

        suggestions = json.loads(json_content)
        return suggestions
    except json.JSONDecodeError as e:
        logger.error(f"JSONデコードエラー: {e}")
        logger.error(f"受け取った応答: {content}")
        raise
    except Exception as e:
        logger.error(f"予期せぬエラーが発生しました: {e}")
        logger.error(f"受け取った応答: {content}")
        raise

@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def generate_english_slug(title, gemini_api_key):
    logger.info("Geminiを使用して英語のslugを生成中...")
    os.environ['GEMINI_API_KEY'] = gemini_api_key
    prompt = f"""
    以下の日本語のタイトルを英語に翻訳し、WordPressのslugとして適切な形式に変換してください。
    slugは短く、簡潔で、URLに適した形式にしてください。

    日本語タイトル: {title}

    回答は以下のJSONフォーマットで提供してください：
    {{
        "slug": "英語のslug"
    }}
    """

    response = completion(
        model="gemini/gemini-1.5-pro-latest", 
        messages=[{"role": "user", "content": prompt}]
    )
    logger.info("英語のslugの生成が完了しました。")

    try:
        content = response.choices[0].message.content
        json_match = re.search(r'```json\s*(.*?)\s*```', content, re.DOTALL)
        if json_match:
            json_content = json_match.group(1)
        else:
            json_content = content

        result = json.loads(json_content)
        return result['slug']
    except Exception as e:
        logger.error(f"英語のslug生成中にエラーが発生しました: {e}")
        logger.error(f"受け取った応答: {content}")
        raise

def save_json_to_file(data, filename, folder_path):
    file_path = os.path.join(folder_path, filename)
    with open(file_path, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    logger.info(f"JSONファイルを保存しました: {file_path}")

# メイン処理
# README.mdファイルを読み込む
readme_path = os.path.join(folder_path, 'README.md')
if not os.path.exists(readme_path):
    print(f"エラー: {readme_path} が見つかりません。")
    sys.exit(1)

with open(readme_path, 'r', encoding='utf-8') as file:
    content = file.read()

# タイトルを抽出（最初の#で始まる行）
title = next((line.strip('# ') for line in content.split('\n') if line.startswith('#')), "デフォルトタイトル")

# 既存のカテゴリとタグを取得
category_map, tag_map = get_existing_categories_and_tags(f"{BASE_URL}/wp-json/wp/v2", (AUTH_USER, AUTH_PASS))

# カテゴリとタグを提案
suggestions = suggest_categories_and_tags(content, list(category_map.keys()), list(tag_map.keys()), GEMINI_API_KEY)

# タイトルから英語のslugを生成
p_slug = generate_english_slug(title, GEMINI_API_KEY)

# カテゴリとタグの名前をIDに変換
category_ids = [category_map.get(cat['name']) for cat in suggestions['categories'] if cat['name'] in category_map]
tag_ids = [tag_map.get(tag['name']) for tag in suggestions['tags'] if tag['name'] in tag_map]

# 新しいカテゴリとタグを作成
for cat in suggestions['categories']:
    if cat['name'] not in category_map:
        response = requests.post(f"{BASE_URL}/wp-json/wp/v2/categories", json={'name': cat['name']}, auth=(AUTH_USER, AUTH_PASS))
        if response.status_code == 201:
            new_cat = response.json()
            category_ids.append(new_cat['id'])
            category_map[cat['name']] = new_cat['id']

for tag in suggestions['tags']:
    if tag['name'] not in tag_map:
        response = requests.post(f"{BASE_URL}/wp-json/wp/v2/tags", json={'name': tag['name']}, auth=(AUTH_USER, AUTH_PASS))
        if response.status_code == 201:
            new_tag = response.json()
            tag_ids.append(new_tag['id'])
            tag_map[tag['name']] = new_tag['id']

# suggestionsとp_slugをJSONファイルに保存
metadata = {
    "slug": p_slug,
    "categories": [{"name": cat['name'], "id": category_map.get(cat['name'])} for cat in suggestions['categories']],
    "tags": [{"name": tag['name'], "id": tag_map.get(tag['name'])} for tag in suggestions['tags']]
}
save_json_to_file(metadata, 'metadata.json', folder_path)

# 投稿内容
p_title = title
p_content = content
p_status = "draft"

payload = {
    'title': p_title,
    'content': p_content,
    'status': p_status,
    'slug': p_slug,
    'categories': category_ids,
    'tags': tag_ids
}

headers = {'content-type': "Application/json"}

# 記事を投稿
r = requests.post(END_POINT_URL, data=json.dumps(payload), headers=headers, auth=(AUTH_USER, AUTH_PASS))
response_data = json.loads(r.text)

logger.info(f"記事を投稿しました: {response_data}")
# サムネイル画像をアップロード
media_endpoint = f"{BASE_URL}/wp-json/wp/v2/media"

# サムネイル画像のファイル名を確認
thumbnail_files = [f for f in os.listdir(folder_path) if f.lower().startswith('thumb.') and f.lower().endswith(('.png', '.jpg', '.jpeg'))]

if thumbnail_files:
    thumb_file = thumbnail_files[0]
    thumb_path = os.path.join(folder_path, thumb_file)
    file_extension = os.path.splitext(thumb_file)[1].lower()

    # Content-Typeを設定
    if file_extension == '.png':
        content_type = 'image/png'
    elif file_extension in ['.jpg', '.jpeg']:
        content_type = 'image/jpeg'

    headers = {
        "Content-Disposition": f'attachment; filename="{thumb_file}"',
        "Content-Type": content_type
    }

    with open(thumb_path, 'rb') as img:
        media_response = requests.post(
            media_endpoint,
            headers=headers,
            data=img,
            auth=(AUTH_USER, AUTH_PASS)
        )

    # アップロードした画像のIDを取得
    media_id = json.loads(media_response.text)['id']

    # サムネイルを設定
    update_payload = {
        'featured_media': media_id
    }

    update_response = requests.post(
        f"{END_POINT_URL}{response_data['id']}",
        json=update_payload,
        auth=(AUTH_USER, AUTH_PASS)
    )

    print(update_response.text)
else:
    print("サムネイル画像が見つかりません。")

# 結果をコンソールに出力
result = {
    "title": p_title,
    "slug": p_slug,
    "categories": [{"name": cat['name'], "id": category_map.get(cat['name'])} for cat in suggestions['categories']],
    "tags": [{"name": tag['name'], "id": tag_map.get(tag['name'])} for tag in suggestions['tags']]
}

print(json.dumps(result, ensure_ascii=False, indent=2))

コメント

タイトルとURLをコピーしました