dice-counter/web.py

# /// script
# requires-python = ">=3.11"
# dependencies = ["flask==3.1.3", "qrcode==8.2", "waitress==3.0.2"]
# ///
"""Webové počitadlo pro domácí síť.

Telefon zapisuje skóre (`/game/<id>`), notebook ukazuje zápisník (`/board`).
Server-rendered HTML, formuláře přes POST/redirect/GET, žádný build step.
"""

import io
import os
import shutil
import socket
import sys
import tempfile
from datetime import date
from pathlib import Path

import qrcode
from flask import (Flask, abort, g, redirect, render_template, request,
                   send_file, url_for)
from markupsafe import Markup, escape
from waitress import serve

import console
import net
import stats
import storage
import version
from core import FINAL_SCORE, Game, GameOver

def bundled(folder: str) -> str:
    """Cesta k šablonám a stylům.

    Ze zdrojáků leží vedle tohoto souboru. V binárce je PyInstaller rozbalí
    do dočasného adresáře, na který ukazuje `sys._MEIPASS`.
    """
    root = Path(getattr(sys, "_MEIPASS", Path(__file__).parent))
    return str(root / folder)


app = Flask(__name__,
            template_folder=bundled("templates"),
            static_folder=bundled("static"))

PORT = int(os.environ.get("DICE_PORT", 8000))

# Kolik kol tabule ukáže. Zbytek historie je na hráčích, ne na tabuli —
# ta má ukazovat stav, ne archiv.
BOARD_ROUNDS = 15


# --- připojení k databázi ---------------------------------------------------

def db():
    if "db" not in g:
        g.db = storage.connect()
    return g.db


@app.teardown_appcontext
def close_db(_exception):
    conn = g.pop("db", None)
    if conn is not None:
        conn.close()


# --- adresa a QR kód --------------------------------------------------------

def address_key() -> str:
    """Otisk všech adres. Tabule ho hlídá: jakmile se změní, načte se znovu,
    protože nabídnuté adresy i QR kódy už neplatí."""
    return ",".join(address.ip for address in net.addresses())


def links(path: str = "/") -> dict:
    """Adresy, na kterých je počitadlo dostupné, i s cíli pro QR kódy.

    Hledá se při každém načtení — notebook se během večera může přesunout
    z domácí WiFi na hotspot a zpátky. První v pořadí je ta nejnadějnější,
    zbytek si člověk může na tabuli přepnout.
    """
    return {
        "choices": [
            {"ip": address.ip, "label": address.label,
             "url": address.url(PORT) + path}
            for address in net.addresses()
        ],
        "port": PORT,
    }


def runs(row: list[bool]) -> list[tuple[int, int]]:
    """Souvislé tmavé úseky řádku jako (začátek, délka).

    Jeden široký obdélník místo řady čtverečků. Na tabuli visí QR kód pro
    každou adresu zvlášť, takže se to na velikosti stránky pozná.
    """
    found = []
    start = None

    for x, dark in enumerate([*row, False]):
        if dark and start is None:
            start = x
        elif not dark and start is not None:
            found.append((start, x - start))
            start = None

    return found


def qr_svg(data: str) -> Markup:
    """QR jako SVG složené z obdélníků — bez Pillow a bez práce s obrázky."""
    code = qrcode.QRCode(box_size=1, border=2)
    code.add_data(data)
    code.make(fit=True)
    matrix = code.get_matrix()
    size = len(matrix)

    rects = [
        f'<rect x="{x}" y="{y}" width="{width}" height="1"/>'
        for y, row in enumerate(matrix)
        for x, width in runs(row)
    ]

    return Markup(
        f'<svg class="qr" viewBox="0 0 {size} {size}" role="img" '
        f'aria-label="QR kód s adresou {escape(data)}" '
        f'shape-rendering="crispEdges">{"".join(rects)}</svg>'
    )


@app.context_processor
def verze():
    """Verze do patičky. Když někdo hlásí chybu, tohle je první otázka."""
    return {"verze": version.current()}


