Journal und Werkstatt

1manfactory.com - Blog of Jürgen Schulze

Stack: Linux, Nginx, MySQL, PHP, Docker, Oracle VM VirtualBox

1manfactory.com - Blog of Jürgen Schulze
Autor

Live-Status

ModusWerkstatt-Journal
Feed302 Artikel
Themen52 Kategorien
Artikel
12.05.2026

21. Logging

Allgemein

Logging ist kein Zusatzmodul, sondern Bestandteil der Betriebssicherheit. Das System muss später rekonstruieren können:

was erzeugt wurde
was geprüft wurde
was freigegeben wurde
was veröffentlicht wurde
was fehlgeschlagen ist
welcher Screenshot dazugehört
welcher OpenAI-Aufruf Kosten verursacht hat

Für das MVP reicht JSONL.

storage/logs/events.jsonl
storage/logs/errors.jsonl
storage/logs/openai_usage.jsonl
storage/logs/platform_responses.jsonl

JSONL bedeutet:

eine JSON-Zeile pro Ereignis
append-only
leicht per grep, jq oder Python auswertbar
kein Datenbankserver notwendig

21.1 Strukturierte Logs

Nicht sinnvoll:

Post veröffentlicht
Fehler beim Upload
OpenAI fertig

Sinnvoll:

{
  "timestamp": "2026-05-12T09:30:00+02:00",
  "level": "info",
  "event": "post_published",
  "platform": "linkedin",
  "reference_id": "draft_20260512_001",
  "payload": {
    "status": "published",
    "screenshot_path": "storage/screenshots/linkedin/draft_20260512_001_published.png"
  }
}

Pflichtfelder:

timestamp
level
event
reference_id
platform
payload

Log-Level:

debug
info
warning
error
critical

Event-Namen sollten stabil und maschinenlesbar sein:

post_draft_created
quality_gate_passed
quality_gate_failed
post_approved
post_rejected
publish_started
post_published
post_failed
post_blocked
interaction_candidate_created
comment_suggested
comment_rejected
openai_request_completed
platform_warning_detected
screenshot_saved

21.2 Aktionsprotokoll

Das Aktionsprotokoll speichert normale Systemereignisse.

Datei:

storage/logs/events.jsonl

Beispiele:

{
  "timestamp": "2026-05-12T09:00:00+02:00",
  "level": "info",
  "event": "post_draft_created",
  "platform": "linkedin",
  "reference_id": "draft_20260512_001",
  "payload": {
    "source_type": "manual_topic",
    "source_id": "topic_001",
    "post_type": "workshop",
    "prompt_name": "workshop_note",
    "prompt_version": "1.0.0",
    "model": "gpt-4.1-mini"
  }
}


{
  "timestamp": "2026-05-12T09:10:00+02:00",
  "level": "info",
  "event": "quality_gate_passed",
  "platform": "linkedin",
  "reference_id": "draft_20260512_001",
  "payload": {
    "checks": {
      "hard_reject": true,
      "character_limit": true,
      "link": true,
      "generic": true,
      "duplicate": true,
      "similarity": true,
      "tone": true,
      "sales_pressure": true
    }
  }
}


{
  "timestamp": "2026-05-12T09:30:00+02:00",
  "level": "info",
  "event": "post_published",
  "platform": "linkedin",
  "reference_id": "draft_20260512_001",
  "payload": {
    "screenshot_path": "storage/screenshots/linkedin/draft_20260512_001_published.png",
    "platform_post_url": null
  }
}

21.3 Fehlerprotokoll

Fehler gehören in eine eigene Datei.

Datei:

storage/logs/errors.jsonl

Beispiel:

{
  "timestamp": "2026-05-12T09:31:00+02:00",
  "level": "error",
  "event": "post_failed",
  "platform": "linkedin",
  "reference_id": "draft_20260512_001",
  "payload": {
    "error_type": "PlaywrightTimeoutError",
    "error_message": "Post button not found.",
    "screenshot_path": "storage/screenshots/linkedin/draft_20260512_001_failed.png"
  }
}

Kritische Fehler:

{
  "timestamp": "2026-05-12T09:31:00+02:00",
  "level": "critical",
  "event": "platform_warning_detected",
  "platform": "linkedin",
  "reference_id": "draft_20260512_001",
  "payload": {
    "reason": "security check detected",
    "action": "platform_blocked",
    "screenshot_path": "storage/screenshots/linkedin/draft_20260512_001_blocked.png"
  }
}

