From 4092854b0bea9e1fc02f29222c6f7cd876e565a5 Mon Sep 17 00:00:00 2001
From: shuishen <1109946754@qq.com>
Date: Mon, 17 Aug 2026 09:43:20 +0800
Subject: [PATCH] feat:项目基础整理
---
scripts/serve_workbench_console.py | 256 ++++++++++++++++++++++++++++++++++++++++++++++++--
1 files changed, 242 insertions(+), 14 deletions(-)
diff --git a/scripts/serve_workbench_console.py b/scripts/serve_workbench_console.py
index 83126a9..302eb37 100644
--- a/scripts/serve_workbench_console.py
+++ b/scripts/serve_workbench_console.py
@@ -1,44 +1,275 @@
-"""Serve the local GeoAI Workbench console from the repository root."""
+"""Serve the local GeoAI Workbench console and its narrow local-run APIs."""
from __future__ import annotations
import argparse
+import base64
+import binascii
+import json
import os
-from pathlib import PurePosixPath
+import re
+import subprocess
+import threading
+from datetime import UTC, datetime
from http import HTTPStatus
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
-from pathlib import Path
+from pathlib import Path, PurePosixPath
+from typing import Any
from urllib.parse import unquote, urlsplit
+from uuid import uuid4
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 6173
+MAX_REQUEST_BYTES = 128 * 1024 * 1024
+MAX_FILE_BYTES = 96 * 1024 * 1024
+MAX_IMAGES_PER_RUN = 12
ALLOWED_PATH_PREFIXES = (
"apps/workbench-console",
"shared/outputs",
"shared/data/raw/01-object-detection",
)
+SAFE_FILE_NAME = re.compile(r"[^A-Za-z0-9._-]+")
+RUN_LOCK = threading.Lock()
+
+
+class ApiError(ValueError):
+ """A request error that can be shown to the local console user."""
+
+
+def safe_file_name(value: str, expected_suffixes: set[str]) -> str:
+ name = Path(value).name
+ suffix = Path(name).suffix.lower()
+ if suffix not in expected_suffixes:
+ raise ApiError(f"Unsupported file type: {suffix or '(none)'}.")
+ stem = SAFE_FILE_NAME.sub("_", Path(name).stem).strip("._") or "upload"
+ return f"{stem[:80]}{suffix}"
+
+
+def decode_upload(payload: dict[str, Any], expected_suffixes: set[str]) -> tuple[str, bytes]:
+ if not isinstance(payload, dict) or not isinstance(payload.get("name"), str) or not isinstance(payload.get("content"), str):
+ raise ApiError("Each uploaded file must include name and Base64 content.")
+ name = safe_file_name(payload["name"], expected_suffixes)
+ try:
+ content = base64.b64decode(payload["content"], validate=True)
+ except (binascii.Error, ValueError) as exc:
+ raise ApiError(f"Invalid Base64 file content for {name}.") from exc
+ if not content:
+ raise ApiError(f"Uploaded file is empty: {name}.")
+ if len(content) > MAX_FILE_BYTES:
+ raise ApiError(f"Uploaded file exceeds {MAX_FILE_BYTES // (1024 * 1024)} MB: {name}.")
+ return name, content
+
+
+def make_run_id(prefix: str) -> str:
+ return f"{prefix}-{datetime.now(UTC):%Y%m%d-%H%M%S}-{uuid4().hex[:6]}"
+
+
+def relative_path(root: Path, path: Path) -> str:
+ return path.relative_to(root).as_posix()
+
+
+def load_json(path: Path) -> dict[str, Any]:
+ try:
+ payload = json.loads(path.read_text(encoding="utf-8"))
+ except (OSError, json.JSONDecodeError):
+ return {}
+ return payload if isinstance(payload, dict) else {}
+
+
+def trajectory_runs(root: Path) -> list[dict[str, Any]]:
+ output_root = root / "shared" / "outputs" / "15-trajectory-analysis"
+ records: list[dict[str, Any]] = []
+ for metadata_path in output_root.rglob("run_metadata.json"):
+ artifact = metadata_path.parent
+ if not (artifact / "trajectory_summary.csv").is_file() or not (artifact / "events.json").is_file():
+ continue
+ metadata = load_json(metadata_path)
+ case_id = str(metadata.get("case_id") or artifact.name)
+ is_real = case_id == "tian-dun-flight-19578"
+ records.append(
+ {
+ "id": case_id,
+ "label": "田墩实飞" if is_real else case_id,
+ "note": "区域相交是来源数据的空间结果，不是违规结论。" if is_real else "本地实验运行结果，可继续查看结构化输出。",
+ "artifactRoot": relative_path(root, artifact),
+ "showSpatialContext": (artifact / "zones.geojson").is_file() and (artifact / "reference_routes.geojson").is_file(),
+ "showFlyableZones": (artifact / "flyable_zones.geojson").is_file(),
+ "createdAt": str(metadata.get("created_at") or ""),
+ }
+ )
+ return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)
+
+
+def detection_runs(root: Path) -> list[dict[str, Any]]:
+ output_root = root / "shared" / "outputs" / "01-object-detection"
+ records: list[dict[str, Any]] = []
+ for metadata_path in output_root.rglob("run_metadata.json"):
+ artifact = metadata_path.parent
+ if not (artifact / "detections.json").is_file():
+ continue
+ metadata = load_json(metadata_path)
+ input_value = str(metadata.get("input_dir") or "")
+ input_dir = Path(input_value) if input_value else root / "shared" / "data" / "raw" / "01-object-detection"
+ try:
+ input_root = relative_path(root, input_dir.resolve())
+ except ValueError:
+ continue
+ run_id = artifact.name if artifact != output_root else "baseline"
+ records.append(
+ {
+ "id": run_id,
+ "label": "既有基线结果" if run_id == "baseline" else run_id,
+ "note": "CPU 基线：人员与常见车辆；树木不在当前模型有效类别内。",
+ "artifactRoot": relative_path(root, artifact),
+ "inputRoot": input_root,
+ "createdAt": str(metadata.get("created_at") or ""),
+ }
+ )
+ return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)
class WorkbenchConsoleHandler(SimpleHTTPRequestHandler):
- """Read-only static handler rooted at the workbench repository."""
+ """Static UI plus fixed, local-only ingestion and experiment commands."""
+
+ server_version = "GeoAIWorkbench/1.0"
+
+ @property
+ def root(self) -> Path:
+ return Path(self.directory).resolve()
def do_GET(self) -> None: # noqa: N802 - inherited standard-library method name
- if urlsplit(self.path).path == "/":
+ path = urlsplit(self.path).path
+ if path == "/api/trajectory/runs":
+ self.send_json(HTTPStatus.OK, {"runs": trajectory_runs(self.root)})
+ return
+ if path == "/api/object-detection/runs":
+ self.send_json(HTTPStatus.OK, {"runs": detection_runs(self.root)})
+ return
+ if path == "/":
self.send_response(HTTPStatus.FOUND)
self.send_header("Location", "/apps/workbench-console/")
self.end_headers()
return
super().do_GET()
+ def do_POST(self) -> None: # noqa: N802 - inherited standard-library method name
+ path = urlsplit(self.path).path
+ try:
+ payload = self.read_json_body()
+ if path == "/api/trajectory/runs":
+ self.send_json(HTTPStatus.CREATED, {"run": self.create_trajectory_run(payload)})
+ return
+ if path == "/api/object-detection/runs":
+ self.send_json(HTTPStatus.CREATED, {"run": self.create_detection_run(payload)})
+ return
+ self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."})
+ except ApiError as exc:
+ self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+ except subprocess.TimeoutExpired:
+ self.send_json(HTTPStatus.GATEWAY_TIMEOUT, {"error": "The local run exceeded its time limit; no existing result was overwritten."})
+ except Exception as exc: # pragma: no cover - defensive server boundary
+ self.log_error("local run failed: %s", exc)
+ self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Local run failed. Check the console terminal for details."})
+
+ def do_OPTIONS(self) -> None: # noqa: N802
+ self.send_response(HTTPStatus.NO_CONTENT)
+ self.send_header("Allow", "GET, POST, OPTIONS")
+ self.end_headers()
+
+ def read_json_body(self) -> dict[str, Any]:
+ content_length = self.headers.get("Content-Length")
+ if content_length is None or not content_length.isdigit():
+ raise ApiError("A JSON request body with Content-Length is required.")
+ size = int(content_length)
+ if size <= 0 or size > MAX_REQUEST_BYTES:
+ raise ApiError(f"Request must be between 1 byte and {MAX_REQUEST_BYTES // (1024 * 1024)} MB.")
+ if "application/json" not in self.headers.get("Content-Type", ""):
+ raise ApiError("Content-Type must be application/json.")
+ try:
+ payload = json.loads(self.rfile.read(size).decode("utf-8"))
+ except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+ raise ApiError("Request body is not valid UTF-8 JSON.") from exc
+ if not isinstance(payload, dict):
+ raise ApiError("JSON request body must be an object.")
+ return payload
+
+ def run_command(self, command: list[str], timeout: int) -> None:
+ completed = subprocess.run(command, cwd=self.root, capture_output=True, text=True, timeout=timeout, check=False)
+ if completed.returncode:
+ message = (completed.stderr or completed.stdout or "Unknown script error.").strip().splitlines()[-1]
+ raise ApiError(f"Processing failed: {message[:600]}")
+
+ def create_trajectory_run(self, payload: dict[str, Any]) -> dict[str, Any]:
+ files = payload.get("files")
+ if not isinstance(files, dict):
+ raise ApiError("Trajectory request must contain a files object.")
+ required = {
+ "flight": {".xlsx"},
+ "route": {".kmz"},
+ "restricted": {".geojson"},
+ }
+ decoded = {key: decode_upload(files.get(key), suffixes) for key, suffixes in required.items()}
+ flyable = decode_upload(files["flyable"], {".gzip"}) if files.get("flyable") else None
+ run_id = make_run_id("trajectory")
+ raw_root = self.root / "shared" / "data" / "raw" / "15-trajectory-analysis" / "runs" / run_id
+ paths = {"flight": raw_root / "tracks" / decoded["flight"][0], "route": raw_root / "routes" / decoded["route"][0], "restricted": raw_root / "areas" / decoded["restricted"][0]}
+ for key, path in paths.items():
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_bytes(decoded[key][1])
+ if flyable:
+ flyable_path = raw_root / "areas" / flyable[0]
+ flyable_path.write_bytes(flyable[1])
+ processed = self.root / "shared" / "data" / "processed" / "15-trajectory-analysis" / run_id
+ output_parent = self.root / "shared" / "outputs" / "15-trajectory-analysis" / "runs" / run_id
+ python = self.root / ".venvs" / "15-trajectory-analysis" / "Scripts" / "python.exe"
+ if not python.is_file():
+ raise ApiError("Trajectory virtual environment is unavailable. Run the capability setup first.")
+ with RUN_LOCK:
+ self.run_command([str(python), str(self.root / "capabilities" / "15-trajectory-analysis" / "prepare_real_flight.py"), "--raw-dir", str(raw_root), "--output", str(processed), "--case-id", run_id], 300)
+ self.run_command([str(python), str(self.root / "capabilities" / "15-trajectory-analysis" / "run_trajectory_analysis.py"), "--input", str(processed / f"{run_id}.case.json"), "--output", str(output_parent)], 300)
+ artifact = output_parent / run_id
+ if not (artifact / "run_metadata.json").is_file():
+ raise ApiError("Trajectory script finished without the expected result metadata.")
+ return next(item for item in trajectory_runs(self.root) if item["id"] == run_id)
+
+ def create_detection_run(self, payload: dict[str, Any]) -> dict[str, Any]:
+ uploads = payload.get("images")
+ if not isinstance(uploads, list) or not uploads:
+ raise ApiError("Object-detection request must include at least one image.")
+ if len(uploads) > MAX_IMAGES_PER_RUN:
+ raise ApiError(f"A local run accepts at most {MAX_IMAGES_PER_RUN} images.")
+ decoded = [decode_upload(item, {".jpg", ".jpeg", ".png"}) for item in uploads]
+ if len({name.casefold() for name, _ in decoded}) != len(decoded):
+ raise ApiError("Uploaded image names must be unique within one run.")
+ run_id = make_run_id("detection")
+ raw_root = self.root / "shared" / "data" / "raw" / "01-object-detection" / "runs" / run_id
+ raw_root.mkdir(parents=True, exist_ok=False)
+ for name, content in decoded:
+ (raw_root / name).write_bytes(content)
+ output = self.root / "shared" / "outputs" / "01-object-detection" / "runs" / run_id
+ python = self.root / ".venvs" / "01-object-detection" / "Scripts" / "python.exe"
+ if not python.is_file():
+ raise ApiError("Object-detection virtual environment is unavailable. Run the capability setup first.")
+ with RUN_LOCK:
+ self.run_command([str(python), str(self.root / "capabilities" / "01-object-detection" / "run_detection.py"), "--input", str(raw_root), "--output", str(output)], 1200)
+ if not (output / "run_metadata.json").is_file():
+ raise ApiError("Detection script finished without the expected result metadata.")
+ return next(item for item in detection_runs(self.root) if item["id"] == run_id)
+
+ def send_json(self, status: HTTPStatus, payload: dict[str, Any]) -> None:
+ body = json.dumps(payload, 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(body)))
+ self.end_headers()
+ self.wfile.write(body)
+
def translate_path(self, path: str) -> str:
"""Expose only the static UI and artifacts required by the local console."""
decoded_path = unquote(urlsplit(path).path).lstrip("/")
requested = PurePosixPath(decoded_path)
- is_allowed = any(
- decoded_path == prefix or decoded_path.startswith(f"{prefix}/")
- for prefix in ALLOWED_PATH_PREFIXES
- )
+ is_allowed = any(decoded_path == prefix or decoded_path.startswith(f"{prefix}/") for prefix in ALLOWED_PATH_PREFIXES)
if ".." in requested.parts or not is_allowed:
return os.fspath(Path(self.directory) / ".console-forbidden")
if decoded_path == "apps/workbench-console" or decoded_path.startswith("apps/workbench-console/"):
@@ -71,13 +302,10 @@
app_dir = root / "apps" / "workbench-console"
if not (root / "shared").is_dir() or not (app_dir / "dist" / "index.html").is_file():
raise SystemExit(f"Not a GeoAI Workbench root: {root}")
-
- handler = lambda *handler_args, **handler_kwargs: WorkbenchConsoleHandler( # noqa: E731
- *handler_args, directory=os.fspath(root), **handler_kwargs
- )
+ handler = lambda *handler_args, **handler_kwargs: WorkbenchConsoleHandler(*handler_args, directory=os.fspath(root), **handler_kwargs) # noqa: E731
server = ThreadingHTTPServer((args.host, args.port), handler)
print(f"GeoAI Workbench console: http://{args.host}:{args.port}")
- print(f"Serving built console and read-only artifacts from: {root}")
+ print("Local runs use fixed capability scripts and create a new run directory.")
try:
server.serve_forever()
except KeyboardInterrupt:
--
Gitblit v1.9.3