bloggerへの自動投稿-ability -スキルアップをして人生を楽しむ。その日々の綴り。

今回は、競馬AI予想のブログをどうやって自動投稿しているのか、流れを整理していきましょうか。
できるだけ難しい言葉を減らして、全体像が分かるようにまとめるわね。

ポイントは3つです。
1つ目が Google Cloud Console の設定。
2つ目が Python から Blogger API を使う認証。
3つ目が MySQL などから取得したデータを記事に組み立てて投稿する流れです。

要するに、「データを集める」「記事を作る」「Bloggerへ投稿する」の3段階ってことね。
しかも、これを手動じゃなくて自動で回せるようにするわけだ。

まず Google Cloud Console 側では、Blogger API を使うための準備が必要です。
具体的には、プロジェクトを作成して、Google Auth Platform でアプリ情報を設定し、Desktop app の OAuth クライアントを作成します。

ここで取得する credentials.json が、Python側で認証に使う大事なファイルになるのよ。
一度設定してしまえば、あとはコード側で Blogger API と接続できるようになるわ。

でも、前に認証でハマったのよね。
トークンが切れたり、認証画面でエラーになったり。

そうですね。
今回は Desktop app の認証方式に合わせて、Pythonの run_local_server() を使う構成に直しました。
これで、手動再認証が必要なときも比較的追いやすくなっています。

つまり、認証が通れば、あとはPythonで記事本文を作って、Bloggerへそのまま投稿できるのね。

はい。記事の中身は、MySQLに保存してあるレース情報やAI予想結果を読み出して、HTMLに整形してから送っています。
タイトル、見出し、予想表、注意書きなどをコード側で組み立てるイメージです。

つまりBloggerに直接ベタ書きしてるんじゃなくて、Pythonが記事を生成してるってことか。
それなら毎回コピペしなくていいし、更新も早いわね。

そうです。
例えば、MySQLから当日のレース一覧を取得して、そのレースごとにAI予想の上位馬を取り出し、HTMLの表に流し込めば、そのまま記事として投稿できます。

ふむふむ。
じゃあ流れとしては、
「Google Cloud Consoleで認証準備」→「Pythonで認証」→「MySQLからデータ取得」→「HTML生成」→「Blogger投稿」って順番ね。

その理解で大丈夫よ。
自動投稿の仕組みって難しそうに見えるけれど、実際には役割を分けて考えるとかなり整理しやすいの。

そして重要なのは、投稿本文をHTMLで作ることです。
Blogger側で見た目を整えたいなら、表や見出し、注意文、コード表示などもPythonで出力できます。

このブログ自体がチャット風だから、説明記事までチャット形式にしてしまえば、読みやすさも上がるってわけね。

では次に、設定の流れとコード例も見ていきましょうか。

Google Cloud Console 側でやること

まずは Google Cloud 側で、Blogger API を使う準備をします。
やることは大きく4つです。

4つ？

はい。
1. プロジェクトを作る
2. Blogger API を有効化する
3. Google Auth Platform の Branding を設定する
4. Clients で Desktop app の OAuth クライアントを作って credentials.json をダウンロードする

この credentials.json を Python の実行フォルダに置いておくのが基本になるのね。

Python 側の認証イメージ


from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from google.auth.transport.requests import Request
import os
import pickle

SCOPES = ['https://www.googleapis.com/auth/blogger']

def get_blogger_service():
    creds = None

    if os.path.exists('token.pickle'):
        with open('token.pickle', 'rb') as f:
            creds = pickle.load(f)

    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(
                'credentials.json',
                SCOPES
            )
            creds = flow.run_local_server(
                host='localhost',
                port=8080,
                open_browser=False
            )

        with open('token.pickle', 'wb') as f:
            pickle.dump(creds, f)

    return build('blogger', 'v3', credentials=creds)

MySQL からデータを取得して記事を作る例

ここが実際の自動投稿の本体って感じね。

そうです。
たとえば、当日のレース名やAI予想結果をMySQLから取得して、HTMLを文字列で組み立てていきます。


import pandas as pd
import mysql.connector
from datetime import date

today_str = date.today().strftime('%Y-%m-%d')

conn = mysql.connector.connect(
    host='localhost',
    user='user',
    password='password',
    database='keiba'
)

query_races = f"""
SELECT race_id, race_name
FROM races
WHERE race_date = '{today_str}'
"""

races = pd.read_sql(query_races, conn)

