AIで自動収益化パイプラインを構築する — makemoneyコマンドの全貌

「コマンドひとつで収益が発生する」——これは誇張ではありません。私は make money というコマンドを実行するだけで、記事の生成、品質チェック、複数プラットフォームへの公開、アナリティクス収集、収益レポート生成までが自動で走る仕組みを構築しました。

この記事では、その自動収益化パイプラインの全設計を解剖します。8つのパイプライン、AEOスコアチェッカー、コスト追跡、障害分離——全てのコンポーネントを実装レベルで解説します。

Last updated: 2026-03-15

この記事でわかること

• 自動収益化パイプラインとは具体的に何をするのか？
• 8つのパイプラインの役割と連携方法
• AEO（Answer Engine Optimization）スコアチェッカーの実装
• コスト追跡と収益性の可視化
• 障害分離によるパイプラインの耐障害性
• cron統合と運用自動化の実際

目次

1. パイプラインアーキテクチャの全体像
2. 8つのパイプラインの詳細
3. AEOスコアチェッカーの実装
4. コスト追跡システム
5. 障害分離とサーキットブレーカー
6. cron統合と自動実行
7. 収益レポートの自動生成
8. 運用から得た最適化のヒント
9. まとめ

---

1. パイプラインアーキテクチャの全体像はどうなっているのか？

なぜ「パイプライン」なのか？

手動で記事を書き、手動で各プラットフォームに投稿し、手動でアナリティクスを確認する——この作業を毎日やると、1日2-3時間かかります。月間60-90時間です。

パイプライン化すると、これが 週10分のモニタリング に短縮されます。

アーキテクチャ概要

┌──────────────┐
│  make money   │  ← エントリポイント（Makefile）
└──────┬───────┘
       │
       ▼
┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│ 1. Content   │────▶│ 2. Quality   │────▶│ 3. Publish   │
│   Generator  │     │    Gate      │     │   Distributor │
└──────────────┘     └──────────────┘     └──────┬───────┘
                                                  │
       ┌──────────────────────────────────────────┤
       ▼                                          ▼
┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│ 4. Analytics │     │ 5. Revenue   │     │ 6. SEO/AEO   │
│   Collector  │     │    Tracker   │     │   Optimizer   │
└──────────────┘     └──────────────┘     └──────────────┘
       │                    │                     │
       ▼                    ▼                     ▼
┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│ 7. Report    │     │ 8. Feedback  │     │ (Error       │
│   Generator  │     │    Loop      │     │  Recovery)   │
└──────────────┘     └──────────────┘     └──────────────┘

各パイプラインは独立して動作し、一つが失敗しても他に影響しません。これが障害分離の基本原則です。

---

2. 8つのパイプラインはそれぞれ何をするのか？

パイプライン1: コンテンツジェネレーター

# engine/external/content_generator.py
class ContentPipeline:
    """トピック選定から下書き生成まで"""

    def generate(self, topic: str, platform: str) -> Article:
        # 1. トレンド分析でトピック優先度を決定
        priority = self.analyze_trend(topic)

        # 2. プラットフォーム特性に合わせたプロンプト構築
        prompt = self.build_prompt(topic, platform, priority)

        # 3. 記事生成（Claude API）
        draft = self.llm_generate(prompt)

        # 4. メタデータ付与
        article = Article(
            title=draft.title,
            body=draft.body,
            platform=platform,
            tags=self.extract_tags(draft),
            seo_description=self.generate_seo_desc(draft),
            created_at=datetime.now(),
        )
        return article

パイプライン2: 品質ゲート

生成された記事は自動で品質チェックを通過します。

