Forem: German Yamil

I Used Python and Claude to Write, Validate, and Publish a Bilingual Ebook — Full Cost, Timeline, and Output

German Yamil — Thu, 16 Apr 2026 02:29:57 +0000

The book you're reading about was written by the pipeline that produced it. That's not a marketing line — it's the only honest way to describe what happened. The ebook teaches Python automation. The pipeline that created it is Python automation. If the pipeline worked, the book is proof. If the book is wrong, so is the pipeline.

That recursive test is what made this worth building.

What I Built

The core is a state machine. Every chapter moves through four states: PENDING → RUNNING → DONE → NEEDS_REVIEW. Each transition gets logged to a SQLite database before anything else happens. The reason: LLM calls are expensive, slow, and non-deterministic. If the process crashes at chapter 7 out of 10, you do not want to regenerate chapters 1 through 6.

Crash recovery was not a nice-to-have. It was required after the first real run died halfway through on a network timeout and I realized I had no way to resume without reprocessing everything.

Here is what a typical pipeline run looked like in the terminal:

[2025-03-14 09:12:04] pipeline  INFO  Starting pipeline run — 10 chapters queued
[2025-03-14 09:12:05] ch_01     INFO  State: PENDING → RUNNING
[2025-03-14 09:14:31] ch_01     INFO  Content generated — 2,187 words
[2025-03-14 09:14:32] ch_01     INFO  AST validation: PASS (3 code blocks, 0 errors)
[2025-03-14 09:14:33] ch_01     INFO  Subprocess test: PASS (all scripts executed)
[2025-03-14 09:14:33] ch_01     INFO  State: RUNNING → DONE
[2025-03-14 09:14:34] ch_02     INFO  State: PENDING → RUNNING
...
[2025-03-14 13:47:09] pipeline  INFO  All chapters DONE — beginning EPUB compilation
[2025-03-14 13:47:44] pipeline  INFO  English EPUB: build/output_en.epub (1.2 MB)
[2025-03-14 13:52:18] pipeline  INFO  Spanish EPUB: build/output_es.epub (1.3 MB)
[2025-03-14 13:52:19] pipeline  INFO  Run complete — 4h 35m elapsed

The pipeline ran. Then it ran again when I changed the chapter structure. Then again when the translation QA caught ratio errors. The state machine meant each re-run only touched what actually needed to change.

The Actual Output

Ten chapters. Approximately 22,000 words total. The subject matter is Python automation for developers — file handling, subprocess management, API wrappers, CLI tools, testing patterns.

Each chapter shipped with at least one complete Python script. Ten scripts in total, all of them validated two ways before they made it into the manuscript:

AST validation — Python's ast.parse() confirmed the code was syntactically valid. This caught the hallucinated syntax errors before they became the reader's problem.
Subprocess testing — every script was actually executed in an isolated environment. Output was checked for exceptions and non-zero exit codes.

The book exists in two languages: English and Spanish. Translation was not a one-pass operation. A semantic QA step compared word ratios, checked that code blocks were preserved exactly, and flagged any chapter where the translation diverged structurally from the source. The first attempt failed that QA check on three chapters. The second pass passed all of them.

Cover image was generated with Imagen 4. One prompt, several iterations, final selection. The cover in the published version is the one that came out of that process.

Total active work time to get from blank project to two published EPUBs: 4 to 6 hours spread across about two weeks. Most of that time was debugging the pipeline itself, not writing content.

The Cost

This is the part people usually obscure. Here is the full accounting:

Tool	Cost
Claude Code Pro	$20/month
Gumroad	Free (revenue share on sales)
KDP	Free (royalty share on sales)
Python, SQLite, Pandoc	Free
Everything else	Free

One paid tool. Twenty dollars a month. That is the entire infrastructure cost.

Break-even is two sales at $19.99. The math is not complicated.

What Was Hard

Three things caused real problems:

LLM hallucinating code that does not compile. This happened more than expected. The model would generate a function that looked plausible in prose context but would throw a SyntaxError or reference a variable that was never defined. The AST validation step was built specifically because of this. Without it, bad code would have shipped inside chapters that looked fine in a document preview.

Crashes mid-pipeline. The first full run died on a network timeout after generating chapter 4. Without state persistence, that meant starting over. After implementing the SQLite state machine, subsequent crashes recovered by reading existing state and picking up from the last RUNNING chapter. This should have been designed in from the start. It was not.

Translation quality drifting from source structure. The first Spanish translation pass was technically fluent but structurally inconsistent. Some chapters had word counts that were disproportionately shorter or longer than the English versions. More importantly, a few code examples had been partially paraphrased in the surrounding explanation — acceptable for natural language, not acceptable for technical instructions where precision matters. The QA layer was added after that first pass failed inspection. The second pass was clean.

None of these problems were surprising in retrospect. They were all predictable failure modes of using an LLM in a production pipeline. The engineering response to each was straightforward. The time cost was real.

Was It Worth It?

The pipeline exists. The book exists. Two EPUBs in two languages, validated code, cover image, published on Gumroad and KDP. That part worked.

Whether it was worth it depends on a question I cannot answer yet: whether developers find the content useful enough to pay for it. The pipeline produced something real. The market will determine if it produced something valuable.

If you want to evaluate it directly: Python Automation for Developers is available on Gumroad for $19.99. There is a 30-day refund policy. If you buy it and the code does not work or the content does not deliver value, you can get your money back.

That is the only honest pitch I can make.

What Is Next

Book 2 is already in planning. Same pipeline. Different subject matter. The system is the asset — the pipeline generalizes to any technical content that can be structured as chapters with embedded, testable code.

The first run proved the pipeline worked. The second run will prove it was repeatable. If the output from book 2 is structurally similar in quality to book 1, that is the more interesting result: not that I built a thing, but that the thing scales.

That is the actual experiment. The book was just the first data point.

Automating Digital Product Launches: From EPUB to Gumroad Listing with Python

German Yamil — Wed, 15 Apr 2026 22:37:55 +0000

Publishing a digital product manually — upload file, write description, set price, add discount codes, verify everything is live — takes 30–45 minutes of copy-pasting and clicking. If you publish multiple products or update them frequently, that time compounds. This article automates the entire Gumroad launch flow from Python: create/update the listing, upload the EPUB, generate the description from your chapter summaries, configure discount codes, and run a pre-launch checklist before going live.

Gumroad API: Create and Update a Product

Gumroad's API uses form-encoded POST requests and returns JSON. Authentication is a bearer token from your account settings.

import os
import requests
from pathlib import Path

GUMROAD_API_BASE = "https://api.gumroad.com/v2"
ACCESS_TOKEN = os.environ["GUMROAD_ACCESS_TOKEN"]

def _gumroad_headers() -> dict:
    return {"Authorization": f"Bearer {ACCESS_TOKEN}"}


def create_product(
    name: str,
    price_cents: int,
    description: str,
    preview_url: str | None = None,
) -> dict:
    """Create a new Gumroad product. Returns the product dict."""
    payload = {
        "name": name,
        "price": price_cents,
        "description": description,
        "published": False,  # don't publish until checklist passes
    }
    if preview_url:
        payload["preview_url"] = preview_url

    r = requests.post(
        f"{GUMROAD_API_BASE}/products",
        headers=_gumroad_headers(),
        data=payload,
    )
    r.raise_for_status()
    product = r.json()["product"]
    print(f"Product created: {product['id']} — {product['name']}")
    return product


def update_product(product_id: str, **fields) -> dict:
    """Update fields on an existing Gumroad product."""
    r = requests.put(
        f"{GUMROAD_API_BASE}/products/{product_id}",
        headers=_gumroad_headers(),
        data=fields,
    )
    r.raise_for_status()
    return r.json()["product"]

Uploading a File to a Product

Gumroad requires the file upload as a multipart form. For large EPUB files, stream from disk rather than loading into memory:

def upload_product_file(product_id: str, file_path: Path) -> dict:
    """Upload a file to a Gumroad product. Returns the product file dict."""
    if not file_path.exists():
        raise FileNotFoundError(f"File not found: {file_path}")

    file_size_mb = file_path.stat().st_size / (1024 * 1024)
    print(f"Uploading {file_path.name} ({file_size_mb:.1f} MB)...")

    with file_path.open("rb") as f:
        r = requests.post(
            f"{GUMROAD_API_BASE}/products/{product_id}/product_files",
            headers=_gumroad_headers(),
            files={"file": (file_path.name, f, "application/epub+zip")},
        )
    r.raise_for_status()
    product_file = r.json()["product_file"]
    print(f"Upload complete: {product_file['id']}")
    return product_file

Generating a Product Description from Chapter Summaries

Rather than writing the description manually, generate it from the chapter summaries your pipeline already produced:

import anthropic
import json
from pathlib import Path

def generate_product_description(summaries_path: Path, title: str) -> str:
    """
    Read chapter summaries from a JSON file and generate a Gumroad
    product description using an LLM.
    """
    summaries = json.loads(summaries_path.read_text())
    chapter_list = "\n".join(
        f"- Chapter {s['number']}: {s['title']} — {s['summary']}"
        for s in summaries
    )

    client = anthropic.Anthropic()
    message = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": (
                f"Write a Gumroad product description for a technical ebook titled '{title}'.\n\n"
                f"Chapter summaries:\n{chapter_list}\n\n"
                "Requirements:\n"
                "- 200-300 words\n"
                "- Lead with the core problem the book solves\n"
                "- List 5-7 specific things the reader will be able to do\n"
                "- No hype words (revolutionary, game-changing, etc.)\n"
                "- End with who this book is NOT for\n"
                "- Plain text, no Markdown headings"
            ),
        }],
    )
    return message.content[0].text

Setting Up Discount Codes via API

Launch pricing and promo codes can be configured programmatically:

def create_offer_code(
    product_id: str,
    code: str,
    amount_off_cents: int,
    max_purchase_count: int | None = None,
) -> dict:
    """
    Create a discount code for a product.
    amount_off_cents: discount amount in cents (e.g. 500 = $5 off).
    """
    payload = {
        "offer_code": code,
        "amount_off": amount_off_cents,
    }
    if max_purchase_count:
        payload["max_purchase_count"] = max_purchase_count

    r = requests.post(
        f"{GUMROAD_API_BASE}/products/{product_id}/offer_codes",
        headers=_gumroad_headers(),
        data=payload,
    )
    r.raise_for_status()
    offer = r.json()["offer_code"]
    print(f"Offer code created: {offer['name']} — {amount_off_cents/100:.2f} off")
    return offer

Webhook Listener for Sales Events

A minimal FastAPI webhook to track sales in real-time:

from fastapi import FastAPI, Request
import logging

app = FastAPI()
log = logging.getLogger("gumroad.webhook")


@app.post("/webhooks/gumroad")
async def gumroad_sale(request: Request):
    """
    Gumroad sends a POST with form data on each sale.
    Verify the seller_id matches yours before processing.
    """
    data = await request.form()
    sale_data = dict(data)

    seller_id = sale_data.get("seller_id")
    expected_seller_id = os.environ["GUMROAD_SELLER_ID"]
    if seller_id != expected_seller_id:
        log.warning(f"Webhook seller_id mismatch: {seller_id}")
        return {"status": "ignored"}

    product_name = sale_data.get("product_name")
    sale_price  = sale_data.get("price")
    buyer_email = sale_data.get("email")
    sale_id     = sale_data.get("sale_id")

    log.info(f"Sale {sale_id}: {product_name} — ${int(sale_price)/100:.2f} — {buyer_email}")
    # Add your fulfillment logic here: send a receipt, update a dashboard, etc.

    return {"status": "ok"}

Run with: uvicorn webhook:app --host 0.0.0.0 --port 8000

Pre-Launch Checklist as Code

Before publishing, run a script that verifies everything is in place:

import sys

def run_launch_checklist(product_id: str, epub_path: Path) -> bool:
    checks = []

    # 1. EPUB exists and is a valid size
    if epub_path.exists() and epub_path.stat().st_size > 10_000:
        checks.append(("EPUB file exists and >10KB", True, ""))
    else:
        checks.append(("EPUB file exists and >10KB", False, str(epub_path)))

    # 2. Product exists on Gumroad
    r = requests.get(
        f"{GUMROAD_API_BASE}/products/{product_id}",
        headers=_gumroad_headers(),
    )
    if r.ok:
        product = r.json()["product"]
        checks.append(("Product exists on Gumroad", True, ""))

        # 3. Price is set
        price_ok = product.get("price", 0) > 0
        checks.append(("Price is set", price_ok, f"price={product.get('price')}"))

        # 4. Description is not empty
        desc_ok = len(product.get("description", "")) > 100
        checks.append(("Description is >100 chars", desc_ok, ""))

        # 5. At least one product file is attached
        files_r = requests.get(
            f"{GUMROAD_API_BASE}/products/{product_id}/product_files",
            headers=_gumroad_headers(),
        )
        files_ok = files_r.ok and len(files_r.json().get("product_files", [])) > 0
        checks.append(("Product file is uploaded", files_ok, ""))
    else:
        checks.append(("Product exists on Gumroad", False, f"HTTP {r.status_code}"))

    # 6. At least one discount code exists
    codes_r = requests.get(
        f"{GUMROAD_API_BASE}/products/{product_id}/offer_codes",
        headers=_gumroad_headers(),
    )
    codes_ok = codes_r.ok and len(codes_r.json().get("offer_codes", [])) > 0
    checks.append(("Discount code configured", codes_ok, ""))

    # Print results
    all_pass = True
    print("\n--- Launch Checklist ---")
    for label, passed, note in checks:
        status = "PASS" if passed else "FAIL"
        print(f"  [{status}] {label}" + (f" ({note})" if note else ""))
        if not passed:
            all_pass = False

    print(f"\nResult: {'READY TO PUBLISH' if all_pass else 'NOT READY — fix failures above'}")
    return all_pass


if __name__ == "__main__":
    product_id = os.environ["GUMROAD_PRODUCT_ID"]
    epub_path  = Path("outputs/book.epub")
    if not run_launch_checklist(product_id, epub_path):
        sys.exit(1)
    # Only publish if checklist passes
    update_product(product_id, published=True)
    print("Product is now live.")

The checklist approach means a bad deploy fails loudly with a specific failure message, not silently with a misconfigured listing.

Full pipeline + source code: germy5.gumroad.com/l/xhxkzz — $19.99, 30-day refund.

Building a Retry System with Exponential Backoff for LLM API Calls in Python

German Yamil — Wed, 15 Apr 2026 22:37:09 +0000

LLM APIs fail. Rate limits hit at inconvenient times, 503s appear during high-traffic periods, and timeouts spike when the model is under load. If your pipeline doesn't have a retry strategy, a transient failure kills an hours-long job. This article builds a complete retry system: a generic backoff decorator, specific handling for 429 rate-limit responses, and a circuit breaker to stop retrying when a service is genuinely down.

Why LLM APIs Fail

Understanding the failure mode shapes your retry strategy:

429 Too Many Requests: you've exceeded your rate limit. The Retry-After header tells you exactly how long to wait. Exponential backoff without reading this header wastes time or retries too early.
503 Service Unavailable: the provider is overloaded. Retryable — back off and try again.
500 Internal Server Error: server-side bug. Sometimes retryable (transient), sometimes not. Retry with a cap.
408 Request Timeout: the request took too long. Retryable.
400 Bad Request: your payload is malformed. Not retryable — retrying won't fix a broken prompt.
401 Unauthorized / 403 Forbidden: auth issue. Not retryable — fix your API key first.
422 Unprocessable Entity: your request is semantically invalid. Not retryable.

The rule: 4xx errors are generally your fault (except 429 and 408), 5xx errors are the server's fault.

Generic Retry Decorator with Exponential Backoff + Jitter

import time
import random
import logging
import functools
from typing import Callable, TypeVar, Any

log = logging.getLogger("pipeline.retry")

RETRYABLE_STATUS_CODES = {408, 429, 500, 502, 503, 504}

F = TypeVar("F", bound=Callable[..., Any])


def with_retry(
    max_attempts: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    backoff_factor: float = 2.0,
    jitter: bool = True,
) -> Callable[[F], F]:
    """
    Decorator that retries a function on retryable HTTP errors.
    Uses exponential backoff with optional full jitter.
    """
    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            attempt = 0
            while attempt < max_attempts:
                try:
                    return func(*args, **kwargs)

                except RetryableError as e:
                    attempt += 1
                    if attempt >= max_attempts:
                        log.error(f"Giving up after {max_attempts} attempts: {e}")
                        raise

                    delay = min(base_delay * (backoff_factor ** (attempt - 1)), max_delay)
                    if jitter:
                        # Full jitter: uniform random between 0 and delay
                        # Prevents thundering herd when many workers retry simultaneously
                        delay = random.uniform(0, delay)

                    if e.retry_after is not None:
                        delay = max(delay, e.retry_after)

                    log.warning(
                        f"Attempt {attempt}/{max_attempts} failed ({e}). "
                        f"Retrying in {delay:.1f}s"
                    )
                    time.sleep(delay)

                except NonRetryableError:
                    raise  # no point retrying

        return wrapper  # type: ignore
    return decorator


class RetryableError(Exception):
    def __init__(self, message: str, status_code: int, retry_after: float | None = None):
        super().__init__(message)
        self.status_code = status_code
        self.retry_after = retry_after


class NonRetryableError(Exception):
    def __init__(self, message: str, status_code: int):
        super().__init__(message)
        self.status_code = status_code

Specific Handling for 429: Read the Retry-After Header

The Retry-After header is either a number of seconds or an HTTP date. Parse both:

from email.utils import parsedate_to_datetime
from datetime import datetime, timezone


def parse_retry_after(header_value: str | None) -> float | None:
    """
    Parse the Retry-After header.
    Returns seconds to wait, or None if the header is absent/unparseable.
    """
    if not header_value:
        return None

    # Try it as a plain number of seconds first
    try:
        return float(header_value)
    except ValueError:
        pass

    # Try it as an HTTP date
    try:
        retry_at = parsedate_to_datetime(header_value)
        now = datetime.now(tz=timezone.utc)
        seconds = (retry_at - now).total_seconds()
        return max(0.0, seconds)
    except Exception:
        return None


def classify_http_error(status_code: int, headers: dict, body: str) -> Exception:
    """Convert an HTTP error response into the appropriate exception type."""
    if status_code in RETRYABLE_STATUS_CODES:
        retry_after = None
        if status_code == 429:
            retry_after = parse_retry_after(headers.get("Retry-After"))
        return RetryableError(
            f"HTTP {status_code}: {body[:200]}",
            status_code=status_code,
            retry_after=retry_after,
        )
    else:
        return NonRetryableError(
            f"HTTP {status_code}: {body[:200]}",
            status_code=status_code,
        )

Circuit Breaker: Stop Retrying After N Consecutive Failures

A circuit breaker prevents your pipeline from hammering an API that's genuinely down. After N consecutive failures it "opens" and fast-fails all calls. After a cooldown period it enters "half-open" state and allows one test call through.

import threading
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"        # normal operation
    OPEN = "open"            # failing fast
    HALF_OPEN = "half_open"  # testing recovery


class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 60.0,
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._last_failure_time: float | None = None
        self._lock = threading.Lock()

    @property
    def state(self) -> CircuitState:
        with self._lock:
            if self._state == CircuitState.OPEN:
                if (
                    self._last_failure_time is not None
                    and time.monotonic() - self._last_failure_time >= self.recovery_timeout
                ):
                    self._state = CircuitState.HALF_OPEN
                    log.info("Circuit breaker entering HALF_OPEN — testing recovery")
            return self._state

    def record_success(self):
        with self._lock:
            self._failure_count = 0
            self._state = CircuitState.CLOSED

    def record_failure(self):
        with self._lock:
            self._failure_count += 1
            self._last_failure_time = time.monotonic()
            if self._failure_count >= self.failure_threshold:
                if self._state != CircuitState.OPEN:
                    log.error(
                        f"Circuit breaker OPEN after {self._failure_count} consecutive failures"
                    )
                self._state = CircuitState.OPEN

    def call(self, func: Callable, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            raise RuntimeError("Circuit breaker is OPEN — not attempting call")
        try:
            result = func(*args, **kwargs)
            self.record_success()
            return result
        except Exception:
            self.record_failure()
            raise

Putting It All Together

import requests

circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60.0)


@with_retry(max_attempts=5, base_delay=1.0, max_delay=60.0)
def call_llm_api(prompt: str, model: str, api_key: str) -> str:
    """Make a retryable LLM API call wrapped in a circuit breaker."""
    def _call():
        response = requests.post(
            "https://api.anthropic.com/v1/messages",
            headers={
                "x-api-key": api_key,
                "anthropic-version": "2023-06-01",
                "content-type": "application/json",
            },
            json={
                "model": model,
                "max_tokens": 4096,
                "messages": [{"role": "user", "content": prompt}],
            },
            timeout=120,
        )

        if not response.ok:
            raise classify_http_error(
                response.status_code,
                dict(response.headers),
                response.text,
            )

        return response.json()["content"][0]["text"]

    return circuit_breaker.call(_call)

With this setup: transient 503s retry automatically with backoff, 429s wait the exact number of seconds the server requests, 400s fail immediately without wasting retries, and five consecutive failures open the circuit and surface a clear error to your pipeline rather than thrashing for minutes.

Full pipeline + source code: germy5.gumroad.com/l/xhxkzz — $19.99, 30-day refund.

Extracting and Validating Code Blocks from LLM-Generated Markdown in Python

German Yamil — Wed, 15 Apr 2026 22:37:00 +0000

LLMs produce inconsistent Markdown. A prompt that returns a clean python ` block on one run might return `Python, `python, or a bare block with no language tag on the next. If your pipeline ingests that output and tries to save or execute the code, inconsistency breaks everything downstream. This article shows how to extract fenced code blocks reliably, validate them, fix common LLM-introduced errors, and integrate the whole thing into a chapter-level pipeline.

Why LLMs Hallucinate Code Fences

LLMs are trained on diverse Markdown across the web. Code fences appear as:

python ` (correct)
`Python (capitalized language tag)
python ` (space before language tag)
`py (alias)
` (no language tag)
Indented four-space blocks (old Markdown style)
Code inside a <pre> tag (if the model pulls from HTML training data)

They also occasionally emit mismatched fence lengths (

` vs `

`) or forget the closing fence entirely, leaving the rest of the document inside the block. Your extractor has to handle all of these.

Regex vs AST: When to Use Each

Use regex for extraction: you want to pull blocks out of arbitrary text without a full Markdown parse. It's fast, handles partial/malformed documents, and runs on raw strings.

Use AST (via Python's ast.parse()) for validation of the extracted Python code: it catches syntax errors that regex cannot detect, without executing the code.

Never use regex to validate Python syntax — and never exec() untrusted code to check if it runs.

Extracting All Fenced Code Blocks

`python
import re
from dataclasses import dataclass
from typing import Optional

@dataclass
class CodeBlock:
language: str # normalized to lowercase, e.g. "python"
source: str # raw code content
index: int # position in document (0-based)

FENCE_PATTERN = re.compile(
r"^(`{3,}|~{3,})\s*(?P[a-zA-Z0-9_+-])\s\n" # opening fence + optional lang
r"(?P.?)" # code content (non-greedy) r"^\1\s$", # closing fence (same length) re.MULTILINE | re.DOTALL, )

LANG_ALIASES = {
"py": "python",
"python3": "python",
"Python": "python",
"js": "javascript",
"ts": "typescript",
"sh": "bash",
"shell": "bash",
"": "unknown",
}

def extract_code_blocks(markdown: str) -> list[CodeBlock]:
"""Extract all fenced code blocks from a Markdown string."""
blocks = []
for i, match in enumerate(FENCE_PATTERN.finditer(markdown)):
raw_lang = match.group("lang").strip()
lang = LANG_ALIASES.get(raw_lang, raw_lang.lower())
code = match.group("code")
blocks.append(CodeBlock(language=lang, source=code, index=i))
return blocks
`

The key detail: the closing fence regex (^\1) backreferences the opening fence, so ` only closes with `

, not `. This handles nested or mismatched fences correctly.

Detecting and Fixing Common LLM Code Issues

Missing Imports

LLMs often use libraries in code examples without importing them. This heuristic scan catches the most common cases:

`python
import ast

COMMON_MISSING_IMPORTS = {
"Path": "from pathlib import Path",
"dataclass": "from dataclasses import dataclass, field",
"Optional": "from typing import Optional",
"List": "from typing import List",
"Dict": "from typing import Dict",
"datetime": "from datetime import datetime",
"json": "import json",
"os": "import os",
"re": "import re",
"logging": "import logging",
}

def detect_missing_imports(code: str) -> list[str]:
"""Return suggested import statements for symbols used but not imported."""
try:
tree = ast.parse(code)
except SyntaxError:
return [] # syntax errors handled separately

# Collect all names defined by imports in this block
imported_names = set()
for node in ast.walk(tree):
    if isinstance(node, ast.Import):
        for alias in node.names:
            imported_names.add(alias.asname or alias.name.split(".")[0])
    elif isinstance(node, ast.ImportFrom):
        for alias in node.names:
            imported_names.add(alias.asname or alias.name)

# Collect all Name nodes used at top level
used_names = {node.id for node in ast.walk(tree) if isinstance(node, ast.Name)}

suggestions = []
for name, import_stmt in COMMON_MISSING_IMPORTS.items():
    if name in used_names and name not in imported_names:
        suggestions.append(import_stmt)

return suggestions

Wrong Indentation

LLMs sometimes produce mixed tabs and spaces. Python 3 raises TabError on this. Normalize before parsing:

`python def fix_indentation(code: str) -> str: """Replace tabs with 4 spaces to avoid TabError on parse.""" return code.expandtabs(4) `

Validate with AST

`python def validate_python(code: str) -> tuple[bool, Optional[str]]: """ Try to parse code as Python AST. Returns (is_valid, error_message). """ try: ast.parse(fix_indentation(code)) return True, None except SyntaxError as e: return False, f"SyntaxError at line {e.lineno}: {e.msg}" except ValueError as e: return False, f"ValueError: {e}" `

Integration with a Chapter Pipeline

The full extraction step in a chapter-level pipeline: parse the LLM response, validate Python blocks, prepend missing imports, and store code separately from prose.

`python
from pathlib import Path
import json

def process_chapter_output(
markdown: str,
chapter_id: str,
output_dir: Path,
) -> dict:
"""
Extract code from a chapter's Markdown output.
Returns a summary dict with validation results.
"""
blocks = extract_code_blocks(markdown)
results = {"chapter_id": chapter_id, "blocks": []}

code_dir = output_dir / "code" / chapter_id
code_dir.mkdir(parents=True, exist_ok=True)

for block in blocks:
    result = {
        "index": block.index,
        "language": block.language,
        "valid": None,
        "error": None,
        "missing_imports": [],
        "file": None,
    }

    if block.language == "python":
        fixed = fix_indentation(block.source)
        is_valid, error = validate_python(fixed)
        result["valid"] = is_valid
        result["error"] = error

        if is_valid:
            suggestions = detect_missing_imports(fixed)
            result["missing_imports"] = suggestions

            # Prepend missing imports and save
            final_code = "\n".join(suggestions) + ("\n\n" if suggestions else "") + fixed
            filename = code_dir / f"block_{block.index:02d}.py"
            filename.write_text(final_code)
            result["file"] = str(filename)

    results["blocks"].append(result)

# Save summary
summary_path = output_dir / "code" / f"{chapter_id}_summary.json"
summary_path.write_text(json.dumps(results, indent=2))

valid_count = sum(1 for b in results["blocks"] if b.get("valid") is True)
print(f"Chapter {chapter_id}: {len(blocks)} blocks, {valid_count} valid Python")
return results

This gives you a clean separation between prose (the full Markdown file) and code (individually saved .py files with an accompanying JSON summary). When a block fails AST validation, you know exactly which chapter and which block index to fix, without searching through thousands of words of text.

Full pipeline + source code: germy5.gumroad.com/l/xhxkzz — $19.99, 30-day refund.

How to Structure a Python Project for Long-Running LLM Batch Jobs

German Yamil — Wed, 15 Apr 2026 22:36:15 +0000

LLM batch jobs are not like web requests. They run for hours, they call expensive APIs, and they fail partway through. If your project structure doesn't account for this from the start, you'll spend more time recovering from crashes than writing code. This article covers a directory layout, config validation, logging, graceful shutdown, and environment variable handling that hold up in production.

Directory Layout

Start with a layout that separates concerns clearly:

my_pipeline/
├── src/
│   ├── __init__.py
│   ├── pipeline.py       # main orchestration
│   ├── llm_client.py     # API calls
│   ├── parser.py         # output parsing
│   └── storage.py        # file I/O and checkpoints
├── checkpoints/          # progress state (gitignore this)
├── outputs/              # final generated files
│   ├── chapters/
│   └── covers/
├── logs/                 # rotating log files
├── config.py             # Config dataclass
├── main.py               # entrypoint
├── .env                  # secrets (never commit)
├── .env.example          # committed template
├── requirements.txt
└── .gitignore

The checkpoints/ directory is the most important one. It stores progress state so a 3-hour job that crashes at chapter 18 of 25 can resume from chapter 18, not chapter 1.

Config Dataclass with Validation

Avoid scattered os.getenv() calls throughout your code. Centralise all config in one place and validate it at startup — fail fast, fail loudly.

# config.py
import os
from dataclasses import dataclass, field
from pathlib import Path


@dataclass
class PipelineConfig:
    # Paths
    output_dir: Path = field(default_factory=lambda: Path("outputs"))
    checkpoint_dir: Path = field(default_factory=lambda: Path("checkpoints"))
    log_dir: Path = field(default_factory=lambda: Path("logs"))

    # LLM settings
    api_key: str = field(default_factory=lambda: os.environ["ANTHROPIC_API_KEY"])
    model: str = "claude-opus-4-5"
    max_tokens: int = 4096
    temperature: float = 0.7

    # Batch settings
    batch_size: int = 5
    delay_between_calls: float = 2.0

    def __post_init__(self):
        self._validate()
        self._create_directories()

    def _validate(self):
        if not self.api_key:
            raise ValueError("ANTHROPIC_API_KEY is not set")
        if not 0.0 <= self.temperature <= 1.0:
            raise ValueError(f"temperature must be 0.0-1.0, got {self.temperature}")
        if self.max_tokens < 256:
            raise ValueError(f"max_tokens too low: {self.max_tokens}")
        if self.batch_size < 1:
            raise ValueError(f"batch_size must be >= 1, got {self.batch_size}")

    def _create_directories(self):
        for d in [self.output_dir, self.checkpoint_dir, self.log_dir]:
            d.mkdir(parents=True, exist_ok=True)


def load_config(**overrides) -> PipelineConfig:
    """Load config from environment, then apply any overrides."""
    return PipelineConfig(**overrides)

Logging: Rotating File Handler + Console + JSON

Use Python's standard logging module with two handlers: a rotating file handler for persistent logs and a console handler for live feedback. For machine-readable logs (useful if you ever pipe to a log aggregator), add JSON formatting.

# src/logging_setup.py
import json
import logging
import sys
from logging.handlers import RotatingFileHandler
from pathlib import Path


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_obj = {
            "time": self.formatTime(record, self.datefmt),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        if record.exc_info:
            log_obj["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_obj)


def setup_logging(log_dir: Path, level: int = logging.INFO) -> logging.Logger:
    logger = logging.getLogger("pipeline")
    logger.setLevel(level)
    logger.handlers.clear()

    # Rotating file handler: 10 MB per file, keep 5 backups
    log_file = log_dir / "pipeline.log"
    file_handler = RotatingFileHandler(log_file, maxBytes=10 * 1024 * 1024, backupCount=5)
    file_handler.setFormatter(JSONFormatter())
    file_handler.setLevel(logging.DEBUG)

    # Console handler: human-readable
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setFormatter(
        logging.Formatter("%(asctime)s  %(levelname)-8s  %(message)s", datefmt="%H:%M:%S")
    )
    console_handler.setLevel(logging.INFO)

    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    return logger

Graceful Shutdown: Save Checkpoint Before Exiting

Long-running jobs must handle SIGINT (Ctrl+C) and SIGTERM (kill signal from a scheduler or container orchestrator) without losing progress. The trick is a shared flag that the main loop checks each iteration.

# src/shutdown.py
import signal
import logging

log = logging.getLogger("pipeline")

class GracefulShutdown:
    """Set shutdown.requested = True when SIGINT or SIGTERM is received."""

    def __init__(self):
        self.requested = False
        signal.signal(signal.SIGINT,  self._handler)
        signal.signal(signal.SIGTERM, self._handler)

    def _handler(self, signum, frame):
        sig_name = signal.Signals(signum).name
        log.warning(f"Signal {sig_name} received — finishing current item then stopping")
        self.requested = True

Use it in your main loop:

# src/pipeline.py
import json
from pathlib import Path

def run(items: list, config, logger, shutdown) -> None:
    checkpoint_file = config.checkpoint_dir / "progress.json"

    # Load existing checkpoint
    completed = set()
    if checkpoint_file.exists():
        state = json.loads(checkpoint_file.read_text())
        completed = set(state.get("completed", []))
        logger.info(f"Resuming: {len(completed)}/{len(items)} already done")

    try:
        for item in items:
            if shutdown.requested:
                logger.info("Shutdown requested — stopping loop")
                break

            if item["id"] in completed:
                logger.info(f"Skipping {item['id']} (already done)")
                continue

            process_item(item, config, logger)
            completed.add(item["id"])

            # Save checkpoint after every item
            checkpoint_file.write_text(json.dumps({"completed": list(completed)}))

    finally:
        logger.info(f"Checkpoint saved: {len(completed)}/{len(items)} complete")

Environment Variable Loading with python-dotenv

# main.py
from dotenv import load_dotenv

load_dotenv()  # loads .env before anything else imports os.environ

from config import load_config
from src.logging_setup import setup_logging
from src.shutdown import GracefulShutdown
from src.pipeline import run

def main():
    config = load_config()
    logger = setup_logging(config.log_dir)
    shutdown = GracefulShutdown()

    logger.info(f"Pipeline starting — model={config.model}, batch_size={config.batch_size}")

    items = load_work_items()  # your data source
    run(items, config, logger, shutdown)
    logger.info("Pipeline finished")

if __name__ == "__main__":
    main()

The .env.example file should list every required variable with placeholder values:

# .env.example
ANTHROPIC_API_KEY=sk-ant-...

Copy it to .env, fill in real values, and never commit .env. Add it to .gitignore.

Why This Structure Pays Off

When your 4-hour generation job crashes at chapter 22, you lose one in-progress item instead of all 25. When a teammate clones the repo, they get a ValueError at startup telling them exactly which env var is missing. When something goes wrong at 2am, the rotating JSON logs give you a timestamp-accurate audit trail.

This is unglamorous infrastructure work, but it's what separates a script from a pipeline.

Full pipeline + source code: germy5.gumroad.com/l/xhxkzz — $19.99, 30-day refund.

Generating Ebook Covers Programmatically with Python and the Imagen API

German Yamil — Wed, 15 Apr 2026 22:36:06 +0000

Every ebook needs two covers: one for KDP (1600x2560px at 300 DPI) and one for Gumroad (1280x720px). Designing both by hand for every book iteration wastes time. This article walks through a Python pipeline that calls an image generation API, post-processes the result with Pillow, validates dimensions and file size, and saves both variants automatically.

Why Dimensions Matter

KDP rejects covers that don't meet their specs. The required ratio is 1:1.6 (width:height). At 1600x2560px and 300 DPI, the physical print size is roughly 5.3" x 8.5" — standard trade paperback. Gumroad thumbnails display at 1280x720px (16:9), which is entirely different. If you upload your KDP cover to Gumroad, it will be cropped or letterboxed.

File size limits also apply: KDP caps cover images at 50MB (you'll rarely hit this with JPEG), and Gumroad has a 50MB limit for product thumbnails. Always validate before uploading.

Calling an Image Generation API

The pattern below works with any REST-based image generation API (Imagen, Stability AI, DALL-E, etc.). Swap the endpoint and auth header for your provider.

import os
import requests
import base64
from pathlib import Path

API_URL = "https://your-image-api-endpoint/generate"
API_KEY = os.environ["IMAGE_API_KEY"]

def generate_cover_image(prompt: str) -> bytes:
    """Call image generation API and return raw image bytes."""
    payload = {
        "prompt": prompt,
        "width": 1600,
        "height": 2560,
        "num_outputs": 1,
        "output_format": "png",
    }
    headers = {"Authorization": f"Bearer {API_KEY}"}
    response = requests.post(API_URL, json=payload, headers=headers, timeout=120)
    response.raise_for_status()

    data = response.json()
    # Most APIs return base64-encoded image data
    image_b64 = data["images"][0]
    return base64.b64decode(image_b64)

Save the returned bytes to disk before any post-processing. Never process in-memory only — if the next step crashes, you lose the API call.

def save_raw(image_bytes: bytes, path: Path) -> None:
    path.write_bytes(image_bytes)
    print(f"Saved raw image: {path} ({len(image_bytes) / 1024:.1f} KB)")

Pillow Post-Processing

Raw API output is rarely production-ready. You need to resize, set DPI metadata, add a text overlay, and export at the correct quality.

from PIL import Image, ImageDraw, ImageFont
import io

def add_text_overlay(img: Image.Image, title: str, subtitle: str) -> Image.Image:
    """Add title and subtitle text to the bottom third of the cover."""
    draw = ImageDraw.Draw(img)
    width, height = img.size

    # Semi-transparent dark band at the bottom
    overlay = Image.new("RGBA", img.size, (0, 0, 0, 0))
    overlay_draw = ImageDraw.Draw(overlay)
    overlay_draw.rectangle(
        [(0, int(height * 0.65)), (width, height)],
        fill=(0, 0, 0, 180),
    )
    img = Image.alpha_composite(img.convert("RGBA"), overlay)

    draw = ImageDraw.Draw(img)

    # Use a default PIL font — replace with a TTF for production
    try:
        title_font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 80)
        sub_font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 48)
    except OSError:
        title_font = ImageFont.load_default()
        sub_font = ImageFont.load_default()

    # Center title
    title_bbox = draw.textbbox((0, 0), title, font=title_font)
    title_x = (width - (title_bbox[2] - title_bbox[0])) // 2
    draw.text((title_x, int(height * 0.70)), title, fill="white", font=title_font)

    # Center subtitle
    sub_bbox = draw.textbbox((0, 0), subtitle, font=sub_font)
    sub_x = (width - (sub_bbox[2] - sub_bbox[0])) // 2
    draw.text((sub_x, int(height * 0.82)), subtitle, fill="#cccccc", font=sub_font)

    return img.convert("RGB")

Saving Both Variants

From one generated image, produce both the KDP and Gumroad versions:

def save_kdp_cover(img: Image.Image, output_path: Path) -> Path:
    """Resize to 1600x2560 at 300 DPI and save as JPEG."""
    kdp = img.resize((1600, 2560), Image.LANCZOS)
    kdp.save(output_path, format="JPEG", quality=95, dpi=(300, 300))
    print(f"KDP cover saved: {output_path}")
    return output_path

def save_gumroad_cover(img: Image.Image, output_path: Path) -> Path:
    """
    Crop center of the KDP cover to 16:9, then resize to 1280x720.
    The KDP cover is portrait; we take the upper portion for Gumroad.
    """
    width, height = img.size
    target_h = int(width * 720 / 1280)  # height that gives 16:9 at this width
    top = int(height * 0.10)            # start 10% down to avoid empty sky
    crop_box = (0, top, width, top + target_h)
    cropped = img.crop(crop_box)
    gumroad = cropped.resize((1280, 720), Image.LANCZOS)
    gumroad.save(output_path, format="JPEG", quality=90)
    print(f"Gumroad cover saved: {output_path}")
    return output_path

File Validation Before Use

Before shipping covers to KDP or uploading to Gumroad, validate dimensions and file size programmatically:

def validate_cover(path: Path, expected_w: int, expected_h: int, max_mb: float = 50.0) -> bool:
    """Return True if the cover meets dimension and size requirements."""
    errors = []

    size_mb = path.stat().st_size / (1024 * 1024)
    if size_mb > max_mb:
        errors.append(f"File too large: {size_mb:.1f} MB (max {max_mb} MB)")

    with Image.open(path) as img:
        w, h = img.size
        if w != expected_w or h != expected_h:
            errors.append(f"Wrong dimensions: {w}x{h} (expected {expected_w}x{expected_h})")

    if errors:
        for e in errors:
            print(f"VALIDATION ERROR: {e}")
        return False

    print(f"Validation passed: {path.name} ({size_mb:.2f} MB, {expected_w}x{expected_h})")
    return True

# Usage
kdp_path   = Path("outputs/cover_kdp.jpg")
gumrd_path = Path("outputs/cover_gumroad.jpg")

assert validate_cover(kdp_path,   1600, 2560, max_mb=50)
assert validate_cover(gumrd_path, 1280,  720, max_mb=50)

Putting It Together

def run_cover_pipeline(title: str, subtitle: str, prompt: str) -> None:
    out = Path("outputs")
    out.mkdir(exist_ok=True)

    raw_bytes = generate_cover_image(prompt)
    raw_path  = out / "cover_raw.png"
    save_raw(raw_bytes, raw_path)

    with Image.open(raw_path) as base_img:
        final = add_text_overlay(base_img.copy(), title, subtitle)

    kdp_path   = out / "cover_kdp.jpg"
    gumrd_path = out / "cover_gumroad.jpg"

    save_kdp_cover(final, kdp_path)
    save_gumroad_cover(final, gumrd_path)

    validate_cover(kdp_path,   1600, 2560)
    validate_cover(gumrd_path, 1280,  720)

if __name__ == "__main__":
    run_cover_pipeline(
        title="Python Automation",
        subtitle="Build pipelines that ship",
        prompt="Minimalist tech book cover, dark background, glowing circuit patterns, professional",
    )

The entire pipeline runs in under 90 seconds including the API call. Swap the font paths for your system and drop in your preferred image API credentials. The validation step will catch any spec violations before you waste time on a KDP rejection.

Full pipeline + source code: germy5.gumroad.com/l/xhxkzz — $19.99, 30-day refund.

The $20/Month Solo Creator Stack: Building and Selling Technical Ebooks with Python and AI

German Yamil — Wed, 15 Apr 2026 21:43:26 +0000

I have one paid tool in my entire publishing stack. Everything else is free. Here is the honest accounting.

The Full Stack

Tool	Purpose	Cost
Claude Code Pro	Writing, editing, code generation, pipeline automation	$20/month
Gumroad	Direct sales, payments, delivery	Free (10% fee per sale)
KDP (Kindle Direct Publishing)	Amazon distribution	Free (royalty split)
Canva	Cover design	Free tier
NotebookLM	Research, source synthesis	Free
ConvertKit	Email list (up to 1,000 subscribers)	Free tier

The $20/month goes to Claude Code Pro. That is the only subscription. Everything else is either free at the usage levels a solo creator needs, or it is fee-on-revenue — meaning you only pay when you earn.

This matters because fee-on-revenue tools do not create a fixed cost problem. If a book sells nothing in a month, the only cash out is $20.

The Financial Model

Two pricing tiers, two different distribution strategies:

Gumroad — $19.99 list price

Gumroad takes 10%: $2.00
Net per sale: $17.99
Audience: engineers who find you through search, social, or email

KDP — $9.99 list price

Amazon takes ~30% for digital delivery: ~$3.00
Net per sale: ~$6.99
Audience: Amazon shoppers browsing by topic

Break-even

With $20/month in fixed costs, the math is simple:

Gumroad only: 2 sales × $17.99 = $35.98 → covered
KDP only: 3 sales × $6.99 = $20.97 → covered
Mixed: 1 Gumroad + 1 KDP = $17.99 + $6.99 = $24.98 → covered

Two sales per month is the floor. Everything above that is margin.

Python Break-Even Calculator

Rather than doing this in a spreadsheet, here is a script that calculates break-even dynamically for any pricing configuration:

from dataclasses import dataclass
from math import ceil

@dataclass
class Channel:
    name: str
    list_price: float
    platform_fee_pct: float  # as a decimal, e.g. 0.10 for 10%

    @property
    def net_per_sale(self) -> float:
        return self.list_price * (1 - self.platform_fee_pct)


def break_even(
    fixed_monthly_cost: float,
    channels: list[Channel],
    weights: list[float],  # expected fraction of sales per channel, must sum to 1.0
) -> dict:
    """
    Calculate how many total sales per month are needed to cover fixed costs,
    given a mix of channels with different net revenue per sale.

    Example:
        channels = [gumroad, kdp]
        weights  = [0.6, 0.4]   # 60% of sales via Gumroad, 40% via KDP
    """
    assert abs(sum(weights) - 1.0) < 1e-6, "Weights must sum to 1.0"
    assert len(channels) == len(weights)

    # weighted average net revenue per sale across channels
    avg_net = sum(c.net_per_sale * w for c, w in zip(channels, weights))

    raw = fixed_monthly_cost / avg_net
    sales_needed = ceil(raw)

    return {
        "avg_net_per_sale": round(avg_net, 2),
        "raw_break_even": round(raw, 2),
        "sales_needed": sales_needed,
        "breakdown": [
            {
                "channel": c.name,
                "net_per_sale": round(c.net_per_sale, 2),
                "weight": w,
                "sales_share": ceil(sales_needed * w),
            }
            for c, w in zip(channels, weights)
        ],
    }


if __name__ == "__main__":
    gumroad = Channel("Gumroad", list_price=19.99, platform_fee_pct=0.10)
    kdp = Channel("KDP", list_price=9.99, platform_fee_pct=0.30)

    result = break_even(
        fixed_monthly_cost=20.00,
        channels=[gumroad, kdp],
        weights=[0.6, 0.4],
    )

    print(f"Weighted avg net/sale: ${result['avg_net_per_sale']}")
    print(f"Break-even: {result['sales_needed']} sales/month")
    print()
    for row in result["breakdown"]:
        print(
            f"  {row['channel']}: ${row['net_per_sale']}/sale "
            f"({int(row['weight']*100)}% of sales → "
            f"~{row['sales_share']} sales needed)"
        )

Running this with the default values:

Weighted avg net/sale: $13.59
Break-even: 2 sales/month
  Gumroad: $17.99/sale (60% of sales → ~2 sales needed)
  KDP: $6.99/sale (40% of sales → ~1 sale needed)

You can adjust weights to model different scenarios — a book that gets heavy Amazon traffic versus one that sells primarily through direct email.

Time Investment Per Book

I track active hours per project. A technical ebook (8,000–15,000 words, with code examples) takes:

Research and outline: 1–1.5 hours (NotebookLM for source synthesis, Claude for structure)
Writing and code: 2–3 hours (Claude Code handles boilerplate; I review and direct)
Cover design: 30 minutes (Canva template, swap text and colors)
Formatting and upload: 30–45 minutes (Pandoc pipeline, KDP upload, Gumroad listing)

Total active time: 4–6 hours per book.

This is not passive time — it is focused work. But it is a weekend afternoon, not a month-long project.

Revenue Projections

These numbers assume modest, realistic sales — not a viral launch:

Single book

Period	Sales/month	Gumroad net	Annual
Conservative	4	$71.96	$863
Moderate	6	$107.94	$1,295

A single book earning at the conservative end, after costs, puts roughly $827/year in your pocket. Not life-changing, but it is compounding: the book keeps selling while you write the next one.

20-book catalog

Assuming each book averages 4 sales/month across both channels with a 60/40 Gumroad/KDP split:

Per book monthly net: ~$52 (blended across channels)
20 books × $52 = $1,040/month gross
Less $20 fixed cost = $1,020/month net
Annual: ~$12,240

The practical monthly figure, accounting for variance and slower months, lands around $1,079/month. That is a meaningful secondary income on a catalog you built in 80–120 hours of actual work spread across a year.

What Makes This Model Work

The key is treating each book as a durable asset rather than a one-time effort. Technical content in stable domains — Python, APIs, database design, deployment patterns — stays relevant for 2–3 years without significant updates.

The pipeline automation (Pandoc build, QA checks, metadata validation) means the marginal cost of publishing book two through twenty drops to nearly zero in setup time. You are reusing the same tooling, the same distribution accounts, the same email list.

The $20/month is not a cost. It is rent on the infrastructure that makes this possible at solo-creator speed.

The full pipeline — including the translation layer, EPUB build system, QA checks, and Gumroad/KDP upload automation — is documented in a practical guide at https://germy5.gumroad.com/l/xhxkzz for $19.99, with a 30-day refund if it does not deliver value.

Assembling Validated EPUB3 Ebooks with Pandoc and Python — Metadata Guardrails and Build Gates

German Yamil — Wed, 15 Apr 2026 21:37:25 +0000

Pandoc is excellent at producing valid EPUB3 from Markdown. Where it falls short is telling you that your metadata still says "TBD" or that three chapters are not finished yet. That is your job, and the right place to do it is before Pandoc ever runs.

This article covers the guardrail layer I put between raw source files and the final EPUB build.

The Problem with Running Pandoc Too Early

An EPUB with [TBD] in the subtitle field, or with a chapter that ends mid-sentence because a writer placeholder was never filled in, is not a draft — it is a broken product. The fix is not discipline; it is a gate that fails loudly when the source is not ready.

I enforce two gates:

Metadata validation — reject any YAML front matter containing placeholder strings.
Completion gate — read a checkpoint.json file where each chapter has a status, and refuse to build if any chapter is not DONE.

Gate 1: Metadata Validator

import yaml
import sys
from pathlib import Path

PLACEHOLDER_PATTERNS = ["[TBD]", "TODO", "XXX", "FIXME", "PLACEHOLDER"]

def load_metadata(metadata_path: Path) -> dict:
    with open(metadata_path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)

def validate_metadata(metadata: dict, source_path: Path) -> list[str]:
    """
    Walk every string value in the metadata dict and flag placeholders.
    Returns a list of error strings; empty list means clean.
    """
    errors = []

    def check_value(key_path: str, value):
        if isinstance(value, str):
            for pattern in PLACEHOLDER_PATTERNS:
                if pattern in value:
                    errors.append(
                        f"  [{source_path}] '{key_path}' contains '{pattern}': {value!r}"
                    )
        elif isinstance(value, dict):
            for k, v in value.items():
                check_value(f"{key_path}.{k}", v)
        elif isinstance(value, list):
            for i, item in enumerate(value):
                check_value(f"{key_path}[{i}]", item)

    for key, val in metadata.items():
        check_value(key, val)

    return errors

def run_metadata_gate(metadata_path: Path) -> None:
    metadata = load_metadata(metadata_path)
    errors = validate_metadata(metadata, metadata_path)
    if errors:
        print("[FAIL] Metadata contains placeholders:")
        for e in errors:
            print(e)
        sys.exit(1)
    print(f"[OK] Metadata clean: {metadata_path}")

A typical metadata.yaml looks like:

title: "Building Production APIs with FastAPI"
subtitle: "A Practical Guide for Backend Engineers"
author: "Yamil Salinas"
language: en
rights: "© 2026 Yamil Salinas. All rights reserved."
publisher: "Self-published"

If someone adds subtitle: "[TBD]" and forgets to come back to it, the build stops immediately with a clear message.

Gate 2: Chapter Completion Check

The checkpoint.json file is maintained by the writing pipeline — each chapter entry gets updated to DONE when it passes review. The build gate reads this file and refuses to proceed if anything is still in progress.

import json
from pathlib import Path

def load_checkpoint(checkpoint_path: Path) -> dict:
    with open(checkpoint_path, "r", encoding="utf-8") as f:
        return json.load(f)

def run_completion_gate(checkpoint_path: Path) -> None:
    """
    Expects checkpoint.json of the form:
    {
      "chapters": {
        "01_introduction.md": "DONE",
        "02_setup.md": "DONE",
        "03_routing.md": "IN_PROGRESS"
      }
    }
    Exits with code 1 if any chapter is not DONE.
    """
    checkpoint = load_checkpoint(checkpoint_path)
    chapters = checkpoint.get("chapters", {})
    not_done = {k: v for k, v in chapters.items() if v != "DONE"}

    if not_done:
        print("[FAIL] Completion gate: the following chapters are not DONE:")
        for chapter, status in not_done.items():
            print(f"  {chapter}: {status}")
        sys.exit(1)

    print(f"[OK] All {len(chapters)} chapter(s) are DONE.")

This keeps the checkpoint file as the single source of truth for build readiness. Writers update it; the build script reads it.

The Pandoc Build Step

Once both gates pass, Pandoc runs as a subprocess. I always pin the output format to epub3 and pass a CSS file for consistent styling:

import subprocess
from pathlib import Path

def build_epub(
    chapter_files: list[Path],
    metadata_path: Path,
    cover_image: Path,
    css_path: Path,
    output_path: Path,
) -> None:
    """
    Invoke Pandoc to assemble an EPUB3 from the validated chapter files.
    All paths must be absolute to avoid CWD surprises.
    """
    cmd = [
        "pandoc",
        "--from", "markdown+smart",
        "--to", "epub3",
        "--metadata-file", str(metadata_path),
        "--css", str(css_path),
        "--epub-cover-image", str(cover_image),
        "--toc",
        "--toc-depth=2",
        "--output", str(output_path),
    ] + [str(f) for f in chapter_files]

    print(f"[BUILD] Running Pandoc → {output_path.name}")
    result = subprocess.run(cmd, capture_output=True, text=True)

    if result.returncode != 0:
        print("[FAIL] Pandoc error:")
        print(result.stderr)
        sys.exit(1)

    print("[OK] Pandoc build succeeded.")

Passing --metadata-file separately from the chapter files means the YAML never contaminates the Markdown source. Each chapter file is pure prose.

Gate 3: File Size Validation

A successful Pandoc exit code does not mean the EPUB is sane. An EPUB under a certain size threshold almost certainly lost assets — images did not embed, CSS was missing, or chapters were accidentally excluded.

def validate_output_size(output_path: Path, min_size_kb: int = 50) -> None:
    """
    Fail the build if the output EPUB is suspiciously small.
    Adjust min_size_kb based on your typical book size.
    """
    size_kb = output_path.stat().st_size / 1024

    if size_kb < min_size_kb:
        print(
            f"[FAIL] Output file is only {size_kb:.1f} KB "
            f"(expected at least {min_size_kb} KB). "
            f"Check that all chapters and assets were included."
        )
        sys.exit(1)

    print(f"[OK] Output size: {size_kb:.1f} KB")

Putting It Together

def main():
    root = Path("/path/to/your/book")
    chapter_files = sorted((root / "chapters").glob("*.md"))

    run_metadata_gate(root / "metadata.yaml")
    run_completion_gate(root / "checkpoint.json")
    build_epub(
        chapter_files=chapter_files,
        metadata_path=root / "metadata.yaml",
        cover_image=root / "assets" / "cover.png",
        css_path=root / "assets" / "epub.css",
        output_path=root / "dist" / "book.epub",
    )
    validate_output_size(root / "dist" / "book.epub", min_size_kb=100)
    print("[DONE] Build complete.")

if __name__ == "__main__":
    main()

Run this in CI or as a pre-publish step. The script is entirely deterministic — same inputs produce the same EPUB, and broken inputs exit non-zero so no downstream step proceeds.

Why This Matters

Every time I skipped one of these gates during development, something slipped through that I caught only while doing a final read of the assembled EPUB. Adding 15 minutes to write the gate saves hours of post-production cleanup. It also means I can hand off the repository to anyone and the build either works cleanly or tells them exactly why it did not.

If you want to see the full pipeline that wraps these gates — including the translation step, KDP formatting, and distribution automation — it is documented in a practical guide at https://germy5.gumroad.com/l/xhxkzz for $19.99, with a 30-day refund guarantee.

How I Translated a Technical Ebook from Spanish to English with Semantic QA in Python

German Yamil — Wed, 15 Apr 2026 21:37:19 +0000

When I decided to publish a technical ebook in English after writing it in Spanish, my first instinct was to paste chapters into a translation interface and call it done. That lasts about ten minutes before you notice that every code block came back garbled with translated variable names and broken syntax. Here is the pipeline I actually ended up building.

The Core Problem: Prose and Code Are Not the Same

A translation API should never see your code blocks. Variable names, function signatures, and string literals are not natural language — sending them through a translation model at best returns them unchanged, and at worst invents plausible-sounding nonsense.

The solution is a fence detector that splits each chapter into typed segments before anything touches an API.

import re
from dataclasses import dataclass
from typing import List

@dataclass
class Segment:
    content: str
    is_code: bool

def split_segments(text: str) -> List[Segment]:
    """
    Split markdown text into alternating prose and fenced code segments.
    Fence patterns: ```

lang ...

 ``` or ~~~ ... ~~~
    """
    pattern = re.compile(
        r'(```

[\w]*\n[\s\S]*?

```|~~~[\w]*\n[\s\S]*?~~~)',
        re.MULTILINE
    )
    segments: List[Segment] = []
    last_end = 0

    for match in pattern.finditer(text):
        # prose before this code block
        prose = text[last_end:match.start()]
        if prose.strip():
            segments.append(Segment(content=prose, is_code=False))
        # the code block itself
        segments.append(Segment(content=match.group(), is_code=True))
        last_end = match.end()

    # any trailing prose
    remainder = text[last_end:]
    if remainder.strip():
        segments.append(Segment(content=remainder, is_code=False))

    return segments

This gives you a clean list of segments where is_code=True segments pass through untouched.

Calling the Translation API

The actual translation call is straightforward. I am showing a generic pattern here — swap in whatever provider you use:

import httpx
import os

TRANSLATE_URL = "https://api.your-provider.com/translate"
API_KEY = os.environ["TRANSLATION_API_KEY"]

def translate_text(text: str, source_lang: str = "es", target_lang: str = "en") -> str:
    """
    Send a single prose segment to the translation API.
    Returns the translated string.
    """
    response = httpx.post(
        TRANSLATE_URL,
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "text": text,
            "source": source_lang,
            "target": target_lang,
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()["translated_text"]

For chapters longer than a few thousand words, chunk the prose segments further to stay within token limits. I found 1,500-word chunks safe for most providers.

QA: Two Modes Depending on What You Have Available

After translation, you need to verify the output is not garbage. I implemented two modes: a semantic check using embeddings if OPENAI_API_KEY is present, and a word-count ratio fallback if it is not.

import os
import math

def qa_segment(original: str, translated: str) -> dict:
    """
    Returns {"mode": str, "score": float, "passed": bool}.
    Mode is "cosine" when OPENAI_API_KEY is set, "word_ratio" otherwise.
    """
    if os.getenv("OPENAI_API_KEY"):
        return _cosine_qa(original, translated)
    return _word_ratio_qa(original, translated)


def _cosine_qa(original: str, translated: str) -> dict:
    """
    Embed both strings and compute cosine similarity.
    Requires: pip install openai
    Threshold: 0.75 (cross-lingual embeddings stay high for equivalent content).
    """
    from openai import OpenAI
    client = OpenAI()

    def embed(text: str):
        resp = client.embeddings.create(
            model="text-embedding-3-small",
            input=text[:8000],  # truncate to model limit
        )
        return resp.data[0].embedding

    vec_a = embed(original)
    vec_b = embed(translated)

    dot = sum(a * b for a, b in zip(vec_a, vec_b))
    mag_a = math.sqrt(sum(a ** 2 for a in vec_a))
    mag_b = math.sqrt(sum(b ** 2 for b in vec_b))
    similarity = dot / (mag_a * mag_b)

    return {"mode": "cosine", "score": round(similarity, 4), "passed": similarity >= 0.75}


def _word_ratio_qa(original: str, translated: str) -> dict:
    """
    Fallback: compare word counts. A good translation of ES→EN prose
    typically expands by 0–25%. Flag anything outside 0.6–1.5x.
    """
    orig_words = len(original.split())
    trans_words = len(translated.split())
    ratio = trans_words / orig_words if orig_words else 0
    passed = 0.6 <= ratio <= 1.5
    return {"mode": "word_ratio", "score": round(ratio, 4), "passed": passed}

Running in cosine mode gives you semantic confidence. Running in word-ratio mode gives you a cheap structural sanity check. Both are useful; knowing which mode you are in matters for interpreting the output.

Reassembling the Document

After translation and QA, stitch the segments back together in order:

def translate_chapter(text: str) -> str:
    segments = split_segments(text)
    output_parts = []
    failures = []

    for i, seg in enumerate(segments):
        if seg.is_code:
            output_parts.append(seg.content)
            continue

        translated = translate_text(seg.content)
        qa_result = qa_segment(seg.content, translated)

        if not qa_result["passed"]:
            failures.append({
                "segment_index": i,
                "mode": qa_result["mode"],
                "score": qa_result["score"],
                "original_preview": seg.content[:120],
            })

        output_parts.append(translated)

    if failures:
        print(f"[WARN] {len(failures)} segment(s) failed QA:")
        for f in failures:
            print(f"  [{f['mode']}] segment {f['segment_index']} "
                  f"score={f['score']}: {f['original_preview']!r}")

    return "\n".join(output_parts)

Failed segments get logged with enough context to review them manually. The book still assembles — you decide whether to re-translate or accept.

What This Solved

Running this against a 60,000-word technical ebook produced output where every code block came through intact and 94% of prose segments passed QA on the first pass. The remaining 6% were mostly segments with heavy domain jargon that the word-ratio check flagged; a single re-read was enough to confirm they were fine.

The pipeline adds maybe 90 minutes of compute time for a full book, runs unattended, and gives you a concrete QA report rather than a vague "it looks okay."

If you want to see how this fits into a full ebook production pipeline — from raw manuscript through translation, formatting, and distribution — I documented the entire workflow in a guide available at https://germy5.gumroad.com/l/xhxkzz for $19.99, with a 30-day refund if it does not deliver value.

Building Crash-Recovery State Machines for Long-Running LLM Pipelines in Python

German Yamil — Wed, 15 Apr 2026 21:25:14 +0000

LLM pipelines fail. API timeouts, rate limits, unexpected token counts, network blips — any of these can kill a long-running job mid-way through.

If you restart from zero every time, you're wasting money and time. Here's a state machine pattern with atomic checkpoints that lets you resume exactly where you left off.

The Core Problem

Imagine a pipeline that processes 10 chapters:

Chapter 1 → DONE
Chapter 2 → DONE
...
Chapter 7 → [process crashes here]
Chapter 8 → not started
...

Without recovery, you restart from Chapter 1. With a state machine, you restart from Chapter 7.

The State Machine

Four states cover every situation:

from enum import Enum

class ChapterStatus(Enum):
    PENDING = "PENDING"           # Not started
    RUNNING = "RUNNING"           # Currently being processed
    DONE = "DONE"                 # Completed successfully
    NEEDS_REVIEW = "NEEDS_REVIEW" # Failed, requires human intervention

State transitions:

PENDING → RUNNING    (when processing starts)
RUNNING → DONE       (on success)
RUNNING → NEEDS_REVIEW  (on failure)
RUNNING → PENDING    (crash recovery — automatic on restart)

Atomic Checkpoint File

Store state in a JSON file, updated atomically:

import json
import os
import tempfile
from pathlib import Path

CHECKPOINT_FILE = "pipeline_checkpoint.json"

def init_checkpoint(chapter_ids: list[str]) -> None:
    """Create checkpoint file if it doesn't exist."""
    if os.path.exists(CHECKPOINT_FILE):
        return  # Don't overwrite existing progress

    data = {
        "version": 1,
        "chapters": [
            {"id": ch_id, "status": "PENDING", "error": None}
            for ch_id in chapter_ids
        ]
    }
    _write_checkpoint(data)

def _write_checkpoint(data: dict) -> None:
    """Write checkpoint atomically using temp file + rename."""
    tmp_path = CHECKPOINT_FILE + ".tmp"
    with open(tmp_path, "w") as f:
        json.dump(data, f, indent=2)
    os.replace(tmp_path, CHECKPOINT_FILE)  # Atomic on POSIX systems

def read_checkpoint() -> dict:
    with open(CHECKPOINT_FILE) as f:
        return json.load(f)

def update_chapter_status(
    chapter_id: str,
    status: str,
    error: str = None
) -> None:
    data = read_checkpoint()
    chapter = next(c for c in data["chapters"] if c["id"] == chapter_id)
    chapter["status"] = status
    chapter["error"] = error
    _write_checkpoint(data)

Key detail: os.replace() is atomic on POSIX systems. If the process crashes during the write, the old checkpoint file survives intact. You never end up with a corrupted checkpoint.

Crash Recovery

This is the key function — run it at startup before anything else:

def recover_running_orphans() -> int:
    """
    Reset any RUNNING chapters back to PENDING.

    A chapter in RUNNING state at startup means the last process
    crashed while working on it. Reset to PENDING so it retries.

    Returns the number of chapters recovered.
    """
    data = read_checkpoint()
    recovered = 0

    for chapter in data["chapters"]:
        if chapter["status"] == "RUNNING":
            chapter["status"] = "PENDING"
            chapter["error"] = "recovered_from_crash"
            recovered += 1

    if recovered:
        _write_checkpoint(data)
        print(f"⚠️  Recovered {recovered} orphaned chapter(s) from previous crash")

    return recovered

The Main Loop

import time

def run_pipeline(process_fn, max_retries: int = 3) -> None:
    """
    Main pipeline loop with crash recovery and retry logic.
    process_fn(chapter_id) -> (success: bool, error: str | None)
    """
    # 1. Recover from any previous crash
    recover_running_orphans()

    data = read_checkpoint()
    pending = [c for c in data["chapters"] if c["status"] == "PENDING"]

    print(f"📋 {len(pending)} chapters to process")

    for chapter in pending:
        chapter_id = chapter["id"]

        # 2. Mark as RUNNING before we start
        update_chapter_status(chapter_id, "RUNNING")
        print(f"⚙️  Processing {chapter_id}...")

        success = False
        last_error = None

        # 3. Try up to max_retries times
        for attempt in range(max_retries):
            try:
                ok, error = process_fn(chapter_id)
                if ok:
                    success = True
                    break
                else:
                    last_error = error
                    print(f"  Attempt {attempt+1} failed: {error[:100]}")
                    time.sleep(2 ** attempt)  # Exponential backoff
            except Exception as e:
                last_error = str(e)
                print(f"  Attempt {attempt+1} exception: {e}")
                time.sleep(2 ** attempt)

        # 4. Update final status
        if success:
            update_chapter_status(chapter_id, "DONE")
            print(f"  ✅ DONE")
        else:
            update_chapter_status(chapter_id, "NEEDS_REVIEW", error=last_error)
            print(f"  ❌ NEEDS_REVIEW: {last_error}")

    # 5. Summary
    data = read_checkpoint()
    counts = {}
    for c in data["chapters"]:
        counts[c["status"]] = counts.get(c["status"], 0) + 1
    print(f"\n📊 Final: {counts}")

Status Dashboard

Quick view of pipeline state:

def print_dashboard() -> None:
    data = read_checkpoint()

    status_icons = {
        "PENDING": "⏳",
        "RUNNING": "🔄",
        "DONE": "✅",
        "NEEDS_REVIEW": "❌",
    }

    print("\n=== Pipeline Status ===")
    for chapter in data["chapters"]:
        icon = status_icons.get(chapter["status"], "?")
        err = f" — {chapter['error'][:60]}" if chapter.get("error") else ""
        print(f"  {icon} {chapter['id']}: {chapter['status']}{err}")

    done = sum(1 for c in data["chapters"] if c["status"] == "DONE")
    total = len(data["chapters"])
    print(f"\nProgress: {done}/{total} ({done/total*100:.0f}%)\n")

Usage

def my_llm_call(chapter_id: str) -> tuple[bool, str]:
    """Your actual processing logic goes here."""
    # Call your LLM API, validate output, etc.
    # Return (True, None) on success
    # Return (False, "error message") on failure
    try:
        # ... your code here
        return True, None
    except Exception as e:
        return False, str(e)

# Initialize (only creates file if it doesn't exist)
init_checkpoint(["ch01", "ch02", "ch03", "ch04", "ch05"])

# Run — safe to Ctrl+C and restart at any point
run_pipeline(my_llm_call)

Why Not a Database?

For pipelines under ~1000 items, a JSON file is simpler and more portable:

No setup, no connection string, no credentials
git diff checkpoint.json shows you exactly what changed
Easy to manually edit if you need to reset a specific chapter
Works on any machine without installing dependencies

Switch to SQLite when you need concurrent writers or >10k items.

I built this pattern while creating an automated ebook pipeline that generates, validates, and publishes technical books end-to-end for $20/month in API costs.

The full pipeline (10 scripts, complete state machine, EPUB output): germy5.gumroad.com/l/xhxkzz — $19.99.

How to Validate Python Code Examples in Technical Ebooks Using AST and Subprocess

German Yamil — Wed, 15 Apr 2026 21:25:07 +0000

Most technical ebooks have code that doesn't run. The author wrote it, it looked right, but it was never executed. Readers discover this the hard way.

Here's a two-layer validation system that makes it structurally impossible to ship broken code examples.

Why Two Layers?

A single validator isn't enough:

Syntax-only (AST): Fast, but doesn't catch import errors, missing files, or runtime failures
Runtime-only (subprocess): Slow if run on every keystroke, and running code in your main process is dangerous

The solution: run them in order. Syntax first (cheap), then runtime (thorough).

Layer 1: AST Syntax Check

import ast
from typing import tuple

def check_syntax(code: str) -> tuple[bool, str]:
    """
    Returns (is_valid, error_message).
    Pure static analysis — no code is executed.
    """
    try:
        ast.parse(code)
        return True, ""
    except SyntaxError as e:
        return False, f"SyntaxError at line {e.lineno}: {e.msg}"
    except ValueError as e:
        return False, f"ValueError: {e}"

This handles:

Missing colons, mismatched brackets
Invalid escape sequences
f-string syntax errors
Invalid encoding declarations

It does NOT handle: missing imports, undefined variables, runtime logic errors.

Layer 2: Subprocess Isolation

import subprocess
import tempfile
import os
from typing import tuple

def check_runtime(code: str, timeout: int = 10) -> tuple[bool, str]:
    """
    Executes code in an isolated subprocess within a temporary directory.
    Returns (is_valid, error_output).
    """
    with tempfile.TemporaryDirectory() as tmpdir:
        # Write code to a temp file
        script_path = os.path.join(tmpdir, "validate_script.py")
        with open(script_path, "w", encoding="utf-8") as f:
            f.write(code)

        result = subprocess.run(
            ["python3", script_path],
            capture_output=True,
            text=True,
            timeout=timeout,
            cwd=tmpdir,           # Working directory = temp dir
            env={**os.environ},   # Inherit environment
        )

        if result.returncode == 0:
            return True, ""

        # Return first 500 chars of stderr for diagnosis
        return False, result.stderr[:500].strip()

Key design decisions:

tempfile.TemporaryDirectory() — auto-cleaned up when the with block exits
cwd=tmpdir — relative file paths in the script resolve to the temp directory, not your project root
timeout=10 — prevents infinite loops from hanging your pipeline
capture_output=True — stderr gives you the actual Python traceback

Combining Both Layers

from enum import Enum
import json
import os

class ValidationResult(Enum):
    PASS = "PASS"
    SYNTAX_FAIL = "SYNTAX_FAIL"
    RUNTIME_FAIL = "RUNTIME_FAIL"

def validate_code_snippet(
    code: str,
    timeout: int = 10
) -> dict:
    """
    Full validation pipeline for a code snippet.
    Returns structured result dict.
    """
    # Layer 1: syntax
    syntax_ok, syntax_err = check_syntax(code)
    if not syntax_ok:
        return {
            "result": ValidationResult.SYNTAX_FAIL.value,
            "error": syntax_err,
            "layer": 1,
        }

    # Layer 2: runtime
    runtime_ok, runtime_err = check_runtime(code, timeout)
    if not runtime_ok:
        return {
            "result": ValidationResult.RUNTIME_FAIL.value,
            "error": runtime_err,
            "layer": 2,
        }

    return {
        "result": ValidationResult.PASS.value,
        "error": None,
        "layer": None,
    }

Integrating With a Chapter Pipeline

If you're building a content pipeline, hook this into your state machine:

import json

CHECKPOINT_FILE = "checkpoint.json"

def process_chapter(chapter_id: str, code_snippets: list[str]):
    """
    Validates all code snippets in a chapter.
    Sets chapter status to NEEDS_REVIEW if any fail.
    """
    with open(CHECKPOINT_FILE, "r+") as f:
        data = json.load(f)
        chapter = next(c for c in data["chapters"] if c["id"] == chapter_id)

        chapter["status"] = "RUNNING"
        f.seek(0)
        json.dump(data, f, indent=2)
        f.truncate()

    # Validate each snippet
    failed_snippets = []
    for i, snippet in enumerate(code_snippets):
        result = validate_code_snippet(snippet)
        if result["result"] != "PASS":
            failed_snippets.append({
                "snippet_index": i,
                "result": result,
            })

    # Update checkpoint
    with open(CHECKPOINT_FILE, "r+") as f:
        data = json.load(f)
        chapter = next(c for c in data["chapters"] if c["id"] == chapter_id)

        if failed_snippets:
            chapter["status"] = "NEEDS_REVIEW"
            chapter["failed_snippets"] = failed_snippets
        else:
            chapter["status"] = "DONE"

        f.seek(0)
        json.dump(data, f, indent=2)
        f.truncate()

    return len(failed_snippets) == 0

What This Catches

Real failures this system blocked during testing:

Error Type	Layer Caught	Example
Missing colon	1 (AST)	`def foo()`
Undefined variable	2 (subprocess)	`print(undefined_var)`
Import not found	2 (subprocess)	`import nonexistent_pkg`
Infinite loop	2 (timeout)	`while True: pass`
File not found	2 (subprocess)	`open("missing.csv")`
Wrong indentation	1 (AST)	Misaligned blocks

What This Doesn't Catch

Be honest about limitations:

Logic errors — code runs but produces wrong output
Side effects — scripts that write to disk or network
Long-running scripts — adjust timeout accordingly
Scripts with required user input — input() calls will hang

For a technical ebook pipeline, these are acceptable tradeoffs. You want to verify the code examples compile and execute cleanly, not full unit test coverage.

Complete Example

# Test the validator
snippets = [
    # Valid
    """
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

print(fibonacci(10))
""",
    # Syntax error
    """
def broken(
    print("missing closing paren")
""",
    # Runtime error
    """
import this_package_does_not_exist
""",
]

for i, snippet in enumerate(snippets):
    result = validate_code_snippet(snippet.strip())
    status = "✅" if result["result"] == "PASS" else "❌"
    print(f"Snippet {i+1}: {status} {result['result']}")
    if result["error"]:
        print(f"  Error: {result['error'][:100]}")

Output:

Snippet 1: ✅ PASS
Snippet 2: ❌ SYNTAX_FAIL
  Error: SyntaxError at line 2: '(' was never closed
Snippet 3: ❌ RUNTIME_FAIL
  Error: ModuleNotFoundError: No module named 'this_package_does_not_exist'

This validation system is part of a larger pipeline I built for producing technical ebooks end-to-end: writing, validation, translation QA, EPUB assembly, and marketing asset generation — all for $20/month.

If you want the complete pipeline with all scripts: germy5.gumroad.com/l/xhxkzz ($19.99, 30-day refund).

I Built a Python Pipeline That Writes, Validates, Translates, and Publishes Ebooks — $20/Month Total Cost

German Yamil — Sat, 11 Apr 2026 13:34:03 +0000

I spent a few weeks building a Python pipeline that does something I couldn't find anywhere else: it writes a technical ebook, validates every code snippet in it, translates it to English, assembles the EPUB, and generates marketing assets — all for $20/month in API costs.

The book it produced is its own proof of concept. Every chapter in it was processed by the exact pipeline it describes.

Here's how it works.

The Architecture: A State Machine

The core problem with LLM-driven pipelines is crash recovery. If your process dies at chapter 7 of 10, you don't want to restart from zero.

The solution is a state machine with atomic checkpoints:

class ChapterStatus(Enum):
    PENDING = "PENDING"
    RUNNING = "RUNNING"
    DONE = "DONE"
    NEEDS_REVIEW = "NEEDS_REVIEW"

Every chapter lives in checkpoint.json with one of these four states. Before processing, the system calls:

def cp_recover_running_orphans(checkpoint_path: str) -> int:
    """
    Resets any RUNNING chapters back to PENDING.
    Called at startup to recover from crashes.
    """
    with open(checkpoint_path, "r+") as f:
        data = json.load(f)
        recovered = 0
        for chapter in data["chapters"]:
            if chapter["status"] == "RUNNING":
                chapter["status"] = "PENDING"
                recovered += 1
        if recovered:
            f.seek(0)
            json.dump(data, f, indent=2)
            f.truncate()
    return recovered

If the process crashes mid-chapter, the next run resets it to PENDING and retries. No manual intervention. No lost work.

The Validation Suite: Why Ebooks Fail

Most technical ebooks have code that doesn't run. The author wrote it, it looked right, but it was never executed.

The pipeline runs every script through two validation layers before a chapter can advance to DONE:

Layer 1: AST Parse

import ast

def validate_syntax(code: str) -> tuple[bool, str]:
    try:
        ast.parse(code)
        return True, "OK"
    except SyntaxError as e:
        return False, f"SyntaxError at line {e.lineno}: {e.msg}"

This catches syntax errors cheaply, without executing anything.

Layer 2: Runtime Check in Isolation

import subprocess
import tempfile
import os

def validate_runtime(code: str, timeout: int = 10) -> tuple[bool, str]:
    with tempfile.TemporaryDirectory() as tmpdir:
        script_path = os.path.join(tmpdir, "script.py")
        with open(script_path, "w") as f:
            f.write(code)

        result = subprocess.run(
            ["python3", script_path],
            capture_output=True,
            text=True,
            timeout=timeout,
            cwd=tmpdir
        )

        if result.returncode == 0:
            return True, "OK"
        return False, result.stderr[:500]

The script is written to a temp file and executed in an isolated subprocess. If it fails for any reason — import error, runtime exception, timeout — the chapter goes to NEEDS_REVIEW.

Critical security note: Never use exec() on f-string interpolated code in your main process. The isolation is the point.

The Translation Pipeline: Honest QA

The book was written in Spanish. The pipeline translates it to English with a semantic QA layer that doesn't lie about what it's doing.

The key challenge: code blocks can't be translated. Python is Python in any language.

The fence detector splits each chapter into segments:

import re

FENCE_PATTERN = re.compile(r"(```

[\w]*\n.*?

```)", re.DOTALL)

def split_segments(text: str) -> list[dict]:
    parts = FENCE_PATTERN.split(text)
    segments = []
    for part in parts:
        if FENCE_PATTERN.match(part):
            segments.append({"type": "code", "content": part})
        else:
            segments.append({"type": "prose", "content": part})
    return segments

Only prose segments go to the translation API. Code blocks are preserved exactly. After translation, the segments are reassembled in order.

The QA layer is honest about its mode:

def qa_translation(original: str, translated: str) -> dict:
    if os.getenv("OPENAI_API_KEY"):
        score = cosine_similarity_via_embeddings(original, translated)
        mode = "SEMANTIC"
    else:
        # Word ratio: honest fallback
        orig_words = len(original.split())
        trans_words = len(translated.split())
        score = min(orig_words, trans_words) / max(orig_words, trans_words)
        mode = "WORD_RATIO_ONLY"

    return {
        "score": score,
        "mode": mode,
        "pass": score >= 0.75
    }

If you don't have an OpenAI key, it tells you it's using word ratio. No TF-IDF disguised as semantic similarity.

The Build System: Safe EPUB Assembly

The Pandoc build has three guardrails:

1. Merge gate — only chapters with status DONE are included:

def merge_chapters(checkpoint_path: str, output_path: str):
    with open(checkpoint_path) as f:
        data = json.load(f)

    done_chapters = [
        ch for ch in data["chapters"] 
        if ch["status"] == "DONE"
    ]

    if len(done_chapters) != len(data["chapters"]):
        raise ValueError(f"Not all chapters are DONE. Aborting merge.")

    # ... merge and write to output_path

2. YAML guardrail — rejects placeholder metadata:

PLACEHOLDER_PATTERNS = ["[TBD]", "[INSERT", "TODO", "XXX"]

def validate_yaml_metadata(metadata: dict):
    for key, value in metadata.items():
        for pattern in PLACEHOLDER_PATTERNS:
            if pattern in str(value):
                raise ValueError(f"Placeholder found in {key}: {value}")

3. Pandoc invocation — epub3 with injected CSS:

pandoc manuscript.md \
  --epub-metadata=metadata.yaml \
  --epub-cover-image=cover.jpg \
  --css=style.css \
  --to=epub3 \
  -o book.epub

The Financial Model

Fixed cost:  $20/month (Claude Code Pro)
Gumroad:     $19.99 → $17.99 net (Gumroad takes ~10%)
KDP:         $9.99  → $6.99 net  (Amazon takes 30%)

Break-even:  ceil($20 / $17.99) = 2 Gumroad sales/month

Conservative year 1 (3 Gumroad + 5 KDP sales/month):
  Gumroad: 3 × $17.99 × 12 = $647.64
  KDP:     5 × $6.99  × 12 = $419.40
  Costs:   $20 × 12        = $240.00
  Net:                       $827.04 ... wait, let me recalc

Actually at those numbers: $647.64 + $419.40 - $240 = $827.04/year
Conservative catalog model (3 Gumroad/month): $3,790 net/year

The pipeline scales linearly. Book 2 takes the same 4-6 hours of active work as Book 1. At 20 books × 3 Gumroad sales/month = $1,079/month net.

What the Pipeline Produces

10 chapters, ~22,000 words
10 Python scripts — all AST-validated and runtime-tested
EPUB in English and Spanish
Cover generated with Imagen 4
Marketing assets — audio overview, pitch deck, social posts

The complete pipeline, all scripts, and both EPUBs are at:
germy5.gumroad.com/l/xhxkzz — $19.99, 30-day refund.

If you have questions about any specific part of the architecture, ask in the comments.