WEBサイトの情報を収集し、OpenAIで要約してWPに自動投稿するプログラム【Python】

　コンピュータ　 1517

はじめに　

　最近ChatGPTが話題になっているので、いくつかのテストプログラムをChatGPTを利用して作成してきたが、今回はタイトルにあるように少し複雑なプログラムを作成することに取り組んだ。ウェブスクレイピングする場合に特に重要となる正規表現を使って必要な部分を切り抜くプログラムを書くのは普通は手間がかかるのだが、ChatGPTを使うことでスムーズに作成することができた。
　今回作成したプログラムでは、WEBサイトから記事を収集し、OpenAIのAPIを使って要約文章を作成し、WordPressに自動投稿することができる。

WEBサイトからの記事の収集方法について

　まず記事の一覧を取得し、それから個別の記事の内容を取得している。RSSによって記事一覧を取得する場合は問題ないが、それ以外の場合には、Webサイトごとに異なる方法で情報を取得する必要がある。そのため、個別に分岐して処理を行っている。記事の内容は、HTMLの中の特定のClassかIDの名前の部分を取得する。

# main関数の一部抜粋
    for city_name, rss_url, element_id, is_rss in website_urls:
        if is_rss:
            feed = feedparser.parse(rss_url)
            for entry in feed.entries:
                date = entry.published if 'published' in entry else entry.updated if 'updated' in entry else entry.created
                if isinstance(date, str):
                    #date = datetime.strptime(date, date_format)
                    date = parser.parse(date, tzinfos={'JST': gettz('Asia/Tokyo')})
                    entry_title = entry.title
                    url = entry.link
                    data = extract_article(url, element_id, city_name, entry_title, date)
                    if not data is None:
                        json_data.append(data)
        else:
            data = extract_data_from_website(rss_url, element_id,city_name)
            if data is None:
                continue
            json_data.append(data)
# RSS非対応のWebサイトのパース
def extract_data_from_website(rss_url, element_id, city_name):
    """
    RSS非対応のWebサイトのパース関数
    :param rss_url: 取得したい記事のURL
    :param element_id: 取得したい要素のid
    :param city_name: 記事が属する地域
    :return: 取得した記事情報の辞書
    """
    response = requests.get(rss_url)
    detected_encoding = chardet.detect(response.content)['encoding']
    response.encoding = detected_encoding
    soup = BeautifulSoup(response.text, 'html.parser')
    data ={}

    if city_name == "XX地域":
        # 個別の処理
        data = extract_article(url, element_id, city_name, entry_title, date)
        return data

OpenAIを使って要約文章を作成する方法について

　OpenAIのAPIを使用するには、pythonライブラリの"openai"をインポートする。
　APIのパラメータはデフォルトの設定を使用しているが、トークン数の制限に引っかからないように調整するために、MeCabの形態素解析を使用して文章を形態素に分解し、計算している。制限を超える文章の部分は、文末から削除している。

# OpenAPIを使って要約を作成する関数
def create_summary(prompt):
    """
    文章の要約を取得する関数
    :param prompt: 要約したい文章
    :return: 要約文章
    """
    # 文章の長さを取得
    debug_print(prompt)
    if prompt is None:return
    threshold = 50
    prompt_tokens = [token.split("\t")[0] for token in m.parse(prompt).split("\n")]
    prompt_length = int(len(prompt_tokens) * 1.035)
    debug_print(prompt_length)
    if prompt_length <= threshold: return
    if(max_tokens - prompt_length < 0):
        new_prompt_tokens = prompt_tokens[:-(prompt_length - max_tokens)]
        prompt = "".join(new_prompt_tokens)

    # OpenAIのAPIを実行
    completions = openai.Completion.create(
        engine=model_engine,
        prompt=prompt,
        temperature=0.7,
        max_tokens=max_tokens,
        stop=None,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
    )

    # APIの結果を返す
    message = completions.choices[0].text
    if not completions.choices[0].text:return

    # 消費したトークン数を取得
    debug_print("tokens_consumed:" + str(completions.usage.get("total_tokens")))

    return message

