364 lines
14 KiB
Python
364 lines
14 KiB
Python
import os
|
|
import secrets
|
|
import sqlite3
|
|
from datetime import date, datetime
|
|
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
|
|
from functools import wraps
|
|
from pathlib import Path
|
|
|
|
from flask import Flask, jsonify, redirect, render_template, request, session, url_for
|
|
import yaml
|
|
|
|
|
|
def normalize_url_prefix(value):
|
|
if not isinstance(value, str):
|
|
raise RuntimeError("flask.url_prefix must be a string")
|
|
value = value.strip()
|
|
if not value or value == "/":
|
|
return ""
|
|
return f"/{value.strip('/')}"
|
|
|
|
|
|
class UrlPrefixMiddleware:
|
|
"""Mount a WSGI application below a configurable URL prefix."""
|
|
|
|
def __init__(self, application, prefix):
|
|
self.application = application
|
|
self.prefix = prefix
|
|
|
|
def __call__(self, environ, start_response):
|
|
path = environ.get("PATH_INFO", "")
|
|
if self.prefix and path != self.prefix and not path.startswith(f"{self.prefix}/"):
|
|
start_response("404 Not Found", [("Content-Type", "text/plain; charset=utf-8")])
|
|
return [b"Not Found"]
|
|
|
|
if self.prefix:
|
|
environ["SCRIPT_NAME"] = f"{environ.get('SCRIPT_NAME', '')}{self.prefix}"
|
|
environ["PATH_INFO"] = path[len(self.prefix):] or "/"
|
|
return self.application(environ, start_response)
|
|
|
|
|
|
def load_yaml_config(path):
|
|
config_path = Path(path).expanduser().resolve()
|
|
if not config_path.is_file():
|
|
raise RuntimeError(
|
|
f"Configuration file not found: {config_path}. "
|
|
"Copy config.example.yaml to config.yaml and set its values."
|
|
)
|
|
|
|
with config_path.open(encoding="utf-8") as config_file:
|
|
values = yaml.safe_load(config_file) or {}
|
|
|
|
auth = values.get("auth", {})
|
|
flask_config = values.get("flask", {})
|
|
amount_config = values.get("amounts", {})
|
|
required = {
|
|
"SECRET_KEY": flask_config.get("secret_key"),
|
|
"APP_USERNAME": auth.get("username"),
|
|
"APP_PASSWORD": auth.get("password"),
|
|
"URL_PREFIX": normalize_url_prefix(flask_config.get("url_prefix", "/fatima")),
|
|
"LOG_FILE": flask_config.get("log_file", "/tmp/fatima.log"),
|
|
}
|
|
configured_database = flask_config.get("database")
|
|
if configured_database is not None:
|
|
if not isinstance(configured_database, str) or not configured_database.strip():
|
|
raise RuntimeError(f"{config_path}: flask.database must be a non-empty path")
|
|
database_path = Path(configured_database).expanduser()
|
|
if not database_path.is_absolute():
|
|
database_path = config_path.parent / database_path
|
|
required["DATABASE"] = str(database_path.resolve())
|
|
missing = [
|
|
name for name, value in required.items()
|
|
if name != "URL_PREFIX" and (not isinstance(value, str) or not value)
|
|
]
|
|
if missing:
|
|
raise RuntimeError(f"Missing or empty values in {config_path}: {', '.join(missing)}")
|
|
presets = amount_config.get("presets", ["55", "27.50"])
|
|
if not isinstance(presets, list) or len(presets) != 2:
|
|
raise RuntimeError(f"{config_path}: amounts.presets must contain exactly two amounts")
|
|
try:
|
|
normalized_presets = []
|
|
for preset in presets:
|
|
amount = Decimal(str(preset))
|
|
if not amount.is_finite() or amount <= 0:
|
|
raise InvalidOperation
|
|
normalized_presets.append(f"{amount:.2f}".rstrip("0").rstrip("."))
|
|
except (InvalidOperation, ValueError):
|
|
raise RuntimeError(f"{config_path}: amount presets must be positive numbers") from None
|
|
required["PRESET_AMOUNTS"] = normalized_presets
|
|
return required
|
|
|
|
|
|
def database_summary(database_path):
|
|
with sqlite3.connect(database_path) as database:
|
|
amount_count = database.execute("SELECT COUNT(*) FROM date_choices").fetchone()[0]
|
|
status_count = database.execute("SELECT COUNT(*) FROM date_statuses").fetchone()[0]
|
|
recorded_count, first_date, last_date = database.execute(
|
|
"""
|
|
SELECT COUNT(*), MIN(recorded_date), MAX(recorded_date) FROM (
|
|
SELECT choice_date AS recorded_date FROM date_choices
|
|
UNION
|
|
SELECT status_date AS recorded_date FROM date_statuses
|
|
)
|
|
"""
|
|
).fetchone()
|
|
return amount_count, status_count, recorded_count, first_date, last_date
|
|
|
|
|
|
def log_database_summary(database_path, log_path):
|
|
amounts, statuses, recorded_dates, first_date, last_date = database_summary(database_path)
|
|
date_range = f"{first_date} through {last_date}" if first_date else "no recorded dates"
|
|
message = (
|
|
"Fatima database: "
|
|
f"path={database_path} | "
|
|
f"amounts={amounts} | statuses={statuses} | "
|
|
f"recorded_dates={recorded_dates} | range={date_range}"
|
|
)
|
|
print(message, flush=True)
|
|
try:
|
|
with log_path.open("a", encoding="utf-8") as log_file:
|
|
timestamp = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
log_file.write(f"{timestamp} | {message}\n")
|
|
except OSError as error:
|
|
print(f"Could not write database summary to {log_path}: {error}", flush=True)
|
|
|
|
|
|
def create_app(test_config=None):
|
|
app = Flask(__name__)
|
|
config_path = os.environ.get("APP_CONFIG", Path(app.root_path) / "config.yaml")
|
|
app.config.from_mapping(load_yaml_config(config_path))
|
|
app.config.setdefault(
|
|
"DATABASE",
|
|
str(Path(__file__).resolve().parent / "instance" / "calendar.sqlite"),
|
|
)
|
|
|
|
# Environment variables remain useful when deploying without a local file.
|
|
for name in ("SECRET_KEY", "APP_USERNAME", "APP_PASSWORD", "DATABASE", "LOG_FILE"):
|
|
if os.environ.get(name):
|
|
app.config[name] = os.environ[name]
|
|
if "URL_PREFIX" in os.environ:
|
|
app.config["URL_PREFIX"] = normalize_url_prefix(os.environ["URL_PREFIX"])
|
|
if test_config:
|
|
app.config.update(test_config)
|
|
app.config["URL_PREFIX"] = normalize_url_prefix(app.config["URL_PREFIX"])
|
|
app.config["DATABASE"] = str(Path(app.config["DATABASE"]).expanduser().resolve())
|
|
app.config["LOG_FILE"] = str(Path(app.config["LOG_FILE"]).expanduser().resolve())
|
|
app.wsgi_app = UrlPrefixMiddleware(app.wsgi_app, app.config["URL_PREFIX"])
|
|
|
|
Path(app.config["DATABASE"]).parent.mkdir(parents=True, exist_ok=True)
|
|
with sqlite3.connect(app.config["DATABASE"]) as database:
|
|
database.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS date_choices (
|
|
choice_date TEXT PRIMARY KEY,
|
|
amount_cents INTEGER NOT NULL CHECK (amount_cents >= 0),
|
|
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
"""
|
|
)
|
|
database.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS date_statuses (
|
|
status_date TEXT PRIMARY KEY,
|
|
day_status TEXT NOT NULL DEFAULT 'worked'
|
|
CHECK (day_status IN ('worked', 'vacation', 'sick')),
|
|
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
"""
|
|
)
|
|
|
|
if not app.config.get("TESTING"):
|
|
log_database_summary(app.config["DATABASE"], Path(app.config["LOG_FILE"]))
|
|
|
|
def get_database():
|
|
connection = sqlite3.connect(app.config["DATABASE"])
|
|
connection.row_factory = sqlite3.Row
|
|
return connection
|
|
|
|
def login_required(view):
|
|
@wraps(view)
|
|
def wrapped_view(*args, **kwargs):
|
|
if not session.get("authorized"):
|
|
return redirect(url_for("login"))
|
|
return view(*args, **kwargs)
|
|
|
|
return wrapped_view
|
|
|
|
@app.route("/login", methods=("GET", "POST"))
|
|
def login():
|
|
if session.get("authorized"):
|
|
return redirect(url_for("calendar"))
|
|
|
|
error = None
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "")
|
|
password = request.form.get("password", "")
|
|
valid_user = secrets.compare_digest(username, app.config["APP_USERNAME"])
|
|
valid_password = secrets.compare_digest(password, app.config["APP_PASSWORD"])
|
|
if valid_user and valid_password:
|
|
session.clear()
|
|
session["authorized"] = True
|
|
return redirect(url_for("calendar"))
|
|
error = "That username or password is not correct."
|
|
|
|
return render_template("login.html", error=error)
|
|
|
|
@app.route("/")
|
|
@login_required
|
|
def calendar():
|
|
with get_database() as database:
|
|
choices = {
|
|
row["choice_date"]: row["amount_cents"]
|
|
for row in database.execute("SELECT choice_date, amount_cents FROM date_choices")
|
|
}
|
|
statuses = {
|
|
row["status_date"]: row["day_status"]
|
|
for row in database.execute("SELECT status_date, day_status FROM date_statuses")
|
|
}
|
|
return render_template(
|
|
"calendar.html",
|
|
today=date.today().isoformat(),
|
|
choices=choices,
|
|
statuses=statuses,
|
|
preset_amounts=app.config["PRESET_AMOUNTS"],
|
|
)
|
|
|
|
@app.post("/api/choices")
|
|
@login_required
|
|
def save_choice():
|
|
data = request.get_json(silent=True) or {}
|
|
choice_date = data.get("date", "")
|
|
try:
|
|
date.fromisoformat(choice_date)
|
|
except (TypeError, ValueError):
|
|
return jsonify(error="Enter a valid date."), 400
|
|
|
|
try:
|
|
amount = Decimal(str(data.get("amount", "")).replace(",", "."))
|
|
if not amount.is_finite() or amount < 0:
|
|
raise InvalidOperation
|
|
amount_cents = int((amount * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
|
|
except (InvalidOperation, ValueError):
|
|
return jsonify(error="Enter a valid amount of zero or more."), 400
|
|
|
|
with get_database() as database:
|
|
database.execute(
|
|
"""
|
|
INSERT INTO date_choices (choice_date, amount_cents)
|
|
VALUES (?, ?)
|
|
ON CONFLICT(choice_date) DO UPDATE SET
|
|
amount_cents = excluded.amount_cents,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
""",
|
|
(choice_date, amount_cents),
|
|
)
|
|
return jsonify(date=choice_date, amount_cents=amount_cents)
|
|
|
|
@app.post("/api/statuses")
|
|
@login_required
|
|
def save_status():
|
|
data = request.get_json(silent=True) or {}
|
|
status_date = data.get("date", "")
|
|
day_status = data.get("status", "")
|
|
try:
|
|
date.fromisoformat(status_date)
|
|
except (TypeError, ValueError):
|
|
return jsonify(error="Enter a valid date."), 400
|
|
if day_status not in {"worked", "vacation", "sick"}:
|
|
return jsonify(error="Choose worked, vacation, or sick."), 400
|
|
|
|
with get_database() as database:
|
|
database.execute(
|
|
"""
|
|
INSERT INTO date_statuses (status_date, day_status)
|
|
VALUES (?, ?)
|
|
ON CONFLICT(status_date) DO UPDATE SET
|
|
day_status = excluded.day_status,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
""",
|
|
(status_date, day_status),
|
|
)
|
|
return jsonify(date=status_date, status=day_status)
|
|
|
|
@app.post("/api/delete-date")
|
|
@login_required
|
|
def delete_date_data():
|
|
data = request.get_json(silent=True) or {}
|
|
selected_date = data.get("date", "")
|
|
try:
|
|
date.fromisoformat(selected_date)
|
|
except (TypeError, ValueError):
|
|
return jsonify(error="Enter a valid date."), 400
|
|
|
|
with get_database() as database:
|
|
database.execute("DELETE FROM date_choices WHERE choice_date = ?", (selected_date,))
|
|
database.execute("DELETE FROM date_statuses WHERE status_date = ?", (selected_date,))
|
|
return jsonify(date=selected_date, deleted=True)
|
|
|
|
@app.get("/report")
|
|
@login_required
|
|
def report():
|
|
current_year = request.args.get("year", default=date.today().year, type=int)
|
|
if current_year < 1900 or current_year > 9998:
|
|
return redirect(url_for("report"))
|
|
start_date = f"{current_year}-01-01"
|
|
end_date = f"{current_year + 1}-01-01"
|
|
with get_database() as database:
|
|
rows = database.execute(
|
|
"""
|
|
WITH recorded_dates AS (
|
|
SELECT choice_date AS selected_date FROM date_choices
|
|
WHERE choice_date >= ? AND choice_date < ?
|
|
UNION
|
|
SELECT status_date AS selected_date FROM date_statuses
|
|
WHERE status_date >= ? AND status_date < ?
|
|
)
|
|
SELECT
|
|
COALESCE(date_statuses.day_status, 'worked') AS category,
|
|
COUNT(*) AS day_count,
|
|
COALESCE(SUM(date_choices.amount_cents), 0) AS amount_cents
|
|
FROM recorded_dates
|
|
LEFT JOIN date_choices
|
|
ON date_choices.choice_date = recorded_dates.selected_date
|
|
LEFT JOIN date_statuses
|
|
ON date_statuses.status_date = recorded_dates.selected_date
|
|
GROUP BY category
|
|
""",
|
|
(start_date, end_date, start_date, end_date),
|
|
).fetchall()
|
|
|
|
values = {row["category"]: row for row in rows}
|
|
categories = [
|
|
{
|
|
"key": key,
|
|
"label": label,
|
|
"days": values[key]["day_count"] if key in values else 0,
|
|
"amount_cents": values[key]["amount_cents"] if key in values else 0,
|
|
}
|
|
for key, label in (
|
|
("worked", "Worked"),
|
|
("vacation", "Vacation"),
|
|
("sick", "Sick"),
|
|
)
|
|
]
|
|
return render_template(
|
|
"report.html",
|
|
year=current_year,
|
|
categories=categories,
|
|
total_days=sum(item["days"] for item in categories),
|
|
total_cents=sum(item["amount_cents"] for item in categories),
|
|
)
|
|
|
|
@app.post("/logout")
|
|
def logout():
|
|
session.clear()
|
|
return redirect(url_for("login"))
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=int(os.environ.get("PORT", "9009")))
|