Was hat das System heute getan?
Was ist fehlgeschlagen?
Welche Beiträge wurden veröffentlicht?
Welche Interaktionen wurden vorgeschlagen?
Welche OpenAI-Kosten sind entstanden?
Welche Daten müssen manuell ergänzt werden?

storage/logs/events.jsonl
storage/logs/errors.jsonl
storage/logs/openai_usage.jsonl
data/post_drafts.json
data/interaction_candidates.json

Datum
veröffentlichte Beiträge
fehlgeschlagene Beiträge
blockierte Plattformzustände
erzeugte Entwürfe
freigegebene Beiträge
Kommentarvorschläge
OpenAI-Token
offene Review-Queue

TAGESREPORT 2026-05-12

Beiträge:
- veröffentlicht: 1
- fehlgeschlagen: 0
- blockiert: 0
- offen in Review: 3

Interaktionen:
- Kandidaten: 4
- Vorschläge: 2
- freigegeben: 0
- ausgeführt: 0

OpenAI:
- Requests: 6
- Input-Tokens: 2.410
- Output-Tokens: 980
- Total-Tokens: 3.390

Fehler:
- keine

Beiträge pro Plattform
Fehler pro Plattform
häufige Beitragstypen
genutzte Themen
Interaktionsvorschläge
Tokenverbrauch
manuell ergänzte Reichweitenwerte

WOCHENREPORT 2026-W20

LinkedIn:
- Beiträge: 3
- Fehler: 0
- Interaktionsvorschläge: 4

Instagram:
- Beiträge: 2
- Fehler: 1
- häufigster Fehler: image upload timeout

Facebook:
- Beiträge: 2
- Fehler: 0

OpenAI:
- Requests: 31
- Total-Tokens: 18.240

data/post_drafts.json

status == published
published_at im Zeitraum

{
  "id": "draft_20260512_001",
  "platform": "linkedin",
  "post_type": "workshop",
  "status": "published",
  "text": "Gemeinfreie Texte sind frei verfügbar, aber nicht automatisch verlagsfertig.",
  "published_at": "2026-05-12T09:30:00+02:00",
  "screenshot_path": "storage/screenshots/linkedin/draft_20260512_001_published.png"
}

ID
Plattform
Typ
Veröffentlichungszeit
Screenshot
Textauszug

data/post_drafts.json
storage/logs/errors.jsonl

status == failed
status == blocked
failed_at
blocked_at
error_message
screenshot_path

Fehlgeschlagene Beiträge:

draft_20260512_002
Plattform: instagram
Status: failed
Fehler: Caption textbox not found.
Screenshot: storage/screenshots/instagram/draft_20260512_002_failed.png

failed: technischer Fehler
blocked: Login, CAPTCHA, Sicherheitsprüfung oder Plattformwarnung

data/interaction_candidates.json

candidate
analyzed
suggested
approved
executed
rejected
failed
blocked

neue Kandidaten
neue Kommentarvorschläge
freigegebene Vorschläge
ausgeführte Interaktionen
blockierte Interaktionen

Interaktionen 2026-05-12:

suggested:
- interaction_20260512_001 / linkedin / external_comment
- interaction_20260512_002 / linkedin / comment_reply

rejected:
- interaction_20260512_003 / blocked topic: politics

storage/logs/openai_usage.jsonl

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

Requests
Input-Tokens
Output-Tokens
Total-Tokens
nach Modell
nach Use-Case
nach Plattform

cost_estimates:
  gpt-4.1-mini:
    input_per_1m_tokens: 0.40
    output_per_1m_tokens: 1.60

input_cost = input_tokens / 1_000_000 * input_price
output_cost = output_tokens / 1_000_000 * output_price

data/manual_metrics.json

[
  {
    "post_id": "draft_20260512_001",
    "platform": "linkedin",
    "recorded_at": "2026-05-13T09:00:00+02:00",
    "impressions": 420,
    "likes": 7,
    "comments": 1,
    "shares": 0,
    "clicks": 3
  }
]

Impressionen
Likes
Kommentare
Shares
Klicks

Klicks
Kommentare mit Substanz
Newsletter-Anmeldungen
Shop-Besuche
Blog-Besuche

src/report.py

