Clean things up a bit
This commit is contained in:
184
app.py
184
app.py
@@ -1,15 +1,5 @@
|
||||
#!/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,
|
||||
@@ -17,45 +7,26 @@ from flask import (
|
||||
jsonify
|
||||
)
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Flask Configuration
|
||||
# -----------------------------------------------------------------------------
|
||||
LEGACY_TIMEZONE = timezone.utc
|
||||
|
||||
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,
|
||||
@@ -65,161 +36,76 @@ def initialize_database():
|
||||
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.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()
|
||||
))
|
||||
|
||||
""", (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
|
||||
@@ -238,94 +124,59 @@ def index():
|
||||
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,
|
||||
@@ -337,78 +188,51 @@ def index():
|
||||
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
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user