DEV Community

A3E Ecosystem
A3E Ecosystem

Posted on

How I Automated My Entire Content Distribution Stack with 200 Lines of Python

#ai #automation #programming #webdev

Every developer I know has the same content problem: you write something good, then spend more time figuring out where and how to post it than you spent writing it. LinkedIn wants different formatting than Dev.to. Hashnode has its own publication ID. X needs brevity, Mastodon needs federation. The platforms are fragmented; your time is not.

Here's how I solved it.

The Architecture

The core idea is a 4-tier routing decision that fires for every (brand × platform) pair:

1. API tier     — direct REST call (Bluesky, Dev.to, Hashnode, Gumroad, Mastodon)
2. Bridge tier  — Chrome extension bridge (X/Twitter, Threads, Instagram, LinkedIn)
3. Playwright   — headless persistent-context (TikTok, Bandcamp, Etsy, KDP)
4. Desktop      — PyAutoGUI for native file dialogs (YouTube, rare edge cases)
Enter fullscreen mode Exit fullscreen mode

Each tier has a tier_api(), tier_bridge(), or tier_browser() method on a BasePublisher subclass. The routing logic picks the best available tier at runtime based on what credentials exist:

class BasePublisher:
    DEFAULT_TIERS = ["api", "bridge", "browser"]

    def publish_with_fallback(self, content, metadata, content_type="post"):
        for tier in self.DEFAULT_TIERS:
            method = getattr(self, f"tier_{tier}", None)
            if method is None:
                continue
            result = method(content, metadata)
            if result.ok:
                return result
        return PublishResult.fail(...)
Enter fullscreen mode Exit fullscreen mode

Simple. Each platform subclass overrides DEFAULT_TIERS to reflect what's actually implemented. Bluesky only has ["api"]. X has ["bridge", "api"] — bridge first because it uses the logged-in session, API as fallback if OAuth credentials exist.

The Gate Layer

Before any publish fires, a gate checks four things:

def can_publish_now(brand: str, platform: str) -> tuple[bool, str]:
    limits = limits_for(brand, platform)

    # 1. Hard daily cap
    if today_count(brand, platform) >= limits["daily_max"]:
        return False, "daily_max hit"

    # 2. Burst prevention (platform-wide, cross-brand)
    if last_60min_count(platform) >= limits["burst_limit_60min"]:
        return False, f"burst_limit tripped"

    # 3. Minimum gap between posts for this brand
    if time_since_last_post(brand, platform) < limits["min_gap_minutes"] * 60:
        return False, "min_gap not met"

    # 4. Consecutive failure circuit breaker
    if consecutive_fail_count(brand, platform) >= FAIL_THRESHOLD:
        return False, "circuit_breaker: credentials may be expired"

    return True, "ok"
Enter fullscreen mode Exit fullscreen mode

The burst limit is cross-brand and platform-wide. This matters because multiple brands can share the same underlying account — you don't want 9 brands firing at once and tripping Twitter's rate limiter.

The Race Condition Fix

Here's the subtle part. If you have 4 parallel CPU wakes all checking the gate simultaneously:

Wake A: checks gate → 0 posts in last 60min → allowed ✓
Wake B: checks gate → 0 posts in last 60min → allowed ✓  ← RACE
Wake C: checks gate → 0 posts in last 60min → allowed ✓  ← RACE
Wake A: posts, logs → 1 post
Wake B: posts, logs → 2 posts  ← rate limit violation
Enter fullscreen mode Exit fullscreen mode

The fix: a platform-wide file-lock wraps the entire gate-check → publish → log sequence:

with platform_lock(platform, timeout=90.0):
    allowed, reason = can_publish_now(brand, platform)
    if not allowed:
        return gated_outcome(reason)
    return do_publish_and_log(brand, platform, content)
Enter fullscreen mode Exit fullscreen mode

Now wake B has to wait for wake A's log entry to land before its gate check runs. It will see count=1, the burst limit will trip, and it will correctly back off.

URL Verification

The last piece: every publish must return a real post URL, not a profile homepage. A publish_verifier module checks the URL against platform-specific regex patterns:

PLATFORM_PATTERNS = {
    "bluesky": [
        r"https://bsky\.app/profile/[^/]+/post/[A-Za-z0-9_-]+",  # real post
        r"https://bsky\.app/profile/[^/]+",                        # profile (ok with flag)
    ],
    "x_twitter": [
        r"https://(?:twitter|x)\.com/[^/]+/status/\d+",  # real post
        r"https://(?:twitter|x)\.com/[^/]+",              # profile (ok: bridge doesn't return permalinks)
    ],
    ...
}
Enter fullscreen mode Exit fullscreen mode

The profile_only_ok flag lets you accept profile URLs for platforms where the bridge path can't return a post permalink — X's compose box doesn't redirect you to the tweet after posting. You get the profile as confirmation it worked; the tweet is live. This is distinct from a failed publish that returns the platform homepage.

Results

After wiring this up across 9 content brands and 15+ platforms:

  • Bluesky: 100% success rate when gate is open
  • Hashnode: 4/4 real post URLs today
  • X/Twitter: ~42% real posts (rest are rate-gated correctly)
  • Threads: ~68% real posts (burst limit enforces natural spacing)
  • YouTube: blocked by API quota + CDP file-upload protection (that's a separate war)

The 200 lines of routing logic eliminated manual posting. The gate layer eliminated spam. The race-condition fix eliminated the parallel-wake over-posting bug. The URL verifier eliminated phantom success logs.

The code is unglamorous, but it works.

codevault builds developer tools and automation utilities. The publisher stack is part of a larger autonomous business operations system.

Related: When automation starts surfacing unexpected data anomalies and you are chasing hardware ghosts, the bug is often in the compiler, not the silicon. The One-Line Fix That Took 24 Hours to Find covers how a missing volatile qualifier corrupted sensor readings across 3,000 boots in a production embedded system.

Top comments (0)