CONTENT_QUALITY_CHECKS = {
    "word_count":     lambda a: 800 <= len(a.body) <= 5000,
    "heading_depth":  lambda a: count_headings(a.body) >= 3,
    "code_examples":  lambda a: count_code_blocks(a.body) >= 1,
    "no_secrets":     lambda a: not contains_secrets(a.body),
    "seo_meta":       lambda a: len(a.seo_description) >= 50,
    "aeo_score":      lambda a: calculate_aeo_score(a.body) >= 80,
    "freshness":      lambda a: has_date_signal(a.body),
    "links_valid":    lambda a: all_links_reachable(a.body),
    "factual_check":  lambda a: no_obvious_errors(a.body),
    "originality":    lambda a: plagiarism_score(a.body) < 0.15,
}

def run_content_gate(article: Article) -> GateResult:
    results = {name: check(article) for name, check in CONTENT_QUALITY_CHECKS.items()}
    score = sum(results.values()) / len(results) * 10
    return GateResult(passed=score >= 7.0, score=score, details=results)

スコア7.0未満の記事は自動で修正パイプラインに回され、改善版が再生成されます。

パイプライン3: マルチプラットフォーム配信

# engine/external/distributor.py
PLATFORMS = {
    "note":   NotePublisher(),
    "zenn":   ZennPublisher(),
    "qiita":  QiitaPublisher(),
    "devto":  DevToPublisher(),
}

async def distribute(article: Article, targets: list[str]):
    """複数プラットフォームに同時配信"""
    tasks = []
    for platform in targets:
        publisher = PLATFORMS[platform]
        adapted = publisher.adapt(article)  # プラットフォーム固有のフォーマット変換
        tasks.append(publisher.publish(adapted))

    results = await asyncio.gather(*tasks, return_exceptions=True)

    for platform, result in zip(targets, results):
        if isinstance(result, Exception):
            logger.error(f"{platform} 配信失敗: {result}")
            await notify_failure(platform, result)
        else:
            logger.info(f"{platform} 配信成功: {result.url}")
            await record_distribution(platform, result)

---

ここから先は有料エリアです。残りの5つのパイプライン（Analytics、Revenue Tracker、SEO/AEO Optimizer、Report Generator、Feedback Loop）の完全な実装、AEOスコアチェッカーの詳細アルゴリズム、コスト追跡の設計、そしてcron統合の実践的な設定を公開します。

---

3. AEOスコアチェッカーはどう実装するのか？

AEO（Answer Engine Optimization）とは何か？

AEOとは、AIアシスタント（ChatGPT、Claude、Perplexity等）が回答を生成する際に、あなたのコンテンツが引用元として選ばれるための最適化です。従来のSEO（検索エンジン最適化）の次世代版です。

2026年の現在、Google検索の40%以上がAI Overview（AI概要）付きで表示されます。あなたの記事がAI Overviewに引用されるかどうかは、収益に直結します。

AEOスコアの計算方法

# engine/external/aeo_checker.py
def calculate_aeo_score(content: str) -> int:
    """0-100のAEOスコアを返す"""
    score = 0
    max_score = 100

    # 1. 質問形式の見出し (25点)
    headings = extract_headings(content)
    question_headings = [h for h in headings if is_question(h)]
    question_ratio = len(question_headings) / max(len(headings), 1)
    score += min(25, int(question_ratio * 35))

    # 2. 直接的な回答 (25点)
    # 見出しの直後に簡潔な回答文があるか
    for heading in question_headings:
        answer = get_first_paragraph_after(content, heading)
        if answer and len(answer) <= 300:
            score += 5  # 各質問見出しに対して最大5点
    score = min(score, 50)  # 上限50点

    # 3. 構造化データ (20点)
    has_table = "| " in content and " |" in content
    has_list = bool(re.findall(r'^[-*] ', content, re.MULTILINE))
    has_code = "```" in content
    has_definition = bool(re.findall(r'\*\*[^*]+\*\*とは', content))
    score += sum([
        has_table * 5,
        has_list * 5,
        has_code * 5,
        has_definition * 5,
    ])

    # 4. 鮮度シグナル (15点)
    has_date = bool(re.search(r'(Last updated|更新日|2026)', content))
    has_version = bool(re.search(r'v\d+\.\d+', content))
    has_current_ref = bool(re.search(r'(2026年|最新|現在)', content))
    score += sum([
        has_date * 5,
        has_version * 5,
        has_current_ref * 5,
    ])

    # 5. 引用可能性 (15点)
    # 短く自己完結した定義文の数
    definitions = re.findall(
        r'\*\*[^*]+\*\*[はとは]+[^。]+。', content
    )
    score += min(15, len(definitions) * 3)

    return min(score, max_score)

