#!/usr/bin/env python3 from __future__ import annotations import os from pathlib import Path from urllib.parse import urljoin import requests from flask import ( Flask, request, redirect, url_for, jsonify, send_from_directory, render_template, ) BASE = Path("/srv/www") STATIC_DIR = BASE / "static" STATIC_DIR.mkdir(exist_ok=True) # Where the Mirage API lives MIRAGE_API_BASE = os.environ.get( "MIRAGE_API_BASE", "http://127.0.0.1:5151/api/v1" ) # How mirrors are exposed over HTTP (for links) MIRROR_HTTP_BASE = os.environ.get("MIRROR_HTTP_BASE", "/mirrors/") app = Flask(__name__) def _api_url(path: str) -> str: return urljoin(MIRAGE_API_BASE.rstrip("/") + "/", path.lstrip("/")) def api_get(path: str, **kwargs): resp = requests.get(_api_url(path), timeout=5, **kwargs) resp.raise_for_status() return resp.json() def api_post(path: str, json=None, **kwargs): resp = requests.post(_api_url(path), json=json, timeout=10, **kwargs) return resp # -------------------- ROUTES -------------------- @app.route("/static/") def static_file(filename): return send_from_directory(STATIC_DIR, filename) @app.route("/", methods=["GET"]) def index(): # Ask Mirage for mirrors data = api_get("mirrors") mirrors = data.get("mirrors", []) cats = set() rows = [] for m in mirrors: categories = m.get("categories") or [] for c in categories: cats.add(c) raw = m.get("last_updated") disp = raw.replace("T", " ").replace("Z", " UTC") if raw else None rows.append({ "slug": m["slug"], "categories": categories, "categories_joined": ", ".join(categories), "url": m["url"], "status": m.get("status") or "idle", "last_updated_raw": raw, "last_updated": disp, }) return render_template( "index.html", mirrors=rows, categories=sorted(cats), error=None, ) @app.route("/add", methods=["POST"]) def add_mirror_route(): slug = (request.form.get("slug") or "").strip() categories_raw = (request.form.get("categories") or "").strip() url = (request.form.get("url") or "").strip() ignore_robots = bool(request.form.get("ignore_robots")) error = None if not slug or not categories_raw or not url: error = "Slug, categories, and URL are required." elif " " in slug: error = "Slug cannot contain spaces." categories = [c.strip() for c in categories_raw.split(",") if c.strip()] if error: # Same re-render pattern as before, but using Mirage API now data = api_get("mirrors") mirrors = data.get("mirrors", []) cats = set() rows = [] for m in mirrors: cs = m.get("categories") or [] for c in cs: cats.add(c) raw = m.get("last_updated") disp = raw.replace("T", " ").replace("Z", " UTC") if raw else None rows.append({ "slug": m["slug"], "categories": cs, "categories_joined": ", ".join(cs), "url": m["url"], "status": m.get("status") or "idle", "last_updated_raw": raw, "last_updated": disp, }) return ( render_template("index.html", mirrors=rows, categories=sorted(cats), error=error), 400, ) payload = { "slug": slug, "url": url, "categories": categories, "ignore_robots": ignore_robots, "enqueue": True, } resp = api_post("mirrors", json=payload) if resp.status_code >= 400: try: msg = resp.json().get("error", resp.text) except Exception: msg = resp.text or "Failed to create mirror." # Same error re-render as above data = api_get("mirrors") mirrors = data.get("mirrors", []) cats = set() rows = [] for m in mirrors: cs = m.get("categories") or [] for c in cs: cats.add(c) raw = m.get("last_updated") disp = raw.replace("T", " ").replace("Z", " UTC") if raw else None rows.append({ "slug": m["slug"], "categories": cs, "categories_joined": ", ".join(cs), "url": m["url"], "status": m.get("status") or "idle", "last_updated_raw": raw, "last_updated": disp, }) return ( render_template("index.html", mirrors=rows, categories=sorted(cats), error=msg), 400, ) return redirect(url_for("index")) @app.route("/update/", methods=["POST"]) def trigger_update(slug): api_post(f"mirrors/{slug}/update") return redirect(url_for("index")) @app.route("/status", methods=["GET"]) def status(): data = api_get("mirrors") mirrors = data.get("mirrors", []) out = [] for m in mirrors: raw = m.get("last_updated") disp = raw.replace("T", " ").replace("Z", " UTC") if raw else None out.append({ "slug": m["slug"], "categories": m.get("categories") or [], "url": m["url"], "status": m.get("status") or "idle", "last_updated": raw, "last_updated_display": disp or "", }) return jsonify({"mirrors": out}) @app.route("/logs/") def log_view(slug): # Template just polls /logs//tail via JS return render_template("log.html", slug=slug) @app.route("/logs//tail") def log_tail(slug): try: resp = requests.get(_api_url(f"mirrors/{slug}/log"), params={"bytes": 65536}, timeout=5) except requests.RequestException: return "", 200 if resp.status_code >= 400: return "", 200 return resp.text, 200, {"Content-Type": "text/plain; charset=utf-8"} @app.route("/search", methods=["GET"]) def content_search(): q = (request.args.get("q") or "").strip() if not q: return jsonify({"results": []}) try: resp = requests.get(_api_url("search"), params={"q": q}, timeout=10) except requests.RequestException: return jsonify({ "results": [{ "path": "(error)", "line": 0, "url": "", "snippet": "Search request to Mirage API failed.", }] }) if resp.status_code >= 400: return jsonify({ "results": [{ "path": "(error)", "line": 0, "url": "", "snippet": f"Mirage API returned {resp.status_code}.", }] }) data = resp.json() results = [] for r in data.get("results", []): rel_path = r.get("path", "") lineno = int(r.get("line", 0)) snippet = r.get("snippet") or "" # Web UI decides how mirrors are exposed over HTTP: url = MIRROR_HTTP_BASE.rstrip("/") + "/" + rel_path.replace("\\", "/") results.append({ "path": rel_path, "line": lineno, "url": url, "snippet": snippet, }) return jsonify({"results": results}) if __name__ == "__main__": app.run(host="127.0.0.1", port=5000, debug=False)