Artikel

20. Scheduler

Allgemein

Der Scheduler steuert wiederkehrende Abläufe:

Beiträge erzeugen
Quality-Gate ausführen
Entwürfe automatisch freigeben
freigegebene Beiträge veröffentlichen
Interaktionen prüfen
Reports erzeugen
Tageslimits kontrollieren

Für das MVP reicht Cron. Ein Python-Scheduler ist sinnvoll, wenn die Logik stärker in der Anwendung bleiben soll.

Grundregel:

Cron startet nur.
Python entscheidet.

20.1 Einfacher Cronjob

Cron eignet sich für feste Tagesabläufe.

Beispiel:

08:00 Beiträge erzeugen
08:10 Quality-Gate ausführen
08:20 Auto-Approval prüfen
09:00 freigegebenen Beitrag veröffentlichen
17:00 Kommentare unter eigenen Beiträgen prüfen
18:00 Tagesreport erzeugen

Cron ruft einzelne Kommandos auf:

cd /opt/social-publisher && .venv/bin/python -m src.generate_posts
cd /opt/social-publisher && .venv/bin/python -m src.publisher publish-next
cd /opt/social-publisher && .venv/bin/python -m src.interaction_worker analyze

Wichtig:

absolute Pfade verwenden
virtuelle Umgebung explizit verwenden
Logs umleiten
keine parallelen Läufe
Dry-Run zuerst produktiv testen

20.2 Python-Scheduler

Ein Python-Scheduler ist sinnvoll, wenn mehrere Regeln geprüft werden müssen:

Ist heute ein erlaubter Posting-Tag?
Liegt die aktuelle Uhrzeit im Posting-Zeitfenster?
Wurde heute bereits gepostet?
Ist der Kill-Switch aktiv?
Gibt es freigegebene Beiträge?
Ist die Plattform aktiv?

Für das MVP reicht ein eigener einfacher Scheduler ohne externe Bibliothek.

Ablauf:

config.yaml laden
Kill-Switch prüfen
Plattformen durchlaufen
Zeitfenster prüfen
Tageslimits prüfen
Queue prüfen
nächsten freigegebenen Beitrag veröffentlichen
Ergebnis loggen

Cron startet dann nur noch:

python -m src.scheduler run

20.3 Posting-Zeitfenster

Zeitfenster stehen in config.yaml.

timezone: "Europe/Berlin"

posting_days:
  linkedin:
    - "monday"
    - "wednesday"
    - "friday"

posting_windows:
  linkedin:
    - start: "08:30"
      end: "10:30"
    - start: "16:00"
      end: "18:00"

quiet_hours:
  start: "21:00"
  end: "07:00"

Prüflogik:

from datetime import datetime
from datetime import time
from zoneinfo import ZoneInfo


class PostingWindowChecker:
    def __init__(self, timezone: str) -> None:
        self.timezone = timezone

    def is_allowed_now(
        self,
        allowed_days: list[str],
        windows: list[dict[str, str]],
    ) -> bool:
        now = datetime.now(ZoneInfo(self.timezone))
        weekday = now.strftime("%A").lower()

        if weekday not in allowed_days:
            return False

        current_time = now.time()

        for window in windows:
            start_time = self._parse_time(window["start"])
            end_time = self._parse_time(window["end"])

            if current_time >= start_time and current_time <= end_time:
                return True

        return False

    def _parse_time(self, value: str) -> time:
        parts = value.split(":")

        if len(parts) != 2:
            raise RuntimeError("Invalid time format: " + value)

        hour = int(parts[0])
        minute = int(parts[1])

        return time(hour=hour, minute=minute)

20.4 Queue-Verarbeitung

Der Scheduler verarbeitet nur Beiträge mit Status:

approved

Ablauf:

approved posts laden
nach Plattform filtern
ältesten Beitrag wählen
PublishRequest bauen
Publisher ausführen
Status aktualisieren
Screenshot speichern
Log schreiben

Nicht verarbeitet werden:

draft
rejected
published
failed
blocked
dry_run

Beispiel:

