Artikel

9. Content-Generator

Allgemein

Der Content-Generator ist die zentrale Komponente für die Erstellung von Beitragsentwürfen. Er nimmt normalisierte Eingangsdaten, wählt ein Thema, bestimmt Plattform und Beitragstyp, rendert den passenden Prompt, ruft OpenAI auf und speichert das Ergebnis als PostDraft.

Der Generator veröffentlicht nichts.

Er erzeugt nur Entwürfe.

ContentSourceItem
    ↓
Themenauswahl
    ↓
Prompt-Auswahl
    ↓
OpenAI-Aufruf
    ↓
technische Nachprüfung
    ↓
PostDraft speichern

9.1 Generator-Service

Der Generator-Service verbindet mehrere Komponenten:

ContentSourceCollector
PromptBuilder
OpenAiClient
PostTextValidator
JsonStore oder SQLiteStore
IdFactory

Minimales Modell für einen Entwurf:

from dataclasses import dataclass
from datetime import datetime
from typing import Any


@dataclass(frozen=True)
class PostDraft:
    id: str
    source_type: str
    source_id: str
    platform: str
    post_type: str
    status: str
    text: str
    image_path: str | None
    target_url: str | None
    prompt_name: str
    prompt_version: str
    model: str
    quality_result: dict[str, Any] | None
    created_at: datetime
    updated_at: datetime

ID-Erzeugung:

from datetime import datetime
from zoneinfo import ZoneInfo


class IdFactory:
    def create_post_draft_id(self) -> str:
        now = datetime.now(ZoneInfo("Europe/Berlin"))

        return "draft_" + now.strftime("%Y%m%d_%H%M%S_%f")

Generator-Service:

from datetime import datetime
from zoneinfo import ZoneInfo


class ContentGeneratorService:
    def __init__(
        self,
        prompt_builder: PromptBuilder,
        openai_client: OpenAiClient,
        id_factory: IdFactory,
        text_validator: PostTextValidator,
    ) -> None:
        self.prompt_builder = prompt_builder
        self.openai_client = openai_client
        self.id_factory = id_factory
        self.text_validator = text_validator

    def generate_draft(
        self,
        item: ContentSourceItem,
        platform: str,
        post_type: str,
        safety_identifier: str,
        model: str,
    ) -> PostDraft:
        prompt_result = self.prompt_builder.build_for_content_item(
            platform=platform,
            post_type=post_type,
            item=item,
        )

        request = OpenAiTextRequest(
            model=model,
            instructions="Erzeuge Social-Media-Texte für einen kleinen deutschen E-Book-Verlag.",
            input_text=prompt_result.input_text,
            safety_identifier=safety_identifier,
            max_output_tokens=prompt_result.max_output_tokens,
        )

        result = self.openai_client.create_text(request)

        cleaned_text = result.text.strip()

        self.text_validator.validate(
            platform=platform,
            post_type=post_type,
            text=cleaned_text,
        )

        now = datetime.now(ZoneInfo("Europe/Berlin"))

        return PostDraft(
            id=self.id_factory.create_post_draft_id(),
            source_type=item.source_type,
            source_id=item.id,
            platform=platform,
            post_type=post_type,
            status="draft",
            text=cleaned_text,
            image_path=item.image_path,
            target_url=item.url,
            prompt_name=prompt_result.prompt_name,
            prompt_version=prompt_result.prompt_version,
            model=result.model,
            quality_result=None,
            created_at=now,
            updated_at=now,
        )

9.2 Themenauswahl

Die Themenauswahl verhindert, dass der Generator zufällig oder monoton arbeitet.

Eingangsdaten können viele Quellen enthalten:

Bücher
Blogartikel
Zitate
Covermotive
manuelle Themen

Für das MVP reicht eine einfache Auswahl nach Priorität und Plattformhinweis.

Beispielmodell:

from dataclasses import dataclass


@dataclass(frozen=True)
class ContentSourceItem:
    id: str
    source_type: str
    title: str
    body: str
    author: str | None
    book_title: str | None
    url: str | None
    image_path: str | None
    tags: list[str]
    platform_hint: str | None
    post_type_hint: str | None

Themenauswahl:

