refactor: simplify task and history services

Remove retired query paths and repository compatibility branches, and share task test fixtures. Cancel pending questions and resolve their notifications when recovering expired runs.
2026-11-09 17:55:00 +08:00
parent 5197a93163
commit 6a4478bfe4
10 changed files with 464 additions and 816 deletions
@@ -3,7 +3,7 @@ from __future__ import annotations

 import json
 from datetime import datetime, timedelta, timezone
-from typing import Iterable
+from typing import TYPE_CHECKING, Iterable
 from uuid import NAMESPACE_URL, UUID, uuid4, uuid5

 from app.auth.permissions import (
@@ -77,6 +77,10 @@ from app.services.operational_scada_coverage import require_complete_scada_evide
 from app.services.time_comparison_chart import validate_time_comparison_chart


+if TYPE_CHECKING:
+    from app.infra.db.postgresql.operational_tasks import OperationalTaskRepository
+
+
 LEASE_DURATION = timedelta(seconds=60)


@@ -89,7 +93,7 @@ def _task_json(model: OperationalTaskSummary) -> dict:


 class OperationalTaskService:
-    def __init__(self, repository, context, permissions: Iterable[str], *, clock=None,
+    def __init__(self, repository: OperationalTaskRepository, context, permissions: Iterable[str], *, clock=None,
                 execution_scope: str = "production"):
        self.repo = repository
        self.context = context
@@ -114,42 +118,14 @@ class OperationalTaskService:
            require(owner == command.agent_instance_id,
                    "AGENT_INSTANCE_MISMATCH", "Agent instance does not own this run", status=409)

-    @property
-    def _supports_runs(self) -> bool:
-        return hasattr(self.repo, "get_active_run") and hasattr(self.repo, "update_run")
-
    async def _active_run(self, task: dict) -> dict | None:
-        if self._supports_runs:
-            return await self.repo.get_active_run(str(task["task_id"]))
-        if task.get("status") not in {"queued", "running"}:
-            return None
-        now = self._now()
-        return _json(OperationalTaskRun(
-            run_id=uuid5(NAMESPACE_URL, f"legacy-run:{task['task_id']}:{task['execution_generation']}"),
-            task_id=task["task_id"],
-            generation=task["execution_generation"],
-            execution_request_id=task["execution_request_id"],
-            status="running" if task["status"] == "running" else "queued",
-            revision=task["revision"],
-            agent_session_id=task.get("agent_session_id"),
-            lease_expires_at=task.get("agent_lease_expires_at"),
-            heartbeat_at=task.get("updated_at") if task["status"] == "running" else None,
-            started_at=task.get("created_at") if task["status"] == "running" else None,
-            dispatch_attempts=0,
-            next_dispatch_at=task.get("created_at") or now,
-            created_at=task.get("created_at") or now,
-            updated_at=task.get("updated_at") or now,
-        ))
+        return await self.repo.get_active_run(str(task["task_id"]))

-    async def _append_event(self, **kwargs) -> int | None:
-        append_event = getattr(self.repo, "append_event", None)
-        return await append_event(**kwargs) if append_event is not None else None
+    async def _append_event(self, **kwargs) -> int:
+        return await self.repo.append_event(**kwargs)

    async def _list_access_requests(self, task_id: UUID | str) -> list[dict]:
-        list_access_requests = getattr(self.repo, "list_access_requests", None)
-        if list_access_requests is None:
-            return []
-        return await list_access_requests(str(task_id))
+        return await self.repo.list_access_requests(str(task_id))

    def _permission_view(self) -> OperationalTaskPermissions:
        permissions = self.permissions
@@ -286,8 +262,7 @@ class OperationalTaskService:
        actions = await self.repo.list_actions(str(task_id))
        access_requests = await self._list_access_requests(task_id)
        active_run = await self._active_run(task)
-        latest_run = (await self.repo.get_latest_run(str(task_id))
-                      if hasattr(self.repo, "get_latest_run") else active_run)
+        latest_run = await self.repo.get_latest_run(str(task_id))
        progress = await self.repo.get_latest_progress(str(task_id), task["execution_generation"])
        work_order_snapshot = None
        if task["task_type"] == "external_work_order" and task["source"].get("external_id"):
@@ -397,9 +372,8 @@ class OperationalTaskService:
            replay = await self._request_replay(scope, command.request_id, body)
            if replay:
                return TaskAccessRequest.model_validate(replay["access_request"])
-            availability = getattr(self.repo, "access_requests_available", None)
            require(
-                availability is None or await availability(),
+                await self.repo.access_requests_available(),
                "DEPENDENCY_UNAVAILABLE",
                "Task data access requests are not enabled for this project database",
                status=503,
@@ -616,8 +590,7 @@ class OperationalTaskService:
        return OperationalTaskRun.model_validate(await self._run(task_id, run_id))

    async def _fail_run_messages(self, run_id, reason: str, now: datetime) -> None:
-        if hasattr(self.repo, "fail_run_messages"):
-            await self.repo.fail_run_messages(str(run_id), reason, now)
+        await self.repo.fail_run_messages(str(run_id), reason, now)

    async def fail_run(
        self, task_id: UUID, run_id: UUID, command: AgentRunFailureCommand
@@ -772,7 +745,7 @@ class OperationalTaskService:
                # Lease recovery already holds a row lock. Try rather than wait
                # for the shared run lock to avoid reversing the lock order of
                # message/finish transactions; a busy run is checked next tick.
-                if hasattr(self.repo, "try_lock") and not await self.repo.try_lock(f"run:{run['run_id']}"):
+                if not await self.repo.try_lock(f"run:{run['run_id']}"):
                    continue
                task_id = UUID(str(run["task_id"]))
                expected = run["revision"]
@@ -784,6 +757,7 @@ class OperationalTaskService:
                timed_out = OperationalTaskRun.model_validate(run)
                await self.repo.update_run(_json(timed_out), expected)
                await self._fail_run_messages(run["run_id"], "Run lease expired", now)
+                await task_questions.cancel_pending(self, task_id, now)
                await self._run_event(timed_out, now)
                task = await self._task(task_id)
                timeout_count = await self.repo.consecutive_timed_out_runs(str(task_id))
@@ -1017,25 +991,9 @@ class OperationalTaskService:
                lease_expires_at=now + LEASE_DURATION,
                updated_at=now,
            )
-            if self._supports_runs:
-                await self.repo.update_run(
-                    _json(OperationalTaskRun.model_validate(active_run)), run_expected
-                )
-            else:
-                # Compatibility for pre-015 repository adapters only.
-                if command.report:
-                    if task["task_type"] == "event_investigation":
-                        task["status"] = self._investigation_status(actions.values(), task) if command.report.stage == "final" else "running"
-                    else:
-                        task["status"] = self._status_from_plan(current_plan.plan.nodes, actions.values()) if current_plan else "running"
-                    if task["status"] in {"waiting_human", "completed", "failed", "cancelled"}:
-                        task["lifecycle_status"] = task["status"]
-                    task["agent_lease_expires_at"] = (
-                        now + LEASE_DURATION if task["status"] == "running" else None
-                    )
-                    await self.repo.update_task(
-                        _task_json(OperationalTaskSummary.model_validate(task))
-                    )
+            await self.repo.update_run(
+                _json(OperationalTaskRun.model_validate(active_run)), run_expected
+            )
            await self._task_event(task, now)
            response = {"plan_revision": plan_revision, "report_revision": report_revision}
            if progress_revision is not None:
@@ -1988,13 +1946,12 @@ class OperationalTaskService:
        return DispatchDueSchedulesResult(task_ids=task_ids, dispatched_at=now)

    async def _require_no_pending_calculations(self, task_id: UUID, run: dict) -> None:
-        if hasattr(self.repo, "pending_calculation_jobs"):
-            pending = await self.repo.pending_calculation_jobs(
-                str(task_id), str(run["run_id"]), str(run["execution_request_id"]),
-            )
-            require(not pending, "PENDING_CALCULATIONS",
-                    "Wait for submitted calculations before publishing a final report or completing the run",
-                    refs=pending, status=409, recovery="retry_same_request")
+        pending = await self.repo.pending_calculation_jobs(
+            str(task_id), str(run["run_id"]), str(run["execution_request_id"]),
+        )
+        require(not pending, "PENDING_CALCULATIONS",
+                "Wait for submitted calculations before publishing a final report or completing the run",
+                refs=pending, status=409, recovery="retry_same_request")

    async def _finish_run(
        self, task_id: UUID, run_id: UUID, command: AgentRunCommand,
@@ -2024,11 +1981,10 @@ class OperationalTaskService:
            actions = []
            if outcome == "succeeded":
                await self._require_no_pending_calculations(task_id, run)
-                if hasattr(self.repo, "pending_run_messages"):
-                    pending_messages = await self.repo.pending_run_messages(str(run_id))
-                    require(not pending_messages, "PENDING_AGENT_MESSAGES",
-                            "Run has messages not yet entered into Agent context",
-                            status=409, recovery="retry_same_request", refs=pending_messages)
+                pending_messages = await self.repo.pending_run_messages(str(run_id))
+                require(not pending_messages, "PENDING_AGENT_MESSAGES",
+                        "Run has messages not yet entered into Agent context",
+                        status=409, recovery="retry_same_request", refs=pending_messages)
                plan_document = await self.repo.get_latest_plan(str(task_id))
                report_document = await self.repo.get_latest_report(str(task_id))
                plan = (
@@ -2158,8 +2114,6 @@ class OperationalTaskService:
        from app.infra.db.postgresql.task_alerts import TaskAlertRepository
        from app.services.task_alerts import TaskAlertService

-        if not hasattr(self.repo, "list_investigation_alerts"):
-            return
        for alert in await self.repo.list_investigation_alerts(task_id):
            command = TaskAlertEvent.model_validate({
                **{key: value for key, value in alert.items() if key not in {"status", "updated_at", "revision"}},
@@ -2364,42 +2318,41 @@ class OperationalTaskService:

    async def _requeue_task(self, task: dict, now: datetime) -> None:
        await task_questions.cancel_pending(self, UUID(str(task["task_id"])), now)
-        if self._supports_runs:
-            active_run = await self.repo.get_active_run(str(task["task_id"]))
-            while active_run is not None:
-                await self.repo.lock(f"run:{active_run['run_id']}")
-                # Completion may have won the run lock after our first read.
-                refreshed_run = await self.repo.get_active_run(str(task["task_id"]))
-                if refreshed_run is None or str(refreshed_run["run_id"]) == str(active_run["run_id"]):
-                    active_run = refreshed_run
-                    break
+        active_run = await self.repo.get_active_run(str(task["task_id"]))
+        while active_run is not None:
+            await self.repo.lock(f"run:{active_run['run_id']}")
+            # Completion may have won the run lock after our first read.
+            refreshed_run = await self.repo.get_active_run(str(task["task_id"]))
+            if refreshed_run is None or str(refreshed_run["run_id"]) == str(active_run["run_id"]):
                active_run = refreshed_run
-            if active_run is not None and active_run["status"] == "queued":
-                # Multiple inputs can arrive before the worker claims the next
-                # generation. They are all visible when that queued run starts,
-                # so coalesce them instead of violating the one-active-run rule.
-                task.update(
-                    lifecycle_status="open", status="queued",
-                    revision=task["revision"] + 1,
-                    execution_generation=active_run["generation"],
-                    execution_request_id=str(active_run["execution_request_id"]),
-                    agent_session_id=None, agent_lease_expires_at=None,
-                    updated_at=now,
-                )
-                return
-            if active_run is not None:
-                expected = active_run["revision"]
-                active_run.update(
-                    status="cancelled", revision=expected + 1,
-                    lease_expires_at=None, finished_at=now,
-                    failure_code="SUPERSEDED_BY_REQUEUE",
-                    failure_detail="New task input superseded the active run",
-                    updated_at=now,
-                )
-                cancelled = OperationalTaskRun.model_validate(active_run)
-                await self.repo.update_run(_json(cancelled), expected)
-                await self._fail_run_messages(active_run["run_id"], "Run superseded by requeue", now)
-                await self._run_event(cancelled, now)
+                break
+            active_run = refreshed_run
+        if active_run is not None and active_run["status"] == "queued":
+            # Multiple inputs can arrive before the worker claims the next
+            # generation. They are all visible when that queued run starts,
+            # so coalesce them instead of violating the one-active-run rule.
+            task.update(
+                lifecycle_status="open", status="queued",
+                revision=task["revision"] + 1,
+                execution_generation=active_run["generation"],
+                execution_request_id=str(active_run["execution_request_id"]),
+                agent_session_id=None, agent_lease_expires_at=None,
+                updated_at=now,
+            )
+            return
+        if active_run is not None:
+            expected = active_run["revision"]
+            active_run.update(
+                status="cancelled", revision=expected + 1,
+                lease_expires_at=None, finished_at=now,
+                failure_code="SUPERSEDED_BY_REQUEUE",
+                failure_detail="New task input superseded the active run",
+                updated_at=now,
+            )
+            cancelled = OperationalTaskRun.model_validate(active_run)
+            await self.repo.update_run(_json(cancelled), expected)
+            await self._fail_run_messages(active_run["run_id"], "Run superseded by requeue", now)
+            await self._run_event(cancelled, now)
        generation = task["execution_generation"] + 1
        task_id = UUID(task["task_id"])
        task.update(
@@ -2409,9 +2362,8 @@ class OperationalTaskService:
            agent_session_id=None, agent_lease_expires_at=None, updated_at=now,
        )
        run = self._new_run(task_id, generation, now)
-        if self._supports_runs:
-            await self.repo.insert_run(_json(run))
-            await self._run_event(run, now, queued=True)
+        await self.repo.insert_run(_json(run))
+        await self._run_event(run, now, queued=True)

    async def _notify(
        self, task_id: UUID, kind: str, title: str, description: str, now: datetime,
@@ -2447,10 +2399,7 @@ class OperationalTaskService:
        now: datetime,
        reason: str,
    ) -> None:
-        resolve = getattr(self.repo, "resolve_notifications", None)
-        if resolve is None:
-            return
-        notifications = await resolve(
+        notifications = await self.repo.resolve_notifications(
            str(task_id),
            kinds=kinds,
            subject_type=subject_type,
@@ -13,9 +13,7 @@ if TYPE_CHECKING:


 async def list_questions(service: "OperationalTaskService", task_id: UUID) -> list[dict]:
-    # Legacy test repositories do not implement question persistence.
-    reader = getattr(service.repo, "list_questions", None)
-    return await reader(str(task_id)) if reader else []
+    return await service.repo.list_questions(str(task_id))


 async def cancel_pending(service: "OperationalTaskService", task_id: UUID, now: datetime) -> None:
@@ -1,11 +1,10 @@
 import time
 from datetime import datetime, timedelta
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List

 import numpy as np
 import pandas as pd
 from psycopg import AsyncConnection
-from uuid import UUID

 from app.algorithms.pipe_health_prediction.survival_predictor import (
    PipeHealthSurvivalPredictor,
@@ -15,14 +14,11 @@ from app.algorithms.scada_cleaning.pressure_series import clean_pressure_data_df
 from app.infra.db.postgresql.network_assets import NetworkAssetRepository
 from app.infra.db.postgresql.scada import ScadaInfoRepository
 from app.infra.db.timescaledb.repositories.realtime import RealtimeRepository
-from app.infra.db.timescaledb.repositories.analysis import AnalysisResultsRepository
 from app.infra.db.timescaledb.repositories.scada import ScadaRepository


 class TimeseriesAnalysisService:
-    """
-    复合查询类，提供跨表查询功能
-    """
+    """SCADA 清洗与管道健康预测；历史查询由 TimeseriesHistoryService 负责。"""

    @staticmethod
    async def _get_project_scada_index(
@@ -31,294 +27,6 @@ class TimeseriesAnalysisService:
        scadas = await ScadaInfoRepository.get_scadas(postgres_conn)
        return {scada["device_id"]: scada for scada in scadas}

-    @staticmethod
-    async def get_scada_associated_realtime_simulation_data(
-        timescale_conn: AsyncConnection,
-        postgres_conn: AsyncConnection,
-        device_ids: List[str],
-        start_time: datetime,
-        end_time: datetime,
-    ) -> Dict[str, List[Dict[str, Any]]]:
-        """
-        获取 SCADA 关联的 link/node 模拟值
-
-        根据传入的 SCADA device_ids，找到关联的 link/node，
-        并根据对应的 type，查询对应的模拟数据
-
-        Args:
-            timescale_conn: TimescaleDB 异步连接
-            postgres_conn: PostgreSQL 异步连接
-            device_ids: SCADA 设备ID列表
-            start_time: 开始时间
-            end_time: 结束时间
-
-        Returns:
-            模拟数据字典，以 device_id 为键，值为数据列表，每个数据包含 time, value 和 scada_id
-
-        Raises:
-            ValueError: 当 SCADA 设备未找到或字段无效时
-        """
-        scada_by_id = await TimeseriesAnalysisService._get_project_scada_index(postgres_conn)
-        link_devices: dict[str, str] = {}
-        node_devices: dict[str, str] = {}
-        for device_id in device_ids:
-            target_scada = scada_by_id.get(device_id)
-            if not target_scada:
-                raise ValueError(f"SCADA device {device_id} not found")
-            scada_type = target_scada["device_type"]
-            if scada_type in {"pipe_flow", "flow"}:
-                link_devices[device_id] = target_scada["link_id"]
-            elif scada_type == "pressure":
-                node_devices[device_id] = target_scada["node_id"]
-            else:
-                raise ValueError(f"Unknown SCADA type: {scada_type}")
-
-        link_series = await RealtimeRepository.get_link_fields_by_ids_time_range(
-            timescale_conn, start_time, end_time,
-            list(dict.fromkeys(link_devices.values())), "flow",
-        )
-        node_series = await RealtimeRepository.get_node_fields_by_ids_time_range(
-            timescale_conn, start_time, end_time,
-            list(dict.fromkeys(node_devices.values())), "pressure",
-        )
-        return {
-            device_id: [
-                {**item, "scada_id": device_id}
-                for item in (
-                    link_series.get(element_id, [])
-                    if device_id in link_devices
-                    else node_series.get(element_id, [])
-                )
-            ]
-            for device_id, element_id in (link_devices | node_devices).items()
-        }
-
-    @staticmethod
-    async def get_scada_associated_analysis_simulation_data(
-        timescale_conn: AsyncConnection,
-        postgres_conn: AsyncConnection,
-        device_ids: List[str],
-        start_time: datetime,
-        end_time: datetime,
-        run_id: UUID,
-    ) -> Dict[str, List[Dict[str, Any]]]:
-        """
-        获取 SCADA 关联的 link/node 分析模拟值
-
-        根据传入的 SCADA device_ids，找到关联的 link/node，
-        并根据对应的 type，查询对应的模拟数据
-
-        Args:
-            timescale_conn: TimescaleDB 异步连接
-            postgres_conn: PostgreSQL 异步连接
-            device_ids: SCADA 设备ID列表
-            start_time: 开始时间
-            end_time: 结束时间
-
-        Returns:
-            模拟数据字典，以 device_id 为键，值为数据列表，每个数据包含 time, value 和 scada_id
-
-        Raises:
-            ValueError: 当 SCADA 设备未找到或字段无效时
-        """
-        scada_by_id = await TimeseriesAnalysisService._get_project_scada_index(postgres_conn)
-        link_devices: dict[str, str] = {}
-        node_devices: dict[str, str] = {}
-        for device_id in device_ids:
-            target_scada = scada_by_id.get(device_id)
-            if not target_scada:
-                raise ValueError(f"SCADA device {device_id} not found")
-
-            scada_type = target_scada["device_type"]
-            if scada_type in {"pipe_flow", "flow"}:
-                link_devices[device_id] = target_scada["link_id"]
-            elif scada_type == "pressure":
-                node_devices[device_id] = target_scada["node_id"]
-            else:
-                raise ValueError(f"Unknown SCADA type: {scada_type}")
-
-        link_series = await AnalysisResultsRepository.get_series_by_ids(
-            timescale_conn, run_id, "link",
-            list(dict.fromkeys(link_devices.values())), start_time, end_time, "flow",
-        )
-        node_series = await AnalysisResultsRepository.get_series_by_ids(
-            timescale_conn, run_id, "node",
-            list(dict.fromkeys(node_devices.values())), start_time, end_time, "pressure",
-        )
-        return {
-            device_id: [
-                {**item, "scada_id": device_id}
-                for item in (
-                    link_series.get(element_id, [])
-                    if device_id in link_devices
-                    else node_series.get(element_id, [])
-                )
-            ]
-            for device_id, element_id in (link_devices | node_devices).items()
-        }
-
-    @staticmethod
-    async def get_realtime_simulation_data(
-        timescale_conn: AsyncConnection,
-        feature_infos: List[Tuple[str, str]],
-        start_time: datetime,
-        end_time: datetime,
-    ) -> Dict[str, List[Dict[str, Any]]]:
-        """
-        获取 link/node 模拟值
-
-        根据传入的 feature_infos，找到关联的 link/node，
-        并根据对应的 type，查询对应的模拟数据
-
-        Args:
-            timescale_conn: TimescaleDB 异步连接
-            feature_infos: 传入的 feature 信息列表，包含 (element_id, type)
-            start_time: 开始时间
-            end_time: 结束时间
-
-        Returns:
-            模拟数据字典，以 feature_id 为键，值为数据列表，每个数据包含 time, value 和 feature_id
-
-        Raises:
-            ValueError: 当 SCADA 设备未找到或字段无效时
-        """
-        pipe_ids: list[str] = []
-        junction_ids: list[str] = []
-        for feature_id, feature_type in feature_infos:
-            if feature_type.lower() == "pipe":
-                pipe_ids.append(feature_id)
-            elif feature_type.lower() == "junction":
-                junction_ids.append(feature_id)
-            else:
-                raise ValueError(f"Unknown type: {feature_type}")
-        link_series = await RealtimeRepository.get_link_fields_by_ids_time_range(
-            timescale_conn, start_time, end_time, list(dict.fromkeys(pipe_ids)), "flow"
-        )
-        node_series = await RealtimeRepository.get_node_fields_by_ids_time_range(
-            timescale_conn, start_time, end_time,
-            list(dict.fromkeys(junction_ids)), "pressure",
-        )
-        return {
-            feature_id: [
-                {**item, "feature_id": feature_id}
-                for item in (
-                    link_series.get(feature_id, [])
-                    if feature_type.lower() == "pipe"
-                    else node_series.get(feature_id, [])
-                )
-            ]
-            for feature_id, feature_type in feature_infos
-        }
-
-    @staticmethod
-    async def get_analysis_simulation_data(
-        timescale_conn: AsyncConnection,
-        feature_infos: List[Tuple[str, str]],
-        start_time: datetime,
-        end_time: datetime,
-        run_id: UUID,
-    ) -> Dict[str, List[Dict[str, Any]]]:
-        """
-        获取 link/node 分析模拟值
-
-        根据传入的 feature_infos，找到关联的 link/node，
-        并根据对应的 type，查询对应的模拟数据
-
-        Args:
-            timescale_conn: TimescaleDB 异步连接
-            feature_infos: 传入的 feature 信息列表，包含 (element_id, type)
-            start_time: 开始时间
-            end_time: 结束时间
-            run_id: 分析运行 ID
-
-        Returns:
-            模拟数据字典，以 feature_id 为键，值为数据列表，每个数据包含 time, value 和 feature_id
-
-        Raises:
-            ValueError: 当类型无效时
-        """
-        pipe_ids: list[str] = []
-        junction_ids: list[str] = []
-        for feature_id, feature_type in feature_infos:
-            if feature_type.lower() == "pipe":
-                pipe_ids.append(feature_id)
-            elif feature_type.lower() == "junction":
-                junction_ids.append(feature_id)
-            else:
-                raise ValueError(f"Unknown type: {feature_type}")
-        link_series = await AnalysisResultsRepository.get_series_by_ids(
-            timescale_conn, run_id, "link", list(dict.fromkeys(pipe_ids)),
-            start_time, end_time, "flow",
-        )
-        node_series = await AnalysisResultsRepository.get_series_by_ids(
-            timescale_conn, run_id, "node", list(dict.fromkeys(junction_ids)),
-            start_time, end_time, "pressure",
-        )
-        return {
-            feature_id: [
-                {**item, "feature_id": feature_id}
-                for item in (
-                    link_series.get(feature_id, [])
-                    if feature_type.lower() == "pipe"
-                    else node_series.get(feature_id, [])
-                )
-            ]
-            for feature_id, feature_type in feature_infos
-        }
-
-    @staticmethod
-    async def get_element_associated_scada_data(
-        timescale_conn: AsyncConnection,
-        postgres_conn: AsyncConnection,
-        element_id: str,
-        start_time: datetime,
-        end_time: datetime,
-        use_cleaned: bool = False,
-    ) -> Optional[Any]:
-        """
-        获取 link/node 关联的 SCADA 监测值
-
-        根据传入的 link/node id，匹配 SCADA 信息，
-        如果存在关联的 SCADA device_id，获取实际的监测数据
-
-        Args:
-            timescale_conn: TimescaleDB 异步连接
-            postgres_conn: PostgreSQL 异步连接
-            element_id: link 或 node 的 ID
-            start_time: 开始时间
-            end_time: 结束时间
-            use_cleaned: 是否使用清洗后的数据 (True: "cleaned_value", False: "monitored_value")
-
-        Returns:
-            SCADA 监测数据值，如果没有找到则返回 None
-
-        Raises:
-            ValueError: 当元素类型无效时
-        """
-
-        scada_by_id = await TimeseriesAnalysisService._get_project_scada_index(postgres_conn)
-        associated_scada = next(
-            (
-                scada
-                for scada in scada_by_id.values()
-                if (scada.get("node_id") or scada.get("link_id")) == element_id
-            ),
-            None,
-        )
-
-        if not associated_scada:
-            return None
-
-        device_id = associated_scada["device_id"]
-
-        data_field = "cleaned_value" if use_cleaned else "monitored_value"
-
-        res = await ScadaRepository.get_scada_field_by_id_time_range(
-            timescale_conn, [device_id], start_time, end_time, data_field
-        )
-
-        return {element_id: res.get(device_id, [])}
-
    @staticmethod
    async def clean_scada_data(
        timescale_conn: AsyncConnection,
@@ -57,6 +57,33 @@ def _points(
    ]


+def _history_series(
+    *, element_type: HistoryElementType, element_id: str,
+    source: HistorySource, source_unit: str, device_id: str | None = None,
+    unit_inferred: bool = False, points: list[HistoryPoint] | None = None,
+) -> ElementHistorySeries:
+    metric = _target_metric(element_type)
+    return ElementHistorySeries(
+        element_id=element_id, element_type=element_type, device_id=device_id,
+        metric=metric, source=source, source_unit=source_unit,
+        display_unit=display_unit_for_metric(metric.value),
+        unit_inferred=unit_inferred, points=points if points is not None else [],
+    )
+
+
+async def _resolve_result_units(
+    postgres_conn: AsyncConnection, query: ElementHistoryQuery,
+) -> tuple[dict[str, Any], bool]:
+    units = await NetworkSettingsRepository.get_result_units(postgres_conn)
+    if query.mode.value != "analysis_comparison":
+        return units, False
+    run = await AnalysisRepository.get_run(postgres_conn, query.run_id)
+    if run is None:
+        raise ValueError(f"analysis run does not exist: {query.run_id}")
+    recorded = (run.get("parameters") or {}).get("result_units")
+    return (recorded, False) if isinstance(recorded, dict) else (units, True)
+
+
 async def _query_page(timescale_conn, postgres_conn, query, selected_devices, limit, offset):
    descriptors = []
    series = []
@@ -66,16 +93,14 @@ async def _query_page(timescale_conn, postgres_conn, query, selected_devices, li
            "source": source, "element_type": element_type,
            "id": device_id if source == "scada" else element_id, "field": field,
        })
-        metric = _target_metric(element_type)
-        series.append(ElementHistorySeries(
+        series.append(_history_series(
            element_id=element_id, element_type=element_type, device_id=device_id,
-            metric=metric, source=(
+            source=(
                HistorySource.SCADA_RAW if field == "monitored_value" else HistorySource.SCADA_CLEANED
            ) if source == "scada" else (
                HistorySource.ANALYSIS_SIMULATION if source == "analysis" else HistorySource.REALTIME_SIMULATION
            ),
-            source_unit=unit, display_unit=display_unit_for_metric(metric.value),
-            unit_inferred=unit_inferred, points=[],
+            source_unit=unit, unit_inferred=unit_inferred,
        ))

    # Stable series ordering makes a page crossing raw/cleaned/model boundaries repeatable.
@@ -84,17 +109,8 @@ async def _query_page(timescale_conn, postgres_conn, query, selected_devices, li
            add(source="scada", element_type=element_type, element_id=element_id,
                device_id=device_id, field=field, unit=device["measurement_unit"])
    if query.mode.value != "observed":
-        units = await NetworkSettingsRepository.get_result_units(postgres_conn)
+        units, inferred = await _resolve_result_units(postgres_conn, query)
        source = "analysis" if query.mode.value == "analysis_comparison" else "realtime"
-        inferred = False
-        if source == "analysis":
-            run = await AnalysisRepository.get_run(postgres_conn, query.run_id)
-            if run is None:
-                raise ValueError(f"analysis run does not exist: {query.run_id}")
-            recorded = (run.get("parameters") or {}).get("result_units")
-            inferred = not isinstance(recorded, dict)
-            if not inferred:
-                units = recorded
        targets = {(t.element_type, t.element_id) for t in query.elements}
        for element_type, element_id in sorted(targets):
            metric = _target_metric(element_type)
@@ -202,21 +218,18 @@ class TimeseriesHistoryService:
            rows = scada_by_device.get(device_id, [])
            if not rows:
                continue
-            metric = _target_metric(element_type)
            unit = device["measurement_unit"]
            for source, field in (
                (HistorySource.SCADA_RAW, "monitored_value"),
                (HistorySource.SCADA_CLEANED, "cleaned_value"),
            ):
                series.append(
-                    ElementHistorySeries(
+                    _history_series(
                        element_id=element_id,
                        element_type=element_type,
                        device_id=device_id,
-                        metric=metric,
                        source=source,
                        source_unit=unit,
-                        display_unit=display_unit_for_metric(metric.value),
                        points=_points(
                            rows,
                            value_field=field,
@@ -227,15 +240,8 @@ class TimeseriesHistoryService:
        if query.mode.value == "observed":
            return ElementHistoryResponse(series=series)

-        current_units = await NetworkSettingsRepository.get_result_units(postgres_conn)
+        result_units, unit_inferred = await _resolve_result_units(postgres_conn, query)
        if query.mode.value == "analysis_comparison":
-            run = await AnalysisRepository.get_run(postgres_conn, query.run_id)
-            if run is None:
-                raise ValueError(f"analysis run does not exist: {query.run_id}")
-            parameters = run.get("parameters") or {}
-            recorded_units = parameters.get("result_units")
-            unit_inferred = not isinstance(recorded_units, dict)
-            result_units = recorded_units if not unit_inferred else current_units
            link_values = await AnalysisResultsRepository.get_series_by_ids(
                timescale_conn,
                query.run_id,
@@ -256,8 +262,6 @@ class TimeseriesHistoryService:
            )
            source = HistorySource.ANALYSIS_SIMULATION
        else:
-            unit_inferred = False
-            result_units = current_units
            link_values = await RealtimeRepository.get_link_fields_by_ids_time_range(
                timescale_conn,
                query.start_time,
@@ -285,13 +289,11 @@ class TimeseriesHistoryService:
                continue
            source_unit = str(result_units.get(metric.value, "")).strip()
            series.append(
-                ElementHistorySeries(
+                _history_series(
                    element_id=target.element_id,
                    element_type=target.element_type,
-                    metric=metric,
                    source=source,
                    source_unit=source_unit,
-                    display_unit=display_unit_for_metric(metric.value),
                    unit_inferred=unit_inferred,
                    points=_points(
                        rows,