Errechnet von Zeitdifferenzen / Timestamps Tagesübergreifend

[cbesi]

Hallo,

ich habe den folgenden Code geschrieben, welcher auch funktioniert.

Hier wird aus einer Datenbank, Zeiten ausgelesen und daraus sollen Pausen sowie die Summe der Arbeitszeit ermittelt werden bis eine Differenz von 6 Std zwischen der letzten Abmeldung und der nächsten Anmeldung sind. Das funktioniert auch, mein Problem ist aber das ich es nicht hinbekomme, wenn tagesübergreifend gearbeitet wird:

Da mein Code endet, wenn die Zeitdifferenz größer 6 Std. ist.
Normal müsste an dem Tag noch einmal geschaut werden ob es nach den 6 Std. wieder eine Anmeldung gibt, diese Zeit müsste dann separat summiert werden.

Kann mir jemand helfen?

Code: Alles auswählen

import datetime
import firebirdsql

import industriezeit
import round_timestamp


def calc_worktime(fromdate,todate,zeitraumid):

    def calc_arbeitszeit_dezimal(arbeitszeit):
        print('Arbeitszeit:', arbeitszeit)
        #global dez_arbeitszeit
        #global dez_pause
        #global helfer
        dez_arbeitszeit = industriezeit.zeit_to_dezimalzeit(arbeitszeit)


        print('Pause:', pause)
        dez_pause = industriezeit.zeit_to_dezimalzeit(pause)
        print("############# DEZIMAL ##############")
        cur.execute("Select Vornamename from stammdatenlohn where helferid = '%s'"%zeitraumid)
        helfer = cur.fetchall()
        print(helfer[0])
        print("Pause-Dez:",dez_pause)
        print("Arbeitszeit-Dez:",dez_arbeitszeit)
        print("####################################")
        return [dez_arbeitszeit,dez_pause,helfer[0]]

    # Verbindung zur Firebird-Datenbank herstellen
    con = firebirdsql.connect(
        host='localhost', database='c:\\db\test).fdb',
        user='xyz', password='123456'
    )

    # Cursor erstellen
    cur = con.cursor()

    # SQL-Abfrage ausführen
    #cur.execute("SELECT * FROM zeitbuchungen where datum > '%s' and datum < '%s' and zeitraumid = '%s' ORDER BY datum, zeit" %(fromdate,todate,zeitraumid))
    cur.execute("SELECT * FROM zeitbuchungen where datum between '%s' and '%s' and zeitraumid = '%s' ORDER BY datum, zeit" %(fromdate,todate,zeitraumid))

    # Ergebnisse abrufen
    results = cur.fetchall()

    # Arbeitszeit und Pausen berechnen
    arbeitszeit = datetime.timedelta()
    pause = datetime.timedelta()
    last_timestamp = None
    last_date = None
    last_pause_begin = None
    abmeldung = None
    abmeldung2 = None
    pausenzahl = 0
    nextday = False

    ## Schleife um die Stempelungen auszuwerten
    for row in results:
        #print(row)
        timestamp = datetime.datetime.combine(row[2], row[3])
        #print("gerundet", round_timestamp.round_zeit(str(timestamp)))

        ##Erste Anmeldung wird gesucht
        if row[1] == 1:
            ##print(row[1], row[2], row[3], "Anmeldung")

            # last_timestamp = timestamp
            anmeldung = timestamp
            last_date = row[2]
            if abmeldung is not None and abmeldung is not 'ENDE':
                ##print("Abmeldung= ", abmeldung, "   Anmeldung= ", anmeldung)
                if (anmeldung - abmeldung).total_seconds() >= 6 * 60 * 60:
                    # Tag beenden
                    print("*************** Tag beenden")
                    letzte_Buchung = row
                    abmeldung = 'ENDE'
                    nextday = True
                    abbruchsrow = row
                    werte = calc_arbeitszeit_dezimal(arbeitszeit)

                else:
                    if last_timestamp is not None:
                        print("Pausen", timestamp, "-", last_timestamp)
                        pause += timestamp - last_timestamp
                        print(pause)
        else:
            ### Pausenzeit wird summiert
            if row[1] == 0 and abmeldung is not 'ENDE':
                # if last_timestamp is not None:
                print(row[1], row[2], row[3], "Abmeldung")
                abmeldung = datetime.datetime.combine(row[2], row[3])
                arbeitszeit += abmeldung - anmeldung
                last_timestamp = abmeldung

                print("Abmeldung= ", abmeldung, "   Anmeldung= ", anmeldung)








    print("Return-Daten",werte[0],werte[1],werte[2])
    #return dez_arbeitszeit,dez_pause,helfer[0]
    return werte[0],werte[1],werte[2]

    # Verbindung schließen
    #cur.close()

    con.close()

