Andrew Kesterson
4ee4ced6eb
Add timezone-aware timestamp support and reporting improvements Store timestamps in ISO-8601 format with timezone offsets Accept client-local timestamps from browser and API Normalize legacy naive timestamps during parsing Add migration script for converting old SQLite timestamps Fix mixed naive/aware datetime comparison errors Render timestamps in browser-local timezone Auto-generate default 7-day reports on page load Expand default dashboard report to include all users Preserve manual single-user report generation Add curl-based validation/test script for API and report flows
415 lines
9.0 KiB
Python
415 lines
9.0 KiB
Python
#!/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
|
|
)
|