Meisterschaft: eine pro Saison, mit zugeordneten Turnieren

Auf Vorgabe des Auftraggebers umgestellt:
* genau eine Meisterschaft je Saison (unique auf saison_jahr)
* turnier.meisterschaft_id (nullable FK) statt bool zaehlt_meisterschaft;
  NULL = "o. M.". Kompat-Property turnier.zaehlt_meisterschaft bleibt.
* Teilwertungen (Erwachsene-Gesamt, Jugend J1-J5) sind (Kategorie, Klasse)-
  Sichten auf die eine Meisterschaft, keine eigenen Zeilen mehr
* meisterschaft_stand: kategorie + klasse_id + Unique je Wertung/Fahrer
* berechne_stand(kategorie=, klasse_id=), wertungen_fuer(), berechne_alle_wertungen()
* Read-API: /meisterschaften/{id}/stand[?wertung=], .../endlauf-qualifikation?wertung=
* Ingest: POST /turniere haengt zaehlende Turniere an die Jahres-Meisterschaft
* Initialmigration neu erzeugt (Projekt noch nicht ausgerollt); Seed: eine
  "HTH-Meisterschaft <jahr>"
* Tests angepasst/ergaenzt (61 gruen)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
linus
2026-09-05 19:52:25 +02:00
co-authored by Claude Sonnet 5
parent 1df633bdec
commit c33aaa1a4b
19 changed files with 513 additions and 199 deletions
+1 -1
View File
@@ -40,6 +40,6 @@ debian/debhelper-build-stamp
.vscode/
*.swp
# Claude Code lokale (persoenliche) Einstellungen
# Claude Code
.claude/
@@ -1,8 +1,8 @@
"""Initiales Schema
Revision ID: df5e2dd5a864
Revision ID: d7d7d8a2cf90
Revises:
Create Date: 2026-09-05 18:08:45.999851+02:00
Create Date: 2026-09-05 19:48:05.733306+02:00
"""
from __future__ import annotations
@@ -12,17 +12,15 @@ from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision: str = 'df5e2dd5a864'
revision: str = 'd7d7d8a2cf90'
down_revision: str | None = None
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# pg_trgm fuer die unscharfe Fahrersuche beim Import (Plan §6.4)
op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm")
# ### commands auto generated by Alembic - please adjust! ###
op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm")
op.create_table('benutzer',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('benutzername', sa.String(length=80), nullable=False),
@@ -105,11 +103,11 @@ def upgrade() -> None:
sa.PrimaryKeyConstraint('id', name=op.f('pk_fahrer'))
)
op.create_index(op.f('ix_fahrer_geburtsjahr'), 'fahrer', ['geburtsjahr'], unique=False)
op.create_index(op.f('ix_fahrer_verein_id'), 'fahrer', ['verein_id'], unique=False)
op.execute(
"CREATE INDEX ix_fahrer_name_trgm ON fahrer USING gin "
"((lower(nachname || ' ' || vorname)) gin_trgm_ops)"
)
op.create_index(op.f('ix_fahrer_verein_id'), 'fahrer', ['verein_id'], unique=False)
op.create_table('klasse',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('saison_jahr', sa.Integer(), nullable=False),
@@ -128,6 +126,16 @@ def upgrade() -> None:
sa.UniqueConstraint('saison_jahr', 'code', name='klasse_saison_code')
)
op.create_index(op.f('ix_klasse_saison_jahr'), 'klasse', ['saison_jahr'], unique=False)
op.create_table('meisterschaft',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('saison_jahr', sa.Integer(), nullable=False),
sa.Column('bezeichnung', sa.String(length=200), nullable=False),
sa.Column('regelwerk', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
sa.ForeignKeyConstraint(['saison_jahr'], ['saison.jahr'], name=op.f('fk_meisterschaft_saison_jahr_saison')),
sa.PrimaryKeyConstraint('id', name=op.f('pk_meisterschaft')),
sa.UniqueConstraint('saison_jahr', name='meisterschaft_saison_eindeutig')
)
op.create_index(op.f('ix_meisterschaft_saison_jahr'), 'meisterschaft', ['saison_jahr'], unique=True)
op.create_table('veranstaltung',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('datum', sa.Date(), nullable=False),
@@ -145,40 +153,11 @@ def upgrade() -> None:
sa.PrimaryKeyConstraint('id', name=op.f('pk_veranstaltung'))
)
op.create_index(op.f('ix_veranstaltung_datum'), 'veranstaltung', ['datum'], unique=False)
op.create_table('meisterschaft',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('saison_jahr', sa.Integer(), nullable=False),
sa.Column('bezeichnung', sa.String(length=200), nullable=False),
sa.Column('kategorie', sa.Enum('J', 'S', 'A', name='meisterschaft_kategorie'), nullable=False),
sa.Column('klasse_id', sa.Integer(), nullable=True),
sa.Column('regelwerk', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
sa.ForeignKeyConstraint(['klasse_id'], ['klasse.id'], name=op.f('fk_meisterschaft_klasse_id_klasse')),
sa.ForeignKeyConstraint(['saison_jahr'], ['saison.jahr'], name=op.f('fk_meisterschaft_saison_jahr_saison')),
sa.PrimaryKeyConstraint('id', name=op.f('pk_meisterschaft'))
)
op.create_index(op.f('ix_meisterschaft_saison_jahr'), 'meisterschaft', ['saison_jahr'], unique=False)
op.create_table('turnier',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('veranstaltung_id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(length=120), nullable=False),
sa.Column('veranstalter_verein_id', sa.Integer(), nullable=True),
sa.Column('zaehlt_meisterschaft', sa.Boolean(), nullable=False),
sa.Column('ist_aktiv', sa.Boolean(), nullable=False),
sa.Column('status', sa.Enum('geplant', 'laufend', 'vorlaeufig', 'final', name='turnier_status'), nullable=False),
sa.Column('erstellt_am', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('geaendert_am', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['veranstalter_verein_id'], ['verein.id'], name=op.f('fk_turnier_veranstalter_verein_id_verein')),
sa.ForeignKeyConstraint(['veranstaltung_id'], ['veranstaltung.id'], name=op.f('fk_turnier_veranstaltung_id_veranstaltung'), ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id', name=op.f('pk_turnier'))
)
op.create_index(op.f('ix_turnier_veranstaltung_id'), 'turnier', ['veranstaltung_id'], unique=False)
op.create_index(
'uq_turnier_ist_aktiv', 'turnier', ['ist_aktiv'], unique=True,
postgresql_where=sa.text('ist_aktiv'),
)
op.create_table('meisterschaft_stand',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('meisterschaft_id', sa.Integer(), nullable=False),
sa.Column('kategorie', sa.Enum('J', 'S', 'A', name='meisterschaft_stand_kategorie'), nullable=False),
sa.Column('klasse_id', sa.Integer(), nullable=True),
sa.Column('fahrer_id', sa.Integer(), nullable=False),
sa.Column('punkte_gesamt', sa.Numeric(precision=9, scale=2), nullable=False),
sa.Column('punkte_je_lauf', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
@@ -187,11 +166,31 @@ def upgrade() -> None:
sa.Column('platz', sa.Integer(), nullable=True),
sa.Column('berechnet_am', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['fahrer_id'], ['fahrer.id'], name=op.f('fk_meisterschaft_stand_fahrer_id_fahrer')),
sa.ForeignKeyConstraint(['klasse_id'], ['klasse.id'], name=op.f('fk_meisterschaft_stand_klasse_id_klasse')),
sa.ForeignKeyConstraint(['meisterschaft_id'], ['meisterschaft.id'], name=op.f('fk_meisterschaft_stand_meisterschaft_id_meisterschaft'), ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id', name=op.f('pk_meisterschaft_stand'))
sa.PrimaryKeyConstraint('id', name=op.f('pk_meisterschaft_stand')),
sa.UniqueConstraint('meisterschaft_id', 'kategorie', 'klasse_id', 'fahrer_id', name='meisterschaft_stand_eindeutig')
)
op.create_index(op.f('ix_meisterschaft_stand_fahrer_id'), 'meisterschaft_stand', ['fahrer_id'], unique=False)
op.create_index(op.f('ix_meisterschaft_stand_meisterschaft_id'), 'meisterschaft_stand', ['meisterschaft_id'], unique=False)
op.create_table('turnier',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('veranstaltung_id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(length=120), nullable=False),
sa.Column('veranstalter_verein_id', sa.Integer(), nullable=True),
sa.Column('meisterschaft_id', sa.Integer(), nullable=True),
sa.Column('ist_aktiv', sa.Boolean(), nullable=False),
sa.Column('status', sa.Enum('geplant', 'laufend', 'vorlaeufig', 'final', name='turnier_status'), nullable=False),
sa.Column('erstellt_am', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('geaendert_am', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['meisterschaft_id'], ['meisterschaft.id'], name=op.f('fk_turnier_meisterschaft_id_meisterschaft'), ondelete='SET NULL'),
sa.ForeignKeyConstraint(['veranstalter_verein_id'], ['verein.id'], name=op.f('fk_turnier_veranstalter_verein_id_verein')),
sa.ForeignKeyConstraint(['veranstaltung_id'], ['veranstaltung.id'], name=op.f('fk_turnier_veranstaltung_id_veranstaltung'), ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id', name=op.f('pk_turnier'))
)
op.create_index(op.f('ix_turnier_meisterschaft_id'), 'turnier', ['meisterschaft_id'], unique=False)
op.create_index(op.f('ix_turnier_veranstaltung_id'), 'turnier', ['veranstaltung_id'], unique=False)
op.create_index('uq_turnier_ist_aktiv', 'turnier', ['ist_aktiv'], unique=True, postgresql_where=sa.text('ist_aktiv'))
op.create_table('wertungsklasse',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('turnier_id', sa.Integer(), nullable=False),
@@ -275,20 +274,20 @@ def downgrade() -> None:
op.drop_table('start')
op.drop_index(op.f('ix_wertungsklasse_turnier_id'), table_name='wertungsklasse')
op.drop_table('wertungsklasse')
op.drop_index('uq_turnier_ist_aktiv', table_name='turnier', postgresql_where=sa.text('ist_aktiv'))
op.drop_index(op.f('ix_turnier_veranstaltung_id'), table_name='turnier')
op.drop_index(op.f('ix_turnier_meisterschaft_id'), table_name='turnier')
op.drop_table('turnier')
op.drop_index(op.f('ix_meisterschaft_stand_meisterschaft_id'), table_name='meisterschaft_stand')
op.drop_index(op.f('ix_meisterschaft_stand_fahrer_id'), table_name='meisterschaft_stand')
op.drop_table('meisterschaft_stand')
op.drop_index('uq_turnier_ist_aktiv', table_name='turnier')
op.drop_index(op.f('ix_turnier_veranstaltung_id'), table_name='turnier')
op.drop_table('turnier')
op.drop_index(op.f('ix_meisterschaft_saison_jahr'), table_name='meisterschaft')
op.drop_table('meisterschaft')
op.drop_index(op.f('ix_veranstaltung_datum'), table_name='veranstaltung')
op.drop_table('veranstaltung')
op.drop_index(op.f('ix_meisterschaft_saison_jahr'), table_name='meisterschaft')
op.drop_table('meisterschaft')
op.drop_index(op.f('ix_klasse_saison_jahr'), table_name='klasse')
op.drop_table('klasse')
op.drop_index(op.f('ix_fahrer_verein_id'), table_name='fahrer')
op.drop_index('ix_fahrer_name_trgm', table_name='fahrer', postgresql_using='gin')
op.drop_index(op.f('ix_fahrer_geburtsjahr'), table_name='fahrer')
op.drop_table('fahrer')
op.drop_table('verein')
+26 -17
View File
@@ -13,7 +13,7 @@ from pydantic import BaseModel, Field
from sqlalchemy import select
from oamc.matching.fahrer import fahrer_zusammenfuehren
from oamc.meisterschaft.berechnung import berechne_stand
from oamc.meisterschaft.berechnung import berechne_alle_wertungen
from oamc.models import (
Geraet,
ImportJob,
@@ -21,7 +21,7 @@ from oamc.models import (
Start,
Zeitmessung,
)
from oamc.models.enums import GeraetTyp, ZeitmessungStatus
from oamc.models.enums import GeraetTyp, Kategorie, ZeitmessungStatus
from oamc.security.deps import Admin, Auswerter, DbDep
from oamc.security.tokens import erzeuge_geraete_token
from oamc.wertung.service import turnier_auswerten
@@ -170,28 +170,37 @@ def import_jobs(db: DbDep, _auswerter: Auswerter, limit: int = 50):
@router.post("/meisterschaften/{meisterschaft_id}/berechnen")
def meisterschaft_berechnen(db: DbDep, _auswerter: Auswerter, meisterschaft_id: int):
"""Stand neu berechnen und materialisieren (``meisterschaft_stand``)."""
"""Alle Wertungen der Jahresmeisterschaft neu berechnen und materialisieren
(``meisterschaft_stand``)."""
bericht = berechne_stand(db, meisterschaft_id)
berichte = berechne_alle_wertungen(db, meisterschaft_id)
db.query(MeisterschaftStand).filter(
MeisterschaftStand.meisterschaft_id == meisterschaft_id
).delete()
for z in bericht.zeilen:
db.add(
MeisterschaftStand(
meisterschaft_id=meisterschaft_id,
fahrer_id=z.fahrer_id,
punkte_gesamt=z.punkte_gesamt,
punkte_je_lauf={str(k): float(v) for k, v in z.punkte_je_lauf.items()},
gestrichene_laeufe=z.gestrichene_laeufe,
anzahl_siege=z.anzahl_siege,
platz=z.platz,
gesamt = 0
hinweise: set[str] = set()
for b in berichte:
hinweise.update(b.hinweise)
for z in b.zeilen:
gesamt += 1
db.add(
MeisterschaftStand(
meisterschaft_id=meisterschaft_id,
kategorie=Kategorie(b.kategorie),
klasse_id=b.klasse_id,
fahrer_id=z.fahrer_id,
punkte_gesamt=z.punkte_gesamt,
punkte_je_lauf={str(k): float(v) for k, v in z.punkte_je_lauf.items()},
gestrichene_laeufe=z.gestrichene_laeufe,
anzahl_siege=z.anzahl_siege,
platz=z.platz,
)
)
)
return {
"meisterschaft_id": meisterschaft_id,
"zeilen": len(bericht.zeilen),
"hinweise": bericht.hinweise,
"wertungen": len(berichte),
"zeilen": gesamt,
"hinweise": sorted(hinweise),
}
+14 -3
View File
@@ -48,22 +48,33 @@ def veranstaltung_anlegen(db: DbDep, _key: IngestKey, daten: VeranstaltungCreate
@router.post("/turniere", status_code=201)
def turnier_anlegen(db: DbDep, _key: IngestKey, daten: TurnierCreate):
if db.get(Veranstaltung, daten.veranstaltung_id) is None:
veranstaltung = db.get(Veranstaltung, daten.veranstaltung_id)
if veranstaltung is None:
raise HTTPException(404, "Veranstaltung nicht gefunden")
verein = None
if daten.veranstalter_verein:
from oamc.importer.ingest import finde_verein
verein = finde_verein(db, daten.veranstalter_verein)
# Zaehlt das Turnier zur Jahreswertung, wird es an die (eine) Meisterschaft
# der Saison gehaengt.
meisterschaft_id = None
if daten.zaehlt_meisterschaft:
from oamc.importer.ingest import meisterschaft_der_saison
m = meisterschaft_der_saison(db, veranstaltung.datum.year)
meisterschaft_id = m.id if m else None
t = Turnier(
veranstaltung_id=daten.veranstaltung_id,
name=daten.name,
veranstalter_verein_id=verein.id if verein else None,
zaehlt_meisterschaft=daten.zaehlt_meisterschaft,
meisterschaft_id=meisterschaft_id,
)
db.add(t)
db.flush()
return {"id": t.id}
return {"id": t.id, "meisterschaft_id": meisterschaft_id}
@router.post("/turniere/{turnier_id}/starts", response_model=ImportBericht)
+76 -19
View File
@@ -16,7 +16,12 @@ from oamc.api.v1.serialisierung import (
turnier_kurz,
veranstaltung_kurz,
)
from oamc.meisterschaft.berechnung import berechne_stand, endlauf_qualifikation
from oamc.meisterschaft.berechnung import (
berechne_alle_wertungen,
berechne_stand,
endlauf_qualifikation,
wertungen_fuer,
)
from oamc.models import (
Fahrer,
Meisterschaft,
@@ -161,27 +166,34 @@ def fahrer_historie(db: DbDep, fahrer_id: int):
@router.get("/meisterschaften")
def liste_meisterschaften(db: DbDep, saison: int | None = None):
stmt = select(Meisterschaft)
"""Pro Saison genau eine Meisterschaft."""
stmt = select(Meisterschaft).order_by(Meisterschaft.saison_jahr.desc())
if saison:
stmt = stmt.where(Meisterschaft.saison_jahr == saison)
return [
{
"id": m.id,
"bezeichnung": m.bezeichnung,
"saison": m.saison_jahr,
"kategorie": m.kategorie.value,
}
for m in db.scalars(stmt)
]
out = []
for m in db.scalars(stmt):
out.append(
{
"id": m.id,
"bezeichnung": m.bezeichnung,
"saison": m.saison_jahr,
"anzahl_turniere": len(m.turniere),
"wertungen": [
{"code": w.code, "bezeichnung": w.bezeichnung, "kategorie": w.kategorie.value,
"klasse_id": w.klasse_id}
for w in wertungen_fuer(db, m.id)
],
}
)
return out
@router.get("/meisterschaften/{meisterschaft_id}/stand")
def meisterschaft_stand(db: DbDep, meisterschaft_id: int):
bericht = berechne_stand(db, meisterschaft_id)
def _wertung_json(db, bericht) -> dict:
return {
"meisterschaft_id": bericht.meisterschaft_id,
"saison": bericht.saison_jahr,
"wertung": bericht.wertung,
"kategorie": bericht.kategorie,
"klasse_id": bericht.klasse_id,
"laeufe_gesamt": bericht.laeufe_gesamt,
"laeufe_gewertet": bericht.laeufe_gewertet,
"hinweise": bericht.hinweise,
@@ -198,11 +210,56 @@ def meisterschaft_stand(db: DbDep, meisterschaft_id: int):
}
@router.get("/meisterschaften/{meisterschaft_id}/endlauf-qualifikation")
def meisterschaft_quali(db: DbDep, meisterschaft_id: int, plaetze: int = 10):
vorschlaege = endlauf_qualifikation(db, meisterschaft_id, plaetze)
@router.get("/meisterschaften/{meisterschaft_id}/stand")
def meisterschaft_stand(
db: DbDep,
meisterschaft_id: int,
wertung: str | None = Query(default=None, description='z. B. "erwachsene" oder "J 3"'),
):
"""Stand der Jahresmeisterschaft. Ohne ``wertung`` alle Teilwertungen."""
if wertung:
gewuenscht = next(
(w for w in wertungen_fuer(db, meisterschaft_id) if w.code.lower() == wertung.lower()),
None,
)
if gewuenscht is None:
raise HTTPException(404, f"Wertung {wertung!r} unbekannt")
bericht = berechne_stand(
db, meisterschaft_id, kategorie=gewuenscht.kategorie, klasse_id=gewuenscht.klasse_id
)
return {"meisterschaft_id": meisterschaft_id, **_wertung_json(db, bericht)}
berichte = berechne_alle_wertungen(db, meisterschaft_id)
return {
"meisterschaft_id": meisterschaft_id,
"saison": berichte[0].saison_jahr if berichte else None,
"wertungen": [_wertung_json(db, b) for b in berichte],
}
@router.get("/meisterschaften/{meisterschaft_id}/endlauf-qualifikation")
def meisterschaft_quali(
db: DbDep,
meisterschaft_id: int,
wertung: str = Query(default="erwachsene"),
plaetze: int = 10,
):
gewuenscht = next(
(w for w in wertungen_fuer(db, meisterschaft_id) if w.code.lower() == wertung.lower()), None
)
if gewuenscht is None:
raise HTTPException(404, f"Wertung {wertung!r} unbekannt")
vorschlaege = endlauf_qualifikation(
db,
meisterschaft_id,
kategorie=gewuenscht.kategorie,
klasse_id=gewuenscht.klasse_id,
plaetze=plaetze,
)
return {
"meisterschaft_id": meisterschaft_id,
"wertung": gewuenscht.code,
"hinweis": "Vorschlag — die endgueltige Festlegung trifft der Bereichsleiter (Plan §3.4).",
"plaetze": plaetze,
"vorschlaege": [
+20 -6
View File
@@ -14,7 +14,7 @@ from sqlalchemy.orm import selectinload
from oamc.api.v1.serialisierung import ergebnis_zeile
from oamc.config import get_settings
from oamc.frontend.templating import templates
from oamc.meisterschaft.berechnung import berechne_stand
from oamc.meisterschaft.berechnung import berechne_stand, wertungen_fuer
from oamc.models import Fahrer, Meisterschaft, Start, Turnier, Veranstaltung
from oamc.security.deps import DbDep
@@ -217,17 +217,31 @@ def meisterschaft_uebersicht(request: Request, db: DbDep):
@router.get("/meisterschaft/{meisterschaft_id}")
def meisterschaft_detail(request: Request, db: DbDep, meisterschaft_id: int):
if db.get(Meisterschaft, meisterschaft_id) is None:
def meisterschaft_detail(
request: Request, db: DbDep, meisterschaft_id: int, wertung: str | None = None
):
m = db.get(Meisterschaft, meisterschaft_id)
if m is None:
raise HTTPException(404, "Meisterschaft nicht gefunden")
bericht = berechne_stand(db, meisterschaft_id)
alle = wertungen_fuer(db, meisterschaft_id)
aktiv = next((w for w in alle if w.code.lower() == (wertung or "").lower()), alle[0])
bericht = berechne_stand(
db, meisterschaft_id, kategorie=aktiv.kategorie, klasse_id=aktiv.klasse_id
)
fahrer_ids = [z.fahrer_id for z in bericht.zeilen]
fahrer_map = {
f.id: f for f in db.scalars(select(Fahrer).where(Fahrer.id.in_([z.fahrer_id for z in bericht.zeilen])))
f.id: f for f in db.scalars(select(Fahrer).where(Fahrer.id.in_(fahrer_ids)))
}
return templates.TemplateResponse(
request,
"public/meisterschaft_detail.html",
{"bericht": bericht, "fahrer_map": fahrer_map},
{
"meisterschaft": m,
"wertungen": alle,
"aktive_wertung": aktiv,
"bericht": bericht,
"fahrer_map": fahrer_map,
},
)
+2
View File
@@ -3,6 +3,7 @@
from oamc.importer.ingest import (
finde_klasse,
finde_verein,
meisterschaft_der_saison,
verarbeite_ergebnisse,
verarbeite_starts,
)
@@ -10,6 +11,7 @@ from oamc.importer.ingest import (
__all__ = [
"finde_klasse",
"finde_verein",
"meisterschaft_der_saison",
"verarbeite_ergebnisse",
"verarbeite_starts",
]
+27
View File
@@ -15,10 +15,13 @@ from sqlalchemy import func, select
from sqlalchemy.orm import Session
from oamc.matching import MatchStatus, match_oder_neu, normalisiere
from oamc.meisterschaft.berechnung import PLATZHALTER_PUNKTETABELLE
from oamc.models import (
Ergebnis,
ImportJob,
Klasse,
Meisterschaft,
Saison,
Start,
Turnier,
Verein,
@@ -63,6 +66,30 @@ def _saison_fuer_turnier(db: Session, turnier: Turnier) -> int:
return turnier.veranstaltung.datum.year
def meisterschaft_der_saison(db: Session, jahr: int, *, anlegen: bool = True) -> Meisterschaft | None:
"""Die (einzige) Meisterschaft einer Saison — bei Bedarf anlegen.
Modell: pro Saison genau eine Meisterschaft; die zaehlenden Turniere haengen
daran (``turnier.meisterschaft_id``)."""
m = db.scalars(select(Meisterschaft).where(Meisterschaft.saison_jahr == jahr)).first()
if m is not None or not anlegen:
return m
if db.get(Saison, jahr) is None:
return None # ohne Saison-Stammdatensatz keine Meisterschaft
m = Meisterschaft(
saison_jahr=jahr,
bezeichnung=f"HTH-Meisterschaft {jahr}",
regelwerk={
"punkte": {str(k): v for k, v in PLATZHALTER_PUNKTETABELLE.items()},
"_hinweis": "O-1 offen: PLATZHALTER-Punktetabelle, nicht offiziell!",
},
)
db.add(m)
db.flush()
return m
def _idempotenz_treffer(db: Session, schluessel: str | None) -> ImportJob | None:
if not schluessel:
return None
+14 -4
View File
@@ -1,23 +1,33 @@
"""Meisterschafts-/Pokalwertung (HTH) — AP 3.
BLOCKIERT durch offene Frage O-1: die echte Punktetabelle (Platz -> Punkte) ist
noch nicht bekannt. Bis dahin arbeitet dieses Modul mit
``PLATZHALTER_PUNKTETABELLE`` (klar als solche markiert) bzw. mit der Tabelle aus
``meisterschaft.regelwerk['punkte']``.
Eine Meisterschaft pro Saison; darin mehrere Wertungen (Erwachsene-Gesamt + je
Jugendklasse). BLOCKIERT durch offene Frage O-1: die echte Punktetabelle
(Platz -> Punkte) ist noch nicht bekannt — bis dahin
``PLATZHALTER_PUNKTETABELLE`` bzw. ``meisterschaft.regelwerk['punkte']``.
"""
from oamc.meisterschaft.berechnung import (
PLATZHALTER_PUNKTETABELLE,
MeisterschaftsBericht,
StandZeile,
Wertung,
berechne_alle_wertungen,
berechne_stand,
endlauf_qualifikation,
gewertete_laeufe,
punkte_fuer_platz,
wertungen_fuer,
)
__all__ = [
"PLATZHALTER_PUNKTETABELLE",
"MeisterschaftsBericht",
"StandZeile",
"Wertung",
"berechne_stand",
"berechne_alle_wertungen",
"wertungen_fuer",
"endlauf_qualifikation",
"gewertete_laeufe",
"punkte_fuer_platz",
]
+115 -33
View File
@@ -1,14 +1,18 @@
"""Meisterschaftsstand, Streichresultate, Tie-Break, Endlaufqualifikation.
Regeln (Plan §3.4):
* Erwachsene: ein Gesamtklassement A + S (klassenuebergreifend).
* Jugend: Pokalwertung klassenweise J 1 - J 5.
* Streichresultate nach Tabelle in ``saison.streichresultate_tabelle``
Modell: **eine Meisterschaft pro Saison**, dazu gehoeren die Turniere mit
``turnier.meisterschaft_id``. Innerhalb dieser einen Meisterschaft werden mehrere
**Wertungen** gerechnet (Plan §3.4):
* Erwachsene: ein Gesamtklassement A + S (``kategorie=S``, ``klasse_id=None``)
* Jugend: Pokalwertung je Klasse J 1 - J 5 (``kategorie=J``, ``klasse_id`` gesetzt)
Weitere Regeln:
* Streichresultate nach ``saison.streichresultate_tabelle``
("Turniere in der Saison" -> "davon gewertet").
* Nur Laeufe mit ``turnier.zaehlt_meisterschaft``.
* Tie-Break bei Punktgleichheit: Majoritaet der Siege.
* Endlauf: die besten 10 Erwachsenen bzw. 10 Jugendlichen — die Anwendung
*schlaegt vor*, entscheidet nicht (Bereichsleiter).
* Endlauf: die besten 10 je Wertung — die Anwendung *schlaegt vor*, entscheidet
nicht (Bereichsleiter).
Die Punktetabelle ist offene Frage **O-1**. ``PLATZHALTER_PUNKTETABELLE`` ist
NICHT die echte Tabelle und erzeugt bei Verwendung einen Hinweis im Bericht.
@@ -50,9 +54,8 @@ def punkte_fuer_platz(platz: int | None, tabelle: dict[int, int] | None = None)
tab = tabelle or PLATZHALTER_PUNKTETABELLE
if platz in tab:
return Decimal(tab[platz])
# unterhalb der Tabelle: kleinster definierter Wert bzw. Platzhalter
if tabelle:
return Decimal(min(tabelle.values())) if tabelle else Decimal(0)
return Decimal(min(tabelle.values()))
return Decimal(PLATZHALTER_PUNKTE_AB_15)
@@ -61,18 +64,26 @@ def gewertete_laeufe(anzahl_turniere: int, tabelle: dict[str, int] | None) -> in
if not tabelle:
return anzahl_turniere
# exakter Eintrag?
key = str(anzahl_turniere)
if key in tabelle:
return int(tabelle[key])
# sonst: naechstkleinerer definierter Schluessel
kleinere = sorted((int(k) for k in tabelle), reverse=True)
for k in kleinere:
for k in sorted((int(k) for k in tabelle), reverse=True):
if k <= anzahl_turniere:
return int(tabelle[str(k)])
return anzahl_turniere
@dataclass
class Wertung:
"""Eine Teilwertung innerhalb der Jahresmeisterschaft."""
code: str # "erwachsene" | "J 1" ... "J 5"
bezeichnung: str
kategorie: Kategorie
klasse_id: int | None
endlauf_plaetze: int = 10
@dataclass
class StandZeile:
fahrer_id: int
@@ -87,14 +98,53 @@ class StandZeile:
class MeisterschaftsBericht:
meisterschaft_id: int
saison_jahr: int
wertung: str # code der Wertung
kategorie: str
klasse_id: int | None
laeufe_gesamt: int
laeufe_gewertet: int
zeilen: list[StandZeile]
hinweise: list[str] = field(default_factory=list)
def _relevante_starts(db: Session, m: Meisterschaft, turnier_ids: list[int]) -> list[Start]:
def wertungen_fuer(db: Session, meisterschaft_id: int) -> list[Wertung]:
"""Die Wertungen der Jahresmeisterschaft: Erwachsene-Gesamt + je Jugendklasse
der Saison."""
m = db.get(Meisterschaft, meisterschaft_id)
if m is None:
raise ValueError(f"Meisterschaft {meisterschaft_id} nicht gefunden")
wertungen = [
Wertung(
code="erwachsene",
bezeichnung="Erwachsene (A + S, klassenuebergreifend)",
kategorie=Kategorie.STAMM,
klasse_id=None,
)
]
jugend = db.scalars(
select(Klasse)
.where(Klasse.saison_jahr == m.saison_jahr, Klasse.kategorie == Kategorie.JUGEND)
.order_by(Klasse.sortierung)
).all()
for k in jugend:
wertungen.append(
Wertung(
code=k.code,
bezeichnung=f"Jugend {k.code}",
kategorie=Kategorie.JUGEND,
klasse_id=k.id,
)
)
return wertungen
def _relevante_starts(
db: Session,
turnier_ids: list[int],
kategorie: Kategorie,
klasse_id: int | None,
) -> list[Start]:
stmt = (
select(Start)
.join(Ergebnis, Ergebnis.start_id == Start.id)
@@ -105,9 +155,9 @@ def _relevante_starts(db: Session, m: Meisterschaft, turnier_ids: list[int]) ->
Start.status == StartStatus.GESTARTET,
)
)
if m.kategorie == Kategorie.JUGEND:
if m.klasse_id is not None:
stmt = stmt.where(Start.klasse_id == m.klasse_id)
if kategorie == Kategorie.JUGEND:
if klasse_id is not None:
stmt = stmt.where(Start.klasse_id == klasse_id)
else:
stmt = stmt.where(Klasse.kategorie == Kategorie.JUGEND)
else:
@@ -116,7 +166,15 @@ def _relevante_starts(db: Session, m: Meisterschaft, turnier_ids: list[int]) ->
return list(db.scalars(stmt))
def berechne_stand(db: Session, meisterschaft_id: int) -> MeisterschaftsBericht:
def berechne_stand(
db: Session,
meisterschaft_id: int,
*,
kategorie: Kategorie = Kategorie.STAMM,
klasse_id: int | None = None,
) -> MeisterschaftsBericht:
"""Stand EINER Wertung der Jahresmeisterschaft."""
m = db.get(Meisterschaft, meisterschaft_id)
if m is None:
raise ValueError(f"Meisterschaft {meisterschaft_id} nicht gefunden")
@@ -134,29 +192,27 @@ def berechne_stand(db: Session, meisterschaft_id: int) -> MeisterschaftsBericht:
if regelwerk.get("_hinweis"):
hinweise.append(str(regelwerk["_hinweis"]))
# Turniere dieser Jahresmeisterschaft, deren Veranstaltung (vor-)final ist
turniere = list(
db.scalars(
select(Turnier)
.join(Veranstaltung, Veranstaltung.id == Turnier.veranstaltung_id)
.where(
Veranstaltung.datum.is_not(None),
Turnier.zaehlt_meisterschaft.is_(True),
Turnier.meisterschaft_id == meisterschaft_id,
Veranstaltung.status.in_(["vorlaeufig", "final"]),
)
.order_by(Veranstaltung.datum, Turnier.id)
)
)
# nur Turniere der Saison der Meisterschaft
turniere = [t for t in turniere if t.veranstaltung.datum.year == m.saison_jahr]
turnier_ids = [t.id for t in turniere]
roh: dict[int, dict[int, Decimal]] = {} # fahrer_id -> {turnier_id: punkte}
siege: dict[int, int] = {}
if turnier_ids:
for s in _relevante_starts(db, m, turnier_ids):
for s in _relevante_starts(db, turnier_ids, kategorie, klasse_id):
p = punkte_fuer_platz(s.ergebnis.platzierung, punkte_tab)
bucket = roh.setdefault(s.fahrer_id, {})
# Mehrfachstart in derselben Kategorie am selben Tag: bestes Ergebnis zaehlt
# Mehrfachstart am selben Tag: bestes Ergebnis zaehlt
bucket[s.turnier_id] = max(bucket.get(s.turnier_id, Decimal(0)), p)
if s.ergebnis.platzierung == 1:
siege[s.fahrer_id] = siege.get(s.fahrer_id, 0) + 1
@@ -185,7 +241,9 @@ def berechne_stand(db: Session, meisterschaft_id: int) -> MeisterschaftsBericht:
return MeisterschaftsBericht(
meisterschaft_id=meisterschaft_id,
saison_jahr=m.saison_jahr,
kategorie=m.kategorie.value,
wertung="erwachsene" if kategorie != Kategorie.JUGEND else _klassencode(db, klasse_id),
kategorie=kategorie.value,
klasse_id=klasse_id,
laeufe_gesamt=len(turnier_ids),
laeufe_gewertet=n_gewertet,
zeilen=zeilen,
@@ -193,6 +251,24 @@ def berechne_stand(db: Session, meisterschaft_id: int) -> MeisterschaftsBericht:
)
def _klassencode(db: Session, klasse_id: int | None) -> str:
if klasse_id is None:
return "jugend-gesamt"
k = db.get(Klasse, klasse_id)
return k.code if k else str(klasse_id)
def berechne_alle_wertungen(db: Session, meisterschaft_id: int) -> list[MeisterschaftsBericht]:
berichte = []
for w in wertungen_fuer(db, meisterschaft_id):
berichte.append(
berechne_stand(
db, meisterschaft_id, kategorie=w.kategorie, klasse_id=w.klasse_id
)
)
return berichte
@dataclass
class QualiVorschlag:
fahrer_id: int
@@ -203,17 +279,23 @@ class QualiVorschlag:
def endlauf_qualifikation(
db: Session, meisterschaft_id: int, plaetze: int = 10
db: Session,
meisterschaft_id: int,
*,
kategorie: Kategorie = Kategorie.STAMM,
klasse_id: int | None = None,
plaetze: int = 10,
) -> list[QualiVorschlag]:
"""Vorschlag Top-N fuer den Endlauf. Bei Punktgleichheit an der Grenze wird
``tie_break_relevant`` gesetzt — die endgueltige Festlegung trifft der
Bereichsleiter (Plan §3.4)."""
"""Vorschlag Top-N fuer den Endlauf EINER Wertung. Bei Punktgleichheit an der
Grenze wird ``tie_break_relevant`` gesetzt — die endgueltige Festlegung trifft
der Bereichsleiter (Plan §3.4)."""
bericht = berechne_stand(db, meisterschaft_id)
zeilen = bericht.zeilen[: plaetze + 5] # etwas Puffer fuer die Grenzbetrachtung
bericht = berechne_stand(db, meisterschaft_id, kategorie=kategorie, klasse_id=klasse_id)
grenz_punkte = (
bericht.zeilen[plaetze - 1].punkte_gesamt if len(bericht.zeilen) >= plaetze else None
)
vorschlaege: list[QualiVorschlag] = []
grenz_punkte = bericht.zeilen[plaetze - 1].punkte_gesamt if len(bericht.zeilen) >= plaetze else None
for z in zeilen:
for z in bericht.zeilen[: plaetze + 5]:
vorschlaege.append(
QualiVorschlag(
fahrer_id=z.fahrer_id,
+40 -13
View File
@@ -1,9 +1,18 @@
"""Meisterschafts-/Pokalwertung (HTH) — Plan §3.4.
Modell (Vorgabe des Auftraggebers): **pro Saison genau EINE Meisterschaft**. Zu
dieser Meisterschaft gehoeren die Turniere, die in die Jahreswertung zaehlen
(``turnier.meisterschaft_id``). Innerhalb der einen Meisterschaft gibt es mehrere
**Wertungen**:
* Erwachsene: ein Gesamtklassement A + S (klassenuebergreifend)
* Jugend: Pokalwertung klassenweise J 1 - J 5
* Streichresultate nach Tabelle in ``saison.streichresultate_tabelle``
* Tie-Break: Majoritaet der Siege
Diese Wertungen sind keine eigenen Tabellen, sondern (Kategorie, Klasse)-Sichten
auf die eine Meisterschaft — abgebildet als Spalten in ``meisterschaft_stand``.
* Streichresultate nach ``saison.streichresultate_tabelle``
* Tie-Break bei Punktgleichheit: Majoritaet der Siege
* Die Anwendung *schlaegt* die Endlaufqualifikation *vor*, entscheidet nicht.
"""
@@ -11,7 +20,7 @@ from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, Integer, Numeric, String, func
from sqlalchemy import DateTime, ForeignKey, Integer, Numeric, String, UniqueConstraint, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
@@ -20,39 +29,57 @@ from oamc.models.enums import Kategorie
class Meisterschaft(Base):
"""Genau eine pro Saison. Container fuer die zaehlenden Turniere und das
Regelwerk (Punktetabelle, Streichregel-Override, Endlauf-Parameter)."""
__tablename__ = "meisterschaft"
__table_args__ = (UniqueConstraint("saison_jahr", name="meisterschaft_saison_eindeutig"),)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
saison_jahr: Mapped[int] = mapped_column(ForeignKey("saison.jahr"), nullable=False, index=True)
bezeichnung: Mapped[str] = mapped_column(String(200), nullable=False)
kategorie: Mapped[Kategorie] = mapped_column(
sa_enum(Kategorie, "meisterschaft_kategorie"), nullable=False
saison_jahr: Mapped[int] = mapped_column(
ForeignKey("saison.jahr"), nullable=False, index=True, unique=True
)
# nur bei Jugend gesetzt (klassenweise J1..J5); bei Erwachsenen NULL
klasse_id: Mapped[int | None] = mapped_column(ForeignKey("klasse.id"))
# Punktetabelle (Platz -> Punkte), Streichregel-Override, Endlauf-Parameter.
bezeichnung: Mapped[str] = mapped_column(String(200), nullable=False)
# Punktetabelle (Platz -> Punkte), Streich-Override, Endlauf-Parameter.
# Solange O-1 offen ist, steht hier die (klar markierte) Platzhalter-Tabelle.
regelwerk: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
turniere: Mapped[list[Turnier]] = relationship( # noqa: F821
back_populates="meisterschaft"
)
staende: Mapped[list[MeisterschaftStand]] = relationship(
back_populates="meisterschaft", cascade="all, delete-orphan"
)
class MeisterschaftStand(Base):
"""Materialisiert — nach jedem Turnier neu berechnet."""
"""Materialisiert — nach jedem Turnier neu berechnet, je **Wertung**
(Kategorie + optional Klasse) innerhalb der einen Meisterschaft."""
__tablename__ = "meisterschaft_stand"
__table_args__ = (
UniqueConstraint(
"meisterschaft_id",
"kategorie",
"klasse_id",
"fahrer_id",
name="meisterschaft_stand_eindeutig",
),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
meisterschaft_id: Mapped[int] = mapped_column(
ForeignKey("meisterschaft.id", ondelete="CASCADE"), nullable=False, index=True
)
# Wertung: Erwachsene-Gesamt (kategorie=S, klasse_id=NULL) oder Jugend je Klasse
kategorie: Mapped[Kategorie] = mapped_column(
sa_enum(Kategorie, "meisterschaft_stand_kategorie"), nullable=False
)
klasse_id: Mapped[int | None] = mapped_column(ForeignKey("klasse.id"))
fahrer_id: Mapped[int] = mapped_column(ForeignKey("fahrer.id"), nullable=False, index=True)
punkte_gesamt: Mapped[float] = mapped_column(Numeric(9, 2), nullable=False, default=0)
# {turnier_id: punkte}
punkte_je_lauf: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
# Liste der turnier_ids, die als Streichresultat NICHT zaehlen
gestrichene_laeufe: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
anzahl_siege: Mapped[int] = mapped_column(Integer, nullable=False, default=0) # Tie-Break
platz: Mapped[int | None] = mapped_column(Integer)
+13 -2
View File
@@ -57,8 +57,11 @@ class Turnier(Base, TimestampMixin):
)
name: Mapped[str] = mapped_column(String(120), nullable=False) # "Reinheim" / "Hainstadt"
veranstalter_verein_id: Mapped[int | None] = mapped_column(ForeignKey("verein.id"))
# Nicht jede Veranstaltung zaehlt zur Meisterschaft ("o. M.", Plan §3.4).
zaehlt_meisterschaft: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
# Pro Saison gibt es EINE Meisterschaft; ein Turnier gehoert zu ihr, wenn es
# zaehlt. NULL = "o. M." (ohne Meisterschaft, Plan §3.4).
meisterschaft_id: Mapped[int | None] = mapped_column(
ForeignKey("meisterschaft.id", ondelete="SET NULL"), index=True
)
# Genau ein Turnier kann als "aktiv" markiert sein -> Geraete-API-Endpunkte
# /turniere/aktiv/*. Partieller Unique-Index in der Migration.
ist_aktiv: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
@@ -69,6 +72,9 @@ class Turnier(Base, TimestampMixin):
)
veranstaltung: Mapped[Veranstaltung] = relationship(back_populates="turniere")
meisterschaft: Mapped[Meisterschaft | None] = relationship( # noqa: F821
back_populates="turniere"
)
wertungsklassen: Mapped[list[Wertungsklasse]] = relationship(
back_populates="turnier", cascade="all, delete-orphan", order_by="Wertungsklasse.sortierung"
)
@@ -76,6 +82,11 @@ class Turnier(Base, TimestampMixin):
back_populates="turnier", cascade="all, delete-orphan"
)
@property
def zaehlt_meisterschaft(self) -> bool:
"""Kompatibilitaets-Sicht: zaehlt zur Jahresmeisterschaft?"""
return self.meisterschaft_id is not None
class Wertungsklasse(Base):
"""Pro Turnier zusammengefasste Klassen (Plan §3.2): z. B. "S 5, S 9, S 3, S 6, S 7".
+7 -16
View File
@@ -123,24 +123,15 @@ def seed_saison_2026(db: Session) -> dict[str, int]:
"_hinweis": "O-1 offen: PLATZHALTER-Punktetabelle, nicht offiziell!",
}
# Genau EINE Meisterschaft pro Saison. Die Turniere werden ihr spaeter je
# Turnier zugeordnet (turnier.meisterschaft_id); die Teilwertungen (Erwachsene,
# Jugend J 1-J 5) sind Sichten darauf, keine eigenen Zeilen.
_, neu = _get_or_create(
db, Meisterschaft, saison_jahr=2026, bezeichnung="HTH Erwachsene 2026",
defaults={"kategorie": Kategorie.STAMM, "regelwerk": platzhalter_regelwerk},
db,
Meisterschaft,
saison_jahr=2026,
defaults={"bezeichnung": "HTH-Meisterschaft 2026", "regelwerk": platzhalter_regelwerk},
)
zaehler["meisterschaft"] += int(neu)
jugendklassen = db.scalars(
select(Klasse).where(Klasse.saison_jahr == 2026, Klasse.kategorie == Kategorie.JUGEND)
).all()
for k in jugendklassen:
_, neu = _get_or_create(
db, Meisterschaft, saison_jahr=2026, bezeichnung=f"HTH Jugend {k.code} 2026",
defaults={
"kategorie": Kategorie.JUGEND,
"klasse_id": k.id,
"regelwerk": platzhalter_regelwerk,
},
)
zaehler["meisterschaft"] += int(neu)
return zaehler
@@ -1,8 +1,17 @@
{% extends "base.html" %}
{% block title %}Meisterschaftsstand · {{ projekt }}{% endblock %}
{% block title %}{{ meisterschaft.bezeichnung }} · {{ projekt }}{% endblock %}
{% block content %}
<h1>Meisterschaftsstand</h1>
<p>Saison {{ bericht.saison_jahr }} · Kategorie {{ bericht.kategorie }} ·
<h1>{{ meisterschaft.bezeichnung }}</h1>
<nav class="tiles">
{% for w in wertungen %}
<a href="/meisterschaft/{{ meisterschaft.id }}?wertung={{ w.code | urlencode }}"
{% if w.code == aktive_wertung.code %}style="font-weight:700"{% endif %}>{{ w.bezeichnung }}</a>
{% endfor %}
</nav>
<h2>{{ aktive_wertung.bezeichnung }}</h2>
<p>Saison {{ bericht.saison_jahr }} ·
{{ bericht.laeufe_gewertet }} von {{ bericht.laeufe_gesamt }} Läufen gewertet
(Rest = Streichresultate)</p>
@@ -5,7 +5,7 @@
<ul class="plain">
{% for m in meisterschaften %}
<li><a href="/meisterschaft/{{ m.id }}">{{ m.bezeichnung }}</a>
<span class="badge">{{ m.kategorie.value }}</span></li>
<span class="badge">Saison {{ m.saison_jahr }}</span></li>
{% else %}
<li>Keine Meisterschaft angelegt.</li>
{% endfor %}
+10 -2
View File
@@ -10,6 +10,7 @@ from oamc.models import (
Ergebnis,
Fahrer,
Klasse,
Meisterschaft,
Start,
Turnier,
Veranstaltung,
@@ -18,6 +19,11 @@ from oamc.models import (
from oamc.models.enums import StartStatus, TurnierStatus, VeranstaltungStatus
def meisterschaft(db, jahr=2026):
"""Die eine Meisterschaft der Saison (aus dem Seed)."""
return db.scalars(select(Meisterschaft).where(Meisterschaft.saison_jahr == jahr)).one()
def klasse(db, code="J 3"):
return db.scalars(select(Klasse).where(Klasse.code == code, Klasse.saison_jahr == 2026)).one()
@@ -33,11 +39,13 @@ def veranstaltung(db, *, datum=date(2026, 5, 31), titel="Testturnier", status=Ve
return v
def turnier(db, v=None, *, name="Reinheim", zaehlt_meisterschaft=True, aktiv=False,
def turnier(db, v=None, *, name="Reinheim", meisterschaft=None, aktiv=False,
status=TurnierStatus.LAUFEND, flush=True):
"""``meisterschaft``: Meisterschaft-Objekt (zaehlt zur Jahreswertung) oder None."""
v = v or veranstaltung(db)
t = Turnier(
veranstaltung_id=v.id, name=name, zaehlt_meisterschaft=zaehlt_meisterschaft,
veranstaltung_id=v.id, name=name,
meisterschaft_id=meisterschaft.id if meisterschaft is not None else None,
ist_aktiv=aktiv, status=status,
)
db.add(t)
+33
View File
@@ -102,3 +102,36 @@ def test_finalisieren_setzt_status_final(client):
r = client.post(f"/api/v1/turniere/{tid}/finalisieren", headers=HEAD)
assert r.status_code == 200
assert client.db.get(Turnier, tid).status.value == "final"
def test_turnier_wird_an_jahresmeisterschaft_gehaengt(client):
db = client.db
seed_saison_2026(db)
v = f.veranstaltung(db)
db.flush()
r = client.post("/api/v1/turniere", headers=HEAD, json={
"veranstaltung_id": v.id, "name": "Reinheim", "zaehlt_meisterschaft": True,
})
assert r.status_code == 201, r.text
m = f.meisterschaft(db)
assert r.json()["meisterschaft_id"] == m.id
# o. M. -> keine Verknuepfung
r2 = client.post("/api/v1/turniere", headers=HEAD, json={
"veranstaltung_id": v.id, "name": "Pokal o. M.", "zaehlt_meisterschaft": False,
})
assert r2.json()["meisterschaft_id"] is None
def test_meisterschaft_stand_liefert_alle_wertungen(client):
db = client.db
seed_saison_2026(db)
m = f.meisterschaft(db)
r = client.get(f"/api/v1/meisterschaften/{m.id}/stand")
assert r.status_code == 200
codes = {w["wertung"] for w in r.json()["wertungen"]}
assert "erwachsene" in codes
# eine einzelne Wertung gezielt abrufen
r2 = client.get(f"/api/v1/meisterschaften/{m.id}/stand", params={"wertung": "J 3"})
assert r2.status_code == 200
assert r2.json()["wertung"] == "J 3"
+48 -28
View File
@@ -1,18 +1,16 @@
"""Meisterschaftsberechnung (AP 3) — mit PLATZHALTER-Punktetabelle (O-1 offen)."""
"""Meisterschaftsberechnung (AP 3) — eine Meisterschaft pro Saison, mehrere
Wertungen darin. PLATZHALTER-Punktetabelle (O-1 offen)."""
from datetime import date
from decimal import Decimal
from sqlalchemy import select
from oamc.meisterschaft.berechnung import (
PLATZHALTER_PUNKTETABELLE,
berechne_stand,
gewertete_laeufe,
punkte_fuer_platz,
wertungen_fuer,
)
from oamc.models import Meisterschaft
from oamc.models.enums import VeranstaltungStatus
from oamc.models.enums import Kategorie, VeranstaltungStatus
from tests import factories as f
@@ -30,10 +28,19 @@ def test_streichresultate_tabelle():
assert gewertete_laeufe(3, tab) == 3 # unterhalb der Tabelle: kein Streichresultat
def _lauf(db, k, ergebnisse, *, datum, zaehlt=True):
def test_eine_meisterschaft_pro_saison_mit_wertungen(seed_2026):
db = seed_2026
m = f.meisterschaft(db)
wertungen = wertungen_fuer(db, m.id)
codes = {w.code for w in wertungen}
assert "erwachsene" in codes
assert {"J 1", "J 2", "J 3", "J 4", "J 5"} <= codes
def _lauf(db, m, k, ergebnisse, *, datum, zaehlt=True):
"""ergebnisse = [(fahrer, platz), ...]"""
v = f.veranstaltung(db, datum=datum, status=VeranstaltungStatus.FINAL)
t = f.turnier(db, v, zaehlt_meisterschaft=zaehlt)
t = f.turnier(db, v, meisterschaft=m if zaehlt else None)
for i, (fahrer, platz) in enumerate(ergebnisse, start=1):
s = f.start(db, t, fahrer, k, startnummer=i)
e = f.ergebnis(db, s, zeit=Decimal("100"), fehler=0)
@@ -44,48 +51,61 @@ def _lauf(db, k, ergebnisse, *, datum, zaehlt=True):
def test_stand_summiert_punkte_und_zaehlt_siege(seed_2026):
db = seed_2026
m = f.meisterschaft(db)
k = f.klasse(db, "J 3")
a = f.fahrer(db, "Sieger", "Sina", 2014)
b = f.fahrer(db, "Zweite", "Zoe", 2014)
_lauf(db, k, [(a, 1), (b, 2)], datum=date(2026, 4, 1))
_lauf(db, k, [(a, 1), (b, 2)], datum=date(2026, 5, 1))
m = db.scalars(
select(Meisterschaft).where(Meisterschaft.bezeichnung == "HTH Jugend J 3 2026")
).one()
bericht = berechne_stand(db, m.id)
_lauf(db, m, k, [(a, 1), (b, 2)], datum=date(2026, 4, 1))
_lauf(db, m, k, [(a, 1), (b, 2)], datum=date(2026, 5, 1))
bericht = berechne_stand(db, m.id, kategorie=Kategorie.JUGEND, klasse_id=k.id)
zeilen = {z.fahrer_id: z for z in bericht.zeilen}
assert zeilen[a.id].platz == 1
assert zeilen[a.id].punkte_gesamt == Decimal(50) # 2x Platz 1 = 2x25
assert zeilen[a.id].anzahl_siege == 2
assert bericht.hinweise # O-1 Platzhalter-Warnung
assert bericht.wertung == "J 3"
def test_wertungen_sind_getrennt(seed_2026):
db = seed_2026
m = f.meisterschaft(db)
j3, j4 = f.klasse(db, "J 3"), f.klasse(db, "J 4")
kind_j3 = f.fahrer(db, "Drei", "D", 2014)
kind_j4 = f.fahrer(db, "Vier", "V", 2012)
v = f.veranstaltung(db, status=VeranstaltungStatus.FINAL)
t = f.turnier(db, v, meisterschaft=m)
for i, (fahrer, kl) in enumerate([(kind_j3, j3), (kind_j4, j4)], start=1):
s = f.start(db, t, fahrer, kl, startnummer=i)
e = f.ergebnis(db, s, zeit=Decimal("100"), fehler=0)
e.platzierung = 1
db.flush()
stand_j3 = berechne_stand(db, m.id, kategorie=Kategorie.JUGEND, klasse_id=j3.id)
stand_j4 = berechne_stand(db, m.id, kategorie=Kategorie.JUGEND, klasse_id=j4.id)
assert [z.fahrer_id for z in stand_j3.zeilen] == [kind_j3.id]
assert [z.fahrer_id for z in stand_j4.zeilen] == [kind_j4.id]
def test_tie_break_majoritaet_der_siege(seed_2026):
db = seed_2026
m = f.meisterschaft(db)
k = f.klasse(db, "J 3")
a = f.fahrer(db, "Aaa", "A", 2014)
b = f.fahrer(db, "Bbb", "B", 2014)
# beide sammeln gleich viele Punkte, aber A hat mehr Siege
_lauf(db, k, [(a, 1), (b, 6)], datum=date(2026, 4, 1)) # A:25 B:10
_lauf(db, k, [(b, 1), (a, 6)], datum=date(2026, 5, 1)) # B:25 A:10
_lauf(db, k, [(a, 1), (b, 1)], datum=date(2026, 6, 1)) # beide 25, beide Sieg
_lauf(db, k, [(a, 1), (b, 4)], datum=date(2026, 6, 15)) # A:25 (Sieg) B:13
m = db.scalars(
select(Meisterschaft).where(Meisterschaft.bezeichnung == "HTH Jugend J 3 2026")
).one()
bericht = berechne_stand(db, m.id)
_lauf(db, m, k, [(a, 1), (b, 6)], datum=date(2026, 4, 1)) # A:25 B:10
_lauf(db, m, k, [(b, 1), (a, 6)], datum=date(2026, 5, 1)) # B:25 A:10
_lauf(db, m, k, [(a, 1), (b, 1)], datum=date(2026, 6, 1)) # beide 25, beide Sieg
_lauf(db, m, k, [(a, 1), (b, 4)], datum=date(2026, 6, 15)) # A:25 (Sieg) B:13
bericht = berechne_stand(db, m.id, kategorie=Kategorie.JUGEND, klasse_id=k.id)
assert bericht.zeilen[0].fahrer_id == a.id
assert bericht.zeilen[0].anzahl_siege > bericht.zeilen[1].anzahl_siege
def test_laeufe_ohne_meisterschaft_zaehlen_nicht(seed_2026):
db = seed_2026
m = f.meisterschaft(db)
k = f.klasse(db, "J 3")
a = f.fahrer(db, "Solo", "S", 2014)
_lauf(db, k, [(a, 1)], datum=date(2026, 4, 1), zaehlt=False)
m = db.scalars(
select(Meisterschaft).where(Meisterschaft.bezeichnung == "HTH Jugend J 3 2026")
).one()
bericht = berechne_stand(db, m.id)
_lauf(db, m, k, [(a, 1)], datum=date(2026, 4, 1), zaehlt=False)
bericht = berechne_stand(db, m.id, kategorie=Kategorie.JUGEND, klasse_id=k.id)
assert bericht.laeufe_gesamt == 0
assert bericht.zeilen == []
+10 -6
View File
@@ -30,15 +30,19 @@ def test_seed_saison_streichresultate(seed_2026):
assert s.streichresultate_tabelle["4"] == 2
def test_seed_meisterschaften(seed_2026):
def test_seed_meisterschaft(seed_2026):
db = seed_2026
ms = db.scalars(select(Meisterschaft).where(Meisterschaft.saison_jahr == 2026)).all()
bezeichnungen = {m.bezeichnung for m in ms}
assert "HTH Erwachsene 2026" in bezeichnungen
assert "HTH Jugend J 3 2026" in bezeichnungen
# genau EINE Meisterschaft pro Saison
assert len(ms) == 1
assert ms[0].bezeichnung == "HTH-Meisterschaft 2026"
# O-1: Platzhalter-Punktetabelle ist markiert
erw = next(m for m in ms if m.bezeichnung == "HTH Erwachsene 2026")
assert "_hinweis" in erw.regelwerk
assert "_hinweis" in ms[0].regelwerk
from oamc.meisterschaft.berechnung import wertungen_fuer
codes = {w.code for w in wertungen_fuer(db, ms[0].id)}
assert "erwachsene" in codes and "J 3" in codes
def test_seed_vereins_kurznamen(seed_2026):