class QueueSelector:
    def select_next_approved(
        self,
        posts: list[dict[str, object]],
        platform: str,
    ) -> dict[str, object] | None:
        approved_posts: list[dict[str, object]] = []

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

            if post.get("platform") != platform:
                continue

            approved_posts.append(post)

        if len(approved_posts) == 0:
            return None

        return approved_posts[0]

Für SQLite später:

SELECT *
FROM post_drafts
WHERE status = 'approved'
AND platform = ?
ORDER BY updated_at ASC
LIMIT 1;

20.5 Retry-Strategie

Retry darf nicht aggressiv sein.

Unkritische Fehler:

kurzer Timeout
temporärer Netzwerkfehler
OpenAI-Rate-Limit
Element noch nicht geladen

Kritische Fehler:

Login erforderlich
CAPTCHA
Sicherheitsprüfung
Plattformwarnung
unklarer Veröffentlichungsstatus
Post-Button nach Absenden unklar

Regeln:

maximal 2 Versuche pro technischem Schritt
kein zweiter Klick auf „Posten“, wenn Status unklar ist
bei Login/CAPTCHA sofort blocked
nach 3 Plattformfehlern pro Tag Plattform deaktivieren
failed bleibt failed, bis manuell geprüft

Retry-Helfer:

from collections.abc import Callable


class RetryRunner:
    def run(self, action: Callable[[], None], max_attempts: int) -> None:
        attempt = 1

        while attempt <= max_attempts:
            try:
                action()
                return
            except TemporarySchedulerError:
                if attempt == max_attempts:
                    raise

                attempt += 1

Fehlerklassen:

class SchedulerError(Exception):
    pass


class TemporarySchedulerError(SchedulerError):
    pass


class CriticalSchedulerError(SchedulerError):
    pass

Für Veröffentlichungen gilt:

Publisher selbst entscheidet über failed oder blocked.
Scheduler versucht nicht eigenständig erneut zu posten.

20.6 Tageslimits

Tageslimits werden vor jeder Aktion geprüft.

Beispielkonfiguration:

limits:
  max_posts_per_day: 1
  max_comments_per_day: 3
  max_likes_per_day: 5
  max_failures_per_day: 3

Limitprüfung anhand der Queue:

from datetime import datetime
from zoneinfo import ZoneInfo


class DailyLimitChecker:
    def __init__(self, timezone: str) -> None:
        self.timezone = timezone

    def assert_post_allowed(
        self,
        posts: list[dict[str, object]],
        platform: str,
        max_posts_per_day: int,
    ) -> None:
        today = datetime.now(ZoneInfo(self.timezone)).date().isoformat()
        count = 0

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

            if post.get("status") != "published":
                continue

            published_at = post.get("published_at")

            if isinstance(published_at, str) is False:
                continue

            if published_at.startswith(today):
                count += 1

        if count >= max_posts_per_day:
            raise SchedulerError("Daily post limit reached for platform: " + platform)

Fehlerlimit:

class DailyFailureLimitChecker:
    def __init__(self, timezone: str) -> None:
        self.timezone = timezone

    def assert_failure_limit_not_reached(
        self,
        posts: list[dict[str, object]],
        platform: str,
        max_failures_per_day: int,
    ) -> None:
        today = datetime.now(ZoneInfo(self.timezone)).date().isoformat()
        count = 0

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

            failed_at = post.get("failed_at")

            if isinstance(failed_at, str) is False:
                continue

            if failed_at.startswith(today):
                count += 1

        if count >= max_failures_per_day:
            raise SchedulerError("Daily failure limit reached for platform: " + platform)

20.7 Beispiel: scheduler.py

Datei:

src/scheduler.py

import argparse
import json
from datetime import datetime
from datetime import time
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo

from src.publisher import PublisherService
from src.publisher import build_service as build_publisher_service


class SchedulerError(Exception):
    pass


