Artikel
20. Scheduler
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