Автопостинг на 8 платформах: архитектура waterfall, custom publisher’ы и API-ловушки

Построил 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.pyimport refrom pathlib import Pathfrom jinja2 import Environment, FileSystemLoaderTEMPLATES_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_mddef 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"]

Правила платформ:

Адаптация занимает ~30 секунд — читает источник, применяет правила всех платформ, пишет 8 вариантов. Каждый версионируется в git.

---

Стадия 2: Primary Hub — WordPress и Dev.to

Две платформы публикуются первыми. Обе производят URL, необходимые downstream-платформам.

WordPress

WordPress — SEO-якорь. Все социальные тизеры ссылаются на него как на канонический источник. Социальные платформы не индексируются Google; WordPress — да. Без него постится сиротский контент.

# api/publishers/wordpress.pyimport osimport requestsACCESS_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.pyimport osimport requestsAPI_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.pyimport osimport requestsfrom datetime import datetime, timezoneHANDLE = 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}

Три ловушки:

1. uploadBlob ожидает сырые байты с заголовком Content-Type — не multipart files={...}.

2. Таймаут по умолчанию 30 секунд слишком короткий для загрузки 1.5 МБ. Увеличен до 120 секунд после трёх подряд ошибок.

3. Лимит 300 графем — hard. len(text) считает кодпоинты; японские emoji считаются как несколько графем. Нужен regex.findall(r'\X', text).

Mastodon: двухступенчатая медиа-загрузка

# api/publishers/mastodon.pyimport osimport requestsBASE_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.pyimport osimport requestsBOT_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')# Загрузить фото на сервер VKUPLOAD_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.pyimport sysfrom pathlib import Pathfrom concurrent.futures import ThreadPoolExecutorsys.path.insert(0, str(Path(__file__).parent.parent))from api.publishers.bluesky import post as publish_blueskyfrom api.publishers.wordpress import publish_post as publish_wordpressfrom api.publishers.devto import publish_article as publish_devtofrom api.publishers.mastodon import post_status as publish_mastodon, upload_media as upload_mastodon_mediafrom api.publishers.telegram import send_message as send_telegram, send_photo as send_telegram_photofrom api.publishers.paragraph import publish_post as publish_paragraphfrom scripts.adaptation import adapt_for_platformdef 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:

1. Пользователь говорит «опубликуй эту статью»

2. Агент генерирует все варианты, показывает превью

3. Пользователь проверяет Dev.to draft (самый сложный вариант)

4. Пользователь говорит «подтверждаю» — только тогда выполняются API-запросы

Предотвращает ситуацию «ой, я только что опубликовал черновик». Каждый publisher имеет dry_run=True по умолчанию.

---

Краевые случаи: реальная учебная программа

---

Результаты