Fehler werden nicht nur geloggt. Sie müssen auch den Status des betroffenen Objekts verändern:

approved → failed
approved → blocked
candidate → failed
candidate → blocked

21.4 Screenshot-Archiv

Screenshots sind Belege.

Gespeichert wird bei:

Dry-Run
erfolgreicher Veröffentlichung
Fehler
Login erforderlich
CAPTCHA
Plattformwarnung
unklarem Status

Verzeichnisstruktur:

storage/screenshots/
├── linkedin/
│   ├── draft_20260512_001_dry_run.png
│   ├── draft_20260512_001_published.png
│   └── draft_20260512_002_failed.png
├── instagram/
└── facebook/

Dateinamenschema:

{reference_id}_{status}.png

Beispiele:

draft_20260512_001_dry_run.png
draft_20260512_001_published.png
draft_20260512_001_failed.png
draft_20260512_001_blocked.png
interaction_20260512_001_suggested.png

Screenshot-Log:

{
  "timestamp": "2026-05-12T09:30:00+02:00",
  "level": "info",
  "event": "screenshot_saved",
  "platform": "linkedin",
  "reference_id": "draft_20260512_001",
  "payload": {
    "status": "published",
    "screenshot_path": "storage/screenshots/linkedin/draft_20260512_001_published.png"
  }
}

21.5 API-Kostenprotokoll

OpenAI-Aufrufe müssen getrennt protokolliert werden.

Datei:

storage/logs/openai_usage.jsonl

Gespeichert wird:

Zeitpunkt
Response-ID
Modell
Prompt-Name
Prompt-Version
Reference-ID
Input-Tokens
Output-Tokens
Total-Tokens
Use-Case

Beispiel:

{
  "timestamp": "2026-05-12T09:00:00+02:00",
  "level": "info",
  "event": "openai_request_completed",
  "platform": "linkedin",
  "reference_id": "draft_20260512_001",
  "payload": {
    "response_id": "resp_123",
    "model": "gpt-4.1-mini",
    "use_case": "post_generation",
    "prompt_name": "workshop_note",
    "prompt_version": "1.0.0",
    "input_tokens": 420,
    "output_tokens": 180,
    "total_tokens": 600
  }
}

Ein einfacher Tagesreport kann daraus später berechnen:

OpenAI-Aufrufe pro Tag
Tokens pro Tag
Tokens pro Modell
Tokens pro Use-Case
Kostenabschätzung

Das Log speichert Tokens, nicht zwingend Euro-Beträge. Preise ändern sich; Kosten können beim Report anhand aktueller Preisparameter berechnet werden.

21.6 Plattformantworten speichern

Plattformantworten sind bei Playwright nicht immer API-Antworten. Meist sind es UI-Zustände.

Gespeichert werden sollte:

aktuelle URL
Seitentitel
Status
sichtbarer Hinweistext
Screenshot
Fehlertyp

Datei:

storage/logs/platform_responses.jsonl

Beispiel:

{
  "timestamp": "2026-05-12T09:30:00+02:00",
  "level": "info",
  "event": "platform_response",
  "platform": "linkedin",
  "reference_id": "draft_20260512_001",
  "payload": {
    "url": "https://www.linkedin.com/feed/",
    "title": "Feed | LinkedIn",
    "status": "after_publish_click",
    "screenshot_path": "storage/screenshots/linkedin/draft_20260512_001_published.png"
  }
}

Bei Fehler:

{
  "timestamp": "2026-05-12T09:31:00+02:00",
  "level": "warning",
  "event": "platform_response",
  "platform": "linkedin",
  "reference_id": "draft_20260512_002",
  "payload": {
    "url": "https://www.linkedin.com/checkpoint/challenge/",
    "title": "Security Verification | LinkedIn",
    "status": "blocked",
    "detected_fragment": "security verification",
    "screenshot_path": "storage/screenshots/linkedin/draft_20260512_002_blocked.png"
  }
}

Optional kann ein kurzer HTML-Auszug gespeichert werden. Nicht sinnvoll ist das vollständige Archivieren ganzer Social-Media-Seiten.

