127 lines
4.0 KiB
Python
127 lines
4.0 KiB
Python
# topology_api/routes/snapshot.py
|
|
from datetime import datetime
|
|
|
|
from flask import Blueprint, request, jsonify
|
|
from topology_api.queries.snapshot_queries import (
|
|
resolve_snapshot_time, fetch_snapshot, fetch_layout, fetch_missions,
|
|
)
|
|
|
|
snapshot_bp = Blueprint("snapshot", __name__)
|
|
GHC06_LAYOUT_START = datetime.fromisoformat("2025-02-15T00:00:00+00:00")
|
|
|
|
|
|
def _layout_key_for_snapshot(requested_mission, snapshot_at):
|
|
"""Keep historical mission views pinned to their contemporary layout.
|
|
|
|
GHC-05 overlaps GHC-06 chronologically, so timestamp-only selection would
|
|
incorrectly redraw the GHC-05 terminal snapshot with the later expansion
|
|
layout. Explicit mission views therefore take precedence over time.
|
|
"""
|
|
if requested_mission:
|
|
try:
|
|
mission_number = int(requested_mission.rsplit("-", 1)[-1])
|
|
except (TypeError, ValueError):
|
|
mission_number = None
|
|
return (
|
|
"engineering_overview_ghc06_v1"
|
|
if mission_number is not None and mission_number >= 6
|
|
else None
|
|
)
|
|
|
|
snapshot_dt = datetime.fromisoformat(snapshot_at.replace("Z", "+00:00"))
|
|
return (
|
|
"engineering_overview_ghc06_v1"
|
|
if snapshot_dt >= GHC06_LAYOUT_START
|
|
else None
|
|
)
|
|
|
|
|
|
def _parse_iso_timestamp(value):
|
|
if value is None:
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
return None
|
|
return parsed.isoformat()
|
|
|
|
|
|
@snapshot_bp.route("/api/v1/bases/<base_code>/missions")
|
|
def get_missions(base_code):
|
|
if base_code != "guanghan":
|
|
return jsonify({
|
|
"error": "base_not_found",
|
|
"message": f"Base '{base_code}' not found.",
|
|
}), 404
|
|
return jsonify({"missions": fetch_missions(base_code)})
|
|
|
|
|
|
@snapshot_bp.route("/api/v1/bases/<base_code>/snapshot")
|
|
def get_snapshot(base_code):
|
|
if base_code != "guanghan":
|
|
return jsonify({
|
|
"error": "base_not_found",
|
|
"message": f"Base '{base_code}' not found. Only 'guanghan' is available.",
|
|
}), 404
|
|
|
|
requested_mission = request.args.get("mission")
|
|
requested_at = request.args.get("at")
|
|
normalized_at = _parse_iso_timestamp(requested_at)
|
|
|
|
if requested_at is not None and normalized_at is None:
|
|
return jsonify({
|
|
"error": "invalid_time",
|
|
"message": (
|
|
"Parameter 'at' must be an ISO 8601 timestamp with a timezone."
|
|
),
|
|
}), 400
|
|
|
|
snapshot_at, anchor_source = resolve_snapshot_time(
|
|
requested_mission=requested_mission,
|
|
requested_at=normalized_at,
|
|
)
|
|
|
|
if anchor_source == "mission_not_found":
|
|
return jsonify({
|
|
"error": "mission_not_found",
|
|
"message": f"Mission '{requested_mission}' not found.",
|
|
}), 404
|
|
|
|
if anchor_source == "mission_has_no_timed_event":
|
|
return jsonify({
|
|
"error": "mission_has_no_timed_event",
|
|
"message": f"Mission '{requested_mission}' has no timed events.",
|
|
}), 409
|
|
|
|
if anchor_source == "no_events":
|
|
return jsonify({
|
|
"base": {"code": "guanghan", "name": "广寒基地"},
|
|
"query": {
|
|
"requestedAt": requested_at,
|
|
"requestedMission": requested_mission,
|
|
"snapshotAt": None,
|
|
"anchorSource": "no_events",
|
|
},
|
|
"components": [],
|
|
"connections": [],
|
|
})
|
|
|
|
data = fetch_snapshot(snapshot_at)
|
|
layout_key = _layout_key_for_snapshot(requested_mission, snapshot_at)
|
|
layout = fetch_layout(layout_key=layout_key)
|
|
|
|
return jsonify({
|
|
"base": data["base"],
|
|
"query": {
|
|
"requestedAt": requested_at,
|
|
"requestedMission": requested_mission,
|
|
"snapshotAt": snapshot_at,
|
|
"anchorSource": anchor_source,
|
|
},
|
|
"components": data["components"],
|
|
"connections": data["connections"],
|
|
"layout": layout,
|
|
})
|