Restructure into backend/ and frontend/ subprojects

- backend/ uses proper Python src layout (src/media_library_viewer_api/)
  with pyproject.toml, hatchling build, and PYTHONPATH=src convention
- frontend/ is a Vite + React + TypeScript SPA
- archive/ preserves the original Streamlit prototype for reference
- Cleaned up root to only contain docs, license, and subproject dirs
- Updated README for the new dual-subproject architecture
This commit is contained in:
2026-04-30 21:48:46 +02:00
parent 3c432473e5
commit 51b10438a9
47 changed files with 127 additions and 130 deletions
+23
View File
@@ -0,0 +1,23 @@
"""Convenience Streamlit entrypoint.
The real application lives in :mod:`media_library_viewer.app` under ``src/`` so
that the project can be packaged and reused by other frontends later. This file
keeps the simple development command working:
streamlit run app.py
"""
from __future__ import annotations
import sys
from pathlib import Path
SRC = Path(__file__).resolve().parent / "src"
if str(SRC) not in sys.path:
sys.path.insert(0, str(SRC))
from media_library_viewer.app import main # noqa: E402
if __name__ == "__main__":
main()
+42
View File
@@ -0,0 +1,42 @@
[build-system]
requires = ["hatchling>=1.24"]
build-backend = "hatchling.build"
[project]
name = "media-library-viewer"
version = "0.1.0"
description = "Streamlit app for browsing Jellyfin libraries and inspecting remote media files over SSH."
readme = "README.md"
requires-python = ">=3.11"
license = { text = "MIT" }
authors = [{ name = "Media Library Viewer contributors" }]
dependencies = [
"streamlit>=1.35",
"streamlit-aggrid>=1.0",
"requests>=2.31",
"paramiko>=3.4",
"python-dotenv>=1.0",
"pandas>=2.0",
]
[project.optional-dependencies]
dev = [
"ruff>=0.4",
"pytest>=8.0",
]
[tool.hatch.build.targets.wheel]
packages = ["src/media_library_viewer"]
[tool.ruff]
line-length = 120
target-version = "py311"
src = ["src", "tests"]
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B"]
ignore = ["E501"]
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]
+1
View File
@@ -0,0 +1 @@
-e .
@@ -0,0 +1,4 @@
"""Media Library Viewer package."""
__all__ = ["__version__"]
__version__ = "0.1.0"
+277
View File
@@ -0,0 +1,277 @@
"""Streamlit UI adapter for Media Library Viewer.
This module is intentionally thin. Reusable logic lives in:
- ``clients/`` for Jellyfin, SSH, and resource collection integrations.
- ``domain/`` for UI-independent normalization of Jellyfin/media metadata.
- ``services/`` for application services such as the SQLite media index.
- ``ui/`` for Streamlit rendering grouped by feature area.
Keeping those boundaries makes it easier to replace this Streamlit frontend with
another frontend later while preserving the backend/domain code.
"""
from __future__ import annotations
import json
import posixpath
from typing import Any
import streamlit as st
from media_library_viewer.clients.jellyfin import JellyfinClient
from media_library_viewer.clients.ssh import RemoteSSHClient
from media_library_viewer.config import load_config
from media_library_viewer.ui.dashboard import render_media_overview, render_now_playing, render_resource_dashboard
from media_library_viewer.ui.file_browser import render_file_browser, set_file_browser_path
from media_library_viewer.ui.media import render_media_tab
from media_library_viewer.ui.preview import render_ssh_tools
@st.cache_resource(show_spinner=False)
def get_jellyfin_client(base_url: str, api_key: str) -> JellyfinClient:
"""Return a cached Jellyfin client for the current server/API-key pair."""
return JellyfinClient(base_url, api_key)
@st.cache_data(ttl=300, show_spinner=False)
def cached_users(base_url: str, api_key: str):
"""Cache the Jellyfin user list to avoid repeated auth-scoped requests."""
return get_jellyfin_client(base_url, api_key).users()
@st.cache_resource(show_spinner=False)
def get_ssh_client(host: str, username: str, port: int, key_filename: str, password: str) -> RemoteSSHClient:
"""Return a cached SSH connection wrapper for the active remote server."""
client = RemoteSSHClient(
host=host,
username=username,
port=port,
key_filename=key_filename or None,
password=password or None,
)
client.connect()
return client
@st.cache_data(ttl=60, show_spinner=False)
def cached_libraries(base_url: str, api_key: str, user_id: str):
"""Cache library views for the selected Jellyfin user."""
return get_jellyfin_client(base_url, api_key).libraries(user_id)
@st.cache_data(ttl=300, show_spinner=False)
def cached_media_counts(base_url: str, api_key: str, user_id: str):
"""Cache dashboard-level media counts for movies/series/episodes."""
return get_jellyfin_client(base_url, api_key).media_counts(user_id)
@st.cache_data(ttl=300, show_spinner=False)
def cached_library_counts(base_url: str, api_key: str, user_id: str):
"""Cache per-library item counts for dashboard breakdown."""
client = get_jellyfin_client(base_url, api_key)
libraries = client.libraries(user_id)
return client.library_item_counts(user_id, libraries)
@st.cache_data(ttl=15, show_spinner=False)
def cached_active_sessions(base_url: str, api_key: str):
"""Cache active Jellyfin sessions briefly for dashboard now-playing status."""
return get_jellyfin_client(base_url, api_key).active_sessions()
@st.cache_data(ttl=30, show_spinner=False)
def cached_dir_listing(host: str, username: str, port: int, key_filename: str, password: str, path: str):
"""Cache remote directory listings briefly for snappier browsing."""
ssh = get_ssh_client(host, username, port, key_filename, password)
result = ssh.list_dir(path)
if result.exit_status != 0:
raise RuntimeError(result.stderr or result.stdout)
return json.loads(result.stdout)
@st.cache_data(ttl=300, show_spinner=False)
def cached_ffprobe_preview(host: str, username: str, port: int, key_filename: str, password: str, path: str):
"""Cache selected-file ffprobe previews so repeated reruns stay responsive."""
ssh = get_ssh_client(host, username, port, key_filename, password)
return ssh.ffprobe_json(path)
def apply_remote_path_prefix(path: str, prefix: str) -> str:
"""Apply an optional fallback prefix for Jellyfin->SSH path handoff."""
if not path:
return path
normalized_prefix = (prefix or "").strip()
if not normalized_prefix:
return path
normalized_prefix = normalized_prefix.rstrip("/")
if path == normalized_prefix or path.startswith(normalized_prefix + "/"):
return posixpath.normpath(path)
if path.startswith("/"):
return posixpath.normpath(normalized_prefix + path)
return posixpath.normpath(posixpath.join(normalized_prefix, path))
def map_path_to_media_root(path: str, media_root: str) -> str:
"""Map a Jellyfin path to the configured SSH media root when possible.
Example:
- path: ``/media/shows/Show/E01.mkv``
- media_root: ``/srv/media``
- result: ``/srv/media/shows/Show/E01.mkv``
If the path is already under ``media_root``, it is returned unchanged.
If the final segment of ``media_root`` (e.g. ``media``) appears in the
Jellyfin path, the prefix up to that segment is replaced by ``media_root``.
"""
if not path:
return path
normalized_root = (media_root or "").strip()
if not normalized_root:
return path
normalized_root = posixpath.normpath(normalized_root)
raw_parts = [part for part in str(path).split("/") if part]
if not raw_parts:
return path
path_absolute = "/" + "/".join(raw_parts)
if path_absolute == normalized_root or path_absolute.startswith(normalized_root + "/"):
return path_absolute
root_anchor = posixpath.basename(normalized_root)
if not root_anchor:
return path
if root_anchor in raw_parts:
anchor_index = raw_parts.index(root_anchor)
remainder_parts = raw_parts[anchor_index + 1 :]
return posixpath.join(normalized_root, *remainder_parts) if remainder_parts else normalized_root
return path
def resolve_remote_media_path(path: str, media_root: str, fallback_prefix: str) -> str:
"""Resolve Jellyfin paths to SSH-visible paths.
Strategy:
1. Prefer mapping to ``REMOTE_MEDIA_ROOT`` when it can anchor on the root
basename (e.g. ``media`` in ``/srv/media``).
2. If no mapping happened, apply optional fallback prefix.
"""
if not path:
return path
mapped = map_path_to_media_root(path, media_root)
if mapped and mapped != path:
return mapped
return apply_remote_path_prefix(mapped or path, fallback_prefix)
def credentials_panel():
"""Render connection settings and return normalized Jellyfin/SSH inputs."""
cfg = load_config()
with st.sidebar:
st.header("Connections")
with st.expander("Jellyfin", expanded=True):
jellyfin_url = st.text_input("URL", value=cfg.jellyfin.url, placeholder="https://jellyfin.example.com")
jellyfin_api_key = st.text_input("API key", value=cfg.jellyfin.api_key, type="password")
jellyfin_user_id = st.text_input(
"User ID override",
value=cfg.jellyfin.user_id,
help="Optional. API keys are not user sessions, so the app normally lists users with /Users and lets you choose one.",
)
with st.expander("SSH", expanded=True):
ssh_host = st.text_input("Host", value=cfg.ssh.host)
ssh_username = st.text_input("Username", value=cfg.ssh.username)
ssh_port = st.number_input("Port", min_value=1, max_value=65535, value=cfg.ssh.port)
ssh_key = st.text_input("Private key path", value=cfg.ssh.key_filename)
ssh_password = st.text_input("Password / passphrase", value=cfg.ssh.password, type="password")
media_root = st.text_input("Default media root", value=cfg.ssh.media_root, placeholder="/mnt/media")
remote_path_prefix = st.text_input(
"Path prefix for Jellyfin paths",
value=cfg.ssh.path_prefix,
placeholder="/srv",
help="Fallback prefix when REMOTE_MEDIA_ROOT mapping is not enough.",
)
return jellyfin_url, jellyfin_api_key, jellyfin_user_id, ssh_host, ssh_username, int(ssh_port), ssh_key, ssh_password, media_root, remote_path_prefix
def main():
"""Application entrypoint used by the root ``app.py`` wrapper."""
# Set page config per Streamlit session. Doing this at module import time is
# unreliable here because the root wrapper imports ``main`` from this module,
# and Python may reuse the already-imported module on browser reloads.
st.set_page_config(page_title="Media Library Viewer", layout="wide")
st.title("Media Library Viewer")
st.caption("Phase 1: Jellyfin browser + SSH filesystem inspection + safe remote job templates")
jellyfin_url, jellyfin_api_key, jellyfin_user_id, ssh_host, ssh_username, ssh_port, ssh_key, ssh_password, media_root, remote_path_prefix = credentials_panel()
if not jellyfin_url or not jellyfin_api_key:
st.info("Enter Jellyfin connection details in the sidebar.")
return
try:
client = get_jellyfin_client(jellyfin_url, jellyfin_api_key)
users = cached_users(jellyfin_url, jellyfin_api_key)
except Exception as exc:
st.error(f"Jellyfin connection failed: {exc}")
st.caption("Note: this app uses an API key with GET /Users. /Users/Me is only reliable for user access tokens, not server API keys.")
return
if jellyfin_user_id:
user_id = jellyfin_user_id
else:
if not users:
st.error("No Jellyfin users returned from /Users. Set JELLYFIN_USER_ID manually.")
return
user_options = {f"{user.get('Name', 'Unnamed')} ({user['Id']})": user["Id"] for user in users}
with st.sidebar:
selected_user = st.selectbox("Jellyfin user", list(user_options.keys()))
user_id = user_options[selected_user]
tab_dashboard, tab_monitoring, tab_media, tab_files = st.tabs(
["Dashboard", "Monitoring", "Media", "File browser"]
)
with tab_dashboard:
render_now_playing(cached_active_sessions, jellyfin_url, jellyfin_api_key)
st.divider()
if not ssh_host or not ssh_username:
st.info("Enter SSH connection details in the sidebar for server monitoring overview.")
else:
ssh_args = (ssh_host, ssh_username, ssh_port, ssh_key, ssh_password)
render_resource_dashboard(get_ssh_client, ssh_args, media_root or "/", detailed=False)
st.divider()
render_media_overview(cached_media_counts, cached_library_counts, jellyfin_url, jellyfin_api_key, user_id)
with tab_monitoring:
if not ssh_host or not ssh_username:
st.info("Enter SSH connection details in the sidebar.")
else:
ssh_args = (ssh_host, ssh_username, ssh_port, ssh_key, ssh_password)
render_resource_dashboard(get_ssh_client, ssh_args, media_root or "/", detailed=True)
libraries = cached_libraries(jellyfin_url, jellyfin_api_key, user_id)
def set_prefixed_file_browser_path(path: str, selected_path: str | None = None, reset_filters: bool = True) -> None:
set_file_browser_path(
resolve_remote_media_path(path, media_root, remote_path_prefix),
resolve_remote_media_path(selected_path, media_root, remote_path_prefix) if selected_path else None,
reset_filters,
)
with tab_media:
render_media_tab(client, user_id, libraries, set_prefixed_file_browser_path)
with tab_files:
if not ssh_host or not ssh_username:
st.info("Enter SSH connection details in the sidebar.")
return
ssh_args = (ssh_host, ssh_username, ssh_port, ssh_key, ssh_password)
selected_path = render_file_browser(cached_dir_listing, ssh_args, media_root or "/")
ssh = get_ssh_client(ssh_host, ssh_username, ssh_port, ssh_key, ssh_password)
render_ssh_tools(ssh, ssh_args, selected_path, cached_ffprobe_preview)
if __name__ == "__main__":
main()
@@ -0,0 +1,6 @@
"""External service clients.
Modules in this package talk to systems outside the app: Jellyfin over HTTP,
the media server over SSH, and the lightweight remote resource collector.
They should not import Streamlit.
"""
@@ -0,0 +1,167 @@
"""Jellyfin HTTP API client.
This module is deliberately independent from Streamlit. It wraps only the API
calls the app currently needs and returns plain Python dictionaries/lists so a
future FastAPI/React frontend can reuse the same client.
"""
from __future__ import annotations
from typing import Any
import requests
# Jellyfin validates Fields against its ItemFields enum. Keep this list to
# documented/commonly supported optional fields; invalid names cause 400s.
DEFAULT_FIELDS = ",".join(
[
"DateCreated",
"Genres",
"MediaSources",
"Overview",
"Path",
"People",
"PremiereDate",
"ProviderIds",
"Tags",
]
)
class JellyfinClient:
"""Small wrapper around the Jellyfin/Emby-compatible HTTP API."""
def __init__(self, base_url: str, api_key: str, timeout: int = 30):
if not base_url:
raise ValueError("Jellyfin URL is required")
if not api_key:
raise ValueError("Jellyfin API key is required")
# Use the server root, not the web UI path. Users often paste
# https://host/web; API endpoints live at https://host/...
self.base_url = base_url.rstrip("/")
if self.base_url.endswith("/web"):
self.base_url = self.base_url[:-4]
self.api_key = api_key
self.timeout = timeout
self.session = requests.Session()
self.session.headers.update(
{
"X-Emby-Token": api_key,
"Accept": "application/json",
"X-Emby-Authorization": 'MediaBrowser Client="MediaLibraryViewer", Device="Streamlit", DeviceId="streamlit", Version="0.1"',
}
)
def get(self, path: str, **params: Any) -> dict[str, Any]:
"""GET a Jellyfin endpoint and include useful response text on errors."""
clean_params = {k: v for k, v in params.items() if v is not None and v != ""}
response = self.session.get(
f"{self.base_url}{path}", params=clean_params, timeout=self.timeout
)
try:
response.raise_for_status()
except requests.HTTPError as exc:
detail = response.text[:500]
raise requests.HTTPError(
f"{response.status_code} for {response.url}: {detail}",
response=response,
) from exc
return response.json()
def users(self) -> list[dict[str, Any]]:
"""List users visible to this API key.
Jellyfin API keys are server-level tokens, not user session tokens, so
/Users/Me often fails with API-key auth. The user id selected here is
then used for user-scoped library endpoints.
"""
return self.get("/Users")
def libraries(self, user_id: str) -> list[dict[str, Any]]:
"""Return top-level library views visible to the selected Jellyfin user."""
return self.get(f"/Users/{user_id}/Views").get("Items", [])
def items(
self,
user_id: str,
parent_id: str | None = None,
start_index: int = 0,
limit: int = 50,
search: str | None = None,
include_item_types: str | None = None,
recursive: bool = True,
sort_by: str = "SortName",
sort_order: str = "Ascending",
) -> dict[str, Any]:
"""Return a paginated item list for a user/library.
This is used by both the visual library browser and the media-index
builder. Keep arguments close to Jellyfin's own query parameters so the
service layer can request server-side pagination and basic sorting.
"""
return self.get(
f"/Users/{user_id}/Items",
ParentId=parent_id,
StartIndex=start_index,
Limit=limit,
SearchTerm=search,
IncludeItemTypes=include_item_types,
Recursive=str(recursive).lower(),
Fields=DEFAULT_FIELDS,
SortBy=sort_by,
SortOrder=sort_order,
)
def item_count(self, user_id: str, include_item_types: str, parent_id: str | None = None) -> int:
"""Return a count using Jellyfin's TotalRecordCount without fetching rows."""
response = self.get(
f"/Users/{user_id}/Items",
ParentId=parent_id,
Recursive="true",
IncludeItemTypes=include_item_types,
Limit=0,
)
return int(response.get("TotalRecordCount", 0))
def media_counts(self, user_id: str) -> dict[str, int]:
"""Return dashboard-level counts for the main media types."""
return {
"movies": self.item_count(user_id, "Movie"),
"series": self.item_count(user_id, "Series"),
"episodes": self.item_count(user_id, "Episode"),
}
def library_item_counts(self, user_id: str, libraries: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Return per-library item counts broken down by type for the dashboard."""
results = []
for lib in libraries:
lib_id = lib.get("Id")
lib_name = lib.get("Name", "Unknown")
lib_type = lib.get("CollectionType", "")
if not lib_id:
continue
movies = self.item_count(user_id, "Movie", parent_id=lib_id)
series = self.item_count(user_id, "Series", parent_id=lib_id)
episodes = self.item_count(user_id, "Episode", parent_id=lib_id)
total = self.item_count(user_id, "Movie,Episode,Video,Audio,Series", parent_id=lib_id)
results.append({
"library": lib_name,
"type": lib_type,
"movies": movies,
"series": series,
"episodes": episodes,
"total": total,
})
return results
def active_sessions(self, active_within_seconds: int = 300) -> list[dict[str, Any]]:
"""Return currently active sessions that have a now-playing item."""
payload = self.get("/Sessions", ActiveWithinSeconds=active_within_seconds)
sessions = payload if isinstance(payload, list) else []
return [session for session in sessions if session.get("NowPlayingItem")]
def image_url(self, item_id: str, image_type: str = "Primary") -> str:
"""Build an authenticated image URL suitable for st.image/browser use."""
return f"{self.base_url}/Items/{item_id}/Images/{image_type}?api_key={self.api_key}"
@@ -0,0 +1,316 @@
"""Remote resource collection helpers.
The app does not require Prometheus, Netdata, or sysstat. Instead it can install
and manage a tiny POSIX-sh collector under /tmp on the remote server. The
collector samples Linux /proc and /sys counters every 10 seconds and appends JSON
Lines. This module starts/stops the collector and reads those JSONL samples.
"""
from __future__ import annotations
import json
import shlex
from dataclasses import dataclass
from typing import Any
from media_library_viewer.clients.ssh import RemoteSSHClient
# POSIX shell script copied to the remote server by start_resource_collector().
# Keep this script bash-free because many NAS/media servers have minimal shells.
COLLECTOR_SCRIPT = r'''#!/bin/sh
set -u
OUT="${1:-/tmp/media_library_viewer_metrics.jsonl}"
INTERVAL="${2:-10}"
RETENTION_SECONDS="${3:-604800}"
MAX_LINES="${4:-70000}"
PRUNE_EVERY_SAMPLES="${5:-60}"
mkdir -p "$(dirname "$OUT")"
echo "collector starting at $(date -Is 2>/dev/null || date), interval=${INTERVAL}s, retention=${RETENTION_SECONDS}s, max_lines=${MAX_LINES}, out=${OUT}"
read_cpu() {
awk '/^cpu / {print $2+$3+$4+$5+$6+$7+$8+$9+$10, $5+$6, $6}' /proc/stat
}
read_mem_pct() {
awk '
/^MemTotal:/ {total=$2}
/^MemAvailable:/ {avail=$2}
END {if (total > 0) printf "%.2f", (total-avail)*100/total; else printf "0"}
' /proc/meminfo
}
read_net_bytes() {
awk '
NR > 2 {
split($0, parts, ":")
iface = parts[1]
stats = parts[2]
gsub(/^[ \t]+|[ \t]+$/, "", iface)
gsub(/^[ \t]+|[ \t]+$/, "", stats)
if (iface == "lo" || iface == "" || stats == "") next
split(stats, values, /[ \t]+/)
# /proc/net/dev after the colon:
# receive bytes are field 1, transmit bytes are field 9.
# Trim the stats block before split; otherwise leading whitespace can make
# values[1] empty in some awk implementations, resulting in zero rates.
rx += values[1] + 0
tx += values[9] + 0
}
END {printf "%.0f %.0f", rx, tx}
' /proc/net/dev
}
read_disk_bytes() {
read_sectors=0
written_sectors=0
for dev in /sys/block/*; do
[ -r "$dev/stat" ] || continue
name="$(basename "$dev")"
case "$name" in
loop*|ram*|fd*|sr*) continue ;;
esac
# Linux /sys/block/<dev>/stat fields: 3=sectors read, 7=sectors written.
# Use POSIX sh parsing instead of bash arrays so this works on minimal systems.
set -- $(cat "$dev/stat")
sectors_read="${3:-0}"
sectors_written="${7:-0}"
read_sectors=$((read_sectors + sectors_read))
written_sectors=$((written_sectors + sectors_written))
done
printf "%s %s" "$((read_sectors * 512))" "$((written_sectors * 512))"
}
set -- $(read_cpu)
prev_total="${1:-0}"
prev_idle="${2:-0}"
prev_iowait="${3:-0}"
set -- $(read_net_bytes)
prev_rx="${1:-0}"
prev_tx="${2:-0}"
set -- $(read_disk_bytes)
prev_disk_read="${1:-0}"
prev_disk_write="${2:-0}"
prev_ts="$(date +%s)"
sample_count=0
prune_metrics_file() {
[ -f "$OUT" ] || return 0
cutoff="$1"
tmp="${OUT}.$$.tmp"
awk -v cutoff="$cutoff" '
match($0, /"ts":[0-9]+/) {
ts = substr($0, RSTART + 5, RLENGTH - 5)
if (ts >= cutoff) print $0
}
' "$OUT" | tail -n "$MAX_LINES" > "$tmp" && mv "$tmp" "$OUT"
rm -f "$tmp"
}
while true; do
sleep "$INTERVAL"
now_ts="$(date +%s)"
dt=$((now_ts - prev_ts))
if [ "$dt" -le 0 ]; then dt=1; fi
set -- $(read_cpu)
total="${1:-0}"
idle="${2:-0}"
iowait="${3:-0}"
set -- $(read_net_bytes)
rx="${1:-0}"
tx="${2:-0}"
set -- $(read_disk_bytes)
disk_read="${1:-0}"
disk_write="${2:-0}"
mem_pct="$(read_mem_pct)"
total_delta=$((total - prev_total))
idle_delta=$((idle - prev_idle))
iowait_delta=$((iowait - prev_iowait))
rx_delta=$((rx - prev_rx))
tx_delta=$((tx - prev_tx))
disk_read_delta=$((disk_read - prev_disk_read))
disk_write_delta=$((disk_write - prev_disk_write))
cpu_pct="$(awk -v total="$total_delta" -v idle="$idle_delta" 'BEGIN {if (total > 0) printf "%.2f", (total-idle)*100/total; else printf "0"}')"
iowait_pct="$(awk -v total="$total_delta" -v iow="$iowait_delta" 'BEGIN {if (total > 0) printf "%.2f", iow*100/total; else printf "0"}')"
rx_bytes_per_sec="$(awk -v bytes="$rx_delta" -v dt="$dt" 'BEGIN {if (bytes >= 0) printf "%.0f", bytes/dt; else printf "0"}')"
tx_bytes_per_sec="$(awk -v bytes="$tx_delta" -v dt="$dt" 'BEGIN {if (bytes >= 0) printf "%.0f", bytes/dt; else printf "0"}')"
rx_bps="$(awk -v bytes="$rx_delta" -v dt="$dt" 'BEGIN {if (bytes >= 0) printf "%.0f", bytes*8/dt; else printf "0"}')"
tx_bps="$(awk -v bytes="$tx_delta" -v dt="$dt" 'BEGIN {if (bytes >= 0) printf "%.0f", bytes*8/dt; else printf "0"}')"
disk_read_bps="$(awk -v bytes="$disk_read_delta" -v dt="$dt" 'BEGIN {if (bytes >= 0) printf "%.0f", bytes/dt; else printf "0"}')"
disk_write_bps="$(awk -v bytes="$disk_write_delta" -v dt="$dt" 'BEGIN {if (bytes >= 0) printf "%.0f", bytes/dt; else printf "0"}')"
printf '{"ts":%s,"cpu_pct":%s,"iowait_pct":%s,"mem_pct":%s,"net_rx_bytes_per_sec":%s,"net_tx_bytes_per_sec":%s,"net_rx_bps":%s,"net_tx_bps":%s,"disk_read_bps":%s,"disk_write_bps":%s}\n' \
"$now_ts" "$cpu_pct" "$iowait_pct" "$mem_pct" "$rx_bytes_per_sec" "$tx_bytes_per_sec" "$rx_bps" "$tx_bps" "$disk_read_bps" "$disk_write_bps" >> "$OUT"
sample_count=$((sample_count + 1))
if [ $((sample_count % PRUNE_EVERY_SAMPLES)) -eq 0 ]; then
prune_metrics_file "$((now_ts - RETENTION_SECONDS))"
fi
prev_total="$total"
prev_idle="$idle"
prev_iowait="$iowait"
prev_rx="$rx"
prev_tx="$tx"
prev_disk_read="$disk_read"
prev_disk_write="$disk_write"
prev_ts="$now_ts"
done
'''
@dataclass(frozen=True)
class ResourceMonitorPaths:
"""Remote file locations used by the lightweight resource collector."""
metrics_file: str = "/tmp/media_library_viewer_metrics.jsonl"
pid_file: str = "/tmp/media_library_viewer_metrics.pid"
script_file: str = "/tmp/media_library_viewer_metrics_collector.sh"
log_file: str = "/tmp/media_library_viewer_metrics.log"
def start_resource_collector(
ssh: RemoteSSHClient,
interval_seconds: int = 10,
retention_seconds: int = 7 * 24 * 60 * 60,
max_lines: int = 70_000,
paths: ResourceMonitorPaths = ResourceMonitorPaths(),
) -> str:
"""Install and start the remote metrics collector if it is not running.
Starting a fresh collector removes old metrics/log files because schema
changes during development can otherwise leave mixed JSONL records behind.
The collector prunes its own metrics file to 7 days / max_lines.
"""
command = f"""
cat > {shlex.quote(paths.script_file)} <<'MLV_RESOURCE_COLLECTOR'
{COLLECTOR_SCRIPT}
MLV_RESOURCE_COLLECTOR
chmod +x {shlex.quote(paths.script_file)}
if [ -f {shlex.quote(paths.pid_file)} ] && kill -0 "$(cat {shlex.quote(paths.pid_file)})" 2>/dev/null; then
echo "already running pid=$(cat {shlex.quote(paths.pid_file)})"
else
rm -f {shlex.quote(paths.metrics_file)} {shlex.quote(paths.log_file)}
nohup {shlex.quote(paths.script_file)} {shlex.quote(paths.metrics_file)} {int(interval_seconds)} {int(retention_seconds)} {int(max_lines)} >> {shlex.quote(paths.log_file)} 2>&1 &
echo $! > {shlex.quote(paths.pid_file)}
echo "started pid=$(cat {shlex.quote(paths.pid_file)})"
fi
"""
result = ssh.run(command, timeout=20)
if result.exit_status != 0:
raise RuntimeError(result.stderr or result.stdout or "failed to start resource collector")
return result.stdout.strip()
def stop_resource_collector(ssh: RemoteSSHClient, paths: ResourceMonitorPaths = ResourceMonitorPaths()) -> str:
"""Stop the remote collector process if the pid file points to one."""
command = f"""
if [ -f {shlex.quote(paths.pid_file)} ]; then
pid="$(cat {shlex.quote(paths.pid_file)})"
if kill -0 "$pid" 2>/dev/null; then
kill "$pid"
echo "stopped pid=$pid"
else
echo "not running"
fi
rm -f {shlex.quote(paths.pid_file)}
else
echo "not running"
fi
"""
result = ssh.run(command, timeout=20)
if result.exit_status != 0:
raise RuntimeError(result.stderr or result.stdout or "failed to stop resource collector")
return result.stdout.strip()
def restart_resource_collector(
ssh: RemoteSSHClient,
interval_seconds: int = 10,
retention_seconds: int = 7 * 24 * 60 * 60,
max_lines: int = 70_000,
paths: ResourceMonitorPaths = ResourceMonitorPaths(),
) -> str:
stop_message = stop_resource_collector(ssh, paths)
start_message = start_resource_collector(ssh, interval_seconds, retention_seconds, max_lines, paths)
return f"{stop_message}\n{start_message}"
def resource_collector_status(ssh: RemoteSSHClient, paths: ResourceMonitorPaths = ResourceMonitorPaths()) -> str:
"""Return a short human-readable status string for the dashboard."""
command = f"""
if [ -f {shlex.quote(paths.pid_file)} ] && kill -0 "$(cat {shlex.quote(paths.pid_file)})" 2>/dev/null; then
echo "running pid=$(cat {shlex.quote(paths.pid_file)})"
else
echo "not running"
fi
"""
result = ssh.run(command, timeout=10)
if result.exit_status != 0:
raise RuntimeError(result.stderr or result.stdout or "failed to check collector status")
return result.stdout.strip()
def resource_collector_debug_info(ssh: RemoteSSHClient, paths: ResourceMonitorPaths = ResourceMonitorPaths()) -> str:
"""Collect remote diagnostics for troubleshooting missing metrics."""
command = f"""
echo "status:"
if [ -f {shlex.quote(paths.pid_file)} ]; then
pid="$(cat {shlex.quote(paths.pid_file)})"
echo "pid_file=$pid"
if kill -0 "$pid" 2>/dev/null; then echo "process=running"; else echo "process=not-running"; fi
else
echo "pid_file=missing"
fi
echo "files:"
ls -l {shlex.quote(paths.script_file)} {shlex.quote(paths.metrics_file)} {shlex.quote(paths.log_file)} 2>&1 || true
echo "sample_count:"
if [ -f {shlex.quote(paths.metrics_file)} ]; then wc -l < {shlex.quote(paths.metrics_file)}; else echo 0; fi
echo "last_samples:"
if [ -f {shlex.quote(paths.metrics_file)} ]; then tail -n 5 {shlex.quote(paths.metrics_file)}; fi
echo "log_tail:"
if [ -f {shlex.quote(paths.log_file)} ]; then tail -n 40 {shlex.quote(paths.log_file)}; fi
echo "netdev_snapshot:"
cat /proc/net/dev 2>&1 || true
"""
result = ssh.run(command, timeout=20)
return (result.stdout or "") + (result.stderr or "")
def read_resource_metrics(ssh: RemoteSSHClient, max_lines: int = 1000, paths: ResourceMonitorPaths = ResourceMonitorPaths()) -> list[dict[str, Any]]:
"""Read recent JSONL metric samples from the remote collector file."""
command = f"test -f {shlex.quote(paths.metrics_file)} && tail -n {int(max_lines)} {shlex.quote(paths.metrics_file)} || true"
result = ssh.run(command, timeout=20)
if result.exit_status != 0:
raise RuntimeError(result.stderr or result.stdout or "failed to read resource metrics")
rows = []
for line in result.stdout.splitlines():
line = line.strip()
if not line:
continue
try:
rows.append(json.loads(line))
except json.JSONDecodeError:
continue
return rows
def disk_space(ssh: RemoteSSHClient, path: str = "/") -> dict[str, Any]:
"""Return df information for the filesystem containing ``path``."""
command = (
"df -P -B1 -- "
+ shlex.quote(path or "/")
+ " | awk 'NR==2 {printf \"{\\\"filesystem\\\":\\\"%s\\\",\\\"size\\\":%s,\\\"used\\\":%s,\\\"available\\\":%s,\\\"used_pct\\\":\\\"%s\\\",\\\"mount\\\":\\\"%s\\\"}\", $1,$2,$3,$4,$5,$6}'"
)
result = ssh.run(command, timeout=20)
if result.exit_status != 0 or not result.stdout.strip():
raise RuntimeError(result.stderr or result.stdout or "failed to read disk space")
return json.loads(result.stdout)
@@ -0,0 +1,146 @@
"""SSH client helpers for remote filesystem and media inspection.
All command execution goes through ``/bin/sh -c`` and all paths inserted into
commands are shell-quoted by callers. This is important for two reasons:
1. The remote login shell may be fish/csh/etc.; internal commands are POSIX sh.
2. Media paths frequently contain spaces and punctuation.
"""
from __future__ import annotations
import json
import posixpath
import shlex
from dataclasses import dataclass
from typing import Any
import paramiko
@dataclass
class CommandResult:
"""Plain result object returned by remote command execution."""
command: str
exit_status: int
stdout: str
stderr: str
class RemoteSSHClient:
"""SSH helper for read-only inspection plus explicit job execution."""
def __init__(
self,
host: str,
username: str,
port: int = 22,
key_filename: str | None = None,
password: str | None = None,
timeout: int = 20,
):
if not host or not username:
raise ValueError("SSH host and username are required")
self.host = host
self.username = username
self.port = port
self.key_filename = key_filename or None
self.password = password or None
self.timeout = timeout
self._client: paramiko.SSHClient | None = None
def connect(self) -> paramiko.SSHClient:
"""Create or reuse the Paramiko connection.
Unknown host keys are rejected. Users should connect once manually with
ssh so the server is present in known_hosts.
"""
if self._client:
return self._client
client = paramiko.SSHClient()
client.load_system_host_keys()
client.set_missing_host_key_policy(paramiko.RejectPolicy())
client.connect(
self.host,
port=self.port,
username=self.username,
key_filename=self.key_filename,
password=self.password,
timeout=self.timeout,
)
self._client = client
return client
def close(self) -> None:
if self._client:
self._client.close()
self._client = None
def run(self, command: str, timeout: int | None = None) -> CommandResult:
"""Run a command through POSIX sh, independent of the user's login shell.
Paramiko asks the SSH server to execute a command using the account's
default shell. If that shell is fish/csh/etc., POSIX snippets containing
`if ...; then`, pipes, redirects, or heredocs can fail. All internal app
commands and job templates are written for POSIX shell, so explicitly
dispatch through `/bin/sh -c`.
"""
client = self.connect()
shell_command = f"/bin/sh -c {shlex.quote(command)}"
stdin, stdout, stderr = client.exec_command(shell_command, timeout=timeout or self.timeout)
exit_status = stdout.channel.recv_exit_status()
return CommandResult(
command=command,
exit_status=exit_status,
stdout=stdout.read().decode(errors="replace"),
stderr=stderr.read().decode(errors="replace"),
)
def list_dir(self, path: str) -> CommandResult:
"""List one remote directory as JSON.
The command first verifies that ``path`` is a directory. Without that
guard, running ``find`` on a file can look like an empty directory, which
was a source of file-browser confusion. Output is NUL-delimited before
Python serializes it, making spaces in filenames safe.
"""
# JSON-ish output: type, size, mtime epoch, filename. Handles spaces/newlines reasonably via NUL boundaries.
quoted = shlex.quote(path)
not_dir_message = shlex.quote(f"Not a directory: {path}")
command = (
f"test -d {quoted} || "
f"{{ echo {not_dir_message} >&2; exit 20; }}; "
f"find {quoted} -maxdepth 1 -mindepth 1 -printf "
"'%y\\t%s\\t%T@\\t%f\\0' | python3 -c "
+ shlex.quote(
"import sys,json; data=sys.stdin.buffer.read().split(b'\\0'); "
"rows=[]\n"
"for row in data:\n"
" if not row: continue\n"
" t,s,m,n=row.decode('utf-8','replace').split('\\t',3)\n"
" rows.append({'type':t,'size':int(s),'mtime':float(m),'name':n})\n"
"print(json.dumps(rows))"
)
)
return self.run(command)
def stat_path(self, path: str) -> CommandResult:
"""Run stat for a remote file or directory path."""
quoted = shlex.quote(path)
return self.run(f"stat --printf='%F\\n%s bytes\\n%y\\n%n\\n' {quoted}")
def ffprobe_json(self, path: str) -> dict[str, Any]:
"""Run ffprobe and parse JSON output for a remote media file."""
quoted = shlex.quote(path)
result = self.run(
"ffprobe -v error -show_format -show_streams -print_format json " + quoted,
timeout=60,
)
if result.exit_status != 0:
raise RuntimeError(result.stderr or result.stdout or "ffprobe failed")
return json.loads(result.stdout)
@staticmethod
def join(parent: str, child: str) -> str:
return posixpath.normpath(posixpath.join(parent, child))
@@ -0,0 +1,49 @@
"""Configuration loading for the app.
Configuration is intentionally environment/.env based so credentials stay out of
source control and the same package can be reused by different frontends or
process managers.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from pathlib import Path
from dotenv import load_dotenv
load_dotenv()
@dataclass(frozen=True)
class JellyfinConfig:
"""Jellyfin connection settings."""
url: str = os.getenv("JELLYFIN_URL", "")
api_key: str = os.getenv("JELLYFIN_API_KEY", "")
user_id: str = os.getenv("JELLYFIN_USER_ID", "")
@dataclass(frozen=True)
class SSHConfig:
"""SSH connection settings for remote file/resource access."""
host: str = os.getenv("SSH_HOST", "")
username: str = os.getenv("SSH_USERNAME", "")
port: int = int(os.getenv("SSH_PORT", "22"))
key_filename: str = os.getenv("SSH_KEY_FILENAME", str(Path.home() / ".ssh" / "id_rsa"))
password: str = os.getenv("SSH_PASSWORD", "")
media_root: str = os.getenv("REMOTE_MEDIA_ROOT", "")
path_prefix: str = os.getenv("REMOTE_PATH_PREFIX", "")
@dataclass(frozen=True)
class AppConfig:
jellyfin: JellyfinConfig = JellyfinConfig()
ssh: SSHConfig = SSHConfig()
def load_config() -> AppConfig:
"""Build an AppConfig snapshot from the current environment/.env file."""
return AppConfig()
@@ -0,0 +1,5 @@
"""Domain-level helpers and normalization code.
Domain modules convert external data into stable app concepts and should remain
independent of Streamlit or any future frontend framework.
"""
@@ -0,0 +1,161 @@
"""Media-domain normalization helpers.
Jellyfin item JSON is nested and inconsistent across item types. This module
flattens Jellyfin items into stable dictionaries suitable for storage in the
SQLite media index and display by any frontend.
"""
from __future__ import annotations
from typing import Any
import pandas as pd
from media_library_viewer.utils import human_size, ticks_to_minutes
def first_media_source(item: dict[str, Any]) -> dict[str, Any]:
"""Return the first Jellyfin media source, or an empty dict."""
sources = item.get("MediaSources") or []
return sources[0] if sources else {}
def media_streams(item: dict[str, Any], stream_type: str | None = None) -> list[dict[str, Any]]:
"""Return flattened media streams from all media sources.
Jellyfin usually nests streams under MediaSources, while some endpoints may
expose stream-like fields differently. This function gives callers one place
to get streams and optionally filter by type.
"""
streams = []
for source in item.get("MediaSources") or []:
streams.extend(source.get("MediaStreams") or [])
if stream_type is None:
return streams
return [stream for stream in streams if str(stream.get("Type") or stream.get("codec_type") or "").lower() == stream_type.lower()]
def stream_value(stream: dict[str, Any], *keys: str) -> Any:
for key in keys:
if key in stream and stream[key] not in (None, ""):
return stream[key]
return None
def is_hdr_item(item: dict[str, Any]) -> bool:
"""Best-effort HDR detection from Jellyfin video stream metadata."""
hdr_markers = {"hdr", "hdr10", "hdr10+", "dolbyvision", "dovi", "hlg", "pq", "smpte2084", "bt2020"}
for stream in media_streams(item, "Video"):
values = [
stream_value(stream, "VideoRange", "video_range"),
stream_value(stream, "VideoRangeType", "video_range_type"),
stream_value(stream, "ColorTransfer", "color_transfer"),
stream_value(stream, "ColorPrimaries", "color_primaries"),
stream_value(stream, "ColorSpace", "color_space"),
stream_value(stream, "DvVersionMajor", "dv_version_major"),
stream_value(stream, "Hdr10PlusPresent", "hdr10_plus_present"),
]
normalized = " ".join(str(value).lower() for value in values if value not in (None, "", False, 0))
if any(marker in normalized for marker in hdr_markers):
return True
return False
def format_date_added(value: str | None) -> str:
if not value:
return ""
try:
return pd.to_datetime(value).strftime("%Y-%m-%d")
except Exception:
return str(value)
def timestamp_date_added(value: str | None) -> int | None:
if not value:
return None
try:
return int(pd.to_datetime(value).timestamp())
except Exception:
return None
def format_rate_bits_decimal(bits_per_second: float | int | str | None) -> str:
if bits_per_second in (None, ""):
return ""
try:
value = float(bits_per_second)
except (TypeError, ValueError):
return str(bits_per_second)
for unit in ["bps", "Kbps", "Mbps", "Gbps", "Tbps"]:
if value < 1000 or unit == "Tbps":
return f"{value:.1f} {unit}"
value /= 1000
return f"{value:.1f} Tbps"
def normalize_media_item(item: dict[str, Any], library_id: str = "", library_name: str = "") -> dict[str, Any]:
"""Flatten one Jellyfin item into an indexable row.
The returned row contains both display strings (``size``, ``bitrate``) and
numeric sort fields (``size_bytes``, ``bitrate_bps``, ``date_added_ts``).
"""
source = first_media_source(item)
video_streams = media_streams(item, "Video")
video = video_streams[0] if video_streams else {}
size = source.get("Size") or source.get("size")
bitrate = source.get("Bitrate") or source.get("bitrate") or item.get("Bitrate")
width = stream_value(video, "Width", "width")
height = stream_value(video, "Height", "height")
season_number = item.get("ParentIndexNumber")
episode_number = item.get("IndexNumber")
hdr = is_hdr_item(item)
return {
"id": item.get("Id", ""),
"title": item.get("Name", ""),
"series": item.get("SeriesName", ""),
"season": f"S{int(season_number):02d}" if season_number is not None else item.get("SeasonName", ""),
"season_number": int(season_number) if season_number is not None else None,
"episode": int(episode_number) if episode_number is not None else None,
"type": item.get("Type", ""),
"year": item.get("ProductionYear"),
"runtime_ticks": item.get("RunTimeTicks"),
"runtime_min": ticks_to_minutes(item.get("RunTimeTicks")),
"size_bytes": int(size) if size not in (None, "") else None,
"size": human_size(size),
"bitrate_bps": int(bitrate) if bitrate not in (None, "") else None,
"bitrate": format_rate_bits_decimal(bitrate),
"hdr": 1 if hdr else 0,
"hdr_label": "yes" if hdr else "",
"video": video.get("Codec") or video.get("codec_name") or "",
"width": int(width) if width not in (None, "") else None,
"height": int(height) if height not in (None, "") else None,
"resolution": f"{width}x{height}" if width and height else "",
"date_added": format_date_added(item.get("DateCreated")),
"date_added_ts": timestamp_date_added(item.get("DateCreated")),
"path": item.get("Path") or source.get("Path") or "",
"library_id": library_id,
"library_name": library_name,
}
def display_media_row(row: dict[str, Any]) -> dict[str, Any]:
"""Convert a SQLite row back into frontend display fields."""
return {
"title": row.get("title", ""),
"series": row.get("series", ""),
"season": row.get("season", ""),
"episode": row.get("episode", ""),
"type": row.get("type", ""),
"year": row.get("year", ""),
"runtime_min": row.get("runtime_min", ""),
"size": row.get("size") or human_size(row.get("size_bytes")),
"bitrate": row.get("bitrate") or format_rate_bits_decimal(row.get("bitrate_bps")),
"hdr": "yes" if row.get("hdr") else "no",
"video": row.get("video", ""),
"resolution": row.get("resolution", ""),
"date_added": row.get("date_added", ""),
"library": row.get("library_name", ""),
"path": row.get("path", ""),
"id": row.get("id", ""),
}
+59
View File
@@ -0,0 +1,59 @@
"""Template-based remote jobs.
Remote jobs are intentionally explicit templates instead of free-form shell input.
This keeps the UI safer and makes future destructive operations easier to wrap in
confirmations/dry-runs.
"""
from __future__ import annotations
import shlex
from dataclasses import dataclass
from typing import Mapping
from media_library_viewer.clients.ssh import CommandResult, RemoteSSHClient
@dataclass(frozen=True)
class JobTemplate:
"""Description and command template for one remote job."""
name: str
description: str
command_template: str
destructive: bool = False
def render(self, values: Mapping[str, str]) -> str:
"""Render the command with shell-quoted template values.
This is what keeps paths with spaces safe when inserted into job commands.
"""
safe_values = {key: shlex.quote(value) for key, value in values.items()}
return self.command_template.format(**safe_values)
# Phase 1 jobs are intentionally conservative. Add your own templates here later.
JOB_TEMPLATES: dict[str, JobTemplate] = {
"disk_usage": JobTemplate(
name="Disk usage for selected path",
description="Runs du -sh on the selected remote path.",
command_template="du -sh {path}",
),
"ffprobe": JobTemplate(
name="ffprobe JSON",
description="Prints raw ffprobe stream/format metadata.",
command_template="ffprobe -v error -show_format -show_streams -print_format json {path}",
),
"dry_run_find_empty_dirs": JobTemplate(
name="Find empty directories dry-run",
description="Lists empty directories under the selected path. Does not delete anything.",
command_template="find {path} -type d -empty -print",
),
}
def run_job(ssh: RemoteSSHClient, job_key: str, path: str, timeout: int = 600) -> CommandResult:
"""Render and execute a configured job template for a selected remote path."""
template = JOB_TEMPLATES[job_key]
command = template.render({"path": path})
return ssh.run(command, timeout=timeout)
@@ -0,0 +1,5 @@
"""Application services.
Services coordinate clients/domain logic into reusable operations. They are the
natural layer to expose through a future HTTP API for a React frontend.
"""
@@ -0,0 +1,278 @@
"""SQLite-backed media inventory service.
This is the main step toward a frontend-agnostic architecture. The Streamlit UI
asks this service to build/query an index, but the same class could be exposed
through FastAPI to a React frontend without rewriting Jellyfin indexing logic.
"""
from __future__ import annotations
import sqlite3
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable
from media_library_viewer.clients.jellyfin import JellyfinClient
from media_library_viewer.domain.media import display_media_row, normalize_media_item
# Local generated database. It is ignored by git and can be rebuilt from
# Jellyfin metadata whenever needed.
DEFAULT_INDEX_PATH = Path(".cache/media_library_viewer/media_index.sqlite")
MEDIA_TYPES = "Movie,Episode,Video"
# Only values from this whitelist are interpolated into ORDER BY. User-selected
# sort keys map to these known SQL snippets to avoid SQL injection.
SORT_COLUMNS = {
"title": "title COLLATE NOCASE",
"series": "series COLLATE NOCASE",
"season": "season_number",
"episode": "episode",
"type": "type COLLATE NOCASE",
"year": "year",
"runtime": "runtime_min",
"size": "size_bytes",
"bitrate": "bitrate_bps",
"hdr": "hdr",
"video": "video COLLATE NOCASE",
"resolution": "height",
"date_added": "date_added_ts",
"library": "library_name COLLATE NOCASE",
"path": "path COLLATE NOCASE",
}
@dataclass(frozen=True)
class MediaIndexStatus:
"""Lightweight status object displayed by the Media tab."""
exists: bool
item_count: int = 0
updated_at: int | None = None
updated_at_label: str = ""
build_duration_seconds: float | None = None
class MediaIndex:
"""SQLite-backed media inventory.
This class is UI-framework independent. Streamlit, a future FastAPI backend,
or a React-facing API can all use this service.
"""
def __init__(self, db_path: Path | str = DEFAULT_INDEX_PATH):
self.db_path = Path(db_path)
self.db_path.parent.mkdir(parents=True, exist_ok=True)
def connect(self) -> sqlite3.Connection:
"""Open a sqlite connection configured to return Row objects."""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def init_schema(self) -> None:
"""Create tables/indexes if this is the first use of the index."""
with self.connect() as conn:
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS media_items (
id TEXT PRIMARY KEY,
title TEXT,
series TEXT,
season TEXT,
season_number INTEGER,
episode INTEGER,
type TEXT,
year INTEGER,
runtime_ticks INTEGER,
runtime_min INTEGER,
size_bytes INTEGER,
bitrate_bps INTEGER,
hdr INTEGER,
video TEXT,
width INTEGER,
height INTEGER,
resolution TEXT,
date_added TEXT,
date_added_ts INTEGER,
path TEXT,
library_id TEXT,
library_name TEXT
);
CREATE TABLE IF NOT EXISTS index_metadata (
key TEXT PRIMARY KEY,
value TEXT
);
CREATE INDEX IF NOT EXISTS idx_media_type ON media_items(type);
CREATE INDEX IF NOT EXISTS idx_media_library ON media_items(library_id);
CREATE INDEX IF NOT EXISTS idx_media_title ON media_items(title COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_media_series ON media_items(series COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_media_date_added ON media_items(date_added_ts);
CREATE INDEX IF NOT EXISTS idx_media_size ON media_items(size_bytes);
CREATE INDEX IF NOT EXISTS idx_media_bitrate ON media_items(bitrate_bps);
"""
)
def set_metadata(self, key: str, value: str | int | float) -> None:
"""Store a small string metadata value, e.g. build duration."""
self.init_schema()
with self.connect() as conn:
conn.execute(
"INSERT OR REPLACE INTO index_metadata (key, value) VALUES (?, ?)",
(key, str(value)),
)
def replace_items(self, rows: Iterable[dict[str, Any]]) -> int:
"""Atomically replace indexed media rows with a freshly built set."""
self.init_schema()
row_list = list(rows)
columns = [
"id",
"title",
"series",
"season",
"season_number",
"episode",
"type",
"year",
"runtime_ticks",
"runtime_min",
"size_bytes",
"bitrate_bps",
"hdr",
"video",
"width",
"height",
"resolution",
"date_added",
"date_added_ts",
"path",
"library_id",
"library_name",
]
placeholders = ",".join(["?"] * len(columns))
with self.connect() as conn:
conn.execute("DELETE FROM media_items")
conn.executemany(
f"INSERT OR REPLACE INTO media_items ({','.join(columns)}) VALUES ({placeholders})",
[[row.get(column) for column in columns] for row in row_list],
)
conn.execute(
"INSERT OR REPLACE INTO index_metadata (key, value) VALUES ('updated_at', ?)",
(str(int(time.time())),),
)
return len(row_list)
def status(self) -> MediaIndexStatus:
"""Return existence, count, update time, and last build duration."""
if not self.db_path.exists():
return MediaIndexStatus(exists=False)
try:
with self.connect() as conn:
item_count = int(conn.execute("SELECT COUNT(*) FROM media_items").fetchone()[0])
updated_row = conn.execute("SELECT value FROM index_metadata WHERE key='updated_at'").fetchone()
duration_row = conn.execute("SELECT value FROM index_metadata WHERE key='build_duration_seconds'").fetchone()
except sqlite3.Error:
return MediaIndexStatus(exists=False)
updated_at = int(updated_row[0]) if updated_row and str(updated_row[0]).isdigit() else None
label = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(updated_at)) if updated_at else ""
build_duration = None
if duration_row:
try:
build_duration = float(duration_row[0])
except (TypeError, ValueError):
build_duration = None
return MediaIndexStatus(
exists=True,
item_count=item_count,
updated_at=updated_at,
updated_at_label=label,
build_duration_seconds=build_duration,
)
def query(
self,
library_id: str | None = None,
library_ids: list[str] | None = None,
media_types: list[str] | None = None,
search: str = "",
hdr_filter: str = "All",
sort_key: str = "title",
sort_order: str = "Ascending",
limit: int = 100,
offset: int = 0,
) -> tuple[list[dict[str, Any]], int]:
"""Query indexed media with full-index filters, sorting, and pagination."""
self.init_schema()
where = []
params: list[Any] = []
if library_ids:
where.append("library_id IN (" + ",".join(["?"] * len(library_ids)) + ")")
params.extend(library_ids)
elif library_id:
where.append("library_id = ?")
params.append(library_id)
if media_types:
where.append("type IN (" + ",".join(["?"] * len(media_types)) + ")")
params.extend(media_types)
if search:
needle = f"%{search.lower()}%"
where.append("(LOWER(title) LIKE ? OR LOWER(series) LIKE ? OR LOWER(path) LIKE ?)")
params.extend([needle, needle, needle])
if hdr_filter == "HDR only":
where.append("hdr = 1")
elif hdr_filter == "SDR/unknown only":
where.append("(hdr IS NULL OR hdr = 0)")
where_sql = " WHERE " + " AND ".join(where) if where else ""
sort_sql = SORT_COLUMNS.get(sort_key, SORT_COLUMNS["title"])
direction = "DESC" if sort_order == "Descending" else "ASC"
# Always add stable tie-breakers.
order_sql = f" ORDER BY {sort_sql} {direction}, series COLLATE NOCASE ASC, season_number ASC, episode ASC, title COLLATE NOCASE ASC"
with self.connect() as conn:
total = int(conn.execute("SELECT COUNT(*) FROM media_items" + where_sql, params).fetchone()[0])
rows = conn.execute(
"SELECT * FROM media_items" + where_sql + order_sql + " LIMIT ? OFFSET ?",
[*params, int(limit), int(offset)],
).fetchall()
return [display_media_row(dict(row)) for row in rows], total
def build_media_index(
client: JellyfinClient,
user_id: str,
libraries: list[dict[str, Any]],
index: MediaIndex | None = None,
page_size: int = 500,
) -> int:
"""Fetch Jellyfin pages for all selected libraries and rebuild the index."""
index = index or MediaIndex()
started_at = time.perf_counter()
normalized_rows: list[dict[str, Any]] = []
for library in libraries:
library_id = library.get("Id")
library_name = library.get("Name", "")
if not library_id:
continue
start = 0
while True:
response = client.items(
user_id=user_id,
parent_id=library_id,
start_index=start,
limit=page_size,
include_item_types=MEDIA_TYPES,
recursive=True,
sort_by="SortName",
sort_order="Ascending",
)
items = response.get("Items", [])
normalized_rows.extend(normalize_media_item(item, library_id, library_name) for item in items)
start += len(items)
total = int(response.get("TotalRecordCount", start))
if not items or start >= total:
break
count = index.replace_items(normalized_rows)
index.set_metadata("build_duration_seconds", f"{time.perf_counter() - started_at:.3f}")
return count
@@ -0,0 +1,5 @@
"""Streamlit UI modules.
Each module renders one major slice of the application so the main app entrypoint
stays small and future frontend replacement is easier to reason about.
"""
@@ -0,0 +1,343 @@
"""Dashboard and resource-tab UI."""
from __future__ import annotations
import time
from typing import Any
import pandas as pd
import streamlit as st
from media_library_viewer.clients.resources import (
disk_space,
read_resource_metrics,
resource_collector_debug_info,
resource_collector_status,
restart_resource_collector,
start_resource_collector,
stop_resource_collector,
)
from media_library_viewer.utils import human_size
def format_rate_bytes(bytes_per_second: float | int | None) -> str:
if bytes_per_second is None:
return ""
return f"{human_size(bytes_per_second)}/s"
def rate_scale(max_value: float | int | None) -> tuple[float, str]:
value = abs(float(max_value or 0))
units = [(1_000_000_000_000, "TB/s"), (1_000_000_000, "GB/s"), (1_000_000, "MB/s"), (1_000, "KB/s"), (1, "B/s")]
for divisor, suffix in units:
if value >= divisor or divisor == 1:
return float(divisor), suffix
return 1.0, units[-1][1]
def scaled_rate_chart_df(chart_df: pd.DataFrame, columns: list[str], labels: list[str]) -> tuple[pd.DataFrame, str]:
max_value = chart_df[columns].max(numeric_only=True).max()
divisor, suffix = rate_scale(max_value)
scaled = chart_df[columns].copy() / divisor
scaled.columns = [f"{label} ({suffix})" for label in labels]
return scaled, suffix
def format_elapsed(seconds: float | int | None) -> str:
if seconds is None:
return ""
seconds = float(seconds)
if seconds < 60:
return f"{seconds:.1f}s"
minutes = int(seconds // 60)
remainder = seconds % 60
if minutes < 60:
return f"{minutes}m {remainder:.0f}s"
hours = minutes // 60
minutes = minutes % 60
return f"{hours}h {minutes}m"
def render_media_overview(cached_media_counts, cached_library_counts, base_url: str, api_key: str,
user_id: str) -> None:
"""Render dashboard counts for movies/series/episodes and per-library breakdown."""
st.subheader("Media library overview")
try:
counts = cached_media_counts(base_url, api_key, user_id)
except Exception as exc:
st.warning(f"Could not load Jellyfin media counts: {exc}")
return
# Top-level totals
total_items = counts.get("movies", 0) + counts.get("series", 0) + counts.get("episodes", 0)
top_cols = st.columns(4)
top_cols[0].metric("Total items", f"{total_items:,}")
top_cols[1].metric("Movies", f"{counts.get('movies', 0):,}")
top_cols[2].metric("Series", f"{counts.get('series', 0):,}")
top_cols[3].metric("Episodes", f"{counts.get('episodes', 0):,}")
# Per-library breakdown
try:
lib_counts = cached_library_counts(base_url, api_key, user_id)
except Exception as exc:
st.caption(f"Could not load per-library counts: {exc}")
return
if not lib_counts:
return
st.markdown("**Libraries**")
movie_libs = [e for e in lib_counts if e.get("type") == "movies"]
tv_libs = [e for e in lib_counts if e.get("type") == "tvshows"]
if movie_libs and tv_libs:
left_col, right_col = st.columns(2)
elif movie_libs:
left_col = st.container()
right_col = None
elif tv_libs:
left_col = None
right_col = st.container()
else:
return
if movie_libs and left_col:
with left_col:
st.caption("Movie libraries")
for entry in movie_libs:
with st.container(border=True):
st.markdown(f"**{entry['library']}**")
m_cols = st.columns(2)
m_cols[0].metric("Movies", f"{entry['movies']:,}")
if tv_libs and right_col:
with right_col:
st.caption("TV libraries")
for entry in tv_libs:
with st.container(border=True):
st.markdown(f"**{entry['library']}**")
m_cols = st.columns(2)
m_cols[0].metric("Series", f"{entry['series']:,}")
m_cols[1].metric("Episodes", f"{entry['total']:,}")
def render_now_playing(cached_active_sessions, base_url: str, api_key: str) -> None:
"""Render currently playing users/items and transcoding state."""
st.subheader("Now playing")
try:
sessions = cached_active_sessions(base_url, api_key)
except Exception as exc:
st.warning(f"Could not load active sessions: {exc}")
return
if not sessions:
st.caption("No active playback sessions right now.")
return
rows = []
for session in sessions:
item = session.get("NowPlayingItem") or {}
session_id = session.get("Id") or ""
user_name = session.get("UserName") or "Unknown"
device = session.get("DeviceName") or session.get("Client") or ""
play_state = session.get("PlayState") or {}
paused = bool(play_state.get("IsPaused"))
state_label = "paused" if paused else "playing"
series = item.get("SeriesName") or ""
if series:
title = f"{series} - {item.get('Name', '')}"
else:
title = item.get("Name") or "Unknown"
transcoding = session.get("TranscodingInfo") or {}
is_transcoding = bool(transcoding)
transcode_type = []
if is_transcoding:
if transcoding.get("IsVideoDirect") is False:
transcode_type.append("video")
if transcoding.get("IsAudioDirect") is False:
transcode_type.append("audio")
if not transcode_type:
transcode_type.append("active")
rows.append(
{
"user": user_name,
"title": title,
"type": item.get("Type", ""),
"state": state_label,
"transcoding": "yes" if is_transcoding else "no",
"transcoding_type": ", ".join(transcode_type),
"device": device,
"session_id": session_id,
}
)
st.dataframe(pd.DataFrame(rows), use_container_width=True, hide_index=True)
def render_resource_dashboard(get_ssh_client, ssh_args: tuple, media_root: str, detailed: bool = False) -> None:
"""Render server monitoring summary or detailed charts."""
st.subheader("Server monitoring details" if detailed else "Server overview")
host, username, port, key_filename, password = ssh_args
ssh = get_ssh_client(host, username, port, key_filename, password)
try:
status = resource_collector_status(ssh)
except Exception as exc:
st.error(f"Could not check resource collector status: {exc}")
return
if detailed:
control_col, start_col, restart_col, stop_col, refresh_col = st.columns([2.3, 1, 1, 1, 1])
control_col.caption(f"Collector: `{status}` | sample interval: 10s | retention: 7 days / 70k samples")
if start_col.button("Start metrics", key="monitoring_start_metrics", use_container_width=True):
try:
st.success(start_resource_collector(ssh))
except Exception as exc:
st.error(str(exc))
if restart_col.button("Restart", key="monitoring_restart_metrics", use_container_width=True):
try:
st.success(restart_resource_collector(ssh))
except Exception as exc:
st.error(str(exc))
if stop_col.button("Stop metrics", key="monitoring_stop_metrics", use_container_width=True):
try:
st.info(stop_resource_collector(ssh))
except Exception as exc:
st.error(str(exc))
if refresh_col.button("Refresh", key="monitoring_refresh", use_container_width=True):
st.rerun()
else:
st.caption(f"Collector: `{status}`. Open the Monitoring tab for controls, diagnostics, and detailed charts.")
try:
rows = read_resource_metrics(ssh, max_lines=1000)
except Exception as exc:
st.error(f"Could not read resource metrics: {exc}")
rows = []
disk_path = media_root or "/"
try:
space = disk_space(ssh, disk_path)
used_pct_value = float(str(space.get("used_pct", "0")).rstrip("%") or 0)
disk_cols = st.columns(4)
disk_cols[0].metric("Disk used", human_size(space.get("used")))
disk_cols[1].metric("Disk available", human_size(space.get("available")))
disk_cols[2].metric("Disk total", human_size(space.get("size")))
disk_cols[3].metric("Used percent", f"{used_pct_value:.0f}%")
st.progress(min(max(used_pct_value / 100, 0), 1),
text=f"{space.get('mount', disk_path)} on {space.get('filesystem', '')}")
except Exception as exc:
st.warning(f"Could not read disk space for {disk_path}: {exc}")
if not rows:
if detailed:
st.info(
"No monitoring history yet. Click 'Start metrics' and wait at least 10 seconds for the first sample. If this stays empty, use Restart to install the latest collector script.")
with st.expander("Collector diagnostics"):
try:
st.code(resource_collector_debug_info(ssh))
except Exception as exc:
st.error(f"Could not read collector diagnostics: {exc}")
else:
st.info("No monitoring history yet. Open the Monitoring tab to start the collector.")
return
df = pd.DataFrame(rows)
numeric_columns = [
"ts", "cpu_pct", "iowait_pct", "mem_pct", "net_rx_bytes_per_sec", "net_tx_bytes_per_sec", "disk_read_bps",
"disk_write_bps"
]
for column in numeric_columns:
if column in df.columns:
df[column] = pd.to_numeric(df[column], errors="coerce")
df = df.dropna(subset=["ts"])
df["time"] = pd.to_datetime(df["ts"], unit="s", utc=True).dt.tz_convert(None)
cutoff_ts = time.time() - 3600
all_sample_count = len(df)
df = df[df["ts"] >= cutoff_ts]
if df.empty:
st.info("No samples in the last hour yet.")
if detailed:
with st.expander("Resource sample diagnostics"):
if all_sample_count:
raw_df = pd.DataFrame(rows)
st.write(f"Parsed samples: {all_sample_count}")
st.write(
f"Newest remote sample age: {time.time() - float(raw_df['ts'].astype(float).max()):.0f} seconds")
st.dataframe(raw_df.tail(10), use_container_width=True, hide_index=True)
else:
st.write("No parseable samples found.")
return
df = df.sort_values("ts")
latest = df.iloc[-1]
avg_cpu = df["cpu_pct"].mean()
max_cpu = df["cpu_pct"].max()
avg_iowait = df["iowait_pct"].mean() if "iowait_pct" in df.columns else 0
max_iowait = df["iowait_pct"].max() if "iowait_pct" in df.columns else 0
avg_mem = df["mem_pct"].mean()
max_mem = df["mem_pct"].max()
avg_net_down = df["net_rx_bytes_per_sec"].mean()
max_net_down = df["net_rx_bytes_per_sec"].max()
avg_net_up = df["net_tx_bytes_per_sec"].mean()
max_net_up = df["net_tx_bytes_per_sec"].max()
avg_disk_read = df["disk_read_bps"].mean()
max_disk_read = df["disk_read_bps"].max()
avg_disk_write = df["disk_write_bps"].mean()
max_disk_write = df["disk_write_bps"].max()
metric_cols = st.columns(7)
metric_cols[0].metric("CPU now", f"{latest['cpu_pct']:.1f}%")
metric_cols[0].caption(f"avg {avg_cpu:.1f}% \npeak {max_cpu:.1f}%")
metric_cols[1].metric("IO wait", f"{latest.get('iowait_pct', 0):.1f}%")
metric_cols[1].caption(f"avg {avg_iowait:.1f}% \npeak {max_iowait:.1f}%")
metric_cols[2].metric("RAM now", f"{latest['mem_pct']:.1f}%")
metric_cols[2].caption(f"avg {avg_mem:.1f}% \npeak {max_mem:.1f}%")
metric_cols[3].metric("Network down", format_rate_bytes(latest["net_rx_bytes_per_sec"]))
metric_cols[3].caption(f"avg {format_rate_bytes(avg_net_down)} \npeak {format_rate_bytes(max_net_down)}")
metric_cols[4].metric("Network up", format_rate_bytes(latest["net_tx_bytes_per_sec"]))
metric_cols[4].caption(f"avg {format_rate_bytes(avg_net_up)} \npeak {format_rate_bytes(max_net_up)}")
metric_cols[5].metric("Disk read", format_rate_bytes(latest["disk_read_bps"]))
metric_cols[5].caption(f"avg {format_rate_bytes(avg_disk_read)} \npeak {format_rate_bytes(max_disk_read)}")
metric_cols[6].metric("Disk write", format_rate_bytes(latest["disk_write_bps"]))
metric_cols[6].caption(f"avg {format_rate_bytes(avg_disk_write)} \npeak {format_rate_bytes(max_disk_write)}")
chart_df = df.set_index("time")
if detailed:
st.markdown("**CPU, IO wait, and RAM - last hour**")
chart_cols = ["cpu_pct", "mem_pct"]
if "iowait_pct" in chart_df.columns:
chart_cols = ["cpu_pct", "iowait_pct", "mem_pct"]
st.line_chart(chart_df[chart_cols], use_container_width=True)
else:
st.caption(
"Detailed CPU/IO wait/RAM charts, network, disk I/O, raw samples, and collector controls are available in the Monitoring tab.")
return
net_down_df, net_down_suffix = scaled_rate_chart_df(chart_df, ["net_rx_bytes_per_sec"], ["download"])
net_up_df, net_up_suffix = scaled_rate_chart_df(chart_df, ["net_tx_bytes_per_sec"], ["upload"])
net_down_col, net_up_col = st.columns(2)
with net_down_col:
st.markdown(f"**Network down - last hour ({net_down_suffix})**")
st.line_chart(net_down_df, use_container_width=True)
with net_up_col:
st.markdown(f"**Network up - last hour ({net_up_suffix})**")
st.line_chart(net_up_df, use_container_width=True)
disk_read_df, disk_read_suffix = scaled_rate_chart_df(chart_df, ["disk_read_bps"], ["read"])
disk_write_df, disk_write_suffix = scaled_rate_chart_df(chart_df, ["disk_write_bps"], ["write"])
disk_read_col, disk_write_col = st.columns(2)
with disk_read_col:
st.markdown(f"**Disk read - last hour ({disk_read_suffix})**")
st.line_chart(disk_read_df, use_container_width=True)
with disk_write_col:
st.markdown(f"**Disk write - last hour ({disk_write_suffix})**")
st.line_chart(disk_write_df, use_container_width=True)
with st.expander("Raw monitoring samples"):
st.dataframe(df.sort_values("time", ascending=False), use_container_width=True, hide_index=True)
@@ -0,0 +1,261 @@
"""SSH file browser UI."""
from __future__ import annotations
import json
from pathlib import PurePosixPath
from typing import Callable
import pandas as pd
import streamlit as st
from st_aggrid import AgGrid, DataReturnMode, GridOptionsBuilder, GridUpdateMode, JsCode
from media_library_viewer.utils import human_size, timestamp_to_local
FILE_BROWSER_FILTER_KEYS = [
"file_browser_kind_filter",
"file_browser_search",
"file_browser_extension_filter",
"file_browser_sort",
"file_browser_descending",
"file_browser_page",
]
def reset_file_browser_filters() -> None:
"""Clear directory-local filters before opening another folder."""
for key in FILE_BROWSER_FILTER_KEYS:
st.session_state.pop(key, None)
def set_file_browser_path(path: str, selected_path: str | None = None, reset_filters: bool = True) -> None:
"""Set current directory and selected path for the File browser."""
if reset_filters:
st.session_state["file_browser_reset_filters_pending"] = True
st.session_state["file_browser_current_dir"] = path
st.session_state["file_browser_selected_path"] = selected_path or path
st.session_state["file_browser_sync_path_input"] = True
def aggrid_selected_rows(response: dict) -> list[dict]:
"""Return selected rows from a streamlit-aggrid response."""
selected_rows = response.get("selected_rows")
if selected_rows is None:
return []
if isinstance(selected_rows, pd.DataFrame):
return selected_rows.to_dict("records")
return list(selected_rows)
def render_file_browser(cached_dir_listing: Callable[..., list[dict]], ssh_args: tuple, initial_path: str) -> str:
"""Render the File browser and return the selected file/folder path."""
host, username, port, key_filename, password = ssh_args
st.subheader("Remote filesystem")
if "file_browser_current_dir" not in st.session_state:
st.session_state["file_browser_current_dir"] = initial_path or "/"
if "file_browser_selected_path" not in st.session_state:
st.session_state["file_browser_selected_path"] = st.session_state["file_browser_current_dir"]
if st.session_state.pop("file_browser_reset_filters_pending", False):
reset_file_browser_filters()
if "file_browser_path_input" not in st.session_state or st.session_state.pop("file_browser_sync_path_input", False):
st.session_state["file_browser_path_input"] = st.session_state["file_browser_current_dir"]
current_dir = st.session_state["file_browser_current_dir"] or "/"
selected = st.session_state.get("file_browser_selected_path", current_dir)
status_col, selected_col = st.columns([1, 2])
status_col.caption(f"Current folder: `{current_dir}`")
selected_col.caption(f"Selected path: `{selected}`")
path_col, refresh_col = st.columns([6, 1])
path_input = path_col.text_input("Remote path", key="file_browser_path_input", label_visibility="collapsed")
requested_path = path_input or "/"
# Navigate when the user edits the path and presses Enter
if requested_path != current_dir:
set_file_browser_path(requested_path)
st.rerun()
if refresh_col.button("Refresh", key="file_browser_refresh", use_container_width=True):
cached_dir_listing.clear()
st.rerun()
try:
rows = cached_dir_listing(host, username, port, key_filename, password, current_dir)
except Exception as exc:
st.error(f"Could not list directory `{current_dir}`: {exc}")
return st.session_state.get("file_browser_selected_path")
display_rows = []
for row in rows:
kind = "dir" if row["type"] == "d" else "file"
name = row["name"]
extension = PurePosixPath(name).suffix.lower() if kind == "file" else ""
full_path = str(PurePosixPath(current_dir) / name)
display_rows.append(
{
"kind": kind,
"label": "[DIR]" if kind == "dir" else "[FILE]",
"name": name,
"display_name": f"{'[DIR]' if kind == 'dir' else '[FILE]'} {name}",
"extension": extension,
"size_bytes": int(row["size"]),
"size": "-" if kind == "dir" else human_size(row["size"]),
"mtime": float(row["mtime"]),
"modified": timestamp_to_local(row["mtime"]),
"path": full_path,
}
)
total_count = len(display_rows)
dir_count = sum(1 for r in display_rows if r["kind"] == "dir")
file_count = total_count - dir_count
total_file_size = sum(r["size_bytes"] for r in display_rows if r["kind"] == "file")
st.caption(
f"Entries: {total_count} | Directories: {dir_count} | Files: {file_count} | File size: {human_size(total_file_size)}"
)
with st.container(border=True):
filter_col, search_col, ext_col, sort_col, order_col, page_size_col = st.columns([1.1, 2.3, 1.1, 1.2, 1, 1])
kind_filter = filter_col.selectbox("Show", ["All", "Directories", "Files"], key="file_browser_kind_filter", label_visibility="collapsed")
search_term = search_col.text_input("Search", placeholder="Search names", key="file_browser_search", label_visibility="collapsed")
known_exts = sorted({r["extension"] for r in display_rows if r["extension"]})
extension_filter = ext_col.selectbox("Ext", ["All"] + known_exts, key="file_browser_extension_filter", label_visibility="collapsed")
sort_by = sort_col.selectbox("Sort", ["Name", "Kind", "Size", "Modified"], key="file_browser_sort", label_visibility="collapsed")
descending = order_col.toggle("Desc", value=False, key="file_browser_descending")
page_size = page_size_col.selectbox("Rows", [10, 25, 50, 100, 200], index=1, key="file_browser_page_size", label_visibility="collapsed")
filtered_rows = display_rows
if kind_filter == "Directories":
filtered_rows = [r for r in filtered_rows if r["kind"] == "dir"]
elif kind_filter == "Files":
filtered_rows = [r for r in filtered_rows if r["kind"] == "file"]
if search_term:
needle = search_term.lower()
filtered_rows = [r for r in filtered_rows if needle in r["name"].lower()]
if extension_filter != "All":
filtered_rows = [r for r in filtered_rows if r["extension"] == extension_filter]
sort_key_map = {
"Name": lambda r: (r["kind"] != "dir", r["name"].lower()),
"Kind": lambda r: (r["kind"], r["name"].lower()),
"Size": lambda r: (r["kind"] != "dir", r["size_bytes"]),
"Modified": lambda r: r["mtime"],
}
filtered_rows.sort(key=sort_key_map[sort_by], reverse=descending)
filtered_count = len(filtered_rows)
page_count = max(1, (filtered_count + page_size - 1) // page_size)
page_col, summary_col = st.columns([1, 5])
if st.session_state.get("file_browser_page", 1) > page_count:
st.session_state["file_browser_page"] = page_count
page_number = page_col.number_input("Page", min_value=1, max_value=page_count, value=1, step=1, key="file_browser_page")
start = (int(page_number) - 1) * page_size
end = start + page_size
page_rows = filtered_rows[start:end]
summary_col.caption(f"Showing {start + 1 if filtered_count else 0}-{min(end, filtered_count)} of {filtered_count} matching entries")
parent_path = str(PurePosixPath(current_dir).parent)
visible_rows = []
if current_dir != "/":
visible_rows.append(
{
"kind": "up",
"label": "[UP]",
"name": "..",
"display_name": "[UP] ..",
"extension": "",
"size_bytes": 0,
"size": "-",
"mtime": 0.0,
"modified": "",
"path": parent_path,
}
)
visible_rows.extend(page_rows)
if not display_rows:
st.info("Directory is empty.")
elif not page_rows:
st.info("No entries match the current filters.")
if not visible_rows:
with st.expander("File browser diagnostics"):
st.write(f"Current folder: `{current_dir}`")
st.write(f"Path input: `{requested_path}`")
st.write(f"Selected path: `{selected}`")
st.write(f"Raw entries returned by remote listing: {len(rows)}")
return selected
table_rows = [
{
"type": row["kind"],
"name": row["name"],
"ext": row["extension"],
"size": row["size"],
"modified": row["modified"],
"path": row["path"],
}
for row in visible_rows
]
table_df = pd.DataFrame(table_rows)
grid_builder = GridOptionsBuilder.from_dataframe(table_df)
grid_builder.configure_default_column(editable=False, resizable=True, sortable=False, filter=False)
grid_builder.configure_column("type", width=90)
grid_builder.configure_column("name", flex=2)
grid_builder.configure_column("ext", width=90)
grid_builder.configure_column("size", width=120)
grid_builder.configure_column("modified", width=180)
grid_builder.configure_column("path", hide=True)
grid_builder.configure_selection(selection_mode="single", use_checkbox=False)
grid_options = grid_builder.build()
grid_options["rowSelection"] = {
"mode": "singleRow",
"checkboxes": False,
"headerCheckbox": False,
"enableClickSelection": True,
}
grid_options["suppressRowClickSelection"] = False
grid_options["suppressCellFocus"] = True
grid_options["onCellClicked"] = JsCode(
"""
function(event) {
if (event && event.node) {
event.node.setSelected(true, true);
}
}
"""
)
grid_response = AgGrid(
table_df,
gridOptions=grid_options,
height=min(360, 33 * (len(table_rows) + 1)),
fit_columns_on_grid_load=True,
data_return_mode=DataReturnMode.AS_INPUT,
update_mode=GridUpdateMode.SELECTION_CHANGED,
key="file_browser_grid",
theme="streamlit",
allow_unsafe_jscode=True,
)
selected_rows = aggrid_selected_rows(grid_response)
if selected_rows:
picked_row = selected_rows[0]
picked_path = picked_row.get("path")
picked_type = picked_row.get("type")
action_token = f"{picked_type}:{picked_path}"
# Open directories immediately on row select, including the [UP] row.
if picked_type in {"dir", "up"} and picked_path and picked_path != current_dir:
if st.session_state.get("file_browser_last_row_action") != action_token:
st.session_state["file_browser_last_row_action"] = action_token
set_file_browser_path(picked_path)
st.rerun()
elif picked_path:
st.session_state["file_browser_selected_path"] = picked_path
st.session_state["file_browser_last_row_action"] = action_token
st.caption("Select a directory row to open it (including [UP] ..). Select a file row to target metadata/jobs.")
return st.session_state.get("file_browser_selected_path", current_dir)
@@ -0,0 +1,215 @@
"""Media index tab UI."""
from __future__ import annotations
from pathlib import PurePosixPath
from typing import Any, Callable
import pandas as pd
import streamlit as st
from st_aggrid import AgGrid, DataReturnMode, GridOptionsBuilder, GridUpdateMode, JsCode
from media_library_viewer.services.media_index import MediaIndex, build_media_index
def aggrid_selected_rows(response: dict[str, Any]) -> list[dict[str, Any]]:
"""Return selected rows from a streamlit-aggrid response."""
selected_rows = response.get("selected_rows")
if selected_rows is None:
return []
if isinstance(selected_rows, pd.DataFrame):
return selected_rows.to_dict("records")
return list(selected_rows)
def format_elapsed(seconds: float | int | None) -> str:
if seconds is None:
return ""
seconds = float(seconds)
if seconds < 60:
return f"{seconds:.1f}s"
minutes = int(seconds // 60)
remainder = seconds % 60
if minutes < 60:
return f"{minutes}m {remainder:.0f}s"
hours = minutes // 60
minutes = minutes % 60
return f"{hours}h {minutes}m"
def render_media_tab(
client,
user_id: str,
libraries: list[dict[str, Any]],
set_file_browser_path: Callable[[str, str | None, bool], None],
) -> None:
"""Render the SQLite-backed media inventory tab."""
st.subheader("Media inventory")
st.caption("SQLite-backed index for full-library sorting/filtering. File size/bitrate/HDR are based on Jellyfin media source metadata, not a full ffprobe scan.")
index = MediaIndex()
status = index.status()
status_col, build_col, refresh_col = st.columns([3.5, 1.2, 1])
if status.exists:
status_parts = [f"Index: {status.item_count:,} items"]
if status.updated_at_label:
status_parts.append(f"updated {status.updated_at_label}")
if status.build_duration_seconds is not None:
status_parts.append(f"last build took {format_elapsed(status.build_duration_seconds)}")
status_col.caption(" | ".join(status_parts))
else:
status_col.warning("No local media index yet. Build it to enable full-library sorting and filtering.")
if build_col.button("Build index", key="media_index_build", use_container_width=True):
with st.spinner("Building media index from Jellyfin. This can take a while for large libraries..."):
count = build_media_index(client, user_id, libraries, index)
st.success(f"Indexed {count:,} media items.")
st.rerun()
if refresh_col.button("Refresh", key="media_index_refresh", use_container_width=True):
st.rerun()
if not index.status().exists:
st.info("The Media tab uses a local SQLite index so sorting by size, bitrate, HDR, codec, season, and episode works across the whole library rather than just the current Jellyfin page.")
return
library_options = {lib["Name"]: lib["Id"] for lib in libraries}
filter_col, type_col, search_col, page_size_col, page_col = st.columns([1.9, 1.8, 2.4, 1.1, 1])
selected_libraries = filter_col.multiselect(
"Libraries",
list(library_options.keys()),
default=list(library_options.keys()),
key="media_inventory_libraries",
)
media_types = type_col.multiselect(
"Types",
["Movie", "Episode", "Video"],
default=["Movie", "Episode"],
key="media_inventory_types",
)
search = search_col.text_input("Search", key="media_inventory_search")
page_size = page_size_col.selectbox("Rows", [50, 100, 250, 500], index=1, key="media_inventory_page_size")
page = page_col.number_input("Page", min_value=1, value=1, step=1, key="media_inventory_page")
sort_options = {
"Title": "title",
"Series": "series",
"Season": "season",
"Episode": "episode",
"Type": "type",
"Year": "year",
"Runtime": "runtime",
"Size": "size",
"Bitrate": "bitrate",
"HDR": "hdr",
"Video codec": "video",
"Resolution": "resolution",
"Date added": "date_added",
"Library": "library",
"Path": "path",
}
sort_col, order_col, hdr_col = st.columns([1.4, 1.1, 1.2])
sort_label = sort_col.selectbox("Sort", list(sort_options.keys()), key="media_inventory_sort")
sort_order_label = order_col.selectbox("Order", ["Ascending", "Descending"], key="media_inventory_sort_order")
hdr_filter = hdr_col.selectbox("HDR filter", ["All", "HDR only", "SDR/unknown only"], key="media_inventory_hdr_filter")
if not selected_libraries:
st.info("Select at least one library to show indexed media.")
return
rows, total = index.query(
library_ids=[library_options[name] for name in selected_libraries],
media_types=media_types or ["Movie", "Episode", "Video"],
search=search,
hdr_filter=hdr_filter,
sort_key=sort_options[sort_label],
sort_order=sort_order_label,
limit=int(page_size),
offset=(int(page) - 1) * int(page_size),
)
st.caption(f"Showing {len(rows)} of {total:,} indexed matching items. Sort and filters apply to the full local index.")
columns = [
"title", "series", "season", "episode", "type", "year", "runtime_min",
"size", "bitrate", "hdr", "video", "resolution", "date_added", "library", "path", "id",
]
if not rows:
st.info("No media found for the current filters.")
return
table_df = pd.DataFrame(rows)[columns].fillna("")
selected_media_path = st.session_state.get("media_inventory_selected_path")
grid_builder = GridOptionsBuilder.from_dataframe(table_df)
grid_builder.configure_default_column(editable=False, resizable=True, sortable=False, filter=False, autoSize=True)
grid_builder.configure_column("title", header_name="Title", minWidth=150)
grid_builder.configure_column("series", header_name="Series", minWidth=120)
grid_builder.configure_column("season", header_name="Season", maxWidth=95)
grid_builder.configure_column("episode", header_name="Episode", maxWidth=105)
grid_builder.configure_column("type", header_name="Type", maxWidth=100)
grid_builder.configure_column("year", header_name="Year", maxWidth=90)
grid_builder.configure_column("runtime_min", header_name="Runtime (min)", maxWidth=125)
grid_builder.configure_column("size", header_name="Size", maxWidth=120)
grid_builder.configure_column("bitrate", header_name="Bitrate", maxWidth=125)
grid_builder.configure_column("hdr", header_name="HDR", maxWidth=80)
grid_builder.configure_column("video", header_name="Video codec", maxWidth=120)
grid_builder.configure_column("resolution", header_name="Resolution", maxWidth=120)
grid_builder.configure_column("date_added", header_name="Date added", maxWidth=120)
grid_builder.configure_column("library", header_name="Library", maxWidth=140)
grid_builder.configure_column("path", header_name="Path", minWidth=200)
grid_builder.configure_column("id", hide=True)
grid_builder.configure_selection(selection_mode="single", use_checkbox=False)
grid_options = grid_builder.build()
grid_options["autoSizeStrategy"] = {"type": "fitCellContents"}
grid_options["rowSelection"] = {
"mode": "singleRow",
"checkboxes": False,
"headerCheckbox": False,
"enableClickSelection": True,
}
grid_options["suppressRowClickSelection"] = False
grid_options["suppressCellFocus"] = True
grid_options["onCellClicked"] = JsCode(
"""
function(event) {
if (event && event.node) {
event.node.setSelected(true, true);
}
}
"""
)
grid_response = AgGrid(
table_df,
gridOptions=grid_options,
height=min(650, 35 * (len(rows) + 1)),
fit_columns_on_grid_load=True,
data_return_mode=DataReturnMode.AS_INPUT,
update_mode=GridUpdateMode.SELECTION_CHANGED,
key="media_inventory_grid",
theme="streamlit",
allow_unsafe_jscode=True,
)
selected_rows = aggrid_selected_rows(grid_response)
if selected_rows:
selected_media_path = selected_rows[0].get("path")
if selected_media_path:
st.session_state["media_inventory_selected_path"] = selected_media_path
# Auto-sync File browser location from Media row selection.
# Guarded by last-synced path to avoid reapplying on every rerun.
last_synced = st.session_state.get("media_inventory_last_synced_path")
if selected_media_path != last_synced:
set_file_browser_path(str(PurePosixPath(selected_media_path).parent), selected_media_path)
st.session_state["media_inventory_last_synced_path"] = selected_media_path
if selected_media_path:
st.caption(f"Selected media path: `{selected_media_path}` (File browser folder synced automatically)")
else:
st.caption("Select a table row to automatically sync its containing folder to the File browser tab.")
with st.expander("Notes"):
st.write(
"The Media tab now queries a local SQLite index, so sorting/filtering applies across the indexed library. "
"Rebuild the index after Jellyfin scans or metadata changes. Full ffprobe enrichment for every item can be added later as a background index extension."
)
@@ -0,0 +1,129 @@
"""Selected-file preview and remote path tools UI."""
from __future__ import annotations
from typing import Any, Callable
import pandas as pd
import streamlit as st
from media_library_viewer.jobs import JOB_TEMPLATES, run_job
from media_library_viewer.utils import (
ffprobe_format_summary,
is_known_video_file,
summarize_audio_streams,
summarize_streams,
summarize_subtitle_streams,
summarize_video_streams,
)
def render_ffprobe_sections(ffprobe_data: dict[str, Any]) -> None:
"""Render ffprobe output in separate container/video/audio/subtitle sections."""
format_summary = ffprobe_format_summary(ffprobe_data)
video_rows = summarize_video_streams(ffprobe_data)
audio_rows = summarize_audio_streams(ffprobe_data)
subtitle_rows = summarize_subtitle_streams(ffprobe_data)
st.markdown("**Container**")
st.dataframe(pd.DataFrame([format_summary]), use_container_width=True, hide_index=True)
st.markdown("**Video**")
if video_rows:
st.dataframe(pd.DataFrame(video_rows), use_container_width=True, hide_index=True)
else:
st.caption("No video streams found.")
st.markdown("**Audio**")
if audio_rows:
st.dataframe(pd.DataFrame(audio_rows), use_container_width=True, hide_index=True)
else:
st.caption("No audio streams found.")
st.markdown("**Subtitles**")
if subtitle_rows:
st.dataframe(pd.DataFrame(subtitle_rows), use_container_width=True, hide_index=True)
else:
st.caption("No subtitle streams found.")
def render_selected_file_preview(
ssh_args: tuple,
selected_path: str | None,
cached_ffprobe_preview: Callable[..., dict[str, Any]],
) -> None:
"""Run and render a blocking ffprobe preview for selected known video files."""
with st.container(border=True):
st.markdown("**Selected file preview**")
if not selected_path:
st.caption("Select a file to preview media metadata.")
return
st.caption(f"Path: `{selected_path}`")
if not is_known_video_file(selected_path):
st.caption("Automatic ffprobe preview runs for known video file extensions only.")
return
refresh_col, status_col = st.columns([1.2, 5])
if refresh_col.button("Reload preview", key="preview_reload", use_container_width=True):
cached_ffprobe_preview.clear()
st.rerun()
host, username, port, key_filename, password = ssh_args
try:
with st.spinner("Running ffprobe preview..."):
ffprobe_data = cached_ffprobe_preview(host, username, port, key_filename, password, selected_path)
except Exception as exc:
status_col.error(f"ffprobe failed: {exc}")
return
status_col.success("ffprobe preview loaded.")
render_ffprobe_sections(ffprobe_data)
with st.expander("Raw ffprobe JSON"):
st.json(ffprobe_data)
def render_ssh_tools(
ssh,
ssh_args: tuple,
selected_path: str | None,
cached_ffprobe_preview: Callable[..., dict[str, Any]],
) -> None:
"""Render selected-path diagnostics and safe job templates."""
render_selected_file_preview(ssh_args, selected_path, cached_ffprobe_preview)
if not selected_path:
return
st.subheader("Disk metadata and jobs")
tabs = st.tabs(["ffprobe", "stat", "jobs"])
with tabs[0]:
if st.button("Run ffprobe on selected path", key="tools_run_ffprobe"):
try:
data = ssh.ffprobe_json(selected_path)
render_ffprobe_sections(data)
with st.expander("All streams table"):
st.dataframe(pd.DataFrame(summarize_streams(data)), use_container_width=True, hide_index=True)
with st.expander("Raw ffprobe JSON"):
st.json(data)
except Exception as exc:
st.error(str(exc))
with tabs[1]:
if st.button("Run stat", key="tools_run_stat"):
result = ssh.stat_path(selected_path)
st.code(result.stdout or result.stderr)
with tabs[2]:
st.warning("Jobs run commands on the remote server. Phase 1 includes safe/read-only templates only.")
job_key = st.selectbox("Job", list(JOB_TEMPLATES.keys()), format_func=lambda k: JOB_TEMPLATES[k].name)
st.caption(JOB_TEMPLATES[job_key].description)
command_preview = JOB_TEMPLATES[job_key].render({"path": selected_path})
st.code(command_preview, language="bash")
if st.button("Run selected job", key="tools_run_selected_job"):
result = run_job(ssh, job_key, selected_path)
st.write(f"Exit status: `{result.exit_status}`")
if result.stdout:
st.code(result.stdout)
if result.stderr:
st.error(result.stderr)
+227
View File
@@ -0,0 +1,227 @@
"""Formatting and ffprobe summarization helpers.
These helpers are intentionally UI-framework independent. Streamlit renders the
returned dictionaries/dataframes, but another frontend can reuse the same
summaries.
"""
from __future__ import annotations
from datetime import datetime
from pathlib import PurePosixPath
from typing import Any
VIDEO_FILE_EXTENSIONS = {
".3g2",
".3gp",
".avi",
".divx",
".flv",
".m2ts",
".m4v",
".mkv",
".mov",
".mp4",
".mpeg",
".mpg",
".mts",
".ogm",
".ogv",
".rmvb",
".ts",
".vob",
".webm",
".wmv",
}
def ticks_to_minutes(ticks: int | None) -> int | None:
"""Convert Jellyfin/Emby 100-nanosecond ticks to rounded minutes."""
if not ticks:
return None
return round(ticks / 10_000_000 / 60)
def human_size(num: int | float | None) -> str:
"""Format a byte count as B/KB/MB/GB/etc."""
if num is None:
return ""
value = float(num)
for unit in ["B", "KB", "MB", "GB", "TB", "PB"]:
if value < 1024 or unit == "PB":
return f"{value:.1f} {unit}" if unit != "B" else f"{int(value)} B"
value /= 1024
return f"{value:.1f} PB"
def timestamp_to_local(ts: float | None) -> str:
if ts is None:
return ""
return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")
def is_known_video_file(path: str | None) -> bool:
"""Return True when a path extension is one we should ffprobe automatically."""
if not path:
return False
return PurePosixPath(path).suffix.lower() in VIDEO_FILE_EXTENSIONS
def format_duration(seconds: str | int | float | None) -> str:
if seconds in (None, ""):
return ""
try:
total = float(seconds)
except (TypeError, ValueError):
return str(seconds)
hours = int(total // 3600)
minutes = int((total % 3600) // 60)
secs = int(total % 60)
return f"{hours:02d}:{minutes:02d}:{secs:02d}"
def format_bitrate(bit_rate: str | int | float | None) -> str:
if bit_rate in (None, ""):
return ""
try:
value = float(bit_rate)
except (TypeError, ValueError):
return str(bit_rate)
if value >= 1_000_000:
return f"{value / 1_000_000:.2f} Mbps"
if value >= 1_000:
return f"{value / 1_000:.0f} kbps"
return f"{value:.0f} bps"
def _tags(stream: dict[str, Any]) -> dict[str, Any]:
return stream.get("tags") or {}
def _disposition(stream: dict[str, Any], key: str) -> str:
value = (stream.get("disposition") or {}).get(key)
return "yes" if value == 1 else ""
def _side_data_types(stream: dict[str, Any]) -> str:
values = []
for item in stream.get("side_data_list") or []:
if item.get("side_data_type"):
values.append(item["side_data_type"])
return ", ".join(values)
def ffprobe_format_summary(ffprobe: dict[str, Any]) -> dict[str, str]:
"""Summarize ffprobe container/format-level metadata."""
fmt = ffprobe.get("format") or {}
return {
"filename": fmt.get("filename", ""),
"format": fmt.get("format_name", ""),
"format_long": fmt.get("format_long_name", ""),
"duration": format_duration(fmt.get("duration")),
"size": human_size(float(fmt["size"])) if fmt.get("size") else "",
"bit_rate": format_bitrate(fmt.get("bit_rate")),
"stream_count": str(fmt.get("nb_streams", "")),
}
def summarize_video_streams(ffprobe: dict[str, Any]) -> list[dict[str, Any]]:
"""Return video-only stream rows with video/HDR-related fields."""
rows = []
for stream in ffprobe.get("streams", []):
if stream.get("codec_type") != "video":
continue
tags = _tags(stream)
rows.append(
{
"index": stream.get("index"),
"codec": stream.get("codec_name"),
"profile": stream.get("profile"),
"resolution": f"{stream.get('width', '')}x{stream.get('height', '')}",
"pix_fmt": stream.get("pix_fmt"),
"bit_rate": format_bitrate(stream.get("bit_rate")),
"avg_fps": stream.get("avg_frame_rate"),
"color_range": stream.get("color_range"),
"color_space": stream.get("color_space"),
"color_transfer": stream.get("color_transfer"),
"color_primaries": stream.get("color_primaries"),
"side_data": _side_data_types(stream),
"language": tags.get("language"),
"title": tags.get("title"),
"default": _disposition(stream, "default"),
}
)
return rows
def summarize_audio_streams(ffprobe: dict[str, Any]) -> list[dict[str, Any]]:
"""Return audio-only stream rows with channel/language/default fields."""
rows = []
for stream in ffprobe.get("streams", []):
if stream.get("codec_type") != "audio":
continue
tags = _tags(stream)
rows.append(
{
"index": stream.get("index"),
"codec": stream.get("codec_name"),
"profile": stream.get("profile"),
"channels": stream.get("channels"),
"layout": stream.get("channel_layout"),
"sample_rate": stream.get("sample_rate"),
"bit_rate": format_bitrate(stream.get("bit_rate")),
"language": tags.get("language"),
"title": tags.get("title"),
"default": _disposition(stream, "default"),
"forced": _disposition(stream, "forced"),
}
)
return rows
def summarize_subtitle_streams(ffprobe: dict[str, Any]) -> list[dict[str, Any]]:
"""Return subtitle-only stream rows with language/forced/default fields."""
rows = []
for stream in ffprobe.get("streams", []):
if stream.get("codec_type") != "subtitle":
continue
tags = _tags(stream)
rows.append(
{
"index": stream.get("index"),
"codec": stream.get("codec_name"),
"codec_long": stream.get("codec_long_name"),
"language": tags.get("language"),
"title": tags.get("title"),
"default": _disposition(stream, "default"),
"forced": _disposition(stream, "forced"),
"hearing_impaired": _disposition(stream, "hearing_impaired"),
}
)
return rows
def summarize_streams(ffprobe: dict[str, Any]) -> list[dict[str, Any]]:
rows = []
for stream in ffprobe.get("streams", []):
rows.append(
{
"index": stream.get("index"),
"type": stream.get("codec_type"),
"codec": stream.get("codec_name"),
"profile": stream.get("profile"),
"width": stream.get("width"),
"height": stream.get("height"),
"pix_fmt": stream.get("pix_fmt"),
"color_transfer": stream.get("color_transfer"),
"color_primaries": stream.get("color_primaries"),
"color_space": stream.get("color_space"),
"bit_rate": format_bitrate(stream.get("bit_rate")),
"channels": stream.get("channels"),
"sample_rate": stream.get("sample_rate"),
"language": stream.get("tags", {}).get("language"),
"title": stream.get("tags", {}).get("title"),
}
)
return rows
View File