timeclock/app.py

#!/usr/bin/env python3

"""
Minimal Time Clock Application
"""

import sqlite3

from datetime import datetime, timedelta, timezone

LEGACY_TIMEZONE = timezone.utc

from flask import (
    Flask,
    request,
    render_template,
    jsonify
)

# -----------------------------------------------------------------------------
# Flask Configuration
# -----------------------------------------------------------------------------

app = Flask(__name__)

DATABASE = "database/timeclock.db"

# -----------------------------------------------------------------------------
# Database Helpers
# -----------------------------------------------------------------------------

def get_db_connection():
    """
    Create and return a sqlite3 database connection.
    """

    conn = sqlite3.connect(DATABASE)
    conn.row_factory = sqlite3.Row

    return conn


def initialize_database():
    """
    Create database tables if they do not already exist.
    """

    conn = get_db_connection()

    cursor = conn.cursor()

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS entries (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            entrytype TEXT NOT NULL CHECK(entrytype IN ('in', 'out')),
            ts TEXT NOT NULL,
            FOREIGN KEY(user_id) REFERENCES users(id)
        )
    """)

    conn.commit()

    cursor.execute("SELECT COUNT(*) AS count FROM users")

    result = cursor.fetchone()

    if result["count"] == 0:
        cursor.execute("INSERT INTO users (name) VALUES ('Alice')")
        cursor.execute("INSERT INTO users (name) VALUES ('Bob')")
        cursor.execute("INSERT INTO users (name) VALUES ('Charlie')")

        conn.commit()

    conn.close()

# -----------------------------------------------------------------------------
# Time Helpers
# -----------------------------------------------------------------------------

def parse_iso_datetime(value):
    """
    Parse timestamps into timezone-aware datetime objects.

    Handles both:
        2026-05-29 08:00:00
    and:
        2026-05-29T08:00:00-04:00
    """

    #
    # Support legacy sqlite format:
    #   2026-05-29 08:00:00
    #

    normalized = value.replace(" ", "T")

    dt = datetime.fromisoformat(normalized)

    #
    # If timestamp is naive, assume legacy timezone.
    #

    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=LEGACY_TIMEZONE)

    return dt

def utc_now():
    """
    Return current UTC timestamp.
    """

    return datetime.now(timezone.utc)


# -----------------------------------------------------------------------------
# Utility Functions
# -----------------------------------------------------------------------------

def create_entry(user_id, entry_type, client_timestamp=None):
    """
    Create a clock in/out entry for the specified user.
    """

    conn = get_db_connection()

    cursor = conn.cursor()

    if client_timestamp:
        timestamp = parse_iso_datetime(client_timestamp)
    else:
        timestamp = utc_now()

    cursor.execute("""
        INSERT INTO entries (user_id, entrytype, ts)
        VALUES (?, ?, ?)
    """, (
        user_id,
        entry_type,
        timestamp.isoformat()
    ))

    conn.commit()

    conn.close()


def generate_report(user_id, begin_date, end_date):
    """
    Generate total worked hours for a user between two dates.
    """

    conn = get_db_connection()

    cursor = conn.cursor()

    begin_dt = parse_iso_datetime(begin_date)
    end_dt = parse_iso_datetime(end_date)

    cursor.execute("""
        SELECT entrytype, ts
        FROM entries
        WHERE user_id = ?
        ORDER BY ts ASC
    """, (user_id,))

    rows = cursor.fetchall()

    conn.close()

    total_seconds = 0

    clock_in_time = None

    for row in rows:

        timestamp = parse_iso_datetime(row["ts"])

        if timestamp < begin_dt or timestamp > end_dt:
            continue

        if row["entrytype"] == "in":
            clock_in_time = timestamp

        elif row["entrytype"] == "out" and clock_in_time is not None:

            delta = timestamp - clock_in_time

            total_seconds += delta.total_seconds()

            clock_in_time = None

    total_hours = total_seconds / 3600.0

    return round(total_hours, 2)

# -----------------------------------------------------------------------------
# Web UI Routes
# -----------------------------------------------------------------------------

@app.route("/", methods=["GET", "POST"])
def index():
    """
    Main application page.
    """

    conn = get_db_connection()

    users = conn.execute("""
        SELECT id, name
        FROM users
        ORDER BY name
    """).fetchall()

    clocked_in_users = conn.execute("""
        SELECT u.id, u.name, last_entry.ts
        FROM users u
        JOIN (
            SELECT e.user_id, e.entrytype, e.ts
            FROM entries e
            INNER JOIN (
                SELECT user_id, MAX(ts) AS max_ts
                FROM entries
                GROUP BY user_id
            ) latest
            ON e.user_id = latest.user_id
            AND e.ts = latest.max_ts
        ) last_entry
        ON u.id = last_entry.user_id
        WHERE last_entry.entrytype = 'in'
        ORDER BY u.name
    """).fetchall()

    conn.close()

    clocked_in_list = []

    for user in clocked_in_users:

        ts = parse_iso_datetime(user["ts"])

        clocked_in_list.append({
            "id": user["id"],
            "name": user["name"],
            "since": ts.isoformat()
        })

    report_hours = None

    all_user_reports = []

    selected_user_id = None

    now = utc_now()

    default_begin = now - timedelta(days=7)

    default_end = now

    if request.method == "POST":

        selected_user_id = request.form.get("user_id")

        action = request.form.get("action")

        client_timestamp = request.form.get("client_timestamp")

        if action == "clock_in":

            create_entry(
                selected_user_id,
                "in",
                client_timestamp
            )

        elif action == "clock_out":

            create_entry(
                selected_user_id,
                "out",
                client_timestamp
            )

        elif action == "report":

            begin_date = request.form.get("begin_date")

            end_date = request.form.get("end_date")

            report_hours = generate_report(
                selected_user_id,
                begin_date,
                end_date
            )

            default_begin = parse_iso_datetime(begin_date)

            default_end = parse_iso_datetime(end_date)

    else:

        #
        # Automatically generate last 7 day report
        # for ALL users
        #

        for user in users:

            hours = generate_report(
                user["id"],
                default_begin.isoformat(),
                default_end.isoformat()
            )

            all_user_reports.append({
                "id": user["id"],
                "name": user["name"],
                "hours": hours
            })

    return render_template(
        "index.html",
        users=users,
        report_hours=report_hours,
        all_user_reports=all_user_reports,
        clocked_in_users=clocked_in_list,
        default_begin=default_begin.isoformat(timespec="minutes"),
        default_end=default_end.isoformat(timespec="minutes"),
        selected_user_id=selected_user_id
    )

# -----------------------------------------------------------------------------
# CRUD API
# -----------------------------------------------------------------------------

@app.route("/api/users", methods=["GET"])
def api_get_users():

    conn = get_db_connection()

    users = conn.execute("""
        SELECT id, name
        FROM users
    """).fetchall()

    conn.close()

    return jsonify([dict(user) for user in users])


@app.route("/api/entries", methods=["GET"])
def api_get_entries():

    conn = get_db_connection()

    entries = conn.execute("""
        SELECT id, user_id, entrytype, ts
        FROM entries
        ORDER BY ts DESC
    """).fetchall()

    conn.close()

    return jsonify([dict(entry) for entry in entries])


@app.route("/api/entries", methods=["POST"])
def api_create_entry():

    data = request.get_json()

    user_id = data.get("user_id")

    entry_type = data.get("entrytype")

    client_timestamp = data.get("timestamp")

    if entry_type not in ("in", "out"):

        return jsonify({
            "error": "Invalid entrytype"
        }), 400

    create_entry(
        user_id,
        entry_type,
        client_timestamp
    )

    return jsonify({
        "status": "success"
    }), 201

# -----------------------------------------------------------------------------
# Application Entry Point
# -----------------------------------------------------------------------------

if __name__ == "__main__":

    initialize_database()

    app.run(
        host="0.0.0.0",
        port=5000,
        debug=False
    )