9. Content-Generator
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