Artikel
24. CLI-Werkzeuge
Die CLI ist die zentrale Bedienoberfläche für das MVP. Sie verbindet Generator, Review-Queue, Publisher, Logs und Reports.
Ziel:
ein Einstiegspunkt
klare Kommandos
keine versteckte Magie
Dry-Run als Standard
Live-Betrieb nur explizit
Empfohlener Aufruf:
python -m src.main <command>
24.1 Beiträge generieren
Ein Beitrag wird aus einer Content-Quelle erzeugt und als Entwurf gespeichert.
python -m src.main generate --platform linkedin --type workshop
Mit Varianten:
python -m src.main generate --platform linkedin --type workshop --variants 3
Mit konkreter Quelle:
python -m src.main generate --platform linkedin --type book_intro --source-id book_001
Erwartete Ausgabe:
DRAFT_CREATED
draft_20260512_093000_123456
24.2 Entwürfe anzeigen
Alle Entwürfe:
python -m src.main queue list
Nur offene Entwürfe:
python -m src.main queue list --status draft
Einzelnen Entwurf anzeigen:
python -m src.main queue show draft_20260512_093000_123456
Ausgabe:
ID: draft_20260512_093000_123456
Platform: linkedin
Type: workshop
Status: draft
Text:
Gemeinfreie Texte sind frei verfügbar, aber nicht automatisch verlagsfertig.
24.3 Beitrag freigeben
python -m src.main queue approve draft_20260512_093000_123456
Ergebnis:
APPROVED
draft_20260512_093000_123456
Die Freigabe ändert nur den Status.
draft → approved
Veröffentlicht wird dadurch noch nichts.
24.4 Beitrag ablehnen
python -m src.main queue reject draft_20260512_093000_123456
Ergebnis:
REJECTED
draft_20260512_093000_123456
Statuswechsel:
draft → rejected
Abgelehnte Beiträge bleiben im Archiv erhalten.
24.5 Posting ausführen
Dry-Run ist Standard:
python -m src.main publish draft_20260512_093000_123456
Live-Veröffentlichung:
python -m src.main publish draft_20260512_093000_123456 --live
Nächsten freigegebenen Beitrag veröffentlichen:
python -m src.main publish-next
Live:
python -m src.main publish-next --live
Ergebnis bei Dry-Run:
DRY_RUN
draft_20260512_093000_123456
Ergebnis bei Erfolg:
PUBLISHED
draft_20260512_093000_123456
Ergebnis bei Fehler:
FAILED
draft_20260512_093000_123456
Post button not found.
Ergebnis bei kritischem Plattformzustand:
BLOCKED
draft_20260512_093000_123456
Login required.
24.6 Queue prüfen
Queue-Status zusammenfassen:
python -m src.main queue status
Beispielausgabe:
QUEUE STATUS
draft: 4
approved: 2
published: 12
rejected: 3
failed: 1
blocked: 0
Offene Veröffentlichungen:
python -m src.main queue list --status approved
Fehlerhafte Beiträge:
python -m src.main queue list --status failed
Blockierte Beiträge:
python -m src.main queue list --status blocked
24.7 Logs anzeigen
Letzte Events:
python -m src.main logs events --limit 20
Letzte Fehler:
python -m src.main logs errors --limit 20
OpenAI-Nutzung:
python -m src.main logs openai --limit 20
Plattformantworten:
python -m src.main logs platform --limit 20
Beispielausgabe:
2026-05-12T09:30:00+02:00 info post_published linkedin draft_20260512_001
2026-05-12T09:31:00+02:00 error post_failed instagram draft_20260512_002
24.8 Report erzeugen
Tagesreport:
python -m src.main report daily --date 2026-05-12
Wochenreport:
python -m src.main report weekly --start 2026-05-11 --end 2026-05-17
Ausgabe:
TAGESREPORT 2026-05-12 bis 2026-05-12
Beiträge:
- veröffentlicht: 1
- fehlgeschlagen: 0
- blockiert: 0
- offen: 3
OpenAI:
- Requests: 6
- Total-Tokens: 3390
24.9 Beispiel: main.py mit argparse
Datei:
src/main.py
import argparse
import json
from pathlib import Path
from typing import Any
class MainCliError(Exception):
pass
class LogPrinter:
def __init__(self, root_dir: Path) -> None:
self.root_dir = root_dir
def print_events(self, log_name: str, limit: int) -> None:
log_file = self._get_log_file(log_name)
if log_file.exists() is False:
print("NO_LOG_FILE")
print(str(log_file))
return
lines = log_file.read_text(encoding="utf-8").splitlines()
selected_lines = lines[-limit:]
for line in selected_lines:
payload = json.loads(line)
if isinstance(payload, dict) is False:
continue
timestamp = str(payload.get("timestamp", ""))
level = str(payload.get("level", ""))
event = str(payload.get("event", ""))
platform = str(payload.get("platform", ""))
reference_id = str(payload.get("reference_id", ""))
print(
timestamp
+ " "
+ level.ljust(8)
+ event.ljust(28)
+ platform.ljust(12)
+ reference_id
)
def _get_log_file(self, log_name: str) -> Path:
if log_name == "events":
return self.root_dir / "storage" / "logs" / "events.jsonl"
if log_name == "errors":
return self.root_dir / "storage" / "logs" / "errors.jsonl"
if log_name == "openai":
return self.root_dir / "storage" / "logs" / "openai_usage.jsonl"
if log_name == "platform":
return self.root_dir / "storage" / "logs" / "platform_responses.jsonl"
raise MainCliError("Unsupported log name: " + log_name)
class QueueStatusPrinter:
def __init__(self, root_dir: Path) -> None:
self.root_dir = root_dir
def print_status(self) -> None:
queue_file = self.root_dir / "data" / "post_drafts.json"
if queue_file.exists() is False:
print("NO_QUEUE")
return
with queue_file.open("r", encoding="utf-8") as file:
payload = json.load(file)
if isinstance(payload, list) is False:
raise MainCliError("Queue must contain a list.")
counts: dict[str, int] = {
"draft": 0,
"approved": 0,
"published": 0,
"rejected": 0,
"failed": 0,
"blocked": 0,
"dry_run": 0,
}
for item in payload:
if isinstance(item, dict) is False:
continue
status = item.get("status")
if isinstance(status, str) is False:
continue
current_count = counts.get(status)
if current_count is None:
counts[status] = 1
else:
counts[status] = current_count + 1
print("QUEUE STATUS")
print("")
for status, count in counts.items():
print(status.ljust(10) + str(count))
def run_generate(args: argparse.Namespace) -> None:
from src.generate_posts import main as generate_main
generate_main()
def run_queue(args: argparse.Namespace) -> None:
from src.review_queue import build_service
root_dir = Path(__file__).resolve().parent.parent
service = build_service(root_dir)
if args.queue_command == "list":
service.list_drafts(args.status)
return
if args.queue_command == "show":
service.show(args.draft_id)
return
if args.queue_command == "approve":
service.approve(args.draft_id)
return
if args.queue_command == "reject":
service.reject(args.draft_id)
return
if args.queue_command == "edit":
service.edit(args.draft_id)
return
if args.queue_command == "auto-approve":
service.auto_approve()
return
if args.queue_command == "status":
QueueStatusPrinter(root_dir).print_status()
return
raise MainCliError("Unsupported queue command: " + str(args.queue_command))
def run_publish(args: argparse.Namespace) -> None:
from src.publisher import build_service
root_dir = Path(__file__).resolve().parent.parent
dry_run = True
if args.live is True:
dry_run = False
service = build_service(root_dir, dry_run)
service.publish_one(args.post_id)
def run_publish_next(args: argparse.Namespace) -> None:
from src.publisher import build_service
root_dir = Path(__file__).resolve().parent.parent
dry_run = True
if args.live is True:
dry_run = False
service = build_service(root_dir, dry_run)
service.publish_next()
def run_logs(args: argparse.Namespace) -> None:
root_dir = Path(__file__).resolve().parent.parent
printer = LogPrinter(root_dir)
printer.print_events(
log_name=args.log_name,
limit=args.limit,
)
def run_report(args: argparse.Namespace) -> None:
from src.report import ReportService
from src.report import parse_date
root_dir = Path(__file__).resolve().parent.parent
service = ReportService(root_dir)
if args.report_command == "daily":
service.daily(parse_date(args.date))
return
if args.report_command == "weekly":
service.weekly(
start_date=parse_date(args.start),
end_date=parse_date(args.end),
)
return
raise MainCliError("Unsupported report command: " + str(args.report_command))
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="social-publisher",
)
subparsers = parser.add_subparsers(
dest="command",
required=True,
)
generate_parser = subparsers.add_parser("generate")
generate_parser.add_argument("--platform", required=False)
generate_parser.add_argument("--type", required=False)
generate_parser.add_argument("--source-id", required=False)
generate_parser.add_argument("--variants", required=False, type=int)
queue_parser = subparsers.add_parser("queue")
queue_subparsers = queue_parser.add_subparsers(
dest="queue_command",
required=True,
)
queue_list_parser = queue_subparsers.add_parser("list")
queue_list_parser.add_argument("--status", required=False)
queue_show_parser = queue_subparsers.add_parser("show")
queue_show_parser.add_argument("draft_id")
queue_approve_parser = queue_subparsers.add_parser("approve")
queue_approve_parser.add_argument("draft_id")
queue_reject_parser = queue_subparsers.add_parser("reject")
queue_reject_parser.add_argument("draft_id")
queue_edit_parser = queue_subparsers.add_parser("edit")
queue_edit_parser.add_argument("draft_id")
queue_subparsers.add_parser("auto-approve")
queue_subparsers.add_parser("status")
publish_parser = subparsers.add_parser("publish")
publish_parser.add_argument("post_id")
publish_parser.add_argument("--live", action="store_true")
publish_next_parser = subparsers.add_parser("publish-next")
publish_next_parser.add_argument("--live", action="store_true")
logs_parser = subparsers.add_parser("logs")
logs_parser.add_argument(
"log_name",
choices=[
"events",
"errors",
"openai",
"platform",
],
)
logs_parser.add_argument("--limit", required=False, type=int, default=20)
report_parser = subparsers.add_parser("report")
report_subparsers = report_parser.add_subparsers(
dest="report_command",
required=True,
)
daily_parser = report_subparsers.add_parser("daily")
daily_parser.add_argument("--date", required=True)
weekly_parser = report_subparsers.add_parser("weekly")
weekly_parser.add_argument("--start", required=True)
weekly_parser.add_argument("--end", required=True)
return parser
def main() -> None:
parser = build_parser()
args = parser.parse_args()
if args.command == "generate":
run_generate(args)
return
if args.command == "queue":
run_queue(args)
return
if args.command == "publish":
run_publish(args)
return
if args.command == "publish-next":
run_publish_next(args)
return
if args.command == "logs":
run_logs(args)
return
if args.command == "report":
run_report(args)
return
raise MainCliError("Unsupported command: " + str(args.command))
if __name__ == "__main__":
main()
Kommandoreferenz
python -m src.main generate --platform linkedin --type workshop
python -m src.main queue list
python -m src.main queue list --status draft
python -m src.main queue show draft_20260512_001
python -m src.main queue edit draft_20260512_001
python -m src.main queue approve draft_20260512_001
python -m src.main queue reject draft_20260512_001
python -m src.main queue auto-approve
python -m src.main queue status
python -m src.main publish draft_20260512_001
python -m src.main publish draft_20260512_001 --live
python -m src.main publish-next
python -m src.main publish-next --live
python -m src.main logs events --limit 20
python -m src.main logs errors --limit 20
python -m src.main logs openai --limit 20
python -m src.main report daily --date 2026-05-12
python -m src.main report weekly --start 2026-05-11 --end 2026-05-17
Ergebnis dieses Kapitels
Die CLI stellt jetzt bereit:
Beitragserzeugung
Queue-Verwaltung
Freigabe
Ablehnung
Bearbeitung
Dry-Run
Live-Publishing
Queue-Status
Log-Anzeige
Tagesreport
Wochenreport
Damit ist das System ohne Weboberfläche bedienbar. Für das MVP reicht diese CLI vollständig aus.
0 Kommentare