スコア別の改善アクション

スコア	評価	アクション
90-100	優秀	そのまま公開
80-89	良好	軽微な調整で公開可
60-79	改善必要	質問見出しの追加、構造化データの強化
40-59	要修正	記事構成の大幅な見直し
0-39	再生成	トピック選定からやり直し

---

4. コスト追跡システムはどう設計するのか？

なぜコスト追跡が必要なのか？

自動化パイプラインの罠は、コストが見えにくいことです。Claude API呼び出し、各プラットフォームの手数料、インフラコスト——これらを可視化しなければ、「稼いでいるように見えて赤字」になります。

コストレイヤーの分解

# workspace/revenue/cost_tracker.py
COST_LAYERS = {
    "llm_api": {
        "provider": "anthropic",
        "model": "claude-sonnet-4-20250514",
        "input_cost_per_1k": 0.003,   # ドル/1Kトークン
        "output_cost_per_1k": 0.015,
    },
    "platform_fees": {
        "note": 0.15,      # 15%の手数料
        "gumroad": 0.10,   # 10%の手数料
        "zenn_badge": 0.0,  # バッジは手数料なし
    },
    "infrastructure": {
        "vercel": 0,        # Hobby無料枠
        "supabase": 0,      # 無料枠
        "domain": 1500,     # 年間 ¥1,500
    },
}

def calculate_article_cost(article: Article, generation_log: dict) -> Cost:
    """1記事あたりの生成コスト"""
    input_tokens = generation_log["input_tokens"]
    output_tokens = generation_log["output_tokens"]
    retries = generation_log.get("retries", 0)

    llm_cost = (
        input_tokens / 1000 * COST_LAYERS["llm_api"]["input_cost_per_1k"]
        + output_tokens / 1000 * COST_LAYERS["llm_api"]["output_cost_per_1k"]
    ) * (1 + retries)  # リトライ分を加算

    return Cost(
        llm=llm_cost,
        platform_fee_rate=COST_LAYERS["platform_fees"].get(article.platform, 0),
        total_fixed=llm_cost,
        currency="USD",
    )

収益性ダッシュボード

def generate_profitability_report(period: str = "monthly") -> dict:
    """収益性レポート"""
    revenue = sum_revenue(period)
    costs = sum_costs(period)
    articles_published = count_articles(period)

    return {
        "period": period,
        "gross_revenue": revenue,
        "total_costs": costs,
        "net_profit": revenue - costs,
        "margin": (revenue - costs) / max(revenue, 1) * 100,
        "cost_per_article": costs / max(articles_published, 1),
        "revenue_per_article": revenue / max(articles_published, 1),
        "roi": (revenue - costs) / max(costs, 1) * 100,
        "break_even_articles": (
            costs / max(revenue / max(articles_published, 1), 1)
        ),
    }

---

5. 障害分離はどう実装するのか？

パイプライン間の障害伝播を防ぐ方法

最も重要な設計原則：1つのパイプラインの障害が他に波及してはならない。

