import os
import re
import json
import time
import hmac
import hashlib
import secrets
import requests
import shutil
import urllib3
from datetime import datetime, timedelta
from functools import wraps
from concurrent.futures import ThreadPoolExecutor, as_completed
from flask import Flask, render_template, request, jsonify, redirect, url_for, session
from werkzeug.security import generate_password_hash, check_password_hash
# Suppress SSL warnings for Unraid's self-signed certificate
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Build identity, shown in the dashboard footer. Bump these when cutting a
# deploy so the running container is identifiable at a glance — the Unraid
# Docker tab caches stale container IDs, so the footer is the reliable way
# to confirm a `docker compose up -d --build` actually took effect.
# Both are env-overridable if the deploy pipeline ever injects them.
APP_VERSION = os.environ.get("APP_VERSION", "1.6.2")
APP_VERSION_DATE = os.environ.get("APP_VERSION_DATE", "2026-08-27")
# Must point at a path that survives `docker compose up --build` — the
# container filesystem does not. See the data volume in docker-compose.yml.
SETTINGS_FILE = os.environ.get(
"SETTINGS_FILE", os.path.join(os.path.dirname(__file__), "settings.json")
)
def load_settings():
try:
with open(SETTINGS_FILE) as f:
return json.load(f)
except Exception:
return {}
def save_settings(data):
existing = load_settings()
existing.update(data)
os.makedirs(os.path.dirname(SETTINGS_FILE) or ".", exist_ok=True)
with open(SETTINGS_FILE, "w") as f:
json.dump(existing, f, indent=2)
def delete_setting(key):
"""Remove a key outright. save_settings() only ever merges, so it can't
take something away — used to retire the plaintext shared password once
real accounts exist. Returns True if the key was there."""
settings = load_settings()
if key not in settings:
return False
del settings[key]
os.makedirs(os.path.dirname(SETTINGS_FILE) or ".", exist_ok=True)
with open(SETTINGS_FILE, "w") as f:
json.dump(settings, f, indent=2)
return True
app = Flask(__name__)
app.secret_key = os.environ.get("SECRET_KEY") or secrets.token_hex(32)
def friendly_error(raw, service=""):
"""Turn a raw exception string into a short, human-readable cause.
The service dicts carry str(e) from requests/urllib3, which is useful for
debugging but unreadable on the dashboard. Templates render this instead
and keep the raw string behind a details disclosure.
"""
s = str(raw or "")
low = s.lower()
name = service or "The service"
if low.strip() == "timeout":
return f"{name} took too long to answer and was dropped at the 5s page deadline."
if "connecttimeout" in low or "timed out" in low or "read timed out" in low:
return f"{name} didn't respond — the connection timed out."
if "connection refused" in low or "newconnectionerror" in low:
return f"{name} refused the connection — check that it's running and the port is right."
if "name or service not known" in low or "nameresolution" in low or "failed to resolve" in low:
return f"{name}'s hostname couldn't be resolved — check the host setting."
if "no route to host" in low or "network is unreachable" in low:
return f"{name} is unreachable from this container — check the network."
if "401" in s or "unauthorized" in low or "403" in s or "forbidden" in low:
return f"{name} rejected the API key."
if "404" in s or "not found" in low:
return f"{name} returned 404 — check the URL or base path."
if any(code in s for code in ("500", "502", "503", "504")):
return f"{name} returned a server error."
if "ssl" in low or "certificate" in low:
return f"{name}'s TLS handshake failed — check the certificate settings."
return f"{name} couldn't be reached."
_SECRET_PARAM_RE = re.compile(
r"(?i)\b(apikey|api_key|token|password)=([^&\s'\")]+)"
)
def redact_secrets(raw):
"""Strip API keys/tokens out of raw exception text before it's shown in
the UI. Some services (SABnzbd, Tautulli) embed the key in the request
URL, so it ends up verbatim in requests/urllib3 exception strings.
"""
return _SECRET_PARAM_RE.sub(r"\1=[redacted]", str(raw or ""))
app.jinja_env.filters["friendly_error"] = friendly_error
app.jinja_env.filters["redact_secrets"] = redact_secrets
app.config["SESSION_COOKIE_PATH"] = "/"
app.config["SESSION_COOKIE_HTTPONLY"] = True
app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
app.config["SESSION_COOKIE_SECURE"] = False # ProxyFix handles HTTPS; keep False so cookie is sent over internal HTTP too
# Support running under a subpath (e.g. /media/) behind a reverse proxy.
# SCRIPT_NAME is injected into every WSGI request so Flask's url_for() and
# redirects all use the correct prefix automatically.
# ProxyFix also reads X-Forwarded-Prefix sent by NPM/nginx.
from werkzeug.middleware.proxy_fix import ProxyFix
_script_name = os.environ.get("SCRIPT_NAME", "").rstrip("/")
class ScriptNameMiddleware:
"""Inject SCRIPT_NAME into every WSGI environ so Flask knows its prefix."""
def __init__(self, wsgi_app, script_name):
self.app = wsgi_app
self.script_name = script_name
def __call__(self, environ, start_response):
if self.script_name:
environ["SCRIPT_NAME"] = self.script_name
path = environ.get("PATH_INFO", "")
if path.startswith(self.script_name):
environ["PATH_INFO"] = path[len(self.script_name):] or "/"
return self.app(environ, start_response)
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1)
if _script_name:
app.wsgi_app = ScriptNameMiddleware(app.wsgi_app, _script_name)
@app.context_processor
def inject_version():
"""Make the build identity available to every template."""
return {"app_version": APP_VERSION, "app_version_date": APP_VERSION_DATE}
# ── Authentication ──────────────────────────────────────────────────────────
LOGIN_PASSWORD = os.environ.get("LOGIN_PASSWORD", "")
_runtime_password = None # overrides env var after change-password call
# scrypt (Werkzeug's default) needs a hashlib built against OpenSSL, which
# isn't guaranteed — the macOS system Python has no hashlib.scrypt at all.
# pbkdf2 is pure hashlib and available everywhere this runs.
PASSWORD_HASH_METHOD = "pbkdf2:sha256"
_dummy_hash = None # lazily built; see verify_credentials
def get_active_password():
if _runtime_password is not None:
return _runtime_password
# Check settings.json for a persisted password change
saved = load_settings().get("login_password")
if saved:
return saved
return LOGIN_PASSWORD
def _hash_password(pw):
"""SHA-256 of the legacy single shared password.
Only used for the pre-multi-user fallback below. New accounts are hashed
with pbkdf2 + a per-user salt via load_users()/verify_credentials.
"""
return hashlib.sha256(pw.encode()).hexdigest()
def load_users():
"""Username -> pbkdf2 hash, read fresh from settings.json.
Users live in settings.json rather than an env var for two reasons: a
pbkdf2 hash contains '$', which docker compose interpolates when it reads
.env, and settings.json is already the bind-mounted file that survives
`up -d --build`. Reading it per-call also means an added user takes effect
without a restart, and applies to every gunicorn worker at once.
"""
users = load_settings().get("users") or {}
if not isinstance(users, dict):
return {}
return {u: h for u, h in users.items() if isinstance(h, str) and h}
def verify_credentials(username, password):
"""True if these credentials are valid.
Multi-user as soon as settings.json holds a non-empty `users` map;
until then this falls back to the original single shared password so
deploying this doesn't lock an existing install out. The username is
ignored on that legacy path.
"""
global _dummy_hash
users = load_users()
if users:
stored = users.get(username)
if not stored:
# Verify against a throwaway hash anyway, so a wrong username
# costs the same time as a wrong password and can't be probed.
if _dummy_hash is None:
_dummy_hash = generate_password_hash(
secrets.token_hex(16), method=PASSWORD_HASH_METHOD
)
check_password_hash(_dummy_hash, password)
return False
return check_password_hash(stored, password)
active = get_active_password()
if not active:
return False
return hmac.compare_digest(_hash_password(password), _hash_password(active))
def login_required(f):
@wraps(f)
def decorated(*args, **kwargs):
if not session.get("authenticated"):
return redirect(url_for("login", next=request.path))
return f(*args, **kwargs)
return decorated
SONARR_HOST = os.environ.get("ARR_HOST", "localhost")
RADARR_HOST = os.environ.get("ARR_HOST", "localhost")
SONARR_URL = os.environ.get("SONARR_URL", f"http://{SONARR_HOST}:8989")
RADARR_URL = os.environ.get("RADARR_URL", f"http://{RADARR_HOST}:7878")
SONARR_API_KEY = os.environ.get("SONARR_API_KEY", "")
RADARR_API_KEY = os.environ.get("RADARR_API_KEY", "")
SABNZBD_HOST = os.environ.get("SABNZBD_HOST", os.environ.get("ARR_HOST", "localhost"))
SABNZBD_PORT = os.environ.get("SABNZBD_PORT", "8080")
SABNZBD_URL = f"http://{SABNZBD_HOST}:{SABNZBD_PORT}"
SABNZBD_API_KEY = os.environ.get("SABNZBD_API_KEY", "")
TAUTULLI_HOST = os.environ.get("TAUTULLI_HOST", os.environ.get("ARR_HOST", "localhost"))
TAUTULLI_PORT = os.environ.get("TAUTULLI_PORT", "8181")
TAUTULLI_URL = f"http://{TAUTULLI_HOST}:{TAUTULLI_PORT}/tautulli"
TAUTULLI_API_KEY = os.environ.get("TAUTULLI_API_KEY", "")
PROWLARR_HOST = os.environ.get("PROWLARR_HOST", os.environ.get("ARR_HOST", "localhost"))
PROWLARR_PORT = os.environ.get("PROWLARR_PORT", "9696")
PROWLARR_URL = f"http://{PROWLARR_HOST}:{PROWLARR_PORT}/prowlarr"
PROWLARR_API_KEY = os.environ.get("PROWLARR_API_KEY", "")
OMBI_HOST = os.environ.get("OMBI_HOST", os.environ.get("ARR_HOST", "localhost"))
OMBI_PORT = os.environ.get("OMBI_PORT", "3579")
OMBI_URL = f"http://{OMBI_HOST}:{OMBI_PORT}"
OMBI_API_KEY = os.environ.get("OMBI_API_KEY", "")
# A key saved via the Settings page lives in settings.json and must win over
# the .env value here — otherwise "Save" on that page silently does nothing
# for anything that actually calls out to the service.
_saved_settings = load_settings()
SONARR_API_KEY = _saved_settings.get("sonarr_api_key") or SONARR_API_KEY
RADARR_API_KEY = _saved_settings.get("radarr_api_key") or RADARR_API_KEY
SABNZBD_API_KEY = _saved_settings.get("sabnzbd_api_key") or SABNZBD_API_KEY
TAUTULLI_API_KEY = _saved_settings.get("tautulli_api_key") or TAUTULLI_API_KEY
PROWLARR_API_KEY = _saved_settings.get("prowlarr_api_key") or PROWLARR_API_KEY
OMBI_API_KEY = _saved_settings.get("ombi_api_key") or OMBI_API_KEY
RT_HEADERS = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Accept-Language": "en-US,en;q=0.9",
}
# RT cache: refresh every 4 hours
_rt_cache = {}
RT_CACHE_TTL = 4 * 3600 # seconds
# IMDB description cache — persists for container lifetime, keyed by "title|year"
_imdb_desc_cache = {}
# Tracks titles currently being fetched so we don't double-fetch
_imdb_in_flight = set()
# SABnzbd history cache: refresh every 10 minutes (history doesn't change fast)
_sab_history_cache = {"data": None, "expires": 0}
# Sonarr / Radarr library cache: refresh every 5 minutes.
# The full movie list (703 items = 4.4 MB) is the single biggest source of
# dashboard latency — caching it cuts Radarr from ~1.3 s to <10 ms on warm loads.
_sonarr_cache = {"data": None, "expires": 0}
_radarr_cache = {"data": None, "expires": 0}
ARR_CACHE_TTL = 5 * 60 # 5 minutes
# Tautulli stats cache: top shows/movies/users/history refresh every 2 minutes.
# Active streams (get_activity) are always fetched live.
_tautulli_stats_cache = {"data": None, "expires": 0}
TAUTULLI_CACHE_TTL = 2 * 60 # 2 minutes
def _fetch_imdb_description(title, year=""):
"""Fetch a single IMDB description. Returns "" on any failure."""
cache_key = f"{title}|{year}"
if cache_key in _imdb_desc_cache:
return _imdb_desc_cache[cache_key]
try:
query = requests.utils.quote(f"{title} {year}".strip())
search_url = f"https://www.imdb.com/find/?q={query}&type=tt&s=tt"
resp = requests.get(search_url, headers=RT_HEADERS, timeout=8)
resp.raise_for_status()
match = re.search(r'href="(/title/tt\d+/)[^"]*"', resp.text)
if not match:
_imdb_desc_cache[cache_key] = ""
return ""
title_resp = requests.get(f"https://www.imdb.com{match.group(1)}", headers=RT_HEADERS, timeout=8)
title_resp.raise_for_status()
jld = re.search(r'', title_resp.text, re.DOTALL)
desc = json.loads(jld.group(1)).get("description", "") if jld else ""
_imdb_desc_cache[cache_key] = desc
return desc
except Exception:
_imdb_desc_cache[cache_key] = ""
return ""
def _enrich_rt_descriptions_bg(items):
"""Background enrichment: fetch IMDB descriptions for items that don't have one yet.
Uses 10 parallel workers. Safe to call from a daemon thread."""
to_fetch = [
item for item in items
if not item.get("error")
and not item.get("description")
and f"{item.get('title','')}|{item.get('year','')}" not in _imdb_desc_cache
and f"{item.get('title','')}|{item.get('year','')}" not in _imdb_in_flight
]
if not to_fetch:
return
keys = {f"{i.get('title','')}|{i.get('year','')}": i for i in to_fetch}
_imdb_in_flight.update(keys.keys())
try:
def _fetch(item):
desc = _fetch_imdb_description(item.get("title", ""), item.get("year", ""))
item["description"] = desc
with ThreadPoolExecutor(max_workers=10) as ex:
list(ex.map(_fetch, to_fetch))
finally:
_imdb_in_flight.difference_update(keys.keys())
# Ombi cache: refresh every 4 hours
_ombi_cache = {"data": None, "expires": 0}
def _rt_cached(key, fetch_fn):
"""Return cached RT data, re-fetching if older than 4 hours."""
import time
now = time.time()
entry = _rt_cache.get(key)
if entry and entry.get("expires", 0) > now:
return entry["data"]
data = fetch_fn()
_rt_cache[key] = {"data": data, "expires": now + RT_CACHE_TTL}
return data
# ---------------------------------------------------------------------------
# Sonarr
# ---------------------------------------------------------------------------
def sonarr_get(endpoint):
headers = {"X-Api-Key": SONARR_API_KEY}
resp = requests.get(f"{SONARR_URL}/api/v3/{endpoint}", headers=headers, timeout=4)
resp.raise_for_status()
return resp.json()
def sonarr_post(endpoint, payload):
headers = {"X-Api-Key": SONARR_API_KEY, "Content-Type": "application/json"}
resp = requests.post(f"{SONARR_URL}/api/v3/{endpoint}", headers=headers, json=payload, timeout=4)
resp.raise_for_status()
return resp.json()
def get_sonarr_summary():
global _sonarr_cache
try:
now = time.time()
if _sonarr_cache["data"] is not None and now < _sonarr_cache["expires"]:
series = _sonarr_cache["data"]
else:
series = sonarr_get("series")
_sonarr_cache = {"data": series, "expires": now + ARR_CACHE_TTL}
wanted = sonarr_get("wanted/missing?pageSize=1")
total_series = len(series)
monitored = sum(1 for s in series if s.get("monitored"))
missing_episodes = wanted.get("totalRecords", 0)
sonarr_titles = {s.get("title", "").lower() for s in series}
return {
"status": "ok",
"total_series": total_series,
"monitored_series": monitored,
"missing_episodes": missing_episodes,
"sonarr_titles": list(sonarr_titles),
}
except Exception as e:
return {"status": "error", "error": str(e), "sonarr_titles": []}
# ---------------------------------------------------------------------------
# Radarr
# ---------------------------------------------------------------------------
def radarr_get(endpoint):
headers = {"X-Api-Key": RADARR_API_KEY}
resp = requests.get(f"{RADARR_URL}/api/v3/{endpoint}", headers=headers, timeout=4)
resp.raise_for_status()
return resp.json()
def radarr_post(endpoint, payload):
headers = {"X-Api-Key": RADARR_API_KEY, "Content-Type": "application/json"}
resp = requests.post(f"{RADARR_URL}/api/v3/{endpoint}", headers=headers, json=payload, timeout=4)
resp.raise_for_status()
return resp.json()
def get_radarr_summary():
global _radarr_cache
try:
now = time.time()
if _radarr_cache["data"] is not None and now < _radarr_cache["expires"]:
movies = _radarr_cache["data"]
else:
movies = radarr_get("movie")
_radarr_cache = {"data": movies, "expires": now + ARR_CACHE_TTL}
total_movies = len(movies)
monitored = sum(1 for m in movies if m.get("monitored"))
downloaded = sum(1 for m in movies if m.get("hasFile"))
missing = total_movies - downloaded
radarr_titles = {m.get("title", "").lower() for m in movies}
return {
"status": "ok",
"total_movies": total_movies,
"monitored_movies": monitored,
"downloaded_movies": downloaded,
"missing_movies": missing,
"radarr_titles": list(radarr_titles),
}
except Exception as e:
return {"status": "error", "error": str(e), "radarr_titles": []}
# ---------------------------------------------------------------------------
# SABnzbd
# ---------------------------------------------------------------------------
def sabnzbd_get(mode, extra_params=""):
url = f"{SABNZBD_URL}/api?output=json&apikey={SABNZBD_API_KEY}&mode={mode}{extra_params}"
resp = requests.get(url, timeout=4)
resp.raise_for_status()
return resp.json()
def _format_speed(kbps):
"""Format KB/s into a human-readable string."""
if kbps is None:
return "0 B/s"
if kbps >= 1024 * 1024:
return f"{kbps / (1024 * 1024):.1f} GB/s"
if kbps >= 1024:
return f"{kbps / 1024:.1f} MB/s"
return f"{kbps:.0f} KB/s"
def _format_size(mb):
"""Format MB into a human-readable string."""
if mb is None:
return "0 MB"
if mb >= 1024:
return f"{mb / 1024:.1f} GB"
return f"{mb:.0f} MB"
def _get_sab_history_cached():
"""Fetch SABnzbd history with a 10-minute cache — avoids slow 1000-item fetch on every page load."""
global _sab_history_cache
now = time.time()
if _sab_history_cache["data"] is not None and now < _sab_history_cache["expires"]:
return _sab_history_cache["data"]
try:
history = sabnzbd_get("history", "&limit=1000")
h = history.get("history", {})
slots = h.get("slots", [])
week_size = h.get("week_size", "0 B")
day_size = h.get("day_size", "0 B")
seven_days_ago = now - (7 * 24 * 3600)
recent = [s for s in slots if s.get("completed", 0) >= seven_days_ago and s.get("status") != "Failed"]
completed_7d = len(recent)
speeds = []
for s in recent:
duration = s.get("download_time", 0) or 0
size_bytes = s.get("bytes", 0) or 0
if duration > 0 and size_bytes > 0:
speeds.append(size_bytes / duration)
avg_bytes_per_sec = sum(speeds) / len(speeds) if speeds else 0
avg_speed = _format_speed(avg_bytes_per_sec / 1024)
result = {"week_size": week_size, "day_size": day_size,
"completed_7d": completed_7d, "avg_speed": avg_speed}
_sab_history_cache = {"data": result, "expires": now + 600}
return result
except Exception:
return {"week_size": "—", "day_size": "—", "completed_7d": "—", "avg_speed": "—"}
def get_sabnzbd_summary():
try:
# Queue only — fast, no history fetch here
queue = sabnzbd_get("queue")
q = queue.get("queue", {})
speed_str = str(q.get("speed", "0")).strip()
speed_kbps = float(q.get("kbpersec", 0) or 0)
if speed_kbps > 0 and speed_str and speed_str != "0":
display_speed = speed_str + "/s" if not speed_str.endswith("/s") else speed_str
else:
display_speed = "0 B/s"
status = q.get("status", "Unknown")
queue_count = int(q.get("noofslots_total", q.get("noofslots", 0)))
sizeleft = q.get("sizeleft", "0 B")
timeleft = q.get("timeleft", "0:00:00")
hist = _get_sab_history_cached()
return {
"status": "ok",
"speed": display_speed,
"speed_kbps": speed_kbps,
"sab_status": status,
"queue_count": queue_count,
"sizeleft": sizeleft,
"timeleft": timeleft,
"completed_7d": hist["completed_7d"],
"week_size": hist["week_size"],
"day_size": hist["day_size"],
"avg_speed": hist["avg_speed"],
}
except Exception as e:
return {"status": "error", "error": str(e)}
# ---------------------------------------------------------------------------
# Tautulli
# ---------------------------------------------------------------------------
def tautulli_get(cmd, extra=""):
url = f"{TAUTULLI_URL}/api/v2?apikey={TAUTULLI_API_KEY}&cmd={cmd}{extra}"
resp = requests.get(url, timeout=4)
resp.raise_for_status()
return resp.json().get("response", {}).get("data", {})
def get_tautulli_summary():
global _tautulli_stats_cache
try:
now = time.time()
# Stats (top shows/movies/users/history) are cached for 2 minutes
if _tautulli_stats_cache["data"] is not None and now < _tautulli_stats_cache["expires"]:
cached = _tautulli_stats_cache["data"]
top_tv_list = cached["top_tv_list"]
top_movie_list = cached["top_movie_list"]
top_played = cached["top_played"]
top_users = cached["top_users"]
else:
# Top 3 most watched TV shows (30d)
tv_data = tautulli_get("get_home_stats", "&stat_id=top_tv&stats_count=3&stats_type=plays&time_range=30")
tv_rows = tv_data.get("rows", []) if isinstance(tv_data, dict) else []
top_tv_list = [{"title": r.get("title", "Unknown"), "plays": r.get("total_plays", 0)} for r in tv_rows]
# Top 3 most watched movies (30d)
movie_data = tautulli_get("get_home_stats", "&stat_id=top_movies&stats_count=3&stats_type=plays&time_range=30")
movie_rows = movie_data.get("rows", []) if isinstance(movie_data, dict) else []
top_movie_list = [{"title": r.get("title", "Unknown"), "plays": r.get("total_plays", 0)} for r in movie_rows]
# Top played per user from recent history
history_data = tautulli_get("get_history", "&length=100&order_column=date&order_dir=desc")
history_slots = []
if isinstance(history_data, dict):
history_slots = history_data.get("data", [])
if isinstance(history_slots, dict):
history_slots = history_slots.get("data", [])
play_counts = {}
for entry in history_slots:
user = entry.get("friendly_name") or entry.get("user", "Unknown")
if user.lower() == "ninja_hippo":
continue
title = entry.get("full_title") or entry.get("title", "Unknown")
media_type = entry.get("media_type", "")
key = (title, user)
if key not in play_counts:
play_counts[key] = {"title": title, "user": user, "plays": 0, "media_type": media_type}
play_counts[key]["plays"] += 1
top_played = sorted(play_counts.values(), key=lambda x: x["plays"], reverse=True)[:3]
# Top 3 viewers (30d) excluding ninja_hippo
users_data = tautulli_get("get_home_stats", "&stat_id=top_users&stats_count=10&stats_type=plays&time_range=30")
users_rows = users_data.get("rows", []) if isinstance(users_data, dict) else []
top_users = []
for r in users_rows:
username = r.get("friendly_name") or r.get("user", "")
if username.lower() == "ninja_hippo":
continue
top_users.append({"username": username, "plays": r.get("total_plays", 0), "thumb": r.get("user_thumb", "")})
if len(top_users) >= 3:
break
_tautulli_stats_cache = {
"data": {"top_tv_list": top_tv_list, "top_movie_list": top_movie_list,
"top_played": top_played, "top_users": top_users},
"expires": now + TAUTULLI_CACHE_TTL,
}
# Active streams always fetched live (changes second-to-second)
activity = tautulli_get("get_activity")
stream_count = int(activity.get("stream_count", 0) or 0) if isinstance(activity, dict) else 0
raw_sessions = activity.get("sessions", []) if isinstance(activity, dict) else []
sessions = []
for s in raw_sessions:
sessions.append({
"user": s.get("friendly_name") or s.get("user", "Unknown"),
"title": s.get("full_title") or s.get("title", "Unknown"),
"media_type": s.get("media_type", ""),
"state": s.get("state", ""),
})
top_tv = top_tv_list[0] if top_tv_list else None
top_movie = top_movie_list[0] if top_movie_list else None
return {
"status": "ok",
"top_tv": top_tv,
"top_tv_list": top_tv_list,
"top_movie": top_movie,
"top_movie_list": top_movie_list,
"top_played": top_played,
"top_users": top_users,
"active_streams": stream_count,
"sessions": sessions,
}
except Exception as e:
return {"status": "error", "error": str(e)}
# ---------------------------------------------------------------------------
# Ombi
# ---------------------------------------------------------------------------
def ombi_get(endpoint):
headers = {"ApiKey": OMBI_API_KEY}
resp = requests.get(f"{OMBI_URL}/api/v1/{endpoint}", headers=headers, timeout=4)
resp.raise_for_status()
return resp.json()
def _ombi_status_label(available, approved, denied):
if denied:
return "Denied"
if available:
return "Available"
if approved:
return "Approved"
return "Pending"
def get_ombi_summary():
global _ombi_cache
now = time.time()
if _ombi_cache["data"] is not None and now < _ombi_cache["expires"]:
return _ombi_cache["data"]
try:
movies_raw = ombi_get("Request/movie")
tv_raw = ombi_get("Request/tv")
# Flatten all requests with common fields
all_requests = []
for m in movies_raw:
user = m.get("requestedUser", {})
alias = (user.get("alias") or user.get("userAlias") or user.get("userName") or "Unknown").split("@")[0]
if alias.lower() == "ninja_hippo":
continue
all_requests.append({
"title": m.get("title", "Unknown"),
"type": "Movie",
"requested_date": m.get("requestedDate", ""),
"user": alias,
"status": _ombi_status_label(m.get("available"), m.get("approved"), m.get("denied")),
"poster": m.get("posterPath", ""),
})
for show in tv_raw:
for child in show.get("childRequests", []):
user = child.get("requestedUser", {})
alias = (user.get("alias") or user.get("userAlias") or user.get("userName") or "Unknown").split("@")[0]
if alias.lower() == "ninja_hippo":
continue
# Determine child status from season approvals
approved = child.get("approved", False)
available = any(
ep.get("available", False)
for season in child.get("seasonRequests", [])
for ep in season.get("episodes", [])
)
denied = child.get("denied", False)
all_requests.append({
"title": show.get("title", "Unknown"),
"type": "TV",
"requested_date": child.get("requestedDate", ""),
"user": alias,
"status": _ombi_status_label(available, approved, denied),
"poster": show.get("posterPath", ""),
})
# Sort by requested date descending, take 5 most recent
all_requests.sort(key=lambda x: x["requested_date"], reverse=True)
recent = all_requests[:10]
# Tidy date display
for r in recent:
d = r["requested_date"]
r["date_display"] = d[:10] if d else "—"
total_pending = sum(1 for r in all_requests if r["status"] == "Pending")
total_requests = len(all_requests)
result = {
"status": "ok",
"recent": recent,
"total_requests": total_requests,
"total_pending": total_pending,
}
_ombi_cache["data"] = result
_ombi_cache["expires"] = now + (4 * 3600)
return result
except Exception as e:
return {"status": "error", "error": str(e)}
# ---------------------------------------------------------------------------
# Prowlarr
# ---------------------------------------------------------------------------
def prowlarr_get(endpoint):
headers = {"X-Api-Key": PROWLARR_API_KEY}
resp = requests.get(f"{PROWLARR_URL}/api/v1/{endpoint}", headers=headers, timeout=4)
resp.raise_for_status()
return resp.json()
# Grab/query counts change slowly and the extra round-trip sits inside the
# dashboard's 5s wall-clock deadline, so this is cached harder than the
# up/down status it decorates. Failures are not cached — a stats outage
# should not blank the share column for ten minutes after it recovers.
PROWLARR_STATS_TTL = 10 * 60
PROWLARR_STATS_DAYS = 30
_prowlarr_stats_cache = {"data": None, "expires": 0}
def get_prowlarr_stats():
"""Per-indexer query/grab counts for the last PROWLARR_STATS_DAYS days.
Returns {indexer_id: {"queries": int, "grabs": int, "failed_queries": int}}
or {} if Prowlarr's stats endpoint is unavailable — callers must treat an
empty result as "unknown", not "zero activity".
"""
now = time.time()
if _prowlarr_stats_cache["data"] is not None and now < _prowlarr_stats_cache["expires"]:
return _prowlarr_stats_cache["data"]
end = datetime.utcnow()
start = end - timedelta(days=PROWLARR_STATS_DAYS)
fmt = "%Y-%m-%dT%H:%M:%SZ"
try:
raw = prowlarr_get(
f"indexerstats?startDate={start.strftime(fmt)}&endDate={end.strftime(fmt)}"
)
except Exception:
return {}
stats = {}
for row in (raw.get("indexers") or []) if isinstance(raw, dict) else []:
stats[row.get("indexerId")] = {
"queries": row.get("numberOfQueries", 0) or 0,
"grabs": row.get("numberOfGrabs", 0) or 0,
"failed_queries": row.get("numberOfFailedQueries", 0) or 0,
}
_prowlarr_stats_cache["data"] = stats
_prowlarr_stats_cache["expires"] = now + PROWLARR_STATS_TTL
return stats
def get_prowlarr_summary():
try:
indexers = prowlarr_get("indexer")
# indexerstatus returns only blocked/failed indexers
try:
failed_raw = prowlarr_get("indexerstatus")
failed_ids = {s.get("indexerId") for s in failed_raw} if isinstance(failed_raw, list) else set()
except Exception:
failed_ids = set()
stats = get_prowlarr_stats()
total_grabs = sum(v["grabs"] for v in stats.values())
total_queries = sum(v["queries"] for v in stats.values())
result = []
for idx in indexers:
enabled = idx.get("enable", True)
blocked = idx.get("id") in failed_ids
up = enabled and not blocked
st = stats.get(idx.get("id"))
result.append({
"id": idx.get("id"),
"name": idx.get("name", "Unknown"),
"protocol": idx.get("protocol", ""),
"enabled": enabled,
"up": up,
"has_stats": st is not None,
"grabs": st["grabs"] if st else 0,
"queries": st["queries"] if st else 0,
"failed_queries": st["failed_queries"] if st else 0,
# Share of everything actually pulled down in the window. Falls
# back to query share when nothing has been grabbed yet, so a
# quiet month still ranks the indexers by who is doing the work.
"grab_share": round(100 * st["grabs"] / total_grabs, 1)
if st and total_grabs else 0.0,
"query_share": round(100 * st["queries"] / total_queries, 1)
if st and total_queries else 0.0,
})
# The share the card actually renders, so templates never have to
# know which metric won.
result[-1]["share"] = (result[-1]["grab_share"] if total_grabs
else result[-1]["query_share"])
# Busiest first — the card's job is to answer "who is pulling the
# weight", so name order would bury the answer.
rank_key = "grabs" if total_grabs else "queries"
result.sort(key=lambda i: (-i[rank_key], i["name"].lower()))
total = len(result)
up_count = sum(1 for i in result if i["up"])
return {
"status": "ok",
"indexers": result,
"total": total,
"up_count": up_count,
"has_stats": bool(stats),
"stats_days": PROWLARR_STATS_DAYS,
"total_grabs": total_grabs,
"total_queries": total_queries,
"share_metric": "grabs" if total_grabs else "queries",
}
except Exception as e:
return {"status": "error", "error": str(e), "indexers": []}
# ---------------------------------------------------------------------------
# Unraid system stats
# ---------------------------------------------------------------------------
UNRAID_HOST = os.environ.get("UNRAID_HOST", os.environ.get("ARR_HOST", ""))
UNRAID_API_KEY = os.environ.get("UNRAID_API_KEY", "")
def get_unraid_stats():
"""Fetch Unraid disk/system stats via GraphQL API, falling back to shutil."""
def fmt_bytes(b):
b = int(b or 0)
if b >= 1024**4: return f"{b/1024**4:.1f} TB"
if b >= 1024**3: return f"{b/1024**3:.1f} GB"
if b >= 1024**2: return f"{b/1024**2:.1f} MB"
return f"{b/1024:.0f} KB"
def _fs_bytes(v):
"""Unraid's GraphQL reports fsSize/fsFree/fsUsed/size in KB, not bytes
— the schema says so explicitly ('(KB) Total Size of the FS'). Reading
them as bytes understated every figure by 1024x, so a 13 TB disk
rendered as '13.0 GB'. The shutil fallback below already returns real
bytes and must not go through this."""
return int(v or 0) * 1024
if UNRAID_HOST and UNRAID_API_KEY:
# HTTPS on 443 with SSL verification disabled (Unraid uses a self-signed cert)
url = f"https://{UNRAID_HOST}:443/graphql"
headers = {"x-api-key": UNRAID_API_KEY, "Content-Type": "application/json"}
# Parity is deliberately not requested: it mirrors the array rather
# than holding a filesystem, so it has no size or usage to report and
# only ever rendered as a name and a temperature.
query = """{
vars { version }
array {
state
disks { name device fsSize fsFree type temp isSpinning }
caches { name device fsSize fsFree type temp isSpinning }
}
}"""
try:
resp = requests.post(url, json={"query": query}, headers=headers,
timeout=4, verify=False)
if resp.ok:
d = resp.json().get("data", {})
if d:
arr = d.get("array", {})
disks = arr.get("disks", [])
data_disks = [dk for dk in disks if dk.get("type") == "DATA"]
total_bytes = sum(_fs_bytes(dk.get("fsSize")) for dk in data_disks)
free_bytes = sum(_fs_bytes(dk.get("fsFree")) for dk in data_disks)
used_bytes = total_bytes - free_bytes
used_pct = round(used_bytes / total_bytes * 100) if total_bytes else 0
def disk_row(dk):
dtotal = _fs_bytes(dk.get("fsSize"))
dfree = _fs_bytes(dk.get("fsFree"))
dused = dtotal - dfree
return {
"name": dk.get("name") or dk.get("device") or "disk",
"device": dk.get("device", ""),
"temp": dk.get("temp"),
"spinning": dk.get("isSpinning"),
"total": fmt_bytes(dtotal),
"used": fmt_bytes(dused),
"free": fmt_bytes(dfree),
"used_pct": round(dused / dtotal * 100) if dtotal else 0,
}
return {
"status": "ok",
"source": "api",
"array_state": arr.get("state", ""),
"os_version": d.get("vars", {}).get("version", ""),
"disk_count": len(data_disks),
"array_total": fmt_bytes(total_bytes),
"array_used": fmt_bytes(used_bytes),
"array_free": fmt_bytes(free_bytes),
"array_used_pct": used_pct,
"disks": [disk_row(dk) for dk in data_disks],
"caches": [disk_row(dk) for dk in arr.get("caches", [])],
}
except Exception:
pass
# Fallback: container-visible disk usage
try:
mount_paths = ["/mnt/user", "/data", "/"]
chosen = next((p for p in mount_paths if os.path.exists(p)), "/")
usage = shutil.disk_usage(chosen)
used_pct = round(usage.used / usage.total * 100) if usage.total else 0
return {
"status": "ok",
"source": "local",
"mount": chosen,
"array_total": fmt_bytes(usage.total),
"array_used": fmt_bytes(usage.used),
"array_free": fmt_bytes(usage.free),
"array_used_pct": used_pct,
}
except Exception as e:
return {"status": "error", "error": str(e)}
# ---------------------------------------------------------------------------
# Rotten Tomatoes scraping
# ---------------------------------------------------------------------------
def _scrape_rt_jsonld(html):
match = re.search(r'', html, re.DOTALL)
if not match:
return []
try:
data = json.loads(match.group(1))
inner = data.get("itemListElement", {})
if isinstance(inner, dict):
return inner.get("itemListElement", [])
if isinstance(inner, list):
return inner
except Exception:
pass
return []
def _scrape_rt_tiles(html):
tiles = re.findall(r']*>(.*?)', html, re.DOTALL)
results = []
for tile in tiles:
audience_match = re.search(r'slot="audienceScore"[^>]*>\s*([^<\s][^<]*?)\s*<', tile)
audience_score = audience_match.group(1).strip() if audience_match else ""
certified_match = re.search(r'certified="(true|false)"', tile)
certified = certified_match.group(1) == "true" if certified_match else False
results.append({"audience_score": audience_score, "certified": certified})
return results
def _fetch_rt_movies():
try:
resp = requests.get(
"https://www.rottentomatoes.com/browse/movies_in_theaters/sort:top_box_office",
headers=RT_HEADERS, timeout=4
)
resp.raise_for_status()
html = resp.text
items = _scrape_rt_jsonld(html)[:30]
tiles = _scrape_rt_tiles(html)
movies = []
for i, item in enumerate(items):
tile = tiles[i] if i < len(tiles) else {}
rating = item.get("aggregateRating")
critics_score = str(rating.get("ratingValue", "")) if rating else ""
try:
if critics_score and int(critics_score) < 50:
continue
except ValueError:
pass
movies.append({
"title": item.get("name", "Unknown"),
"year": str(item.get("dateCreated", ""))[:4],
"poster": item.get("image", ""),
"rt_url": item.get("url", ""),
"critics_score": critics_score,
"audience_score": tile.get("audience_score", ""),
"certified": tile.get("certified", False),
"description": "",
})
return movies
except Exception as e:
return [{"error": str(e)}]
def _fetch_rt_tv():
try:
resp = requests.get(
"https://www.rottentomatoes.com/browse/tv_series_browse/sort:popular",
headers=RT_HEADERS, timeout=4
)
resp.raise_for_status()
html = resp.text
items = _scrape_rt_jsonld(html)[:30]
tiles = _scrape_rt_tiles(html)
shows = []
for i, item in enumerate(items):
tile = tiles[i] if i < len(tiles) else {}
rating = item.get("aggregateRating")
critics_score = str(rating.get("ratingValue", "")) if rating else ""
try:
if critics_score and int(critics_score) < 50:
continue
except ValueError:
pass
shows.append({
"title": item.get("name", "Unknown"),
"year": str(item.get("dateCreated", ""))[:4],
"poster": item.get("image", ""),
"rt_url": item.get("url", ""),
"critics_score": critics_score,
"audience_score": tile.get("audience_score", ""),
"certified": tile.get("certified", False),
"description": "",
})
return shows
except Exception as e:
return [{"error": str(e)}]
def get_rt_movies():
data = _rt_cached("rt_movies", _fetch_rt_movies)
# Kick off background IMDB enrichment (non-blocking)
import threading
threading.Thread(target=_enrich_rt_descriptions_bg, args=(data,), daemon=True).start()
return data
def get_rt_tv():
data = _rt_cached("rt_tv", _fetch_rt_tv)
import threading
threading.Thread(target=_enrich_rt_descriptions_bg, args=(data,), daemon=True).start()
return data
# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------
@app.route("/login", methods=["GET", "POST"])
def login():
error = None
multi_user = bool(load_users())
if request.method == "POST":
username = request.form.get("username", "").strip()
entered = request.form.get("password", "")
if verify_credentials(username, entered):
session["authenticated"] = True
session["user"] = username if multi_user else ""
session.permanent = True
# Always redirect to dashboard — avoids bare-path redirect issues
# when running behind a reverse proxy subpath (e.g. /media/)
return redirect(url_for("dashboard"))
# Deliberately doesn't say which half was wrong.
error = ("Incorrect username or password." if multi_user
else "Incorrect password. Try again.")
# If nothing is configured at all, auto-login (dev mode)
if not multi_user and not get_active_password():
session["authenticated"] = True
return redirect(url_for("dashboard"))
return render_template("login.html", error=error, multi_user=multi_user)
@app.route("/logout")
def logout():
session.clear()
return redirect(url_for("login"))
@app.route("/api/unraid")
@login_required
def api_unraid():
return jsonify(get_unraid_stats())
@app.route("/api/rt/descriptions")
@login_required
def api_rt_descriptions():
"""Return cached IMDB descriptions for all RT items. Called by the page after load.
Also triggers enrichment in this worker if the cache is cold (handles multi-worker case)."""
import threading
movies = _rt_cache.get("rt_movies", {}).get("data", [])
tv = _rt_cache.get("rt_tv", {}).get("data", [])
# If this worker hasn't loaded RT data yet, trigger it and tell client to retry
if not movies and not tv:
threading.Thread(target=get_rt_movies, daemon=True).start()
threading.Thread(target=get_rt_tv, daemon=True).start()
return jsonify({"descriptions": {}, "pending": True})
all_items = [i for i in movies + tv if not i.get("error")]
# If this worker has no descriptions yet, kick off enrichment now
needs_fetch = [
i for i in all_items
if f"{i.get('title','')}|{i.get('year','')}" not in _imdb_desc_cache
and f"{i.get('title','')}|{i.get('year','')}" not in _imdb_in_flight
]
if needs_fetch:
threading.Thread(target=_enrich_rt_descriptions_bg, args=(all_items,), daemon=True).start()
result = {}
for item in all_items:
key = f"{item.get('title','')}|{item.get('year','')}"
desc = _imdb_desc_cache.get(key, None)
if desc is not None:
result[item.get("title", "")] = desc
still_pending = bool(needs_fetch) or any(
_imdb_desc_cache.get(f"{i.get('title','')}|{i.get('year','')}", None) is None
for i in all_items
)
return jsonify({"descriptions": result, "pending": still_pending})
@app.route("/")
@login_required
def dashboard():
# Fetch all service data in parallel with a hard 5-second wall-clock deadline.
# Any service that hasn't responded in time returns a graceful error dict so
# the page always renders quickly — critical for remote/cellular access.
DASHBOARD_TIMEOUT = 5.0
tasks = {
"sonarr": get_sonarr_summary,
"radarr": get_radarr_summary,
"sabnzbd": get_sabnzbd_summary,
"tautulli": get_tautulli_summary,
"prowlarr": get_prowlarr_summary,
"ombi": get_ombi_summary,
"unraid": get_unraid_stats,
"rt_movies": get_rt_movies,
"rt_tv": get_rt_tv,
}
results = {}
with ThreadPoolExecutor(max_workers=len(tasks)) as ex:
futures = {ex.submit(fn): name for name, fn in tasks.items()}
deadline = time.time() + DASHBOARD_TIMEOUT
for future in as_completed(futures, timeout=DASHBOARD_TIMEOUT):
name = futures[future]
try:
results[name] = future.result(timeout=max(0.1, deadline - time.time()))
except Exception as e:
results[name] = {"status": "error", "error": str(e)}
# Fill in any tasks that didn't complete within the deadline
for name in tasks:
if name not in results:
results[name] = {"status": "error", "error": "timeout"}
sonarr = results["sonarr"]
radarr = results["radarr"]
sabnzbd = results["sabnzbd"]
tautulli = results["tautulli"]
prowlarr = results["prowlarr"]
ombi = results["ombi"]
unraid = results["unraid"]
sonarr_titles = set(sonarr.get("sonarr_titles", []))
radarr_titles = set(radarr.get("radarr_titles", []))
_rt_movies_raw = results["rt_movies"]
_rt_tv_raw = results["rt_tv"]
rt_movies = [m for m in _rt_movies_raw if not m.get("error") and m.get("title", "").lower() not in radarr_titles][:10]
if not rt_movies and _rt_movies_raw:
rt_movies = _rt_movies_raw
rt_tv = [s for s in _rt_tv_raw if not s.get("error") and s.get("title", "").lower() not in sonarr_titles][:10]
if not rt_tv and _rt_tv_raw:
rt_tv = _rt_tv_raw
s = load_settings()
timezone = s.get("timezone", "America/New_York")
# Public-facing URLs for card header links (via NPM reverse proxy)
arr_host_public = f"https://hippofam.com"
service_urls = {
"sonarr": f"{arr_host_public}/sonarr",
"radarr": f"{arr_host_public}/radarr",
"sabnzbd": f"{arr_host_public}/sab",
"tautulli": f"{arr_host_public}/tautulli",
"prowlarr": f"{arr_host_public}/prowlarr",
"ombi": f"{arr_host_public}/ombi",
"unraid": f"https://{UNRAID_HOST}",
}
return render_template(
"dashboard.html",
sonarr=sonarr, radarr=radarr, sabnzbd=sabnzbd, tautulli=tautulli,
prowlarr=prowlarr, ombi=ombi, unraid=unraid,
rt_movies=rt_movies, rt_tv=rt_tv,
sonarr_titles=sonarr_titles, radarr_titles=radarr_titles,
timezone=timezone,
service_urls=service_urls,
)
@app.route("/api/sabnzbd")
@login_required
def api_sabnzbd():
"""Lightweight real-time endpoint — queue only (no history re-fetch)."""
try:
queue = sabnzbd_get("queue")
q = queue.get("queue", {})
speed_str = str(q.get("speed", "0")).strip()
speed_kbps = float(q.get("kbpersec", 0) or 0)
if speed_kbps > 0 and speed_str and speed_str != "0":
display_speed = speed_str + "/s" if not speed_str.endswith("/s") else speed_str
else:
display_speed = "0 B/s"
status = q.get("status", "Unknown")
queue_count = int(q.get("noofslots_total", q.get("noofslots", 0)))
sizeleft = q.get("sizeleft", "0 B")
timeleft = q.get("timeleft", "0:00:00")
resp = jsonify({
"status": "ok",
"speed": display_speed,
"speed_kbps": speed_kbps,
"sab_status": status,
"queue_count": queue_count,
"sizeleft": sizeleft,
"timeleft": timeleft,
})
resp.headers["Cache-Control"] = "no-store"
return resp
except Exception as e:
return jsonify({"status": "error", "error": str(e)}), 500
@app.route("/api/prowlarr")
@login_required
def api_prowlarr():
"""Real-time Prowlarr indexer status."""
return jsonify(get_prowlarr_summary())
@app.route("/api/tautulli")
@login_required
def api_tautulli():
"""Real-time Tautulli active streams."""
try:
activity = tautulli_get("get_activity")
stream_count = int(activity.get("stream_count", 0) or 0) if isinstance(activity, dict) else 0
raw_sessions = activity.get("sessions", []) if isinstance(activity, dict) else []
sessions = []
for s in raw_sessions:
sessions.append({
"user": s.get("friendly_name") or s.get("user", "Unknown"),
"title": s.get("full_title") or s.get("title", "Unknown"),
"media_type": s.get("media_type", ""),
"state": s.get("state", ""),
})
return jsonify({"status": "ok", "stream_count": stream_count, "sessions": sessions})
except Exception as e:
return jsonify({"status": "error", "error": str(e)}), 500
@app.route("/api/ombi")
@login_required
def api_ombi():
"""Return cached Ombi data."""
return jsonify(get_ombi_summary())
@app.route("/api/services")
@login_required
def api_services():
"""Header status lights — lightweight reachability check for all services."""
def check(fn):
try:
r = fn()
return r.get("status") == "ok"
except Exception:
return False
# Quick queue-only checks for speed
def sonarr_ok():
sonarr_get("series?pageSize=1")
return True
def radarr_ok():
radarr_get("movie?pageSize=1")
return True
def sabnzbd_ok():
sabnzbd_get("queue")
return True
def tautulli_ok():
tautulli_get("get_activity")
return True
def prowlarr_ok():
prowlarr_get("health")
return True
def ombi_ok():
ombi_get("Settings/about")
return True
results = {}
for name, fn in [("sonarr", sonarr_ok), ("radarr", radarr_ok),
("sabnzbd", sabnzbd_ok), ("tautulli", tautulli_ok),
("prowlarr", prowlarr_ok), ("ombi", ombi_ok)]:
try:
fn()
results[name] = True
except Exception:
results[name] = False
return jsonify(results)
@app.route("/api/tautulli/never-watched")
@login_required
def api_never_watched():
"""Movies and shows in Radarr/Sonarr that have never been played in Tautulli."""
try:
# Get all history titles from Tautulli (last 500 entries)
hist = tautulli_get("get_history", "&length=500&order_column=date&order_dir=desc")
played_titles = set()
if isinstance(hist, dict):
slots = hist.get("data", [])
if isinstance(slots, dict):
slots = slots.get("data", [])
for s in slots:
t = s.get("full_title") or s.get("title") or s.get("parent_title", "")
if t:
played_titles.add(t.lower())
never = []
# Sonarr series with episodes on disk, never played
try:
series_list = sonarr_get("series")
for s in series_list:
size = s.get("sizeOnDisk", 0)
if size == 0:
continue
title = s.get("title", "")
if title.lower() not in played_titles:
never.append({
"type": "show",
"title": title,
"year": s.get("year", ""),
"size": size,
"sonarr_id": s.get("id"),
})
except Exception:
pass
# Sort by size descending
never.sort(key=lambda x: x["size"], reverse=True)
# Format size
def fmt(b):
gb = b / (1024**3)
if gb >= 1:
return f"{gb:.1f} GB"
return f"{b / (1024**2):.0f} MB"
for item in never:
item["size_display"] = fmt(item["size"])
return jsonify({"status": "ok", "items": never[:30]})
except Exception as e:
return jsonify({"status": "error", "error": str(e)}), 500
@app.route("/api/delete/movie", methods=["POST"])
@login_required
def api_delete_movie():
data = request.get_json()
radarr_id = data.get("radarr_id")
if not radarr_id:
return jsonify({"error": "No radarr_id"}), 400
try:
headers = {"X-Api-Key": RADARR_API_KEY}
resp = requests.delete(
f"{RADARR_URL}/api/v3/movie/{radarr_id}?deleteFiles=true",
headers=headers, timeout=15
)
resp.raise_for_status()
return jsonify({"success": True})
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route("/api/delete/show", methods=["POST"])
@login_required
def api_delete_show():
data = request.get_json()
sonarr_id = data.get("sonarr_id")
if not sonarr_id:
return jsonify({"error": "No sonarr_id"}), 400
try:
headers = {"X-Api-Key": SONARR_API_KEY}
resp = requests.delete(
f"{SONARR_URL}/api/v3/series/{sonarr_id}?deleteFiles=true",
headers=headers, timeout=15
)
resp.raise_for_status()
return jsonify({"success": True})
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route("/settings", methods=["GET"])
@login_required
def settings_page():
s = load_settings()
return render_template("settings.html",
timezone=s.get("timezone", "America/New_York"),
sonarr_api_key=SONARR_API_KEY,
radarr_api_key=RADARR_API_KEY,
sabnzbd_api_key=SABNZBD_API_KEY,
tautulli_api_key=TAUTULLI_API_KEY,
prowlarr_api_key=PROWLARR_API_KEY,
ombi_api_key=OMBI_API_KEY,
arr_host=os.environ.get("ARR_HOST", ""),
)
@app.route("/api/settings", methods=["POST"])
@login_required
def api_save_settings():
data = request.get_json()
allowed = {"timezone", "sonarr_api_key", "radarr_api_key", "sabnzbd_api_key",
"tautulli_api_key", "prowlarr_api_key", "ombi_api_key", "arr_host"}
filtered = {k: v for k, v in data.items() if k in allowed}
save_settings(filtered)
return jsonify({"success": True})
@app.route("/api/settings", methods=["GET"])
@login_required
def api_get_settings():
s = load_settings()
return jsonify({
"timezone": s.get("timezone", "America/New_York"),
})
@app.route("/api/change-password", methods=["POST"])
@login_required
def api_change_password():
global _runtime_password
data = request.get_json()
current = data.get("current", "")
new_pw = data.get("new", "")
confirm = data.get("confirm", "")
if not current or not new_pw or not confirm:
return jsonify({"error": "All fields are required"}), 400
if new_pw != confirm:
return jsonify({"error": "New passwords do not match"}), 400
if len(new_pw) < 6:
return jsonify({"error": "Password must be at least 6 characters"}), 400
users = load_users()
if users:
# Multi-user: change only the signed-in account's own password.
me = session.get("user", "")
if me not in users or not verify_credentials(me, current):
return jsonify({"error": "Current password is incorrect"}), 403
users[me] = generate_password_hash(new_pw, method=PASSWORD_HASH_METHOD)
save_settings({"users": users})
# No _runtime_password here: load_users() re-reads settings.json on
# every check, so the change is live in every worker immediately.
return jsonify({"success": True})
if not verify_credentials("", current):
return jsonify({"error": "Current password is incorrect"}), 403
# Legacy single-password install: persist and update the runtime value.
save_settings({"login_password": new_pw})
_runtime_password = new_pw
return jsonify({"success": True})
@app.route("/api/sonarr/lookup")
@login_required
def api_sonarr_lookup():
q = request.args.get("q", "").strip()
if not q:
return jsonify({"error": "No query provided"}), 400
try:
results = sonarr_get(f"series/lookup?term={requests.utils.quote(q)}")
if not results:
return jsonify({})
s = results[0]
title = s.get("title", "")
sonarr_titles = {x.get("title", "").lower() for x in sonarr_get("series")}
return jsonify({
"title": title,
"year": s.get("year", ""),
"network": s.get("network", ""),
"status": s.get("status", ""),
"exists": title.lower() in sonarr_titles,
})
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route("/api/radarr/lookup")
@login_required
def api_radarr_lookup():
q = request.args.get("q", "").strip()
if not q:
return jsonify({"error": "No query provided"}), 400
try:
results = radarr_get(f"movie/lookup?term={requests.utils.quote(q)}")
if not results:
return jsonify({})
m = results[0]
title = m.get("title", "")
radarr_titles = {x.get("title", "").lower() for x in radarr_get("movie")}
return jsonify({
"title": title,
"year": m.get("year", ""),
"studio": m.get("studio", ""),
"runtime": m.get("runtime", ""),
"exists": title.lower() in radarr_titles,
})
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route("/add-to-radarr", methods=["POST"])
@login_required
def add_to_radarr():
data = request.get_json()
title = data.get("title")
if not title:
return jsonify({"error": "No title provided"}), 400
try:
results = radarr_get(f"movie/lookup?term={requests.utils.quote(title)}")
if not results:
return jsonify({"error": "Movie not found in Radarr lookup"}), 404
movie = results[0]
root_folders = radarr_get("rootfolder")
quality_profiles = radarr_get("qualityprofile")
if not root_folders or not quality_profiles:
return jsonify({"error": "No root folders or quality profiles configured in Radarr"}), 500
payload = {
"title": movie.get("title"),
"qualityProfileId": quality_profiles[0]["id"],
"titleSlug": movie.get("titleSlug"),
"images": movie.get("images", []),
"tmdbId": movie.get("tmdbId"),
"year": movie.get("year"),
"rootFolderPath": root_folders[0]["path"],
"monitored": True,
"addOptions": {"searchForMovie": True},
}
result = radarr_post("movie", payload)
return jsonify({"success": True, "title": result.get("title")})
except requests.HTTPError as e:
if e.response is not None and e.response.status_code == 400:
return jsonify({"error": "Movie may already exist in Radarr"}), 409
return jsonify({"error": str(e)}), 500
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route("/add-to-sonarr", methods=["POST"])
@login_required
def add_to_sonarr():
data = request.get_json()
title = data.get("title")
if not title:
return jsonify({"error": "No title provided"}), 400
try:
results = sonarr_get(f"series/lookup?term={requests.utils.quote(title)}")
if not results:
return jsonify({"error": "Series not found in Sonarr lookup"}), 404
series = results[0]
root_folders = sonarr_get("rootfolder")
quality_profiles = sonarr_get("qualityprofile")
if not root_folders or not quality_profiles:
return jsonify({"error": "No root folders or quality profiles configured in Sonarr"}), 500
payload = {
"title": series.get("title"),
"qualityProfileId": quality_profiles[0]["id"],
"titleSlug": series.get("titleSlug"),
"images": series.get("images", []),
"tvdbId": series.get("tvdbId"),
"year": series.get("year"),
"rootFolderPath": root_folders[0]["path"],
"monitored": True,
"seasonFolder": True,
"addOptions": {
"searchForMissingEpisodes": True,
"monitor": "all",
},
}
result = sonarr_post("series", payload)
return jsonify({"success": True, "title": result.get("title")})
except requests.HTTPError as e:
if e.response is not None and e.response.status_code == 400:
return jsonify({"error": "Series may already exist in Sonarr"}), 409
return jsonify({"error": str(e)}), 500
except Exception as e:
return jsonify({"error": str(e)}), 500
if __name__ == "__main__":
# PORT is read at run time, not import time, so a dev harness can hand us
# a free port. Defaults to 5000 to match EXPOSE/gunicorn and the
# `${HOST_PORT}:5000` mapping in docker-compose.yml — don't change the
# default without changing all three.
app.run(host="0.0.0.0", port=int(os.environ.get("PORT", "5000")))