Artikel
13. Posting-Engine
Die Posting-Engine verarbeitet freigegebene Beiträge und übergibt sie an den passenden Plattformadapter.
Sie erzeugt keine Texte. Sie entscheidet auch nicht über Qualität. Sie veröffentlicht nur Beiträge, deren Status bereits approved ist.
approved post
↓
Publisher
↓
PlatformAdapter
↓
Playwright
↓
Screenshot
↓
published oder failed
13.1 Gemeinsames Interface
Jede Plattform wird über dasselbe Interface angesprochen.
from dataclasses import dataclass
from pathlib import Path
@dataclass(frozen=True)
class PublishRequest:
id: str
platform: str
post_type: str
text: str
image_path: str | None
target_url: str | None
storage_state_path: Path
screenshot_dir: Path
dry_run: bool
@dataclass(frozen=True)
class PublishResult:
post_id: str
platform: str
status: str
screenshot_path: str
platform_post_url: str | None
error_message: str | None
class PlatformAdapter:
def publish(self, request: PublishRequest) -> PublishResult:
raise NotImplementedError
Statuswerte:
published
failed
blocked
dry_run
13.2 Plattformadapter
Die Posting-Engine kennt nur den Plattformnamen. Die konkrete Bedienung liegt im Adapter.
class PlatformAdapterRegistry:
def __init__(self) -> None:
self.adapters: dict[str, PlatformAdapter] = {}
def register(self, platform: str, adapter: PlatformAdapter) -> None:
self.adapters[platform] = adapter
def get(self, platform: str) -> PlatformAdapter:
adapter = self.adapters.get(platform)
if adapter is None:
raise PublisherError("No adapter registered for platform: " + platform)
return adapter
Beispielregistrierung:
registry = PlatformAdapterRegistry()
registry.register("linkedin", LinkedInAdapter())
registry.register("facebook", FacebookAdapter())
registry.register("instagram", InstagramAdapter())
13.3 Text einfügen
Das Einfügen des Textes erfolgt im Adapter.
Grundmuster:
page.get_by_role("textbox").fill(request.text)
Viele Social-Media-Editoren sind aber keine klassischen Textfelder, sondern contenteditable-Elemente. Dann ist fill() nicht immer zuverlässig.
Robuster:
textbox = page.get_by_role("textbox").first
textbox.click()
page.keyboard.insert_text(request.text)
Falls Zeilenumbrüche nicht sauber übernommen werden:
textbox.click()
for character in request.text:
if character == "\n":
page.keyboard.press("Shift+Enter")
else:
page.keyboard.insert_text(character)
Für das MVP reicht meistens:
textbox.click()
page.keyboard.insert_text(request.text)
13.4 Bild hochladen
Bild-Upload erfolgt über ein input[type=file].
Beispiel:
file_input = page.locator("input[type=file]").first
file_input.set_input_files(request.image_path)
Vorher prüfen:
from pathlib import Path
class UploadFileValidator:
def validate(self, image_path: str | None) -> None:
if image_path is None:
return
path = Path(image_path)
if path.exists() is False:
raise PublisherError("Image file does not exist: " + image_path)
if path.is_file() is False:
raise PublisherError("Image path is not a file: " + image_path)
allowed_suffixes = [
".jpg",
".jpeg",
".png",
".webp",
]
if path.suffix.lower() not in allowed_suffixes:
raise PublisherError("Unsupported image file type: " + image_path)
Bei Instagram ist ein Bild Pflicht. Bei LinkedIn und Facebook optional.
13.5 Link hinzufügen
Links werden als Teil des Textes angehängt oder in einem eigenen Feld eingefügt, wenn die Plattform das vorsieht.
Für das MVP:
class PostTextBuilder:
def build(self, text: str, target_url: str | None, include_link: bool) -> str:
if include_link is False:
return text
if target_url is None:
return text
return text.rstrip() + "\n\n" + target_url
Regel:
LinkedIn: Link optional anhängen
Facebook: Link optional anhängen
Instagram: Link nicht automatisch in Caption anhängen
Kommentare: nie Link anhängen
13.6 Post absenden
Der Klick auf „Posten“ ist der kritischste Schritt.
Grundregel:
Nach dem Klick nicht blind erneut klicken.
Ablauf:
Text einfügen
Bild hochladen
Dry-Run prüfen
Screenshot vor Absenden optional speichern
Post-Button klicken
auf UI-Reaktion warten
Screenshot speichern
Status setzen
Beispiel:
submit_button = page.get_by_role("button", name="Posten")
submit_button.click(timeout=10_000)
Bei unklarem Status:
nicht erneut klicken
Screenshot speichern
Status failed
manuelle Prüfung
13.7 Screenshot nach Veröffentlichung
Nach jedem Veröffentlichungsversuch wird ein Screenshot gespeichert.
class PublisherScreenshotService:
def save(self, page: Page, screenshot_dir: Path, post_id: str, status: str) -> Path:
screenshot_dir.mkdir(parents=True, exist_ok=True)
screenshot_path = screenshot_dir / (post_id + "_" + status + ".png")
page.screenshot(
path=str(screenshot_path),
full_page=True,
)
return screenshot_path
Dateinamen:
storage/screenshots/linkedin/draft_20260512_001_published.png
storage/screenshots/linkedin/draft_20260512_001_failed.png
storage/screenshots/linkedin/draft_20260512_001_dry_run.png
13.8 Ergebnis speichern
Die Engine aktualisiert den Beitrag nach dem Veröffentlichungsversuch.
Bei Erfolg:
{
"status": "published",
"published_at": "2026-05-12T10:00:00+02:00",
"screenshot_path": "storage/screenshots/linkedin/draft_20260512_001_published.png",
"platform_post_url": null,
"error_message": null
}
Bei Fehler:
{
"status": "failed",
"failed_at": "2026-05-12T10:00:00+02:00",
"screenshot_path": "storage/screenshots/linkedin/draft_20260512_001_failed.png",
"platform_post_url": null,
"error_message": "Post button not found."
}
Repository-Methode:
class PublishedResultApplier:
def apply(self, draft: dict[str, object], result: PublishResult) -> dict[str, object]:
draft["status"] = result.status
draft["screenshot_path"] = result.screenshot_path
draft["platform_post_url"] = result.platform_post_url
draft["error_message"] = result.error_message
draft["updated_at"] = self._now()
if result.status == "published":
draft["published_at"] = self._now()
if result.status == "failed":
draft["failed_at"] = self._now()
if result.status == "blocked":
draft["blocked_at"] = self._now()
return draft
def _now(self) -> str:
from datetime import datetime
from zoneinfo import ZoneInfo
return datetime.now(ZoneInfo("Europe/Berlin")).isoformat()
13.9 Beispiel: publisher.py
Datei:
src/publisher.py
import argparse
import json
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo
from playwright.sync_api import Page
from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
from playwright.sync_api import sync_playwright
class PublisherError(Exception):
pass
class PublisherBlockedError(PublisherError):
pass
@dataclass(frozen=True)
class PublishRequest:
id: str
platform: str
post_type: str
text: str
image_path: str | None
target_url: str | None
storage_state_path: Path
screenshot_dir: Path
dry_run: bool
@dataclass(frozen=True)
class PublishResult:
post_id: str
platform: str
status: str
screenshot_path: str
platform_post_url: str | None
error_message: str | None
class PlatformAdapter:
def publish(self, request: PublishRequest) -> PublishResult:
raise NotImplementedError
class PublisherScreenshotService:
def save(self, page: Page, screenshot_dir: Path, post_id: str, status: str) -> Path:
screenshot_dir.mkdir(parents=True, exist_ok=True)
screenshot_path = screenshot_dir / (post_id + "_" + status + ".png")
page.screenshot(
path=str(screenshot_path),
full_page=True,
)
return screenshot_path
class UploadFileValidator:
def validate(self, image_path: str | None) -> None:
if image_path is None:
return
path = Path(image_path)
if path.exists() is False:
raise PublisherError("Image file does not exist: " + image_path)
if path.is_file() is False:
raise PublisherError("Image path is not a file: " + image_path)
allowed_suffixes = [
".jpg",
".jpeg",
".png",
".webp",
]
if path.suffix.lower() not in allowed_suffixes:
raise PublisherError("Unsupported image file type: " + image_path)
class PostTextBuilder:
def build(self, platform: str, text: str, target_url: str | None) -> str:
if platform == "instagram":
return text
if target_url is None:
return text
return text.rstrip() + "\n\n" + target_url
class LinkedInAdapter(PlatformAdapter):
def __init__(self) -> None:
self.screenshot_service = PublisherScreenshotService()
self.upload_file_validator = UploadFileValidator()
self.post_text_builder = PostTextBuilder()
def publish(self, request: PublishRequest) -> PublishResult:
self._validate_request(request)
with sync_playwright() as playwright:
browser = playwright.chromium.launch(headless=True)
context = browser.new_context(storage_state=str(request.storage_state_path))
page = context.new_page()
page.set_default_timeout(10_000)
page.set_default_navigation_timeout(30_000)
try:
page.goto(
"https://www.linkedin.com/feed/",
wait_until="domcontentloaded",
timeout=30_000,
)
self._assert_logged_in(page)
page.get_by_role("button", name="Beitrag beginnen").click(timeout=10_000)
textbox = page.get_by_role("textbox").first
textbox.click(timeout=10_000)
post_text = self.post_text_builder.build(
platform=request.platform,
text=request.text,
target_url=request.target_url,
)
page.keyboard.insert_text(post_text)
if request.image_path is not None:
self._upload_image(page, request.image_path)
if request.dry_run is True:
screenshot_path = self.screenshot_service.save(
page=page,
screenshot_dir=request.screenshot_dir,
post_id=request.id,
status="dry_run",
)
browser.close()
return PublishResult(
post_id=request.id,
platform=request.platform,
status="dry_run",
screenshot_path=str(screenshot_path),
platform_post_url=None,
error_message=None,
)
page.get_by_role("button", name="Posten").click(timeout=10_000)
page.wait_for_timeout(3000)
screenshot_path = self.screenshot_service.save(
page=page,
screenshot_dir=request.screenshot_dir,
post_id=request.id,
status="published",
)
browser.close()
return PublishResult(
post_id=request.id,
platform=request.platform,
status="published",
screenshot_path=str(screenshot_path),
platform_post_url=None,
error_message=None,
)
except PublisherBlockedError as exception:
screenshot_path = self.screenshot_service.save(
page=page,
screenshot_dir=request.screenshot_dir,
post_id=request.id,
status="blocked",
)
browser.close()
return PublishResult(
post_id=request.id,
platform=request.platform,
status="blocked",
screenshot_path=str(screenshot_path),
platform_post_url=None,
error_message=str(exception),
)
except PlaywrightTimeoutError as exception:
screenshot_path = self.screenshot_service.save(
page=page,
screenshot_dir=request.screenshot_dir,
post_id=request.id,
status="failed",
)
browser.close()
return PublishResult(
post_id=request.id,
platform=request.platform,
status="failed",
screenshot_path=str(screenshot_path),
platform_post_url=None,
error_message=str(exception),
)
except PublisherError as exception:
screenshot_path = self.screenshot_service.save(
page=page,
screenshot_dir=request.screenshot_dir,
post_id=request.id,
status="failed",
)
browser.close()
return PublishResult(
post_id=request.id,
platform=request.platform,
status="failed",
screenshot_path=str(screenshot_path),
platform_post_url=None,
error_message=str(exception),
)
def _validate_request(self, request: PublishRequest) -> None:
if request.storage_state_path.exists() is False:
raise PublisherError("Storage state file does not exist: " + str(request.storage_state_path))
if request.text.strip() == "":
raise PublisherError("Post text must not be empty.")
self.upload_file_validator.validate(request.image_path)
def _assert_logged_in(self, page: Page) -> None:
current_url = page.url.lower()
if "login" in current_url:
raise PublisherBlockedError("Login required.")
body_text = page.locator("body").inner_text(timeout=10_000).lower()
blocked_fragments = [
"captcha",
"sicherheitsüberprüfung",
"security check",
"unusual activity",
]
for blocked_fragment in blocked_fragments:
if blocked_fragment in body_text:
raise PublisherBlockedError("Blocked platform state detected: " + blocked_fragment)
def _upload_image(self, page: Page, image_path: str) -> None:
image_button_names = [
"Bild hinzufügen",
"Medien hinzufügen",
"Foto hinzufügen",
]
clicked = False
for button_name in image_button_names:
locator = page.get_by_role("button", name=button_name)
if locator.count() > 0:
locator.first.click(timeout=10_000)
clicked = True
break
if clicked is False:
raise PublisherError("Image upload button not found.")
file_input = page.locator("input[type=file]").first
file_input.set_input_files(image_path)
class FacebookAdapter(PlatformAdapter):
def publish(self, request: PublishRequest) -> PublishResult:
return PublishResult(
post_id=request.id,
platform=request.platform,
status="failed",
screenshot_path="",
platform_post_url=None,
error_message="FacebookAdapter is not implemented yet.",
)
class InstagramAdapter(PlatformAdapter):
def publish(self, request: PublishRequest) -> PublishResult:
return PublishResult(
post_id=request.id,
platform=request.platform,
status="failed",
screenshot_path="",
platform_post_url=None,
error_message="InstagramAdapter is not implemented yet.",
)
class PlatformAdapterRegistry:
def __init__(self) -> None:
self.adapters: dict[str, PlatformAdapter] = {}
def register(self, platform: str, adapter: PlatformAdapter) -> None:
self.adapters[platform] = adapter
def get(self, platform: str) -> PlatformAdapter:
adapter = self.adapters.get(platform)
if adapter is None:
raise PublisherError("No adapter registered for platform: " + platform)
return adapter
class ReviewQueueStore:
def __init__(self, file_path: Path) -> None:
self.file_path = file_path
def load(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 PublisherError("Post draft JSON must contain a list.")
drafts: list[dict[str, Any]] = []
for item in payload:
if isinstance(item, dict) is False:
raise PublisherError("Post draft JSON contains non-object item.")
drafts.append(item)
return drafts
def save(self, drafts: list[dict[str, Any]]) -> None:
self.file_path.parent.mkdir(parents=True, exist_ok=True)
with self.file_path.open("w", encoding="utf-8") as file:
json.dump(drafts, file, ensure_ascii=False, indent=2)
file.write("\n")
class ReviewQueueRepository:
def __init__(self, store: ReviewQueueStore) -> None:
self.store = store
def find_by_id(self, post_id: str) -> dict[str, Any]:
drafts = self.store.load()
for draft in drafts:
if draft.get("id") == post_id:
return draft
raise PublisherError("Post not found: " + post_id)
def update(self, updated_post: dict[str, Any]) -> None:
drafts = self.store.load()
updated_drafts: list[dict[str, Any]] = []
found = False
updated_id = updated_post.get("id")
for draft in drafts:
if draft.get("id") == updated_id:
updated_drafts.append(updated_post)
found = True
else:
updated_drafts.append(draft)
if found is False:
raise PublisherError("Post not found: " + str(updated_id))
self.store.save(updated_drafts)
def list_approved(self) -> list[dict[str, Any]]:
drafts = self.store.load()
result: list[dict[str, Any]] = []
for draft in drafts:
if draft.get("status") == "approved":
result.append(draft)
return result
class PublishedResultApplier:
def apply(self, draft: dict[str, Any], result: PublishResult) -> dict[str, Any]:
draft["status"] = result.status
draft["screenshot_path"] = result.screenshot_path
draft["platform_post_url"] = result.platform_post_url
draft["error_message"] = result.error_message
draft["updated_at"] = self._now()
if result.status == "published":
draft["published_at"] = self._now()
if result.status == "failed":
draft["failed_at"] = self._now()
if result.status == "blocked":
draft["blocked_at"] = self._now()
if result.status == "dry_run":
draft["dry_run_at"] = self._now()
return draft
def _now(self) -> str:
return datetime.now(ZoneInfo("Europe/Berlin")).isoformat()
class PublishRequestFactory:
def __init__(self, root_dir: Path, dry_run: bool) -> None:
self.root_dir = root_dir
self.dry_run = dry_run
def create(self, draft: dict[str, Any]) -> PublishRequest:
post_id = self._read_string(draft, "id")
platform = self._read_string(draft, "platform")
post_type = self._read_string(draft, "post_type")
text = self._read_string(draft, "text")
image_path = self._read_optional_string(draft, "image_path")
target_url = self._read_optional_string(draft, "target_url")
return PublishRequest(
id=post_id,
platform=platform,
post_type=post_type,
text=text,
image_path=image_path,
target_url=target_url,
storage_state_path=self.root_dir / "storage" / "sessions" / (platform + ".json"),
screenshot_dir=self.root_dir / "storage" / "screenshots" / platform,
dry_run=self.dry_run,
)
def _read_string(self, payload: dict[str, Any], key: str) -> str:
value = payload.get(key)
if isinstance(value, str) is False:
raise PublisherError("Field must be string: " + key)
if value.strip() == "":
raise PublisherError("Field must not be empty: " + key)
return value
def _read_optional_string(self, payload: dict[str, Any], key: str) -> str | None:
value = payload.get(key)
if value is None:
return None
if isinstance(value, str) is False:
raise PublisherError("Field must be string or null: " + key)
if value.strip() == "":
return None
return value
class PublisherService:
def __init__(
self,
repository: ReviewQueueRepository,
registry: PlatformAdapterRegistry,
request_factory: PublishRequestFactory,
result_applier: PublishedResultApplier,
) -> None:
self.repository = repository
self.registry = registry
self.request_factory = request_factory
self.result_applier = result_applier
def publish_one(self, post_id: str) -> None:
draft = self.repository.find_by_id(post_id)
status = draft.get("status")
if status != "approved":
raise PublisherError("Post must have status approved. Current status: " + str(status))
request = self.request_factory.create(draft)
adapter = self.registry.get(request.platform)
result = adapter.publish(request)
updated_draft = self.result_applier.apply(draft, result)
self.repository.update(updated_draft)
print(result.status.upper())
print(result.post_id)
if result.error_message is not None:
print(result.error_message)
def publish_next(self) -> None:
approved_posts = self.repository.list_approved()
if len(approved_posts) == 0:
print("NO_APPROVED_POSTS")
return
first_post = approved_posts[0]
post_id = first_post.get("id")
if isinstance(post_id, str) is False:
raise PublisherError("Approved post id must be a string.")
self.publish_one(post_id)
def build_registry() -> PlatformAdapterRegistry:
registry = PlatformAdapterRegistry()
registry.register("linkedin", LinkedInAdapter())
registry.register("facebook", FacebookAdapter())
registry.register("instagram", InstagramAdapter())
return registry
def build_service(root_dir: Path, dry_run: bool) -> PublisherService:
store = ReviewQueueStore(root_dir / "data" / "post_drafts.json")
repository = ReviewQueueRepository(store)
registry = build_registry()
request_factory = PublishRequestFactory(root_dir, dry_run)
result_applier = PublishedResultApplier()
return PublisherService(
repository=repository,
registry=registry,
request_factory=request_factory,
result_applier=result_applier,
)
def main() -> None:
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="command", required=True)
publish_parser = subparsers.add_parser("publish")
publish_parser.add_argument("post_id")
publish_parser.add_argument("--live", action="store_true")
next_parser = subparsers.add_parser("publish-next")
next_parser.add_argument("--live", action="store_true")
args = parser.parse_args()
root_dir = Path(__file__).resolve().parent.parent
dry_run = True
if args.command == "publish":
if args.live is True:
dry_run = False
service = build_service(root_dir, dry_run)
service.publish_one(args.post_id)
return
if args.command == "publish-next":
if args.live is True:
dry_run = False
service = build_service(root_dir, dry_run)
service.publish_next()
return
raise PublisherError("Unsupported command: " + str(args.command))
if __name__ == "__main__":
main()
Aufrufe
Dry-Run für einen Beitrag:
python -m src.publisher publish draft_20260512_001
Live-Veröffentlichung:
python -m src.publisher publish draft_20260512_001 --live
Nächsten freigegebenen Beitrag im Dry-Run testen:
python -m src.publisher publish-next
Nächsten freigegebenen Beitrag live veröffentlichen:
python -m src.publisher publish-next --live
Ergebnis dieses Kapitels
Die Posting-Engine kann jetzt:
freigegebene Beiträge laden
Plattformadapter auswählen
Text einfügen
Bilder hochladen
Links anhängen
Dry-Run ausführen
Post absenden
Screenshots speichern
Ergebnis in der Queue speichern
Status auf dry_run, published, failed oder blocked setzen
Die Plattformlogik bleibt in Adapterklassen. Der Publisher selbst bleibt generisch.
0 Kommentare