if __name__ == "__main__":
    fromdate = '18.07.2023'
    todate = '20.07.2023'
    zeitraumid = 6433

    calc_worktime(fromdate,todate,zeitraumid)

[Sirius3]

round_timestamp wird importiert aber nicht benutzt.
Man definiert keine Funktionen innerhalb von Funktionen; dort kommt auch ein Cursor vor, der gar nicht als Argument übergeben wird. Das selbe gilt für die Pause!
Man formatiert keine Werte in SQL-Statements hinein, dafür gibt es Platzhalter; welcher das für firebirdsql ist, mußt Du in deren Dokumentation nachschlagen.
Wenn man für eine Abfrage exakt einen Datensatz erwartet, benutzt man fetchone statt fetchall.
Es ist komisch, dass man über eine Zeitraum-ID zu einer Helfer-ID kommt, für mich sieht das nach einer falschen Abfrage aus.
Warum heißt das Tabellenfeld Vornamename? Gibt es auch einen Nachnamenamen?
In einer Tabelle stammdatenlohn würde ich Stammdaten zum Lohn erwarten und keine Vornamenamen für irgendwelche Helfer. Wer hat die Datenbank aufgebaut? Die mußt Du dringend überarbeiten, damit die Tabellen und Felder sinnvoll benannt sind, denn so herrscht totale Verwirrung.
Gibt es neben einer Pause im Dezember auch eine Pause im August?
Wenn man mehrere Rückgabewerte hat, packt man die in ein Tuple und nicht in eine Liste.