WPに自動投稿する方法について

　pythonライブラリのwordpress_xmlrpcをインポートする。これにより、WordPressに投稿するための機能を使用することができる。
　xmlrpcとは、WordPressが提供する、外部からの投稿や編集を行うためのAPIである。
　記事のタイトルや要約はjson_dataに格納しているので、HTML用に整形する。
　今回は、オプションでドラフト状態にしているが、直接投稿することも可能である。

# 作成した記事のデータをWordPressに投稿する関数
def create_wp_post(json_data, wp_url, wp_id, wp_password):
    """
    作成した記事のデータをWordPressに投稿する関数
    
    Parameters:
    - json_data (list) : 記事データのjson
    - wp_url (str) : WordPressのURL
    - wp_id (str) : WordPressのアカウントのID
    - wp_password (str) : WordPressのアカウントのパスワード
    """

    #データがNoneだった場合は処理をやめる
    if json_data is None:
        debug_print("json_data is None. Exiting...")
        return
    else:
        #wp用コンテンツ整形処理
        content =""
        previous_city_name = ""
        post_tags = set()

        table_of_contents = "<h2>目次</h2>"
        title_count = 0

        for data in json_data:
            if data is None:
                continue
            city_name = data["city_name"]
            title = data.get("entry_title")
            summary = data.get("text_abs")
            url = data["url"]
            if title is None or summary is None:
                continue
            title_count += 1

            post_tags.add(city_name)
            if city_name != previous_city_name:
                if previous_city_name:
                    table_of_contents += "</ul>"
                content = content + "<h3 id='" + city_name + "'>" + city_name + "</h3>"
                table_of_contents += "<p><a href='#" + city_name + "'>" + city_name + "</a>"
                table_of_contents += "<ul>"

                previous_city_name = city_name

            # 記事部分
            content = content + "<h4><a href='" + url + "'>" + title + "</a></h4>"
            content = content + "<p><pre>　</pre>" + summary + "</p>"

            # 目次部分
            table_of_contents += "<li>" + title + "</li>"

        title_count_text = str(title_count) + "件" if title_count > 1 else "1件"
        title_summary = " ( " + title_count_text + " ) "

        table_of_contents = "<div class='wp-block-quote'>" + table_of_contents + "</div>"
        content = table_of_contents + content

        if title_count ==0:return
        debug_print(content)

        now = datetime.now()
        #which="publish"
        which="draft"

        #クライアント呼び出し
        wp_client = Client(wp_url, wp_id,wp_password)

        #投稿処理
        wp_post = WordPressPost()
        wp_post.post_status = which
        wp_post.title = "{}年{}月{}日({})のXX地域ニュース{}".format(now.year, now.month, now.day, now.strftime("%A"), title_summary )
        wp_post.content = content
        wp_post.terms_names = {
        "post_tag": list(post_tags),
        "category": ['XXXXXX'],
        }
        wp_client.call(NewPost(wp_post)) 
        return

ソースコードとデモ動画

　作成したプログラムの全体のソースコードは以下のとおり。

# coding: utf-8
debug_mode = True
 
import feedparser # RSSフィードのパーサー
import json # JSONフォーマットのデータ操作
import pytz # タイムゾーンのライブラリ
import requests # HTTPリクエストを送信するためのライブラリ
import openai # OpenAIのAPIを使用するためのライブラリ
import deepl # Deepl翻訳APIを使用するためのライブラリ
import MeCab # 形態素解析を行うためのライブラリ
import re # 正規表現を使用するためのライブラリ
import ssl # SSL接続に関するライブラリ
import locale # 地域や言語によって異なる表記などのルールを扱うためのライブラリ
import time # 時間に関するライブラリ
import chardet # 文字エンコーディングを検出するためのライブラリ

