Simon Gardling
b9cd380a22
All checks were successful
Build and Deploy / deploy (push) Successful in 2m40s
143 lines
4.2 KiB
Python
143 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
JELLYFIN_URL = os.environ.get("JELLYFIN_URL", "http://127.0.0.1:8096")
|
|
GRAFANA_URL = os.environ.get("GRAFANA_URL", "http://127.0.0.1:3000")
|
|
STATE_FILE = os.environ.get("STATE_FILE", "/var/lib/jellyfin-annotations/state.json")
|
|
POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL", "30"))
|
|
|
|
|
|
def get_api_key():
|
|
cred_dir = os.environ.get("CREDENTIALS_DIRECTORY")
|
|
if cred_dir:
|
|
return Path(cred_dir, "jellyfin-api-key").read_text().strip()
|
|
for p in ["/run/agenix/jellyfin-api-key"]:
|
|
if Path(p).exists():
|
|
return Path(p).read_text().strip()
|
|
sys.exit("ERROR: Cannot find jellyfin-api-key")
|
|
|
|
|
|
def http_json(method, url, body=None):
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=data,
|
|
headers={"Content-Type": "application/json", "Accept": "application/json"},
|
|
method=method,
|
|
)
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
return json.loads(resp.read())
|
|
|
|
|
|
def get_active_sessions(api_key):
|
|
try:
|
|
req = urllib.request.Request(
|
|
f"{JELLYFIN_URL}/Sessions?api_key={api_key}",
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
sessions = json.loads(resp.read())
|
|
return [s for s in sessions if s.get("NowPlayingItem")]
|
|
except Exception as e:
|
|
print(f"Error fetching sessions: {e}", file=sys.stderr)
|
|
return None
|
|
|
|
|
|
def format_label(session):
|
|
user = session.get("UserName", "Unknown")
|
|
item = session.get("NowPlayingItem", {}) or {}
|
|
name = item.get("Name", "Unknown")
|
|
series = item.get("SeriesName", "")
|
|
season = item.get("ParentIndexNumber")
|
|
episode = item.get("IndexNumber")
|
|
media_type = item.get("Type", "Unknown")
|
|
|
|
if series and season and episode:
|
|
return f"{user}: {series} S{season:02d}E{episode:02d} - {name}"
|
|
elif series:
|
|
return f"{user}: {series} - {name}"
|
|
elif media_type == "Movie":
|
|
return f"{user}: {name} (movie)"
|
|
return f"{user}: {name}"
|
|
|
|
|
|
def load_state():
|
|
try:
|
|
with open(STATE_FILE) as f:
|
|
return json.load(f)
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
|
|
def save_state(state):
|
|
os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
|
|
tmp = STATE_FILE + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
json.dump(state, f)
|
|
os.replace(tmp, STATE_FILE)
|
|
|
|
|
|
def grafana_post(label, start_ms):
|
|
try:
|
|
result = http_json(
|
|
"POST",
|
|
f"{GRAFANA_URL}/api/annotations",
|
|
{"time": start_ms, "text": label, "tags": ["jellyfin"]},
|
|
)
|
|
return result.get("id")
|
|
except Exception as e:
|
|
print(f"Error posting annotation: {e}", file=sys.stderr)
|
|
return None
|
|
|
|
|
|
def grafana_close(grafana_id, end_ms):
|
|
try:
|
|
http_json(
|
|
"PATCH",
|
|
f"{GRAFANA_URL}/api/annotations/{grafana_id}",
|
|
{"timeEnd": end_ms},
|
|
)
|
|
except Exception as e:
|
|
print(f"Error closing annotation {grafana_id}: {e}", file=sys.stderr)
|
|
|
|
|
|
def main():
|
|
api_key = get_api_key()
|
|
state = load_state()
|
|
|
|
while True:
|
|
now_ms = int(time.time() * 1000)
|
|
sessions = get_active_sessions(api_key)
|
|
|
|
if sessions is not None:
|
|
current_ids = {s["Id"] for s in sessions}
|
|
|
|
for s in sessions:
|
|
sid = s["Id"]
|
|
if sid not in state:
|
|
label = format_label(s)
|
|
grafana_id = grafana_post(label, now_ms)
|
|
if grafana_id is not None:
|
|
state[sid] = {
|
|
"grafana_id": grafana_id,
|
|
"label": label,
|
|
"start_ms": now_ms,
|
|
}
|
|
save_state(state)
|
|
|
|
for sid in [k for k in state if k not in current_ids]:
|
|
info = state.pop(sid)
|
|
grafana_close(info["grafana_id"], now_ms)
|
|
save_state(state)
|
|
|
|
time.sleep(POLL_INTERVAL)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|