332 lines
9.6 KiB
Python
332 lines
9.6 KiB
Python
import os
|
|
import secrets
|
|
import sqlite3
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
|
|
from dotenv import load_dotenv
|
|
from flask import (
|
|
Flask,
|
|
abort,
|
|
g,
|
|
jsonify,
|
|
redirect,
|
|
render_template,
|
|
request,
|
|
session,
|
|
url_for,
|
|
)
|
|
|
|
load_dotenv()
|
|
|
|
|
|
def create_app() -> Flask:
|
|
app = Flask(__name__, static_folder="static", template_folder="templates")
|
|
|
|
secret_key = os.environ.get("SECRET_KEY")
|
|
if not secret_key:
|
|
# Safe default for local dev; in production set SECRET_KEY in .env
|
|
secret_key = secrets.token_urlsafe(32)
|
|
app.secret_key = secret_key
|
|
|
|
app.config["DB_PATH"] = os.environ.get("DB_PATH", "./data/app.db")
|
|
app.config["AUTH_USERNAME"] = os.environ.get("AUTH_USERNAME", "")
|
|
app.config["AUTH_PASSWORD"] = os.environ.get("AUTH_PASSWORD", "")
|
|
|
|
# Ensure DB directory exists.
|
|
# - If DB_PATH is a bare filename (no directory), dirname() is "".
|
|
# - If DB_PATH is absolute (e.g. /data/app.db), dirname() is "/data".
|
|
db_dir = os.path.dirname(app.config["DB_PATH"]) or "."
|
|
os.makedirs(db_dir, exist_ok=True)
|
|
|
|
@app.before_request
|
|
def _open_db():
|
|
g.db = sqlite3.connect(app.config["DB_PATH"])
|
|
g.db.row_factory = sqlite3.Row
|
|
|
|
@app.after_request
|
|
def _close_db(resp):
|
|
db = getattr(g, "db", None)
|
|
if db is not None:
|
|
db.close()
|
|
return resp
|
|
|
|
def init_db() -> None:
|
|
db = sqlite3.connect(app.config["DB_PATH"])
|
|
try:
|
|
db.executescript(
|
|
"""
|
|
PRAGMA journal_mode=WAL;
|
|
PRAGMA foreign_keys=ON;
|
|
|
|
CREATE TABLE IF NOT EXISTS todo_items (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
text TEXT NOT NULL,
|
|
done INTEGER NOT NULL DEFAULT 0,
|
|
created_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS expenses (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
description TEXT NOT NULL,
|
|
jpy REAL NOT NULL,
|
|
eur REAL NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
);
|
|
"""
|
|
)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
init_db()
|
|
|
|
def is_authed() -> bool:
|
|
return bool(session.get("authed"))
|
|
|
|
def require_auth() -> None:
|
|
if not is_authed():
|
|
abort(401)
|
|
|
|
@app.get("/")
|
|
def index():
|
|
if not is_authed():
|
|
return redirect(url_for("login"))
|
|
return render_template("index.html")
|
|
|
|
@app.get("/login")
|
|
def login():
|
|
if is_authed():
|
|
return redirect(url_for("index"))
|
|
return render_template("login.html")
|
|
|
|
@app.post("/login")
|
|
def login_post():
|
|
username = (request.form.get("username") or "").strip()
|
|
password = request.form.get("password") or ""
|
|
|
|
if not app.config["AUTH_USERNAME"] or not app.config["AUTH_PASSWORD"]:
|
|
# Misconfiguration: refuse login if credentials are not set.
|
|
abort(500)
|
|
|
|
if username == app.config["AUTH_USERNAME"] and password == app.config["AUTH_PASSWORD"]:
|
|
session.clear()
|
|
session["authed"] = True
|
|
return redirect(url_for("index"))
|
|
|
|
return render_template("login.html", error="Credenziali non valide")
|
|
|
|
@app.post("/logout")
|
|
def logout():
|
|
session.clear()
|
|
return redirect(url_for("login"))
|
|
|
|
# -----------------
|
|
# API: TODO
|
|
# -----------------
|
|
|
|
@app.get("/api/todos")
|
|
def api_todos_list():
|
|
require_auth()
|
|
rows = g.db.execute(
|
|
"SELECT id, text, done, created_at FROM todo_items ORDER BY id DESC"
|
|
).fetchall()
|
|
return jsonify(
|
|
[
|
|
{
|
|
"id": r["id"],
|
|
"text": r["text"],
|
|
"done": bool(r["done"]),
|
|
"created_at": r["created_at"],
|
|
}
|
|
for r in rows
|
|
]
|
|
)
|
|
|
|
@app.post("/api/todos")
|
|
def api_todos_create():
|
|
require_auth()
|
|
data = request.get_json(silent=True) or {}
|
|
text = (data.get("text") or "").strip()
|
|
if not text:
|
|
return jsonify({"error": "text is required"}), 400
|
|
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
cur = g.db.execute(
|
|
"INSERT INTO todo_items(text, done, created_at) VALUES(?, 0, ?)",
|
|
(text, now),
|
|
)
|
|
g.db.commit()
|
|
return jsonify({"id": cur.lastrowid, "text": text, "done": False, "created_at": now}), 201
|
|
|
|
@app.patch("/api/todos/<int:todo_id>")
|
|
def api_todos_update(todo_id: int):
|
|
require_auth()
|
|
data = request.get_json(silent=True) or {}
|
|
|
|
fields = []
|
|
params = []
|
|
|
|
if "text" in data:
|
|
text = (data.get("text") or "").strip()
|
|
if not text:
|
|
return jsonify({"error": "text cannot be empty"}), 400
|
|
fields.append("text = ?")
|
|
params.append(text)
|
|
|
|
if "done" in data:
|
|
done = 1 if bool(data.get("done")) else 0
|
|
fields.append("done = ?")
|
|
params.append(done)
|
|
|
|
if not fields:
|
|
return jsonify({"error": "no fields to update"}), 400
|
|
|
|
params.append(todo_id)
|
|
cur = g.db.execute(
|
|
f"UPDATE todo_items SET {', '.join(fields)} WHERE id = ?",
|
|
tuple(params),
|
|
)
|
|
g.db.commit()
|
|
if cur.rowcount == 0:
|
|
return jsonify({"error": "not found"}), 404
|
|
return jsonify({"ok": True})
|
|
|
|
@app.delete("/api/todos/<int:todo_id>")
|
|
def api_todos_delete(todo_id: int):
|
|
require_auth()
|
|
cur = g.db.execute("DELETE FROM todo_items WHERE id = ?", (todo_id,))
|
|
g.db.commit()
|
|
if cur.rowcount == 0:
|
|
return jsonify({"error": "not found"}), 404
|
|
return jsonify({"ok": True})
|
|
|
|
# -----------------
|
|
# API: EXPENSES
|
|
# -----------------
|
|
|
|
@app.get("/api/expenses")
|
|
def api_expenses_list():
|
|
require_auth()
|
|
rows = g.db.execute(
|
|
"SELECT id, description, jpy, eur, created_at FROM expenses ORDER BY id DESC"
|
|
).fetchall()
|
|
return jsonify(
|
|
[
|
|
{
|
|
"id": r["id"],
|
|
"description": r["description"],
|
|
"jpy": r["jpy"],
|
|
"eur": r["eur"],
|
|
"created_at": r["created_at"],
|
|
}
|
|
for r in rows
|
|
]
|
|
)
|
|
|
|
@app.post("/api/expenses")
|
|
def api_expenses_create():
|
|
require_auth()
|
|
data = request.get_json(silent=True) or {}
|
|
description = (data.get("description") or "").strip()
|
|
jpy = data.get("jpy")
|
|
eur = data.get("eur")
|
|
|
|
if not description:
|
|
return jsonify({"error": "description is required"}), 400
|
|
|
|
try:
|
|
jpy_f = float(jpy)
|
|
eur_f = float(eur)
|
|
except (TypeError, ValueError):
|
|
return jsonify({"error": "jpy and eur must be numbers"}), 400
|
|
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
cur = g.db.execute(
|
|
"INSERT INTO expenses(description, jpy, eur, created_at) VALUES(?, ?, ?, ?)",
|
|
(description, jpy_f, eur_f, now),
|
|
)
|
|
g.db.commit()
|
|
return (
|
|
jsonify(
|
|
{
|
|
"id": cur.lastrowid,
|
|
"description": description,
|
|
"jpy": jpy_f,
|
|
"eur": eur_f,
|
|
"created_at": now,
|
|
}
|
|
),
|
|
201,
|
|
)
|
|
|
|
@app.patch("/api/expenses/<int:expense_id>")
|
|
def api_expenses_update(expense_id: int):
|
|
require_auth()
|
|
data = request.get_json(silent=True) or {}
|
|
|
|
fields = []
|
|
params = []
|
|
|
|
if "description" in data:
|
|
description = (data.get("description") or "").strip()
|
|
if not description:
|
|
return jsonify({"error": "description cannot be empty"}), 400
|
|
fields.append("description = ?")
|
|
params.append(description)
|
|
|
|
if "jpy" in data:
|
|
try:
|
|
jpy_f = float(data.get("jpy"))
|
|
except (TypeError, ValueError):
|
|
return jsonify({"error": "jpy must be a number"}), 400
|
|
fields.append("jpy = ?")
|
|
params.append(jpy_f)
|
|
|
|
if "eur" in data:
|
|
try:
|
|
eur_f = float(data.get("eur"))
|
|
except (TypeError, ValueError):
|
|
return jsonify({"error": "eur must be a number"}), 400
|
|
fields.append("eur = ?")
|
|
params.append(eur_f)
|
|
|
|
if not fields:
|
|
return jsonify({"error": "no fields to update"}), 400
|
|
|
|
params.append(expense_id)
|
|
cur = g.db.execute(
|
|
f"UPDATE expenses SET {', '.join(fields)} WHERE id = ?",
|
|
tuple(params),
|
|
)
|
|
g.db.commit()
|
|
if cur.rowcount == 0:
|
|
return jsonify({"error": "not found"}), 404
|
|
return jsonify({"ok": True})
|
|
|
|
@app.delete("/api/expenses/<int:expense_id>")
|
|
def api_expenses_delete(expense_id: int):
|
|
require_auth()
|
|
cur = g.db.execute("DELETE FROM expenses WHERE id = ?", (expense_id,))
|
|
g.db.commit()
|
|
if cur.rowcount == 0:
|
|
return jsonify({"error": "not found"}), 404
|
|
return jsonify({"ok": True})
|
|
|
|
# -----------------
|
|
# Errors
|
|
# -----------------
|
|
|
|
@app.errorhandler(401)
|
|
def _unauthorized(_e):
|
|
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", "8000")), debug=False)
|