Artikel
5. Datenmodell
Das Datenmodell bildet die Arbeitsobjekte des Systems ab:
Bücher
Autoren
Beitragsentwürfe
freigegebene Beiträge
veröffentlichte Beiträge
Interaktionskandidaten
Interaktionsprotokolle
Für das MVP reichen JSON-Dateien. Sobald mehrere Plattformen, wiederkehrende Jobs, Reports und Dublettenprüfung dazukommen, ist SQLite sinnvoller.
5.1 Book
Book beschreibt ein Buch als Content-Quelle.
Minimalfelder:
id
title
author_id
language
category
description
shop_url
cover_image_path
tags
created_at
updated_at
Beispiel als JSON:
{
"id": "book_001",
"title": "Stolz und Vorurteil",
"author_id": "author_001",
"language": "de",
"category": "klassiker",
"description": "Ein Gesellschaftsroman über Herkunft, Heirat, Stolz und Selbsttäuschung.",
"shop_url": "https://example.com/stolz-und-vorurteil",
"cover_image_path": "storage/images/stolz-und-vorurteil.jpg",
"tags": [
"klassiker",
"jane-austen",
"gesellschaftsroman"
],
"created_at": "2026-05-11T09:00:00+02:00",
"updated_at": "2026-05-11T09:00:00+02:00"
}
Python-Modell:
from dataclasses import dataclass
from datetime import datetime
@dataclass(frozen=True)
class Book:
id: str
title: str
author_id: str
language: str
category: str
description: str
shop_url: str
cover_image_path: str
tags: list[str]
created_at: datetime
updated_at: datetime
5.2 Author
Author beschreibt den Autor eines Buches.
Minimalfelder:
id
name
birth_year
death_year
country
description
tags
created_at
updated_at
Beispiel:
{
"id": "author_001",
"name": "Jane Austen",
"birth_year": 1775,
"death_year": 1817,
"country": "Großbritannien",
"description": "Englische Schriftstellerin, bekannt für Gesellschaftsromane mit präziser Beobachtung sozialer Verhältnisse.",
"tags": [
"klassiker",
"englische-literatur",
"gesellschaftsroman"
],
"created_at": "2026-05-11T09:00:00+02:00",
"updated_at": "2026-05-11T09:00:00+02:00"
}
Python-Modell:
from dataclasses import dataclass
from datetime import datetime
@dataclass(frozen=True)
class Author:
id: str
name: str
birth_year: int | None
death_year: int | None
country: str
description: str
tags: list[str]
created_at: datetime
updated_at: datetime
5.3 PostDraft
PostDraft ist ein erzeugter, aber noch nicht freigegebener Beitrag.
Minimalfelder:
id
source_type
source_id
platform
post_type
status
text
image_path
target_url
prompt_name
prompt_version
model
quality_result
created_at
updated_at
Beispiel:
{
"id": "draft_20260511_001",
"source_type": "book",
"source_id": "book_001",
"platform": "linkedin",
"post_type": "book_intro",
"status": "draft",
"text": "Jane Austens Stolz und Vorurteil ist weniger eine romantische Komödie als eine genaue Gesellschaftsbeobachtung.",
"image_path": null,
"target_url": "https://example.com/stolz-und-vorurteil",
"prompt_name": "linkedin_book_intro",
"prompt_version": "1.0.0",
"model": "gpt-4.1-mini",
"quality_result": null,
"created_at": "2026-05-11T09:10:00+02:00",
"updated_at": "2026-05-11T09:10:00+02:00"
}
Python-Modell:
from dataclasses import dataclass
from datetime import datetime
from typing import Any
@dataclass(frozen=True)
class PostDraft:
id: str
source_type: str
source_id: str
platform: str
post_type: str
status: str
text: str
image_path: str | None
target_url: str | None
prompt_name: str
prompt_version: str
model: str
quality_result: dict[str, Any] | None
created_at: datetime
updated_at: datetime
5.4 ApprovedPost
ApprovedPost ist ein freigegebener Beitrag. Er darf veröffentlicht werden.
Minimalfelder:
id
draft_id
platform
post_type
text
image_path
target_url
approved_by
approval_mode
approved_at
scheduled_at
status
Beispiel:
{
"id": "approved_20260511_001",
"draft_id": "draft_20260511_001",
"platform": "linkedin",
"post_type": "book_intro",
"text": "Jane Austens Stolz und Vorurteil ist weniger eine romantische Komödie als eine genaue Gesellschaftsbeobachtung.",
"image_path": null,
"target_url": "https://example.com/stolz-und-vorurteil",
"approved_by": "system",
"approval_mode": "manual",
"approved_at": "2026-05-11T09:20:00+02:00",
"scheduled_at": "2026-05-11T09:30:00+02:00",
"status": "approved"
}
Python-Modell:
from dataclasses import dataclass
from datetime import datetime
@dataclass(frozen=True)
class ApprovedPost:
id: str
draft_id: str
platform: str
post_type: str
text: str
image_path: str | None
target_url: str | None
approved_by: str
approval_mode: str
approved_at: datetime
scheduled_at: datetime | None
status: str
5.5 PublishedPost
PublishedPost beschreibt einen tatsächlich veröffentlichten Beitrag.
Minimalfelder:
id
approved_post_id
platform
post_type
text
image_path
target_url
platform_post_url
screenshot_path
published_at
status
log_id
Beispiel:
{
"id": "published_20260511_001",
"approved_post_id": "approved_20260511_001",
"platform": "linkedin",
"post_type": "book_intro",
"text": "Jane Austens Stolz und Vorurteil ist weniger eine romantische Komödie als eine genaue Gesellschaftsbeobachtung.",
"image_path": null,
"target_url": "https://example.com/stolz-und-vorurteil",
"platform_post_url": null,
"screenshot_path": "storage/screenshots/linkedin/post_20260511_001_success.png",
"published_at": "2026-05-11T09:30:00+02:00",
"status": "published",
"log_id": "log_20260511_001"
}
Python-Modell:
from dataclasses import dataclass
from datetime import datetime
@dataclass(frozen=True)
class PublishedPost:
id: str
approved_post_id: str
platform: str
post_type: str
text: str
image_path: str | None
target_url: str | None
platform_post_url: str | None
screenshot_path: str
published_at: datetime
status: str
log_id: str
5.6 InteractionCandidate
InteractionCandidate beschreibt eine mögliche Interaktion, bevor sie ausgeführt wird.
Typische Fälle:
Antwort auf Kommentar unter eigenem Beitrag
Kommentar unter fremdem Beitrag
Like-Kandidat
Repost-Kandidat
Minimalfelder:
id
platform
interaction_type
source_url
source_author
source_text
relevance_score
risk_score
status
suggested_text
created_at
updated_at
Beispiel:
{
"id": "interaction_20260511_001",
"platform": "linkedin",
"interaction_type": "comment_reply",
"source_url": "https://www.linkedin.com/feed/update/example",
"source_author": "Max Beispiel",
"source_text": "Interessanter Punkt zur Sichtbarkeit kleiner Anbieter.",
"relevance_score": 0.82,
"risk_score": 0.12,
"status": "suggested",
"suggested_text": "Der Punkt ist im Buchmarkt ähnlich: Technisch kann ein kleines Angebot sehr gut sein, aber Auffindbarkeit entscheidet oft früher als Qualität.",
"created_at": "2026-05-11T10:00:00+02:00",
"updated_at": "2026-05-11T10:01:00+02:00"
}
Python-Modell:
from dataclasses import dataclass
from datetime import datetime
@dataclass(frozen=True)
class InteractionCandidate:
id: str
platform: str
interaction_type: str
source_url: str
source_author: str
source_text: str
relevance_score: float
risk_score: float
status: str
suggested_text: str | None
created_at: datetime
updated_at: datetime
5.7 InteractionLog
InteractionLog protokolliert eine ausgeführte oder fehlgeschlagene Interaktion.
Minimalfelder:
id
candidate_id
platform
interaction_type
executed_text
status
screenshot_path
executed_at
error_message
Beispiel:
{
"id": "interaction_log_20260511_001",
"candidate_id": "interaction_20260511_001",
"platform": "linkedin",
"interaction_type": "comment_reply",
"executed_text": "Der Punkt ist im Buchmarkt ähnlich: Technisch kann ein kleines Angebot sehr gut sein, aber Auffindbarkeit entscheidet oft früher als Qualität.",
"status": "executed",
"screenshot_path": "storage/screenshots/linkedin/interaction_20260511_001_success.png",
"executed_at": "2026-05-11T10:10:00+02:00",
"error_message": null
}
Python-Modell:
from dataclasses import dataclass
from datetime import datetime
@dataclass(frozen=True)
class InteractionLog:
id: str
candidate_id: str
platform: str
interaction_type: str
executed_text: str | None
status: str
screenshot_path: str | None
executed_at: datetime
error_message: str | None
5.8 JSON-Dateien vs. SQLite
Für den Start reicht JSON.
Vorteile:
leicht lesbar
leicht versionierbar
kein Datenbank-Setup
gut für MVP
direkt manuell korrigierbar
Nachteile:
keine sauberen Abfragen
schwierig bei parallelen Prozessen
keine Transaktionen
Dublettenprüfung umständlicher
Reporting schwächer
SQLite ist sinnvoll, sobald mehr als ein einfacher Ablauf existiert.
Vorteile:
relationale Struktur
Transaktionen
einfache Reports
bessere Statusabfragen
bessere Dublettenprüfung
kein Server nötig
stabiler für Scheduler
Empfohlene Entwicklung:
Phase 1: JSON-Dateien
Phase 2: SQLite
Phase 3: optional PostgreSQL oder MariaDB
Für dieses Projekt ist SQLite der sinnvolle Produktivstandard, sobald automatisches Posting aktiv ist.
JSON-Dateien für MVP
data/authors.json
data/books.json
data/post_drafts.json
data/approved_posts.json
data/published_posts.json
data/interaction_candidates.json
data/interaction_logs.json
Beispiel: data/post_drafts.json
[
{
"id": "draft_20260511_001",
"source_type": "book",
"source_id": "book_001",
"platform": "linkedin",
"post_type": "book_intro",
"status": "draft",
"text": "Jane Austens Stolz und Vorurteil ist weniger eine romantische Komödie als eine genaue Gesellschaftsbeobachtung.",
"image_path": null,
"target_url": "https://example.com/stolz-und-vorurteil",
"prompt_name": "linkedin_book_intro",
"prompt_version": "1.0.0",
"model": "gpt-4.1-mini",
"quality_result": null,
"created_at": "2026-05-11T09:10:00+02:00",
"updated_at": "2026-05-11T09:10:00+02:00"
}
]
Ein einfacher JSON-Store:
import json
from pathlib import Path
from typing import Any
class JsonStore:
def __init__(self, file_path: Path) -> None:
self.file_path = file_path
def load_list(self) -> list[dict[str, Any]]:
if self.file_path.exists() is False:
return []
with self.file_path.open("r", encoding="utf-8") as file:
data = json.load(file)
if isinstance(data, list) is False:
raise RuntimeError(f"JSON file must contain a list: {self.file_path}")
result: list[dict[str, Any]] = []
for item in data:
if isinstance(item, dict) is False:
raise RuntimeError(f"JSON list contains non-object item: {self.file_path}")
result.append(item)
return result
def save_list(self, data: 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(data, file, ensure_ascii=False, indent=2)
file.write("\n")
5.9 Beispiel-Schema SQLite
SQLite-Datei:
data/social_publisher.sqlite
Schema:
CREATE TABLE authors (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
birth_year INTEGER NULL,
death_year INTEGER NULL,
country TEXT NOT NULL,
description TEXT NOT NULL,
tags_json TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE books (
id TEXT PRIMARY KEY,
title TEXT NOT NULL,
author_id TEXT NOT NULL,
language TEXT NOT NULL,
category TEXT NOT NULL,
description TEXT NOT NULL,
shop_url TEXT NOT NULL,
cover_image_path TEXT NOT NULL,
tags_json TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY (author_id) REFERENCES authors (id)
);
CREATE TABLE post_drafts (
id TEXT PRIMARY KEY,
source_type TEXT NOT NULL,
source_id TEXT NOT NULL,
platform TEXT NOT NULL,
post_type TEXT NOT NULL,
status TEXT NOT NULL,
text TEXT NOT NULL,
image_path TEXT NULL,
target_url TEXT NULL,
prompt_name TEXT NOT NULL,
prompt_version TEXT NOT NULL,
model TEXT NOT NULL,
quality_result_json TEXT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE approved_posts (
id TEXT PRIMARY KEY,
draft_id TEXT NOT NULL,
platform TEXT NOT NULL,
post_type TEXT NOT NULL,
text TEXT NOT NULL,
image_path TEXT NULL,
target_url TEXT NULL,
approved_by TEXT NOT NULL,
approval_mode TEXT NOT NULL,
approved_at TEXT NOT NULL,
scheduled_at TEXT NULL,
status TEXT NOT NULL,
FOREIGN KEY (draft_id) REFERENCES post_drafts (id)
);
CREATE TABLE published_posts (
id TEXT PRIMARY KEY,
approved_post_id TEXT NOT NULL,
platform TEXT NOT NULL,
post_type TEXT NOT NULL,
text TEXT NOT NULL,
image_path TEXT NULL,
target_url TEXT NULL,
platform_post_url TEXT NULL,
screenshot_path TEXT NOT NULL,
published_at TEXT NOT NULL,
status TEXT NOT NULL,
log_id TEXT NOT NULL,
FOREIGN KEY (approved_post_id) REFERENCES approved_posts (id)
);
CREATE TABLE interaction_candidates (
id TEXT PRIMARY KEY,
platform TEXT NOT NULL,
interaction_type TEXT NOT NULL,
source_url TEXT NOT NULL,
source_author TEXT NOT NULL,
source_text TEXT NOT NULL,
relevance_score REAL NOT NULL,
risk_score REAL NOT NULL,
status TEXT NOT NULL,
suggested_text TEXT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE interaction_logs (
id TEXT PRIMARY KEY,
candidate_id TEXT NOT NULL,
platform TEXT NOT NULL,
interaction_type TEXT NOT NULL,
executed_text TEXT NULL,
status TEXT NOT NULL,
screenshot_path TEXT NULL,
executed_at TEXT NOT NULL,
error_message TEXT NULL,
FOREIGN KEY (candidate_id) REFERENCES interaction_candidates (id)
);
CREATE TABLE event_logs (
id TEXT PRIMARY KEY,
event TEXT NOT NULL,
level TEXT NOT NULL,
platform TEXT NULL,
reference_id TEXT NULL,
payload_json TEXT NOT NULL,
created_at TEXT NOT NULL
);
Nützliche Indizes:
CREATE INDEX idx_books_author_id
ON books (author_id);
CREATE INDEX idx_post_drafts_status
ON post_drafts (status);
CREATE INDEX idx_post_drafts_platform_status
ON post_drafts (platform, status);
CREATE INDEX idx_approved_posts_status
ON approved_posts (status);
CREATE INDEX idx_approved_posts_scheduled_at
ON approved_posts (scheduled_at);
CREATE INDEX idx_published_posts_platform_published_at
ON published_posts (platform, published_at);
CREATE INDEX idx_interaction_candidates_status
ON interaction_candidates (status);
CREATE INDEX idx_interaction_candidates_platform_status
ON interaction_candidates (platform, status);
CREATE INDEX idx_event_logs_created_at
ON event_logs (created_at);
Initialisierung per Python:
import sqlite3
from pathlib import Path
class SqliteSchemaInstaller:
def __init__(self, database_path: Path, schema_path: Path) -> None:
self.database_path = database_path
self.schema_path = schema_path
def install(self) -> None:
if self.schema_path.exists() is False:
raise RuntimeError(f"Schema file does not exist: {self.schema_path}")
self.database_path.parent.mkdir(parents=True, exist_ok=True)
schema_sql = self.schema_path.read_text(encoding="utf-8")
with sqlite3.connect(self.database_path) as connection:
connection.executescript(schema_sql)
connection.commit()
Beispielkommando:
mkdir -p database
nano database/schema.sql
Dann:
from pathlib import Path
from src.sqlite_schema_installer import SqliteSchemaInstaller
def main() -> None:
root_dir = Path(__file__).resolve().parent.parent
installer = SqliteSchemaInstaller(
database_path=root_dir / "data" / "social_publisher.sqlite",
schema_path=root_dir / "database" / "schema.sql",
)
installer.install()
if __name__ == "__main__":
main()
Empfohlener MVP-Stand
Für den ersten lauffähigen Stand:
JSON verwenden
Dataclasses definieren
Status sauber führen
später SQLite ergänzen
IDs stabil erzeugen
Zeitpunkte immer als ISO-String speichern
Tags und Prüfberichte als JSON speichern
Für den ersten produktiven Automationsbetrieb:
SQLite verwenden
Transaktionen nutzen
Statuswechsel atomar speichern
Indizes für Statusabfragen anlegen
Event-Log zentral führen
Screenshots nur als Dateipfad speichern
0 Kommentare