Agentive
自動化ラボ

AIでSNS投稿を自動スケジュール — X/note/Instagramの一括管理

約5分で読めます

全体アーキテクチャ

1本の記事やアイデアから、複数SNS向けの投稿を自動生成し、最適な時間に自動投稿するパイプラインを構築する。

[記事/アイデア]

[AI投稿生成] → X用(140字) / note用(長文) / Instagram用(キャプション)

[投稿キュー登録] → 最適時間を自動計算

[スケジューラー] → 指定時刻に自動投稿

[分析] → エンゲージメント収集 → 次回の投稿改善

AI投稿生成

1つのソースから各プラットフォームに最適化した投稿を生成する。

from anthropic import Anthropic

client = Anthropic()

def generate_posts(source_content, platforms):
    """1つのコンテンツから複数SNS向け投稿を生成"""
    prompt = f"""以下のコンテンツから各SNS向けの投稿を生成してください。

## ソースコンテンツ
{source_content}

## 生成対象
{', '.join(platforms)}

## 各プラットフォームの制約
- X: 140文字以内、ハッシュタグ2-3個、フック重視
- note: 500-1000文字、見出し付き、読者への問いかけで締める
- Instagram: 150文字キャプション、ハッシュタグ10-15個

## 出力形式(JSON)
{{"x": "...", "note": "...", "instagram": "..."}}
"""

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        messages=[{"role": "user", "content": prompt}]
    )
    return json.loads(response.content[0].text)

投稿キュー管理

生成した投稿をキューに登録し、投稿状態を管理する。

import sqlite3
from datetime import datetime, timedelta

class PostQueue:
    def __init__(self, db_path="posts.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS posts (
                id INTEGER PRIMARY KEY,
                platform TEXT,
                content TEXT,
                scheduled_at TEXT,
                status TEXT DEFAULT 'queued',
                posted_at TEXT,
                engagement TEXT
            )
        """)

    def add(self, platform, content, scheduled_at):
        """投稿をキューに追加"""
        self.conn.execute(
            "INSERT INTO posts (platform, content, scheduled_at)"
            " VALUES (?, ?, ?)",
            (platform, content, scheduled_at.isoformat())
        )
        self.conn.commit()

    def get_due_posts(self):
        """投稿時刻が到来した投稿を取得"""
        now = datetime.now().isoformat()
        return self.conn.execute(
            "SELECT * FROM posts "
            "WHERE status = 'queued' AND scheduled_at <= ?",
            (now,)
        ).fetchall()

    def mark_posted(self, post_id):
        """投稿完了をマーク"""
        self.conn.execute(
            "UPDATE posts SET status = 'posted', "
            "posted_at = ? WHERE id = ?",
            (datetime.now().isoformat(), post_id)
        )
        self.conn.commit()

最適投稿時間の計算

過去のエンゲージメントデータから、プラットフォームごとの最適投稿時間を算出する。

OPTIMAL_TIMES = {
    "x": {
        "weekday": ["07:30", "12:00", "18:30", "21:00"],
        "weekend": ["09:00", "14:00", "20:00"]
    },
    "note": {
        "weekday": ["06:00", "12:00", "20:00"],
        "weekend": ["10:00", "15:00"]
    },
    "instagram": {
        "weekday": ["12:00", "19:00", "21:30"],
        "weekend": ["11:00", "16:00", "20:00"]
    }
}

def get_next_optimal_time(platform, after=None):
    """次の最適投稿時間を取得"""
    after = after or datetime.now()
    is_weekend = after.weekday() >= 5
    day_type = "weekend" if is_weekend else "weekday"

    times = OPTIMAL_TIMES[platform][day_type]

    for t in times:
        hour, minute = map(int, t.split(":"))
        candidate = after.replace(
            hour=hour, minute=minute, second=0
        )
        if candidate > after:
            return candidate

    # 今日の最適時間が全て過ぎていたら翌日の最初
    tomorrow = after + timedelta(days=1)
    is_weekend = tomorrow.weekday() >= 5
    day_type = "weekend" if is_weekend else "weekday"
    first_time = OPTIMAL_TIMES[platform][day_type][0]
    hour, minute = map(int, first_time.split(":"))
    return tomorrow.replace(
        hour=hour, minute=minute, second=0
    )

上記はデフォルト値。実際にはエンゲージメント履歴から学習し、自分のフォロワーに最適な時間帯を特定する。

マルチプラットフォーム投稿

各プラットフォームのAPIに投稿する統一インターフェース。

class MultiPlatformPoster:
    def __init__(self):
        self.posters = {
            "x": self._post_x,
            "note": self._post_note,
            "instagram": self._post_instagram
        }

    async def post(self, platform, content):
        poster = self.posters.get(platform)
        if not poster:
            raise ValueError(f"Unknown: {platform}")
        return await poster(content)

    async def _post_x(self, content):
        """X APIで投稿"""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "https://api.twitter.com/2/tweets",
                headers={
                    "Authorization": f"Bearer {X_TOKEN}"
                },
                json={"text": content}
            )
            return resp.json()

    async def _post_note(self, content):
        """note APIで下書き保存"""
        save_to_file(
            f"drafts/note/{datetime.now():%Y%m%d}.md",
            content
        )
        return {"status": "draft_saved"}

    async def _post_instagram(self, content):
        """Instagram Graph APIで投稿"""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"https://graph.facebook.com/v18.0/"
                f"{IG_USER_ID}/media",
                json={
                    "caption": content,
                    "image_url": generate_image_url(content),
                    "access_token": IG_TOKEN
                }
            )
            return resp.json()

スケジューラーの起動

import asyncio

async def scheduler_loop():
    """メインスケジューラーループ"""
    queue = PostQueue()
    poster = MultiPlatformPoster()

    while True:
        due_posts = queue.get_due_posts()
        for post in due_posts:
            post_id = post[0]
            platform, content = post[1], post[2]
            try:
                await poster.post(platform, content)
                queue.mark_posted(post_id)
                print(f"Posted to {platform}: "
                      f"{content[:50]}...")
            except Exception as e:
                print(f"Failed: {platform} - {e}")
        await asyncio.sleep(60)  # 1分ごとにチェック

asyncio.run(scheduler_loop())

投稿カレンダーの例

曜日XnoteInstagram
7:30 / 12:00-19:00
12:00 / 18:3020:00-
7:30 / 21:00-21:30
12:00 / 18:3012:0019:00
7:30 / 21:0020:00-
9:00 / 14:00-11:00
14:00 / 20:0010:0016:00

週に合計14-18投稿を自動管理。手作業なら毎日30分かかるところを完全自動化。

関連記事

A

Agentive 編集部

AIエージェントを実際に使い倒す個人開発者。サイト制作の自動化を実践しながら、その知見を発信しています。