135 lines
4.0 KiB
Python
135 lines
4.0 KiB
Python
"""Die Runde: wer mitmacht, und wer gerade wer ist.
|
|
|
|
Der Host pflegt die Namensliste. Alle anderen wählen sich einen davon aus.
|
|
Kein Passwort — für acht Freunde, die sich ohnehin über Discord hören, wäre
|
|
das nur eine Hürde ohne Nutzen.
|
|
|
|
Die Zuordnung hängt an derselben Cookie-Sitzung wie Vorschläge und
|
|
Wertungen. Damit steht der Name serverseitig fest und kommt nicht mehr aus
|
|
einem Textfeld, in das jeder schreiben kann, was er will.
|
|
"""
|
|
|
|
import sqlite3
|
|
from datetime import datetime, timezone
|
|
|
|
from .db import connect
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE,
|
|
created_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS user_sessions (
|
|
session_id TEXT PRIMARY KEY,
|
|
user_id INTEGER NOT NULL,
|
|
seit TEXT NOT NULL
|
|
);
|
|
"""
|
|
|
|
|
|
def init():
|
|
with connect() as conn:
|
|
conn.executescript(SCHEMA)
|
|
|
|
|
|
def alle():
|
|
with connect() as conn:
|
|
rows = conn.execute("SELECT * FROM users ORDER BY name COLLATE NOCASE").fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def anlegen(name):
|
|
"""Gibt None zurück, wenn der Name schon vergeben ist."""
|
|
name = (name or "").strip()[:40]
|
|
if not name:
|
|
return None
|
|
try:
|
|
with connect() as conn:
|
|
conn.execute(
|
|
"INSERT INTO users (name, created_at) VALUES (?, ?)",
|
|
(name, datetime.now(timezone.utc).isoformat(timespec="seconds")),
|
|
)
|
|
except sqlite3.IntegrityError:
|
|
return None
|
|
return von_name(name)
|
|
|
|
|
|
def von_name(name):
|
|
with connect() as conn:
|
|
row = conn.execute("SELECT * FROM users WHERE name = ?", (name,)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def entfernen(user_id):
|
|
with connect() as conn:
|
|
conn.execute("DELETE FROM user_sessions WHERE user_id = ?", (user_id,))
|
|
return conn.execute("DELETE FROM users WHERE id = ?", (user_id,)).rowcount
|
|
|
|
|
|
def waehlen(session_id, user_id):
|
|
"""Gibt None zurück, wenn es den Namen nicht gibt."""
|
|
with connect() as conn:
|
|
da = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
|
|
if not da:
|
|
return None
|
|
conn.execute(
|
|
"INSERT INTO user_sessions (session_id, user_id, seit) VALUES (?, ?, ?) "
|
|
"ON CONFLICT (session_id) DO UPDATE SET "
|
|
"user_id = excluded.user_id, seit = excluded.seit",
|
|
(
|
|
session_id,
|
|
user_id,
|
|
datetime.now(timezone.utc).isoformat(timespec="seconds"),
|
|
),
|
|
)
|
|
return dict(da)
|
|
|
|
|
|
def abwaehlen(session_id):
|
|
with connect() as conn:
|
|
return conn.execute(
|
|
"DELETE FROM user_sessions WHERE session_id = ?", (session_id,)
|
|
).rowcount
|
|
|
|
|
|
def aktuell(session_id):
|
|
"""Der Benutzer dieser Sitzung, oder None."""
|
|
with connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT u.* FROM users u "
|
|
"JOIN user_sessions s ON s.user_id = u.id "
|
|
"WHERE s.session_id = ?",
|
|
(session_id,),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def besitzer(session_id):
|
|
"""Die user_id dieser Sitzung, oder None ohne gewählten Namen.
|
|
|
|
Sammlungen, Vorschläge und Wertungen werden damit beschriftet. Ein
|
|
Namenswechsel im selben Browser lässt die alten Einträge dann in Ruhe —
|
|
sie gehören weiter dem Namen, unter dem sie entstanden sind.
|
|
"""
|
|
with connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT user_id FROM user_sessions WHERE session_id = ?",
|
|
(session_id,),
|
|
).fetchone()
|
|
return row["user_id"] if row else None
|
|
|
|
|
|
def gehoert_mir(zeile, session_id, user_id):
|
|
"""Ein Eintrag gehört mir, wenn der Name passt — sonst die Sitzung."""
|
|
if zeile["user_id"] is not None or user_id is not None:
|
|
return zeile["user_id"] == user_id
|
|
return zeile["session_id"] == session_id
|
|
|
|
|
|
def name_von(session_id):
|
|
"""Nur der Name, für Vorschläge und Wertungen."""
|
|
benutzer = aktuell(session_id)
|
|
return benutzer["name"] if benutzer else None
|