class SchedulerStore:
    def __init__(self, file_path: Path) -> None:
        self.file_path = file_path

    def load_posts(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 SchedulerError("Post queue must contain a list.")

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

        for item in payload:
            if isinstance(item, dict) is False:
                raise SchedulerError("Post queue contains non-object item.")

            posts.append(item)

        return posts


class SchedulerConfigStore:
    def __init__(self, file_path: Path) -> None:
        self.file_path = file_path

    def load(self) -> dict[str, Any]:
        import yaml

        if self.file_path.exists() is False:
            raise SchedulerError("Missing config.yaml.")

        with self.file_path.open("r", encoding="utf-8") as file:
            payload = yaml.safe_load(file)

        if isinstance(payload, dict) is False:
            raise SchedulerError("config.yaml must contain an object.")

        return payload


class PostingWindowChecker:
    def __init__(self, timezone: str) -> None:
        self.timezone = timezone

    def is_allowed_now(
        self,
        allowed_days: list[str],
        windows: list[dict[str, str]],
    ) -> bool:
        now = datetime.now(ZoneInfo(self.timezone))
        weekday = now.strftime("%A").lower()

        if weekday not in allowed_days:
            return False

        current_time = now.time()

        for window in windows:
            start_time = self._parse_time(window["start"])
            end_time = self._parse_time(window["end"])

            if current_time >= start_time and current_time <= end_time:
                return True

        return False

    def _parse_time(self, value: str) -> time:
        parts = value.split(":")

        if len(parts) != 2:
            raise SchedulerError("Invalid time format: " + value)

        hour = int(parts[0])
        minute = int(parts[1])

        return time(hour=hour, minute=minute)


class DailyLimitChecker:
    def __init__(self, timezone: str) -> None:
        self.timezone = timezone

    def assert_post_allowed(
        self,
        posts: list[dict[str, Any]],
        platform: str,
        max_posts_per_day: int,
    ) -> None:
        today = datetime.now(ZoneInfo(self.timezone)).date().isoformat()
        count = 0

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

            if post.get("status") != "published":
                continue

            published_at = post.get("published_at")

            if isinstance(published_at, str) is False:
                continue

            if published_at.startswith(today):
                count += 1

        if count >= max_posts_per_day:
            raise SchedulerError("Daily post limit reached for platform: " + platform)


class DailyFailureLimitChecker:
    def __init__(self, timezone: str) -> None:
        self.timezone = timezone

    def assert_failure_limit_not_reached(
        self,
        posts: list[dict[str, Any]],
        platform: str,
        max_failures_per_day: int,
    ) -> None:
        today = datetime.now(ZoneInfo(self.timezone)).date().isoformat()
        count = 0

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

            failed_at = post.get("failed_at")

            if isinstance(failed_at, str) is False:
                continue

            if failed_at.startswith(today):
                count += 1

        if count >= max_failures_per_day:
            raise SchedulerError("Daily failure limit reached for platform: " + platform)


class KillSwitchChecker:
    def assert_allowed(self, raw_config: dict[str, Any], platform: str) -> None:
        kill_switch = raw_config.get("kill_switch")

        if isinstance(kill_switch, dict) is False:
            raise SchedulerError("Missing kill_switch configuration.")

        enabled = kill_switch.get("enabled")

        if enabled is True:
            reason = kill_switch.get("reason")
            raise SchedulerError("Global kill-switch enabled: " + str(reason))

        platform_kill_switch = raw_config.get("platform_kill_switch")

        if isinstance(platform_kill_switch, dict) is False:
            raise SchedulerError("Missing platform_kill_switch configuration.")

        platform_config = platform_kill_switch.get(platform)

        if isinstance(platform_config, dict) is False:
            raise SchedulerError("Missing platform kill-switch: " + platform)

        platform_enabled = platform_config.get("enabled")

        if platform_enabled is True:
            reason = platform_config.get("reason")
            raise SchedulerError("Platform kill-switch enabled for " + platform + ": " + str(reason))


class QueueSelector:
    def has_approved_post(
        self,
        posts: list[dict[str, Any]],
        platform: str,
    ) -> bool:
        for post in posts:
            if post.get("platform") != platform:
                continue

            if post.get("status") == "approved":
                return True

        return False


class SchedulerService:
    def __init__(
        self,
        root_dir: Path,
        store: SchedulerStore,
        config_store: SchedulerConfigStore,
        publisher_service: PublisherService,
    ) -> None:
        self.root_dir = root_dir
        self.store = store
        self.config_store = config_store
        self.publisher_service = publisher_service
        self.kill_switch_checker = KillSwitchChecker()
        self.queue_selector = QueueSelector()

    def run(self) -> None:
        raw_config = self.config_store.load()
        posts = self.store.load_posts()

        timezone = self._read_timezone(raw_config)

        platforms = self._read_platforms(raw_config)

        for platform in platforms:
            try:
                self._run_platform(
                    raw_config=raw_config,
                    posts=posts,
                    platform=platform,
                    timezone=timezone,
                )
            except SchedulerError as exception:
                print("SCHEDULER_SKIP")
                print(platform)
                print(str(exception))

    def _run_platform(
        self,
        raw_config: dict[str, Any],
        posts: list[dict[str, Any]],
        platform: str,
        timezone: str,
    ) -> None:
        self.kill_switch_checker.assert_allowed(raw_config, platform)

        if self._is_platform_enabled(raw_config, platform) is False:
            raise SchedulerError("Platform is disabled.")

        allowed_days = self._read_posting_days(raw_config, platform)
        windows = self._read_posting_windows(raw_config, platform)

        posting_window_checker = PostingWindowChecker(timezone)

        if posting_window_checker.is_allowed_now(allowed_days, windows) is False:
            raise SchedulerError("Outside posting window.")

        limits = self._read_limits(raw_config)

        daily_limit_checker = DailyLimitChecker(timezone)
        daily_limit_checker.assert_post_allowed(
            posts=posts,
            platform=platform,
            max_posts_per_day=limits["max_posts_per_day"],
        )

        failure_limit_checker = DailyFailureLimitChecker(timezone)
        failure_limit_checker.assert_failure_limit_not_reached(
            posts=posts,
            platform=platform,
            max_failures_per_day=limits["max_failures_per_day"],
        )

        if self.queue_selector.has_approved_post(posts, platform) is False:
            raise SchedulerError("No approved post for platform.")

        self.publisher_service.publish_next()

    def _read_timezone(self, raw_config: dict[str, Any]) -> str:
        app = raw_config.get("app")

        if isinstance(app, dict) is True:
            timezone = app.get("timezone")

            if isinstance(timezone, str) is True:
                return timezone

        timezone = raw_config.get("timezone")

        if isinstance(timezone, str) is False:
            raise SchedulerError("Missing timezone.")

        return timezone

    def _read_platforms(self, raw_config: dict[str, Any]) -> list[str]:
        platforms = raw_config.get("platforms")

        if isinstance(platforms, dict) is False:
            raise SchedulerError("Missing platforms configuration.")

        result: list[str] = []

        for platform_name in platforms.keys():
            if isinstance(platform_name, str) is False:
                continue

            result.append(platform_name)

        return result

    def _is_platform_enabled(self, raw_config: dict[str, Any], platform: str) -> bool:
        platforms = raw_config.get("platforms")

        if isinstance(platforms, dict) is False:
            raise SchedulerError("Missing platforms configuration.")

        platform_config = platforms.get(platform)

        if isinstance(platform_config, dict) is False:
            raise SchedulerError("Missing platform configuration: " + platform)

        enabled = platform_config.get("enabled")

        if isinstance(enabled, bool) is False:
            raise SchedulerError("platform.enabled must be boolean for: " + platform)

        return enabled

    def _read_posting_days(self, raw_config: dict[str, Any], platform: str) -> list[str]:
        posting_days = raw_config.get("posting_days")

        if isinstance(posting_days, dict) is False:
            raise SchedulerError("Missing posting_days configuration.")

        platform_days = posting_days.get(platform)

        if isinstance(platform_days, list) is False:
            raise SchedulerError("Missing posting days for platform: " + platform)

        result: list[str] = []

        for day in platform_days:
            if isinstance(day, str) is False:
                raise SchedulerError("Posting day must be string.")

            result.append(day)

        return result

    def _read_posting_windows(self, raw_config: dict[str, Any], platform: str) -> list[dict[str, str]]:
        posting_windows = raw_config.get("posting_windows")

        if isinstance(posting_windows, dict) is False:
            raise SchedulerError("Missing posting_windows configuration.")

        platform_windows = posting_windows.get(platform)

        if isinstance(platform_windows, list) is False:
            raise SchedulerError("Missing posting windows for platform: " + platform)

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

        for item in platform_windows:
            if isinstance(item, dict) is False:
                raise SchedulerError("Posting window must be object.")

            start = item.get("start")
            end = item.get("end")

            if isinstance(start, str) is False:
                raise SchedulerError("Posting window start must be string.")

            if isinstance(end, str) is False:
                raise SchedulerError("Posting window end must be string.")

            result.append(
                {
                    "start": start,
                    "end": end,
                }
            )

        return result

    def _read_limits(self, raw_config: dict[str, Any]) -> dict[str, int]:
        limits = raw_config.get("limits")

        if isinstance(limits, dict) is False:
            raise SchedulerError("Missing limits configuration.")

        max_posts_per_day = limits.get("max_posts_per_day")
        max_failures_per_day = limits.get("max_failures_per_day")

        if isinstance(max_posts_per_day, int) is False:
            raise SchedulerError("limits.max_posts_per_day must be integer.")

        if isinstance(max_failures_per_day, int) is False:
            raise SchedulerError("limits.max_failures_per_day must be integer.")

        return {
            "max_posts_per_day": max_posts_per_day,
            "max_failures_per_day": max_failures_per_day,
        }


def build_service(root_dir: Path, dry_run: bool) -> SchedulerService:
    store = SchedulerStore(root_dir / "data" / "post_drafts.json")
    config_store = SchedulerConfigStore(root_dir / "config.yaml")
    publisher_service = build_publisher_service(root_dir, dry_run)

    return SchedulerService(
        root_dir=root_dir,
        store=store,
        config_store=config_store,
        publisher_service=publisher_service,
    )


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("command")
    parser.add_argument("--live", action="store_true")

    args = parser.parse_args()

    if args.command != "run":
        raise SchedulerError("Unsupported command: " + args.command)

    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.run()


if __name__ == "__main__":
    main()

Aufrufe

Dry-Run:

python -m src.scheduler run

Live:

python -m src.scheduler run --live

20.8 Beispiel-Crontab

Crontab öffnen:

crontab -e

Beispiel für Dry-Run-Testbetrieb:

SHELL=/bin/bash
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin

0 8 * * 1-5 cd /opt/social-publisher && .venv/bin/python -m src.generate_posts >> storage/logs/cron.log 2>&1
10 8 * * 1-5 cd /opt/social-publisher && .venv/bin/python -m src.review_queue auto-approve >> storage/logs/cron.log 2>&1
*/15 8-18 * * 1-5 cd /opt/social-publisher && .venv/bin/python -m src.scheduler run >> storage/logs/cron.log 2>&1
0 17 * * 1-5 cd /opt/social-publisher && .venv/bin/python -m src.interaction_worker analyze >> storage/logs/cron.log 2>&1

Live-Betrieb:

SHELL=/bin/bash
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin

0 8 * * 1-5 cd /opt/social-publisher && .venv/bin/python -m src.generate_posts >> storage/logs/cron.log 2>&1
10 8 * * 1-5 cd /opt/social-publisher && .venv/bin/python -m src.review_queue auto-approve >> storage/logs/cron.log 2>&1
*/15 8-18 * * 1-5 cd /opt/social-publisher && .venv/bin/python -m src.scheduler run --live >> storage/logs/cron.log 2>&1
0 17 * * 1-5 cd /opt/social-publisher && .venv/bin/python -m src.interaction_worker analyze >> storage/logs/cron.log 2>&1

Mit Lockfile gegen parallele Läufe:

*/15 8-18 * * 1-5 flock -n /tmp/social-publisher.lock bash -lc 'cd /opt/social-publisher && .venv/bin/python -m src.scheduler run --live' >> /opt/social-publisher/storage/logs/cron.log 2>&1

Ergebnis dieses Kapitels

Der Scheduler kann jetzt:

per Cron gestartet werden
Posting-Zeitfenster prüfen
Posting-Tage prüfen
Kill-Switches beachten
Tageslimits prüfen
Fehlerlimits prüfen
freigegebene Beiträge aus der Queue verarbeiten
Dry-Run und Live-Betrieb trennen
parallele Läufe per Lockfile verhindern

Damit ist der automatische Tagesbetrieb vorbereitet.

0 Kommentare