The full AI content pipeline: from idea to post

TREND → RESEARCH → OUTLINE → TEXT → ADAPTATIONS → IMAGE → VIDEO → VOICE → PUBLISHING

{
  "brand": {
    "name": "Acme Realty",
    "voice": "Friendly expert. Not a salesperson. Facts + stories.",
    "audience": "Americans aged 35–55 who are thinking about moving to or investing in Ecuador",
    "languages": ["en"],
    "channels": ["newsletter", "youtube", "instagram", "linkedin"]
  },
  "week": "2026-10-05",
  "posts": [
    {
      "id": "post-001",
      "publish_at": "2026-10-05T09:00:00-05:00",
      "topic": "What it really costs to live in Cuenca in 2026",
      "content_type": "educational",
      "formats": {
        "blog": { "words": 1200, "status": "pending" },
        "newsletter": { "sections": 3, "status": "pending" },
        "youtube_script": { "duration_min": 8, "status": "pending" },
        "twitter": { "tweets": 5, "status": "pending" },
        "linkedin": { "status": "pending" }
      },
      "assets": {
        "cover_image": null,
        "short_video": null,
        "voice_over": null
      },
      "keywords": ["cost of living in Cuenca", "cost of living in Ecuador", "moving to Ecuador"],
      "approved": false
    }
  ]
}

import anthropic
import json
import os

claude = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])


def text_of(response) -> str:
    """Collects the text of a reply: newer models may put "thinking" blocks before the text."""
    return "".join(block.text for block in response.content if block.type == "text")


def generate_main_article(topic: str, keywords: list[str],
                           brand_voice: str, word_count: int = 1200) -> str:
    """Generates a full SEO-optimized article with Claude Sonnet."""
    response = claude.messages.create(
        model="claude-sonnet-5-5",  # current model IDs: see Anthropic's documentation
        max_tokens=8000,  # with headroom: the model's "thinking" also counts toward this limit
        messages=[{
            "role": "user",
            "content": f"""Write a blog article.

Topic: {topic}
Length: {word_count} words
SEO keywords: {', '.join(keywords)}
Brand voice: {brand_voice}

Structure:
1. H1 headline (with the main keyword)
2. Introduction (150 words, a hook + a promise)
3. 3–4 H2 sections with specific facts and numbers
4. Practical tips (bulleted list)
5. Conclusion + call to action

Requirements:
- Facts only, no filler like "this is very important"
- Specific numbers and examples; don't invent numbers: if you don't have solid data, mark the spot "to verify"
- Conversational but expert tone
- Language: American English, Markdown formatting"""
        }]
    )
    return text_of(response)


def adapt_to_all_platforms(main_article: str, brand_context: str,
                            topic: str) -> dict:
    """
    Turns one base article into content for every platform.
    Claude Haiku: fast and cheap (a couple of cents).
    """
    response = claude.messages.create(
        model="claude-haiku-4-5",  # Haiku 4.5: retirement from the API is possible no earlier than October 15, 2026; check the IDs in Anthropic's documentation
        max_tokens=6000,
        messages=[{
            "role": "user",
            "content": f"""You are the brand's content strategist. Brand: {brand_context}

Main article on the topic "{topic}":
---
{main_article}
---

Adapt it into the following formats. Return ONLY valid JSON, with no explanations and no triple backticks around it:

{{
  "newsletter": {{
    "subject_line": "Subject line (short and specific)",
    "sections": [
      "Section 1: the main story (up to 1,000 characters, with emoji, friendly)",
      "Section 2: goes deeper on one idea (800 characters)",
      "Section 3: a practical tip + CTA (600 characters)"
    ]
  }},
  "twitter_thread": [
    "Post 1/5: hook (up to 280 characters)",
    "Post 2/5: key fact",
    "Post 3/5: an example or a story",
    "Post 4/5: an unexpected angle",
    "Post 5/5: takeaway + link"
  ],
  "linkedin_post": "LinkedIn version (300–400 words, professional tone)",
  "instagram_caption": "Instagram version (150–200 words + 15 hashtags)",
  "youtube_description": "YouTube description (300 words, timestamps, keywords)"
}}"""
        }]
    )
    return json.loads(text_of(response))