# engine/common/circuit_breaker.py
class CircuitBreaker:
    """サーキットブレーカーパターン"""

    def __init__(self, name: str, threshold: int = 3, reset_timeout: int = 300):
        self.name = name
        self.failure_count = 0
        self.threshold = threshold
        self.reset_timeout = reset_timeout  # 秒
        self.state = "CLOSED"  # CLOSED / OPEN / HALF_OPEN
        self.last_failure_time = None

    async def execute(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if self._should_try_reset():
                self.state = "HALF_OPEN"
            else:
                raise CircuitOpenError(f"{self.name}: 回路オープン中")

        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.threshold:
            self.state = "OPEN"
            logger.warning(
                f"CircuitBreaker '{self.name}': OPEN "
                f"({self.failure_count}回連続失敗)"
            )

    def _on_success(self):
        self.failure_count = 0
        self.state = "CLOSED"

    def _should_try_reset(self) -> bool:
        elapsed = time.time() - (self.last_failure_time or 0)
        return elapsed >= self.reset_timeout

各パイプラインのサーキットブレーカー設定

BREAKERS = {
    "note_publish":   CircuitBreaker("note",   threshold=3, reset_timeout=300),
    "qiita_publish":  CircuitBreaker("qiita",  threshold=3, reset_timeout=300),
    "devto_publish":  CircuitBreaker("devto",   threshold=5, reset_timeout=600),
    "analytics":      CircuitBreaker("analytics", threshold=3, reset_timeout=180),
    "llm_api":        CircuitBreaker("llm",     threshold=2, reset_timeout=120),
}

---

6. cron統合はどう行うのか？

実際のcron設定

# scripts/cron-content-pipeline.sh
#!/bin/bash
set -e

AEGIS_DIR="${AEGIS_HOME:-$HOME/Workspace/Work/aegis}"
LOG_DIR="$AEGIS_DIR/workspace/content/analytics"
TIMESTAMP=$(date +%Y-%m-%d_%H%M%S)

log() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a "$LOG_DIR/pipeline.log"
}

# 1. コンテンツ生成 (月水金 9:00)
generate_content() {
    log "コンテンツ生成パイプライン開始"
    cd "$AEGIS_DIR"

    # トレンドスキャンで今日のトピックを決定
    TOPIC=$(python3 -c "
from engine.external.trend_scanner import get_top_topic
print(get_top_topic('japanese_tech'))
")

    # 記事生成
    python3 scripts/generate-article.py \
        --topic "$TOPIC" \
        --platforms "note,qiita,zenn" \
        --quality-gate 7.0 \
        --aeo-minimum 80

    log "コンテンツ生成完了: $TOPIC"
}

# 2. アナリティクス収集 (毎日 21:00)
collect_analytics() {
    log "アナリティクス収集開始"
    python3 scripts/cron-analytics.sh
    log "アナリティクス収集完了"
}

# 3. 収益レポート (毎週日曜 10:00)
generate_revenue_report() {
    log "収益レポート生成開始"
    python3 -c "
from workspace.revenue.report_generator import generate_weekly_report
report = generate_weekly_report()
print(f'週次収益: ¥{report[\"gross_revenue\"]:,.0f}')
print(f'純利益: ¥{report[\"net_profit\"]:,.0f}')
print(f'記事数: {report[\"articles_published\"]}')
"
    log "収益レポート生成完了"
}

# メインルーティング
case "${1:-all}" in
    generate)  generate_content ;;
    analytics) collect_analytics ;;
    revenue)   generate_revenue_report ;;
    all)
        generate_content
        collect_analytics
        ;;
esac

crontabの設定例

# AIコンテンツパイプライン
0 9 * * 1,3,5  /path/to/aegis/scripts/cron-content-pipeline.sh generate
0 21 * * *     /path/to/aegis/scripts/cron-content-pipeline.sh analytics
0 10 * * 0     /path/to/aegis/scripts/cron-content-pipeline.sh revenue

---

7. 収益レポートはどう自動生成するのか？

ledger.jsonl形式のデータ構造