HTML-Auszug nur bei Fehlern
keine vollständigen Feeds archivieren
keine personenbezogenen Daten unnötig speichern

21.7 Beispiel: logger.py

Datei:

src/logger.py


import json
import traceback
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo


class LoggerError(Exception):
    pass


@dataclass(frozen=True)
class LogEvent:
    level: str
    event: str
    platform: str | None
    reference_id: str | None
    payload: dict[str, Any]


class JsonlLogger:
    def __init__(self, log_file: Path, timezone: str) -> None:
        self.log_file = log_file
        self.timezone = timezone

    def write(self, log_event: LogEvent) -> None:
        self.log_file.parent.mkdir(parents=True, exist_ok=True)

        payload = {
            "timestamp": datetime.now(ZoneInfo(self.timezone)).isoformat(),
            "level": log_event.level,
            "event": log_event.event,
            "platform": log_event.platform,
            "reference_id": log_event.reference_id,
            "payload": log_event.payload,
        }

        with self.log_file.open("a", encoding="utf-8") as file:
            file.write(json.dumps(payload, ensure_ascii=False))
            file.write("\n")


class ApplicationLogger:
    def __init__(self, root_dir: Path, timezone: str) -> None:
        self.root_dir = root_dir
        self.timezone = timezone

        self.event_logger = JsonlLogger(
            log_file=self.root_dir / "storage" / "logs" / "events.jsonl",
            timezone=self.timezone,
        )

        self.error_logger = JsonlLogger(
            log_file=self.root_dir / "storage" / "logs" / "errors.jsonl",
            timezone=self.timezone,
        )

        self.openai_usage_logger = JsonlLogger(
            log_file=self.root_dir / "storage" / "logs" / "openai_usage.jsonl",
            timezone=self.timezone,
        )

        self.platform_response_logger = JsonlLogger(
            log_file=self.root_dir / "storage" / "logs" / "platform_responses.jsonl",
            timezone=self.timezone,
        )

    def info(
        self,
        event: str,
        platform: str | None,
        reference_id: str | None,
        payload: dict[str, Any],
    ) -> None:
        self.event_logger.write(
            LogEvent(
                level="info",
                event=event,
                platform=platform,
                reference_id=reference_id,
                payload=payload,
            )
        )

    def warning(
        self,
        event: str,
        platform: str | None,
        reference_id: str | None,
        payload: dict[str, Any],
    ) -> None:
        self.event_logger.write(
            LogEvent(
                level="warning",
                event=event,
                platform=platform,
                reference_id=reference_id,
                payload=payload,
            )
        )

    def error(
        self,
        event: str,
        platform: str | None,
        reference_id: str | None,
        exception: Exception,
        payload: dict[str, Any],
    ) -> None:
        error_payload = dict(payload)
        error_payload["error_type"] = exception.__class__.__name__
        error_payload["error_message"] = str(exception)
        error_payload["traceback"] = traceback.format_exc()

        self.error_logger.write(
            LogEvent(
                level="error",
                event=event,
                platform=platform,
                reference_id=reference_id,
                payload=error_payload,
            )
        )

    def critical(
        self,
        event: str,
        platform: str | None,
        reference_id: str | None,
        exception: Exception,
        payload: dict[str, Any],
    ) -> None:
        error_payload = dict(payload)
        error_payload["error_type"] = exception.__class__.__name__
        error_payload["error_message"] = str(exception)
        error_payload["traceback"] = traceback.format_exc()

        self.error_logger.write(
            LogEvent(
                level="critical",
                event=event,
                platform=platform,
                reference_id=reference_id,
                payload=error_payload,
            )
        )

    def openai_usage(
        self,
        platform: str | None,
        reference_id: str | None,
        response_id: str,
        model: str,
        use_case: str,
        prompt_name: str,
        prompt_version: str,
        input_tokens: int,
        output_tokens: int,
        total_tokens: int,
    ) -> None:
        self.openai_usage_logger.write(
            LogEvent(
                level="info",
                event="openai_request_completed",
                platform=platform,
                reference_id=reference_id,
                payload={
                    "response_id": response_id,
                    "model": model,
                    "use_case": use_case,
                    "prompt_name": prompt_name,
                    "prompt_version": prompt_version,
                    "input_tokens": input_tokens,
                    "output_tokens": output_tokens,
                    "total_tokens": total_tokens,
                },
            )
        )

    def platform_response(
        self,
        platform: str,
        reference_id: str,
        status: str,
        url: str,
        title: str,
        screenshot_path: str | None,
        extra: dict[str, Any],
    ) -> None:
        payload: dict[str, Any] = {
            "status": status,
            "url": url,
            "title": title,
            "screenshot_path": screenshot_path,
        }

        for key, value in extra.items():
            payload[key] = value

        self.platform_response_logger.write(
            LogEvent(
                level="info",
                event="platform_response",
                platform=platform,
                reference_id=reference_id,
                payload=payload,
            )
        )


