Полный контент-конвейер — от идеи до публикации

ТРЕНД → ИССЛЕДОВАНИЕ → ПЛАН → ТЕКСТ → АДАПТАЦИИ → ИЗОБРАЖЕНИЕ → ВИДЕО → ГОЛОС → ПУБЛИКАЦИЯ

{
  "brand": {
    "name": "Acme Realty",
    "voice": "Дружелюбный эксперт. Не продажник. Факты + истории.",
    "audience": "Русскоязычные 35–55 лет, рассматривают переезд или инвестиции в Эквадор",
    "languages": ["ru"],
    "channels": ["telegram", "youtube", "instagram", "linkedin"]
  },
  "week": "2026-10-05",
  "posts": [
    {
      "id": "post-001",
      "publish_at": "2026-10-05T09:00:00-05:00",
      "topic": "Сколько реально стоит жизнь в Куэнке в 2026",
      "content_type": "educational",
      "formats": {
        "blog": { "words": 1200, "status": "pending" },
        "telegram": { "posts": 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": ["жизнь в Куэнке", "стоимость жизни Эквадор", "переезд Эквадор"],
      "approved": false
    }
  ]
}

import anthropic
import json
import os

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


def text_of(response) -> str:
    """Собирает текст ответа: у новых моделей перед текстом могут идти блоки «размышления»."""
    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:
    """Пишет полную статью с учётом поисковых запросов (SEO) через Claude Sonnet."""
    response = claude.messages.create(
        model="claude-sonnet-5-5",  # актуальные идентификаторы моделей: документация Anthropic
        max_tokens=8000,  # с запасом: «размышления» модели тоже входят в этот лимит
        messages=[{
            "role": "user",
            "content": f"""Напиши статью для блога.

Тема: {topic}
Объём: {word_count} слов
Ключевые слова SEO: {', '.join(keywords)}
Голос бренда: {brand_voice}

Структура:
1. H1 заголовок (с главным ключевым словом)
2. Введение (150 слов, крючок + обещание)
3. 3–4 раздела H2 с конкретными фактами и цифрами
4. Практические советы (маркированный список)
5. Вывод + призыв к действию

Требования:
- Только факты, без общих слов "очень важно"
- Конкретные цифры и примеры; не выдумывай цифры: если точных данных нет, пометь место «проверить»
- Разговорный но экспертный тон
- Язык: русский, форматирование Markdown"""
        }]
    )
    return text_of(response)


def adapt_to_all_platforms(main_article: str, brand_context: str,
                            topic: str) -> dict:
    """
    Из одной базовой статьи генерирует контент для всех платформ.
    Claude Haiku — быстро и экономно (пара центов).
    """
    response = claude.messages.create(
        model="claude-haiku-4-5",  # Haiku 4.5: вывод из API возможен не раньше 15.10.2026, сверь идентификаторы в документации Anthropic
        max_tokens=6000,
        messages=[{
            "role": "user",
            "content": f"""Ты контент-стратег бренда. Бренд: {brand_context}

Основная статья на тему "{topic}":
---
{main_article}
---

Адаптируй в следующие форматы. Верни ТОЛЬКО валидный JSON, без пояснений и без тройных обратных кавычек вокруг:

{{
  "telegram_posts": [
    {{
      "text": "Пост 1 (до 1000 символов, с эмодзи, дружелюбный)",
      "delay_hours": 0
    }},
    {{
      "text": "Пост 2 — углубляет одну мысль (800 символов)",
      "delay_hours": 24
    }},
    {{
      "text": "Пост 3 — практический совет + призыв к действию (600 символов)",
      "delay_hours": 48
    }}
  ],
  "twitter_thread": [
    "Твит 1/5: крючок (до 280 символов)",
    "Твит 2/5: ключевой факт",
    "Твит 3/5: пример или история",
    "Твит 4/5: неожиданный угол",
    "Твит 5/5: вывод + ссылка"
  ],
  "linkedin_post": "LinkedIn версия (300–400 слов, деловой тон)",
  "instagram_caption": "Instagram версия (150–200 слов + 15 хэштегов)",
  "youtube_description": "Описание YouTube (300 слов, таймкоды, ключевые слова)"
}}"""
        }]
    )
    return json.loads(text_of(response))


def generate_youtube_script(topic: str, duration_minutes: int,
                              brand_voice: str) -> str:
    """Пишет сценарий видео для YouTube с временными метками."""
    words_per_minute = 130    # примерный темп речи; замерь свой и поправь
    target_words = duration_minutes * words_per_minute

    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=8000,
        messages=[{
            "role": "user",
            "content": f"""Напиши сценарий видео для YouTube.

Тема: {topic}
Длительность: {duration_minutes} мин (~{target_words} слов)
Голос: {brand_voice}

Формат каждого блока:
[MM:SS] НАЗВАНИЕ БЛОКА
(Режиссёрская пометка: что показать на экране, какие перебивки)
Текст ведущего...

Структура:
[00:00] КРЮЧОК — первые 30 секунд, самое важное
[00:30] ВСТУПЛЕНИЕ — кто ведущий, о чём видео
[01:00] ОСНОВНАЯ ЧАСТЬ — 3–4 блока по 2–3 минуты
[{duration_minutes-1}:00] ВЫВОД — резюме + призыв к действию
[{duration_minutes-1}:30] КОНЦОВКА — подписка, следующее видео

Пиши живым разговорным языком, как будто говоришь с другом."""
        }]
    )
    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:
    """Публикует пост в Telegram канал."""
    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)}


