91 lines
2.7 KiB
Python
91 lines
2.7 KiB
Python
from http import HTTPStatus
|
|
from http.server import SimpleHTTPRequestHandler
|
|
import json
|
|
import os
|
|
import socketserver
|
|
from typing import Optional
|
|
from urllib.parse import urlparse, parse_qs
|
|
|
|
PORT = 8000
|
|
|
|
class MyHandler(SimpleHTTPRequestHandler):
|
|
DATA_DIR = "metadata"
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(
|
|
*args,
|
|
directory="public",
|
|
**kwargs
|
|
)
|
|
self.query: dict = {}
|
|
self.data: Optional[dict|list] = None
|
|
|
|
def read_body_data(self):
|
|
self.log_message("Reading body data")
|
|
try:
|
|
raw_data = self.rfile.read(int(self.headers["Content-Length"]))
|
|
self.data = json.loads(raw_data)
|
|
except:
|
|
self.send_error(HTTPStatus.NOT_ACCEPTABLE, "Malformed JSON body")
|
|
self.log_error("Malformed JSON body")
|
|
return False
|
|
return True
|
|
|
|
def do_GET(self):
|
|
self.query = parse_qs(urlparse(self.path).query)
|
|
if self.path.startswith("/api/"):
|
|
self.handle_api_get(self.path.removeprefix("/api/").removesuffix("/"))
|
|
return
|
|
super().do_GET()
|
|
|
|
def do_POST(self):
|
|
self.query = parse_qs(urlparse(self.path).query)
|
|
if self.path.startswith("/api/"):
|
|
self.handle_api_post(self.path.removeprefix("/api/").removesuffix("/"))
|
|
return
|
|
self.send_error(HTTPStatus.NOT_FOUND)
|
|
|
|
def handle_api_get(self, path: str):
|
|
print(f"API request at {path}")
|
|
if path == "files":
|
|
files: list[str] = self.get_files()
|
|
self.send_json(files)
|
|
return
|
|
|
|
def handle_api_post(self, path: str):
|
|
if path == "file":
|
|
if self.read_body_data():
|
|
data = self.get_file(self.data["file"])
|
|
if data is None:
|
|
self.send_error(HTTPStatus.NOT_FOUND)
|
|
self.log_message("File not found")
|
|
else:
|
|
self.log_message("Got file")
|
|
self.send_json(data)
|
|
|
|
def send_json(self, data: dict|list):
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(data).encode("utf-8"))
|
|
|
|
def get_files(self):
|
|
return os.listdir(self.DATA_DIR)
|
|
|
|
def get_file(self, filename: str) -> Optional[dict|list]:
|
|
if filename not in self.get_files():
|
|
return None
|
|
with open(os.path.join(self.DATA_DIR, filename), "r") as f:
|
|
data = json.load(f)
|
|
return data
|
|
|
|
|
|
def main():
|
|
with socketserver.TCPServer(("", PORT), MyHandler) as httpd:
|
|
print(f"Serving on port {PORT}")
|
|
httpd.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|