213 lines
6.8 KiB
Python
213 lines
6.8 KiB
Python
"""Tiny stdlib HTTP server for the demo client page.
|
||
|
||
- Serves `index.html` (and other sibling files) from this directory.
|
||
- Exposes `GET /labels.json`, which parses the repo's
|
||
`app/resources/consumable_classifier_labels.yaml` and returns its label
|
||
list so the page can prefill the candidate-consumables input.
|
||
- ``--live`` 使用 dev 依赖 ``livereload``：修改本目录下 HTML/JS/CSS
|
||
或更新 ``labels.json``/YAML 后，浏览器自动刷新（需 ``uv run --group dev``）。
|
||
|
||
Run:
|
||
python scripts/demo_client/server.py # 127.0.0.1:38081
|
||
python scripts/demo_client/server.py -p 9000 # custom port
|
||
uv run --group dev python scripts/demo_client/server.py --live
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import re
|
||
import sys
|
||
import threading
|
||
import time
|
||
from http import HTTPStatus
|
||
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
REPO_ROOT = SCRIPT_DIR.parents[1]
|
||
LABELS_YAML = REPO_ROOT / "app" / "resources" / "consumable_classifier_labels.yaml"
|
||
# ``--live`` 模式写入静态 `labels.json`，与 livereload 兼容（勿提交，见 .gitignore）
|
||
LABELS_JSON_ARTIFACT = SCRIPT_DIR / "labels.json"
|
||
|
||
|
||
def _load_labels_with_pyyaml(path: Path) -> list[str] | None:
|
||
try:
|
||
import yaml # type: ignore[import-untyped]
|
||
except ImportError:
|
||
return None
|
||
try:
|
||
data: Any = yaml.safe_load(path.read_text(encoding="utf-8"))
|
||
except (OSError, yaml.YAMLError):
|
||
return None
|
||
if not isinstance(data, dict):
|
||
return None
|
||
names = data.get("names")
|
||
if isinstance(names, dict):
|
||
try:
|
||
items = sorted(names.items(), key=lambda kv: int(kv[0]))
|
||
except (TypeError, ValueError):
|
||
items = list(names.items())
|
||
return [str(v) for _, v in items]
|
||
if isinstance(names, list):
|
||
return [str(v) for v in names]
|
||
return None
|
||
|
||
|
||
def _load_labels_fallback(path: Path) -> list[str]:
|
||
"""Minimal parser for the known labels.yaml shape: `<int>: <text>` under `names:`."""
|
||
labels: list[tuple[int, str]] = []
|
||
in_names = False
|
||
pattern = re.compile(r"^\s+(\d+)\s*:\s*(.+?)\s*$")
|
||
try:
|
||
text = path.read_text(encoding="utf-8")
|
||
except OSError:
|
||
return []
|
||
for line in text.splitlines():
|
||
stripped = line.strip()
|
||
if not stripped or stripped.startswith("#"):
|
||
continue
|
||
if stripped.startswith("names:"):
|
||
in_names = True
|
||
continue
|
||
if in_names and not line.startswith((" ", "\t")):
|
||
in_names = False
|
||
if not in_names:
|
||
continue
|
||
match = pattern.match(line)
|
||
if match:
|
||
labels.append((int(match.group(1)), match.group(2)))
|
||
labels.sort(key=lambda kv: kv[0])
|
||
return [name for _, name in labels]
|
||
|
||
|
||
def load_labels() -> list[str]:
|
||
if not LABELS_YAML.is_file():
|
||
return []
|
||
labels = _load_labels_with_pyyaml(LABELS_YAML)
|
||
if labels is None:
|
||
labels = _load_labels_fallback(LABELS_YAML)
|
||
return labels
|
||
|
||
|
||
def write_labels_json_artifact() -> None:
|
||
body = json.dumps({"labels": load_labels()}, ensure_ascii=False) + "\n"
|
||
LABELS_JSON_ARTIFACT.write_text(body, encoding="utf-8")
|
||
|
||
|
||
def _spawn_labels_yaml_poll() -> None:
|
||
p = LABELS_YAML
|
||
state: list[float] = [0.0]
|
||
if p.is_file():
|
||
state[0] = p.stat().st_mtime
|
||
|
||
def loop() -> None:
|
||
while True:
|
||
time.sleep(1.0)
|
||
if not p.is_file():
|
||
continue
|
||
m = p.stat().st_mtime
|
||
if m > state[0]:
|
||
state[0] = m
|
||
try:
|
||
write_labels_json_artifact()
|
||
except OSError as e:
|
||
print(
|
||
f"[demo-client] labels.json 写入失败: {e}",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
threading.Thread(target=loop, daemon=True, name="DemoClientLabelsYamlWatch").start()
|
||
|
||
|
||
def run_livereload(host: str, port: int) -> None:
|
||
try:
|
||
from livereload import Server
|
||
except ImportError as exc: # pragma: no cover
|
||
print(
|
||
"``--live`` 需要 dev 依赖。请执行: uv sync --group dev",
|
||
file=sys.stderr,
|
||
)
|
||
raise SystemExit(1) from exc
|
||
|
||
write_labels_json_artifact()
|
||
_spawn_labels_yaml_poll()
|
||
server = Server()
|
||
server.watch(str(SCRIPT_DIR))
|
||
if LABELS_YAML.is_file():
|
||
server.watch(str(LABELS_YAML))
|
||
url = f"http://{host}:{port}/"
|
||
print(f"Demo client (livereload) at {url}")
|
||
print(f" static dir : {SCRIPT_DIR}")
|
||
print(f" labels yaml: {LABELS_YAML} -> {LABELS_JSON_ARTIFACT.name}")
|
||
print("Press Ctrl+C to stop.")
|
||
server.serve(
|
||
root=str(SCRIPT_DIR),
|
||
host=host,
|
||
port=port,
|
||
open_url=False,
|
||
debug=False,
|
||
)
|
||
|
||
|
||
class DemoHandler(SimpleHTTPRequestHandler):
|
||
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
||
super().__init__(*args, directory=str(SCRIPT_DIR), **kwargs)
|
||
|
||
def do_GET(self) -> None: # noqa: N802 (stdlib override)
|
||
if self.path.split("?", 1)[0] == "/labels.json":
|
||
self._send_labels()
|
||
return
|
||
super().do_GET()
|
||
|
||
def _send_labels(self) -> None:
|
||
body = json.dumps({"labels": load_labels()}, ensure_ascii=False).encode("utf-8")
|
||
self.send_response(HTTPStatus.OK)
|
||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def log_message(self, format: str, *args: Any) -> None: # noqa: A002
|
||
sys.stderr.write(
|
||
"[demo-client] %s - %s\n" % (self.address_string(), format % args)
|
||
)
|
||
|
||
|
||
def run_plain(host: str, port: int) -> None:
|
||
server = ThreadingHTTPServer((host, port), DemoHandler)
|
||
url = f"http://{host}:{port}/"
|
||
print(f"Demo client serving at {url}")
|
||
print(f" static dir : {SCRIPT_DIR}")
|
||
print(f" labels yaml: {LABELS_YAML}")
|
||
print("Press Ctrl+C to stop.")
|
||
try:
|
||
server.serve_forever()
|
||
except KeyboardInterrupt:
|
||
print("\nShutting down.")
|
||
finally:
|
||
server.server_close()
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="Operation room demo client server")
|
||
parser.add_argument("--host", default="127.0.0.1")
|
||
parser.add_argument("-p", "--port", type=int, default=38081)
|
||
parser.add_argument(
|
||
"--live",
|
||
action="store_true",
|
||
help="浏览器热重载（需 uv sync --group dev；会生成同目录 labels.json）",
|
||
)
|
||
args = parser.parse_args()
|
||
if args.live:
|
||
run_livereload(args.host, args.port)
|
||
else:
|
||
run_plain(args.host, args.port)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|