class ScreenshotArchiveLogger:
    def __init__(self, application_logger: ApplicationLogger) -> None:
        self.application_logger = application_logger

    def log_saved(
        self,
        platform: str,
        reference_id: str,
        status: str,
        screenshot_path: Path,
    ) -> None:
        self.application_logger.info(
            event="screenshot_saved",
            platform=platform,
            reference_id=reference_id,
            payload={
                "status": status,
                "screenshot_path": str(screenshot_path),
            },
        )

Beispielnutzung

from pathlib import Path

from src.logger import ApplicationLogger


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

    logger = ApplicationLogger(
        root_dir=root_dir,
        timezone="Europe/Berlin",
    )

    logger.info(
        event="post_draft_created",
        platform="linkedin",
        reference_id="draft_20260512_001",
        payload={
            "source_type": "manual_topic",
            "source_id": "topic_001",
            "post_type": "workshop",
            "prompt_name": "workshop_note",
            "prompt_version": "1.0.0",
            "model": "gpt-4.1-mini",
        },
    )

    logger.openai_usage(
        platform="linkedin",
        reference_id="draft_20260512_001",
        response_id="resp_123",
        model="gpt-4.1-mini",
        use_case="post_generation",
        prompt_name="workshop_note",
        prompt_version="1.0.0",
        input_tokens=420,
        output_tokens=180,
        total_tokens=600,
    )


if __name__ == "__main__":
    main()

Logger im Publisher verwenden

Ausschnitt:

logger.info(
    event="publish_started",
    platform=request.platform,
    reference_id=request.id,
    payload={
        "post_type": request.post_type,
        "dry_run": request.dry_run,
    },
)

Bei Erfolg:

logger.info(
    event="post_published",
    platform=request.platform,
    reference_id=request.id,
    payload={
        "screenshot_path": str(screenshot_path),
        "platform_post_url": platform_post_url,
    },
)

Bei Fehler:

logger.error(
    event="post_failed",
    platform=request.platform,
    reference_id=request.id,
    exception=exception,
    payload={
        "screenshot_path": str(screenshot_path),
    },
)

Logger im OpenAI-Client verwenden

Ausschnitt:

logger.openai_usage(
    platform=platform,
    reference_id=reference_id,
    response_id=result.response_id,
    model=result.model,
    use_case="post_generation",
    prompt_name=prompt_name,
    prompt_version=prompt_version,
    input_tokens=result.input_tokens,
    output_tokens=result.output_tokens,
    total_tokens=result.total_tokens,
)

CLI-Auswertung mit jq

Letzte Ereignisse:

tail -n 20 storage/logs/events.jsonl | jq

Fehler anzeigen:

cat storage/logs/errors.jsonl | jq

OpenAI-Token summieren:

jq -s 'map(.payload.total_tokens) | add' storage/logs/openai_usage.jsonl

Veröffentlichte Beiträge anzeigen:

jq 'select(.event == "post_published")' storage/logs/events.jsonl

Fehler nach Plattform filtern:

jq 'select(.platform == "linkedin")' storage/logs/errors.jsonl

Ergebnis dieses Kapitels

Das Logging-System liefert jetzt:

strukturierte JSONL-Logs
Aktionsprotokoll
Fehlerprotokoll
Screenshot-Archiv
OpenAI-Tokenprotokoll
Plattformantworten
auswertbare Eventnamen
klare Referenz auf Beitrag oder Interaktion

Damit ist der Betrieb nachvollziehbar. Jede Veröffentlichung, jeder Fehler und jeder relevante API-Aufruf lässt sich später rekonstruieren.

Ähnliche Beiträge:

0 Kommentare