全自動生成AIメディアを作ってみた

はじめに

自分で読む用に、AIが海外の生成AI関連のニュースやブログを自動で収集してきて翻訳してくれるスクリプトを書いたら、思いのほか使える感じだったので翻訳記事をブログとして公開することにしました。こちらです。

この記事では、やり方や記事を紹介します。部分的にでもマネすれば生成AIに限らず自分の興味のある分野のメディアが作れると思います。

全体的な流れ

全体的な流れとしては以下のようになります。

1. Google Alert
   記事を収集します。
2. Email parser by Zapier
   届いたメールをパースしてZapierに渡します。
3. AWS Lambda
   記事のタイトルや本文を和訳してWordpressにポストします。

Google Alert

Google Alertは以下のような感じで自分の興味があるワードを設定すると、関連する記事を収集してくれてメールで届けてくれる無料のサービスです。

届いたメールはこんな感じです。

あれ？メールの中に「日本語に翻訳」というボタンがありますね。そうなんです。届いた記事を日本語で読みたいだけなら、このボタンを押せば日本語に翻訳してくれるので、この後の工程はいらないです。お疲れ様でした！
でも、メールだと読まないんですね。これが。なので、僕はWordpressに投稿することにしました。
ちなみにこのキーワードはいくつでも設定できます。

Email Parser by Zapier

Google Alertはメールで届くので、届いたメールを加工するには受け口が必要です。それをしてくれるのが、Email Parser by Zapierです。

ここで自分のGメールアカウントへのアクセスなどを許可して以下のように設定します。

Initila Templateというところに、届いたメールが表示されるので、切り出したい部分を設定します。ここでは、Google Alertに設定したキーワードが入っている部分をKeyword、メール本文のヘッダフッタを除いた主要部分をbodyと言う風に設定しています。

Zapier

ここまでで、自分の設定したキーワードで記事を収集し、メールで届いた内容をパースして情報だけ抜き取る準備ができました。いろんなツールをうまいこと繋げて処理してくれるZapierで処理しましょう。

今回作ったZapierのフローは本当に単純です。

さっき設定したEmail parser by Zapierからもらった情報をAWSのLambdaに投げるだけです。Wordpressに投げるとかもLambda関数のPythonでやってしまっています。ZapierからWordpressへ投稿することもできるはずですし、その方がメンテとかやりやすいとは思いますが、Google Alertが投げてくるメールの中に複数の記事が含まれているので、こうしてあります。ひとつのメールにまとめられた複数の記事をばらして、複数のWordpressの記事にすることもできると思うのですが、Zapierでループを作ったりする方法がわからず、Pythonで処理してしまっています。Pythonは甘え。

ZapierのEmail Parser by Zapier設定

ZapierのEmail Parser by Zapier設定でEventはNew Email(新規メールが届いた)を指定します。

Accountは最初に作ったEmail Parser by Zapierのアカウントを指定してください。

ZapierのAWS Lambda設定

AWS Lambdaの方の設定では、EventはInvoke Functionを指定してください。

Accountは別途AWSで作成したLambda用のアカウントを設定してください。

Actionで、なんていうFunctionをinvokeするかを設定しますが、ここで引数としてEmail Parser by Zapierで設定したkeywordsとbodyを渡してください。これでAWSのLambdaにあるPythonスクリプトに検索キーワードと本文が渡ります。

Lambda Handler

最後にAWSのLambda Handlerですね。Pythonで書いております。使っているライブラリは以下のような感じです。諸事情で古いライブラリを指定していたりします。

requirements.txt

urllib3<2
BeautifulSoup4
google-cloud-translate==2.0.1
fastapi==0.99.0
openai==0.28
requests
tweepy

今回はLayersとか使っていないので、

pip install -r requirements.txt -t ./package

みたいにtargetディレクトリを作りzipでまとめてアップロードしています。
肝心のlambda_function.pyはこちら

lambda_function.py

import json
import re
import requests
from bs4 import BeautifulSoup
from google.cloud import translate_v2 as translate
import openai
import hashlib
import urllib.parse
import base64
import tweepy

# Set up the OpenAI API key
openai.api_key = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
# url2pngのAPIキー
url2png_apikey="xxxxxxxxxxxxx"
url2png_secret="xxxxxxxxxxxxxxxxxx"

