218 lines
7.1 KiB
Python
218 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass
|
|
from datetime import date, datetime
|
|
from typing import Any, Iterable, Iterator, Optional
|
|
|
|
|
|
DB_PATH = os.getenv("DB_PATH", os.path.join("data", "voley.sqlite3"))
|
|
|
|
|
|
def _ensure_parent_dir(path: str) -> None:
|
|
parent = os.path.dirname(path)
|
|
if parent:
|
|
os.makedirs(parent, exist_ok=True)
|
|
|
|
|
|
def connect() -> sqlite3.Connection:
|
|
_ensure_parent_dir(DB_PATH)
|
|
conn = sqlite3.connect(DB_PATH, check_same_thread=False)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA foreign_keys = ON;")
|
|
return conn
|
|
|
|
|
|
@contextmanager
|
|
def get_conn() -> Iterator[sqlite3.Connection]:
|
|
conn = connect()
|
|
try:
|
|
yield conn
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def init_db() -> None:
|
|
with get_conn() as conn:
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS players (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL,
|
|
active INTEGER NOT NULL DEFAULT 1,
|
|
created_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS attendance (
|
|
player_id INTEGER NOT NULL,
|
|
session_date TEXT NOT NULL,
|
|
present INTEGER NOT NULL DEFAULT 1,
|
|
updated_at TEXT NOT NULL,
|
|
PRIMARY KEY (player_id, session_date),
|
|
FOREIGN KEY (player_id) REFERENCES players(id) ON DELETE CASCADE
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_attendance_date ON attendance(session_date);
|
|
"""
|
|
)
|
|
|
|
|
|
def now_iso() -> str:
|
|
return datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
|
|
|
|
|
|
def iso_date(d: date) -> str:
|
|
return d.isoformat()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Player:
|
|
id: int
|
|
name: str
|
|
active: bool
|
|
|
|
|
|
def list_players(active_only: bool = True) -> list[Player]:
|
|
sql = "SELECT id, name, active FROM players"
|
|
params: list[Any] = []
|
|
if active_only:
|
|
sql += " WHERE active=1"
|
|
sql += " ORDER BY name COLLATE NOCASE ASC"
|
|
with get_conn() as conn:
|
|
rows = conn.execute(sql, params).fetchall()
|
|
return [Player(id=int(r["id"]), name=str(r["name"]), active=bool(r["active"])) for r in rows]
|
|
|
|
|
|
def list_players_all() -> list[Player]:
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
"SELECT id, name, active FROM players ORDER BY active DESC, name COLLATE NOCASE ASC"
|
|
).fetchall()
|
|
return [Player(id=int(r["id"]), name=str(r["name"]), active=bool(r["active"])) for r in rows]
|
|
|
|
|
|
def add_player(name: str) -> int:
|
|
name = (name or "").strip()
|
|
if not name:
|
|
raise ValueError("name is required")
|
|
with get_conn() as conn:
|
|
cur = conn.execute(
|
|
"INSERT INTO players(name, active, created_at) VALUES (?, 1, ?)",
|
|
(name, now_iso()),
|
|
)
|
|
return int(cur.lastrowid)
|
|
|
|
|
|
def set_player_active(player_id: int, active: bool) -> None:
|
|
with get_conn() as conn:
|
|
conn.execute("UPDATE players SET active=? WHERE id=?", (1 if active else 0, int(player_id)))
|
|
|
|
|
|
def rename_player(player_id: int, name: str) -> None:
|
|
name = (name or "").strip()
|
|
if not name:
|
|
raise ValueError("name is required")
|
|
with get_conn() as conn:
|
|
conn.execute("UPDATE players SET name=? WHERE id=?", (name, int(player_id)))
|
|
|
|
|
|
def get_attendance_map(session_date: str) -> dict[int, bool]:
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
"SELECT player_id, present FROM attendance WHERE session_date=?",
|
|
(session_date,),
|
|
).fetchall()
|
|
return {int(r["player_id"]): bool(r["present"]) for r in rows}
|
|
|
|
|
|
def toggle_attendance(player_id: int, session_date: str) -> bool:
|
|
with get_conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT present FROM attendance WHERE player_id=? AND session_date=?",
|
|
(int(player_id), session_date),
|
|
).fetchone()
|
|
if row is None:
|
|
conn.execute(
|
|
"INSERT INTO attendance(player_id, session_date, present, updated_at) VALUES (?, ?, 1, ?)",
|
|
(int(player_id), session_date, now_iso()),
|
|
)
|
|
return True
|
|
new_present = 0 if bool(row["present"]) else 1
|
|
conn.execute(
|
|
"UPDATE attendance SET present=?, updated_at=? WHERE player_id=? AND session_date=?",
|
|
(new_present, now_iso(), int(player_id), session_date),
|
|
)
|
|
return bool(new_present)
|
|
|
|
|
|
def list_present_players(session_date: str, active_only: bool = True) -> list[Player]:
|
|
sql = """
|
|
SELECT p.id, p.name, p.active
|
|
FROM attendance a
|
|
JOIN players p ON p.id = a.player_id
|
|
WHERE a.session_date = ? AND a.present = 1
|
|
"""
|
|
params: list[Any] = [session_date]
|
|
if active_only:
|
|
sql += " AND p.active = 1"
|
|
sql += " ORDER BY p.name COLLATE NOCASE ASC"
|
|
with get_conn() as conn:
|
|
rows = conn.execute(sql, params).fetchall()
|
|
return [Player(id=int(r["id"]), name=str(r["name"]), active=bool(r["active"])) for r in rows]
|
|
|
|
|
|
def attendance_counts_by_sunday_in_month(year: int, month: int) -> list[tuple[str, int]]:
|
|
# Возвращает список (YYYY-MM-DD, count) для воскресений в указанном месяце.
|
|
# Считаем количество присутствий (present=1).
|
|
y = int(year)
|
|
m = int(month)
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT session_date, SUM(CASE WHEN present=1 THEN 1 ELSE 0 END) AS cnt
|
|
FROM attendance
|
|
WHERE substr(session_date, 1, 7) = ?
|
|
GROUP BY session_date
|
|
ORDER BY session_date ASC
|
|
""",
|
|
(f"{y:04d}-{m:02d}",),
|
|
).fetchall()
|
|
return [(str(r["session_date"]), int(r["cnt"] or 0)) for r in rows]
|
|
|
|
|
|
def attendance_counts_by_month_in_year(year: int) -> list[tuple[str, int]]:
|
|
# Возвращает список (YYYY-MM, count) — суммарное число посещений по месяцам.
|
|
y = int(year)
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT substr(session_date, 1, 7) AS ym,
|
|
SUM(CASE WHEN present=1 THEN 1 ELSE 0 END) AS cnt
|
|
FROM attendance
|
|
WHERE substr(session_date, 1, 4) = ?
|
|
GROUP BY ym
|
|
ORDER BY ym ASC
|
|
""",
|
|
(f"{y:04d}",),
|
|
).fetchall()
|
|
return [(str(r["ym"]), int(r["cnt"] or 0)) for r in rows]
|
|
|
|
|
|
def get_attendance_history() -> list[dict]:
|
|
"""Возвращает историю посещаемости: (дата, имя игрока, присутствовал)."""
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT a.session_date, p.name, a.present
|
|
FROM attendance a
|
|
JOIN players p ON p.id = a.player_id
|
|
WHERE a.present = 1
|
|
ORDER BY a.session_date DESC, p.name COLLATE NOCASE ASC
|
|
"""
|
|
).fetchall()
|
|
return [{"date": str(r["session_date"]), "name": str(r["name"]), "present": bool(r["present"])} for r in rows]
|
|
|