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", "2.3.0")
APP_VERSION_DATE = os.environ.get("APP_VERSION_DATE", "2026-09-05")
# 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
# A show has to have sat in Sonarr longer than this, untouched, before the
# Admin section will suggest deleting it — keeps something just downloaded
# and not yet gotten to from showing up next to a delete button.
ADMIN_UNWATCHED_MONTHS = 6
# 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 sonarr_series_cached():
"""The full series list, behind the 5-minute cache.
Split out of get_sonarr_summary so the library and home pages can reuse
the same cached payload instead of each paying for the fetch.
"""
global _sonarr_cache
now = time.time()
if _sonarr_cache["data"] is not None and now < _sonarr_cache["expires"]:
return _sonarr_cache["data"]
series = sonarr_get("series")
_sonarr_cache = {"data": series, "expires": now + ARR_CACHE_TTL}
return series
def get_sonarr_summary():
try:
series = sonarr_series_cached()
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 radarr_movies_cached():
"""The full movie list, behind the 5-minute cache. See sonarr_series_cached."""
global _radarr_cache
now = time.time()
if _radarr_cache["data"] is not None and now < _radarr_cache["expires"]:
return _radarr_cache["data"]
movies = radarr_get("movie")
_radarr_cache = {"data": movies, "expires": now + ARR_CACHE_TTL}
return movies
def get_radarr_summary():
try:
movies = radarr_movies_cached()
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_kbps = avg_bytes_per_sec / 1024
avg_speed = _format_speed(avg_speed_kbps)
# Keep the newest completed items too — "Landed today" on the home
# page and the activity feed both read these. Same request as before;
# previously only the aggregates survived.
newest = sorted(recent, key=lambda x: x.get("completed", 0), reverse=True)[:20]
recent_items = [{
"name": s_.get("name", "Unknown"),
"category": s_.get("category", ""),
"size": s_.get("size", ""),
"completed": int(s_.get("completed", 0) or 0),
} for s_ in newest]
result = {"week_size": week_size, "day_size": day_size,
"completed_7d": completed_7d, "avg_speed": avg_speed,
"avg_speed_kbps": avg_speed_kbps, "recent_items": recent_items}
_sab_history_cache = {"data": result, "expires": now + 600}
return result
except Exception:
return {"week_size": "—", "day_size": "—", "completed_7d": "—",
"avg_speed": "—", "avg_speed_kbps": None, "recent_items": []}
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"],
"avg_speed_kbps": hist["avg_speed_kbps"],
"recent_items": hist.get("recent_items", []),
}
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 _ms_to_hm(ms):
"""Milliseconds to a compact '1h 42m' / '38m' label."""
if not ms or ms < 0:
return ""
mins = int(ms // 60000)
h, m = divmod(mins, 60)
return f"{h}h {m:02d}m" if h else f"{m}m"
def _ends_at(remaining_ms):
"""Wall-clock time this stream finishes, as HH:MM. Server-local, which is
the same timezone the header clock renders in."""
if not remaining_ms or remaining_ms < 0:
return ""
return (datetime.now() + timedelta(milliseconds=remaining_ms)).strftime("%H:%M")
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"]
active_users = cached["active_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
# Active viewers (10d) — everyone with a play in the window, for
# the Watch page's "click a name for their 48h history" list.
# A dedicated call rather than reusing top_played/top_users
# above: those are capped at 3 and tuned for different windows
# (all-time-ish top-3, 30d top-10), neither of which is "who's
# been around lately".
cutoff_date = (datetime.now() - timedelta(days=10)).strftime("%Y-%m-%d")
active_data = tautulli_get("get_history", f"&length=1000&order_column=date&order_dir=desc&after={cutoff_date}")
active_slots = []
if isinstance(active_data, dict):
active_slots = active_data.get("data", [])
if isinstance(active_slots, dict):
active_slots = active_slots.get("data", [])
active_by_user = {}
for entry in active_slots:
username = entry.get("friendly_name") or entry.get("user", "Unknown")
if username.lower() == "ninja_hippo":
continue
rec = active_by_user.setdefault(username, {"username": username, "plays": 0, "last_watched_ts": 0})
rec["plays"] += 1
rec["last_watched_ts"] = max(rec["last_watched_ts"], entry.get("date", 0) or 0)
active_users = sorted(active_by_user.values(), key=lambda x: x["last_watched_ts"], reverse=True)
for u in active_users:
u["last_watched"] = (datetime.fromtimestamp(u["last_watched_ts"]).strftime("%d %b · %H:%M")
if u["last_watched_ts"] else "")
_tautulli_stats_cache = {
"data": {"top_tv_list": top_tv_list, "top_movie_list": top_movie_list,
"top_played": top_played, "top_users": top_users,
"active_users": active_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:
try:
progress = int(float(s.get("progress_percent", 0) or 0))
except (TypeError, ValueError):
progress = 0
try:
duration_ms = int(float(s.get("duration", 0) or 0))
offset_ms = int(float(s.get("view_offset", 0) or 0))
except (TypeError, ValueError):
duration_ms = offset_ms = 0
decision = (s.get("transcode_decision") or "").lower()
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", ""),
# Everything below comes from the same get_activity payload.
"progress": max(0, min(100, progress)),
"player": s.get("player", ""),
"quality": s.get("video_full_resolution") or s.get("quality_profile", ""),
"decision": decision,
"year": s.get("year", ""),
"duration_ms": duration_ms,
"offset_ms": offset_ms,
"elapsed": _ms_to_hm(offset_ms),
"remaining": _ms_to_hm(max(0, duration_ms - offset_ms)),
"ends_at": _ends_at(duration_ms - offset_ms),
"art": s.get("art", ""),
"thumb": s.get("thumb", ""),
})
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_users": active_users,
"active_streams": stream_count,
"sessions": sessions,
}
except Exception as e:
return {"status": "error", "error": str(e)}
def get_user_watch_history(username, hours=48):
"""One viewer's plays in the last `hours`, for the Watch page's
per-user drill-down. Fetched on demand, not cached — this is a click
target, not something every page load needs to pay for."""
try:
cutoff_ts = time.time() - hours * 3600
encoded = requests.utils.quote(username)
hist = tautulli_get("get_history", f"&user={encoded}&length=200&order_column=date&order_dir=desc")
slots = []
if isinstance(hist, dict):
slots = hist.get("data", [])
if isinstance(slots, dict):
slots = slots.get("data", [])
items = []
for entry in slots:
ts = entry.get("date", 0) or 0
if ts < cutoff_ts:
continue
try:
percent = int(float(entry.get("percent_complete", 0) or 0))
except (TypeError, ValueError):
percent = 0
items.append({
"title": entry.get("full_title") or entry.get("title", "Unknown"),
"media_type": entry.get("media_type", ""),
"percent": max(0, min(100, percent)),
"when": datetime.fromtimestamp(ts).strftime("%d %b · %H:%M") if ts else "",
})
# Note: "items" collides with dict.items in Jinja attribute lookup
# (getattr wins over __getitem__), so the template key is "entries".
return {"status": "ok", "entries": items}
except Exception as e:
return {"status": "error", "error": str(e)}
def _fmt_bytes(b):
if b >= 1024**3:
return f"{b / 1024**3:.1f} GB"
return f"{b / 1024**2:.0f} MB"
def get_admin_unwatched_shows(limit=10, months=ADMIN_UNWATCHED_MONTHS):
"""Sonarr shows with episodes on disk that Tautulli has never played,
largest first — the source list for P105's Admin section.
Excludes anything added to Sonarr within the last `months` months, so a
show that's simply new and hasn't been gotten to yet doesn't turn up
next to a delete button. Matches against Tautulli's full play history
(a large `length`, not the 30/100-play windows the rest of the
dashboard uses) because a recency-limited window could mismark a show
that genuinely was watched — just a while ago — as never-watched,
which matters a lot more here than elsewhere given what's attached to
the wrong answer. Each show also carries whoever's Ombi request
triggered it, best-effort by title match — a show added straight in
Sonarr rather than requested has no requester to find.
"""
try:
series_list = sonarr_series_cached()
except Exception as e:
return {"status": "error", "error": str(e)}
try:
hist = tautulli_get("get_history", "&length=5000&order_column=date&order_dir=desc")
except Exception as e:
return {"status": "error", "error": str(e)}
played_titles = set()
slots = hist.get("data", []) if isinstance(hist, dict) else []
if isinstance(slots, dict):
slots = slots.get("data", [])
for entry in slots:
t = entry.get("grandparent_title") or entry.get("full_title") or entry.get("title", "")
if t:
played_titles.add(t.lower())
cutoff = datetime.utcnow() - timedelta(days=30 * months)
never = []
for s in series_list:
stats = s.get("statistics") or {}
size = stats.get("sizeOnDisk", 0) or s.get("sizeOnDisk", 0)
if not size:
continue
title = s.get("title", "")
if title.lower() in played_titles:
continue
added = s.get("added", "") or ""
try:
added_dt = datetime.strptime(added[:19], "%Y-%m-%dT%H:%M:%S")
if added_dt > cutoff:
continue # too recently added to fairly call it neglected
except (ValueError, TypeError):
pass # no usable added date — don't let a bad field hide a candidate
never.append({
"sonarr_id": s.get("id"),
"title": title,
"year": s.get("year", ""),
"size": size,
"ep_count": stats.get("episodeFileCount", 0),
"added": added,
})
never.sort(key=lambda x: x["size"], reverse=True)
top = never[:limit]
# Sonarr doesn't know who asked for a show — Ombi does, via whoever's
# request triggered the grab. Best-effort only: a show added straight
# in Sonarr, without an Ombi request, has no requester to find, and an
# unreachable Ombi shouldn't take down the rest of the section.
requested_by = {}
try:
for show in ombi_get("Request/tv"):
for child in show.get("childRequests", []):
user = child.get("requestedUser", {})
alias = (user.get("alias") or user.get("userAlias")
or user.get("userName") or "").split("@")[0]
if alias:
requested_by.setdefault(show.get("title", "").lower(), alias)
except Exception:
pass
for show in top:
show["size_display"] = _fmt_bytes(show["size"])
show["requested_by"] = requested_by.get(show["title"].lower(), "")
total_bytes = sum(show["size"] for show in top)
return {
"status": "ok",
"shows": top,
"months": months,
"total_display": _fmt_bytes(total_bytes) if top else "0 MB",
}
# ---------------------------------------------------------------------------
# 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_write(method, endpoint, payload):
"""Ombi's approve/deny endpoints answer 200 with
{"result": false, "errorMessage": ...} on a logical refusal rather than
an HTTP error, so the body has to be checked and not just the status."""
headers = {"ApiKey": OMBI_API_KEY, "Content-Type": "application/json"}
resp = requests.request(method, f"{OMBI_URL}/api/v1/{endpoint}",
headers=headers, json=payload, timeout=10)
resp.raise_for_status()
try:
body = resp.json()
except ValueError:
return {}
if isinstance(body, dict) and body.get("result") is False:
raise RuntimeError(body.get("errorMessage") or "Ombi refused the decision")
return body
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({
"id": m.get("id"),
"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({
# Ombi's tv approve/deny act on the child request, not
# the parent show, so this is the child's id.
"id": child.get("id"),
"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 fetched for its temperature only — it mirrors the array
# rather than holding a filesystem, so the schema notes fsSize/fsFree
# are "Not present on Parity type drive". It gets no table row; the
# card header carries the temp so a hot parity is still visible.
query = """{
vars { version }
array {
state
disks { name device fsSize fsFree type temp isSpinning }
caches { name device fsSize fsFree type temp isSpinning }
parities { name device 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", [])],
# Temperature only — no size fields exist on parity.
"parities": [
{
"name": dk.get("name") or dk.get("device") or "parity",
"device": dk.get("device", ""),
"temp": dk.get("temp"),
"spinning": dk.get("isSpinning"),
}
for dk in arr.get("parities", [])
],
}
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("home"))
# 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("home"))
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})
# ---------------------------------------------------------------------------
# Page plumbing
#
# Each page fetches only the services it actually renders. The old single
# dashboard route fetched all nine every time; now /system never pays for the
# Rotten Tomatoes scrape and /requests never pays for the 4.4 MB Radarr list.
# ---------------------------------------------------------------------------
PAGE_TASKS = {
"sonarr": lambda: get_sonarr_summary(),
"radarr": lambda: get_radarr_summary(),
"sabnzbd": lambda: get_sabnzbd_summary(),
"tautulli": lambda: get_tautulli_summary(),
"prowlarr": lambda: get_prowlarr_summary(),
"ombi": lambda: get_ombi_summary(),
"unraid": lambda: get_unraid_stats(),
"rt_movies": lambda: get_rt_movies(),
"rt_tv": lambda: get_rt_tv(),
"admin_unwatched": lambda: get_admin_unwatched_shows(),
}
PAGE_TIMEOUT = 5.0
def fetch_services(names, timeout=PAGE_TIMEOUT):
"""Fetch the named services in parallel under a hard wall-clock deadline.
Anything that misses the deadline degrades to an error dict rather than
blocking the page — the behaviour the single dashboard route always had,
lifted out so every page gets it.
"""
tasks = {n: PAGE_TASKS[n] for n in names}
results = {}
if not tasks:
return results
with ThreadPoolExecutor(max_workers=len(tasks)) as ex:
futures = {ex.submit(fn): name for name, fn in tasks.items()}
deadline = time.time() + timeout
try:
for future in as_completed(futures, timeout=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)}
except Exception:
pass # as_completed timed out; the fill-in below covers the rest
for name in tasks:
if name not in results:
results[name] = {"status": "error", "error": "timeout"}
return results
def service_urls():
"""Public URLs for the 'open in ' links, via the reverse proxy."""
arr_host_public = "https://hippofam.com"
return {
"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}",
}
# Every destination, for the top of every page but the index, each
# with its own distinct colour. Order matches nav_items. The four
# with a real keyboard shortcut (r/g/y/c) keep their original colours;
# Requests gets magenta (the same --pending magenta a request already
# shows once approved elsewhere — Requests IS "waiting on a human");
# Watch gets blue, ~2.4:1 contrast on black versus the 4.5:1 every
# other body text here clears — a known, deliberate exception, kept
# anyway because a swatch-only accent read as unfinished next to six
# fully-coloured neighbours. That's every colour the locked eight
# palette has to give (black's invisible on black); Activity's colour,
# --tt-orange, is a genuine ninth on top of it — a dithered swatch
# alone read as unfinished, and dithering the word itself renders
# illegible, so it's a deliberate, acknowledged break from "eight
# colours, no ninth" rather than a fix that stays inside it. Settings
# alone keeps plain ink — deliberately, as the one page with no live
# status of its own, so it's the one that gets no colour rather than
# an arbitrary leftover.
FULL_FASTEXT = [
("k-red", "r", "100", "Index", "/"),
("k-blue", "", "101", "Watch", "/watch"),
("k-green", "g", "102", "Library", "/library"),
("k-yellow", "y", "103", "Discover", "/discover"),
("k-magenta", "", "104", "Requests", "/requests"),
("k-cyan", "c", "105", "System", "/system"),
("k-orange", "", "106", "Activity", "/activity"),
("", "", "900", "Settings", "/settings"),
]
# The index's own closing row: just the four letter keys, since the
# index's body already lists every one of the above once.
HOME_FASTEXT = [
("k-red", "r", "100", "Index", "/"),
("k-green", "g", "102", "Library", "/library"),
("k-yellow", "y", "103", "Discover", "/discover"),
("k-cyan", "c", "105", "System", "/system"),
]
@app.context_processor
def inject_page_chrome():
"""Shell-wide context: the nav, the clock's timezone, service links."""
s = load_settings()
return {
"timezone": s.get("timezone", "America/New_York"),
"service_urls": service_urls(),
# (key, page number, name, href). The page number is the
# navigation: it shows in the header, it's what the visitor
# types to jump, and it's the row label on the index.
"nav_items": [
("home", "100", "Index", "/"),
("watch", "101", "Watch", "/watch"),
("library", "102", "Library", "/library"),
("discover", "103", "Discover", "/discover"),
("requests", "104", "Requests", "/requests"),
("system", "105", "System", "/system"),
("activity", "106", "Activity", "/activity"),
("settings", "900", "Settings", "/settings"),
],
# Every page, at the top of every page but the index — which
# overrides this with HOME_FASTEXT, its own short closing row,
# since its body already lists all of these once. The four
# letter keys (r/g/y/c) keep their original destinations; the
# rest have no keyboard shortcut, just a click.
"fastext": FULL_FASTEXT,
}
# ---------------------------------------------------------------------------
# Derived views over data the fetchers already cache
# ---------------------------------------------------------------------------
def _poster_url(images):
for img in images or []:
if img.get("coverType") == "poster":
return img.get("remoteUrl") or img.get("url") or ""
return ""
def _ago(iso):
"""'2024-08-27T10:00:00Z' -> '2 days'. Empty string if unparseable."""
if not iso:
return ""
try:
dt = datetime.strptime(iso[:19], "%Y-%m-%dT%H:%M:%S")
except (ValueError, TypeError):
return ""
days = (datetime.utcnow() - dt).days
if days <= 0:
return "today"
if days == 1:
return "1 day"
if days < 7:
return f"{days} days"
weeks = days // 7
return "1 wk" if weeks == 1 else f"{weeks} wks"
def get_recently_added(limit=6):
"""Newest titles across both libraries.
Derived from the same cached Sonarr/Radarr lists the summaries already
hold — a sort over the `added` field those payloads carry, not a new
API call. Poster URLs come from the same records.
"""
items = []
try:
for sr in sonarr_series_cached():
items.append({
"title": sr.get("title", "Unknown"),
"sub": f"{sr.get('statistics', {}).get('episodeFileCount', 0)} eps",
"kind": "TV",
"added": sr.get("added", "") or "",
"poster": _poster_url(sr.get("images")),
})
except Exception:
pass
try:
for mv in radarr_movies_cached():
if not mv.get("hasFile"):
continue
items.append({
"title": mv.get("title", "Unknown"),
"sub": str(mv.get("year", "")) or "Film",
"kind": "Film",
"added": mv.get("added", "") or "",
"poster": _poster_url(mv.get("images")),
})
except Exception:
pass
items.sort(key=lambda x: x["added"], reverse=True)
items = items[:limit]
for it in items:
it["ago"] = _ago(it["added"])
return items
def search_library(q, limit=24):
"""Titles in Sonarr/Radarr matching q, case-insensitive substring.
Filters the same cached lists get_recently_added uses — no new fetch.
"""
needle = q.strip().lower()
if not needle:
return []
items = []
try:
for sr in sonarr_series_cached():
if needle in sr.get("title", "").lower():
items.append({
"title": sr.get("title", "Unknown"),
"kind": "TV",
"year": str(sr.get("year", "")),
"poster": _poster_url(sr.get("images")),
})
except Exception:
pass
try:
for mv in radarr_movies_cached():
if needle in mv.get("title", "").lower():
items.append({
"title": mv.get("title", "Unknown"),
"kind": "Film",
"year": str(mv.get("year", "")),
"poster": _poster_url(mv.get("images")),
})
except Exception:
pass
items.sort(key=lambda x: x["title"].lower())
return items[:limit]
def arr_search(q, limit=12):
"""Movies and TV matching q, straight from Radarr's and Sonarr's own
lookup endpoints — the same ones /add-to-radarr and /add-to-sonarr
already use to resolve a title before adding. No Ombi round trip and
no second title-matching path to keep in sync with the Add buttons.
"""
term = q.strip()
if not term:
return {"status": "ok", "results": []}
encoded = requests.utils.quote(term)
results = []
radarr_ok = sonarr_ok = True
try:
radarr_titles = {t.lower() for t in get_radarr_summary().get("radarr_titles", [])}
for m in radarr_get(f"movie/lookup?term={encoded}")[:limit]:
title = m.get("title", "")
results.append({
"title": title,
"year": str(m.get("year", "")),
"poster": _poster_url(m.get("images")),
"kind": "Film",
"target": "radarr",
"held": title.lower() in radarr_titles,
})
except Exception:
radarr_ok = False
try:
sonarr_titles = {t.lower() for t in get_sonarr_summary().get("sonarr_titles", [])}
for s in sonarr_get(f"series/lookup?term={encoded}")[:limit]:
title = s.get("title", "")
results.append({
"title": title,
"year": str(s.get("year", "")),
"poster": _poster_url(s.get("images")),
"kind": "TV",
"target": "sonarr",
"held": title.lower() in sonarr_titles,
})
except Exception:
sonarr_ok = False
if not results and not radarr_ok and not sonarr_ok:
return {"status": "error", "error": "Sonarr and Radarr both unreachable", "results": []}
return {"status": "ok", "results": results}
def get_activity_feed(sabnzbd, ombi, limit=25):
"""A single chronological feed assembled from what the fetchers return.
Three real event sources, no new calls: SABnzbd completions, Ombi
requests, and library additions. Anything without a usable timestamp is
dropped rather than guessed at.
"""
events = []
for it in (sabnzbd or {}).get("recent_items", []):
ts = it.get("completed", 0)
if not ts:
continue
events.append({
"ts": ts, "kind": "grabbed", "label": "Grabbed",
"title": it.get("name", "Unknown"),
"detail": " · ".join(x for x in [it.get("size", ""), it.get("category", "")] if x),
})
for r in (ombi or {}).get("recent", []):
iso = r.get("requested_date", "")
try:
ts = datetime.strptime(iso[:19], "%Y-%m-%dT%H:%M:%S").timestamp()
except (ValueError, TypeError):
continue
events.append({
"ts": ts, "kind": "requested", "label": "Requested",
"title": r.get("title", "Unknown"),
"detail": f"{r.get('user', '')} · {r.get('status', '')}",
})
for it in get_recently_added(limit=12):
iso = it.get("added", "")
try:
ts = datetime.strptime(iso[:19], "%Y-%m-%dT%H:%M:%S").timestamp()
except (ValueError, TypeError):
continue
events.append({
"ts": ts, "kind": "added", "label": "Added",
"title": it.get("title", "Unknown"),
"detail": f"{it.get('kind', '')} · {it.get('sub', '')}",
})
events.sort(key=lambda e: e["ts"], reverse=True)
for e in events:
e["when"] = datetime.fromtimestamp(e["ts"]).strftime("%d %b · %H:%M")
return events[:limit]
def _filter_rt(raw, owned_titles, limit=10):
"""Drop anything already in the library; fall back to the unfiltered list
if that leaves nothing (better a stale suggestion than an empty card)."""
filtered = [x for x in raw if not x.get("error")
and x.get("title", "").lower() not in owned_titles][:limit]
return filtered or raw
# ---------------------------------------------------------------------------
# Pages
# ---------------------------------------------------------------------------
@app.route("/")
@login_required
def home():
"""Marquee: what's playing, one line of system state, then what's new."""
r = fetch_services(["sonarr", "radarr", "sabnzbd", "tautulli",
"prowlarr", "ombi", "unraid", "rt_movies", "rt_tv"])
sonarr, radarr = r["sonarr"], r["radarr"]
radarr_titles = set(radarr.get("radarr_titles", []))
sonarr_titles = set(sonarr.get("sonarr_titles", []))
tautulli = r["tautulli"]
sessions = tautulli.get("sessions", []) if tautulli.get("status") == "ok" else []
# The hero shows the furthest-along stream; the rest sit beside it.
hero = max(sessions, key=lambda s: s.get("progress", 0)) if sessions else None
others = [s for s in sessions if s is not hero]
return render_template(
"home.html", page="home", fastext=HOME_FASTEXT,
sonarr=sonarr, radarr=radarr, sabnzbd=r["sabnzbd"], tautulli=tautulli,
prowlarr=r["prowlarr"], ombi=r["ombi"], unraid=r["unraid"],
hero=hero, other_sessions=others,
# 3, not 6: P100's Just Added is a glance, not the library — a
# single fixed row keeps NOW SHOWING holding the lower two-thirds
# of the first viewport per the direction contract, and never
# orphans a lone tile the way a variable-column auto-fill grid
# could at 6. The full feed is what /library is for.
recently_added=get_recently_added(3),
rt_movies=_filter_rt(r["rt_movies"], radarr_titles, 6),
rt_tv=_filter_rt(r["rt_tv"], sonarr_titles, 6),
)
@app.route("/watch")
@login_required
def watch_page():
r = fetch_services(["tautulli"])
selected_user = request.args.get("user", "").strip()
user_history = get_user_watch_history(selected_user) if selected_user else None
return render_template("watch.html", page="watch", tautulli=r["tautulli"],
selected_user=selected_user, user_history=user_history)
@app.route("/library")
@login_required
def library_page():
r = fetch_services(["sonarr", "radarr"])
q = request.args.get("q", "").strip()
return render_template("library.html", page="library",
sonarr=r["sonarr"], radarr=r["radarr"],
recently_added=get_recently_added(12),
q=q, search_results=search_library(q) if q else None)
@app.route("/discover")
@login_required
def discover_page():
r = fetch_services(["sonarr", "radarr", "rt_movies", "rt_tv"])
sonarr_titles = set(r["sonarr"].get("sonarr_titles", []))
radarr_titles = set(r["radarr"].get("radarr_titles", []))
q = request.args.get("q", "").strip()
return render_template("discover.html", page="discover",
rt_movies=_filter_rt(r["rt_movies"], radarr_titles),
rt_tv=_filter_rt(r["rt_tv"], sonarr_titles),
q=q, search=arr_search(q) if q else None)
@app.route("/requests")
@login_required
def requests_page():
r = fetch_services(["ombi"])
return render_template("requests.html", page="requests", ombi=r["ombi"])
@app.route("/system")
@login_required
def system_page():
r = fetch_services(["unraid", "prowlarr", "sonarr", "radarr",
"sabnzbd", "tautulli", "ombi", "admin_unwatched"])
return render_template("system.html", page="system",
unraid=r["unraid"], prowlarr=r["prowlarr"],
sonarr=r["sonarr"], radarr=r["radarr"],
sabnzbd=r["sabnzbd"], tautulli=r["tautulli"],
ombi=r["ombi"], admin_unwatched=r["admin_unwatched"])
@app.route("/api/array/refresh")
@login_required
def api_array_refresh():
"""Re-renders just the Array section's markup for P105's refresh
button. Unraid's stats aren't cached (get_unraid_stats hits the
GraphQL API live on every call), so this is a plain re-fetch — the
button exists for the visitor's own peace of mind, not to bypass a
cache. Returns HTML, not JSON: swapped directly into #array-section
rather than re-implementing the table/bar/disk-row markup in JS."""
return render_template("_array_section.html", unraid=get_unraid_stats())
@app.route("/activity")
@login_required
def activity_page():
r = fetch_services(["sabnzbd", "ombi"])
return render_template("activity.html", page="activity",
events=get_activity_feed(r["sabnzbd"], r["ombi"]))
@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/ombi/decide", methods=["POST"])
@login_required
def api_ombi_decide():
"""Approve or deny one pending request from P104.
Ombi splits this four ways — movie/tv x approve/deny — and approve is a
POST while deny is a PUT, so both halves are named explicitly rather
than derived."""
data = request.get_json(silent=True) or {}
req_id = data.get("id")
kind = (data.get("type") or "").lower()
verdict = (data.get("verdict") or "").lower()
if not req_id:
return jsonify({"error": "No request id"}), 400
if kind not in ("movie", "tv"):
return jsonify({"error": "Unknown request type"}), 400
if verdict not in ("approve", "deny"):
return jsonify({"error": "Unknown verdict"}), 400
try:
if verdict == "approve":
ombi_write("POST", f"Request/{kind}/approve", {"id": req_id})
else:
ombi_write("PUT", f"Request/{kind}/deny", {"id": req_id, "reason": ""})
# Without this the 4-hour cache keeps calling the row pending for the
# rest of the afternoon, and the page contradicts the decision just
# made on it.
_ombi_cache["data"] = None
_ombi_cache["expires"] = 0
return jsonify({"success": True,
"status": "Approved" if verdict == "approve" else "Denied"})
except Exception as e:
return jsonify({"error": str(e)}), 500
@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():
"""Sonarr shows that have never been played in Tautulli — see the
Admin section on P105 (`get_admin_unwatched_shows`), which this
delegates to rather than duplicating the match/sort logic."""
months = request.args.get("months", default=ADMIN_UNWATCHED_MONTHS, type=int)
limit = request.args.get("limit", default=30, type=int)
result = get_admin_unwatched_shows(limit=limit, months=months)
if result.get("status") != "ok":
return jsonify(result), 500
return jsonify(result)
@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()
_radarr_cache["data"] = None
_radarr_cache["expires"] = 0
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()
_sonarr_cache["data"] = None
_sonarr_cache["expires"] = 0
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", page="settings",
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")))