{"date":"2026-03-14","source":"note","type":"article_sale","gross":500,"fee":75,"net":425,"article_id":"005"}
{"date":"2026-03-14","source":"qiita","type":"badge","gross":120,"fee":0,"net":120,"article_id":"010"}
{"date":"2026-03-15","source":"gumroad","type":"product_sale","gross":2499,"fee":250,"net":2249,"product_id":"prompt-pack"}

自動レポート生成

import json
from collections import defaultdict
from datetime import datetime, timedelta

def generate_weekly_report() -> dict:
    """週次収益レポートの自動生成"""
    week_ago = datetime.now() - timedelta(days=7)

    entries = []
    with open("workspace/revenue/ledger.jsonl") as f:
        for line in f:
            entry = json.loads(line.strip())
            if datetime.fromisoformat(entry["date"]) >= week_ago:
                entries.append(entry)

    by_source = defaultdict(float)
    by_type = defaultdict(float)

    for e in entries:
        by_source[e["source"]] += e["net"]
        by_type[e["type"]] += e["net"]

    total_gross = sum(e["gross"] for e in entries)
    total_fee = sum(e["fee"] for e in entries)
    total_net = sum(e["net"] for e in entries)

    return {
        "period": f"{week_ago.date()} - {datetime.now().date()}",
        "gross_revenue": total_gross,
        "total_fees": total_fee,
        "net_revenue": total_net,
        "by_source": dict(by_source),
        "by_type": dict(by_type),
        "articles_published": len(set(
            e.get("article_id") for e in entries if e.get("article_id")
        )),
        "top_performer": max(entries, key=lambda e: e["net"])
            if entries else None,
    }

---

8. 運用から得た最適化のヒント

ヒント1: 公開曜日と時間帯で閲覧数が3倍変わる

実測データ（2026年1-3月）から：

プラットフォーム	最適な公開時間	理由
Qiita	月曜 10:00-11:00	週初めの技術調査
Zenn	水曜 12:00-13:00	昼休みの学習時間
dev.to	火曜 UTC 14:00	北米の午前中

ヒント2: リトライ戦略は指数バックオフ

async def publish_with_retry(publisher, article, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await publisher.publish(article)
        except RateLimitError:
            wait = 2 ** attempt * 30  # 30s, 60s, 120s
            logger.info(f"レート制限。{wait}秒後にリトライ")
            await asyncio.sleep(wait)
        except Exception as e:
            logger.error(f"配信失敗 (試行 {attempt+1}): {e}")
            if attempt == max_retries - 1:
                raise
    raise PublishError("最大リトライ回数超過")

ヒント3: 失敗した記事は捨てない

品質ゲートを通過しなかった記事は、修正候補キューに入ります。翌日のパイプライン実行時に、前日の失敗記事の改善を優先的に試みます。新規生成よりも修正の方がコストが低いためです。

ヒント4: プラットフォーム手数料を意識した価格設定

Gumroad: 10%手数料 → $24.99商品 = $22.49手取り
Zennバッジ: 手数料なし → ¥120 = ¥120手取り

同じコンテンツでも、手数料の低いプラットフォームを優先するだけで利益率が5-15%改善します。

---

9. まとめ — 自動収益化パイプラインの構築チェックリスト

• 8つのパイプラインが独立して動作する
• 品質ゲート（スコア7.0+、AEO 80+）が実装されている
• コスト追跡で収益性が可視化されている
• サーキットブレーカーで障害が分離されている
• cron統合で自動実行されている
• 収益レポートが週次で自動生成される
• リトライ戦略（指数バックオフ）が実装されている
• 失敗記事の修正キューが動いている

自動化の本質は「楽をすること」ではなく、人間の時間を最も価値の高い判断に集中させることです。パイプラインが日常業務を処理している間に、あなたは次の戦略を考えることができます。

---

この記事はAEGISプロジェクトで実稼働している自動収益化パイプラインに基づいています。

価格: ¥800

---

他の記事は note や Zenn でも公開しています。