| """ |
| Lab-Setup β Multi-page lab portal (Hugging Face Space) |
| Stack: Flask Β· Jinja2 Β· vanilla JS/CSS Β· gunicorn |
| """ |
|
|
| import os |
| import sqlite3 |
| from datetime import datetime, timezone |
| from flask import ( |
| Flask, render_template, request, redirect, |
| url_for, session, flash, jsonify, g, |
| ) |
| from werkzeug.middleware.proxy_fix import ProxyFix |
| from dotenv import load_dotenv |
| import markdown |
|
|
| load_dotenv() |
|
|
| app = Flask(__name__) |
| app.wsgi_app = ProxyFix(app.wsgi_app, x_proto=1, x_host=1) |
|
|
| app.secret_key = os.getenv("FLASK_SECRET", "change-me-in-prod") |
| app.config.update( |
| SESSION_COOKIE_HTTPONLY=True, |
| SESSION_COOKIE_SECURE=True, |
| SESSION_COOKIE_SAMESITE="None", |
| ) |
|
|
| ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "bpel123") |
|
|
| DB_PATH = os.getenv("DB_PATH", os.path.join(os.path.dirname(__file__), "lab.db")) |
|
|
| |
| OTREE_SESSION_URL = os.getenv( |
| "OTREE_SESSION_URL", |
| "http://otree-lab-games-790d4693d333.herokuapp.com/room/bpel_lab", |
| ) |
|
|
| |
| |
| |
| |
| |
| SIDEBAR_PAGES = [ |
| ("page_session", "π₯οΈ", "Session"), |
| ("page_logs", "π", "Lab Logs"), |
| ] |
|
|
|
|
| |
| def get_db(): |
| if "db" not in g: |
| g.db = sqlite3.connect(DB_PATH) |
| g.db.row_factory = sqlite3.Row |
| g.db.execute("PRAGMA journal_mode=WAL") |
| return g.db |
|
|
|
|
| @app.teardown_appcontext |
| def close_db(exc): |
| db = g.pop("db", None) |
| if db is not None: |
| db.close() |
|
|
|
|
| def init_db(): |
| db = sqlite3.connect(DB_PATH) |
| db.execute(""" |
| CREATE TABLE IF NOT EXISTS logs ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| title TEXT NOT NULL, |
| body TEXT NOT NULL DEFAULT '', |
| author TEXT NOT NULL DEFAULT '', |
| created_at TEXT NOT NULL, |
| updated_at TEXT NOT NULL |
| ) |
| """) |
| db.commit() |
| db.close() |
|
|
|
|
| init_db() |
|
|
|
|
| |
| @app.before_request |
| def require_login(): |
| allowed = ("login", "static") |
| if request.endpoint in allowed or (request.endpoint and request.endpoint.startswith("static")): |
| return |
| if not session.get("authenticated"): |
| return redirect(url_for("login")) |
|
|
|
|
| |
| @app.route("/login", methods=["GET", "POST"]) |
| def login(): |
| if request.method == "POST": |
| if request.form.get("password") == ADMIN_PASSWORD: |
| session["authenticated"] = True |
| return redirect(url_for("index")) |
| flash("Incorrect password.", "error") |
| return render_template("login.html") |
|
|
|
|
| @app.route("/logout") |
| def logout(): |
| session.clear() |
| return redirect(url_for("login")) |
|
|
|
|
| |
| @app.context_processor |
| def inject_sidebar(): |
| return dict(sidebar_pages=SIDEBAR_PAGES) |
|
|
|
|
| |
| @app.route("/") |
| def index(): |
| return redirect(url_for("page_session")) |
|
|
|
|
| @app.route("/session") |
| def page_session(): |
| return render_template( |
| "pages/session.html", |
| active_page="page_session", |
| otree_url=OTREE_SESSION_URL, |
| ) |
|
|
|
|
| |
| @app.route("/logs") |
| def page_logs(): |
| db = get_db() |
| rows = db.execute( |
| "SELECT * FROM logs ORDER BY created_at DESC" |
| ).fetchall() |
| logs = [] |
| for r in rows: |
| logs.append({ |
| "id": r["id"], |
| "title": r["title"], |
| "body": r["body"], |
| "body_html": markdown.markdown( |
| r["body"], extensions=["fenced_code", "tables", "nl2br"] |
| ), |
| "author": r["author"], |
| "created_at": r["created_at"], |
| "updated_at": r["updated_at"], |
| }) |
| return render_template("pages/logs.html", active_page="page_logs", logs=logs) |
|
|
|
|
| |
| @app.route("/api/logs", methods=["POST"]) |
| def api_logs_create(): |
| data = request.get_json(silent=True) or {} |
| title = (data.get("title") or "").strip() |
| if not title: |
| return jsonify(error="Title is required."), 400 |
| body = (data.get("body") or "").strip() |
| author = (data.get("author") or "").strip() |
| now = datetime.now(timezone.utc).isoformat() |
| db = get_db() |
| cur = db.execute( |
| "INSERT INTO logs (title, body, author, created_at, updated_at) VALUES (?,?,?,?,?)", |
| (title, body, author, now, now), |
| ) |
| db.commit() |
| return jsonify(id=cur.lastrowid), 201 |
|
|
|
|
| @app.route("/api/logs/<int:log_id>", methods=["PUT"]) |
| def api_logs_update(log_id): |
| data = request.get_json(silent=True) or {} |
| title = (data.get("title") or "").strip() |
| if not title: |
| return jsonify(error="Title is required."), 400 |
| body = (data.get("body") or "").strip() |
| author = (data.get("author") or "").strip() |
| now = datetime.now(timezone.utc).isoformat() |
| db = get_db() |
| db.execute( |
| "UPDATE logs SET title=?, body=?, author=?, updated_at=? WHERE id=?", |
| (title, body, author, now, log_id), |
| ) |
| db.commit() |
| return jsonify(ok=True) |
|
|
|
|
| @app.route("/api/logs/<int:log_id>", methods=["DELETE"]) |
| def api_logs_delete(log_id): |
| db = get_db() |
| db.execute("DELETE FROM logs WHERE id=?", (log_id,)) |
| db.commit() |
| return jsonify(ok=True) |
|
|
|
|
| |
| @app.route("/api/health") |
| def api_health(): |
| return jsonify(status="ok") |
|
|
|
|
| |
| if __name__ == "__main__": |
| app.run(host="0.0.0.0", port=5111, debug=True) |
|
|