@app.template_filter("datum")
def datum(value: str) -> str:
    """Z ISO textu udělá 12. 8. 2026."""
    if not value:
        return ""
    den = value[:10].split("-")
    return f"{int(den[2])}. {int(den[1])}. {den[0]}"


@app.template_filter("qr")
def qr_filter(data: str) -> Markup:
    return qr_svg(data) if data else Markup("")


@app.template_filter("kola")
def kola(count: int) -> str:
    if count == 1:
        return "1 kolo"
    if 2 <= count <= 4:
        return f"{count} kola"
    return f"{count} kol"


@app.template_filter("cislo")
def cislo(value: int) -> str:
    """Tisíce se v češtině oddělují mezerou."""
    return f"{value:,}".replace(",", " ")


# --- zápisník ---------------------------------------------------------------

def scoresheet(game: Game) -> tuple[list[dict], int]:
    """Řádky zápisníku po kolech, od nejnovějšího.

    Naposledy odehrané kolo patří nahoru — po padesáti kolech by jinak bylo
    to podstatné mimo obrazovku. Vrací i počet kol, která se nevešla.

    Tahy zapsané před posledním vynulováním se přeškrtnou — přesně tak, jak
    by se škrtaly na papíře.
    """
    wiped_in_round = {}
    for turn in game.turns:
        if turn.reset:
            wiped_in_round[turn.player] = turn.round

    rows = []
    for number, by_player in game.rounds():
        cells = []
        for name in game.names:
            turn = by_player[name]
            cells.append({
                "value": None if turn is None else turn.value,
                "reset": turn is not None and turn.reset,
                "struck": turn is not None and turn.round < wiped_in_round.get(name, 0),
            })
        rows.append({"round": number, "cells": cells})

    rows.reverse()
    hidden = max(0, len(rows) - BOARD_ROUNDS)

    return rows[:BOARD_ROUNDS], hidden


def view(game: Game, game_id: int, row) -> dict:
    """Společný podklad pro zápisník i pro zadávací pohled."""
    totals = game.totals()
    winner = row["winner_name"] if row["status"] == storage.STATUS_FINISHED else ""
    rows, hidden = scoresheet(game)
    return {
        "game": game,
        "game_id": game_id,
        "row": row,
        "totals": totals,
        "rows": rows,
        "hidden": hidden,
        "streaks": {name: game.zero_streak(name) for name in game.names},
        "standings": game.standings(),
        "winner": winner,
        "current": "" if winner else game.current_player,
        "address_key": address_key(),
    }


def load(game_id: int) -> tuple[Game, "storage.sqlite3.Row"]:
    row = storage.game_row(db(), game_id)
    if row is None:
        abort(404)
    return storage.load_game(db(), game_id), row


# --- routy ------------------------------------------------------------------

@app.route("/")
def index():
    return render_template(
        "index.html",
        players=storage.known_players(db()),
        running=storage.running_game(db()),
        final_score=FINAL_SCORE,
    )


@app.post("/game")
def new_game():
    names = [n for n in request.form.getlist("player") if n.strip()]
    if not names:
        return redirect(url_for("index", chyba="Vyber aspoň jednoho hráče."))

    try:
        final_score = int(request.form.get("final_score") or FINAL_SCORE)
    except ValueError:
        return redirect(url_for("index", chyba="Cílové skóre musí být číslo."))

    try:
        game_id = storage.create_game(db(), names, final_score)
    except ValueError as exc:
        return redirect(url_for("index", chyba=str(exc)))

    return redirect(url_for("game_view", game_id=game_id))


@app.route("/game/<int:game_id>")
def game_view(game_id: int):
    game, row = load(game_id)
    return render_template("game.html", chyba=request.args.get("chyba", ""),
                           **view(game, game_id, row))


@app.post("/game/<int:game_id>/turn")
def add_turn(game_id: int):
    game, row = load(game_id)

    try:
        value = int(request.form.get("value", ""))
    except ValueError:
        return redirect(url_for("game_view", game_id=game_id,
                                chyba="Zadej skóre jako číslo."))

    try:
        result = game.add_score(value)
    except GameOver:
        return redirect(url_for("game_view", game_id=game_id))

    storage.add_turn(db(), game_id, result.turn)
    if result.winner:
        storage.finish_game(db(), game_id, result.winner)

    return redirect(url_for("game_view", game_id=game_id))


