Gemini 3.8 Flashで始めるブログ・SNS全自動化：コピペで動くPython自律スクリプトと収益化設計

[情報ソース収集: RSS / トレンドAPI]
                 │
                 ▼
[オーケストレーター: Python 3.12]
                 │
   ┌─────────────┴─────────────┐
   ▼                           ▼
[Gemini 3.8 Flash]     [Gemini 3.8 Flash]
(記事本文・メタデータ生成)   (Xスレッド・要約生成)
   │                           │
   ▼                           ▼
[WordPress REST API]   [X API v2]
(下書き / 即時公開)      (自動スレッドポスト)
                 │
                 ▼
[Google Indexing API (即時インデックス要求)]

pip install google-genai pydantic requests python-dotenv

GEMINI_API_KEY=AIzaSy...あなたのAPIキー
WP_URL=https://your-domain.com
WP_USERNAME=your_admin_user
WP_APP_PASSWORD=xxxx xxxx xxxx xxxx

import os
from typing import List
import requests
from dotenv import load_dotenv
from google import genai
from google.genai import types
from pydantic import BaseModel, Field

load_dotenv()

# -------------------------------------------------------------------
# 1. Pydanticによる出力スキーマ定義（パース崩れを根絶）
# -------------------------------------------------------------------
class SocialThread(BaseModel):
    post_order: int = Field(description="スレッドの投稿順序（1始まり）")
    text: str = Field(description="140文字以内のSNS投稿本文。ハッシュタグ含む")

class ContentPipelineOutput(BaseModel):
    title: str = Field(description="SEO最適化された記事タイトル（32文字前後）")
    slug: str = Field(description="URL用英数字スラッグ（ハイフン区切り）")
    meta_description: str = Field(description="検索エンジン用ディスクリプション（120文字前後）")
    body_html: str = Field(description="WordPress用完全版HTML本文（h2, h3, p, ulタグのみで構成）")
    tags: List[str] = Field(description="関連記事タグ（3〜5個）")
    social_threads: List[SocialThread] = Field(description="SNS用スレッド投稿（3件）")

# -------------------------------------------------------------------
# 2. Gemini 3.8 Flash推論エンジン
# -------------------------------------------------------------------
client = genai.Client(api_key=os.environ.get("GEMINI_API_KEY"))

def generate_content_bundle(keyword_topic: str) -> ContentPipelineOutput:
    prompt = f"""
    テーマ: {keyword_topic}

    読者対象: テックエンジニア、副業実践者
    要件:
    1. 検索意図を満たす極めて実用的な技術解説または検証記事を作成してください。
    2. 読者の離脱を防ぐため、冒頭からBLUF原則（結論先行）で執筆してください。
    3. WordPressのコンテンツエディタに直接貼り付けられるクリーンなHTMLコードをbody_htmlに出力してください（```htmlなどのマークダウン装飾は不要）。
    4. 記事を宣伝するためのフックの効いたXスレッド投稿（各130文字以内）を3本作成してください。
    """

    response = client.models.generate_content(
        model="gemini-3.8-flash",
        contents=prompt,
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=ContentPipelineOutput,
            temperature=0.3,
            max_output_tokens=8192,
        ),
    )

    return ContentPipelineOutput.model_validate_json(response.text)

# -------------------------------------------------------------------
# 3. WordPress REST API自動パブリッシャー
# -------------------------------------------------------------------
def publish_to_wordpress(content: ContentPipelineOutput) -> int:
    wp_url = f"{os.environ.get('WP_URL')}/wp-json/wp/v2/posts"
    auth = (os.environ.get("WP_USERNAME"), os.environ.get("WP_APP_PASSWORD"))

    payload = {
        "title": content.title,
        "slug": content.slug,
        "content": content.body_html,
        "excerpt": content.meta_description,
        "status": "publish",  # 本番稼働初期は 'draft' 推奨
    }

    res = requests.post(wp_url, json=payload, auth=auth, timeout=30)
    if res.status_code not in (200, 201):
        raise RuntimeError(f"WordPress投稿失敗: {res.status_code} - {res.text}")

    post_id = res.json()["id"]
    return post_id

# -------------------------------------------------------------------
# 4. エントリポイント実行部
# -------------------------------------------------------------------
if __name__ == "__main__":
    target_topic = "PythonとDuckDBを用いたローカル大規模データ高速集計テクニック"
    print(f"[*] 生成パイプライン起動: {target_topic}")

    output_bundle = generate_content_bundle(target_topic)
    print(f"[+] 生成完了: {output_bundle.title}")

    post_id = publish_to_wordpress(output_bundle)
    print(f"[+] WordPress公開成功 (Post ID: {post_id})")

    print("\n--- 生成されたSNSスレッド ---")
    for post in output_bundle.social_threads:
        print(f"[{post.post_order}/3] {post.text}")

from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=2, max=60),
    retry=retry_if_exception_type(Exception)
)
def resilient_generate_bundle(topic: str):
    return generate_content_bundle(topic)

headers = {
    "User-Agent": "AutonomousPublisher/1.0 (+https://your-domain.com/bot-info)"
}
res = requests.post(wp_url, json=payload, auth=auth, headers=headers, timeout=30)