feat(lab): seal full RAVNOVES004TREE semantic review
This commit is contained in:
@@ -179,9 +179,77 @@ def _build_vegetation_lab_router(
|
||||
},
|
||||
)
|
||||
|
||||
@router.get("/{result_id}/route-masks/{layer}/{sequence}")
|
||||
def get_full_route_mask(result_id: str, layer: str, sequence: int) -> Response:
|
||||
candidate = _resolve_candidate(root_provider, definition, result_id)
|
||||
manifest = _read_verified(candidate, definition)
|
||||
route = manifest.get("route_full_review")
|
||||
layers = route.get("layers") if isinstance(route, dict) else None
|
||||
frame_count = route.get("frame_count") if isinstance(route, dict) else None
|
||||
selected = layers.get(layer) if isinstance(layers, dict) else None
|
||||
archive = selected.get("mask_archive") if isinstance(selected, dict) else None
|
||||
archive_relative = archive.get("path") if isinstance(archive, dict) else None
|
||||
if (
|
||||
layer not in {"city", "vegetation"}
|
||||
or not isinstance(frame_count, int)
|
||||
or not 0 <= sequence < frame_count
|
||||
or not isinstance(archive_relative, str)
|
||||
):
|
||||
raise HTTPException(status_code=404, detail="Full-route semantic mask not found")
|
||||
relative = PurePosixPath(archive_relative)
|
||||
artifacts = manifest.get("artifacts")
|
||||
if (
|
||||
relative.is_absolute()
|
||||
or str(relative) != archive_relative
|
||||
or any(part in {"", ".", ".."} for part in relative.parts)
|
||||
or relative.suffix != ".zip"
|
||||
or not isinstance(artifacts, list)
|
||||
or not any(
|
||||
isinstance(item, dict)
|
||||
and item.get("path") == archive_relative
|
||||
and item.get("media_type") == "application/zip"
|
||||
for item in artifacts
|
||||
)
|
||||
):
|
||||
raise HTTPException(status_code=404, detail="Full-route semantic mask not found")
|
||||
return _zip_mask_response(candidate.joinpath(*relative.parts), sequence)
|
||||
|
||||
return router
|
||||
|
||||
|
||||
def _zip_mask_response(archive_path: Path, sequence: int) -> Response:
|
||||
member = f"masks/frame-{sequence + 1:06d}.png"
|
||||
try:
|
||||
before = archive_path.stat()
|
||||
with zipfile.ZipFile(archive_path) as frozen:
|
||||
info = frozen.getinfo(member)
|
||||
if info.is_dir() or info.file_size < 8 or info.file_size > 1024 * 1024:
|
||||
raise ValueError("Semantic mask member is invalid")
|
||||
payload = frozen.read(info)
|
||||
after = archive_path.stat()
|
||||
if (
|
||||
before.st_size != after.st_size
|
||||
or before.st_mtime_ns != after.st_mtime_ns
|
||||
or len(payload) != info.file_size
|
||||
):
|
||||
raise ValueError("Semantic mask archive changed during read")
|
||||
except (KeyError, OSError, ValueError, zipfile.BadZipFile):
|
||||
raise HTTPException(
|
||||
status_code=503,
|
||||
detail="Semantic mask failed verification",
|
||||
) from None
|
||||
digest = hashlib.sha256(payload).hexdigest()
|
||||
return Response(
|
||||
content=payload,
|
||||
media_type="image/png",
|
||||
headers={
|
||||
"Cache-Control": "private, max-age=31536000, immutable",
|
||||
"ETag": f'"{digest}"',
|
||||
"X-Content-Type-Options": "nosniff",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _configured_root(provider: RootProvider) -> Path | None:
|
||||
candidate = provider()
|
||||
if candidate is None:
|
||||
|
||||
Reference in New Issue
Block a user