@app.post("/game/<int:game_id>/undo")
def undo(game_id: int):
    storage.undo_turn(db(), game_id)
    return redirect(url_for("game_view", game_id=game_id))


@app.post("/game/<int:game_id>/abandon")
def abandon(game_id: int):
    storage.abandon_game(db(), game_id)
    return redirect(url_for("index"))


@app.route("/board")
def board_redirect():
    game_id = storage.running_game(db())
    if game_id:
        return redirect(url_for("board", game_id=game_id))

    return render_template("board_idle.html", **links(url_for("index")))


@app.route("/board/<int:game_id>")
def board(game_id: int):
    game, row = load(game_id)
    return render_template("board.html",
                           **links(url_for("game_view", game_id=game_id)),
                           **view(game, game_id, row))


@app.route("/board/<int:game_id>/panel")
def board_panel(game_id: int):
    """Fragment, na který se tabule doptává každé dvě vteřiny."""
    game, row = load(game_id)
    return render_template("_panel.html", **view(game, game_id, row))


@app.route("/export")
def export():
    """Databáze ke stažení.

    Dokud se hrálo ze zdrojáků, ležel `dice.db` v pracovním adresáři a záloha
    byla otázka zkopírování. Z binárky se schovává do systémového datového
    adresáře, který na Windows běžný člověk nenajde — tak ať se dá stáhnout
    přes prohlížeč.

    `VACUUM INTO` udělá konzistentní kopii i uprostřed zápisu, na rozdíl od
    prostého přečtení souboru. Výsledek si přečteme do paměti a kopii hned
    zahodíme: databáze počitadla se počítá v kilobajtech a takhle nezůstane
    ležet v dočasném adresáři, když si ji někdo nestáhne celou.
    """
    conn = db()
    conn.commit()                       # VACUUM nesmí běžet v transakci

    docasny = tempfile.mkdtemp()
    try:
        kopie = Path(docasny) / "dice.db"
        conn.execute("VACUUM INTO ?", (str(kopie),))
        data = kopie.read_bytes()
    finally:
        shutil.rmtree(docasny, ignore_errors=True)

    return send_file(io.BytesIO(data), mimetype="application/vnd.sqlite3",
                     as_attachment=True,
                     download_name=f"dice-counter-{date.today().isoformat()}.db")


@app.route("/stats")
def hall():
    conn = db()
    return render_template("stats.html",
                           records=stats.records(conn),
                           careers=stats.careers(conn),
                           games=storage.recent_games(conn))


def announce() -> None:
    """Vypíše, kde všude je počitadlo k mání, a čím je zrovna spuštěné."""
    print(f"dice-counter {version.current()}\n")

    found = net.addresses()

    if found:
        print("Počitadlo je dostupné na:")
        for address in found:
            mark = "→" if address.primary else " "
            print(f" {mark} {address.url(PORT):30} {address.label}")
        print(f"\nTabule na notebook: {found[0].url(PORT)}/board")
    else:
        print("Adresu se nepodařilo zjistit. Spusť server s DICE_HOST=<adresa>.")

    print(f"\nDatabáze: {storage.DB_PATH.resolve()}")


if __name__ == "__main__":
    console.utf8()

    if {"--version", "-V"} & set(sys.argv[1:]):
        print(f"dice-counter {version.current()}")
        sys.exit(0)

    announce()

    # Ne vestavěný server Flasku: ten pod úvodní hlášku vysype červené
    # varování, že takhle se to nemá, a má pravdu. Waitress je čistě
    # pythonní, takže se zabalí do binárky bez řečí, a osm vláken pobere
    # telefony u stolu i tabuli, která se doptává každé dvě vteřiny.
    serve(app, host="0.0.0.0", port=PORT, threads=8)