# WordPress サイトの URL、ユーザー名、パスワードを設定
prompthub_url = 'https://prompthub.info'
prompthub_username = 'xxxxxxxxxxxxxxx'
prompthub_password = 'xxxxxxxxxxxxxxx'

# Twitter API認証情報
consumer_key = "xxxxxxxxxxxxxxxxxxxxxxxxx"
consumer_secret = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
access_token = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
access_token_secret = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

# Tweepy認証
x_client = tweepy.Client(
        consumer_key        = consumer_key,
        consumer_secret     = consumer_secret,
        access_token        = access_token,
        access_token_secret = access_token_secret
    )

# 認証情報を設定
credentials = base64.b64encode(f"{prompthub_username}:{prompthub_password}".encode('utf-8')).decode('utf-8')
headers = {
    'Authorization': f'Basic {credentials}',
    'Content-Type': 'application/json'
}

# URLの抽出
def extract_urls(text):
    pattern = r'<https://www\.google\.com/url\?rct=j&sa=t&url=(.+?)&'
    urls = re.findall(pattern, text)
    return list(set(urls))

# 日本語に翻訳する
def translate_to_japanese(text):
    translate_client = translate.Client()
    translation = translate_client.translate(text, target_language='ja')
    return translation['translatedText']

# スクリーンショットを撮る
def url2png(url, apikey, secret, fullpage=None, max_width=None,
            unique=None, viewport_width=1280, viewport_height=1024):
    data = {
      'url': url,
      'fullpage': 'true' if fullpage else 'false',
      'thumbnail_max_width': max_width,
      'unique': unique,
      'viewport': '{}x{}'.format(viewport_width, viewport_height),
    }
    filtered_data = dict((opt, data[opt]) for opt in data if data[opt])
    query_string = urllib.parse.urlencode(filtered_data)
    token = hashlib.md5('{}{}'.format(query_string, secret).encode('utf-8')).hexdigest()
    return "http://api.url2png.com/v6/{}/{}/png/?{}".format(apikey, token, query_string)

# ハッシュタグを生成
def generate_hashtags(text):
    prompt = f"Please generate 4 short hashtags from the following text. List the hashtags separated by spaces without adding any extra characters like numbers.:\n\n{text}\n\nHashtags:"
    
    response = openai.ChatCompletion.create(
        model="gpt-4-turbo-2024-04-09",
        messages=[
            {"role": "system", "content": "You are an expert on Twitter."},
            {"role": "user", "content": prompt}
        ],
        max_tokens=300,
        temperature=0.7
    )
    hashtags = response['choices'][0]['message']['content']  # チャットモデルの応答形式に合わせてアクセス方法を変更
    return hashtags


# 要約を生成
def generate_summary(text):
    prompt = f"Please read the following technical article and write a summary in Japanese using bullet points. Also, please write your thoughts in Japanese. Just answer with a summary and explanation, and do not add unnecessary wording. Please format the output in HTML. Please do not enclose the output with '```html' and '```'.:\n\n{text}"
    
    response = openai.ChatCompletion.create(
        model="gpt-4-turbo-2024-04-09",
        messages=[
            {"role": "system", "content": "You are an expert in generative AI."},
            {"role": "user", "content": prompt}
        ],
        max_tokens=4000,
        temperature=0.7
    )
    summary = response['choices'][0]['message']['content']  # チャットモデルの応答形式に合わせてアクセス方法を変更
    return summary

# URLからタイトルと本文を抽出
def extract_title_and_body(url):
    try:
        # User-Agent ヘッダを設定
        ua_headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36'
        }
        # URLからページを取得
        response = requests.get(url, headers=ua_headers)
        response.raise_for_status()

        # HTMLを解析
        soup = BeautifulSoup(response.text, 'html.parser')

        # タイトルを抽出
        title_text = soup.find('title').get_text()
        # タイトルが空の場合は処理を中断
        if not title_text:
            return None, None, None
        # タイトルを日本語に翻訳
        title = translate_to_japanese(title_text)
        
        # 本文を抽出し、日本語に翻訳
        body = ''
        for p in soup.find_all('p'):
            text = p.get_text()
            if(len(text) < 30):
                continue
            body += text + '\n'
            
        # 本文が空の場合は処理を中断
        if not body:
            return None, None, None
        
        # bodyから要約を生成
        body_ja = generate_summary(body)
        # bodyからハッシュタグを生成
        hashtags = generate_hashtags(body)
        # Google翻訳のリンクをつける
        google_translate_link = f"https://translate.google.com/translate?sl=auto&tl=ja&hl=ja&u={url}"
        body_ja += f"\n\n<a href=\"{google_translate_link}\" target=\"_blank\">元記事: {url} </a>\n"

        return title, body_ja, hashtags

    except requests.exceptions.RequestException as e:
        print(f"Error occurred while fetching the page: {e}")
        return None, None, None

    except AttributeError as e:
        print(f"Error occurred while parsing the HTML: {e}")
        return None, None, None

