89 lines
2.7 KiB
Python
Executable File
89 lines
2.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import json
|
|
import os
|
|
import sys
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import urlparse
|
|
|
|
view = os.environ["TEST_VIEW"]
|
|
state_path = os.environ["TEST_STATE"]
|
|
codec = "hevc"
|
|
depth = 10
|
|
refresh_attempts = 0
|
|
|
|
|
|
def record_state():
|
|
with open(state_path, "w") as handle:
|
|
json.dump({"codec": codec, "depth": depth, "attempts": refresh_attempts}, handle)
|
|
|
|
|
|
def refresh_media():
|
|
global codec, depth
|
|
link = os.path.join(view, "movies/Test/Test.mkv")
|
|
try:
|
|
target = os.readlink(link)
|
|
except OSError:
|
|
target = ""
|
|
codec, depth = ("h264", 8) if target.startswith("/data/cache/") else ("hevc", 10)
|
|
record_state()
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def send_json(self, value, status=200):
|
|
body = json.dumps(value, separators=(",", ":")).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self):
|
|
path = urlparse(self.path).path
|
|
if path == "/Sessions":
|
|
self.send_json([])
|
|
elif path == "/Items/test-item":
|
|
self.send_json({"Id": "test-item", "Path": "/data/library/movies/Test/Test.mkv"})
|
|
elif path == "/Items":
|
|
self.send_json({
|
|
"Items": [{
|
|
"Id": "test-item",
|
|
"Path": "/data/library/movies/Test/Test.mkv",
|
|
"RunTimeTicks": 6 * 10_000_000,
|
|
"MediaStreams": [{
|
|
"Type": "Video", "Codec": codec, "BitDepth": depth,
|
|
"Profile": "Main 10" if depth == 10 else "High",
|
|
"Width": 1280, "Height": 720,
|
|
}],
|
|
}],
|
|
"TotalRecordCount": 1,
|
|
})
|
|
else:
|
|
self.send_json({"error": "not found"}, 404)
|
|
|
|
def do_POST(self):
|
|
global refresh_attempts
|
|
path = urlparse(self.path).path
|
|
if path.startswith("/Items/test-item/Refresh"):
|
|
refresh_attempts += 1
|
|
if refresh_attempts == 1:
|
|
record_state()
|
|
self.send_response(503)
|
|
self.end_headers()
|
|
return
|
|
refresh_media()
|
|
self.send_response(204)
|
|
self.end_headers()
|
|
elif path == "/Library/Refresh":
|
|
self.send_response(204)
|
|
self.end_headers()
|
|
else:
|
|
self.send_json({"error": "not found"}, 404)
|
|
|
|
def log_message(self, fmt, *args):
|
|
pass
|
|
|
|
|
|
record_state()
|
|
server = ThreadingHTTPServer(("0.0.0.0", int(sys.argv[1])), Handler)
|
|
server.serve_forever()
|