178 lines
8.3 KiB
Python
178 lines
8.3 KiB
Python
"""Read-only source adapters with bounded queries and explicit provenance."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from pydantic import Field
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
from app.auth.permissions import SCADA_VIEW, SIMULATION_VIEW, permissions_for_context
|
|
from app.auth.project_dependencies import ProjectContext
|
|
from app.domain.schemas.operational_data import TaskDataQuery, TaskDataSnapshot
|
|
|
|
|
|
class TaskDataSettings(BaseSettings):
|
|
server_url: str = "http://127.0.0.1:8000"
|
|
peripheral_url: str = ""
|
|
timeout_seconds: float = Field(default=15, ge=1, le=60)
|
|
model_config = SettingsConfigDict(
|
|
env_prefix="OPERATIONAL_DATA_", extra="ignore",
|
|
env_file=Path(__file__).resolve().parents[3] / ".env",
|
|
)
|
|
|
|
|
|
class TaskDataError(Exception):
|
|
def __init__(self, status: int, detail: str):
|
|
super().__init__(detail)
|
|
self.status = status
|
|
self.detail = detail
|
|
|
|
|
|
def _matches_project(value: object, project_id: str) -> bool:
|
|
if isinstance(value, list):
|
|
return all(_matches_project(item, project_id) for item in value)
|
|
if isinstance(value, dict):
|
|
if "project_id" in value and value["project_id"] != project_id:
|
|
return False
|
|
return all(_matches_project(item, project_id) for item in value.values())
|
|
return True
|
|
|
|
|
|
def _select_demo_observations(
|
|
payload: object, query: TaskDataQuery, project_id: str,
|
|
) -> dict[str, object]:
|
|
"""Project a bounded DEMO-* observation query from the stored demo dataset."""
|
|
if not isinstance(payload, dict) or not _matches_project(payload, project_id):
|
|
raise TaskDataError(502, "数据源返回了其他项目的数据")
|
|
points = payload.get("points")
|
|
observations = payload.get("observations")
|
|
if not isinstance(points, list) or not isinstance(observations, list):
|
|
raise TaskDataError(502, "外围合成样本格式无效")
|
|
if not any(isinstance(point, dict) and point.get("point_id") == query.point_id for point in points):
|
|
raise TaskDataError(404, "合成样本测点不存在")
|
|
assert query.start is not None and query.end is not None
|
|
items: list[object] = []
|
|
try:
|
|
for item in observations:
|
|
if not isinstance(item, dict) or item.get("point_id") != query.point_id:
|
|
continue
|
|
observed_at = datetime.fromisoformat(str(item.get("observed_at")))
|
|
if query.start <= observed_at < query.end:
|
|
items.append(item)
|
|
except (TypeError, ValueError) as exc:
|
|
raise TaskDataError(502, "外围合成样本格式无效") from exc
|
|
return {
|
|
"origin": "synthetic",
|
|
"source_id": payload.get("source_id"),
|
|
"source_revision": payload.get("source_revision"),
|
|
"point_id": query.point_id,
|
|
"window": {
|
|
"start": query.start.isoformat(),
|
|
"end": query.end.isoformat(),
|
|
},
|
|
"items": items,
|
|
"truncated": False,
|
|
}
|
|
|
|
|
|
async def read_task_data(
|
|
*, context: ProjectContext, query: TaskDataQuery, credential: str | None,
|
|
settings: TaskDataSettings | None = None,
|
|
transport: httpx.AsyncBaseTransport | None = None,
|
|
) -> TaskDataSnapshot:
|
|
config = settings or TaskDataSettings()
|
|
permission = SCADA_VIEW if query.source.startswith("scada_") else SIMULATION_VIEW
|
|
if permission not in permissions_for_context(context):
|
|
raise TaskDataError(403, "当前身份没有该数据源的读取权限")
|
|
if not credential or not credential.lower().startswith("bearer "):
|
|
raise TaskDataError(401, "数据查询需要当前产品身份")
|
|
peripheral = query.source.startswith("peripheral_")
|
|
base = config.peripheral_url if peripheral else config.server_url
|
|
if not base:
|
|
raise TaskDataError(503, "调查数据源尚未配置")
|
|
paths = {
|
|
"workbench_runtime": "/api/v1/workbench/runtime",
|
|
"scada_devices": "/api/v1/scada-devices",
|
|
"scada_readings": "/api/v1/timeseries/scada-readings",
|
|
"peripheral_points": "/api/sample/v1/monitoring/points",
|
|
"peripheral_work_orders": "/api/sample/v1/work-orders",
|
|
"peripheral_observations": "/api/sample/v1/monitoring/observations",
|
|
"peripheral_test_data": "/api/sample/v1/test-data",
|
|
}
|
|
demo_observations = (
|
|
query.source == "peripheral_observations"
|
|
and query.point_id is not None
|
|
and query.point_id.startswith("DEMO-")
|
|
)
|
|
path = "/api/sample/v1/demo-data" if demo_observations else paths[query.source]
|
|
params: dict[str, str | int] = {"limit": 100}
|
|
if query.source == "scada_readings":
|
|
params.update(device_ids=",".join(query.device_ids), start_time=query.start.isoformat(), end_time=query.end.isoformat())
|
|
elif query.source == "peripheral_observations" and not demo_observations:
|
|
params = {"point_id": query.point_id, "start": query.start.isoformat(), "end": query.end.isoformat()}
|
|
elif demo_observations:
|
|
params = {}
|
|
try:
|
|
async with httpx.AsyncClient(
|
|
base_url=base.rstrip("/"), timeout=config.timeout_seconds, follow_redirects=False,
|
|
trust_env=False, transport=transport,
|
|
) as client:
|
|
async with client.stream("GET", path, params=params, headers={
|
|
"Authorization": credential, "X-Project-Id": str(context.project_id),
|
|
}) as response:
|
|
if response.status_code in {401, 403}:
|
|
raise TaskDataError(response.status_code, "数据源拒绝当前身份或项目权限")
|
|
response.raise_for_status()
|
|
body = bytearray()
|
|
async for chunk in response.aiter_bytes():
|
|
body.extend(chunk)
|
|
if len(body) > 2_000_000:
|
|
raise TaskDataError(422, "查询结果过大，请缩短时间范围或减少设备数量")
|
|
data = json.loads(body)
|
|
except (httpx.HTTPError, ValueError) as exc:
|
|
raise TaskDataError(502, "调查数据查询失败，不能据此形成已核实结论") from exc
|
|
if demo_observations:
|
|
data = _select_demo_observations(data, query, str(context.project_id))
|
|
elif not _matches_project(data, str(context.project_id)):
|
|
raise TaskDataError(502, "数据源返回了其他项目的数据")
|
|
origin = "synthetic" if peripheral else "system_data"
|
|
historical_replay = False
|
|
if query.source in {"peripheral_points", "peripheral_observations"}:
|
|
source_origin = data.get("origin") if isinstance(data, dict) else None
|
|
if source_origin not in {"historical_replay", "synthetic"}:
|
|
raise TaskDataError(502, "监测数据缺少明确来源，不能作为已核实证据")
|
|
historical_replay = source_origin == "historical_replay"
|
|
origin = "system_data" if historical_replay else "synthetic"
|
|
truncated = isinstance(data, dict) and bool(data.get("truncated") or data.get("next_cursor") or data.get("next_offset"))
|
|
if isinstance(data, list) and len(data) > 1000:
|
|
data, truncated = data[:1000], True
|
|
if isinstance(data, dict) and isinstance(data.get("items"), list):
|
|
if len(data["items"]) > 1000:
|
|
data = {**data, "items": data["items"][:1000]}
|
|
truncated = True
|
|
if isinstance(data.get("total"), int) and data["total"] > len(data["items"]):
|
|
truncated = True
|
|
# The digest identifies the returned snapshot, not merely a mutable URL.
|
|
digest = hashlib.sha256(json.dumps(data, ensure_ascii=False, sort_keys=True).encode()).hexdigest()
|
|
if demo_observations:
|
|
assert query.start is not None and query.end is not None
|
|
source_ref = (
|
|
f"peripheral_demo_data:{query.point_id}:"
|
|
f"{query.start.isoformat()}:{query.end.isoformat()}:sha256:{digest}"
|
|
)
|
|
elif historical_replay:
|
|
# Use a distinct identity: old snapshots may have incorrectly labeled
|
|
# this same historical response synthetic. Immutable evidence stays intact.
|
|
source_ref = f"peripheral_historical_replay:{query.source}:sha256:{digest}"
|
|
else:
|
|
source_ref = f"{query.source}:sha256:{digest}"
|
|
return TaskDataSnapshot(
|
|
source=query.source, source_ref=source_ref,
|
|
queried_at=datetime.now(timezone.utc), origin=origin,
|
|
data=data, truncated=truncated,
|
|
)
|