from bs4 import BeautifulSoup # HTMLやXMLのパーサー
from dateutil import parser # 日時のパーサー
from dateutil.tz import gettz # 要求されたタイムゾーンのtzinfoオブジェクトを返すライブラリ
from datetime import datetime, timedelta # 日時に関するクラス
from wordpress_xmlrpc import Client, WordPressPost # WordPressにアクセスするためのライブラリ
from wordpress_xmlrpc.methods.users import GetUserInfo # WordPressのユーザー情報を取得するためのメソッド
from wordpress_xmlrpc.methods.posts import GetPosts, NewPost # WordPressの投稿情報を取得するためのメソッド

# 日本語のロケールを設定
locale.setlocale(locale.LC_TIME, 'ja_JP.UTF-8')

# SSLエラーを回避するための設定
ssl._create_default_https_context = ssl._create_unverified_context

# Tokyoに対応するタイムゾーンを生成
jst = pytz.timezone('Asia/Tokyo')

# MeCabのTaggerインスタンスを生成
m = MeCab.Tagger()

# OpenAPI設定
# APIキーを取得
openai.api_key = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
#モデルを指定
model_engine = "text-davinci-003"
#最大トークン数を指定
max_tokens = 1024

# WordPress XMLRPCでの投稿設定
wp_id="XXXXXXXXXXXX"
wp_password="XXXXXXXXXXXX"
wp_url="https://XXXXXXXX/xmlrpc.php"

# 情報ソース
# 取得対象の日数
days = 3

# ウェブサイトのURLリスト
"""
website_urls:
配列(リスト)変数で、WebサイトのURLを格納している。
各要素はタプルとして、 ('地域名', 'WebサイトのURL', '記事の要素のID', 'RSSかどうかのフラグ') の4つの要素を持つ。
この配列は、 forループによって取り出され、RSSフィードの解析、記事の抽出の処理に利用する。
"""
website_urls = [
    ('XX地域','https://xxxxxx/xxx.xml','main_body', True),
]

# OpenAPIを使って要約を作成する関数
def create_summary(prompt):
    """
    文章の要約を取得する関数
    :param prompt: 要約したい文章
    :return: 要約文章
    """
    # 文章の長さを取得
    debug_print(prompt)
    if prompt is None:return
    threshold = 50
    prompt_tokens = [token.split("\t")[0] for token in m.parse(prompt).split("\n")]
    prompt_length = int(len(prompt_tokens) * 1.035)
    debug_print(prompt_length)
    if prompt_length <= threshold: return
    if(max_tokens - prompt_length < 0):
        new_prompt_tokens = prompt_tokens[:-(prompt_length - max_tokens)]
        prompt = "".join(new_prompt_tokens)

    # OpenAIのAPIを実行
    completions = openai.Completion.create(
        engine=model_engine,
        prompt=prompt,
        temperature=0.7,
        max_tokens=max_tokens,
        stop=None,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
    )

    # APIの結果を返す
    message = completions.choices[0].text
    if not completions.choices[0].text:return

    # 消費したトークン数を取得
    debug_print("tokens_consumed:" + str(completions.usage.get("total_tokens")))

    return message

# 特定の要素を取得する関数
def get_text_from_city_hp(url, element_id):
    """
    特定の要素のテキストを取得する関数
    :param url: 取得したい要素が存在するURL
    :param element_id: 取得したい要素のid
    :return: 取得した要素のテキスト
    """
    # HTTPリクエストを送信しHTMLデータを取得
    response = requests.get(url)
    detected_encoding = chardet.detect(response.content)['encoding']
    response.encoding = detected_encoding
    
    # HTMLデータをパース
    soup = BeautifulSoup(response.text, 'html.parser')
    # idを持つ要素を取得
    element = soup.find(id=element_id) or soup.find(class_=element_id)
    if element:
        return element
    else:
        return

# 個別の記事を取得する関数
def extract_article(url, element_id, city_name, entry_title, date):
    """
    個別の記事を取得する関数
    :param url: 取得したい記事のURL
    :param element_id: 取得したい要素のid
    :param city_name: 記事が属する地域
    :param entry_title: 記事の見出し
    :param date: 記事の日時
    :return: 取得した記事情報の辞書
    """
    # 指定日数以内の記事のみ取得
    if not date.replace(tzinfo=pytz.timezone('Asia/Tokyo')) > datetime.now(jst) - timedelta(days=days):return
    data = {}
    # 記事のテキストを取得
    text = get_text_from_city_hp(url, element_id)
    # 記事の要約を取得
    text_abs = create_summary(text)
    debug_print(text_abs)

    # 取得した記事情報を格納
    date_str = date.strftime('%Y-%m-%d %H:%M:%S')
    data = {'city_name': city_name, 'date': date_str, 'entry_title': entry_title, 'url': url, 'text':text, 'text_abs':text_abs}
    return data


