Yunohost_homePage/scripts/preview.py

#!/usr/bin/env python3
"""Serveur de previsualisation locale (sert src/), avec rechargement
automatique, sauvegarde et export depuis l'editeur visuel (editor.html).

Usage: python3 scripts/preview.py [port]
"""
import http.server
import json
import os
import shutil
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from yunohost_sync import build_app_list, fetch_remote_apps, known_urls, load_env, sync_target

PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8000

SCRIPT_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = SCRIPT_DIR.parent
SRC_DIR = PROJECT_ROOT / "src"
OUTPUT_DIR = PROJECT_ROOT / "output"
DEMO_CATALOG_PATH = SRC_DIR / "catalog.demo.json"

WATCHED_FILES = ["index.html", "style.css", "app.js", "config.json"]
EXPORTED_FILES = ["index.html", "style.css", "app.js", "config.json"]

RELOAD_SNIPPET = b"""
<script>
(function () {
  var lastSeen = null;
  var polling = false;
  function poll() {
    if (polling) { return; }
    polling = true;
    fetch("/__mtime", { cache: "no-store" })
      .then(function (r) { return r.text(); })
      .then(function (text) {
        if (lastSeen === null) {
          lastSeen = text;
        } else if (text !== lastSeen) {
          location.reload();
        }
      })
      .catch(function (err) { console.error("live-reload:", err); })
      .finally(function () { polling = false; });
  }
  setInterval(poll, 1000);
})();
</script>
</body>
"""


def watched_mtime():
    mtimes = []
    for name in WATCHED_FILES:
        try:
            mtimes.append((SRC_DIR / name).stat().st_mtime)
        except OSError:
            pass
    return str(max(mtimes)) if mtimes else "0"


class ReloadingHandler(http.server.SimpleHTTPRequestHandler):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, directory=str(SRC_DIR), **kwargs)

    def do_GET(self):
        if self.path == "/__mtime":
            body = watched_mtime().encode()
            self.send_response(200)
            self.send_header("Content-Type", "text/plain")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return

        if self.path == "/__catalog":
            self._send_catalog()
            return

        if self.path in ("/", "/index.html"):
            try:
                html = (SRC_DIR / "index.html").read_bytes()
            except OSError:
                return super().do_GET()
            html = html.replace(b"</body>", RELOAD_SNIPPET)
            self.send_response(200)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.send_header("Content-Length", str(len(html)))
            self.end_headers()
            self.wfile.write(html)
            return

        super().do_GET()

    def do_POST(self):
        if self.path == "/__save":
            length = int(self.headers.get("Content-Length", 0))
            raw = self.rfile.read(length)
            try:
                data = json.loads(raw)
            except json.JSONDecodeError as exc:
                self.send_error(400, f"JSON invalide: {exc}")
                return
            with open(SRC_DIR / "config.json", "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
                f.write("\n")
            self._send_json({"ok": True})
            return

        if self.path == "/__export":
            OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
            copied = []
            for name in EXPORTED_FILES:
                src = SRC_DIR / name
                if src.exists():
                    shutil.copy(src, OUTPUT_DIR / name)
                    copied.append(name)
            self._send_json({"ok": True, "files": copied, "dir": str(OUTPUT_DIR)})
            return

        self.send_error(404)

    def _send_catalog(self):
        if os.environ.get("HOMEPAGE_DEMO"):
            try:
                with open(DEMO_CATALOG_PATH, encoding="utf-8") as f:
                    apps = json.load(f)
            except OSError as exc:
                self._send_json({"error": str(exc)})
                return
            self._send_json({"apps": apps})
            return

        env = load_env()
        ssh_target, domain, self_app_id = sync_target(env)
        if not ssh_target or not domain:
            self._send_json({"error": "SYNC_SSH_HOST (ou REMOTE_HOST) et PORTAL_DOMAIN manquants dans .env"})
            return
        try:
            with open(SRC_DIR / "config.json", encoding="utf-8") as f:
                config = json.load(f)
            remote = fetch_remote_apps(ssh_target, domain)
            apps = build_app_list(remote, domain, self_app_id, exclude_urls=known_urls(config))
        except Exception as exc:  # SSH/JSON/reseau : on remonte un message lisible
            self._send_json({"error": str(exc)})
            return
        self._send_json({"apps": apps})

    def _send_json(self, payload):
        body = json.dumps(payload).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def end_headers(self):
        # Empeche le navigateur de mettre en cache src/*, sinon un simple
        # rechargement (ou le reload automatique) peut afficher une version
        # perimee et forcer a faire Ctrl+F5.
        self.send_header("Cache-Control", "no-store, must-revalidate")
        super().end_headers()

    def log_message(self, format, *args):
        if args and "__mtime" in str(args[0]):
            return
        super().log_message(format, *args)


if __name__ == "__main__":
    with http.server.ThreadingHTTPServer(("127.0.0.1", PORT), ReloadingHandler) as httpd:
        print(f"Apercu     : http://localhost:{PORT}/")
        print(f"Editeur    : http://localhost:{PORT}/editor.html")
        print("Ctrl+C pour arreter.")
        try:
            httpd.serve_forever()
        except KeyboardInterrupt:
            pass