fix(sensor-placement): enforce project write boundaries

Bind every scheme request to ProjectContext, keep viewer access read-only, reject concurrent optimization jobs without blocking worker threads, and cap export/update payload sizes. Run optimization and workbook work off the event loop.

app/api/v1/endpoints/sensor_placement.py

@@ -41,6 +41,25 @@ def _project_network(network: str, project_context: ProjectContext) -> str:
     return project_context.project_code
 
 
+def _can_modify_project(project_context: ProjectContext, current_user: Any) -> bool:
+    return bool(
+        project_context.project_role in {"owner", "admin", "member"}
+        or getattr(current_user, "role", None) == "admin"
+        or getattr(current_user, "is_superuser", False)
+    )
+
+
+def _require_project_write(
+    project_context: ProjectContext,
+    current_user: Any,
+) -> None:
+    if not _can_modify_project(project_context, current_user):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="当前项目角色为只读，不能修改监测点方案",
+        )
+
+
 def _service_http_error(exc: Exception) -> HTTPException:
     if isinstance(exc, SensorPlacementNotFoundError):
         return HTTPException(
@@ -62,12 +81,16 @@ def _get_scheme_response(
     network: str,
     scheme_id: int,
     current_user: Any,
+    project_context: ProjectContext,
 ) -> dict[str, Any]:
     try:
         scheme = get_sensor_placement_scheme(network, scheme_id)
         return {
             **scheme,
-            "can_edit": can_edit_sensor_placement(current_user, scheme),
+            "can_edit": (
+                _can_modify_project(project_context, current_user)
+                and can_edit_sensor_placement(current_user, scheme)
+            ),
         }
     except (
         SensorPlacementNotFoundError,
@@ -87,6 +110,7 @@ async def optimize_sensor_placement_scheme(
     current_user=Depends(get_current_metadata_user),
 ) -> dict[str, Any]:
     network = _project_network(payload.network, project_context)
+    _require_project_write(project_context, current_user)
     optimizer = (
         pressure_sensor_placement_sensitivity
         if payload.method == "sensitivity"
@@ -103,7 +127,11 @@ async def optimize_sensor_placement_scheme(
         )
         scheme = get_sensor_placement_scheme(network, int(created["id"]))
         return {**scheme, "can_edit": True}
-    except (SensorPlacementValidationError, ValueError) as exc:
+    except (
+        SensorPlacementConflictError,
+        SensorPlacementValidationError,
+        ValueError,
+    ) as exc:
         raise _service_http_error(exc) from exc
     except Exception as exc:
         logger.exception("Sensor placement optimization failed")
@@ -128,6 +156,7 @@ async def get_sensor_placement_scheme_detail(
         _project_network(network, project_context),
         scheme_id,
         current_user,
+        project_context,
     )
 
 
@@ -144,7 +173,13 @@ async def overwrite_sensor_placement_scheme(
     current_user=Depends(get_current_metadata_user),
 ) -> dict[str, Any]:
     network = _project_network(network, project_context)
-    scheme = _get_scheme_response(network, scheme_id, current_user)
+    _require_project_write(project_context, current_user)
+    scheme = _get_scheme_response(
+        network,
+        scheme_id,
+        current_user,
+        project_context,
+    )
     if not scheme["can_edit"]:
         raise HTTPException(status_code=403, detail="无权修改该监测点方案")
 
@@ -176,7 +211,12 @@ async def export_sensor_placement_excel(
     current_user=Depends(get_current_metadata_user),
 ) -> StreamingResponse:
     network = _project_network(network, project_context)
-    scheme = _get_scheme_response(network, scheme_id, current_user)
+    scheme = _get_scheme_response(
+        network,
+        scheme_id,
+        current_user,
+        project_context,
+    )
     if (
         payload.sensor_location != scheme["sensor_location"]
         and not scheme["can_edit"]
@@ -184,7 +224,8 @@ async def export_sensor_placement_excel(
         raise HTTPException(status_code=403, detail="无权导出该方案的未保存草稿")
 
     try:
-        workbook = build_sensor_placement_workbook(
+        workbook = await run_in_threadpool(
+            build_sensor_placement_workbook,
             network=network,
             scheme=scheme,
             sensor_location=payload.sensor_location,