Построил pipeline публикации контента на 8 платформах. Время распространения статьи сократилось с 50 минут до 90 секунд. Рассказываю, почему waterfall обходит parallel, какие API-ловушки встретились, и почему без человека в цикле нельзя.
Постановка задачи
Пишу каждый день. Два года каждая статья заканчивалась одинаково: открыть восемь вкладок, скопировать markdown, убрать форматирование для Telegram, сжать картинки для Mastodon, переписать хук для Bluesky (лимит 300 графем), вставить URL в каждую платформу, записать куда что ушло. Пятьдесят минут механической работы после завершения творческой части.
Попробовал Buffer и Zapier. Оба проваливаются по двум пунктам: не умеют адаптировать markdown в plaintext (Telegram и VK показывают `жирный как текст, не как жирный), и не поддерживают AT Protocol Bluesky или GraphQL Paragraph. Для нишевых платформ кастомные publisher'ы — единственный путь.
Задача: сделать так, чтобы после написания статьи оставалось нажать одну кнопку — и через 90 секунд контент был на всех платформах.
Архитектура: почему waterfall, а не parallel
Первая попытка была наивной: запускать все платформы одновременно. Провалилась сразу.
Проблема — цепочки зависимостей. Тизеры Bluesky нуждаются в URL WordPress, чтобы построить link cards. Нельзя запостить тизер до публикации канонической статьи. Dev.to и Paragraph имеют rate limits, которые при параллельном выполнении запускают каскад 429 на все publisher'ы. Одна платформа подвисает — убивает весь запуск.
Решение: waterfall с избирательной конкурентностью. Четыре стадии, каждая производит артефакты для следующей.
┌─────────────────────────────────────────────────────────────────────┐ │ PIPELINE: WATERFALL С ИЗБИРАТЕЛЬНОЙ КОНКУРЕНТНОСТЬЮ │ ├─────────────────────────────────────────────────────────────────────┤ │ │ │ Стадия 1: Адаптация контента (последовательно) │ │ → Парсинг markdown-источника │ │ → Генерация вариантов через Jinja2-шаблоны │ │ → Сжатие изображений под лимиты платформ (FFmpeg) │ │ │ │ Стадия 2: Primary Hub (последовательно, блокирующе) │ │ → WordPress: публикация полной статьи → получение канонического URL│ │ → Dev.to: публикация markdown-варианта → получение dev.to URL │ │ │ │ Стадия 3: Социальные тизеры (параллельно, 4 потока) │ │ → Bluesky: 300 графем + сжатое изображение + ссылка │ │ → Mastodon: 500 символов + медиа-загрузка + ссылка │ │ → Tumblr: фото-пост + подпись + ссылка │ │ → Telegram: plaintext + сжатое изображение + ссылка │ │ → VK: plaintext + фото + ссылка │ │ │ │ Стадия 4: Архивирование (последовательно) │ │ → Git-коммит со всеми опубликованными URL │ │ → Обновление индекса LLM-Wiki │ │ → Запись execution log для отладки │ │ │ └─────────────────────────────────────────────────────────────────────┘
Почему waterfall? Каждая стадия производит артефакты для следующей. URL WordPress становится канонической ссылкой для всех социальных платформ. Без него постится сиротский контент, исчезающий за 48 часов.
Стадия 1: Адаптация контента
Здесь умирает большинство pipeline'ов. Нельзя постить один markdown в Dev.to, Telegram и Bluesky. Dev.to рендерит ## заголовки нативно. Telegram показывает сырой #. Bluesky вырезает весь markdown и показывает plaintext.
Построил адаптационный слой на Jinja2 — один шаблон на платформу. Источник всегда один markdown-файл в vault LLM-Wiki. Адаптер читает его, применяет правила платформы, генерирует вариант.
# scripts/adaptation.py import re from pathlib import Path from jinja2 import Environment, FileSystemLoader TEMPLATES_DIR = Path(__file__).parent.parent / "templates" env = Environment(loader=FileSystemLoader(TEMPLATES_DIR)) def adapt_for_platform(source_md: str, platform: str) -> str: """Генерация варианта под конкретную платформу из канонического markdown.""" template = env.get_template(f"{platform}.j2") if platform in ("telegram", "vk"): text = re.sub(r'\*\*(.*?)\*\*', r'\1', source_md) # жирный text = re.sub(r'\*(.*?)\*', r'\1', text) # курсив text = re.sub(r'__(.*?)__', r'\1', text) # подчёркивание text = re.sub(r'~~(.*?)~~', r'\1', text) # зачёркивание text = re.sub(r'\[(.*?)\]\((.*?)\)', r'\2', text) # ссылки → голый URL text = re.sub(r'!\[.*?\]\(.*?\)', '', text) # удалить изображения text = re.sub(r'#{1,6}\s+', '', text) # удалить заголовки return template.render(content=text, has_images=False) elif platform == "bluesky": text = re.sub(r'#{1,6}\s+', '', source_md) text = re.sub(r'\*\*(.*?)\*\*', r'\1', text) text = re.sub(r'\n+', ' ', text) return text[:270].strip() + "..." # оставить место под URL elif platform == "devto": text = source_md.replace(".svg)", ".png)") return template.render(content=text, tags=extract_tags(source_md)) elif platform == "wordpress": return template.render(content=markdown_to_html(source_md)) elif platform == "mastodon": text = re.sub(r'\*\*(.*?)\*\*', r'\1', source_md) text = re.sub(r'\n+', ' ', text) return text[:470].strip() + "..." elif platform == "paragraph": return template.render(content=source_md, images_are_external=True) return source_md def extract_tags(md: str) -> list: """Извлечь теги из YAML frontmatter, ограничить до 4 для Dev.to.""" match = re.search(r'^tags:\s*(.+)', md, re.MULTILINE) if match: tags = [t.strip() for t in match.group(1).split(",")] return tags[:4] return ["python", "automation"]
Правила платформ:
Платформа | Markdown | Изображения | Лимит длины | Особенности |
|---|---|---|---|---|
Telegram / VK | Полная очистка | Только превью по голому URL | ~4000 символов | Голые URL для превью, без markdown |
Bluesky | Полная очистка | Blob-загрузка (2 МБ макс) | 300 графем | Подсчёт графем, не байтов |
Mastodon | Полная очистка | Двухступенчатая загрузка | 500 символов |
|
Нативный | Только PNG (SVG ломается) | Нет | Максимум 4 тега, режим draft | |
WordPress | HTML | Только внешние URL (free plan) | Нет | Scope |
Paragraph | Нативный | Только внешние URL | Нет | GitHub raw URL блокируются hotlink |
Habr | Ручная вставка | Загрузка через редактор | Нет | Формальный тон, форматирование кода |
Ручная вставка | Загрузка через редактор | Нет | Аналитический угол, бизнес-контекст |
Адаптация занимает ~30 секунд — читает источник, применяет правила всех платформ, пишет 8 вариантов. Каждый версионируется в git.
Стадия 2: Primary Hub — WordPress и Dev.to
Две платформы публикуются первыми. Обе производят URL, необходимые downstream-платформам.
WordPress
WordPress — SEO-якорь. Все социальные тизеры ссылаются на него как на канонический источник. Социальные платформы не индексируются Google; WordPress — да. Без него постится сиротский контент.
# api/publishers/wordpress.py import os import requests ACCESS_TOKEN = os.environ.get('WORDPRESS_ACCESS_TOKEN') BLOG_ID = os.environ.get('WORDPRESS_BLOG_ID') def publish_post(title: str, content: str, featured_image_url: str = None, categories: list = None, tags: str = "") -> dict: url = f"https://public-api.wordpress.com/rest/v1.2/sites/{BLOG_ID}/posts/new" headers = { "Authorization": f"Bearer {ACCESS_TOKEN}", "Content-Type": "application/json", } payload = { "title": title, "content": content, "status": "publish", } if categories: payload["categories"] = categories if tags: payload["tags"] = tags if featured_image_url: payload["featured_image"] = featured_image_url r = requests.post(url, headers=headers, json=payload, timeout=30) data = r.json() return { "success": "ID" in data, "url": data.get("URL"), "id": data.get("ID") }
Критическое ограничение: Free WordPress.com предоставляет только posts OAuth scope. Scope media требует платного Business ($25/мес). Загрузка изображений через API на free tier — невозможна. Решение: хостить изображения внешне и ссылаться по URL.
Dev.to
Dev.to — технический хаб. Markdown-нативный, code blocks работают, frontmatter теги авто-категоризуют. URL Dev.to становится ссылкой «опубликовано также на».
# api/publishers/devto.py import os import requests API_KEY = os.environ.get('DEVTO_API_KEY') def publish_article(title: str, body: str, tags: list, published: bool = False) -> dict: url = "https://dev.to/api/articles" headers = { "api-key": API_KEY, "Content-Type": "application/json" } payload = { "article": { "title": title, "body_markdown": body, "published": published, "tags": tags[:4] # Hard limit Dev.to: максимум 4 тега } } r = requests.post(url, headers=headers, json=payload, timeout=30) data = r.json() if r.status_code == 201: return {"success": True, "url": data.get("url"), "id": data.get("id")} return {"success": False, "error": data.get("error", f"HTTP {r.status_code}")}
Невидимые лимиты:
Максимум 4 тега. Присылать 5 — получить 422.
SVG не рендерится. Нужно конвертировать в PNG через FFmpeg перед загрузкой.
Режим draft:
published: falseсоздаёт черновик в дашборде.
Обе платформы должны завершиться до старта Стадии 3. Если WordPress упал — pipeline останавливается.
Стадия 3: Социальные тизеры — параллельно
После получения URL от WordPress и Dev.to шесть социальных платформ запускаются параллельно.
Bluesky: AT Protocol
Bluesky не использует REST. Использует AT Protocol — бинарные blob-загрузки, создание записей через JSON-RPC, лимит blob 2 МБ.
# api/publishers/bluesky.py import os import requests from datetime import datetime, timezone HANDLE = os.environ.get('BLUESKY_HANDLE') APP_PASSWORD = os.environ.get('BLUESKY_APP_PASSWORD') BASE_URL = "https://bsky.social/xrpc" def create_session() -> dict: r = requests.post( f"{BASE_URL}/com.atproto.server.createSession", json={"identifier": HANDLE, "password": APP_PASSWORD}, timeout=30 ) data = r.json() return { "success": "accessJwt" in data, "accessJwt": data.get("accessJwt"), "did": data.get("did") } def upload_blob(image_path: str, session: dict) -> dict: ext = image_path.lower().split(".")[-1] mime = {"jpg": "image/jpeg", "jpeg": "image/jpeg", "png": "image/png", "gif": "image/gif"}.get(ext, "image/png") with open(image_path, "rb") as f: data = f.read() r = requests.post( f"{BASE_URL}/com.atproto.repo.uploadBlob", headers={ "Authorization": f"Bearer {session['accessJwt']}", "Content-Type": mime }, data=data, timeout=120 # 1.5 МБ загрузки требуют времени ) blob = r.json().get("blob") return {"success": bool(blob), "blob": blob} def post(text: str, image_path: str = None) -> dict: session = create_session() if not session["success"]: return session now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") record = { "$type": "app.bsky.feed.post", "text": text, "createdAt": now, } if image_path and os.path.exists(image_path): if os.path.getsize(image_path) > 2_000_000: compressed = image_path.replace(".png", "-compressed.jpg") os.system(f"ffmpeg -y -i {image_path} -q:v 2 {compressed}") image_path = compressed blob = upload_blob(image_path, session) if blob["success"]: record["embed"] = { "$type": "app.bsky.embed.images", "images": [{"alt": "", "image": blob["blob"]}] } r = requests.post( f"{BASE_URL}/com.atproto.repo.createRecord", headers={"Authorization": f"Bearer {session['accessJwt']}"}, json={ "repo": session["did"], "collection": "app.bsky.feed.post", "record": record }, timeout=30 ) data = r.json() if "uri" in data: post_id = data["uri"].split("/")[-1] return { "success": True, "url": f"https://bsky.app/profile/{HANDLE}/post/{post\_id}" } return {"success": False, "error": data}
Три ловушки:
uploadBlobожидает сырые байты с заголовкомContent-Type— не multipartfiles={...}.Таймаут по умолчанию 30 секунд слишком короткий для загрузки 1.5 МБ. Увеличен до 120 секунд после трёх подряд ошибок.
Лимит 300 графем — hard.
len(text)считает кодпоинты; японские emoji считаются как несколько графем. Нуженregex.findall(r'\X', text).
Mastodon: двухступенчатая медиа-загрузка
# api/publishers/mastodon.py import os import requests BASE_URL = os.environ.get('MASTODON_URL') ACCESS_TOKEN = os.environ.get('MASTODON_ACCESS_TOKEN') def upload_media(image_path: str) -> dict: url = f"{BASE_URL}/api/v1/media" headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"} with open(image_path, "rb") as f: files = {"file": (os.path.basename(image_path), f)} r = requests.post(url, headers=headers, files=files, timeout=60) data = r.json() return { "success": "id" in data, "id": data.get("id"), "url": data.get("url") } def post_status(text: str, media_ids: list = None) -> dict: url = f"{BASE_URL}/api/v1/statuses" headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"} payload = {"status": text} if media_ids: payload["media_ids[]"] = media_ids r = requests.post(url, headers=headers, data=payload, timeout=30) data = r.json() return { "success": "id" in data, "url": data.get("url") }
Одним запросом создать статус с текстом и файлом — не работает. Два отдельных запроса обязательны.
Telegram: Bot API
# api/publishers/telegram.py import os import requests BOT_TOKEN = os.environ.get('TELEGRAM_BOT_TOKEN') CHAT_ID = os.environ.get('TELEGRAM_CHAT_ID') BASE_URL = f"https://api.telegram.org/bot{BOT\_TOKEN}" def send_message(text: str) -> dict: url = f"{BASE_URL}/sendMessage" payload = { "chat_id": CHAT_ID, "text": text, "parse_mode": "HTML", "disable_web_page_preview": False } r = requests.post(url, json=payload, timeout=30) data = r.json() return {"success": data.get("ok"), "message_id": data.get("result", {}).get("message_id")} def send_photo(image_path: str, caption: str = "") -> dict: url = f"{BASE_URL}/sendPhoto" with open(image_path, "rb") as f: files = {"photo": f} payload = {"chat_id": CHAT_ID, "caption": caption, "parse_mode": "HTML"} r = requests.post(url, data=payload, files=files, timeout=60) data = r.json() return {"success": data.get("ok"), "message_id": data.get("result", {}).get("message_id")}
Особенности Telegram:
parse_mode: HTML— поддерживает<b>,<i>,<a>, но не markdown.disable_web_page_preview: False— генерирует превью по голым URL.Фото-загрузка: multipart POST с файлом, не base64.
Лимит caption: 1024 символа. Большие статьи — только как текст без картинки.
VK: photos.upload + wall.post
VK API — самый сложный из всех. Требует предварительной загрузки фото на сервер VK, получения owner_id и photo_id, затем прикрепления к записи на стене.
# Получить URL сервера для загрузки VK_UPLOAD_URL=$(curl -s "https://api.vk.com/method/photos.getWallUploadServer?access\_token=${VK\_TOKEN}&v=5.199" | jq -r '.response.upload_url') # Загрузить фото на сервер VK UPLOAD_RESPONSE=$(curl -s -F "photo=@${IMAGE_PATH}" "${VK_UPLOAD_URL}") # Сохранить фото на стене SAVE_RESPONSE=$(curl -s "https://api.vk.com/method/photos.saveWallPhoto?server=${SERVER}&photo=${PHOTO}&hash=${HASH}&access\_token=${VK\_TOKEN}&v=5.199") OWNER_ID=$(echo $SAVE_RESPONSE | jq -r '.response[0].owner_id') PHOTO_ID=$(echo $SAVE_RESPONSE | jq -r '.response[0].id') # Опубликовать запись с прикреплённым фото curl -s "https://api.vk.com/method/wall.post?owner\_id=${OWNER\_ID}&message=${TEXT}&attachments=photo${OWNER\_ID}\_${PHOTO\_ID}&access\_token=${VK\_TOKEN}&v=5.199"
Сложности VK:
Токен
access_tokenтребует подтверждения прав через OAuth в браузере.Загрузка фото — двухступенчатая: получить upload_url → POST файл → получить server/photo/hash → сохранить через
photos.saveWallPhoto.Рекламные блоки: группы с >1000 подписчиков требуют пометку как рекламу. Личная страница — не требует.
Rate limit: 3 записи в сутки для личной страницы, 50 для группы.
Tumblr и Paragraph
Следуют аналогичным паттернам — загрузка медиа первым запросом, затем создание поста с media IDs или внешними URL.
Стадия 4: Архивирование и оркестратор
Hermes Agent планирует задачи, но никогда не autopublishes без явного подтверждения.
# scripts/orchestrator.py import sys from pathlib import Path from concurrent.futures import ThreadPoolExecutor sys.path.insert(0, str(Path(__file__).parent.parent)) from api.publishers.bluesky import post as publish_bluesky from api.publishers.wordpress import publish_post as publish_wordpress from api.publishers.devto import publish_article as publish_devto from api.publishers.mastodon import post_status as publish_mastodon, upload_media as upload_mastodon_media from api.publishers.telegram import send_message as send_telegram, send_photo as send_telegram_photo from api.publishers.paragraph import publish_post as publish_paragraph from scripts.adaptation import adapt_for_platform def run_pipeline(article_path: str, image_path: str, dry_run: bool = True): with open(article_path) as f: full_text = f.read() # Стадия 1: Адаптация variants = { "wordpress": adapt_for_platform(full_text, "wordpress"), "devto": adapt_for_platform(full_text, "devto"), "bluesky": adapt_for_platform(full_text, "bluesky"), "mastodon": adapt_for_platform(full_text, "mastodon"), "telegram": adapt_for_platform(full_text, "telegram"), "vk": adapt_for_platform(full_text, "vk"), "tumblr": adapt_for_platform(full_text, "tumblr"), "paragraph": adapt_for_platform(full_text, "paragraph"), } if dry_run: print("[DRY RUN] Варианты платформ сгенерированы:") for platform, text in variants.items(): print(f" {platform}: {len(text)} символов") return # Стадия 2: Primary Hub (последовательно) wp = publish_wordpress({ "title": "Название статьи", "content": variants["wordpress"], "status": "publish", "categories": ["Productivity", "Tools"], "tags": "automation, python, publishing" }) canonical_url = wp.get("url") dev = publish_devto({ "title": "Название статьи", "body_markdown": variants["devto"], "published": False, "tags": ["python", "automation", "publishing"] }) # Стадия 3: Социальные тизеры (параллельно, 6 потоков) def publish_bluesky_trailer(): text = f"{variants['bluesky']} → {canonical_url}" return publish_bluesky(text, image_path) def publish_mastodon_trailer(): media = upload_mastodon_media(image_path) text = f"{variants['mastodon']} → {dev.get('url', canonical_url)}" return publish_mastodon(text, [media["id"]] if media["success"] else []) def publish_tumblr_post(): return publish_tumblr(image_path, variants["tumblr"], canonical_url) def publish_paragraph_article(): return publish_paragraph("Название статьи", variants["paragraph"]) def publish_telegram_message(): return send_telegram_photo(image_path, variants["telegram"]) def publish_vk_post(): return publish_vk(image_path, variants["vk"], canonical_url) with ThreadPoolExecutor(max_workers=6) as executor: futures = { "bluesky": executor.submit(publish_bluesky_trailer), "mastodon": executor.submit(publish_mastodon_trailer), "tumblr": executor.submit(publish_tumblr_post), "paragraph": executor.submit(publish_paragraph_article), "telegram": executor.submit(publish_telegram_message), "vk": executor.submit(publish_vk_post), } social_results = {k: v.result() for k, v in futures.items()} # Стадия 4: Архивирование log_publish_results(wp, dev, social_results) git_commit_with_urls(wp, dev, social_results) update_llm_wiki_index(wp, dev, social_results)
Human-in-the-loop:
Пользователь говорит «опубликуй эту статью»
Агент генерирует все варианты, показывает превью
Пользователь проверяет Dev.to draft (самый сложный вариант)
Пользователь говорит «подтверждаю» — только тогда выполняются API-запросы
Предотвращает ситуацию «ой, я только что опубликовал черновик». Каждый publisher имеет dry_run=True по умолчанию.
Краевые случаи: реальная учебная программа
Краевой случай | Как обнаружил | Решение |
|---|---|---|
Bluesky blob > 2 МБ | Три подряд ошибки вечером четверга, все 400 без тела | Сжатие FFmpeg: |
WordPress free план блокирует media upload | 403 на каждой загрузке изображения | Хостить изображения внешне, ссылаться по URL |
Dev.to SVG сломан | Диаграмма неделю показывалась пустым прямоугольником |
|
Dev.to 5-й тег отклонён | Получил 422, прочитал ошибку | Жёсткое ограничение 4 тега. Без переговоров |
Mastodon двухступенчатая медиа | Три раза опубликовал текст без картинки | Два отдельных запроса: |
Bluesky 300 графем | Пост 305 символов, получил | Подсчёт графем через |
Paragraph hotlink-блокировка | GitHub raw URL показывались как битые картинки | VK CDN URL из |
LinkedIn токен истекает | Посты падали с 401 через 60 дней | Ручное обновление. Для ежемесячной публикации автоматизация не стоит затрат |
Telegram лимит caption | Статья 1200 символов не влезла в caption к фото | Разделение: фото с коротким caption + отдельное сообщение с полным текстом |
VK rate limit | 4-я запись в сутки вернула ошибку | Проверка лимита перед публикацией, откладывание на следующий день |
Markdown-расхождение | Telegram показывал | Полная очистка markdown для Telegram/VK на стадии адаптации |
NotebookLM cookies истекают | Cron задание в 11:00 молча падало на auth | Двухступенчатый cron: 10:30 напоминание → пользователь вставляет cookies → 11:00 публикация |
Результаты
Метрика | До pipeline | После pipeline |
|---|---|---|
Время на статью | 50 минут (8 платформ × ~6 мин) | 90 секунд (автоматизация) + 5 минут проверки |
Платформы | 3–4 в зависимости от энергии | 8 автоматизировано + черновики для 10 дополнительных |
Неудачные загрузки | ~30% (забыл картинку, неправильный формат, истёкший токен) | ~5% (только истечение токена) |
SEO-индексация | Нет — социальный контент эфемерен | WordPress как канонический источник, индексируется Google |
Переключение контекста | 8 вкладок, 8 разных UI, 8 copy-paste | Одна команда, одно превью, одно подтверждение |
Версионирование статей | Нет — распределённые копии расходились | Git отслеживает каждый вариант, платформу, URL |