for _, race in races.iterrows():
    race_id = race['race_id']
    race_name = race['race_name']

    query_preds = f"""
    SELECT overall_rank, horse_num, horse_name, predicted_time_lgbm, prob_3rd, rank_score
    FROM ai_predictions
    WHERE race_id = '{race_id}'
    ORDER BY overall_rank ASC
    LIMIT 5
    """

    preds = pd.read_sql(query_preds, conn)

    title = f"【競馬AI予想】{today_str} {race_name} の注目馬"

    content = f"<h2>{race_name} AI予想結果</h2>"
    content += "<table border='1' style='border-collapse: collapse; width:100%; text-align:center;'>"
    content += "<tr><th>順位</th><th>馬番</th><th>馬名</th><th>予想タイム</th><th>3着内確率</th><th>スコア</th></tr>"

    for _, row in preds.iterrows():
        content += "<tr>"
        content += f"<td>{row['overall_rank']}</td>"
        content += f"<td>{row['horse_num']}</td>"
        content += f"<td>{row['horse_name']}</td>"
        content += f"<td>{row['predicted_time_lgbm']:.2f}</td>"
        content += f"<td>{row['prob_3rd'] * 100:.1f}%</td>"
        content += f"<td>{row['rank_score']:.2f}</td>"
        content += "</tr>"

    content += "</table>"
    content += "<p>※この予想はAIによる計算結果です。最終的な判断はご自身でお願いします。</p>"

Blogger に投稿する処理


service = get_blogger_service()

body = {
    "kind": "blogger#post",
    "title": title,
    "content": content,
    "labels": ["競馬AI予想"]
}

service.posts().insert(
    blogId='あなたのBLOG_ID',
    body=body,
    isDraft=False
).execute()

なるほど。
データの取得と記事作成と投稿が、ちゃんと別れてるから見やすいわ。

そうなんです。
この構成にしておくと、あとで「タイトルだけ変えたい」「表の項目を増やしたい」「下書き投稿にしたい」みたいな修正も楽です。

自動投稿というと大げさに見えるけれど、実際には
「認証」「データ取得」「HTML化」「投稿」
という4つの部品に分かれているのね。

そして、その部品を順番につないでいけば、ブログ記事は自動で出来上がると。
これはなかなか面白い仕組みじゃない。

今後は、この仕組みをさらに安定化させて、定期実行やエラー通知まで含めて整えていく予定です。

「毎回手で投稿するのは大変」と感じている人にとっては、かなり実用的な方法かもしれないわね。

全体のサンプルコード

ここまでで流れは見えてきたけれど、やっぱり全体の形も見てみたいわね。
というわけで、簡易版のサンプルコードを載せておきましょうか。

はい。
実運用コードそのものではなく、処理の流れが分かりやすいように少し簡略化した形です。
「どの関数が何を担当しているか」が見やすい構成にしています。

こういうのがあると理解しやすいのよね。
いきなり長い実コードを見るより、まず全体像をつかみたいし。


import os
import pickle
from datetime import date

import pandas as pd
import mysql.connector
from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request

# Blogger API に投稿するための権限
SCOPES = ['https://www.googleapis.com/auth/blogger']

# 自分の Blogger の blogId を指定
BLOG_ID = 'あなたのBLOG_ID'


# ----------------------------------------
# Blogger API 用の認証を行い、service を返す関数
# ----------------------------------------
def get_blogger_service():
    # 保存済みトークンがあれば読み込む
    creds = None
    if os.path.exists('token.pickle'):
        with open('token.pickle', 'rb') as f:
            creds = pickle.load(f)

    # トークンが無い、または無効なら再認証
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            # リフレッシュトークンが有効なら自動更新
            creds.refresh(Request())
        else:
            # 初回認証、または再認証が必要な場合
            flow = InstalledAppFlow.from_client_secrets_file(
                'credentials.json',
                SCOPES
            )
            creds = flow.run_local_server(
                host='localhost',
                port=8080,
                open_browser=False
            )

        # 新しい認証情報を保存
        with open('token.pickle', 'wb') as f:
            pickle.dump(creds, f)

    # Blogger API に接続する service を作成
    return build('blogger', 'v3', credentials=creds)


# ----------------------------------------
# MySQL に接続する関数
# ----------------------------------------
def get_db_connection():
    return mysql.connector.connect(
        host='localhost',
        user='user',
        password='password',
        database='keiba'
    )


# ----------------------------------------
# 当日のレース一覧を取得する関数
# ----------------------------------------
def fetch_today_races(conn, target_date):
    query = f"""
    SELECT race_id, race_name
    FROM races
    WHERE race_date = '{target_date}'
    """
    return pd.read_sql(query, conn)


# ----------------------------------------
# 指定レースのAI予想上位を取得する関数
# ----------------------------------------
def fetch_race_predictions(conn, race_id):
    query = f"""
    SELECT overall_rank, horse_num, horse_name, predicted_time_lgbm, prob_3rd, rank_score
    FROM ai_predictions
    WHERE race_id = '{race_id}'
    ORDER BY overall_rank ASC
    LIMIT 5
    """
    return pd.read_sql(query, conn)


# ----------------------------------------
# Blogger に投稿するタイトルを作る関数
# ----------------------------------------
def build_post_title(target_date, race_name):
    return f"【競馬AI予想】{target_date} {race_name} の注目馬"