import argparse
import json
from dataclasses import dataclass
from datetime import date
from datetime import datetime
from pathlib import Path
from typing import Any


class ReportError(Exception):
    pass


@dataclass(frozen=True)
class ReportPeriod:
    start_date: date
    end_date: date


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

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

        if isinstance(payload, list) is False:
            raise ReportError("JSON file must contain a list: " + str(file_path))

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

        for item in payload:
            if isinstance(item, dict) is False:
                raise ReportError("JSON list contains non-object item: " + str(file_path))

            result.append(item)

        return result


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

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

        with file_path.open("r", encoding="utf-8") as file:
            for line in file:
                cleaned_line = line.strip()

                if cleaned_line == "":
                    continue

                payload = json.loads(cleaned_line)

                if isinstance(payload, dict) is False:
                    raise ReportError("JSONL line must contain an object: " + str(file_path))

                events.append(payload)

        return events


class DateFilter:
    def filter_by_field(
        self,
        items: list[dict[str, Any]],
        field_name: str,
        period: ReportPeriod,
    ) -> list[dict[str, Any]]:
        result: list[dict[str, Any]] = []

        for item in items:
            value = item.get(field_name)

            if isinstance(value, str) is False:
                continue

            item_date = self._parse_date(value)

            if item_date < period.start_date:
                continue

            if item_date > period.end_date:
                continue

            result.append(item)

        return result

    def filter_events(
        self,
        events: list[dict[str, Any]],
        period: ReportPeriod,
    ) -> list[dict[str, Any]]:
        result: list[dict[str, Any]] = []

        for event in events:
            timestamp = event.get("timestamp")

            if isinstance(timestamp, str) is False:
                continue

            event_date = self._parse_date(timestamp)

            if event_date < period.start_date:
                continue

            if event_date > period.end_date:
                continue

            result.append(event)

        return result

    def _parse_date(self, value: str) -> date:
        normalized_value = value.replace("Z", "+00:00")
        parsed = datetime.fromisoformat(normalized_value)

        return parsed.date()


class PostReportAnalyzer:
    def count_by_status(
        self,
        posts: list[dict[str, Any]],
        status: str,
    ) -> int:
        count = 0

        for post in posts:
            if post.get("status") == status:
                count += 1

        return count

    def count_by_platform(
        self,
        posts: list[dict[str, Any]],
        status: str,
    ) -> dict[str, int]:
        result: dict[str, int] = {}

        for post in posts:
            if post.get("status") != status:
                continue

            platform = post.get("platform")

            if isinstance(platform, str) is False:
                continue

            current_count = result.get(platform)

            if current_count is None:
                result[platform] = 1
            else:
                result[platform] = current_count + 1

        return result

    def get_failed_posts(
        self,
        posts: list[dict[str, Any]],
    ) -> list[dict[str, Any]]:
        result: list[dict[str, Any]] = []

        for post in posts:
            status = post.get("status")

            if status == "failed" or status == "blocked":
                result.append(post)

        return result


class InteractionReportAnalyzer:
    def count_by_status(
        self,
        interactions: list[dict[str, Any]],
    ) -> dict[str, int]:
        result: dict[str, int] = {}

        for interaction in interactions:
            status = interaction.get("status")

            if isinstance(status, str) is False:
                continue

            current_count = result.get(status)

            if current_count is None:
                result[status] = 1
            else:
                result[status] = current_count + 1

        return result


class OpenAiUsageAnalyzer:
    def summarize(self, events: list[dict[str, Any]]) -> dict[str, Any]:
        request_count = 0
        input_tokens = 0
        output_tokens = 0
        total_tokens = 0
        by_model: dict[str, int] = {}
        by_use_case: dict[str, int] = {}

        for event in events:
            if event.get("event") != "openai_request_completed":
                continue

            payload = event.get("payload")

            if isinstance(payload, dict) is False:
                continue

            request_count += 1
            input_tokens += self._read_int(payload, "input_tokens")
            output_tokens += self._read_int(payload, "output_tokens")
            total_tokens += self._read_int(payload, "total_tokens")

            model = payload.get("model")

            if isinstance(model, str) is True:
                current_model_count = by_model.get(model)

                if current_model_count is None:
                    by_model[model] = 1
                else:
                    by_model[model] = current_model_count + 1

            use_case = payload.get("use_case")

            if isinstance(use_case, str) is True:
                current_use_case_count = by_use_case.get(use_case)

                if current_use_case_count is None:
                    by_use_case[use_case] = 1
                else:
                    by_use_case[use_case] = current_use_case_count + 1

        return {
            "request_count": request_count,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_tokens": total_tokens,
            "by_model": by_model,
            "by_use_case": by_use_case,
        }

    def _read_int(self, payload: dict[str, Any], key: str) -> int:
        value = payload.get(key)

        if isinstance(value, int) is False:
            return 0

        return value


