"""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) delegation_bearer: str = "" 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, delegated: bool = False, 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) and not delegated: raise TaskDataError(403, "当前身份没有该数据源的读取权限") if delegated: credential = config.delegation_bearer if not credential: raise TaskDataError(503, "任务授权已批准，但数据源委托身份尚未配置") if not credential.lower().startswith("bearer "): credential = f"Bearer {credential}" 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, )