class TopicSelector:
    def select(
        self,
        items: list[ContentSourceItem],
        platform: str,
        post_type: str | None,
    ) -> ContentSourceItem:
        matching_items: list[ContentSourceItem] = []

        for item in items:
            if item.platform_hint is not None and item.platform_hint != platform:
                continue

            if post_type is not None:
                if item.post_type_hint is not None and item.post_type_hint != post_type:
                    continue

            matching_items.append(item)

        if len(matching_items) == 0:
            raise RuntimeError("No matching content source item found.")

        return matching_items[0]

Das ist bewusst simpel. Später kann erweitert werden um:

Priorität
zuletzt verwendet
Themenmischung
Plattformhistorie
Dublettenprüfung
Kampagnenlogik

Beispiel mit Ausschluss bereits verwendeter Quellen:

class UsedSourceFilter:
    def filter_unused(
        self,
        items: list[ContentSourceItem],
        published_source_keys: set[str],
    ) -> list[ContentSourceItem]:
        result: list[ContentSourceItem] = []

        for item in items:
            source_key = item.source_type + ":" + item.id

            if source_key in published_source_keys:
                continue

            result.append(item)

        return result

9.3 Beitragsvarianten erzeugen

Für ein Thema können mehrere Varianten erzeugt werden.

Beispiele:

sachliche Variante
kürzere Variante
technischere Variante
literarischere Variante
plattformneutrale Variante

Für das MVP reichen zwei oder drei Varianten.

Erweiterung des Prompts über Variantennamen:

from dataclasses import dataclass


@dataclass(frozen=True)
class PostVariantRequest:
    item: ContentSourceItem
    platform: str
    post_type: str
    variant_name: str

Generator-Methode:

class VariantContentGeneratorService:
    def __init__(
        self,
        content_generator_service: ContentGeneratorService,
    ) -> None:
        self.content_generator_service = content_generator_service

    def generate_variants(
        self,
        item: ContentSourceItem,
        platform: str,
        post_type: str,
        variant_names: list[str],
        safety_identifier: str,
        model: str,
    ) -> list[PostDraft]:
        drafts: list[PostDraft] = []

        for variant_name in variant_names:
            variant_item = ContentSourceItem(
                id=item.id + "_" + variant_name,
                source_type=item.source_type,
                title=item.title + " / Variante: " + variant_name,
                body=item.body,
                author=item.author,
                book_title=item.book_title,
                url=item.url,
                image_path=item.image_path,
                tags=item.tags,
                platform_hint=item.platform_hint,
                post_type_hint=item.post_type_hint,
            )

            draft = self.content_generator_service.generate_draft(
                item=variant_item,
                platform=platform,
                post_type=post_type,
                safety_identifier=safety_identifier,
                model=model,
            )

            drafts.append(draft)

        return drafts

Sinnvolle Varianten:

standard
kurz
fachlich
literarisch
werkstatt

Beispiel:

variant_names = [
    "standard",
    "kurz",
    "fachlich",
]

Für produktive Nutzung sollte nicht jede Variante später veröffentlicht werden. Varianten sind Material für die Review-Queue.

9.4 Plattformvarianten erzeugen

Ein Inhalt kann für mehrere Plattformen angepasst werden.

Beispiel:

LinkedIn: fachlicher Gedanke
Instagram: bildnaher Text
Facebook: verständlicher Buchhinweis

Modell:

from dataclasses import dataclass


@dataclass(frozen=True)
class PlatformPostPlan:
    platform: str
    post_type: str

Plattformvarianten-Generator:

class PlatformVariantGenerator:
    def __init__(
        self,
        content_generator_service: ContentGeneratorService,
    ) -> None:
        self.content_generator_service = content_generator_service

    def generate_for_platforms(
        self,
        item: ContentSourceItem,
        plans: list[PlatformPostPlan],
        safety_identifier: str,
        model: str,
    ) -> list[PostDraft]:
        drafts: list[PostDraft] = []

        for plan in plans:
            draft = self.content_generator_service.generate_draft(
                item=item,
                platform=plan.platform,
                post_type=plan.post_type,
                safety_identifier=safety_identifier,
                model=model,
            )

            drafts.append(draft)

        return drafts

Beispiel:

plans = [
    PlatformPostPlan(
        platform="linkedin",
        post_type="workshop",
    ),
    PlatformPostPlan(
        platform="facebook",
        post_type="book_intro",
    ),
    PlatformPostPlan(
        platform="instagram",
        post_type="quote",
    ),
]

Wichtig: Plattformvarianten dürfen nicht nur gekürzte Kopien sein. Jede Plattform braucht einen eigenen Fokus.

LinkedIn: Warum ist das fachlich interessant?
Instagram: Was sieht man, welche Stimmung trägt das Motiv?
Facebook: Was ist für Leser konkret relevant?

9.5 Hashtags erzeugen

Hashtags sollten begrenzt und regelbasiert erzeugt werden. Sie sind kein Ersatz für Inhalt.

Empfohlene Limits:

LinkedIn: 0 bis 3
Instagram: 3 bis 5
Facebook: 0 bis 2

Einfache Hashtag-Generierung aus Tags:

class HashtagGenerator:
    def generate(self, platform: str, tags: list[str]) -> list[str]:
        limit = self._get_limit(platform)

        hashtags: list[str] = []

        for tag in tags:
            hashtag = self._normalize_tag(tag)

            if hashtag == "":
                continue

            if hashtag in hashtags:
                continue

            hashtags.append(hashtag)

            if len(hashtags) >= limit:
                break

        return hashtags

    def _get_limit(self, platform: str) -> int:
        if platform == "linkedin":
            return 3

        if platform == "instagram":
            return 5

        if platform == "facebook":
            return 2

        raise RuntimeError(f"Unsupported platform for hashtags: {platform}")

    def _normalize_tag(self, tag: str) -> str:
        cleaned_tag = tag.strip().lower()

        if cleaned_tag == "":
            return ""

        replacements: dict[str, str] = {
            "ä": "ae",
            "ö": "oe",
            "ü": "ue",
            "ß": "ss",
            "-": "",
            " ": "",
        }

        for search, replacement in replacements.items():
            cleaned_tag = cleaned_tag.replace(search, replacement)

        allowed_characters = "abcdefghijklmnopqrstuvwxyz0123456789"
        normalized = ""

        for character in cleaned_tag:
            if character in allowed_characters:
                normalized = normalized + character

        if normalized == "":
            return ""

        return "#" + normalized

Beispiel:

generator = HashtagGenerator()

hashtags = generator.generate(
    platform="instagram",
    tags=[
        "klassiker",
        "Jane Austen",
        "gesellschaftsroman",
        "E-Book",
    ],
)

print(hashtags)

Mögliche Ausgabe:

['#klassiker', '#janeausten', '#gesellschaftsroman', '#ebook']

Hashtags an Text anhängen:

class HashtagAppender:
    def append(self, text: str, hashtags: list[str]) -> str:
        if len(hashtags) == 0:
            return text

        hashtag_line = " ".join(hashtags)

        return text.rstrip() + "\n\n" + hashtag_line

Bei LinkedIn kann man Hashtags auch komplett weglassen. Für einen sachlichen Verlagsauftritt ist das oft besser.

9.6 Call-to-Action begrenzen

Call-to-Action bedeutet nicht automatisch Verkaufsdruck.

Ungeeignet:

Jetzt kaufen
Unbedingt lesen
Sichern Sie sich Ihr Exemplar
Hier klicken
Nur heute

Geeignet:

Die Ausgabe ist im Shop verfügbar.
Mehr dazu im Blog.
Weitere Titel finden sich im Verlagsprogramm.

Ein CTA sollte selten und dezent sein.

Beispielklasse:

class CallToActionPolicy:
    def __init__(self) -> None:
        self.blocked_fragments = [
            "jetzt kaufen",
            "unbedingt lesen",
            "hier klicken",
            "nur heute",
            "sichern sie sich",
            "greifen sie zu",
            "ein muss",
        ]

    def validate(self, text: str) -> None:
        lowered_text = text.lower()

        for blocked_fragment in self.blocked_fragments:
            if blocked_fragment in lowered_text:
                raise RuntimeError(f"Blocked call-to-action found: {blocked_fragment}")

CTA-Ergänzung nach Regel:

class CallToActionAppender:
    def append_if_allowed(
        self,
        text: str,
        platform: str,
        target_url: str | None,
        include_cta: bool,
    ) -> str:
        if include_cta is False:
            return text

        if target_url is None:
            return text

        if platform == "instagram":
            return text

        cta = "Die Ausgabe ist im Shop verfügbar: " + target_url

        return text.rstrip() + "\n\n" + cta

Für Instagram ist ein Link in der Caption meist weniger sinnvoll. Dort kann der CTA eher entfallen oder auf Profil/Shop allgemein verweisen, wenn das zur eigenen Strategie passt.

Für das MVP:

CTA standardmäßig deaktiviert
CTA nur bei Buchvorstellung
CTA nie bei Kommentaren
CTA nie bei fremden Beiträgen

9.7 Entwürfe speichern

Entwürfe werden nicht überschrieben, sondern angehängt.

JSON-Speicher:

import json
from dataclasses import asdict
from datetime import datetime
from pathlib import Path
from typing import Any


class PostDraftJsonStore:
    def __init__(self, file_path: Path) -> None:
        self.file_path = file_path

    def append(self, draft: PostDraft) -> None:
        existing_drafts = self._load()

        payload = self._to_payload(draft)
        existing_drafts.append(payload)

        self.file_path.parent.mkdir(parents=True, exist_ok=True)

        with self.file_path.open("w", encoding="utf-8") as file:
            json.dump(existing_drafts, file, ensure_ascii=False, indent=2)
            file.write("\n")

    def append_many(self, drafts: list[PostDraft]) -> None:
        existing_drafts = self._load()

        for draft in drafts:
            existing_drafts.append(self._to_payload(draft))

        self.file_path.parent.mkdir(parents=True, exist_ok=True)

        with self.file_path.open("w", encoding="utf-8") as file:
            json.dump(existing_drafts, file, ensure_ascii=False, indent=2)
            file.write("\n")

    def _load(self) -> list[dict[str, Any]]:
        if self.file_path.exists() is False:
            return []

        with self.file_path.open("r", encoding="utf-8") as file:
            payload = json.load(file)

        if isinstance(payload, list) is False:
            raise RuntimeError("Post draft JSON must contain a list.")

        drafts: list[dict[str, Any]] = []

        for item in payload:
            if isinstance(item, dict) is False:
                raise RuntimeError("Post draft JSON contains non-object item.")

            drafts.append(item)

        return drafts

    def _to_payload(self, draft: PostDraft) -> dict[str, Any]:
        return {
            "id": draft.id,
            "source_type": draft.source_type,
            "source_id": draft.source_id,
            "platform": draft.platform,
            "post_type": draft.post_type,
            "status": draft.status,
            "text": draft.text,
            "image_path": draft.image_path,
            "target_url": draft.target_url,
            "prompt_name": draft.prompt_name,
            "prompt_version": draft.prompt_version,
            "model": draft.model,
            "quality_result": draft.quality_result,
            "created_at": draft.created_at.isoformat(),
            "updated_at": draft.updated_at.isoformat(),
        }

Beispieldatei:

data/post_drafts.json

Beispielinhalt:

[
  {
    "id": "draft_20260512_093000_123456",
    "source_type": "book",
    "source_id": "book_001",
    "platform": "linkedin",
    "post_type": "book_intro",
    "status": "draft",
    "text": "Jane Austens Stolz und Vorurteil ist weniger eine romantische Komödie als eine genaue Beobachtung gesellschaftlicher Erwartungen.",
    "image_path": "storage/images/stolz-und-vorurteil.jpg",
    "target_url": "https://example.com/stolz-und-vorurteil",
    "prompt_name": "linkedin_post",
    "prompt_version": "1.0.0",
    "model": "gpt-4.1-mini",
    "quality_result": null,
    "created_at": "2026-05-12T09:30:00+02:00",
    "updated_at": "2026-05-12T09:30:00+02:00"
  }
]

9.8 Beispiel: generate_posts.py

Die Datei erzeugt aus den Content-Quellen einen Entwurf und speichert ihn in data/post_drafts.json.

Datei:

src/generate_posts.py

from pathlib import Path

from src.config import ConfigLoader
from src.content_sources import ContentSourceCollector
from src.generator import ContentGeneratorService
from src.generator import IdFactory
from src.openai_client import OpenAiClient
from src.openai_client import SafetyIdentifierFactory
from src.post_draft_store import PostDraftJsonStore
from src.prompt_library import PromptBuilder
from src.prompt_library import PromptSelector
from src.prompt_library import PromptTemplateRenderer
from src.text_validation import PostTextValidator
from src.topic_selector import TopicSelector