class ManualMetricsAnalyzer:
    def summarize(self, metrics: list[dict[str, Any]]) -> dict[str, int]:
        impressions = 0
        likes = 0
        comments = 0
        shares = 0
        clicks = 0

        for metric in metrics:
            impressions += self._read_int(metric, "impressions")
            likes += self._read_int(metric, "likes")
            comments += self._read_int(metric, "comments")
            shares += self._read_int(metric, "shares")
            clicks += self._read_int(metric, "clicks")

        return {
            "impressions": impressions,
            "likes": likes,
            "comments": comments,
            "shares": shares,
            "clicks": clicks,
        }

    def _read_int(self, payload: dict[str, Any], key: str) -> int:
        value = payload.get(key)

        if isinstance(value, int) is False:
            return 0

        return value


class ReportPrinter:
    def print_report(
        self,
        title: str,
        period: ReportPeriod,
        posts: list[dict[str, Any]],
        failed_posts: list[dict[str, Any]],
        posts_by_platform: dict[str, int],
        interaction_counts: dict[str, int],
        openai_summary: dict[str, Any],
        manual_metrics_summary: dict[str, int],
    ) -> None:
        print(title + " " + period.start_date.isoformat() + " bis " + period.end_date.isoformat())
        print("")

        print("Beiträge:")
        print("- veröffentlicht: " + str(self._count_status(posts, "published")))
        print("- fehlgeschlagen: " + str(self._count_status(posts, "failed")))
        print("- blockiert: " + str(self._count_status(posts, "blocked")))
        print("- offen: " + str(self._count_status(posts, "draft")))
        print("")

        print("Beiträge nach Plattform:")
        for platform, count in posts_by_platform.items():
            print("- " + platform + ": " + str(count))
        print("")

        print("Fehlgeschlagene Beiträge:")
        if len(failed_posts) == 0:
            print("- keine")
        else:
            for post in failed_posts:
                print("- " + str(post.get("id")) + " / " + str(post.get("platform")) + " / " + str(post.get("error_message")))
        print("")

        print("Interaktionen:")
        for status, count in interaction_counts.items():
            print("- " + status + ": " + str(count))
        print("")

        print("OpenAI:")
        print("- Requests: " + str(openai_summary["request_count"]))
        print("- Input-Tokens: " + str(openai_summary["input_tokens"]))
        print("- Output-Tokens: " + str(openai_summary["output_tokens"]))
        print("- Total-Tokens: " + str(openai_summary["total_tokens"]))
        print("")

        print("Manuelle Reichweitenwerte:")
        print("- Impressionen: " + str(manual_metrics_summary["impressions"]))
        print("- Likes: " + str(manual_metrics_summary["likes"]))
        print("- Kommentare: " + str(manual_metrics_summary["comments"]))
        print("- Shares: " + str(manual_metrics_summary["shares"]))
        print("- Klicks: " + str(manual_metrics_summary["clicks"]))

    def _count_status(self, posts: list[dict[str, Any]], status: str) -> int:
        count = 0

        for post in posts:
            if post.get("status") == status:
                count += 1

        return count


