import http.server import socketserver import urllib.parse import urllib.request import json import datetime from pathlib import Path import re PORT = 8001 OREF_ENDPOINT = "https://alerts-history.oref.org.il/Shared/Ajax/GetAlarmsHistory.aspx" BASE_DIR = Path(__file__).resolve().parent CHAGIM_DIR = BASE_DIR / "chagim" CHAGIM_DIR.mkdir(exist_ok=True) def safe_filename(filename): """ Allow only simple .txt filenames. Prevent ../ path traversal. """ if not filename: return None filename = Path(filename).name if not re.fullmatch(r"[A-Za-z0-9_\-א-ת]+\.txt", filename): return None return filename def read_chag_file(path): """ Convert: title|ט' באב event|מנחה|18:00 into JSON. """ title = "" events = [] with open(path, "r", encoding="utf-8") as f: for line in f: line = line.rstrip("\r\n") if not line: continue parts = line.split("|") if parts[0] == "title" and len(parts) >= 2: title = parts[1] elif parts[0] == "event" and len(parts) >= 3: events.append({ "description": parts[1], "time": parts[2] }) return { "title": title, "events": events } def write_chag_file(path, data): """ Convert JSON back to the TXT format. """ title = str(data.get("title", "")) events = data.get("events", []) lines = [ f"title|{title}" ] for event in events: description = str(event.get("description", "")) time = str(event.get("time", "")) lines.append( f"event|{description}|{time}" ) with open(path, "w", encoding="utf-8") as f: f.write("\n".join(lines) + "\n") class Handler(http.server.SimpleHTTPRequestHandler): def end_headers(self) -> None: self.send_header("Access-Control-Allow-Origin", "*") self.send_header( "Access-Control-Allow-Methods", "GET, POST, OPTIONS" ) self.send_header( "Access-Control-Allow-Headers", "Content-Type" ) super().end_headers() def send_json(self, data, status=200): response = json.dumps( data, ensure_ascii=False ).encode("utf-8") self.send_response(status) self.send_header( "Content-Type", "application/json; charset=utf-8" ) self.send_header( "Content-Length", str(len(response)) ) self.end_headers() self.wfile.write(response) def read_json_body(self): length = int( self.headers.get("Content-Length", 0) ) body = self.rfile.read(length) return json.loads(body) def do_POST(self): parsed = urllib.parse.urlparse(self.path) # --------------------------------------------------------- # Existing display configuration API # --------------------------------------------------------- if parsed.path == "/save-display-config": try: config = self.read_json_body() with open( BASE_DIR / "display.json", "w", encoding="utf-8" ) as f: json.dump( config, f, ensure_ascii=False, indent=2 ) self.send_json({"ok": True}) except Exception as e: self.send_json( {"error": str(e)}, 500 ) return # --------------------------------------------------------- # Create new chag file # --------------------------------------------------------- if parsed.path == "/api/chagim": try: data = self.read_json_body() filename = safe_filename( data.get("filename", "") ) if not filename: self.send_json( {"error": "Invalid filename"}, 400 ) return path = CHAGIM_DIR / filename if path.exists(): self.send_json( {"error": "File already exists"}, 409 ) return write_chag_file( path, { "title": data.get("title", ""), "events": data.get("events", []) } ) self.send_json({ "ok": True, "filename": filename }) except Exception as e: self.send_json( {"error": str(e)}, 500 ) return # --------------------------------------------------------- # Save existing chag file # --------------------------------------------------------- if parsed.path.startswith("/api/chagim/"): filename = urllib.parse.unquote( parsed.path[len("/api/chagim/"):] ) filename = safe_filename(filename) if not filename: self.send_json( {"error": "Invalid filename"}, 400 ) return path = CHAGIM_DIR / filename try: data = self.read_json_body() write_chag_file(path, data) self.send_json({ "ok": True, "filename": filename }) except Exception as e: self.send_json( {"error": str(e)}, 500 ) return self.send_error(404, "Not Found") def do_OPTIONS(self) -> None: self.send_response(204) self.end_headers() def do_GET(self): parsed = urllib.parse.urlparse(self.path) # --------------------------------------------------------- # List chagim files # --------------------------------------------------------- if parsed.path == "/api/chagim": try: files = sorted( [ f.name for f in CHAGIM_DIR.iterdir() if f.is_file() and f.suffix.lower() == ".txt" ] ) self.send_json({ "files": files }) except Exception as e: self.send_json( {"error": str(e)}, 500 ) return # --------------------------------------------------------- # Read a specific chag file # --------------------------------------------------------- if parsed.path.startswith("/api/chagim/"): filename = urllib.parse.unquote( parsed.path[len("/api/chagim/"):] ) filename = safe_filename(filename) if not filename: self.send_json( {"error": "Invalid filename"}, 400 ) return path = CHAGIM_DIR / filename if not path.exists(): self.send_json( {"error": "File not found"}, 404 ) return try: data = read_chag_file(path) data["filename"] = filename self.send_json(data) except Exception as e: self.send_json( {"error": str(e)}, 500 ) return # --------------------------------------------------------- # Existing OREF proxy # --------------------------------------------------------- if parsed.path == "/alerts-proxy": qs = parsed.query or "" if qs: url = ( f"{OREF_ENDPOINT}?{qs}" f"&_cb={int(datetime.datetime.now().timestamp())}" ) else: url = ( f"{OREF_ENDPOINT}" f"?_cb={int(datetime.datetime.now().timestamp())}" ) print( f"[{datetime.datetime.now().isoformat(timespec='seconds')}] " f"upstream GET {url}" ) req = urllib.request.Request( url, headers={ "User-Agent": "Mozilla/5.0", "Accept": "application/json,text/plain,*/*", "Cache-Control": "no-cache", "Pragma": "no-cache", }, method="GET", ) try: with urllib.request.urlopen( req, timeout=10 ) as resp: body = resp.read() content_type = resp.headers.get( "Content-Type", "application/json; charset=utf-8" ) status = resp.status self.send_response(status) self.send_header( "Content-Type", content_type ) self.send_header( "Cache-Control", "no-store" ) self.send_header( "Pragma", "no-cache" ) self.end_headers() self.wfile.write(body) except Exception as e: msg = json.dumps( { "error": "alerts-proxy failed", "detail": str(e) }, ensure_ascii=False ).encode("utf-8") self.send_response(502) self.send_header( "Content-Type", "application/json; charset=utf-8" ) self.send_header( "Cache-Control", "no-store" ) self.send_header( "Pragma", "no-cache" ) self.end_headers() self.wfile.write(msg) return # --------------------------------------------------------- # Normal static files # --------------------------------------------------------- return super().do_GET() if __name__ == "__main__": with socketserver.ThreadingTCPServer( ("", PORT), Handler ) as httpd: print( f"Serving on http://0.0.0.0:{PORT}" ) print( f"Chagim directory: {CHAGIM_DIR}" ) httpd.serve_forever()