def main() -> None:
    root_dir = Path(__file__).resolve().parent.parent

    config_loader = ConfigLoader(root_dir)
    config = config_loader.load()

    platform = "linkedin"
    post_type = "workshop"

    collector = ContentSourceCollector(root_dir)
    items = collector.collect()

    topic_selector = TopicSelector()
    item = topic_selector.select(
        items=items,
        platform=platform,
        post_type=post_type,
    )

    renderer = PromptTemplateRenderer(root_dir / "prompts")
    selector = PromptSelector()
    prompt_builder = PromptBuilder(
        renderer=renderer,
        selector=selector,
    )

    openai_client = OpenAiClient()
    id_factory = IdFactory()
    text_validator = PostTextValidator()

    safety_identifier_factory = SafetyIdentifierFactory()
    safety_identifier = safety_identifier_factory.create("publisher-system:null-papier")

    generator = ContentGeneratorService(
        prompt_builder=prompt_builder,
        openai_client=openai_client,
        id_factory=id_factory,
        text_validator=text_validator,
    )

    draft = generator.generate_draft(
        item=item,
        platform=platform,
        post_type=post_type,
        safety_identifier=safety_identifier,
        model=config.raw["openai"]["default_model"],
    )

    store = PostDraftJsonStore(root_dir / "data" / "post_drafts.json")
    store.append(draft)

    print("DRAFT_CREATED")
    print(draft.id)
    print(draft.text)


if __name__ == "__main__":
    main()

Dazu passende Minimaldateien:

data/topics.json

[
  {
    "id": "topic_001",
    "platform": "linkedin",
    "post_type": "workshop",
    "title": "Warum gemeinfreie Texte digitale Nacharbeit brauchen",
    "briefing": "Erkläre sachlich, warum gemeinfreie Texte nicht automatisch gute E-Books ergeben. Themen: OCR-Fehler, Typografie, Metadaten und EPUB3.",
    "tags": [
      "verlagswerkstatt",
      "ebook-produktion",
      "public-domain"
    ],
    "priority": 80,
    "status": "active"
  }
]

Collector-Ergänzung für manuelle Themen:

class ContentSourceCollector:
    def __init__(self, root_dir: Path) -> None:
        self.root_dir = root_dir

    def collect(self) -> list[ContentSourceItem]:
        items: list[ContentSourceItem] = []

        topic_source = JsonManualTopicSource(self.root_dir / "data" / "topics.json")
        topic_normalizer = ManualTopicNormalizer()

        for topic in topic_source.load():
            if topic.status != "active":
                continue

            items.append(topic_normalizer.normalize(topic))

        return items

Textvalidator:

class PostTextValidator:
    def validate(self, platform: str, post_type: str, text: str) -> None:
        cleaned_text = text.strip()

        if cleaned_text == "":
            raise RuntimeError("Post text must not be empty.")

        max_length = self._get_max_length(platform)

        if len(cleaned_text) > max_length:
            raise RuntimeError(f"Post text exceeds limit for platform: {platform}")

        blocked_fragments = [
            "jetzt kaufen",
            "unbedingt lesen",
            "tauchen sie ein",
            "meisterwerk",
            "ein muss",
            "lassen sie sich verzaubern",
        ]

        lowered_text = cleaned_text.lower()

        for blocked_fragment in blocked_fragments:
            if blocked_fragment in lowered_text:
                raise RuntimeError(f"Post text contains blocked fragment: {blocked_fragment}")

    def _get_max_length(self, platform: str) -> int:
        if platform == "linkedin":
            return 900

        if platform == "instagram":
            return 700

        if platform == "facebook":
            return 1200

        raise RuntimeError(f"Unsupported platform: {platform}")

Ausführen:

python -m src.generate_posts

Erwartete Ausgabe:

DRAFT_CREATED
draft_20260512_093000_123456
Gemeinfreie Texte sind frei verfügbar, aber nicht automatisch verlagsfertig...

Ergebnis dieses Kapitels

Der Content-Generator kann jetzt:

normalisierte Content-Quellen verarbeiten
ein passendes Thema auswählen
Prompt und Plattform bestimmen
OpenAI zur Texterzeugung nutzen
mehrere Varianten erzeugen
plattformbezogene Fassungen erstellen
Hashtags regelbasiert ergänzen
Call-to-Action begrenzen
Entwürfe als PostDraft speichern

Damit ist die Rohproduktion der Social-Media-Beiträge vorbereitet. Der nächste Schritt ist die Qualitätsprüfung vor der Freigabe.

0 Kommentare