class ReportService:
    def __init__(self, root_dir: Path) -> None:
        self.root_dir = root_dir
        self.json_reader = JsonFileReader()
        self.jsonl_reader = JsonlFileReader()
        self.date_filter = DateFilter()
        self.post_analyzer = PostReportAnalyzer()
        self.interaction_analyzer = InteractionReportAnalyzer()
        self.openai_analyzer = OpenAiUsageAnalyzer()
        self.manual_metrics_analyzer = ManualMetricsAnalyzer()
        self.printer = ReportPrinter()

    def daily(self, report_date: date) -> None:
        period = ReportPeriod(
            start_date=report_date,
            end_date=report_date,
        )

        self._print(period, "TAGESREPORT")

    def weekly(self, start_date: date, end_date: date) -> None:
        period = ReportPeriod(
            start_date=start_date,
            end_date=end_date,
        )

        self._print(period, "WOCHENREPORT")

    def _print(self, period: ReportPeriod, title: str) -> None:
        posts = self.json_reader.load_list(self.root_dir / "data" / "post_drafts.json")
        interactions = self.json_reader.load_list(self.root_dir / "data" / "interaction_candidates.json")
        manual_metrics = self.json_reader.load_list(self.root_dir / "data" / "manual_metrics.json")
        openai_events = self.jsonl_reader.load_events(self.root_dir / "storage" / "logs" / "openai_usage.jsonl")

        period_posts = self._filter_posts_for_period(posts, period)
        period_interactions = self.date_filter.filter_by_field(interactions, "created_at", period)
        period_manual_metrics = self.date_filter.filter_by_field(manual_metrics, "recorded_at", period)
        period_openai_events = self.date_filter.filter_events(openai_events, period)

        failed_posts = self.post_analyzer.get_failed_posts(period_posts)
        posts_by_platform = self.post_analyzer.count_by_platform(period_posts, "published")
        interaction_counts = self.interaction_analyzer.count_by_status(period_interactions)
        openai_summary = self.openai_analyzer.summarize(period_openai_events)
        manual_metrics_summary = self.manual_metrics_analyzer.summarize(period_manual_metrics)

        self.printer.print_report(
            title=title,
            period=period,
            posts=period_posts,
            failed_posts=failed_posts,
            posts_by_platform=posts_by_platform,
            interaction_counts=interaction_counts,
            openai_summary=openai_summary,
            manual_metrics_summary=manual_metrics_summary,
        )

    def _filter_posts_for_period(
        self,
        posts: list[dict[str, Any]],
        period: ReportPeriod,
    ) -> list[dict[str, Any]]:
        result: list[dict[str, Any]] = []

        for post in posts:
            candidate_fields = [
                "published_at",
                "failed_at",
                "blocked_at",
                "created_at",
            ]

            for field_name in candidate_fields:
                value = post.get(field_name)

                if isinstance(value, str) is False:
                    continue

                item_date = datetime.fromisoformat(value.replace("Z", "+00:00")).date()

                if item_date < period.start_date:
                    continue

                if item_date > period.end_date:
                    continue

                result.append(post)
                break

        return result


def parse_date(value: str) -> date:
    return date.fromisoformat(value)


def main() -> None:
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="command", required=True)

    daily_parser = subparsers.add_parser("daily")
    daily_parser.add_argument("--date", required=True)

    weekly_parser = subparsers.add_parser("weekly")
    weekly_parser.add_argument("--start", required=True)
    weekly_parser.add_argument("--end", required=True)

    args = parser.parse_args()

    root_dir = Path(__file__).resolve().parent.parent
    service = ReportService(root_dir)

    if args.command == "daily":
        service.daily(parse_date(args.date))
        return

    if args.command == "weekly":
        service.weekly(
            start_date=parse_date(args.start),
            end_date=parse_date(args.end),
        )
        return

    raise ReportError("Unsupported command: " + str(args.command))


if __name__ == "__main__":
    main()

python -m src.report daily --date 2026-05-12

python -m src.report weekly --start 2026-05-11 --end 2026-05-17

0 18 * * 1-5 cd /opt/social-publisher && .venv/bin/python -m src.report daily --date $(date +\%F) >> storage/logs/reports.log 2>&1

0 8 * * 1 cd /opt/social-publisher && .venv/bin/python -m src.report weekly --start $(date -d 'last monday -7 days' +\%F) --end $(date -d 'last sunday' +\%F) >> storage/logs/reports.log 2>&1

Tagesreport
Wochenreport
Liste veröffentlichter Beiträge
Liste fehlgeschlagener Beiträge
Interaktionsstatus
OpenAI-Tokenauswertung
manuelle Reichweitenmetriken
CLI-Ausgabe für operative Kontrolle