Artikel
11. Review-Queue
Die Review-Queue ist die Kontrollschicht zwischen Texterzeugung und Veröffentlichung.
Der Content-Generator erzeugt Entwürfe. Das Quality-Gate prüft Mindestkriterien. Die Review-Queue entscheidet, was mit einem Entwurf geschieht:
draft → approved
draft → rejected
approved → published
approved → failed
Für das MVP reicht eine JSON-Datei. Später kann dieselbe Logik auf SQLite umgestellt werden.
11.1 Zweck der Queue
Die Review-Queue erfüllt vier Aufgaben:
Entwürfe sichtbar machen
Entwürfe freigeben
Entwürfe ablehnen
Statuswechsel nachvollziehbar speichern
Ohne Queue würde der Ablauf zu direkt:
OpenAI erzeugt Text → Playwright veröffentlicht
Das ist technisch einfach, aber riskant.
Mit Queue:
OpenAI erzeugt Text
Quality-Gate prüft Text
Entwurf landet in Queue
Freigabe erfolgt manuell oder regelbasiert
Publisher verarbeitet nur approved
Dadurch bleiben die Schritte getrennt.
11.2 Status: draft, approved, rejected, published, failed
Für das MVP reichen fünf Statuswerte.
from enum import Enum
class ReviewStatus(str, Enum):
DRAFT = "draft"
APPROVED = "approved"
REJECTED = "rejected"
PUBLISHED = "published"
FAILED = "failed"
Bedeutung:
draft
Der Beitrag wurde erzeugt, ist aber noch nicht freigegeben.
approved
Der Beitrag darf vom Publisher veröffentlicht werden.
rejected
Der Beitrag wurde verworfen.
published
Der Beitrag wurde veröffentlicht.
failed
Der Veröffentlichungsversuch ist fehlgeschlagen.
Optional später:
quality_failed
scheduled
publishing
blocked
Für das schnelle Handbuch bleibt es zunächst bei fünf Zuständen.
11.3 CLI zur Freigabe
Die CLI muss einfache Aktionen unterstützen:
python -m src.review_queue list
python -m src.review_queue show draft_20260512_001
python -m src.review_queue approve draft_20260512_001
python -m src.review_queue reject draft_20260512_001
python -m src.review_queue edit draft_20260512_001
Für das MVP reicht argparse.
Beispielstruktur:
src/review_queue.py
Die CLI arbeitet auf:
data/post_drafts.json
Beispielobjekt:
{
"id": "draft_20260512_001",
"source_type": "manual_topic",
"source_id": "topic_001",
"platform": "linkedin",
"post_type": "workshop",
"status": "draft",
"text": "Gemeinfreie Texte sind frei verfügbar, aber nicht automatisch verlagsfertig.",
"image_path": null,
"target_url": "https://www.null-papier.de",
"prompt_name": "workshop_note",
"prompt_version": "1.0.0",
"model": "gpt-4.1-mini",
"quality_result": null,
"created_at": "2026-05-12T09:00:00+02:00",
"updated_at": "2026-05-12T09:00:00+02:00"
}
11.4 Vorschau im Terminal
Die Vorschau muss knapp sein.
Für mehrere Entwürfe:
ID PLATFORM TYPE STATUS
draft_20260512_001 linkedin workshop draft
draft_20260512_002 facebook book_intro draft
Für einen einzelnen Entwurf:
ID: draft_20260512_001
Platform: linkedin
Type: workshop
Status: draft
Source: manual_topic:topic_001
Prompt: workshop_note 1.0.0
Model: gpt-4.1-mini
URL: https://www.null-papier.de
Text:
Gemeinfreie Texte sind frei verfügbar, aber nicht automatisch verlagsfertig.
OCR-Korrektur, Typografie und Metadaten bleiben echte Arbeit.
Terminal-Ausgabe:
class ReviewQueuePrinter:
def print_list(self, drafts: list[dict[str, object]]) -> None:
print("ID PLATFORM TYPE STATUS")
print("-------------------------------------------------------------")
for draft in drafts:
print(
str(draft["id"]).ljust(26)
+ " "
+ str(draft["platform"]).ljust(11)
+ " "
+ str(draft["post_type"]).ljust(13)
+ " "
+ str(draft["status"])
)
def print_detail(self, draft: dict[str, object]) -> None:
print("ID: " + str(draft["id"]))
print("Platform: " + str(draft["platform"]))
print("Type: " + str(draft["post_type"]))
print("Status: " + str(draft["status"]))
print("Source: " + str(draft["source_type"]) + ":" + str(draft["source_id"]))
print("Prompt: " + str(draft["prompt_name"]) + " " + str(draft["prompt_version"]))
print("Model: " + str(draft["model"]))
target_url = draft.get("target_url")
if target_url is not None:
print("URL: " + str(target_url))
print("")
print("Text:")
print(str(draft["text"]))
11.5 Bearbeiten eines Entwurfs
Für das MVP reicht Bearbeitung über temporäre Datei im Standardeditor.
Ablauf:
Entwurf laden
Text in temporäre Datei schreiben
Editor öffnen
bearbeiteten Text lesen
Text ersetzen
updated_at setzen
Status bleibt draft
Editor aus Umgebung:
EDITOR
Fallback:
nano
Python:
import os
import subprocess
import tempfile
from datetime import datetime
from zoneinfo import ZoneInfo
class DraftEditor:
def edit_text(self, current_text: str) -> str:
editor = os.environ.get("EDITOR")
if editor is None:
editor = "nano"
with tempfile.NamedTemporaryFile(
mode="w+",
suffix=".txt",
encoding="utf-8",
delete=False,
) as temporary_file:
temporary_file.write(current_text)
temporary_file_path = temporary_file.name
subprocess.run(
[
editor,
temporary_file_path,
],
check=True,
)
with open(temporary_file_path, "r", encoding="utf-8") as file:
edited_text = file.read()
cleaned_text = edited_text.strip()
if cleaned_text == "":
raise RuntimeError("Edited draft text must not be empty.")
return cleaned_text
Nach dem Bearbeiten sollte das Quality-Gate erneut laufen. Für das MVP kann zunächst nur der Text ersetzt werden; sauberer ist:
edit → quality check → draft bleibt draft
11.6 Automatische Freigabe nach Regeln
Automatische Freigabe ist nur für harmlose, standardisierte Inhalte sinnvoll.
Geeignet:
Klassiker-Zitat
Buch des Tages
Blogartikel-Zusammenfassung
Verlagswerkstatt nach festem Prompt
Nicht geeignet:
Kommentare auf fremde Beiträge
Antworten auf Beschwerden
politische oder rechtliche Themen
ungeprüfte neue Formate
Regeln:
Quality-Gate bestanden
Status ist draft
Plattform ist erlaubt
Post-Typ ist erlaubt
Text enthält keinen Link bei Kommentar
Text überschreitet kein Limit
Prompt-Version ist freigegeben
Beispiel:
class AutoApprovalPolicy:
def __init__(self) -> None:
self.allowed_platforms = [
"linkedin",
"facebook",
]
self.allowed_post_types = [
"workshop",
"book_intro",
"blog_summary",
"quote",
]
self.allowed_prompt_versions = [
"1.0.0",
]
def can_auto_approve(self, draft: dict[str, object]) -> bool:
status = draft.get("status")
if status != "draft":
return False
platform = draft.get("platform")
if platform not in self.allowed_platforms:
return False
post_type = draft.get("post_type")
if post_type not in self.allowed_post_types:
return False
prompt_version = draft.get("prompt_version")
if prompt_version not in self.allowed_prompt_versions:
return False
quality_result = draft.get("quality_result")
if isinstance(quality_result, dict) is True:
passed = quality_result.get("passed")
if passed is not True:
return False
return True
Automatische Freigabe sollte protokolliert werden:
{
"event": "draft_auto_approved",
"draft_id": "draft_20260512_001",
"platform": "linkedin",
"post_type": "workshop"
}
11.7 Beispiel: review_queue.py
Datei:
src/review_queue.py
import argparse
import json
import os
import subprocess
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo
class ReviewQueueError(Exception):
pass
class ReviewStatus:
DRAFT = "draft"
APPROVED = "approved"
REJECTED = "rejected"
PUBLISHED = "published"
FAILED = "failed"
class ReviewQueueStore:
def __init__(self, file_path: Path) -> None:
self.file_path = file_path
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 ReviewQueueError("Review queue JSON must contain a list.")
drafts: list[dict[str, Any]] = []
for item in payload:
if isinstance(item, dict) is False:
raise ReviewQueueError("Review queue contains non-object item.")
drafts.append(item)
return drafts
def save(self, drafts: list[dict[str, Any]]) -> None:
self.file_path.parent.mkdir(parents=True, exist_ok=True)
with self.file_path.open("w", encoding="utf-8") as file:
json.dump(drafts, file, ensure_ascii=False, indent=2)
file.write("\n")
class ReviewQueueRepository:
def __init__(self, store: ReviewQueueStore) -> None:
self.store = store
def list_all(self) -> list[dict[str, Any]]:
return self.store.load()
def list_by_status(self, status: str) -> list[dict[str, Any]]:
drafts = self.store.load()
result: list[dict[str, Any]] = []
for draft in drafts:
if draft.get("status") == status:
result.append(draft)
return result
def find_by_id(self, draft_id: str) -> dict[str, Any]:
drafts = self.store.load()
for draft in drafts:
if draft.get("id") == draft_id:
return draft
raise ReviewQueueError("Draft not found: " + draft_id)
def update(self, updated_draft: dict[str, Any]) -> None:
drafts = self.store.load()
updated_drafts: list[dict[str, Any]] = []
found = False
updated_id = updated_draft.get("id")
for draft in drafts:
if draft.get("id") == updated_id:
updated_drafts.append(updated_draft)
found = True
else:
updated_drafts.append(draft)
if found is False:
raise ReviewQueueError("Draft not found: " + str(updated_id))
self.store.save(updated_drafts)
class ReviewQueuePrinter:
def print_list(self, drafts: list[dict[str, Any]]) -> None:
print("ID PLATFORM TYPE STATUS")
print("-------------------------------------------------------------")
for draft in drafts:
draft_id = str(draft.get("id", ""))
platform = str(draft.get("platform", ""))
post_type = str(draft.get("post_type", ""))
status = str(draft.get("status", ""))
print(
draft_id.ljust(26)
+ " "
+ platform.ljust(11)
+ " "
+ post_type.ljust(13)
+ " "
+ status
)
def print_detail(self, draft: dict[str, Any]) -> None:
print("ID: " + str(draft.get("id")))
print("Platform: " + str(draft.get("platform")))
print("Type: " + str(draft.get("post_type")))
print("Status: " + str(draft.get("status")))
print("Source: " + str(draft.get("source_type")) + ":" + str(draft.get("source_id")))
print("Prompt: " + str(draft.get("prompt_name")) + " " + str(draft.get("prompt_version")))
print("Model: " + str(draft.get("model")))
target_url = draft.get("target_url")
if target_url is not None:
print("URL: " + str(target_url))
image_path = draft.get("image_path")
if image_path is not None:
print("Image: " + str(image_path))
print("")
print("Text:")
print(str(draft.get("text")))
class DraftEditor:
def edit_text(self, current_text: str) -> str:
editor = os.environ.get("EDITOR")
if editor is None:
editor = "nano"
with tempfile.NamedTemporaryFile(
mode="w+",
suffix=".txt",
encoding="utf-8",
delete=False,
) as temporary_file:
temporary_file.write(current_text)
temporary_file_path = temporary_file.name
subprocess.run(
[
editor,
temporary_file_path,
],
check=True,
)
with open(temporary_file_path, "r", encoding="utf-8") as file:
edited_text = file.read()
cleaned_text = edited_text.strip()
if cleaned_text == "":
raise ReviewQueueError("Edited draft text must not be empty.")
return cleaned_text
class AutoApprovalPolicy:
def __init__(self) -> None:
self.allowed_platforms = [
"linkedin",
"facebook",
]
self.allowed_post_types = [
"workshop",
"book_intro",
"blog_summary",
"quote",
]
self.allowed_prompt_versions = [
"1.0.0",
]
def can_auto_approve(self, draft: dict[str, Any]) -> bool:
if draft.get("status") != ReviewStatus.DRAFT:
return False
platform = draft.get("platform")
if platform not in self.allowed_platforms:
return False
post_type = draft.get("post_type")
if post_type not in self.allowed_post_types:
return False
prompt_version = draft.get("prompt_version")
if prompt_version not in self.allowed_prompt_versions:
return False
quality_result = draft.get("quality_result")
if isinstance(quality_result, dict) is True:
passed = quality_result.get("passed")
if passed is not True:
return False
return True
class ReviewQueueService:
def __init__(
self,
repository: ReviewQueueRepository,
printer: ReviewQueuePrinter,
editor: DraftEditor,
auto_approval_policy: AutoApprovalPolicy,
) -> None:
self.repository = repository
self.printer = printer
self.editor = editor
self.auto_approval_policy = auto_approval_policy
def list_drafts(self, status: str | None) -> None:
if status is None:
drafts = self.repository.list_all()
else:
drafts = self.repository.list_by_status(status)
self.printer.print_list(drafts)
def show(self, draft_id: str) -> None:
draft = self.repository.find_by_id(draft_id)
self.printer.print_detail(draft)
def approve(self, draft_id: str) -> None:
draft = self.repository.find_by_id(draft_id)
self._assert_status(draft, ReviewStatus.DRAFT)
draft["status"] = ReviewStatus.APPROVED
draft["approved_at"] = self._now()
draft["updated_at"] = self._now()
self.repository.update(draft)
print("APPROVED")
print(draft_id)
def reject(self, draft_id: str) -> None:
draft = self.repository.find_by_id(draft_id)
self._assert_status(draft, ReviewStatus.DRAFT)
draft["status"] = ReviewStatus.REJECTED
draft["rejected_at"] = self._now()
draft["updated_at"] = self._now()
self.repository.update(draft)
print("REJECTED")
print(draft_id)
def edit(self, draft_id: str) -> None:
draft = self.repository.find_by_id(draft_id)
self._assert_status(draft, ReviewStatus.DRAFT)
current_text = draft.get("text")
if isinstance(current_text, str) is False:
raise ReviewQueueError("Draft text must be a string.")
edited_text = self.editor.edit_text(current_text)
draft["text"] = edited_text
draft["updated_at"] = self._now()
self.repository.update(draft)
print("UPDATED")
print(draft_id)
def auto_approve(self) -> None:
drafts = self.repository.list_by_status(ReviewStatus.DRAFT)
approved_count = 0
for draft in drafts:
if self.auto_approval_policy.can_auto_approve(draft) is False:
continue
draft["status"] = ReviewStatus.APPROVED
draft["approved_at"] = self._now()
draft["approval_mode"] = "auto"
draft["updated_at"] = self._now()
self.repository.update(draft)
approved_count += 1
print("AUTO_APPROVED")
print(str(approved_count))
def _assert_status(self, draft: dict[str, Any], expected_status: str) -> None:
current_status = draft.get("status")
if current_status != expected_status:
raise ReviewQueueError(
"Invalid draft status. Expected "
+ expected_status
+ ", got "
+ str(current_status)
)
def _now(self) -> str:
return datetime.now(ZoneInfo("Europe/Berlin")).isoformat()
def build_service(root_dir: Path) -> ReviewQueueService:
store = ReviewQueueStore(root_dir / "data" / "post_drafts.json")
repository = ReviewQueueRepository(store)
printer = ReviewQueuePrinter()
editor = DraftEditor()
auto_approval_policy = AutoApprovalPolicy()
return ReviewQueueService(
repository=repository,
printer=printer,
editor=editor,
auto_approval_policy=auto_approval_policy,
)
def main() -> None:
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="command", required=True)
list_parser = subparsers.add_parser("list")
list_parser.add_argument("--status", required=False)
show_parser = subparsers.add_parser("show")
show_parser.add_argument("draft_id")
approve_parser = subparsers.add_parser("approve")
approve_parser.add_argument("draft_id")
reject_parser = subparsers.add_parser("reject")
reject_parser.add_argument("draft_id")
edit_parser = subparsers.add_parser("edit")
edit_parser.add_argument("draft_id")
subparsers.add_parser("auto-approve")
args = parser.parse_args()
root_dir = Path(__file__).resolve().parent.parent
service = build_service(root_dir)
if args.command == "list":
service.list_drafts(args.status)
return
if args.command == "show":
service.show(args.draft_id)
return
if args.command == "approve":
service.approve(args.draft_id)
return
if args.command == "reject":
service.reject(args.draft_id)
return
if args.command == "edit":
service.edit(args.draft_id)
return
if args.command == "auto-approve":
service.auto_approve()
return
raise ReviewQueueError("Unsupported command: " + str(args.command))
if __name__ == "__main__":
main()
Beispielaufrufe
Alle Entwürfe anzeigen:
python -m src.review_queue list
Nur offene Entwürfe anzeigen:
python -m src.review_queue list --status draft
Entwurf anzeigen:
python -m src.review_queue show draft_20260512_001
Entwurf bearbeiten:
python -m src.review_queue edit draft_20260512_001
Entwurf freigeben:
python -m src.review_queue approve draft_20260512_001
Entwurf ablehnen:
python -m src.review_queue reject draft_20260512_001
Regelbasierte Freigabe:
python -m src.review_queue auto-approve
Ergebnis dieses Kapitels
Die Review-Queue stellt jetzt bereit:
zentrale Freigabeschicht
Statusverwaltung
Terminal-Vorschau
manuelle Freigabe
manuelle Ablehnung
Bearbeitung im Editor
regelbasierte automatische Freigabe
JSON-basierte Speicherung
Damit ist der Übergang vom generierten Entwurf zum veröffentlichbaren Beitrag kontrolliert.
0 Kommentare