# RSS非対応のWebサイトのパース
def extract_data_from_website(rss_url, element_id, city_name):
    """
    RSS非対応のWebサイトのパース関数
    :param rss_url: 取得したい記事のURL
    :param element_id: 取得したい要素のid
    :param city_name: 記事が属する地域
    :return: 取得した記事情報の辞書
    """
    response = requests.get(rss_url)
    detected_encoding = chardet.detect(response.content)['encoding']
    response.encoding = detected_encoding
    soup = BeautifulSoup(response.text, 'html.parser')
    data ={}

    if city_name == "XX地域":
        # 個別の処理
        data = extract_article(url, element_id, city_name, entry_title, date)
        return data

# 作成した記事のデータをWordPressに投稿する関数
def create_wp_post(json_data, wp_url, wp_id, wp_password):
    """
    作成した記事のデータをWordPressに投稿する関数
    
    Parameters:
    - json_data (list) : 記事データのjson
    - wp_url (str) : WordPressのURL
    - wp_id (str) : WordPressのアカウントのID
    - wp_password (str) : WordPressのアカウントのパスワード
    """

    #データがNoneだった場合は処理をやめる
    if json_data is None:
        debug_print("json_data is None. Exiting...")
        return
    else:
        #wp用コンテンツ整形処理
        content =""
        previous_city_name = ""
        post_tags = set()

        table_of_contents = "<h2>目次</h2>"
        title_count = 0

        for data in json_data:
            if data is None:
                continue
            city_name = data["city_name"]
            title = data.get("entry_title")
            summary = data.get("text_abs")
            url = data["url"]
            if title is None or summary is None:
                continue
            title_count += 1

            post_tags.add(city_name)
            if city_name != previous_city_name:
                if previous_city_name:
                    table_of_contents += "</ul>"
                content = content + "<h3 id='" + city_name + "'>" + city_name + "</h3>"
                table_of_contents += "<p><a href='#" + city_name + "'>" + city_name + "</a>"
                table_of_contents += "<ul>"

                previous_city_name = city_name

            # 記事部分
            content = content + "<h4><a href='" + url + "'>" + title + "</a></h4>"
            content = content + "<p><pre>　</pre>" + summary + "</p>"

            # 目次部分
            table_of_contents += "<li>" + title + "</li>"

        title_count_text = str(title_count) + "件" if title_count > 1 else "1件"
        title_summary = " ( " + title_count_text + " ) "

        table_of_contents = "<div class='wp-block-quote'>" + table_of_contents + "</div>"
        content = table_of_contents + content

        if title_count ==0:return
        debug_print(content)

        now = datetime.now()
        #which="publish"
        which="draft"

        #クライアント呼び出し
        wp_client = Client(wp_url, wp_id,wp_password)

        #投稿処理
        wp_post = WordPressPost()
        wp_post.post_status = which
        wp_post.title = "{}年{}月{}日({})のXX地域ニュース{}".format(now.year, now.month, now.day, now.strftime("%A"), title_summary )
        wp_post.content = content
        wp_post.terms_names = {
        "post_tag": list(post_tags),
        "category": ['XXXXXX'],
        }
        wp_client.call(NewPost(wp_post)) 
        return 

# debug用のprint関数
def debug_print(message):
    if debug_mode:
        print(message)

