launchpad/app.py

import os
import sqlite3
from datetime import datetime
from pathlib import Path
from urllib.parse import urlparse

from flask import (
    Flask, render_template, request, redirect, url_for, flash, session, jsonify, abort
)

# ----------------------------------------------------------------------------
# Config
# ----------------------------------------------------------------------------
BASE_DIR = Path(__file__).resolve().parent
DB_PATH = BASE_DIR / "launchpad.db"
UPLOADS_DIR = BASE_DIR / "static" / "uploads"
ALLOWED_IMAGE_EXT = {"png", "jpg", "jpeg", "webp", "gif"}


def create_app():
    app = Flask(__name__, template_folder=str(BASE_DIR / "templates"), static_folder=str(BASE_DIR / "static"))

    # Minimal secrets; override in .env or environment
    app.config["SECRET_KEY"] = os.getenv("SECRET_KEY", "dev-secret-change-me")
    app.config["ADMIN_USER"] = os.getenv("ADMIN_USER", "admin")
    app.config["ADMIN_PASS"] = os.getenv("ADMIN_PASS", "admin123")
    app.config["PUBLIC_BASE_URL"] = os.getenv("PUBLIC_BASE_URL", "").rstrip("/")

    # Ensure DB exists
    init_db()
    # Ensure uploads dir exists
    UPLOADS_DIR.mkdir(parents=True, exist_ok=True)

    # ---------------------------
    # Helpers
    # ---------------------------
    def get_db():
        return sqlite3.connect(DB_PATH)

    def login_required(fn):
        from functools import wraps
        @wraps(fn)
        def wrapper(*args, **kwargs):
            if not session.get("user"):
                return redirect(url_for("login", next=request.path))
            return fn(*args, **kwargs)
        return wrapper

    def validate_url(u: str) -> bool:
        try:
            p = urlparse(u)
            return p.scheme in {"http", "https"} and bool(p.netloc)
        except Exception:
            return False

    def make_absolute(img_path_or_url: str) -> str:
        """Return an absolute https/http URL for images. If already absolute, return as is.
        If it's a /static/... path, prefix with PUBLIC_BASE_URL if set, otherwise request.url_root.
        """
        if not img_path_or_url:
            return img_path_or_url
        low = img_path_or_url.lower()
        if low.startswith("http://") or low.startswith("https://"):
            return img_path_or_url
        # Use configured public base if provided; else request.url_root (requires request context)
        base = app.config.get("PUBLIC_BASE_URL") or request.url_root.rstrip('/')
        if not img_path_or_url.startswith('/'):
            img_path_or_url = '/' + img_path_or_url
        return f"{base}{img_path_or_url}"

    def allowed_image(filename: str) -> bool:
        return "." in filename and filename.rsplit(".", 1)[1].lower() in ALLOWED_IMAGE_EXT

    def save_image(file_storage) -> str | None:
        """Save uploaded image to static/uploads and return its web path beginning with /static/..."""
        from werkzeug.utils import secure_filename
        if not file_storage or not file_storage.filename:
            return None
        if not allowed_image(file_storage.filename):
            return None
        name = secure_filename(file_storage.filename)
        # unique filename
        ts = datetime.utcnow().strftime("%Y%m%d%H%M%S%f")
        ext = name.rsplit(".", 1)[1].lower()
        base = name.rsplit(".", 1)[0]
        final_name = f"{base}_{ts}.{ext}"
        dest = UPLOADS_DIR / final_name
        file_storage.save(dest)
        return f"/static/uploads/{final_name}"

    # ---------------------------
    # Auth
    # ---------------------------
    @app.route("/login", methods=["GET", "POST"])
    def login():
        if request.method == "POST":
            username = request.form.get("username", "").strip()
            password = request.form.get("password", "")
            if username == app.config["ADMIN_USER"] and password == app.config["ADMIN_PASS"]:
                session["user"] = username
                flash("Welcome!", "success")
                return redirect(request.args.get("next") or url_for("apps_list"))
            flash("Invalid credentials", "danger")
        return render_template("login.html")

    @app.route("/logout")
    def logout():
        session.clear()
        flash("Logged out", "info")
        return redirect(url_for("login"))

    # ---------------------------
    # UI Pages
    # ---------------------------
    @app.route("/")
    @login_required
    def apps_list():
        query = "SELECT id, name, image, url, enabled, created_at FROM apps ORDER BY priority DESC, name ASC"
        with get_db() as conn:
            conn.row_factory = sqlite3.Row
            rows = conn.execute(query).fetchall()
        return render_template("apps_list.html", apps=rows)

    @app.route("/apps/new", methods=["GET", "POST"])
    @login_required
    def app_create():
        if request.method == "POST":
            name = request.form.get("name", "").strip()
            image_file = request.files.get("image_file")  # NEW
            image = request.form.get("image", "").strip()
            url = request.form.get("url", "").strip()
            enabled = 1 if request.form.get("enabled") == "on" else 0
            priority = int(request.form.get("priority", 0) or 0)

            if not name:
                flash("Name is required", "danger")
                return render_template("app_form.html", form_action=url_for("app_create"))

            # Prefer uploaded file over URL
            saved_image_path = save_image(image_file) if image_file else None
            if saved_image_path:
                image = saved_image_path
            elif image and not validate_url(image):
                flash("Image must be a valid http(s) URL or upload a file", "danger")
                return render_template("app_form.html", form_action=url_for("app_create"))

            if not validate_url(url):
                flash("App URL must be a valid http(s) URL", "danger")
                return render_template("app_form.html", form_action=url_for("app_create"))

            # Ensure image stored as absolute URL
            image = make_absolute(image) if image else ""

            with get_db() as conn:
                conn.execute(
                    "INSERT INTO apps (name, image, url, enabled, priority, created_at) VALUES (?,?,?,?,?,?)",
                    (name, image, url, enabled, priority, datetime.utcnow().isoformat(timespec='seconds')),
                )
                conn.commit()
            flash("App added", "success")
            return redirect(url_for("apps_list"))

        return render_template("app_form.html", form_action=url_for("app_create"))

    @app.route("/apps/<int:app_id>/edit", methods=["GET", "POST"])
    @login_required
    def app_edit(app_id: int):
        with get_db() as conn:
            conn.row_factory = sqlite3.Row
            row = conn.execute("SELECT id, name, image, url, enabled, priority FROM apps WHERE id=?", (app_id,)).fetchone()
            if not row:
                abort(404)
        if request.method == "POST":
            name = request.form.get("name", "").strip()
            image_file = request.files.get("image_file")  # NEW
            url = request.form.get("url", "").strip()
            enabled = 1 if request.form.get("enabled") == "on" else 0
            priority = int(request.form.get("priority", 0) or 0)

            # Prefer uploaded file > new URL > keep old value
            new_image = row["image"]
            saved_image_path = save_image(image_file) if image_file else None
            if saved_image_path:
                new_image = saved_image_path
            else:
                form_image = request.form.get("image", "").strip()
                if form_image:
                    if not validate_url(form_image):
                        flash("Image must be a valid http(s) URL or upload a file", "danger")
                        return render_template("app_form.html", form_action=url_for("app_edit", app_id=app_id), data=row)
                    new_image = form_image

            if not name:
                flash("Name is required", "danger")
                return render_template("app_form.html", form_action=url_for("app_edit", app_id=app_id), data=row)
            if not validate_url(url):
                flash("App URL must be a valid http(s) URL", "danger")
                return render_template("app_form.html", form_action=url_for("app_edit", app_id=app_id), data=row)

            new_image = make_absolute(new_image) if new_image else ""

            with get_db() as conn:
                conn.execute(
                    "UPDATE apps SET name=?, image=?, url=?, enabled=?, priority=? WHERE id=?",
                    (name, new_image, url, enabled, priority, app_id),
                )
                conn.commit()
            flash("App updated", "success")
            return redirect(url_for("apps_list"))
        return render_template("app_form.html", form_action=url_for("app_edit", app_id=app_id), data=row)

    @app.route("/apps/<int:app_id>/delete", methods=["POST"]) 
    @login_required
    def app_delete(app_id: int):
        with get_db() as conn:
            conn.execute("DELETE FROM apps WHERE id=?", (app_id,))
            conn.commit()
        flash("App deleted", "warning")
        return redirect(url_for("apps_list"))

    # ---------------------------
    # API Endpoint (public or protected — here public for simplicity)
    # ---------------------------
    @app.route("/api/apps", methods=["GET"])
    def api_apps():
        with get_db() as conn:
            conn.row_factory = sqlite3.Row
            rows = conn.execute(
                "SELECT id, name, image, url FROM apps WHERE enabled=1 ORDER BY priority DESC, name ASC"
            ).fetchall()
        data = []
        for r in rows:
            d = dict(r)
            img = d.get("image") or ""
            d["image"] = make_absolute(img) if img else ""
            data.append(d)
        return jsonify({
            "version": 1,
            "updated_at": datetime.utcnow().isoformat(timespec='seconds') + "Z",
            "apps": data,
        })

    return app


# ----------------------------------------------------------------------------
# DB bootstrap
# ----------------------------------------------------------------------------
SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS apps (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    image TEXT,
    url TEXT NOT NULL,
    enabled INTEGER NOT NULL DEFAULT 1,
    priority INTEGER NOT NULL DEFAULT 0,
    created_at TEXT NOT NULL
);
"""


def init_db():
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("PRAGMA journal_mode=WAL;")
        conn.execute(SCHEMA_SQL)
        conn.commit()


# ----------------------------------------------------------------------------
# Dev entrypoint
# ----------------------------------------------------------------------------
if __name__ == "__main__":
    app = create_app()
    app.run(host="0.0.0.0", port=int(os.getenv("PORT", 5000)), debug=True)