# Запуск: asyncio.run(publish_to_telegram("Тестовый пост"))
# Бот должен быть администратором канала; TELEGRAM_CHANNEL_ID — это адрес канала вида @imya_kanala

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:
    """
    Планирует пост через Buffer API (GraphQL).
    due_at: время публикации в формате ISO 8601, UTC, например 2026-10-12T14:00:00.000Z
    Медиа в Buffer API передаются публичной ссылкой: см. документацию.
    """
    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()

Запуск по расписанию (понедельник, 09:00)
  → HTTP: читаем content-calendar.json из GitHub
  → Code: фильтруем approved: true
  → Loop: для каждого поста:
    → Claude API: генерируем статью (Sonnet)
    → Claude API: адаптации платформ (Haiku) [параллельно]
    → Генерация изображения: обложка [параллельно]
    → Buffer API: планируем LinkedIn и Twitter
    → Telegram Bot: планируем 3 Telegram поста
    → Google Docs: сохраняем для финальной проверки
  → Telegram уведомление: "X постов готовы, ожидают проверки"

#!/usr/bin/env bash
# content-factory-orchestrator.sh
# Запускать из папки проекта: bash content-factory-orchestrator.sh (нужна утилита jq)
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 "🏭 Контент-фабрика запущена. Неделя: $WEEK_DATE"

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

if [ -z "$POSTS" ]; then
    log "⚠️ Нет одобренных постов. Останавливаюсь."
    exit 0
fi

for POST_ID in $POSTS; do
    log "▶️ Обрабатываю: $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"

    # Статья и сценарий для YouTube параллельно
    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

    # Адаптации и обложка параллельно
    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 "  ✅ Контент для $POST_ID готов"

    # Планируем публикации
    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 "🎉 Готово! Постов запланировано: $(echo "$POSTS" | wc -w)"

def run_analytics_loop(days_back: int = 7) -> dict:
    """
    Собирает метрики прошлой недели и генерирует темы на следующую.
    """
    # get_*_stats() и update_calendar_with_recommendations() здесь только названы:
    # напиши их под свои площадки (или попроси Claude)
    telegram_stats = get_telegram_stats(days_back)
    youtube_stats = get_youtube_stats(days_back)
    instagram_stats = get_instagram_stats(days_back)

    combined_stats = {
        "telegram": telegram_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"""Ты аналитик контент-стратегии.
Проанализируй результаты за {days_back} дней:

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

Дай структурированный анализ. Верни только JSON, без пояснений и без тройных обратных кавычек вокруг:
{{
  "победители": ["пост + почему сработало"],
  "провалы": ["пост + почему не зашло"],
  "паттерны": ["какой тип контента стабильно работает"],
  "лучшее_время": {{"telegram": "HH:MM", "instagram": "HH:MM"}},
  "темы_на_следующую_неделю": ["тема 1", "тема 2", "тема 3", "тема 4", "тема 5"],
  "изменения_стратегии": ["что поменять в производственном процессе"]
}}

Используй только данные из статистики, не придумывай."""
        }]
    )

    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