def generate_youtube_script(topic: str, duration_minutes: int,
                              brand_voice: str) -> str:
    """Generates a YouTube video script with timestamps."""
    words_per_minute = 130    # a typical speaking pace; time yourself and adjust
    target_words = duration_minutes * words_per_minute

    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=8000,
        messages=[{
            "role": "user",
            "content": f"""Write a script for a YouTube video.

Topic: {topic}
Length: {duration_minutes} min (~{target_words} words)
Voice: {brand_voice}

Format for each block:
[MM:SS] BLOCK TITLE
(Director's note: what to show on screen / B-roll)
Host's lines...

Structure:
[00:00] HOOK: the first 30 seconds, the most important part
[00:30] INTRO: who the host is, what the video is about
[01:00] MAIN PART: 3–4 blocks of 2–3 minutes each
[{duration_minutes-1}:00] WRAP-UP: summary + CTA
[{duration_minutes-1}:30] OUTRO: subscribe, next video

Write in a lively, conversational way, as if you're talking to a friend."""
        }]
    )
    return text_of(response)

import asyncio
from telegram import Bot, InputFile
import os

TELEGRAM_BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
CHANNEL_ID = os.environ["TELEGRAM_CHANNEL_ID"]


async def publish_to_telegram(text: str, image_path: str = None) -> dict:
    """Publishes a post to a Telegram channel."""
    bot = Bot(token=TELEGRAM_BOT_TOKEN)

    if image_path:
        with open(image_path, "rb") as img:
            message = await bot.send_photo(
                chat_id=CHANNEL_ID,
                photo=InputFile(img),
                caption=text,
                parse_mode="Markdown"
            )
    else:
        message = await bot.send_message(
            chat_id=CHANNEL_ID,
            text=text,
            parse_mode="Markdown"
        )

    return {"message_id": message.message_id, "date": str(message.date)}


# To run it: asyncio.run(publish_to_telegram("Test post"))
# The bot must be an administrator of the channel; TELEGRAM_CHANNEL_ID is the channel address, like @your_channel

import json
import os
import requests

BUFFER_API_KEY = os.environ["BUFFER_API_KEY"]
BUFFER_CHANNEL_IDS = {
    "instagram": os.environ["BUFFER_INSTAGRAM_CHANNEL_ID"],
    "linkedin": os.environ["BUFFER_LINKEDIN_CHANNEL_ID"],
    "twitter": os.environ["BUFFER_TWITTER_CHANNEL_ID"],
}


def schedule_to_buffer(text: str, platform: str, due_at: str) -> dict:
    """
    Schedules a post through the Buffer API (GraphQL).
    due_at: publish time in ISO 8601 format, UTC, for example 2026-10-12T14:00:00.000Z
    Media goes to the Buffer API as a public link: see the documentation.
    """
    query = f"""
    mutation {{
      createPost(input: {{
        text: {json.dumps(text, ensure_ascii=False)},
        channelId: "{BUFFER_CHANNEL_IDS[platform]}",
        schedulingType: automatic,
        mode: customScheduled,
        dueAt: "{due_at}"
      }}) {{
        ... on PostActionSuccess {{ post {{ id dueAt }} }}
        ... on MutationError {{ message }}
      }}
    }}
    """
    response = requests.post(
        "https://api.buffer.com",
        headers={"Authorization": f"Bearer {BUFFER_API_KEY}"},
        json={"query": query},
    )
    return response.json()

Run on a schedule (Monday 09:00)
  → HTTP: read content-calendar.json from GitHub
  → Code: keep only approved: true
  → Loop: for each post:
    → Claude API: generate the article (Sonnet)
    → Claude API: platform adaptations (Haiku) [in parallel]
    → Image generation: cover [in parallel]
    → Buffer API: schedule LinkedIn and Twitter/X
    → Newsletter platform: save the issue as a draft
    → Google Docs: save everything for a final review
  → Slack or email notification: "X posts ready, waiting for review"

#!/usr/bin/env bash
# content-factory-orchestrator.sh
# Run it from the project folder: bash content-factory-orchestrator.sh (needs the jq tool)
set -euo pipefail