# カテゴリ名から カテゴリIDを取得する関数
def get_category_id(category_name):
    response = requests.get(f'{prompthub_url}/wp-json/wp/v2/categories', headers=headers)
    categories = response.json()
    for category in categories:
        # カテゴリ名に改行が含まれている場合があるので、改行をスペースに置換して比較
        category_without_newlines = category_name.replace('\n', ' ')
        if category['name'] == category_without_newlines:
            return category['id']
    return None

# WordPressに投稿
def post_to_wordpress(title, body, image_url, category):
    category_names = [f"{category}"] 
    
    # 新しい記事のデータを作成
    post_data = {
        'title': title,
        'content': body,
        'status': 'publish',  # 'draft' にすると下書き保存
        'categories': [],
        'featured_media': 0  # アイキャッチ画像のIDを指定（後で更新）
    }
    
    # カテゴリIDを取得して設定
    for category_name in category_names:
        category_id = get_category_id(category_name)
        if category_id:
            post_data['categories'].append(category_id)

    # 記事を投稿
    response = requests.post(
        f'{prompthub_url}/wp-json/wp/v2/posts', 
        headers=headers, 
        json=post_data)
    post_id = response.json()['id']
    post_link = response.json()['link']

    # 画像をダウンロード
    img_response = requests.get(image_url)
    # ダウンロードに失敗した場合はアイキャッチ画像は設定しない
    if img_response.status_code != 200:
        return post_link

    # 画像をアップロード
    img_headers = {
        'Authorization': f'Basic {credentials}',
        'Content-Type': 'image/png',
        'Content-Disposition': f'attachment; filename={image_url.split("/")[-1]}.png'}
    upl_response = requests.post(
        f'{prompthub_url}/wp-json/wp/v2/media', 
        headers=img_headers, 
        data=img_response.content)

    # アップロードに失敗した場合はアイキャッチ画像は設定しない
    if upl_response.status_code != 201:
        return post_link

    # アイキャッチ画像を記事に設定
    post_data['featured_media'] = upl_response.json()['id']
    response = requests.post(f'{prompthub_url}/wp-json/wp/v2/posts/{post_id}', headers=headers, json=post_data)

    return post_link

# X APIを使ってツイートする
def tweet(title, hashtags, post_url):
    try:
        message = f"{title}\n{hashtags}\n{post_url}"
        x_client.create_tweet(text=message)
    except Exception as e:
        print(f"Error occurred while tweeting: {e}")
    
# Lambdaのハンドラ関数
def lambda_handler(event, context):
    # eventからデータを取得
    category = event['keywords']  # カテゴリ名
    print(f"Category : {category}")
    
    # メール本文からURLを抽出
    urls = extract_urls(event['body'])

    # URLをひとつずつ処理
    for url in urls:
        # タイトルと本文を抽出
        title, body, hashtags = extract_title_and_body(url)
        if title is None:
            continue
        
        print(f"Title : {title}")
        # スクリーンショットを撮る
        ss_url = url2png(url, url2png_apikey, url2png_secret)
        # WordPress に投稿
        post_url = post_to_wordpress(title, body, ss_url, category)
        print(f"Article : {post_url}")
        # Xにツイート
        tweet(title, hashtags, post_url)
    
    return {
        'statusCode': 200,
        'body': json.dumps('PromptHub に記事を投稿しました！')
    }

extract_urlsという関数で、bodyに含まれているurlのリストを作成し、これをひとつずつ処理しています。
extract_title_and_bodyという関数で、urlが示す先のページのタイトルを取得し、日本語に翻訳して返しています。また、ページの本文はChatGPTで要約と解説を日本語で作成して返してます。
post_to_wordpressでWordpressに投稿しています。グループ内だけに公開するならSlackやNotionに投げるのもありかなと思います。
このスクリプトでは、最後にXにツイートも行っています。