72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import requests
|
||
|
|
from datetime import datetime, date, timedelta
|
||
|
|
|
||
|
|
SECRET_DIR = "/home/torjus/.config/sops-nix/secrets"
|
||
|
|
|
||
|
|
def sonarr_url():
|
||
|
|
xdg_dir = os.environ["XDG_RUNTIME_DIR"]
|
||
|
|
if not xdg_dir:
|
||
|
|
raise Exception("XDG_RUNTIME_DIR not set")
|
||
|
|
with open(f"{SECRET_DIR}/sonarr_base_url") as f:
|
||
|
|
return f.read().strip()
|
||
|
|
|
||
|
|
def radarr_url():
|
||
|
|
with open(f"{SECRET_DIR}/radarr_base_url") as f:
|
||
|
|
return f.read().strip()
|
||
|
|
|
||
|
|
def make_header(api_key: str):
|
||
|
|
return {"X-Api-Key": api_key, "Accept": "application/json"}
|
||
|
|
|
||
|
|
def get_sonarr_key():
|
||
|
|
with open(f"{SECRET_DIR}/sonarr_api_key") as f:
|
||
|
|
return f.read().strip()
|
||
|
|
|
||
|
|
def get_radarr_key():
|
||
|
|
with open(f"{SECRET_DIR}/radarr_api_key") as f:
|
||
|
|
return f.read().strip()
|
||
|
|
|
||
|
|
def get_sonarr_history(since: datetime|None=None):
|
||
|
|
api_key = get_sonarr_key()
|
||
|
|
if not since:
|
||
|
|
since = datetime.combine(date.today()-timedelta(days=1),datetime.min.time())
|
||
|
|
url = f"{sonarr_url()}/api/history/since"
|
||
|
|
url += f"?date={since.isoformat()}"
|
||
|
|
response = requests.get(url, headers=make_header(api_key))
|
||
|
|
response.raise_for_status()
|
||
|
|
data = response.json()
|
||
|
|
|
||
|
|
items = []
|
||
|
|
for item in data:
|
||
|
|
if item["eventType"] == "downloadFolderImported":
|
||
|
|
items.append(item["sourceTitle"])
|
||
|
|
return items
|
||
|
|
|
||
|
|
def get_radarr_history(since: datetime|None=None):
|
||
|
|
api_key = get_radarr_key()
|
||
|
|
if not since:
|
||
|
|
since = datetime.combine(date.today()-timedelta(days=7),datetime.min.time())
|
||
|
|
url = f"{radarr_url()}/api/v3/history/since"
|
||
|
|
url += f"?date={since.isoformat()}"
|
||
|
|
response = requests.get(url, headers=make_header(api_key))
|
||
|
|
response.raise_for_status()
|
||
|
|
data = response.json()
|
||
|
|
|
||
|
|
items = []
|
||
|
|
for item in data:
|
||
|
|
if item["eventType"] == "downloadFolderImported":
|
||
|
|
items.append(item["sourceTitle"])
|
||
|
|
return items
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
sonarr_items = get_sonarr_history()
|
||
|
|
radarr_items = get_radarr_history()
|
||
|
|
|
||
|
|
output = {
|
||
|
|
"text": f"Son: {len(sonarr_items)}|Rad: {len(radarr_items)}",
|
||
|
|
"tooltip": "\n".join(radarr_items) + "\n".join(sonarr_items)
|
||
|
|
}
|
||
|
|
print(json.dumps(output))
|