"""Fake Radarr (19891), Sonarr (19892) and SABnzbd (19893) serving the made-up library in catalog.py, for the README screenshots. Only the API calls arr-tui makes are answered; posters come from posters/ (see posters.sh).""" import datetime, json, os, sys, threading, urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import catalog HERE = os.path.dirname(os.path.abspath(__file__)) TODAY = datetime.date.today() GB = 1 << 30 RADARR, SONARR, SAB = 19891, 19892, 19893 def day(n): return (TODAY + datetime.timedelta(days=n)).isoformat() def mediafile(res, codec, dr, audio, extra, lang, gb, name): return {"size": int(gb * GB), "sceneName": name, "relativePath": name + ".mkv", "quality": {"quality": {"name": ("Bluray-%dp" % res) if res == 2160 else ("WEBDL-%dp" % res), "resolution": res}}, "mediaInfo": {"videoCodec": codec, "videoDynamicRange": "HDR" if dr else "", "videoDynamicRangeType": dr, "audioCodec": audio, "audioAdditionalFeatures": extra}, "languages": [{"name": lang}]} def poster(port, kind, id): local = "/MediaCover/%d/poster.jpg" % id return [{"coverType": "poster", "url": local, "remoteUrl": "http://127.0.0.1:%d/remote/%s%d.jpg" % (port, kind, id)}] def movies(): out = [] for (id, title, year, genres, studio, runtime, rating, f, mon, overview, _c) in catalog.MOVIES: m = {"id": id, "title": title, "year": year, "genres": genres, "studio": studio, "runtime": runtime, "overview": overview, "tmdbId": 1000 + id, "titleSlug": title.lower().replace(" ", "-"), "monitored": mon, "hasFile": f is not None, "isAvailable": year < 2026, "status": "released" if year < 2026 else "announced", "path": "/movies/%s (%d)" % (title, year), "images": poster(RADARR, "m", id), "qualityProfileId": 4, "ratings": {"imdb": {"value": rating, "votes": 12000}} if rating else {}} if f: res, codec, dr, audio, extra, lang, gb = f name = "%s.%d.%dp.%s%s" % (title.replace(" ", "."), year, res, (dr.replace(" ", ".") + ".") if dr else "", codec) m["movieFile"] = mediafile(res, codec, dr, audio, extra, lang, gb, name) m["sizeOnDisk"] = int(gb * GB) out.append(m) return out def series(): out = [] for (id, title, year, network, status, seasons, overview, _c) in catalog.SERIES: eps = sum(s[0] for s in seasons); files = sum(s[1] for s in seasons) out.append({"id": id, "title": title, "year": year, "network": network, "status": status, "overview": overview, "runtime": 45, "genres": ["Drama"], "tvdbId": 2000 + id, "titleSlug": title.lower().replace(" ", "-"), "images": poster(SONARR, "s", id), "monitored": True, "ratings": {"value": 8.0, "votes": 900}, "path": "/tv/" + title, "seasons": [{"seasonNumber": i + 1, "monitored": True, "statistics": {"episodeFileCount": f, "episodeCount": e, "totalEpisodeCount": e, "sizeOnDisk": f * 2 * GB}} for i, (e, f) in enumerate(seasons)], "statistics": {"seasonCount": len(seasons), "episodeFileCount": files, "episodeCount": eps, "totalEpisodeCount": eps, "sizeOnDisk": files * 2 * GB}}) return out EPISODE_TITLES = ["Kindling", "The Ridge", "Backdraft", "Smoke Line", "Ember", "Firebreak", "Updraft", "Ash", "Tinder", "Afterglow", "Spark", "Wildland"] def episodes(series_id): s = next(x for x in catalog.SERIES if x[0] == series_id) out, eid = [], series_id * 1000 for si, (count, files) in enumerate(s[5]): for e in range(count): eid += 1 aired = day(-(len(s[5]) - si) * 90 + e * 7) out.append({"id": eid, "seriesId": series_id, "seasonNumber": si + 1, "episodeNumber": e + 1, "title": EPISODE_TITLES[e % len(EPISODE_TITLES)], "airDate": aired, "airDateUtc": aired + "T20:00:00Z", "overview": "The crew faces a long night on the ridge as the wind turns.", "hasFile": e < files, "monitored": True}) return out def releases(title, year): base = title.replace(" ", ".") rows = [ ("%s.%d.2160p.UHD.BluRay.REMUX.DV.HDR.HEVC.TrueHD.Atmos.7.1-FRAMES" % (base, year), "NZBgeek", 68.4, 2160, "Remux-2160p", True, [], 2, ["DV", "HDR10", "Atmos"]), ("%s.%d.2160p.WEB-DL.DDP5.1.Atmos.DV.HDR.H.265-FLUX" % (base, year), "DrunkenSlug", 21.9, 2160, "WEBDL-2160p", True, [], 1, ["DV", "Atmos"]), ("%s.%d.1080p.BluRay.DTS-HD.MA.5.1.x264-SPARKS" % (base, year), "NZBgeek", 14.2, 1080, "Bluray-1080p", True, [], 6, []), ("%s.%d.1080p.WEB-DL.DDP5.1.H.264-NTb" % (base, year), "NZBFinder", 6.8, 1080, "WEBDL-1080p", True, [], 3, []), ("%s.%d.2160p.WEB-DL.DDP5.1.HDR10.HEVC-GERMAN-iNTENSO" % (base, year), "DrunkenSlug", 18.1, 2160, "WEBDL-2160p", False, ["Language is not wanted"], 4, ["HDR10"]), ("%s.%d.720p.WEBRip.x264-ION10" % (base, year), "NZBFinder", 1.9, 720, "WEBRip-720p", False, ["Quality is not wanted in profile"], 9, []), ("%s.%d.1080p.HDTV.x264-CAM" % (base, year), "NZBgeek", 2.4, 1080, "HDTV-1080p", False, ["Release is a CAM", "Quality is not wanted in profile"], 1, []), ("%s.%d.2160p.BluRay.x265.10bit.HDR.DTS-HD.MA-TERMiNAL" % (base, year), "NZBFinder", 34.0, 2160, "Bluray-2160p", True, [], 12, ["HDR10"]), ] out = [] for i, (name, idx, gb, res, q, ok, rej, days, cfs) in enumerate(rows): out.append({"guid": "g%d" % i, "title": name, "indexer": idx, "indexerId": 1, "protocol": "usenet", "size": int(gb * GB), "ageMinutes": days * 1440.0, "qualityWeight": res, "customFormatScore": 100 * len(cfs), "releaseGroup": name.rsplit("-", 1)[-1], "approved": ok, "rejected": not ok, "rejections": rej, "quality": {"quality": {"name": q, "resolution": res}}, "languages": [{"name": "German" if "GERMAN" in name else "English"}], "customFormats": [{"name": c} for c in cfs]}) return out def radarr(path, q): if path.endswith("/system/status"): return {"appName": "Radarr", "version": "5.26.2"} if path.endswith("/movie"): return movies() if "/movie/" in path and path.rsplit("/", 1)[1].isdigit(): return next(x for x in movies() if x["id"] == int(path.rsplit("/", 1)[1])) if path.endswith("/movie/lookup"): term = q.get("term", [""])[0].lower() res = [m for m in movies() if term and term in m["title"].lower()] extra = [{"id": 0, "title": "The Long Orbit: Return", "year": 2027, "overview": "The sequel.", "tmdbId": 9001, "titleSlug": "the-long-orbit-return", "images": [{"coverType": "poster", "remoteUrl": "http://127.0.0.1:%d/remote/m1.jpg" % RADARR}], "status": "announced", "ratings": {}}] return res + (extra if "orbit" in term else []) if path.endswith("/qualityprofile"): return [{"id": 1, "name": "Any"}, {"id": 4, "name": "HD-1080p"}, {"id": 5, "name": "Ultra-HD"}] if path.endswith("/rootfolder"): return [{"id": 1, "path": "/movies", "freeSpace": 3 << 40, "accessible": True}] if path.endswith("/release"): m = next(x for x in catalog.MOVIES if x[0] == int(q["movieId"][0])) return releases(m[1], m[2]) if path.endswith("/queue"): return {"records": [ {"id": 1, "title": "Iron.Meridian.2021.2160p.WEB-DL.DDP5.1.Atmos.HDR.H.265-FLUX", "status": "downloading", "trackedDownloadStatus": "ok", "trackedDownloadState": "downloading", "size": 21.9 * GB, "sizeleft": 7.2 * GB, "timeleft": "00:06:12", "downloadClient": "SABnzbd", "downloadId": "SABnzbd_nzo_a1", "protocol": "usenet", "movie": {"title": "Iron Meridian", "year": 2021}}, {"id": 2, "title": "Static.2018.1080p.BluRay.x264-AMIABLE", "status": "completed", "trackedDownloadStatus": "warning", "trackedDownloadState": "importBlocked", "size": 6.0 * GB, "sizeleft": 0, "timeleft": "", "downloadClient": "SABnzbd", "downloadId": "SABnzbd_nzo_h3", "protocol": "usenet", "statusMessages": [{"title": "Static.2018.1080p", "messages": ["Not an upgrade for existing movie file"]}], "movie": {"title": "Static", "year": 2018}}]} if path.endswith("/calendar"): out = [] for off, mid, kind in [(1, 9, "digitalRelease"), (5, 15, "inCinemas"), (9, 4, "physicalRelease")]: m = dict(next(x for x in movies() if x["id"] == mid)); m[kind] = day(off) + "T00:00:00Z"; out.append(m) return out return [] def sonarr(path, q): if path.endswith("/system/status"): return {"appName": "Sonarr", "version": "4.0.15"} if path.endswith("/series"): return series() if "/series/" in path and path.rsplit("/", 1)[1].isdigit(): return next(x for x in series() if x["id"] == int(path.rsplit("/", 1)[1])) if path.endswith("/episode"): return episodes(int(q["seriesId"][0])) if path.endswith("/episodefile"): return [] if path.endswith("/qualityprofile"): return [{"id": 1, "name": "Any"}, {"id": 4, "name": "HD-1080p"}] if path.endswith("/rootfolder"): return [{"id": 1, "path": "/tv", "freeSpace": 3 << 40, "accessible": True}] if path.endswith("/queue"): return {"records": [ {"id": 7, "title": "Signal.Fire.S03E05.1080p.WEB-DL.DDP5.1.H.264-NTb", "status": "downloading", "trackedDownloadStatus": "ok", "trackedDownloadState": "downloading", "size": 2.1 * GB, "sizeleft": 1.3 * GB, "timeleft": "00:01:40", "downloadClient": "SABnzbd", "downloadId": "SABnzbd_nzo_a2", "protocol": "usenet", "series": {"title": "Signal Fire"}, "episode": {"seasonNumber": 3, "episodeNumber": 5, "title": "Ember"}}]} if path.endswith("/calendar"): out = [] for off, sid, s, e, t in [(0, 1, 3, 7, "Updraft"), (2, 3, 2, 4, "Code Blue"), (4, 5, 1, 3, "Dock"), (7, 1, 3, 8, "Ash"), (11, 3, 2, 5, "Triage")]: ser = next(x for x in series() if x["id"] == sid) out.append({"id": 90000 + off, "seriesId": sid, "seasonNumber": s, "episodeNumber": e, "title": t, "airDate": day(off), "airDateUtc": day(off) + "T20:00:00Z", "monitored": True, "hasFile": False, "overview": "A new episode of " + ser["title"] + ".", "series": ser}) return out return [] def sab(q): mode = q.get("mode", [""])[0] if mode == "version": return {"version": "4.5.1"} if mode == "queue": return {"queue": {"status": "Downloading", "paused": False, "kbpersec": "48213", "speedlimit_abs": "0", "size": "26.4 GB", "sizeleft": "9.1 GB", "timeleft": "0:08:05", "noofslots_total": 3, "slots": [ {"nzo_id": "nzo_a1", "filename": "Iron.Meridian.2021.2160p.WEB-DL.DDP5.1.Atmos.HDR.H.265-FLUX", "cat": "movies", "status": "Downloading", "size": "21.9 GB", "sizeleft": "7.2 GB", "mb": "22425", "mbleft": "7372", "percentage": "67", "timeleft": "0:06:12"}, {"nzo_id": "nzo_a2", "filename": "Signal.Fire.S03E05.1080p.WEB-DL.DDP5.1.H.264-NTb", "cat": "tv", "status": "Downloading", "size": "2.1 GB", "sizeleft": "1.3 GB", "mb": "2150", "mbleft": "1331", "percentage": "38", "timeleft": "0:01:40"}, {"nzo_id": "nzo_a3", "filename": "Hearth.and.Home.S03.1080p.BluRay.x264-ROVERS", "cat": "tv", "status": "Queued", "size": "2.4 GB", "sizeleft": "2.4 GB", "mb": "2457", "mbleft": "2457", "percentage": "0", "timeleft": "0:00:00"}]}} if mode == "history": now = int(datetime.datetime.now().timestamp()) return {"history": {"slots": [ {"nzo_id": "nzo_h1", "name": "The.Long.Orbit.2023.2160p.UHD.BluRay.REMUX.DV.HDR.HEVC.TrueHD.Atmos-FRAMES", "category": "movies", "status": "Completed", "size": "58.2 GB", "fail_message": "", "completed": now - 3600}, {"nzo_id": "nzo_h2", "name": "Signal.Fire.S03E04.1080p.WEB-DL.DDP5.1.H.264-NTb", "category": "tv", "status": "Completed", "size": "2.0 GB", "fail_message": "", "completed": now - 7200}, {"nzo_id": "nzo_h3", "name": "Static.2018.1080p.BluRay.x264-AMIABLE", "category": "movies", "status": "Completed", "size": "6.0 GB", "fail_message": "", "completed": now - 20000}, {"nzo_id": "nzo_h4", "name": "Low.Tide.S02E06.720p.HDTV.x264-KILLERS", "category": "tv", "status": "Failed", "size": "1.1 GB", "fail_message": "Aborted, cannot be completed - https://sabnzbd.org/not-complete", "completed": now - 90000}, {"nzo_id": "nzo_h5", "name": "Neon.Harbour.2023.2160p.WEB-DL.DTS-HD.MA.HDR10Plus.HEVC-SMURF", "category": "movies", "status": "Completed", "size": "44.5 GB", "fail_message": "", "completed": now - 180000}]}} return {} def handler(port): class H(BaseHTTPRequestHandler): def log_message(self, *a): pass def do_GET(self): u = urllib.parse.urlparse(self.path); q = urllib.parse.parse_qs(u.query); path = u.path if path.startswith("/MediaCover/") or path.startswith("/remote/"): if path.startswith("/MediaCover/"): id = path.split("/")[2]; f = ("m" if port == RADARR else "s") + id + ".jpg" else: f = path.rsplit("/", 1)[1] p = os.path.join(HERE, "posters", f) if not os.path.exists(p): self.send_response(404); self.end_headers(); return data = open(p, "rb").read() self.send_response(200); self.send_header("Content-Type", "image/jpeg"); self.send_header("Content-Length", str(len(data))); self.end_headers(); self.wfile.write(data); return out = sab(q) if port == SAB else (radarr(path, q) if port == RADARR else sonarr(path, q)) b = json.dumps(out).encode() self.send_response(200); self.send_header("Content-Type", "application/json"); self.send_header("Content-Length", str(len(b))); self.end_headers(); self.wfile.write(b) return H for port in (RADARR, SONARR): threading.Thread(target=ThreadingHTTPServer(("127.0.0.1", port), handler(port)).serve_forever, daemon=True).start() ThreadingHTTPServer(("127.0.0.1", SAB), handler(SAB)).serve_forever()