Hitonabi
28a12b2e06
Vorher gab es KEINEN Code-Pfad, der je einen Rip ausgelöst hat: kein POST /jobs, kein udev-Daemon (udev_daemon.py existierte nirgends), GET /jobs gab hart [] zurück, Postgres lag komplett brach, der SSE-Stream konnte strukturell nie senden (sse_connections wurde nie befüllt), udevadm lieferte ohne udevd nichts. - POST /jobs: legt Job-Zeile an, schickt worker.tasks.rip_disc via Celery - GET /jobs aus Postgres (running→processing fürs UI) - Disc-Watcher: 3s-ioctl-Poll statt udev, protokolliert Einwurf/Auswurf - /devices über /sys (vendor/model) + ioctl-Status — ehrlich statt leer - /logs + /settings (Settings-Seite sprach vorher gegen 404) - SSE-Fix, udev aus dem API-Image entfernt, Import-Smoke-Test Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
141 lines
3.8 KiB
Python
141 lines
3.8 KiB
Python
"""Job-, Log- und Settings-Persistenz in PostgreSQL (KONZEPT: Postgres für Job-Logs).
|
|
|
|
Die jobs/logs-Tabellendefinition existiert bewusst identisch im Worker
|
|
(docker/worker/db.py) — es gibt kein geteiltes Paket zwischen den Containern.
|
|
Wer die Struktur ändert, ändert BEIDE Dateien. create_all ist idempotent.
|
|
|
|
Bis 23.07. lag Postgres komplett brach: GET /jobs gab hart [] zurück, nichts
|
|
schrieb je eine Zeile — der Job-Verlauf im UI war ein Placebo.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import (
|
|
Column,
|
|
DateTime,
|
|
Integer,
|
|
MetaData,
|
|
String,
|
|
Table,
|
|
Text,
|
|
create_engine,
|
|
select,
|
|
)
|
|
|
|
DATABASE_URL = os.getenv(
|
|
"DATABASE_URL", "postgresql://rippy:rippy@localhost:5432/rippy"
|
|
)
|
|
|
|
engine = create_engine(DATABASE_URL, pool_pre_ping=True)
|
|
metadata = MetaData()
|
|
|
|
jobs = Table(
|
|
"jobs",
|
|
metadata,
|
|
Column("id", String(36), primary_key=True),
|
|
Column("disc_type", String(16)),
|
|
Column("device", String(64)),
|
|
Column("title", String(255)),
|
|
Column("status", String(16), nullable=False, server_default="pending"),
|
|
Column("progress", Integer, nullable=False, server_default="0"),
|
|
Column("output_path", Text),
|
|
Column("error", Text),
|
|
Column("created_at", DateTime(timezone=True)),
|
|
Column("finished_at", DateTime(timezone=True)),
|
|
)
|
|
|
|
logs = Table(
|
|
"logs",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True, autoincrement=True),
|
|
Column("ts", DateTime(timezone=True)),
|
|
Column("level", String(16)),
|
|
Column("source", String(32)),
|
|
Column("message", Text),
|
|
)
|
|
|
|
settings_table = Table(
|
|
"settings",
|
|
metadata,
|
|
Column("key", String(64), primary_key=True),
|
|
Column("value", Text),
|
|
)
|
|
|
|
|
|
def utcnow() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def init_db() -> None:
|
|
"""Legt fehlende Tabellen an (idempotent)."""
|
|
metadata.create_all(engine)
|
|
|
|
|
|
def insert_job(job_id: str, device: str, disc_type: str = None, title: str = None) -> None:
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
jobs.insert().values(
|
|
id=job_id,
|
|
device=device,
|
|
disc_type=disc_type,
|
|
title=title,
|
|
status="pending",
|
|
progress=0,
|
|
created_at=utcnow(),
|
|
)
|
|
)
|
|
|
|
|
|
def list_jobs(limit: int = 100) -> list:
|
|
with engine.connect() as conn:
|
|
zeilen = conn.execute(
|
|
select(jobs).order_by(jobs.c.created_at.desc()).limit(limit)
|
|
).mappings().all()
|
|
return [dict(z) for z in zeilen]
|
|
|
|
|
|
def add_log(level: str, source: str, message: str) -> None:
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
logs.insert().values(ts=utcnow(), level=level, source=source, message=message)
|
|
)
|
|
|
|
|
|
def list_logs(limit: int = 200) -> list:
|
|
with engine.connect() as conn:
|
|
zeilen = conn.execute(
|
|
select(logs).order_by(logs.c.id.desc()).limit(limit)
|
|
).mappings().all()
|
|
return [dict(z) for z in zeilen]
|
|
|
|
|
|
def get_settings(key: str = "ui") -> dict:
|
|
with engine.connect() as conn:
|
|
zeile = conn.execute(
|
|
select(settings_table.c.value).where(settings_table.c.key == key)
|
|
).first()
|
|
if not zeile or not zeile[0]:
|
|
return {}
|
|
try:
|
|
return json.loads(zeile[0])
|
|
except ValueError:
|
|
return {}
|
|
|
|
|
|
def save_settings(werte: dict, key: str = "ui") -> None:
|
|
payload = json.dumps(werte)
|
|
with engine.begin() as conn:
|
|
vorhanden = conn.execute(
|
|
select(settings_table.c.key).where(settings_table.c.key == key)
|
|
).first()
|
|
if vorhanden:
|
|
conn.execute(
|
|
settings_table.update()
|
|
.where(settings_table.c.key == key)
|
|
.values(value=payload)
|
|
)
|
|
else:
|
|
conn.execute(settings_table.insert().values(key=key, value=payload))
|