# main関数
def main():
    """
    メイン関数。各地域のRSSフィード等から記事を抽出し、WordPressに投稿する。
    """
    json_data = list()
    for city_name, rss_url, element_id, is_rss in website_urls:
        if is_rss:
            feed = feedparser.parse(rss_url)
            for entry in feed.entries:
                date = entry.published if 'published' in entry else entry.updated if 'updated' in entry else entry.created
                if isinstance(date, str):
                    #date = datetime.strptime(date, date_format)
                    date = parser.parse(date, tzinfos={'JST': gettz('Asia/Tokyo')})
                    entry_title = entry.title
                    url = entry.link
                    data = extract_article(url, element_id, city_name, entry_title, date)
                    if not data is None:
                        json_data.append(data)
        else:
            data = extract_data_from_website(rss_url, element_id,city_name)
            if data is None:
                continue
            json_data.append(data)

    create_wp_post(json_data, wp_url, wp_id, wp_password)

# main関数の呼び出し
if __name__=="__main__":
    # プログラムのエントリーポイント
    main()

　動作に必要なプラグインは以下を実行することで一度にインストールすることができる。これで、feedparser、pytz、requests、openai、deepl、unidic-lite、dateutil、wordpress_xmlrpcのインストールが一度に行える。

pip install feedparser pytz requests openai deepl unidic-lite python-dateutil python-wordpress-xmlrpc

　以下は、実際に作成したプログラムを使って収集した記事を要約し、テスト中のブログに投稿した際のデモ動画。

おわりに

　自動的に要約記事を書くことができるなんて、非常に便利な時代になったものだと思う。
　ただし、今回のプログラムでは個別の記事ごとにOpenAIのAPIを利用することから、全体の書き振りの統一感を出すことは難しいという問題が残っている。また、ChatGPTに比べると要約精度が明らかに低いので、収集する分野に特化した学習を行って、要約の精度が向上させる必要がある。また、取得する記事ごとのコストを削減するためには、不要な部分を事前に削除する機能も必要だろう。
　なお、取得する記事は著作権の問題がないものに限るので、利用の際は注意して欲しい。

スッキリわかるPython入門 スッキリわかるシリーズ Kindle版




関連記事
" data-srcset="'https://ubun2m.com/wp-content/uploads/2023/01/uranai_suisyou-1.png 150w, https://ubun2m.com/wp-content/uploads/2023/01/uranai_suisyou-1.png 720w" sizes="(max-width: 360px) 100vw, 360px"/>
コンピュータ

ChatGPTが作成したプログラマーのための星座占い

2023年01月13日
" data-srcset="'https://ubun2m.com/wp-content/uploads/2022/06/Screenshot-2022-06-25-at-19.23.26.png 150w, https://ubun2m.com/wp-content/uploads/2022/06/Screenshot-2022-06-25-at-19.23.26.png 720w" sizes="(max-width: 360px) 100vw, 360px"/>
コンピュータ

【Javascript】iPadスプリットビュー用のアイコンを作る

2022年03月14日
" data-srcset="'https://ubun2m.com/wp-content/uploads/2022/02/Untitled_Artwork-4-400x400.png 150w, https://ubun2m.com/wp-content/uploads/2022/02/Untitled_Artwork-4-400x400.png 720w" sizes="(max-width: 360px) 100vw, 360px"/>
コンピュータ

JavaScriptツール置き場

2022年02月28日
コメント
  • 138 まーちゃん 2023年11月25日 2:48 PM

    こんにちは。はじめまして、まーちゃんと申します。
    M様の「WEBサイトの情報を収集し、OpenAIで要約してWPに自動投稿するプログラム【Python】」を拝見しまして、私が探し求めたソフトウェアに出会えたと思いました。
    お尋ねですが、main関数が一部抜粋となっておりますが実際に試してみることは可能でしょうか。何卒よろしくお願いいたします。

    • 139 legato 2023年11月29日 8:01 PM

      まーちゃんさん

      コメントいただきありがとうございます。
      「ソースコードとデモ動画」に記載してあるコードを使っていただければ、とりあえずは動くと思います。APIキーの設定や、スクレイピング対象のWebページの設定などはご自身の環境に合わせて修正してください。
      この記事を書いた時点からOpenAIのAPIもかなり進化しているので、実用的な記事が書けるかもしれませんね。

      legato

ー