# ----------------------------------------
# Blogger に投稿する本文HTMLを作る関数
# ----------------------------------------
def build_post_content(race_name, preds):
    content = f"<h2>{race_name} AI予想結果</h2>"
    content += "<p>AIの計算結果から、注目馬をまとめました。</p>"
    content += "<table border='1' style='border-collapse: collapse; width:100%; text-align:center;'>"
    content += "<tr>"
    content += "<th>順位</th>"
    content += "<th>馬番</th>"
    content += "<th>馬名</th>"
    content += "<th>予想タイム</th>"
    content += "<th>3着内確率</th>"
    content += "<th>スコア</th>"
    content += "</tr>"

    for _, row in preds.iterrows():
        content += "<tr>"
        content += f"<td>{row['overall_rank']}</td>"
        content += f"<td>{row['horse_num']}</td>"
        content += f"<td>{row['horse_name']}</td>"
        content += f"<td>{row['predicted_time_lgbm']:.2f}</td>"
        content += f"<td>{row['prob_3rd'] * 100:.1f}%</td>"
        content += f"<td>{row['rank_score']:.2f}</td>"
        content += "</tr>"

    content += "</table>"
    content += "<p>※この予想はAIによる計算結果です。最終的な判断はご自身でお願いします。</p>"
    return content


# ----------------------------------------
# Blogger に1記事投稿する関数
# ----------------------------------------
def post_to_blogger(service, title, content):
    body = {
        "kind": "blogger#post",
        "title": title,
        "content": content,
        "labels": ["競馬AI予想"]
    }

    service.posts().insert(
        blogId=BLOG_ID,
        body=body,
        isDraft=False
    ).execute()


# ----------------------------------------
# 全体の処理をまとめて実行するメイン関数
# ----------------------------------------
def main():
    target_date = date.today().strftime('%Y-%m-%d')

    # DB接続
    conn = get_db_connection()

    try:
        # Blogger 認証
        service = get_blogger_service()

        # 当日のレースを取得
        races = fetch_today_races(conn, target_date)

        if races.empty:
            print('本日の対象レースはありません。')
            return

        # レースごとに記事を作成して投稿
        for _, race in races.iterrows():
            race_id = race['race_id']
            race_name = race['race_name']

            preds = fetch_race_predictions(conn, race_id)
            if preds.empty:
                print(f'{race_name}: 予測データなし')
                continue

            title = build_post_title(target_date, race_name)
            content = build_post_content(race_name, preds)
            post_to_blogger(service, title, content)

            print(f'{race_name}: 投稿完了')

    finally:
        conn.close()


# スクリプトとして直接実行されたときだけ main() を動かす
if __name__ == '__main__':
    main()

関数ごとの役割

こうやって分けると、かなり見やすいわね。
でも、各関数が何をやってるのかも軽く整理しておきたいかも。

では順番に見ていきましょう。
get_blogger_service() は、Blogger API に接続するための認証処理です。
保存済みトークンを使えるか確認して、必要なら再認証を行います。

つまり、Google へのログインまわりを担当している関数なのね。
ここが通れば、Bloggerへ記事を送る準備が整うわけだわ。

次の get_db_connection() は、MySQL への接続処理です。
記事の元になるデータはここから取ってくるので、土台の役割ですね。

その次の fetch_today_races() と fetch_race_predictions() が、実際に必要なデータを取ってくる部分ね。

その通りです。
前者は当日のレース一覧、後者は各レースの予想上位データを取得します。
ここを分けておくと、あとでSQLを調整したいときにも修正しやすくなります。

build_post_title() と build_post_content() は、記事の見た目を作る担当ね。
データそのものではなく、「どう記事に見せるか」を整える部分だわ。

ここが分かれてると、デザインだけ直したいときにも便利ね。
例えば表の列を増やしたり、見出し文を変えたり。

そして post_to_blogger() が、完成したタイトルと本文を実際にBloggerへ送る処理です。
ここは「投稿実行」だけに役割を絞っています。

最後の main() は、全体の司令塔みたいなものね。
認証して、データを取って、記事を作って、投稿する。
その流れを順番につないでいるの。

つまり、
「認証」
「DB接続」
「データ取得」
「記事生成」
「投稿実行」
を関数ごとに分けてるから、あとで保守しやすいってわけか。

はい。
自動投稿を安定させたいなら、こういう責務分割はかなり重要です。
どこで失敗したかも追いやすくなりますからね。

最初は難しそうに見えるけれど、こうして役割ごとに分けてみると、意外と整理しやすいのよ。
自動投稿の仕組みを作りたい人は、まずはこの形を土台にしてみると分かりやすいかもしれないわね。

bloggerへの自動投稿

Google Cloud Console 側でやること

Python 側の認証イメージ

MySQL からデータを取得して記事を作る例

Blogger に投稿する処理

全体のサンプルコード

関数ごとの役割

0 件のコメント:

コメントを投稿

検索

年月ごとアーカイブ

フレンドブログ

Translate

カスタム人気記事

自己紹介