#!/usr/bin/env python3 """ HEVC Dashboard — runs on the Windows PC (port 5000). Features: - Runs alongside the local worker API by default - Browses local mounted paths (e.g., Z:/TV) - Proxies all worker API calls (status, scan, process, pause, resume, stop) - SSE log stream proxy """ import os import requests from flask import Flask, Response, jsonify, render_template, request, stream_with_context app = Flask(__name__) # --------------------------------------------------------------------------- # Global config (editable at runtime via /config POST) # --------------------------------------------------------------------------- _config: dict = { "worker_url": os.environ.get("WORKER_URL", "http://127.0.0.1:5001").rstrip("/"), "media_root": os.environ.get("MEDIA_ROOT", "").replace("\\", "/"), "temp_dir": os.environ.get("TEMP_DIR", "").replace("\\", "/"), "discord_webhook": os.environ.get("DISCORD_WEBHOOK", ""), "custom_ffmpeg": os.environ.get("CUSTOM_FFMPEG", ""), "custom_ffprobe": os.environ.get("CUSTOM_FFPROBE", ""), "encoder": os.environ.get("ENCODER", "libx265"), "audio_codec": os.environ.get("AUDIO_CODEC", "auto"), "rename_output": os.environ.get("RENAME_OUTPUT", "false").strip().lower() in {"1", "true", "yes", "on"}, "force_reencode": os.environ.get("FORCE_REENCODE", "false").strip().lower() in {"1", "true", "yes", "on"}, } # --------------------------------------------------------------------------- # Dashboard UI # --------------------------------------------------------------------------- @app.route("/") def index(): return render_template( "index.html", worker_url=_config["worker_url"], media_root=_config["media_root"], custom_ffmpeg=_config.get("custom_ffmpeg", ""), custom_ffprobe=_config.get("custom_ffprobe", ""), temp_dir=_config["temp_dir"], discord_webhook=_config.get("discord_webhook", ""), encoder=_config.get("encoder", "libx265"), audio_codec=_config.get("audio_codec", "auto"), rename_output=_config.get("rename_output", False), force_reencode=_config.get("force_reencode", False), ) @app.route("/config", methods=["GET", "POST"]) def config(): if request.method == "POST": data = request.get_json(force=True, silent=True) or {} if "worker_url" in data: _config["worker_url"] = str(data["worker_url"]).rstrip("/") if "media_root" in data: _config["media_root"] = str(data["media_root"]).replace("\\", "/") if "temp_dir" in data: _config["temp_dir"] = str(data["temp_dir"]).replace("\\", "/") if "discord_webhook" in data: _config["discord_webhook"] = str(data["discord_webhook"]).strip() if "custom_ffmpeg" in data: _config["custom_ffmpeg"] = str(data["custom_ffmpeg"]).strip() if "custom_ffprobe" in data: _config["custom_ffprobe"] = str(data["custom_ffprobe"]).strip() if "encoder" in data: _config["encoder"] = str(data["encoder"]).strip() if "audio_codec" in data: _config["audio_codec"] = str(data["audio_codec"]).strip() if "rename_output" in data: _config["rename_output"] = bool(data["rename_output"]) if "force_reencode" in data: _config["force_reencode"] = bool(data["force_reencode"]) return jsonify({"ok": True, "config": dict(_config)}) return jsonify(dict(_config)) # --------------------------------------------------------------------------- # Local directory browser # --------------------------------------------------------------------------- def _normalize_local_path(raw_path: str) -> str: path = (raw_path or "").strip() # On Linux "/" is a real path — only fall back to media_root if truly empty if path == "": path = _config.get("media_root", "") if not path: return "/" if os.name != "nt" else "" return os.path.normpath(path) @app.route("/proxy/browse") def proxy_browse(): path = _normalize_local_path(request.args.get("path", "")) if not os.path.isdir(path): return jsonify({"error": f"Path not found: {path}"}), 404 try: items = sorted(os.listdir(path)) except PermissionError: return jsonify({"error": f"Permission denied: {path}"}), 403 dirs = [] for name in items: full = os.path.join(path, name) if os.path.isdir(full) and not name.startswith("."): dirs.append({"name": name, "full_path": full.replace("\\", "/")}) parent = os.path.dirname(path) if not parent or parent == path: parent = None return jsonify( { "path": path.replace("\\", "/"), "parent": parent.replace("\\", "/") if parent else None, "dirs": dirs, } ) # --------------------------------------------------------------------------- # Worker API proxy # --------------------------------------------------------------------------- def worker_url() -> str: return _config["worker_url"] def _proxy_get(path: str): try: r = requests.get(f"{worker_url()}{path}", timeout=8) return Response( r.content, status=r.status_code, content_type=r.headers.get("Content-Type", "application/json"), ) except requests.exceptions.ConnectionError: return jsonify({"error": "Cannot reach worker", "state": "unreachable"}), 503 except Exception as exc: return jsonify({"error": str(exc)}), 503 def _proxy_post(path: str, body=None): try: r = requests.post(f"{worker_url()}{path}", json=body or {}, timeout=8) return Response(r.content, status=r.status_code, content_type="application/json") except requests.exceptions.ConnectionError: return jsonify({"error": "Cannot reach worker"}), 503 except Exception as exc: return jsonify({"error": str(exc)}), 503 @app.route("/proxy/status") def proxy_status(): return _proxy_get("/api/status") @app.route("/proxy/drives") def proxy_drives(): return _proxy_get("/api/drives") @app.route("/proxy/sysinfo") def proxy_sysinfo(): return _proxy_get("/api/sysinfo") @app.route("/proxy/test-webhook", methods=["POST"]) def proxy_test_webhook(): body = request.get_json(force=True, silent=True) or {} return _proxy_post("/api/test-webhook", body) @app.route("/proxy/scan", methods=["POST"]) def proxy_scan(): body = request.get_json(force=True, silent=True) or {} return _proxy_post("/api/scan", body) @app.route("/proxy/process", methods=["POST"]) def proxy_process(): body = request.get_json(force=True, silent=True) or {} return _proxy_post("/api/process", body) @app.route("/proxy/database/clear", methods=["POST"]) def proxy_database_clear(): return _proxy_post("/api/database/clear") @app.route("/proxy/pause", methods=["POST"]) def proxy_pause(): return _proxy_post("/api/pause") @app.route("/proxy/resume", methods=["POST"]) def proxy_resume(): return _proxy_post("/api/resume") @app.route("/proxy/stop", methods=["POST"]) def proxy_stop(): return _proxy_post("/api/stop") @app.route("/proxy/logs/stream") def proxy_logs_stream(): def generate(): try: r = requests.get(f"{worker_url()}/api/logs/stream", stream=True, timeout=120) for line in r.iter_lines(): if line: yield f"{line.decode()}\n\n" except Exception: pass return Response( stream_with_context(generate()), mimetype="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- if __name__ == "__main__": print("HEVC Dashboard -> http://localhost:5000") print(f"Worker URL -> {worker_url()}") print(f"Media root -> {_config['media_root']}") app.run(host="0.0.0.0", port=5000, debug=False, threaded=True)