Sowohl Connections als auch Cursor sollten wieder geschlossen werden. Am besten per `with closing(...):`-Anweisung, denn Dein close steht nach dem `return´ und wird niemals ausgeführt.
In Zeile 40 formatierst Du dann Werte in ein SQL-Statement hinein, statt Parameter zu benutzen.

Man macht kein *-Select und greift dann auf die Felder per magischen Indizes zu, sondern gibt die Felder beim Select explizit an und benutzt dann Tuple-Unpacking in der for-Schleife.
row[2] und row[3] werden ja aus dem Zusammenhang irgendwie klar, aber was row[1] enthält? Scheint wohl irgendein Flag zu sein, ob man sich an- oder abmeldet.
Es scheint so, als ob Datum und Uhrzeit in zwei getrennten Datenbankfeldern stehen würden. Das macht man nicht, weil für Datum und Uhrzeit jede Datenbank einen passenden DATETIME oder TIMESTAMP-Datentyp kennt!

Du definierst etliche Variablen, von denen die meisten gar nicht gebraucht werden. Die können alle weg.

Außer mit None vergleicht man niemals Werte per `is` oder `is not`. Eine Variable sollte immer nur Werte eines Typs haben, abmeldung ist aber mal ein String und mal eine Zeit.

Du überschreibst mehrmals `werte`, und gibst nur den letzten Wert von `werte` zurück. Das scheint nicht richtig zu sein.


Wenn Du einen Betrieb mit Nachtarbeit hast, und das irgendwie abbilden möchtest, dann darfst Du nicht einfach nicht nach 6 Stunden abbrechen, sondern mußt auf andere Art das Ende des Arbeitstags ermitteln, z.B. weil die "Pause" mehr als 11h dauert.

[__blackjack__]

@cbesi: Bei einige Namen fehlen Unterstriche, dennTextewodieWortenichtgetrenntsind,lassensichschwererlesen.

Kommentare sollen dem Leser einen Mehrwert über den Code geben. Faustregel: Kommentare beschreiben nicht *was* der Code macht, denn das steht da bereits als Code, sondern warum er das macht. Sofern das nicht offensichtlich ist. Offensichtlich ist in aller Regel auch was in der Dokumentation von Python und den verwendeten Bibliotheken steht.

Der Kommentar ``##Erste Anmeldung wird gesucht`` dürfte falsch sein, denn die Schleife verarbeitet ja *alles* was die Abfrage liefert und sicht nicht nur die erste Anmeldung‽

Bei ``row[1]`` wäre echt interessant was das ist, denn der Code vergleicht auf die Zahlen 0 und 1. Ist das tatsächlich eine Zahl oder ist das ein Flag? Letzteres sollte man nicht mit 0 und 1 vergleichen, auch wenn das funktioniert, weil dem Leser vermittelt es ist eine Zahl und auch 2 oder 23, oder 42 wären vielleicht mögliche Werte. Zusammen mit dem Umstand das wir nicht wissen wie das in der Datenbank heisst, ist das ziemlich verwirrend.

Als erstes Datum und Zeit, was ja in der Datenbank eigentlich schon *ein* Wert sein sollte, zu einem Zeitstempel zusammenzusetzen ist ein guter Schritt. Dann sollte man den so erhaltenen Wert aber auch überall verwenden, und nicht innerhalb von dem Code die beiden Werte noch mal zum gleichen Wert zusammensetzen.

Auch wenn der Vergleich mit ``6 * 60 * 60`` halbwegs verständlich ist, würde man da doch eher einen `timedelta`-Wert verwenden. Ist die Bedingung da überhaupt korrekt? Der Tag ist zuende wenn das erste mal über 6 Stunden *am Stück* gearbeitet wurde, oder nicht doch eher wenn die Summe der Arbeitszeiten 6 Stunden überschreitet‽

Sirius3 schrieb das `werte` mehrmals überschrieben wird: Wird es gar nicht, aber das ist nicht einfach zu erkennen bei der Schleife und den vielen Bedingungen. `werte` wird einmal etwas zugewiesen, was dann bis zum Ende nicht mehr verändert wird und dann wird `werte` zurückgegeben. Also sollte man `werte` vielleicht besser `ergebnis` nennen, und eigentlich macht es keinen Sinn nach der einmaligen Zuweisung diese Schleife überhaupt weiterlaufen zu lassen.

Womit dann auch das komische "ENDE" was Sirius3 auch angemerkt hat, verschwinden kann, denn das da alles zuende ist, braucht man nirgends prüfen wenn die Funktion da tatsächlich einfach zuende abgearbeitet ist!

`abmeldung` und `last_timestamp` haben immer den gleichen Wert, da muss man also nicht zwei Namen für verwenden. Dann fällt auch auf, dass innerhalb eines ``if abmeldung is not None:`` ein ``if last_timestamp is not None:`` steht — was *immer* Wahr ist, weil die beiden Namen *immer* an den gleichen Wert gebunden sind.

Diese ganzen `print()`-Ausgaben von jedem Pupskleinen Zwischenergebnis machen das ganze auch schwer lesbar, weil man ständig von Code abgelenkt wird der gar nichts mit den eigentlichen Berechnungen zu tun hat, aber gefühlt mindestens 50% der Funktion ausmacht.

Wenn man den ganzen überflüssigen und unbenutzen Kram raus wirft, sieht das so aus (ungetestet):

Code: Alles auswählen

from datetime import date as Date, datetime as DateTime, timedelta as TimeDelta

import firebirdsql
import industriezeit


def calculate_worktime(from_date, to_date, zeitraum_id):
    with firebirdsql.connect(
        host="localhost",
        database=R"c:\\db\test).fdb",
        user="xyz",
        password="123456",
    ) as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                (
                    "SELECT wie_auch_immer_das_heisst, datum, zeit"
                    " FROM zeitbuchungen"
                    " WHERE datum BETWEEN ? AND ? AND zeitraumid = ?"
                    " ORDER BY datum, zeit"
                ),
                (from_date, to_date, zeitraum_id),
            )
            arbeitszeit = TimeDelta()
            pausenzeit = TimeDelta()
            abmeldung = None
            for wie_auch_immer_das_heisst, datum, zeit in cursor.fetchall():
                #
                # TODO Das sollte in der Datenbank bereits *ein* Wert sein.
                #
                timestamp = DateTime.combine(datum, zeit)
                if wie_auch_immer_das_heisst == 1:
                    anmeldung = timestamp
                    if abmeldung:
                        if anmeldung - abmeldung >= TimeDelta(hours=6):
                            break

                        pausenzeit += anmeldung - abmeldung

                elif wie_auch_immer_das_heisst == 0:
                    abmeldung = timestamp
                    arbeitszeit += abmeldung - anmeldung

            cursor.execute(
                "SELECT vornamename FROM stammdatenlohn WHERE helferid = ?",
                (zeitraum_id,),
            )
            return (
                industriezeit.zeit_to_dezimalzeit(arbeitszeit),
                industriezeit.zeit_to_dezimalzeit(pausenzeit),
                cursor.fetchall()[0],
            )


def main():
    print(calculate_worktime(Date(2023, 7, 18), Date(2023, 7, 20), 6433))


if __name__ == "__main__":
    main()

Die Platzhalter für SQL müsste man sich nicht raussuchen wenn man SQLAlchemy als Abstraktionsschicht verwendet. Minimal, ungetestet, mit „reflection“ des Datenbankinhaltes:

Code: Alles auswählen

from datetime import date as Date, datetime as DateTime, timedelta as TimeDelta

import firebirdsql
import industriezeit
from sqlalchemy import select, create_engine, MetaData


def calculate_worktime(database_url, from_date, to_date, zeitraum_id):
    engine = create_engine(database_url)
    meta_data = MetaData()
    meta_data.reflect(engine)
    zeitbuchung_table = meta_data.tables["zeitbuchungen"]
    lohnstammdaten_table = meta_data.tables["stammdatenlohn"]

    with engine.connect() as connection:
        arbeitszeit = TimeDelta()
        pausenzeit = TimeDelta()
        abmeldung = None
        for zeitbuchung in connection.execute(
            select(zeitbuchung_table)
            .where(
                zeitbuchung_table.c.datum.between(from_date, to_date),
                zeitbuchung_table.c.zeitraumid == zeitraum_id,
            )
            .order_by(zeitbuchung_table.c.datum, zeitbuchung_table.c.zeit)
        ):
            #
            # TODO Das sollte in der Datenbank bereits *ein* Wert sein.
            #
            timestamp = DateTime.combine(zeitbuchung.datum, zeitbuchung.zeit)
            if zeitbuchung.wie_auch_immer_das_heisst == 1:
                anmeldung = timestamp
                if abmeldung:
                    if anmeldung - abmeldung >= TimeDelta(hours=6):
                        break

                    pausenzeit += anmeldung - abmeldung

            elif zeitbuchung.wie_auch_immer_das_heisst == 0:
                abmeldung = timestamp
                arbeitszeit += abmeldung - anmeldung

        return (
            industriezeit.zeit_to_dezimalzeit(arbeitszeit),
            industriezeit.zeit_to_dezimalzeit(pausenzeit),
            connection.scalar(
                select(lohnstammdaten_table.c.vornamename).where(
                    lohnstammdaten_table.c.helferid == zeitraum_id
                )
            ),
        )


def main():
    print(
        calculate_worktime(
            "firebird+fdb://xyz:123456@localhost/c:/db/test).fdb",
            Date(2023, 7, 18),
            Date(2023, 7, 20),
            6433,
        )
    )


if __name__ == "__main__":
    main()

Wenn man die Tabellen nicht dynamisch ermittelt, sondern im Quelltext definiert, weiss der Leser mehr über den Datenbankaufbau. Und wenn man dann noch das ORM verwendet, kann man auch gleich komische Namen in der Datenbank auf sinnvolle(re) Namen im Programm abbilden (ungetestet):

Code: Alles auswählen

from datetime import date as Date, datetime as DateTime, timedelta as TimeDelta

import firebirdsql
import industriezeit
from sqlalchemy import (
    DATE,
    INTEGER,
    TEXT,
    TIME,
    Column,
    MetaData,
    create_engine,
    select,
)
from sqlalchemy.orm import Session, declarative_base

Base = declarative_base()


class Zeitbuchung(Base):
    id = Column(INTEGER, primary_key=True)
    #
    # TODO Oder BOOLEAN oder Enum?
    #
    wie_auch_immer_das_heisst = Column(INTEGER, nullable=False)
    #
    # TODO Die folgendenden beiden Felder solten in der Datenbank bereits *ein*
    #   Feld sein.
    #
    datum = Column(DATE, nullable=False)
    zeit = Column(TIME, nullable=False)
    zeitraum_id = Column("zeitraumid", INTEGER, nullable=False)

    @property
    def timestamp(self):
        return DateTime.combine(self.datum, self.zeit)


class Lohnstammdaten(Base):
    id = Column(INTEGER, primary_key=True)
    vorname = Column("vornamename", TEXT, nullable=False)
    ...
    helfer_id = Column("helferid", INTEGER, nullable=False, unique=True)


def calculate_worktime(database_url, from_date, to_date, zeitraum_id):
    with Session(create_engine(database_url)) as session:
        arbeitszeit = TimeDelta()
        pausenzeit = TimeDelta()
        abmeldung = None
        for zeitbuchung in session.execute(
            select(Zeitbuchung)
            .where(
                Zeitbuchung.datum.between(from_date, to_date),
                Zeitbuchung.zeitraum_id == zeitraum_id,
            )
            .order_by(Zeitbuchung.datum, Zeitbuchung.zeit)
        ):
            if zeitbuchung.wie_auch_immer_das_heisst == 1:
                anmeldung = zeitbuchung.timestamp
                if abmeldung:
                    if anmeldung - abmeldung >= TimeDelta(hours=6):
                        break

                    pausenzeit += anmeldung - abmeldung

            elif zeitbuchung.wie_auch_immer_das_heisst == 0:
                abmeldung = zeitbuchung.timestamp
                arbeitszeit += abmeldung - anmeldung

        return (
            industriezeit.zeit_to_dezimalzeit(arbeitszeit),
            industriezeit.zeit_to_dezimalzeit(pausenzeit),
            session.execute(
                select(Lohnstammdaten.vorname).where(
                    Lohnstammdaten.helfer_id == zeitraum_id
                )
            ).scalar_one(),
        )


def main():
    print(
        calculate_worktime(
            "firebird+fdb://xyz:123456@localhost/c:/db/test).fdb",
            Date(2023, 7, 18),
            Date(2023, 7, 20),
            6433,
        )
    )


if __name__ == "__main__":
    main()

Bei `zeitraum_id` und `helfer_id` hätte man in einer vollständigeren Abbildung der Tabellen dann noch `ForeignKey` und `relationship()` ”Deklarationen”.

[cbesi]

Hallo blackjack, und Sirius3,

danke für Eure Verbesserungsvorschläge, danke für die ausführlichen Erklärungen.

Ich habe mir dies gut durchgelesen und die Lernkurve ist definitiv deutlich gestiegen. DANKE dafür.

Zur Erklärung auf die Datenbank, habe ich keinen Einfluss, diese wird geliefert und darf nicht von mir verändert werden.

Adaptiert habe ich das erste Beispiel von blackjack, mit sqlalchemy bin ich noch so gar nicht vertraut.

Mein grundsätzliches Problem ist aber noch nicht gelöst, ich versuche es besser zu beschreiben.

wie_auch_immer_das_heisst = anmelden in der Tabelle


Wir haben den Fall:

18.07.2023, 00:07 Uhr, Anmelden =1
18.07.2023, 03:05 Uhr, Abmelden=0 (Pause beginnt)
18.07.2023, 03:35 Uhr Anmelden=1 (Pause ende)
18.07.2023, 08:35 Uhr Abmelden=0 (Pause2 beginnt)
18.07.2023, 09:05 Uhr Anmelden=1 (Pause2 ende)
18.07.2023, 10:34 Uhr Abmelden=0
----------------------------------------------------------------------------------------- Schichtende
18.07.2023, 23:48 Uhr Anmelden=1 (Schicht2 beginnt)
19.07.2023, 03:10 Uhr Abmelden=0 (Pause beginnt)
19.07.2023, 03:40 Uhr Anmelden=1
19.07.2023, 11:57 Uhr Abmelden=0
----------------------------------------------------------------------------------------- Schicht 2 vom 18.7

Mit dem Code schaffen wir es nun den ersten Abschnitt / Schicht 1 zu berechnen, ich habe aber das Problem, das ich nicht verstehe wie ich eine 2. Schicht ermitteln und auswerten kann.

Problem ist halt generell, das "Stempelungen" tagesübergreifend, sowie mehrere Schichten an einem Tag sein können (gesetzlich ist es richtig die Auszeit mit 11 Std sein / 6 Std in meinem Code sind gesetzlich nicht korrekt). Grundsätzlich sollen die Stunden immer der Tag sowie der Schicht vom Tag des Arbeitsbeginns zugeordnet werden können:

Mein Code sieht also nun so aus:
(Ja ich habe schon wieder ein nerviges print eingebaut, aber anhand der Ausgabe kann ich die Stimmigkeit leichter ermitteln)

Code: Alles auswählen

from datetime import date as Date, datetime as DateTime, timedelta as TimeDelta

import firebirdsql
import industriezeit


def calculate_worktime(from_date, to_date, zeitraum_id):
    with firebirdsql.connect(
        host="localhost",
        database=R"c:\\db\zu_ersetzen.fdb",
        user="admin",
        password="12345678",
    ) as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                (
                    "SELECT anmelden, datum, zeit"
                    " FROM zeitbuchungen"
                    " WHERE datum BETWEEN ? AND ? AND zeitraumid = ?"
                    " ORDER BY datum, zeit"
                ),
                (from_date, to_date, zeitraum_id),
            )
            arbeitszeit = TimeDelta()
            pausenzeit = TimeDelta()
            abmeldung = None
            for anmelden, datum, zeit in cursor.fetchall():
                print(anmelden, datum, zeit)
                #
                # TODO Das sollte in der Datenbank bereits *ein* Wert sein.
                #
                timestamp = DateTime.combine(datum, zeit)
                if anmelden == 1:
                    anmeldung = timestamp
                    if abmeldung:
                        if anmeldung - abmeldung >= TimeDelta(hours=6):
                            break

                        pausenzeit += anmeldung - abmeldung

                elif anmelden == 0:
                    abmeldung = timestamp
                    arbeitszeit += abmeldung - anmeldung

            cursor.execute(
                "SELECT vornamename FROM stammdatenlohn WHERE helferid = ?",
                (zeitraum_id,),
            )
            return (
                industriezeit.zeit_to_dezimalzeit(arbeitszeit),
                industriezeit.zeit_to_dezimalzeit(pausenzeit),
                cursor.fetchall()[0],
            )


def main():
    print(calculate_worktime(Date(2023, 7, 18), Date(2023, 7, 19), 6433))

if __name__ == "__main__":
    main()

Könnt Ihr mir an der Stelle auch noch einmal helfen?

[__blackjack__]

@cbesi: Ich habe als erstes mal die Ermittlung von Arbeits- und Pausenzeiten aus Datenbankeinträgen in eine eigene Funktion heraus gezogen. Damit ist das nicht mehr auf eine Datenbank angewiesen und man kann das unabhängig testen.

Und dann darf man nicht einfach aufhören wenn die 6 Stunden voll sind und *ein* Ergebnis zurückgeben, sondern muss alle Datensätze verarbeiten. Hier bietet sich eine Generatorfunktion an. Mal als erster Ansatz ein Algorithmus der einfach immer stumpf alle 6 Arbeiststunden eine neue Schicht anfängt:

Code: Alles auswählen

#!/usr/bin/env python3
from datetime import date as Date, datetime as DateTime, timedelta as TimeDelta

import parse

# import firebirdsql
# import industriezeit

TEST_LINE_FORMAT = "{date}, {time} Uhr A{n_or_b}melden={state:d}"
TEST_DATA = """\
18.07.2023, 00:07 Uhr Anmelden=1
18.07.2023, 03:05 Uhr Abmelden=0 (Pause beginnt)
18.07.2023, 03:35 Uhr Anmelden=1 (Pause ende)
18.07.2023, 08:35 Uhr Abmelden=0 (Pause2 beginnt)
18.07.2023, 09:05 Uhr Anmelden=1 (Pause2 ende)
18.07.2023, 10:34 Uhr Abmelden=0
----------------------------------------------------------------------------------------- Schichtende
18.07.2023, 23:48 Uhr Anmelden=1 (Schicht2 beginnt)
19.07.2023, 03:10 Uhr Abmelden=0 (Pause beginnt)
19.07.2023, 03:40 Uhr Anmelden=1
19.07.2023, 11:57 Uhr Abmelden=0
----------------------------------------------------------------------------------------- Schicht 2 vom 18.7""".splitlines()


def iter_test_data(lines):
    for line in lines:
        if not line.startswith("-"):
            match = parse.search(TEST_LINE_FORMAT, line)
            assert "bn"[match["state"]] == match["n_or_b"]
            yield (
                bool(match["state"]),
                DateTime.strptime(match["date"], "%d.%m.%Y").date(),
                DateTime.strptime(match["time"], "%H:%M").time(),
            )


def aggregate_shifts(log_entries):
    arbeitszeit = TimeDelta()
    pausenzeit = TimeDelta()
    abmeldung = None
    for anmelden, datum, zeit in log_entries:
        print(anmelden, datum, zeit)
        timestamp = DateTime.combine(datum, zeit)
        if anmelden == 1:
            anmeldung = timestamp
            if abmeldung:
                if anmeldung - abmeldung >= TimeDelta(hours=6):
                    yield arbeitszeit, pausenzeit
                    arbeitszeit = TimeDelta()
                    pausenzeit = TimeDelta()
                    abmeldung = None
                else:
                    pausenzeit += anmeldung - abmeldung

        elif anmelden == 0:
            abmeldung = timestamp
            arbeitszeit += abmeldung - anmeldung

    if arbeitszeit or pausenzeit:  # XXX Oder ``if abmeldung:``.
        yield arbeitszeit, pausenzeit


def calculate_worktime(from_date, to_date, zeitraum_id):
    with firebirdsql.connect(
        host="localhost",
        database=R"c:\\db\zu_ersetzen.fdb",
        user="admin",
        password="12345678",
    ) as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                (
                    "SELECT anmelden, datum, zeit"
                    " FROM zeitbuchungen"
                    " WHERE datum BETWEEN ? AND ? AND zeitraumid = ?"
                    " ORDER BY datum, zeit"
                ),
                (from_date, to_date, zeitraum_id),
            )
            schichten = [
                (
                    industriezeit.zeit_to_dezimalzeit(arbeitszeit),
                    industriezeit.zeit_to_dezimalzeit(pausenzeit),
                )
                for arbeitszeit, pausenzeit in aggregate_shifts(
                    cursor.fetchall()
                )
            ]

            cursor.execute(
                "SELECT vornamename FROM stammdatenlohn WHERE helferid = ?",
                (zeitraum_id,),
            )
            return (schichten, cursor.fetchall()[0])


def main():
    # print(calculate_worktime(Date(2023, 7, 18), Date(2023, 7, 19), 6433))
    print(list(iter_test_data(TEST_DATA)))
    print(list(aggregate_shifts(iter_test_data(TEST_DATA))))


if __name__ == "__main__":
    main()

Die Frage ist aber ob das überhaupt so einfach geht, ob man das “greedy” lösen kann, und ob das *überhaupt* eindeutig lösbar ist.

Auf jeden Fall ist das ein Fall für Unit-Tests.

[cbesi]

Hallo Blackjack,

ganz großes Danke an Dich.
Ich durfte mal wieder sehr viel lernen.
Konnte den Code adaptieren, habe diesen noch etwas erweitert, nun kommt das dabei raus, was ich haben wollte.

Mega!!!