WEEK_DATE="${1:-$(date +%Y-%m-%d)}"
CALENDAR_FILE="content-calendar.json"
OUTPUT_DIR="./content-output/${WEEK_DATE}"

mkdir -p "$OUTPUT_DIR"
log() { echo "[$(date +%H:%M:%S)] $1"; }

log "🏭 Content factory started. Week: $WEEK_DATE"

POSTS=$(jq -r '.posts[] | select(.approved == true) | .id' "$CALENDAR_FILE")

if [ -z "$POSTS" ]; then
    log "⚠️ No approved posts. Stopping."
    exit 0
fi

for POST_ID in $POSTS; do
    log "▶️ Processing: $POST_ID"
    POST_DIR="${OUTPUT_DIR}/${POST_ID}"
    mkdir -p "$POST_DIR"

    TOPIC=$(jq -r ".posts[] | select(.id == \"$POST_ID\") | .topic" "$CALENDAR_FILE")
    KEYWORDS=$(jq -r ".posts[] | select(.id == \"$POST_ID\") | .keywords | join(\", \")" "$CALENDAR_FILE")
    PUBLISH_AT=$(jq -r ".posts[] | select(.id == \"$POST_ID\") | .publish_at" "$CALENDAR_FILE")

    log "  Topic: $TOPIC"

    # Article and YouTube script in parallel
    python3 scripts/generate_article.py --topic "$TOPIC" --keywords "$KEYWORDS" \
        --output "${POST_DIR}/article.md" &
    python3 scripts/generate_script.py --topic "$TOPIC" --duration 8 \
        --output "${POST_DIR}/youtube-script.md" &

    wait

    # Adaptations and cover image in parallel
    python3 scripts/adapt_platforms.py --article "${POST_DIR}/article.md" \
        --output "${POST_DIR}/adaptations.json" &
    python3 scripts/generate_image.py --topic "$TOPIC" \
        --output "${POST_DIR}/cover.jpg" &

    wait
    log "  ✅ Content for $POST_ID is ready"

    # Schedule the posts
    python3 scripts/schedule_content.py --post-id "$POST_ID" \
        --adaptations "${POST_DIR}/adaptations.json" \
        --cover "${POST_DIR}/cover.jpg" \
        --publish-at "$PUBLISH_AT"
done

log "🎉 Done! Posts scheduled: $(echo "$POSTS" | wc -w)"

def run_analytics_loop(days_back: int = 7) -> dict:
    """
    Collects last week's metrics and generates topics for next week.
    """
    # get_*_stats() and update_calendar_with_recommendations() are only named here:
    # write them for your own platforms (or ask Claude to)
    newsletter_stats = get_newsletter_stats(days_back)
    youtube_stats = get_youtube_stats(days_back)
    instagram_stats = get_instagram_stats(days_back)

    combined_stats = {
        "newsletter": newsletter_stats,
        "youtube": youtube_stats,
        "instagram": instagram_stats,
        "period_days": days_back,
    }

    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=6000,
        messages=[{
            "role": "user",
            "content": f"""You are a content strategy analyst.
Analyze the results for the last {days_back} days:

{json.dumps(combined_stats, ensure_ascii=False, indent=2)}

Give a structured analysis. Return only JSON, with no explanations and no triple backticks around it:
{{
  "winners": ["post + why it worked"],
  "flops": ["post + why it didn't land"],
  "patterns": ["what type of content works consistently"],
  "best_time": {{"newsletter": "HH:MM", "instagram": "HH:MM"}},
  "next_week_topics": ["topic 1", "topic 2", "topic 3", "topic 4", "topic 5"],
  "strategy_changes": ["what to change in the production process"]
}}

Use only the data in these statistics; don't make anything up."""
        }]
    )

    recommendations = json.loads(text_of(response))
    update_calendar_with_recommendations(recommendations)
    return recommendations

mkdir -p content-factory/{scripts,templates,logs}
cd content-factory
touch content-calendar.json content-factory-orchestrator.sh .env \
      scripts/generate_article.py scripts/generate_script.py \
      scripts/adapt_platforms.py scripts/generate_image.py \
      scripts/schedule_content.py

pip install python-telegram-bot