feat:项目基础整理

shuishen

2 days ago 4092854b0bea9e1fc02f29222c6f7cd876e565a5

 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: