#!/usr/bin/env python3 """HTTP glue and command line for the central Blueforce panel (stdlib only). python3 -m web.server # loopback, demo data, read-only python3 -m web.server --data ./fleet.json # a real inventory document python3 -m web.server --port 9100 --privilege full python3 -m web.server --check # verify the data source, no bind ``http.server`` is the whole server: ``fastapi``/``uvicorn``/``flask`` are not in the committed offline package pool, so the panel cannot ship with them (PRODUCT.md, "Stack"). The routing itself lives in ``web/panel.py`` and is a pure function, which is why the test suite can cover every endpoint without binding a port. """ from __future__ import annotations import argparse import http.server import sys from pathlib import Path from typing import Sequence try: # `python3 -m web.server` from . import ( DEFAULT_HOST, DEFAULT_PORT, PANEL_NAME, PANEL_SURFACE, PANEL_TITLE, PANEL_VERSION, ) from . import fleet_data, render from .panel import PanelContext, Response, route except ImportError: # `python3 web/server.py` sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from web import ( # type: ignore[no-redef] DEFAULT_HOST, DEFAULT_PORT, PANEL_NAME, PANEL_SURFACE, PANEL_TITLE, PANEL_VERSION, ) from web import fleet_data, render # type: ignore[no-redef] from web.panel import PanelContext, Response, route # type: ignore[no-redef] __all__ = [ "PanelRequestHandler", "build_context", "main", "make_server", "parse_args", ] class PanelRequestHandler(http.server.BaseHTTPRequestHandler): """One handler for every method: the panel answers, it never acts. Only the two read methods can produce a page; everything else is refused by ``route`` with 405 and the connection is closed, so a write request cannot leave half-read state behind. """ server_version = f"BlueforcePanel/{PANEL_VERSION}" protocol_version = "HTTP/1.1" def do_GET(self) -> None: # noqa: N802 - stdlib naming self._respond("GET") def do_HEAD(self) -> None: # noqa: N802 - stdlib naming self._respond("HEAD") def do_POST(self) -> None: # noqa: N802 - stdlib naming self._respond("POST") def do_PUT(self) -> None: # noqa: N802 - stdlib naming self._respond("PUT") def do_PATCH(self) -> None: # noqa: N802 - stdlib naming self._respond("PATCH") def do_DELETE(self) -> None: # noqa: N802 - stdlib naming self._respond("DELETE") def do_OPTIONS(self) -> None: # noqa: N802 - stdlib naming self._respond("OPTIONS") def do_TRACE(self) -> None: # noqa: N802 - stdlib naming self._respond("TRACE") # -- internals --------------------------------------------------------- def _respond(self, method: str) -> None: context: PanelContext = self.server.context # type: ignore[attr-defined] response: Response = route(context, method, self.path) self.send_response(response.status) self.send_header("Content-Type", response.content_type) for name, value in response.headers: self.send_header(name, value) self.send_header("Content-Length", str(response.content_length)) self.end_headers() if method != "HEAD" and response.body: self.wfile.write(response.body) if method not in ("GET", "HEAD"): self.close_connection = True def log_message(self, format: str, *args: object) -> None: # noqa: A002 sys.stderr.write("[panel] %s %s\n" % (self.address_string(), format % args)) def make_server(context: PanelContext, host: str, port: int) -> http.server.ThreadingHTTPServer: """Bind the panel. Threaded so a slow page never blocks the operator.""" server = http.server.ThreadingHTTPServer((host, port), PanelRequestHandler) server.context = context # type: ignore[attr-defined] return server def build_context(args: argparse.Namespace) -> PanelContext: """Load the inventory and describe the running panel.""" snapshot = fleet_data.load_fleet(args.data) privilege = render.privilege_from_level(args.privilege) server_info = render.ServerInfo( host=args.host, port=args.port, privilege=privilege, python_version=f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}", started_at=fleet_data.timestamp_now(), data_path=args.data, ) return PanelContext(snapshot=snapshot, privilege=privilege, server_info=server_info) def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( prog="web/server.py", description=f"{PANEL_TITLE} — filo yönetim paneli (yalnız okuma).", ) parser.add_argument( "--host", default=DEFAULT_HOST, help=f"Dinlenecek adres (varsayılan {DEFAULT_HOST}: yalnız yerel arayüz).", ) parser.add_argument( "--port", type=int, default=DEFAULT_PORT, help=f"Dinlenecek port (varsayılan {DEFAULT_PORT}).", ) parser.add_argument( "--data", default=None, help=( "Filo envanteri JSON dosyası. Verilmezse ve varsayılan konum boşsa " "etiketli demo veri gösterilir; verilir de okunamazsa panel demo veriye " "düşmez, «VERİ YOK» der." ), ) parser.add_argument( "--privilege", choices=(render.PRIVILEGE_READ_ONLY_LEVEL, render.PRIVILEGE_FULL_LEVEL), default=render.PRIVILEGE_READ_ONLY_LEVEL, help="Yetki düzeyi: salt-okunur (varsayılan) ya da tam yetki.", ) parser.add_argument( "--check", action="store_true", help="Sunucuyu başlatmadan veri kaynağını denetle ve Türkçe durum yazıp çık.", ) parser.add_argument( "--version", action="version", version=f"{PANEL_TITLE} {PANEL_VERSION}", ) return parser.parse_args(argv) def _print_status(context: PanelContext) -> None: snapshot = context.snapshot print(f"{PANEL_NAME} — {PANEL_SURFACE} {PANEL_VERSION}") print(f"Veri kaynağı : {snapshot.source_label}") print(f"Veri durumu : {snapshot.banner_word} — {snapshot.banner_text}") print( f"Cihazlar : {snapshot.total} cihaz · hazır {snapshot.ready_count} · " f"hazır değil {snapshot.not_ready_count} · geçersiz kayıt {len(snapshot.invalid)}" ) print(f"Yetki : {context.privilege.word} — {context.privilege.reason}") print(f"Adres : {context.server_info.bind_address} (yalnız okuma, yerel arayüz)") def main(argv: Sequence[str] | None = None) -> int: args = parse_args(argv) context = build_context(args) _print_status(context) if args.check: print("Denetim tamam; sunucu başlatılmadı (--check).") return 0 server = make_server(context, args.host, args.port) print("Durdurmak için Ctrl+C.") try: server.serve_forever() except KeyboardInterrupt: print("\nPanel durduruldu.") finally: server.server_close() return 0 if __name__ == "__main__": raise SystemExit(main())