Compare commits

...
Author SHA1 Message Date
QiuSW 407ffa17b2 feat: 实现 Brain 匿名检测与单路跟踪 (#14) 2026-08-28 22:39:12 +08:00
ila ba4ec28763 feat: 建立 Brain 可替换视频解码流水线 (#13)
合入 dev,#13 保持待验收。
2026-08-28 22:27:27 +08:00
QiuSW b9b067213f feat: 建立 Brain 可替换视频解码流水线 (#13) 2026-08-28 22:27:04 +08:00
ila 52b368068e feat: 实现 Brain 合成与本地视频输入 (#11)
合入 dev,#11 保持待验收。
2026-08-28 22:20:04 +08:00
QiuSW 7964d61cab feat: 实现 Brain 合成与本地视频输入 (#11) 2026-08-28 22:19:22 +08:00
ila 4a605f6482 Merge pull request #123 from feat/sense-79-ops-alerts
feat: 实现 Sense 运维告警隔离 (#79)
2026-08-28 18:25:33 +08:00
QiuSW d61e6d5ee1 feat: 实现 Sense 运维告警隔离 (#79) 2026-08-28 18:25:07 +08:00
ila 7274bd42f5 Merge PR #122: Sense 可靠投递状态机
Implements #78. Issue remains open for user acceptance.
2026-08-28 17:23:01 +08:00
38 changed files with 2460 additions and 0 deletions
@@ -0,0 +1,27 @@
"""Brain-internal configuration models.
These types are deliberately not a cross-project contract. Adapters for a
future versioned Sense/Brain contract belong in a coordination task.
"""
from .models import (
AreaRule,
BrainInputConfig,
ConfigError,
DirectionalLineRule,
Point,
SourceConfig,
VideoProfile,
parse_input_config,
)
__all__ = [
"AreaRule",
"BrainInputConfig",
"ConfigError",
"DirectionalLineRule",
"Point",
"SourceConfig",
"VideoProfile",
"parse_input_config",
]
+206
View File
@@ -0,0 +1,206 @@
"""Validated project-internal input configuration."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Sequence
INTERNAL_SCHEMA = "brain.internal.input/v1"
_SECRET_KEYS = {"credential", "password", "secret", "token", "username"}
class ConfigError(ValueError):
"""Raised when Brain's project-internal test/runtime config is invalid."""
@dataclass(frozen=True, slots=True)
class Point:
x: float
y: float
@dataclass(frozen=True, slots=True)
class AreaRule:
rule_id: str
points: tuple[Point, ...]
@dataclass(frozen=True, slots=True)
class DirectionalLineRule:
rule_id: str
start: Point
end: Point
trigger_direction: str
@dataclass(frozen=True, slots=True)
class VideoProfile:
profile_id: str
width: int
height: int
fps: float
@dataclass(frozen=True, slots=True)
class SourceConfig:
kind: str
seed: int | None = None
frame_count: int | None = None
path: Path | None = None
chunk_size: int = 64 * 1024
@dataclass(frozen=True, slots=True)
class BrainInputConfig:
schema: str
logical_device_id: str
profile: VideoProfile
source: SourceConfig
areas: tuple[AreaRule, ...] = ()
directional_lines: tuple[DirectionalLineRule, ...] = ()
def _mapping(value: object, field: str) -> Mapping[str, Any]:
if not isinstance(value, Mapping):
raise ConfigError(f"{field} must be an object")
return value
def _text(value: object, field: str) -> str:
if not isinstance(value, str) or not value.strip():
raise ConfigError(f"{field} must be a non-empty string")
return value.strip()
def _integer(value: object, field: str, *, minimum: int = 1) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
raise ConfigError(f"{field} must be an integer >= {minimum}")
return value
def _number(value: object, field: str, *, minimum: float = 0.0) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ConfigError(f"{field} must be a number")
result = float(value)
if result <= minimum:
raise ConfigError(f"{field} must be greater than {minimum}")
return result
def _reject_secrets(value: object, field: str = "config") -> None:
if isinstance(value, Mapping):
for key, child in value.items():
normalized = str(key).strip().lower()
if normalized in _SECRET_KEYS or any(
marker in normalized for marker in ("password", "secret", "token")
):
raise ConfigError(f"{field} must not contain credential field {key!r}")
_reject_secrets(child, f"{field}.{key}")
elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
for index, child in enumerate(value):
_reject_secrets(child, f"{field}[{index}]")
def _point(value: object, field: str) -> Point:
if (
not isinstance(value, Sequence)
or isinstance(value, (str, bytes, bytearray))
or len(value) != 2
):
raise ConfigError(f"{field} must be [x, y]")
x, y = value
if isinstance(x, bool) or isinstance(y, bool) or not isinstance(x, (int, float)) or not isinstance(y, (int, float)):
raise ConfigError(f"{field} coordinates must be numbers")
point = Point(float(x), float(y))
if not 0.0 <= point.x <= 1.0 or not 0.0 <= point.y <= 1.0:
raise ConfigError(f"{field} coordinates must be normalized to 0..1")
return point
def parse_input_config(raw: Mapping[str, Any], *, base_dir: Path | None = None) -> BrainInputConfig:
"""Parse the explicitly versioned Brain-internal input configuration."""
_reject_secrets(raw)
schema = _text(raw.get("schema"), "schema")
if schema != INTERNAL_SCHEMA:
raise ConfigError(f"schema must be {INTERNAL_SCHEMA!r}")
profile_raw = _mapping(raw.get("profile"), "profile")
profile = VideoProfile(
profile_id=_text(profile_raw.get("id"), "profile.id"),
width=_integer(profile_raw.get("width"), "profile.width"),
height=_integer(profile_raw.get("height"), "profile.height"),
fps=_number(profile_raw.get("fps"), "profile.fps"),
)
source_raw = _mapping(raw.get("source"), "source")
kind = _text(source_raw.get("kind"), "source.kind")
if kind == "synthetic":
seed = source_raw.get("seed", 0)
if isinstance(seed, bool) or not isinstance(seed, int):
raise ConfigError("source.seed must be an integer")
source = SourceConfig(
kind=kind,
seed=seed,
frame_count=_integer(source_raw.get("frame_count"), "source.frame_count"),
)
elif kind == "local_file":
configured_path = Path(_text(source_raw.get("path"), "source.path"))
if not configured_path.is_absolute() and base_dir is not None:
configured_path = base_dir / configured_path
source = SourceConfig(
kind=kind,
path=configured_path,
chunk_size=_integer(source_raw.get("chunk_size", 64 * 1024), "source.chunk_size"),
)
else:
raise ConfigError("source.kind must be 'synthetic' or 'local_file'")
areas_raw = raw.get("areas", [])
if not isinstance(areas_raw, list):
raise ConfigError("areas must be an array")
areas: list[AreaRule] = []
for index, item in enumerate(areas_raw):
area = _mapping(item, f"areas[{index}]")
points_raw = area.get("points")
if not isinstance(points_raw, list) or len(points_raw) < 3:
raise ConfigError(f"areas[{index}].points must contain at least three points")
areas.append(
AreaRule(
rule_id=_text(area.get("id"), f"areas[{index}].id"),
points=tuple(_point(point, f"areas[{index}].points[{point_index}]") for point_index, point in enumerate(points_raw)),
)
)
lines_raw = raw.get("directional_lines", [])
if not isinstance(lines_raw, list):
raise ConfigError("directional_lines must be an array")
lines: list[DirectionalLineRule] = []
for index, item in enumerate(lines_raw):
line = _mapping(item, f"directional_lines[{index}]")
direction = _text(line.get("trigger_direction"), f"directional_lines[{index}].trigger_direction")
if direction not in {"left_to_right", "right_to_left"}:
raise ConfigError(
f"directional_lines[{index}].trigger_direction must be left_to_right or right_to_left"
)
lines.append(
DirectionalLineRule(
rule_id=_text(line.get("id"), f"directional_lines[{index}].id"),
start=_point(line.get("start"), f"directional_lines[{index}].start"),
end=_point(line.get("end"), f"directional_lines[{index}].end"),
trigger_direction=direction,
)
)
identifiers = [area.rule_id for area in areas] + [line.rule_id for line in lines]
if len(set(identifiers)) != len(identifiers):
raise ConfigError("rule ids must be unique")
return BrainInputConfig(
schema=schema,
logical_device_id=_text(raw.get("logical_device_id"), "logical_device_id"),
profile=profile,
source=source,
areas=tuple(areas),
directional_lines=tuple(lines),
)
@@ -0,0 +1,16 @@
"""Replaceable Brain-internal video decode pipeline."""
from .models import DecodedFrame, DecoderBackend, DecoderError
from .pipeline import DecoderPipeline, decode_packets
from .raw_rgb import RawRGBDecoder
from .y4m import Y4MDecoder
__all__ = [
"DecodedFrame",
"DecoderBackend",
"DecoderError",
"DecoderPipeline",
"RawRGBDecoder",
"Y4MDecoder",
"decode_packets",
]
+35
View File
@@ -0,0 +1,35 @@
"""Decode-layer ports and frame model."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterable, Iterator, Protocol
from yovision_brain.input import CancellationToken, InputPacket
class DecoderError(RuntimeError):
"""A safe and actionable decode failure."""
@dataclass(frozen=True, slots=True)
class DecodedFrame:
sequence: int
timestamp_ns: int
logical_device_id: str
profile_id: str
width: int
height: int
pixel_format: str
payload: bytes
dimensions_changed: bool = False
class DecoderBackend(Protocol):
media_formats: frozenset[str]
def decode(
self,
packets: Iterable[InputPacket],
cancellation: CancellationToken | None = None,
) -> Iterator[DecodedFrame]: ...
@@ -0,0 +1,47 @@
"""Decoder selection independent of concrete codec libraries."""
from __future__ import annotations
from collections.abc import Iterable, Iterator
from itertools import chain
from yovision_brain.input import CancellationToken, InputPacket
from .models import DecodedFrame, DecoderBackend, DecoderError
from .raw_rgb import RawRGBDecoder
from .y4m import Y4MDecoder
class DecoderPipeline:
def __init__(self, backends: Iterable[DecoderBackend] | None = None) -> None:
selected = tuple(backends) if backends is not None else (RawRGBDecoder(), Y4MDecoder())
self._backends: dict[str, DecoderBackend] = {}
for backend in selected:
for media_format in backend.media_formats:
if media_format in self._backends:
raise ValueError(f"duplicate decoder for media format {media_format!r}")
self._backends[media_format] = backend
def decode(
self,
packets: Iterable[InputPacket],
cancellation: CancellationToken | None = None,
) -> Iterator[DecodedFrame]:
iterator = iter(packets)
if cancellation is not None and cancellation.cancelled:
return
try:
first = next(iterator)
except StopIteration:
return
backend = self._backends.get(first.media_format)
if backend is None:
raise DecoderError(f"no decoder registered for media format {first.media_format!r}")
yield from backend.decode(chain((first,), iterator), cancellation)
def decode_packets(
packets: Iterable[InputPacket],
cancellation: CancellationToken | None = None,
) -> Iterator[DecodedFrame]:
return DecoderPipeline().decode(packets, cancellation)
@@ -0,0 +1,45 @@
"""Pass-through decoder for deterministic RGB24 synthetic frames."""
from __future__ import annotations
from collections.abc import Iterable, Iterator
from yovision_brain.input import CancellationToken, InputPacket
from .models import DecodedFrame, DecoderError
class RawRGBDecoder:
media_formats = frozenset({"rgb24"})
def decode(
self,
packets: Iterable[InputPacket],
cancellation: CancellationToken | None = None,
) -> Iterator[DecodedFrame]:
previous_dimensions: tuple[int, int] | None = None
for packet in packets:
if cancellation is not None and cancellation.cancelled:
return
if packet.media_format != "rgb24":
raise DecoderError(f"raw RGB decoder does not support {packet.media_format!r}")
expected = packet.width * packet.height * 3
if len(packet.payload) != expected:
raise DecoderError(
f"RGB24 frame {packet.sequence} has {len(packet.payload)} bytes; expected {expected}"
)
if packet.timestamp_ns is None:
raise DecoderError(f"RGB24 frame {packet.sequence} has no source timestamp")
dimensions = (packet.width, packet.height)
yield DecodedFrame(
sequence=packet.sequence,
timestamp_ns=packet.timestamp_ns,
logical_device_id=packet.logical_device_id,
profile_id=packet.profile_id,
width=packet.width,
height=packet.height,
pixel_format="rgb24",
payload=packet.payload,
dimensions_changed=previous_dimensions is not None and dimensions != previous_dimensions,
)
previous_dimensions = dimensions
+167
View File
@@ -0,0 +1,167 @@
"""Minimal streaming YUV4MPEG2 decoder for anonymous local fixtures.
The backend intentionally supports only uncompressed C444 streams. Production
codecs and RTSP belong behind the same decoder port in later tasks.
"""
from __future__ import annotations
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
from yovision_brain.input import CancellationToken, InputPacket
from .models import DecodedFrame, DecoderError
_MAX_HEADER_BYTES = 4096
_MAX_FRAME_BYTES = 256 * 1024 * 1024
class _Cancelled(Exception):
pass
class _PacketReader:
def __init__(
self,
packets: Iterable[InputPacket],
cancellation: CancellationToken | None,
) -> None:
self._packets = iter(packets)
self._cancellation = cancellation
self._buffer = bytearray()
self._ended = False
self.first_packet: InputPacket | None = None
def _fill(self) -> bool:
if self._cancellation is not None and self._cancellation.cancelled:
raise _Cancelled
if self._ended:
return False
try:
packet = next(self._packets)
except StopIteration:
self._ended = True
return False
if packet.media_format != "container-bytes":
raise DecoderError(f"Y4M decoder does not support {packet.media_format!r}")
if self.first_packet is None:
self.first_packet = packet
else:
first = self.first_packet
if (packet.logical_device_id, packet.profile_id) != (
first.logical_device_id,
first.profile_id,
):
raise DecoderError("input identity changed inside one local video stream")
self._buffer.extend(packet.payload)
return True
def line(self, *, allow_clean_eof: bool = False) -> bytes | None:
while True:
newline = self._buffer.find(b"\n")
if newline >= 0:
result = bytes(self._buffer[:newline])
del self._buffer[: newline + 1]
return result
if len(self._buffer) > _MAX_HEADER_BYTES:
raise DecoderError("Y4M header exceeds the safe size limit")
if not self._fill():
if not self._buffer and allow_clean_eof:
return None
raise DecoderError("truncated Y4M header")
def exact(self, size: int) -> bytes:
while len(self._buffer) < size:
if not self._fill():
raise DecoderError("truncated Y4M frame payload")
result = bytes(self._buffer[:size])
del self._buffer[:size]
return result
@dataclass(frozen=True, slots=True)
class _Header:
width: int
height: int
fps_numerator: int
fps_denominator: int
def _positive_int(value: bytes, field: str) -> int:
try:
result = int(value)
except ValueError as exc:
raise DecoderError(f"invalid Y4M {field}") from exc
if result <= 0:
raise DecoderError(f"invalid Y4M {field}")
return result
def _parse_header(line: bytes) -> _Header:
parts = line.split()
if not parts or parts[0] != b"YUV4MPEG2":
raise DecoderError("unsupported local video format; expected YUV4MPEG2")
fields = {part[:1]: part[1:] for part in parts[1:] if len(part) > 1}
if fields.get(b"C", b"444") not in {b"444", b"444jpeg"}:
raise DecoderError("unsupported Y4M chroma; only C444 is supported")
width = _positive_int(fields.get(b"W", b""), "width")
height = _positive_int(fields.get(b"H", b""), "height")
fps_parts = fields.get(b"F", b"").split(b":", 1)
if len(fps_parts) != 2:
raise DecoderError("invalid Y4M frame rate")
header = _Header(
width=width,
height=height,
fps_numerator=_positive_int(fps_parts[0], "frame rate numerator"),
fps_denominator=_positive_int(fps_parts[1], "frame rate denominator"),
)
if header.width * header.height * 3 > _MAX_FRAME_BYTES:
raise DecoderError("Y4M frame exceeds the safe size limit")
return header
class Y4MDecoder:
media_formats = frozenset({"container-bytes"})
def decode(
self,
packets: Iterable[InputPacket],
cancellation: CancellationToken | None = None,
) -> Iterator[DecodedFrame]:
reader = _PacketReader(packets, cancellation)
try:
header_line = reader.line()
assert header_line is not None
header = _parse_header(header_line)
first = reader.first_packet
if first is None:
raise DecoderError("local video input is empty")
if (first.width, first.height) != (header.width, header.height):
raise DecoderError(
"Y4M dimensions do not match the configured input profile "
f"({header.width}x{header.height} != {first.width}x{first.height})"
)
interval_ns = round(1_000_000_000 * header.fps_denominator / header.fps_numerator)
frame_size = header.width * header.height * 3
sequence = 0
while True:
frame_header = reader.line(allow_clean_eof=True)
if frame_header is None:
return
if frame_header != b"FRAME":
raise DecoderError(f"invalid Y4M frame header at frame {sequence}")
payload = reader.exact(frame_size)
yield DecodedFrame(
sequence=sequence,
timestamp_ns=sequence * interval_ns,
logical_device_id=first.logical_device_id,
profile_id=first.profile_id,
width=header.width,
height=header.height,
pixel_format="yuv444p",
payload=payload,
)
sequence += 1
except _Cancelled:
return
@@ -0,0 +1,16 @@
"""Brain-internal video input adapters."""
from .factory import build_input_source
from .local_file import LocalFileInput
from .models import CancellationToken, InputError, InputPacket, InputSource
from .synthetic import SyntheticInput
__all__ = [
"CancellationToken",
"InputError",
"InputPacket",
"InputSource",
"LocalFileInput",
"SyntheticInput",
"build_input_source",
]
+17
View File
@@ -0,0 +1,17 @@
"""Construct the configured Brain-internal input adapter."""
from __future__ import annotations
from yovision_brain.config import BrainInputConfig
from .local_file import LocalFileInput
from .models import InputSource
from .synthetic import SyntheticInput
def build_input_source(config: BrainInputConfig) -> InputSource:
if config.source.kind == "synthetic":
return SyntheticInput(config)
if config.source.kind == "local_file":
return LocalFileInput(config)
raise ValueError(f"unsupported Brain input source kind: {config.source.kind}")
@@ -0,0 +1,54 @@
"""Explicit local-file input adapter with safe error reporting."""
from __future__ import annotations
from collections.abc import Iterator
from pathlib import Path
from yovision_brain.config import BrainInputConfig
from .models import CancellationToken, InputError, InputPacket
class LocalFileInput:
def __init__(self, config: BrainInputConfig) -> None:
if config.source.kind != "local_file" or config.source.path is None:
raise ValueError("LocalFileInput requires a local_file source config")
self._config = config
self._path = config.source.path
@property
def source_label(self) -> str:
"""Return a safe label rather than exposing the internal absolute path."""
return self._path.name
def packets(self, cancellation: CancellationToken | None = None) -> Iterator[InputPacket]:
profile = self._config.profile
source = self._config.source
try:
stream = self._path.open("rb")
except FileNotFoundError as exc:
raise InputError(f"local video source not found: {self.source_label}") from exc
except OSError as exc:
raise InputError(f"local video source cannot be opened: {self.source_label}: {exc.strerror}") from exc
with stream:
sequence = 0
while cancellation is None or not cancellation.cancelled:
try:
payload = stream.read(source.chunk_size)
except OSError as exc:
raise InputError(f"local video source read failed: {self.source_label}: {exc.strerror}") from exc
if not payload:
return
yield InputPacket(
sequence=sequence,
timestamp_ns=None,
logical_device_id=self._config.logical_device_id,
profile_id=profile.profile_id,
width=profile.width,
height=profile.height,
media_format="container-bytes",
payload=payload,
)
sequence += 1
+43
View File
@@ -0,0 +1,43 @@
"""Common project-internal input types."""
from __future__ import annotations
from dataclasses import dataclass
from threading import Event
from typing import Iterator, Protocol
class InputError(RuntimeError):
"""A safe, actionable input adapter error."""
class CancellationToken:
"""Thread-safe cooperative cancellation without platform dependencies."""
def __init__(self) -> None:
self._event = Event()
def cancel(self) -> None:
self._event.set()
@property
def cancelled(self) -> bool:
return self._event.is_set()
@dataclass(frozen=True, slots=True)
class InputPacket:
sequence: int
timestamp_ns: int | None
logical_device_id: str
profile_id: str
width: int
height: int
media_format: str
payload: bytes
class InputSource(Protocol):
"""Replaceable source boundary consumed by the future decode layer."""
def packets(self, cancellation: CancellationToken | None = None) -> Iterator[InputPacket]: ...
@@ -0,0 +1,38 @@
"""Deterministic synthetic RGB input for isolated tests and smoke runs."""
from __future__ import annotations
import random
from collections.abc import Iterator
from yovision_brain.config import BrainInputConfig
from .models import CancellationToken, InputPacket
class SyntheticInput:
def __init__(self, config: BrainInputConfig) -> None:
if config.source.kind != "synthetic":
raise ValueError("SyntheticInput requires a synthetic source config")
self._config = config
def packets(self, cancellation: CancellationToken | None = None) -> Iterator[InputPacket]:
source = self._config.source
assert source.seed is not None and source.frame_count is not None
randomizer = random.Random(source.seed)
profile = self._config.profile
frame_size = profile.width * profile.height * 3
interval_ns = round(1_000_000_000 / profile.fps)
for sequence in range(source.frame_count):
if cancellation is not None and cancellation.cancelled:
return
yield InputPacket(
sequence=sequence,
timestamp_ns=sequence * interval_ns,
logical_device_id=self._config.logical_device_id,
profile_id=profile.profile_id,
width=profile.width,
height=profile.height,
media_format="rgb24",
payload=randomizer.randbytes(frame_size),
)
@@ -0,0 +1,16 @@
"""Anonymous detection and single-stream tracking."""
from .detector import LumaBlobDetector, TorchLumaBlobDetector
from .models import BoundingBox, Detection, Detector, DetectorMetadata, TrackedObject
from .tracker import SingleStreamTracker
__all__ = [
"BoundingBox",
"Detection",
"Detector",
"DetectorMetadata",
"LumaBlobDetector",
"SingleStreamTracker",
"TorchLumaBlobDetector",
"TrackedObject",
]
+120
View File
@@ -0,0 +1,120 @@
"""Deterministic anonymous blob detectors with no biometric semantics."""
from __future__ import annotations
from collections.abc import Sequence
from yovision_brain.decode import DecodedFrame, DecoderError
from .models import BoundingBox, Detection, DetectorMetadata
_METADATA = DetectorMetadata(
name="yovision-luma-blob",
version="1.0.0",
source="YoVision Brain first-party deterministic algorithm",
license="No external model license; no learned weights are distributed",
weights="none",
)
def _components(mask: Sequence[Sequence[bool]], minimum_area: int) -> tuple[BoundingBox, ...]:
height = len(mask)
width = len(mask[0]) if height else 0
visited: set[tuple[int, int]] = set()
boxes: list[BoundingBox] = []
for y in range(height):
for x in range(width):
if not mask[y][x] or (x, y) in visited:
continue
pending = [(x, y)]
visited.add((x, y))
points: list[tuple[int, int]] = []
while pending:
current_x, current_y = pending.pop()
points.append((current_x, current_y))
for neighbor in (
(current_x - 1, current_y),
(current_x + 1, current_y),
(current_x, current_y - 1),
(current_x, current_y + 1),
):
nx, ny = neighbor
if 0 <= nx < width and 0 <= ny < height and mask[ny][nx] and neighbor not in visited:
visited.add(neighbor)
pending.append(neighbor)
if len(points) >= minimum_area:
xs, ys = zip(*points)
boxes.append(BoundingBox(min(xs), min(ys), max(xs) + 1, max(ys) + 1))
return tuple(sorted(boxes, key=lambda box: (box.top, box.left, box.bottom, box.right)))
def _validate_frame(frame: DecodedFrame) -> None:
if frame.pixel_format not in {"rgb24", "yuv444p"}:
raise DecoderError(f"anonymous detector does not support pixel format {frame.pixel_format!r}")
expected = frame.width * frame.height * 3
if len(frame.payload) != expected:
raise DecoderError(f"vision frame has {len(frame.payload)} bytes; expected {expected}")
class LumaBlobDetector:
"""Small CPU reference detector used for deterministic integration tests."""
metadata = _METADATA
def __init__(self, *, threshold: int = 200, minimum_area: int = 1) -> None:
if not 0 <= threshold <= 255 or minimum_area < 1:
raise ValueError("invalid luma detector threshold or minimum area")
self._threshold = threshold
self._minimum_area = minimum_area
def detect(self, frame: DecodedFrame) -> tuple[Detection, ...]:
_validate_frame(frame)
if frame.pixel_format == "rgb24":
pixels = [
max(frame.payload[index : index + 3])
for index in range(0, len(frame.payload), 3)
]
else:
pixels = list(frame.payload[: frame.width * frame.height])
mask = [
[pixels[y * frame.width + x] >= self._threshold for x in range(frame.width)]
for y in range(frame.height)
]
return tuple(
Detection(box=box, category="anonymous_target", confidence=1.0)
for box in _components(mask, self._minimum_area)
)
class TorchLumaBlobDetector:
"""PyTorch CPU/GPU smoke backend; it contains no external model weights."""
metadata = DetectorMetadata(
name="yovision-torch-luma-blob",
version="1.0.0",
source="YoVision Brain first-party PyTorch tensor implementation",
license="PyTorch BSD-3-Clause; no external model weights",
weights="none",
)
def __init__(self, *, threshold: int = 200, minimum_area: int = 1, device: str = "cpu") -> None:
self._threshold = threshold
self._minimum_area = minimum_area
self._device = device
def detect(self, frame: DecodedFrame) -> tuple[Detection, ...]:
_validate_frame(frame)
try:
import torch
except ImportError as exc:
raise RuntimeError("PyTorch runtime is required for TorchLumaBlobDetector") from exc
values = torch.tensor(list(frame.payload), dtype=torch.uint8, device=self._device)
if frame.pixel_format == "rgb24":
luma = values.reshape(frame.height, frame.width, 3).amax(dim=2)
else:
luma = values[: frame.width * frame.height].reshape(frame.height, frame.width)
mask = (luma >= self._threshold).cpu().tolist()
return tuple(
Detection(box=box, category="anonymous_target", confidence=1.0)
for box in _components(mask, self._minimum_area)
)
+52
View File
@@ -0,0 +1,52 @@
"""Privacy-preserving vision ports and observations."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Protocol
from yovision_brain.decode import DecodedFrame
@dataclass(frozen=True, slots=True)
class DetectorMetadata:
name: str
version: str
source: str
license: str
weights: str
@dataclass(frozen=True, slots=True)
class BoundingBox:
left: int
top: int
right: int
bottom: int
@property
def area(self) -> int:
return max(0, self.right - self.left) * max(0, self.bottom - self.top)
@dataclass(frozen=True, slots=True)
class Detection:
box: BoundingBox
category: str
confidence: float
@dataclass(frozen=True, slots=True)
class TrackedObject:
track_id: str
box: BoundingBox
category: str
confidence: float
frame_sequence: int
timestamp_ns: int
class Detector(Protocol):
metadata: DetectorMetadata
def detect(self, frame: DecodedFrame) -> tuple[Detection, ...]: ...
@@ -0,0 +1,83 @@
"""Session-local single-stream IoU tracker."""
from __future__ import annotations
from dataclasses import dataclass
from .models import BoundingBox, Detection, TrackedObject
def _iou(first: BoundingBox, second: BoundingBox) -> float:
intersection = BoundingBox(
max(first.left, second.left),
max(first.top, second.top),
min(first.right, second.right),
min(first.bottom, second.bottom),
).area
union = first.area + second.area - intersection
return intersection / union if union else 0.0
@dataclass(slots=True)
class _Track:
track_id: str
detection: Detection
missed: int = 0
class SingleStreamTracker:
"""Tracks anonymous boxes only within one process session and one stream."""
def __init__(self, *, iou_threshold: float = 0.2, max_missed: int = 2) -> None:
if not 0.0 <= iou_threshold <= 1.0 or max_missed < 0:
raise ValueError("invalid tracker threshold or missed-frame limit")
self._iou_threshold = iou_threshold
self._max_missed = max_missed
self._tracks: dict[str, _Track] = {}
self._next_id = 1
def update(
self,
detections: tuple[Detection, ...],
*,
frame_sequence: int,
timestamp_ns: int,
) -> tuple[TrackedObject, ...]:
unmatched_tracks = set(self._tracks)
results: list[TrackedObject] = []
for detection in detections:
candidates = [
(track_id, _iou(self._tracks[track_id].detection.box, detection.box))
for track_id in unmatched_tracks
if self._tracks[track_id].detection.category == detection.category
]
track_id, score = max(candidates, key=lambda item: item[1], default=("", -1.0))
if score < self._iou_threshold:
track_id = f"track-{self._next_id:06d}"
self._next_id += 1
self._tracks[track_id] = _Track(track_id, detection)
else:
unmatched_tracks.remove(track_id)
self._tracks[track_id].detection = detection
self._tracks[track_id].missed = 0
results.append(
TrackedObject(
track_id=track_id,
box=detection.box,
category=detection.category,
confidence=detection.confidence,
frame_sequence=frame_sequence,
timestamp_ns=timestamp_ns,
)
)
for track_id in unmatched_tracks:
track = self._tracks[track_id]
track.missed += 1
if track.missed > self._max_missed:
del self._tracks[track_id]
return tuple(results)
def finish(self) -> tuple[str, ...]:
ended = tuple(sorted(self._tracks))
self._tracks.clear()
return ended
+72
View File
@@ -0,0 +1,72 @@
from __future__ import annotations
from pathlib import Path
import pytest
from yovision_brain.config import ConfigError, parse_input_config
def valid_config() -> dict[str, object]:
return {
"schema": "brain.internal.input/v1",
"logical_device_id": "synthetic-camera-01",
"profile": {"id": "main", "width": 4, "height": 3, "fps": 5},
"source": {"kind": "synthetic", "seed": 17, "frame_count": 3},
"areas": [{"id": "danger-yard", "points": [[0.1, 0.1], [0.9, 0.1], [0.5, 0.8]]}],
"directional_lines": [
{
"id": "gate-line",
"start": [0.2, 0.5],
"end": [0.8, 0.5],
"trigger_direction": "left_to_right",
}
],
}
def test_parse_versioned_internal_config() -> None:
parsed = parse_input_config(valid_config())
assert parsed.schema == "brain.internal.input/v1"
assert parsed.logical_device_id == "synthetic-camera-01"
assert parsed.profile.width == 4
assert parsed.areas[0].rule_id == "danger-yard"
assert parsed.directional_lines[0].trigger_direction == "left_to_right"
def test_relative_local_path_is_bound_to_explicit_base(tmp_path: Path) -> None:
raw = valid_config()
raw["source"] = {"kind": "local_file", "path": "fixture.bin", "chunk_size": 8}
parsed = parse_input_config(raw, base_dir=tmp_path)
assert parsed.source.path == tmp_path / "fixture.bin"
@pytest.mark.parametrize(
("change", "message"),
[
({"schema": "shared.source/v1"}, "schema must be"),
({"logical_device_id": ""}, "logical_device_id"),
({"source": {"kind": "synthetic", "seed": 1, "frame_count": 0}}, "frame_count"),
({"password": "must-not-be-accepted"}, "credential field"),
],
)
def test_invalid_or_secret_config_is_rejected(change: dict[str, object], message: str) -> None:
raw = valid_config()
raw.update(change)
with pytest.raises(ConfigError, match=message):
parse_input_config(raw)
def test_coordinates_and_rule_ids_are_validated() -> None:
raw = valid_config()
raw["areas"] = [{"id": "same", "points": [[0, 0], [2, 0], [0, 1]]}]
with pytest.raises(ConfigError, match="normalized"):
parse_input_config(raw)
raw = valid_config()
raw["areas"] = [{"id": "same", "points": [[0, 0], [1, 0], [0, 1]]}]
raw["directional_lines"] = [
{"id": "same", "start": [0, 0], "end": [1, 1], "trigger_direction": "left_to_right"}
]
with pytest.raises(ConfigError, match="unique"):
parse_input_config(raw)
+104
View File
@@ -0,0 +1,104 @@
from __future__ import annotations
from pathlib import Path
import pytest
from yovision_brain.config import parse_input_config
from yovision_brain.decode import DecoderError, DecoderPipeline, decode_packets
from yovision_brain.input import CancellationToken, InputPacket, LocalFileInput, SyntheticInput
def synthetic_packets():
config = parse_input_config(
{
"schema": "brain.internal.input/v1",
"logical_device_id": "synthetic-01",
"profile": {"id": "main", "width": 2, "height": 1, "fps": 5},
"source": {"kind": "synthetic", "seed": 3, "frame_count": 2},
}
)
return SyntheticInput(config).packets()
def local_packets(path: Path, *, width: int = 2, height: int = 1, chunk_size: int = 5):
config = parse_input_config(
{
"schema": "brain.internal.input/v1",
"logical_device_id": "local-01",
"profile": {"id": "archive", "width": width, "height": height, "fps": 25},
"source": {"kind": "local_file", "path": str(path), "chunk_size": chunk_size},
}
)
return LocalFileInput(config).packets()
def test_rgb24_pipeline_preserves_order_timestamps_and_metadata() -> None:
frames = list(decode_packets(synthetic_packets()))
assert [frame.sequence for frame in frames] == [0, 1]
assert [frame.timestamp_ns for frame in frames] == [0, 200_000_000]
assert all(frame.logical_device_id == "synthetic-01" for frame in frames)
assert all(frame.profile_id == "main" for frame in frames)
assert all((frame.width, frame.height, frame.pixel_format) == (2, 1, "rgb24") for frame in frames)
def test_rgb24_dimension_change_is_explicit() -> None:
packets = [
InputPacket(0, 0, "camera", "main", 1, 1, "rgb24", b"abc"),
InputPacket(1, 1, "camera", "main", 2, 1, "rgb24", b"abcdef"),
]
frames = list(decode_packets(packets))
assert [frame.dimensions_changed for frame in frames] == [False, True]
def test_invalid_rgb_payload_and_unsupported_format_are_clear() -> None:
bad = [InputPacket(0, 0, "camera", "main", 2, 2, "rgb24", b"short")]
with pytest.raises(DecoderError, match="expected 12"):
list(decode_packets(bad))
unknown = [InputPacket(0, 0, "camera", "main", 1, 1, "opaque", b"data")]
with pytest.raises(DecoderError, match="no decoder registered"):
list(DecoderPipeline().decode(unknown))
def test_y4m_local_video_decodes_across_input_chunks(tmp_path: Path) -> None:
video = tmp_path / "anonymous.y4m"
first, second = b"abcdef", b"ghijkl"
video.write_bytes(b"YUV4MPEG2 W2 H1 F25:1 C444\nFRAME\n" + first + b"FRAME\n" + second)
frames = list(decode_packets(local_packets(video)))
assert [frame.payload for frame in frames] == [first, second]
assert [frame.timestamp_ns for frame in frames] == [0, 40_000_000]
assert all(frame.pixel_format == "yuv444p" for frame in frames)
assert all((frame.width, frame.height) == (2, 1) for frame in frames)
assert all(frame.profile_id == "archive" for frame in frames)
def test_y4m_clean_eof_and_cancellation_are_normal(tmp_path: Path) -> None:
video = tmp_path / "empty.y4m"
video.write_bytes(b"YUV4MPEG2 W2 H1 F25:1 C444\n")
assert list(decode_packets(local_packets(video))) == []
token = CancellationToken()
token.cancel()
assert list(decode_packets(local_packets(video), token)) == []
@pytest.mark.parametrize(
("payload", "message"),
[
(b"not-video\n", "expected YUV4MPEG2"),
(b"YUV4MPEG2 W2 H1 F25:1 C420\n", "only C444"),
(b"YUV4MPEG2 W2 H1 F25:1 C444\nFRAME\nabc", "truncated Y4M frame"),
],
)
def test_y4m_damage_and_unsupported_content_are_clear(tmp_path: Path, payload: bytes, message: str) -> None:
video = tmp_path / "broken.y4m"
video.write_bytes(payload)
with pytest.raises(DecoderError, match=message):
list(decode_packets(local_packets(video)))
def test_y4m_profile_dimension_mismatch_is_rejected(tmp_path: Path) -> None:
video = tmp_path / "mismatch.y4m"
video.write_bytes(b"YUV4MPEG2 W2 H1 F25:1 C444\n")
with pytest.raises(DecoderError, match="do not match"):
list(decode_packets(local_packets(video, width=3)))
+5
View File
@@ -0,0 +1,5 @@
# Brain decode fixtures
Decode tests generate tiny anonymous YUV4MPEG2 streams at runtime. Do not add
customer recordings, camera credentials, machine-specific codec paths, or
large model/media artifacts to this directory.
+7
View File
@@ -0,0 +1,7 @@
# Brain input fixtures
This directory may contain only synthetic or anonymous fixtures. Do not add
customer video, camera credentials, personal data, or machine-specific paths.
The current tests generate their tiny local-file payload at runtime so the
repository does not carry a file that could be mistaken for customer media.
+5
View File
@@ -0,0 +1,5 @@
# Brain vision fixtures
Vision tests create anonymous geometric RGB frames in memory. Never add faces,
customer recordings, biometric templates, camera credentials, or unreviewed
model weights to this directory.
+83
View File
@@ -0,0 +1,83 @@
from __future__ import annotations
from pathlib import Path
import pytest
from yovision_brain.config import parse_input_config
from yovision_brain.input import CancellationToken, InputError, LocalFileInput, SyntheticInput, build_input_source
def synthetic_config(seed: int = 9, frame_count: int = 3):
return parse_input_config(
{
"schema": "brain.internal.input/v1",
"logical_device_id": "camera-lab-01",
"profile": {"id": "main", "width": 3, "height": 2, "fps": 4},
"source": {"kind": "synthetic", "seed": seed, "frame_count": frame_count},
}
)
def local_config(path: Path, *, chunk_size: int = 4):
return parse_input_config(
{
"schema": "brain.internal.input/v1",
"logical_device_id": "local-video-01",
"profile": {"id": "archive", "width": 1920, "height": 1080, "fps": 25},
"source": {"kind": "local_file", "path": str(path), "chunk_size": chunk_size},
}
)
def test_synthetic_input_is_deterministic_and_carries_metadata() -> None:
first = list(SyntheticInput(synthetic_config()).packets())
second = list(build_input_source(synthetic_config()).packets())
different = list(SyntheticInput(synthetic_config(seed=10)).packets())
assert first == second
assert [packet.sequence for packet in first] == [0, 1, 2]
assert [packet.timestamp_ns for packet in first] == [0, 250_000_000, 500_000_000]
assert first[0].logical_device_id == "camera-lab-01"
assert first[0].profile_id == "main"
assert first[0].media_format == "rgb24"
assert len(first[0].payload) == 3 * 2 * 3
assert first[0].payload != different[0].payload
def test_synthetic_input_honors_cancellation() -> None:
token = CancellationToken()
packets = SyntheticInput(synthetic_config(frame_count=10)).packets(token)
assert next(packets).sequence == 0
token.cancel()
assert list(packets) == []
def test_local_file_input_reads_chunks_and_finishes_at_eof(tmp_path: Path) -> None:
video = tmp_path / "anonymous-fixture.bin"
video.write_bytes(b"abcdefghij")
source = LocalFileInput(local_config(video))
packets = list(source.packets())
assert source.source_label == "anonymous-fixture.bin"
assert [packet.payload for packet in packets] == [b"abcd", b"efgh", b"ij"]
assert [packet.sequence for packet in packets] == [0, 1, 2]
assert all(packet.timestamp_ns is None for packet in packets)
assert all(packet.media_format == "container-bytes" for packet in packets)
def test_local_file_input_honors_cancellation(tmp_path: Path) -> None:
video = tmp_path / "anonymous-fixture.bin"
video.write_bytes(b"abcdefghij")
token = CancellationToken()
packets = LocalFileInput(local_config(video)).packets(token)
assert next(packets).payload == b"abcd"
token.cancel()
assert list(packets) == []
def test_missing_file_error_is_actionable_without_absolute_path(tmp_path: Path) -> None:
missing = tmp_path / "missing-video.mp4"
with pytest.raises(InputError, match="local video source not found: missing-video.mp4") as caught:
list(LocalFileInput(local_config(missing)).packets())
assert str(tmp_path) not in str(caught.value)
@@ -0,0 +1,69 @@
from __future__ import annotations
import pytest
from yovision_brain.decode import DecodedFrame
from yovision_brain.vision import (
BoundingBox,
Detection,
LumaBlobDetector,
SingleStreamTracker,
TorchLumaBlobDetector,
)
def frame(payload: bytes, *, sequence: int = 0, width: int = 4, height: int = 3) -> DecodedFrame:
return DecodedFrame(sequence, sequence * 40_000_000, "camera", "main", width, height, "rgb24", payload)
def rgb(values: list[int]) -> bytes:
return b"".join(bytes((value, value, value)) for value in values)
def detection(left: int, top: int, right: int, bottom: int) -> Detection:
return Detection(BoundingBox(left, top, right, bottom), "anonymous_target", 0.9)
def test_detector_emits_only_anonymous_observations() -> None:
payload = rgb([0, 255, 255, 0, 0, 255, 255, 0, 0, 0, 0, 0])
result = LumaBlobDetector(minimum_area=2).detect(frame(payload))
assert result == (Detection(BoundingBox(1, 0, 3, 2), "anonymous_target", 1.0),)
assert LumaBlobDetector.metadata.weights == "none"
assert "external model license" in LumaBlobDetector.metadata.license
def test_empty_frame_has_no_detection() -> None:
assert LumaBlobDetector().detect(frame(rgb([0] * 12))) == ()
def test_tracker_keeps_session_id_across_motion_and_short_occlusion() -> None:
tracker = SingleStreamTracker(iou_threshold=0.1, max_missed=2)
first = tracker.update((detection(0, 0, 3, 3),), frame_sequence=0, timestamp_ns=0)
assert first[0].track_id == "track-000001"
assert tracker.update((), frame_sequence=1, timestamp_ns=1) == ()
resumed = tracker.update((detection(1, 0, 4, 3),), frame_sequence=2, timestamp_ns=2)
assert resumed[0].track_id == "track-000001"
assert tracker.finish() == ("track-000001",)
def test_disappeared_track_ends_and_new_target_gets_new_id() -> None:
tracker = SingleStreamTracker(max_missed=1)
first = tracker.update((detection(0, 0, 2, 2),), frame_sequence=0, timestamp_ns=0)
tracker.update((), frame_sequence=1, timestamp_ns=1)
tracker.update((), frame_sequence=2, timestamp_ns=2)
second = tracker.update((detection(0, 0, 2, 2),), frame_sequence=3, timestamp_ns=3)
assert first[0].track_id == "track-000001"
assert second[0].track_id == "track-000002"
def test_track_ids_are_session_local() -> None:
one = SingleStreamTracker().update((detection(0, 0, 1, 1),), frame_sequence=0, timestamp_ns=0)
two = SingleStreamTracker().update((detection(0, 0, 1, 1),), frame_sequence=0, timestamp_ns=0)
assert one[0].track_id == two[0].track_id == "track-000001"
def test_torch_backend_cpu_smoke_uses_no_external_weights() -> None:
pytest.importorskip("torch")
result = TorchLumaBlobDetector().detect(frame(rgb([0, 255] + [0] * 10)))
assert result[0].category == "anonymous_target"
assert TorchLumaBlobDetector.metadata.weights == "none"
@@ -0,0 +1,22 @@
package router
import (
"github.com/gin-gonic/gin"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/ops_alert"
"git.ilapage.cn/ila/yovision/Sense/server/common/actions"
"git.ilapage.cn/ila/yovision/Sense/server/common/middleware"
)
func init() { routerCheckRole = append(routerCheckRole, registerSenseOpsAlertRouter) }
func registerSenseOpsAlertRouter(v1 *gin.RouterGroup, auth *jwt.GinJWTMiddleware) {
api := &ops_alert.API{}
r := v1.Group("/ops-alerts").Use(auth.MiddlewareFunc()).Use(middleware.AuthCheckRole()).Use(actions.PermissionAction())
r.GET("", api.List)
r.GET("/:id", api.Get)
r.POST("/evaluate", api.Evaluate)
r.POST("/:id/acknowledge", api.Acknowledge)
r.POST("/:id/recover", api.Recover)
}
@@ -0,0 +1,32 @@
package router
import (
"github.com/gin-gonic/gin"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
"net/http"
"testing"
)
func TestSenseOpsAlertRoutes(t *testing.T) {
gin.SetMode(gin.TestMode)
engine := gin.New()
registerSenseOpsAlertRouter(engine.Group("/api/v1"), &jwt.GinJWTMiddleware{})
wanted := map[string]bool{
http.MethodGet + " /api/v1/ops-alerts": false,
http.MethodGet + " /api/v1/ops-alerts/:id": false,
http.MethodPost + " /api/v1/ops-alerts/evaluate": false,
http.MethodPost + " /api/v1/ops-alerts/:id/acknowledge": false,
http.MethodPost + " /api/v1/ops-alerts/:id/recover": false,
}
for _, route := range engine.Routes() {
key := route.Method + " " + route.Path
if _, ok := wanted[key]; ok {
wanted[key] = true
}
}
for route, found := range wanted {
if !found {
t.Fatalf("route not registered: %s", route)
}
}
}
+121
View File
@@ -0,0 +1,121 @@
package ops_alert
import (
"errors"
"git.ilapage.cn/ila/yovision/Sense/server/common"
"github.com/gin-gonic/gin"
"github.com/gin-gonic/gin/binding"
"github.com/go-admin-team/go-admin-core/sdk/api"
"github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth/user"
coreService "github.com/go-admin-team/go-admin-core/sdk/service"
"net/http"
"time"
)
const auditSuccess, auditFailure = "1", "2"
type API struct{ api.Api }
func (e *API) service(c *gin.Context) (*Service, error) {
base := coreService.Service{}
if err := e.MakeContext(c).MakeOrm().MakeService(&base).Errors; err != nil {
return nil, err
}
return NewService(base.Orm), nil
}
func (e *API) List(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
request := PageRequest{}
if err = e.MakeContext(c).Bind(&request).Errors; err != nil {
e.audit(c, service, "List", auditFailure, "告警查询条件格式不正确")
e.Error(http.StatusBadRequest, err, "查询条件格式不正确")
return
}
response, err := service.List(request)
if err != nil {
e.audit(c, service, "List", auditFailure, "告警列表查询失败")
e.writeError(err)
return
}
e.audit(c, service, "List", auditSuccess, "读取运维告警列表")
e.OK(response, "查询成功")
}
func (e *API) Get(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
response, err := service.Get(c.Param("id"))
if err != nil {
e.audit(c, service, "Get", auditFailure, "告警详情查询失败")
e.writeError(err)
return
}
e.audit(c, service, "Get", auditSuccess, "读取运维告警详情 "+response.Alert.ID)
e.OK(response, "查询成功")
}
func (e *API) Evaluate(c *gin.Context) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
response, err := service.EvaluateSources(c.Request.Context())
if err != nil {
e.audit(c, service, "Evaluate", auditFailure, "健康事实刷新失败")
e.writeError(err)
return
}
e.audit(c, service, "Evaluate", auditSuccess, "刷新 Sense 内部健康事实")
e.OK(response, "状态已刷新")
}
func (e *API) Acknowledge(c *gin.Context) { e.action(c, "Acknowledge") }
func (e *API) Recover(c *gin.Context) { e.action(c, "Recover") }
func (e *API) action(c *gin.Context, action string) {
service, err := e.service(c)
if err != nil {
e.writeError(err)
return
}
request := ActionRequest{}
if err = e.MakeContext(c).Bind(&request, binding.JSON).Errors; err != nil {
e.audit(c, service, action, auditFailure, "告警操作请求格式不正确")
e.Error(http.StatusBadRequest, err, "请求格式不正确")
return
}
var item Alert
if action == "Acknowledge" {
item, err = service.Acknowledge(c.Request.Context(), c.Param("id"), request, user.GetUserId(c))
} else {
item, err = service.Recover(c.Request.Context(), c.Param("id"), request, user.GetUserId(c))
}
if err != nil {
e.audit(c, service, action, auditFailure, "运维告警状态操作被拒绝")
e.writeError(err)
return
}
e.audit(c, service, action, auditSuccess, action+" "+item.ID)
e.OK(item, "操作成功")
}
func (e *API) audit(c *gin.Context, service *Service, action, status, remark string) {
if err := WriteAudit(service.DB, Audit{Action: action, Method: c.Request.Method, Status: status, Username: user.GetUserName(c), UserID: user.GetUserId(c), ClientIP: common.GetClientIP(c), Route: c.FullPath(), Remark: remark, At: time.Now()}); err != nil {
api.GetRequestLogger(c).Errorf("ops alert audit failed: %s", err.Error())
}
}
func (e *API) writeError(err error) {
switch {
case errors.Is(err, ErrInvalidFilter), errors.Is(err, ErrInvalidReason):
e.Error(http.StatusBadRequest, err, err.Error())
case errors.Is(err, ErrVersionConflict), errors.Is(err, ErrInvalidTransition), errors.Is(err, ErrRecoveryWindowOpen):
e.Error(http.StatusConflict, err, err.Error())
case errors.Is(err, ErrAlertNotFound):
e.Error(http.StatusNotFound, err, err.Error())
default:
e.Error(http.StatusInternalServerError, err, "运维告警操作失败")
}
}
+19
View File
@@ -0,0 +1,19 @@
package ops_alert
import (
adminModels "git.ilapage.cn/ila/yovision/Sense/server/app/admin/models"
"gorm.io/gorm"
"time"
)
type Audit struct {
Action, Method, Status, Username, ClientIP, Route, Remark string
UserID int
At time.Time
}
func WriteAudit(db *gorm.DB, input Audit) error {
model := adminModels.SysOperaLog{Title: "运维告警", BusinessType: "other", Method: "ops_alert.API." + input.Action, RequestMethod: input.Method, OperatorType: "1", OperName: input.Username, OperUrl: input.Route, OperIp: input.ClientIP, Status: input.Status, OperTime: input.At.UTC(), Remark: input.Remark, CreatedAt: input.At.UTC(), UpdatedAt: input.At.UTC()}
model.CreateBy, model.UpdateBy = input.UserID, input.UserID
return db.Create(&model).Error
}
+37
View File
@@ -0,0 +1,37 @@
package ops_alert
import commonDto "git.ilapage.cn/ila/yovision/Sense/server/common/dto"
type PageRequest struct {
commonDto.Pagination `search:"-"`
AlertType string `form:"alertType"`
State string `form:"state"`
Keyword string `form:"keyword"`
}
type Summary struct {
Active int64 `json:"active"`
Unacknowledged int64 `json:"unacknowledged"`
Recovering int64 `json:"recovering"`
}
type PageResponse struct {
Summary Summary `json:"summary"`
List []Alert `json:"list"`
Count int64 `json:"count"`
}
type DetailResponse struct {
Alert Alert `json:"alert"`
Transitions []Transition `json:"transitions"`
}
type ActionRequest struct {
ExpectedVersion int64 `json:"expectedVersion" binding:"required,min=1"`
Reason string `json:"reason" binding:"required"`
}
type EvaluateResponse struct {
Observed int `json:"observed"`
Abnormal int `json:"abnormal"`
}
@@ -0,0 +1,62 @@
package ops_alert
import "time"
const (
StateUnacknowledged = "unacknowledged"
StateAcknowledged = "acknowledged"
StateRecovering = "recovering"
StateRecovered = "recovered"
TypeDeviceOffline = "device_offline"
TypeAuthentication = "authentication_failed"
TypeClockDrift = "clock_drift"
TypeReconciliation = "reconciliation_failed"
TypeMediaShard = "media_shard_failed"
TypeControlTunnel = "control_tunnel_failed"
)
type Alert struct {
ID string `gorm:"size:36;primaryKey" json:"id"`
Fingerprint string `gorm:"size:255;not null;uniqueIndex" json:"fingerprint"`
AlertType string `gorm:"size:32;not null;index" json:"alertType"`
Severity string `gorm:"size:16;not null;index" json:"severity"`
ObjectType string `gorm:"size:32;not null;index" json:"objectType"`
ObjectID string `gorm:"size:96;not null;index" json:"objectId"`
ObjectName string `gorm:"size:128;not null" json:"objectName"`
Location string `gorm:"size:255;not null;default:''" json:"location"`
State string `gorm:"size:24;not null;index" json:"state"`
Title string `gorm:"size:160;not null" json:"title"`
Detail string `gorm:"size:512;not null" json:"detail"`
NextAction string `gorm:"size:255;not null" json:"nextAction"`
LastSourceVersion string `gorm:"size:128;not null;default:''" json:"-"`
OccurrenceCount int `gorm:"not null;default:1" json:"occurrenceCount"`
Cycle int `gorm:"not null;default:1" json:"cycle"`
FirstSeenAt time.Time `gorm:"not null;index" json:"firstSeenAt"`
LastSeenAt time.Time `gorm:"not null;index" json:"lastSeenAt"`
HealthySince *time.Time `json:"healthySince,omitempty"`
AcknowledgedAt *time.Time `json:"acknowledgedAt,omitempty"`
AcknowledgedBy int `gorm:"not null;default:0" json:"acknowledgedBy,omitempty"`
RecoveredAt *time.Time `json:"recoveredAt,omitempty"`
RecoveredBy int `gorm:"not null;default:0" json:"recoveredBy,omitempty"`
OperationalOnly bool `gorm:"not null;default:true" json:"operationalOnly"`
Version int64 `gorm:"not null;default:1" json:"version"`
CreatedAt time.Time `gorm:"not null" json:"createdAt"`
UpdatedAt time.Time `gorm:"not null" json:"updatedAt"`
}
func (Alert) TableName() string { return "sense_ops_alerts" }
type Transition struct {
ID string `gorm:"size:36;primaryKey" json:"id"`
AlertID string `gorm:"size:36;not null;index" json:"alertId"`
Cycle int `gorm:"not null" json:"cycle"`
Action string `gorm:"size:32;not null;index" json:"action"`
FromState string `gorm:"size:24;not null;default:''" json:"fromState"`
ToState string `gorm:"size:24;not null" json:"toState"`
Reason string `gorm:"size:512;not null;default:''" json:"reason"`
ActorUserID int `gorm:"not null;default:0" json:"actorUserId"`
CreatedAt time.Time `gorm:"not null;index" json:"createdAt"`
}
func (Transition) TableName() string { return "sense_ops_alert_transitions" }
+326
View File
@@ -0,0 +1,326 @@
package ops_alert
import (
"context"
"errors"
"fmt"
"strings"
"time"
"unicode/utf8"
"github.com/google/uuid"
"gorm.io/gorm"
"gorm.io/gorm/clause"
deviceModels "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/edge_node"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
)
var (
ErrAlertNotFound = errors.New("运维告警不存在")
ErrInvalidFilter = errors.New("告警筛选条件不正确")
ErrInvalidReason = errors.New("处理说明需为 6 至 256 个字符")
ErrInvalidTransition = errors.New("当前告警状态不允许此操作")
ErrVersionConflict = errors.New("告警状态已变化,请刷新后重试")
ErrRecoveryWindowOpen = errors.New("健康恢复观察窗口尚未结束")
)
const defaultRecoveryWindow = 5 * time.Minute
type Service struct {
DB *gorm.DB
Now func() time.Time
RecoveryWindow time.Duration
}
func NewService(db *gorm.DB) *Service {
return &Service{DB: db, Now: time.Now, RecoveryWindow: defaultRecoveryWindow}
}
func (s *Service) List(request PageRequest) (PageResponse, error) {
if !validOptional(request.AlertType, alertTypes()) || !validOptional(request.State, states()) {
return PageResponse{}, ErrInvalidFilter
}
query := s.DB.Model(&Alert{})
if request.AlertType != "" {
query = query.Where("alert_type = ?", request.AlertType)
}
if request.State != "" {
query = query.Where("state = ?", request.State)
}
if keyword := strings.TrimSpace(request.Keyword); keyword != "" {
like := "%" + keyword + "%"
query = query.Where("object_name LIKE ? OR location LIKE ? OR title LIKE ? OR detail LIKE ?", like, like, like, like)
}
response := PageResponse{}
if err := query.Count(&response.Count).Error; err != nil {
return response, err
}
if err := query.Order("CASE WHEN state = 'unacknowledged' THEN 0 WHEN state = 'acknowledged' THEN 1 WHEN state = 'recovering' THEN 2 ELSE 3 END, last_seen_at DESC").Offset((request.GetPageIndex() - 1) * request.GetPageSize()).Limit(request.GetPageSize()).Find(&response.List).Error; err != nil {
return response, err
}
active := []string{StateUnacknowledged, StateAcknowledged, StateRecovering}
if err := s.DB.Model(&Alert{}).Where("state IN ?", active).Count(&response.Summary.Active).Error; err != nil {
return response, err
}
if err := s.DB.Model(&Alert{}).Where("state = ?", StateUnacknowledged).Count(&response.Summary.Unacknowledged).Error; err != nil {
return response, err
}
if err := s.DB.Model(&Alert{}).Where("state = ?", StateRecovering).Count(&response.Summary.Recovering).Error; err != nil {
return response, err
}
return response, nil
}
func (s *Service) Get(id string) (DetailResponse, error) {
response := DetailResponse{Transitions: []Transition{}}
if err := s.DB.First(&response.Alert, "id = ?", id).Error; errors.Is(err, gorm.ErrRecordNotFound) {
return response, ErrAlertNotFound
} else if err != nil {
return response, err
}
if err := s.DB.Where("alert_id = ?", id).Order("created_at DESC").Find(&response.Transitions).Error; err != nil {
return response, err
}
return response, nil
}
func (s *Service) Acknowledge(ctx context.Context, id string, request ActionRequest, actor int) (Alert, error) {
return s.transition(ctx, id, request, actor, StateUnacknowledged, StateAcknowledged, "acknowledge")
}
func (s *Service) Recover(ctx context.Context, id string, request ActionRequest, actor int) (Alert, error) {
return s.transition(ctx, id, request, actor, StateRecovering, StateRecovered, "recover")
}
func (s *Service) transition(ctx context.Context, id string, request ActionRequest, actor int, expectedState, targetState, action string) (Alert, error) {
reason := strings.TrimSpace(request.Reason)
if utf8.RuneCountInString(reason) < 6 || utf8.RuneCountInString(reason) > 256 {
return Alert{}, ErrInvalidReason
}
if request.ExpectedVersion < 1 {
return Alert{}, ErrVersionConflict
}
now := s.Now().UTC()
result := Alert{}
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var current Alert
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&current, "id = ?", id).Error; errors.Is(err, gorm.ErrRecordNotFound) {
return ErrAlertNotFound
} else if err != nil {
return err
}
if current.Version != request.ExpectedVersion {
return ErrVersionConflict
}
if current.State != expectedState {
return ErrInvalidTransition
}
if action == "recover" && (current.HealthySince == nil || now.Sub(current.HealthySince.UTC()) < s.RecoveryWindow) {
return ErrRecoveryWindowOpen
}
updates := map[string]any{"state": targetState, "version": current.Version + 1, "updated_at": now}
if action == "acknowledge" {
updates["acknowledged_at"], updates["acknowledged_by"] = now, actor
} else {
updates["recovered_at"], updates["recovered_by"] = now, actor
}
write := tx.Model(&Alert{}).Where("id = ? AND version = ? AND state = ?", id, current.Version, expectedState).Updates(updates)
if write.Error != nil {
return write.Error
}
if write.RowsAffected != 1 {
return ErrVersionConflict
}
if err := tx.Create(&Transition{ID: uuid.NewString(), AlertID: id, Cycle: current.Cycle, Action: action, FromState: expectedState, ToState: targetState, Reason: reason, ActorUserID: actor, CreatedAt: now}).Error; err != nil {
return err
}
return tx.First(&result, "id = ?", id).Error
})
return result, err
}
type observation struct {
Fingerprint, AlertType, Severity, ObjectType, ObjectID, ObjectName, Location, Title, Detail, NextAction, SourceVersion string
Abnormal bool
}
func (s *Service) EvaluateSources(ctx context.Context) (EvaluateResponse, error) {
items, err := s.sourceObservations(ctx)
if err != nil {
return EvaluateResponse{}, err
}
response := EvaluateResponse{Observed: len(items)}
for _, item := range items {
if item.Abnormal {
response.Abnormal++
}
if err = s.observe(ctx, item); err != nil {
return response, err
}
}
return response, nil
}
func (s *Service) observe(ctx context.Context, item observation) error {
now := s.Now().UTC()
return s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var current Alert
err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("fingerprint = ?", item.Fingerprint).First(&current).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
if !item.Abnormal {
return nil
}
current = Alert{ID: uuid.NewString(), Fingerprint: item.Fingerprint, AlertType: item.AlertType, Severity: item.Severity, ObjectType: item.ObjectType, ObjectID: item.ObjectID, ObjectName: item.ObjectName, Location: item.Location, State: StateUnacknowledged, Title: item.Title, Detail: item.Detail, NextAction: item.NextAction, LastSourceVersion: item.SourceVersion, OccurrenceCount: 1, Cycle: 1, FirstSeenAt: now, LastSeenAt: now, OperationalOnly: true, Version: 1, CreatedAt: now, UpdatedAt: now}
if err = tx.Create(&current).Error; err != nil {
return err
}
return tx.Create(&Transition{ID: uuid.NewString(), AlertID: current.ID, Cycle: 1, Action: "detected", ToState: StateUnacknowledged, Reason: "健康事实首次满足告警规则", CreatedAt: now}).Error
}
if err != nil {
return err
}
if item.Abnormal {
updates := map[string]any{"severity": item.Severity, "object_name": item.ObjectName, "location": item.Location, "title": item.Title, "detail": item.Detail, "next_action": item.NextAction, "last_seen_at": now, "updated_at": now, "healthy_since": nil}
if item.SourceVersion != current.LastSourceVersion {
updates["last_source_version"], updates["occurrence_count"] = item.SourceVersion, current.OccurrenceCount+1
}
if current.State == StateRecovered {
updates["state"], updates["cycle"], updates["first_seen_at"], updates["occurrence_count"], updates["acknowledged_at"], updates["acknowledged_by"], updates["recovered_at"], updates["recovered_by"] = StateUnacknowledged, current.Cycle+1, now, 1, nil, 0, nil, 0
if err = tx.Create(&Transition{ID: uuid.NewString(), AlertID: current.ID, Cycle: current.Cycle + 1, Action: "reopened", FromState: StateRecovered, ToState: StateUnacknowledged, Reason: "健康事实再次异常", CreatedAt: now}).Error; err != nil {
return err
}
} else if current.State == StateRecovering {
target := StateUnacknowledged
if current.AcknowledgedAt != nil {
target = StateAcknowledged
}
updates["state"] = target
if err = tx.Create(&Transition{ID: uuid.NewString(), AlertID: current.ID, Cycle: current.Cycle, Action: "relapsed", FromState: StateRecovering, ToState: target, Reason: "恢复观察期间再次异常", CreatedAt: now}).Error; err != nil {
return err
}
}
updates["version"] = current.Version + 1
return tx.Model(&Alert{}).Where("id = ? AND version = ?", current.ID, current.Version).Updates(updates).Error
}
if current.State == StateUnacknowledged || current.State == StateAcknowledged {
if err = tx.Create(&Transition{ID: uuid.NewString(), AlertID: current.ID, Cycle: current.Cycle, Action: "health_restored", FromState: current.State, ToState: StateRecovering, Reason: "健康事实已恢复,进入稳定观察窗口", CreatedAt: now}).Error; err != nil {
return err
}
return tx.Model(&Alert{}).Where("id = ? AND version = ?", current.ID, current.Version).Updates(map[string]any{"state": StateRecovering, "healthy_since": now, "version": current.Version + 1, "updated_at": now}).Error
}
return nil
})
}
type admissionProjection struct {
DeviceID, Status, Detail string
CheckedAt time.Time
}
func (admissionProjection) TableName() string { return "sense_admission_results" }
func (s *Service) sourceObservations(ctx context.Context) ([]observation, error) {
now := s.Now().UTC()
result := []observation{}
var devices []deviceModels.Device
if err := s.DB.WithContext(ctx).Find(&devices).Error; err != nil {
return nil, err
}
var admissions []admissionProjection
if err := s.DB.WithContext(ctx).Find(&admissions).Error; err != nil && !isMissingTable(err) {
return nil, err
}
byDevice := map[string]admissionProjection{}
for _, item := range admissions {
byDevice[item.DeviceID] = item
}
for _, device := range devices {
if device.Status == deviceModels.StatusDisabled {
continue
}
state, detail, observed := device.AdapterStatus, "尚未获得设备接入结果", device.UpdatedAt
if item, ok := byDevice[device.ID]; ok {
state, detail, observed = item.Status, item.Detail, item.CheckedAt
}
text := strings.ToLower(state + " " + detail)
version := fmt.Sprintf("%d:%d", device.Version, observed.UnixNano())
base := observation{ObjectType: "device", ObjectID: device.ID, ObjectName: device.Name, Location: device.Location, SourceVersion: version}
result = append(result,
withRule(base, TypeDeviceOffline, "high", "设备离线", detail, "检查设备供电、网络和地址后重新发现", containsAny(text, "offline", "unreachable", "timeout", "离线", "无法连接", "超时")),
withRule(base, TypeAuthentication, "high", "设备认证失败", detail, "更新设备凭据后重新执行接入验证", containsAny(text, "authentication_failed", "unauthorized", "auth", "认证", "凭据")),
withRule(base, TypeClockDrift, "medium", "设备时间漂移", detail, "校准设备时间后重新执行接入验证", containsAny(text, "clock_skew", "clock_drift", "time drift", "时间漂移", "时钟")),
)
}
var routes []media.Route
if err := s.DB.WithContext(ctx).Find(&routes).Error; err != nil && !isMissingTable(err) {
return nil, err
}
for _, route := range routes {
converged := (route.Desired == media.DesiredRunning && (route.Actual == "ready" || route.Actual == "waiting")) || (route.Desired == media.DesiredStopped && route.Actual == "stopped")
base := observation{ObjectType: "media_route", ObjectID: route.ID, ObjectName: route.Path, SourceVersion: fmt.Sprintf("%d", route.Version)}
result = append(result, withRule(base, TypeReconciliation, "medium", "媒体状态未收敛", route.Detail, "在运维中心核对期望态并执行受控重试", !converged))
}
var shards []media_shard.Shard
if err := s.DB.WithContext(ctx).Find(&shards).Error; err != nil && !isMissingTable(err) {
return nil, err
}
for _, shard := range shards {
if shard.Status == media_shard.StatusDisabled {
continue
}
stale := shard.LastProbeAt == nil || now.Sub(shard.LastProbeAt.UTC()) > 30*time.Second
base := observation{ObjectType: "media_shard", ObjectID: shard.ID, ObjectName: shard.Name, SourceVersion: fmt.Sprintf("%d:%d", shard.ConfigVersion, timeValue(shard.LastProbeAt))}
result = append(result, withRule(base, TypeMediaShard, "high", "媒体分片异常", shard.Detail, "检查 MediaMTX 进程和 Control API 后重新探测", shard.Status == media_shard.StatusFailed || stale))
}
var nodes []edge_node.Node
if err := s.DB.WithContext(ctx).Find(&nodes).Error; err != nil && !isMissingTable(err) {
return nil, err
}
for _, node := range nodes {
base := observation{ObjectType: "edge_node", ObjectID: node.ID, ObjectName: node.Name, Location: node.Location, SourceVersion: fmt.Sprintf("%d:%d", node.ProjectionVersion, node.LastHeartbeatAt.UnixNano())}
result = append(result,
withRule(base, TypeDeviceOffline, "high", "边缘节点离线", "心跳超过 90 秒未更新", "检查节点进程、网络和机器身份后等待心跳恢复", now.Sub(node.LastHeartbeatAt.UTC()) > edge_node.HeartbeatTimeout),
withRule(base, TypeControlTunnel, "high", "控制隧道异常", node.ControlTunnelDetail, "检查节点到 Sense 的控制通道,不影响业务预警数据", node.ControlTunnelStatus != edge_node.ChannelReady),
)
}
return result, nil
}
func withRule(base observation, typ, severity, title, detail, next string, abnormal bool) observation {
base.Fingerprint = typ + ":" + base.ObjectType + ":" + base.ObjectID
base.AlertType = typ
base.Severity = severity
base.Title = title
base.Detail = detail
base.NextAction = next
base.Abnormal = abnormal
return base
}
func containsAny(value string, needles ...string) bool {
for _, needle := range needles {
if strings.Contains(value, needle) {
return true
}
}
return false
}
func isMissingTable(err error) bool {
text := strings.ToLower(err.Error())
return strings.Contains(text, "no such table") || strings.Contains(text, "does not exist")
}
func timeValue(value *time.Time) int64 {
if value == nil {
return 0
}
return value.UnixNano()
}
func validOptional(value string, allowed map[string]bool) bool { return value == "" || allowed[value] }
func states() map[string]bool {
return map[string]bool{StateUnacknowledged: true, StateAcknowledged: true, StateRecovering: true, StateRecovered: true}
}
func alertTypes() map[string]bool {
return map[string]bool{TypeDeviceOffline: true, TypeAuthentication: true, TypeClockDrift: true, TypeReconciliation: true, TypeMediaShard: true, TypeControlTunnel: true}
}
@@ -0,0 +1,198 @@
package ops_alert
import (
"context"
"errors"
"testing"
"time"
"github.com/google/uuid"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
adminModels "git.ilapage.cn/ila/yovision/Sense/server/app/admin/models"
deviceModels "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/edge_node"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media_shard"
)
func opsAlertTestService(t *testing.T) (*Service, *gorm.DB, time.Time) {
t.Helper()
db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&Alert{}, &Transition{}, &deviceModels.Device{}, &admissionProjection{}, &media.Route{}, &media_shard.Shard{}, &edge_node.Node{}, &adminModels.SysOperaLog{}); err != nil {
t.Fatal(err)
}
now := time.Date(2026, 8, 28, 8, 0, 0, 0, time.UTC)
service := NewService(db)
service.Now = func() time.Time { return now }
return service, db, now
}
func TestEvaluateSourcesGeneratesSixTypesAndDeduplicates(t *testing.T) {
service, db, now := opsAlertTestService(t)
devices := []deviceModels.Device{
{ID: "offline", Name: "东门摄像机", Location: "东门", Modality: "video", CapabilitiesJSON: "[]", Status: "pending", AdapterStatus: "verification_failed", Version: 1},
{ID: "auth", Name: "仓库摄像机", Location: "仓库", Modality: "video", CapabilitiesJSON: "[]", Status: "pending", AdapterStatus: "verification_failed", Version: 2},
{ID: "clock", Name: "南门摄像机", Location: "南门", Modality: "video", CapabilitiesJSON: "[]", Status: "pending", AdapterStatus: "verification_failed", Version: 3},
}
if err := db.Create(&devices).Error; err != nil {
t.Fatal(err)
}
results := []admissionProjection{
{DeviceID: "offline", Status: "device_offline", Detail: "设备离线,无法连接", CheckedAt: now},
{DeviceID: "auth", Status: "authentication_failed", Detail: "设备认证失败", CheckedAt: now},
{DeviceID: "clock", Status: "clock_skew", Detail: "设备时间漂移", CheckedAt: now},
}
if err := db.Create(&results).Error; err != nil {
t.Fatal(err)
}
if err := db.Create(&media.Route{ID: "route-1", DeviceID: "auth", ProfileToken: "main", Path: "sense_auth", Desired: "running", Actual: "apply_failed", Detail: "配置未收敛", Version: 4, UpdatedAt: now}).Error; err != nil {
t.Fatal(err)
}
probe := now
if err := db.Create(&media_shard.Shard{ID: "shard-1", Name: "媒体服务 A", Mode: "local", ControlAPI: "http://127.0.0.1:9997", Capacity: 16, Status: media_shard.StatusFailed, Detail: "Control API 不可用", LastProbeAt: &probe, ConfigVersion: 1, CreatedAt: now, UpdatedAt: now}).Error; err != nil {
t.Fatal(err)
}
if err := db.Create(&edge_node.Node{ID: "node-1", Name: "主楼边缘节点", Location: "机房", StartedAt: now.Add(-time.Hour), LastHeartbeatAt: now.Add(-10 * time.Second), LastCollectedAt: now.Add(-10 * time.Second), ControlTunnelStatus: edge_node.ChannelUnavailable, ControlTunnelDetail: "TLS 隧道断开", VideoPlaneStatus: edge_node.ChannelReady, ProjectionVersion: 7}).Error; err != nil {
t.Fatal(err)
}
response, err := service.EvaluateSources(context.Background())
if err != nil {
t.Fatal(err)
}
if response.Abnormal != 6 {
t.Fatalf("abnormal=%d want 6", response.Abnormal)
}
var alerts []Alert
if err = db.Find(&alerts).Error; err != nil {
t.Fatal(err)
}
if len(alerts) != 6 {
t.Fatalf("alerts=%d want 6", len(alerts))
}
types := map[string]bool{}
for _, item := range alerts {
types[item.AlertType] = true
if !item.OperationalOnly {
t.Fatal("alert crossed operational boundary")
}
}
for _, typ := range []string{TypeDeviceOffline, TypeAuthentication, TypeClockDrift, TypeReconciliation, TypeMediaShard, TypeControlTunnel} {
if !types[typ] {
t.Fatalf("missing type %s", typ)
}
}
if _, err = service.EvaluateSources(context.Background()); err != nil {
t.Fatal(err)
}
var count int64
db.Model(&Alert{}).Count(&count)
if count != 6 {
t.Fatalf("duplicate active alerts: %d", count)
}
var offline Alert
if err = db.First(&offline, "fingerprint = ?", TypeDeviceOffline+":device:offline").Error; err != nil {
t.Fatal(err)
}
if offline.OccurrenceCount != 1 {
t.Fatalf("unchanged source counted twice: %d", offline.OccurrenceCount)
}
}
func TestAcknowledgeRecoverAndReopenPreserveOneAlert(t *testing.T) {
service, db, now := opsAlertTestService(t)
device := deviceModels.Device{ID: "auth", Name: "仓库摄像机", Modality: "video", CapabilitiesJSON: "[]", Status: "pending", AdapterStatus: "verification_failed", Version: 1}
if err := db.Create(&device).Error; err != nil {
t.Fatal(err)
}
result := admissionProjection{DeviceID: "auth", Status: "authentication_failed", Detail: "认证失败", CheckedAt: now}
if err := db.Create(&result).Error; err != nil {
t.Fatal(err)
}
if _, err := service.EvaluateSources(context.Background()); err != nil {
t.Fatal(err)
}
var alert Alert
if err := db.First(&alert, "alert_type = ?", TypeAuthentication).Error; err != nil {
t.Fatal(err)
}
ack, err := service.Acknowledge(context.Background(), alert.ID, ActionRequest{ExpectedVersion: alert.Version, Reason: "已安排现场人员更新设备凭据"}, 7)
if err != nil {
t.Fatal(err)
}
if ack.State != StateAcknowledged || ack.AcknowledgedBy != 7 {
t.Fatalf("unexpected acknowledgement: %+v", ack)
}
if _, err = service.Acknowledge(context.Background(), alert.ID, ActionRequest{ExpectedVersion: alert.Version, Reason: "并发重复确认不应成功"}, 8); !errors.Is(err, ErrVersionConflict) {
t.Fatalf("want version conflict, got %v", err)
}
if err = db.Model(&admissionProjection{}).Where("device_id = ?", "auth").Updates(map[string]any{"status": "ready", "detail": "接入验证完成", "checked_at": now.Add(time.Minute)}).Error; err != nil {
t.Fatal(err)
}
service.Now = func() time.Time { return now.Add(time.Minute) }
if _, err = service.EvaluateSources(context.Background()); err != nil {
t.Fatal(err)
}
if err = db.First(&alert, "id = ?", alert.ID).Error; err != nil {
t.Fatal(err)
}
if alert.State != StateRecovering || alert.HealthySince == nil {
t.Fatalf("want recovering: %+v", alert)
}
if _, err = service.Recover(context.Background(), alert.ID, ActionRequest{ExpectedVersion: alert.Version, Reason: "恢复窗口尚未结束,不能关闭"}, 7); !errors.Is(err, ErrRecoveryWindowOpen) {
t.Fatalf("want recovery window error, got %v", err)
}
service.Now = func() time.Time { return now.Add(7 * time.Minute) }
recovered, err := service.Recover(context.Background(), alert.ID, ActionRequest{ExpectedVersion: alert.Version, Reason: "设备已稳定在线超过恢复观察窗口"}, 7)
if err != nil {
t.Fatal(err)
}
if recovered.State != StateRecovered {
t.Fatalf("state=%s", recovered.State)
}
if _, err = service.EvaluateSources(context.Background()); err != nil {
t.Fatal(err)
}
if err = db.Model(&admissionProjection{}).Where("device_id = ?", "auth").Updates(map[string]any{"status": "authentication_failed", "detail": "认证再次失败", "checked_at": now.Add(8 * time.Minute)}).Error; err != nil {
t.Fatal(err)
}
service.Now = func() time.Time { return now.Add(8 * time.Minute) }
if _, err = service.EvaluateSources(context.Background()); err != nil {
t.Fatal(err)
}
var count int64
db.Model(&Alert{}).Where("fingerprint = ?", recovered.Fingerprint).Count(&count)
if count != 1 {
t.Fatalf("reopen duplicated alert: %d", count)
}
if err = db.First(&alert, "id = ?", alert.ID).Error; err != nil {
t.Fatal(err)
}
if alert.State != StateUnacknowledged || alert.Cycle != 2 {
t.Fatalf("unexpected reopened alert: %+v", alert)
}
var transitions int64
db.Model(&Transition{}).Where("alert_id = ?", alert.ID).Count(&transitions)
if transitions < 5 {
t.Fatalf("transitions=%d", transitions)
}
}
func TestAuditIsStoredOnlyInGoAdminOperationLog(t *testing.T) {
_, db, now := opsAlertTestService(t)
if err := WriteAudit(db, Audit{Action: "Acknowledge", Method: "POST", Status: "1", Username: "operator", UserID: 9, ClientIP: "127.0.0.1", Route: "/api/v1/ops-alerts/:id/acknowledge", Remark: "确认运维告警", At: now}); err != nil {
t.Fatal(err)
}
var log adminModels.SysOperaLog
if err := db.First(&log).Error; err != nil {
t.Fatal(err)
}
if log.Title != "运维告警" || log.CreateBy != 9 {
t.Fatalf("unexpected audit: %+v", log)
}
}
@@ -0,0 +1,74 @@
package version
import (
"fmt"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/ops_alert"
"git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration"
migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"runtime"
)
func init() {
_, fileName, _, _ := runtime.Caller(0)
migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateSenseOpsAlert)
}
func migrateSenseOpsAlert(db *gorm.DB, version string) error {
return db.Transaction(func(tx *gorm.DB) error {
if err := tx.AutoMigrate(&ops_alert.Alert{}, &ops_alert.Transition{}); err != nil {
return err
}
root, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: senseLayoutMenuName, Title: "视频感知", Icon: "video-camera", Path: "/sense", MenuType: "M", Component: "Layout", Sort: 5, Visible: "0", IsFrame: "1"})
if err != nil {
return err
}
page, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseOpsAlert", Title: "运维告警", Icon: "warning", Path: "ops-alert", Paths: fmt.Sprintf("/0/%d", root.MenuId), MenuType: "C", Permission: "sense:ops-alert:list", ParentId: root.MenuId, Component: "/sense/ops-alert/index", Sort: 11, Visible: "0", IsFrame: "1"})
if err != nil {
return err
}
detail, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseOpsAlertDetail", Title: "查看告警详情", MenuType: "F", Action: "GET", Permission: "sense:ops-alert:detail", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 1, Visible: "1", IsFrame: "1"})
if err != nil {
return err
}
refresh, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseOpsAlertEvaluate", Title: "刷新告警状态", MenuType: "F", Action: "POST", Permission: "sense:ops-alert:evaluate", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 2, Visible: "1", IsFrame: "1"})
if err != nil {
return err
}
ack, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseOpsAlertAcknowledge", Title: "确认运维告警", MenuType: "F", Action: "POST", Permission: "sense:ops-alert:acknowledge", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 3, Visible: "1", IsFrame: "1"})
if err != nil {
return err
}
recover, err := ensureDeviceMenu(tx, migrationModels.SysMenu{MenuName: "SenseOpsAlertRecover", Title: "确认告警恢复", MenuType: "F", Action: "POST", Permission: "sense:ops-alert:recover", ParentId: page.MenuId, Paths: fmt.Sprintf("/0/%d/%d", root.MenuId, page.MenuId), Sort: 4, Visible: "1", IsFrame: "1"})
if err != nil {
return err
}
readPolicies := [][2]string{{"/api/v1/ops-alerts", "GET"}, {"/api/v1/ops-alerts/:id", "GET"}}
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
if err = attachDeviceRole(tx, role, []migrationModels.SysMenu{page, detail}); err != nil {
return err
}
for _, policy := range readPolicies {
if err = tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&deviceCasbinRule{Ptype: "p", V0: role, V1: policy[0], V2: policy[1]}).Error; err != nil {
return err
}
}
}
for _, role := range []string{"implementation_operator", "site_admin"} {
if err = attachDeviceRole(tx, role, []migrationModels.SysMenu{refresh, ack, recover}); err != nil {
return err
}
for _, policy := range [][2]string{{"/api/v1/ops-alerts/evaluate", "POST"}, {"/api/v1/ops-alerts/:id/acknowledge", "POST"}, {"/api/v1/ops-alerts/:id/recover", "POST"}} {
if err = tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&deviceCasbinRule{Ptype: "p", V0: role, V1: policy[0], V2: policy[1]}).Error; err != nil {
return err
}
}
}
if err = rebuildSenseMenuPaths(tx, root.MenuId, "/0"); err != nil {
return err
}
return tx.Create(&common.Migration{Version: version}).Error
})
}
@@ -0,0 +1,39 @@
package version
import (
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/ops_alert"
migrationModels "git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration/models"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"testing"
)
func TestSenseOpsAlertMigrationRBAC(t *testing.T) {
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&migrationModels.SysRole{}, &migrationModels.SysMenu{}, &deviceCasbinRule{}, &common.Migration{}); err != nil {
t.Fatal(err)
}
for _, role := range []string{"implementation_operator", "site_admin", "viewer"} {
if err = db.Create(&migrationModels.SysRole{RoleName: role, RoleKey: role, Status: "2"}).Error; err != nil {
t.Fatal(err)
}
}
const version = "2026082816000_ops_alert.go"
if err = migrateSenseOpsAlert(db, version); err != nil {
t.Fatal(err)
}
var pages, viewerWrites, operatorWrites int64
db.Model(&migrationModels.SysMenu{}).Where("permission = ?", "sense:ops-alert:list").Count(&pages)
db.Model(&deviceCasbinRule{}).Where("v0 = ? AND v1 LIKE ? AND v2 = ?", "viewer", "/api/v1/ops-alerts%", "POST").Count(&viewerWrites)
db.Model(&deviceCasbinRule{}).Where("v0 = ? AND v1 LIKE ? AND v2 = ?", "implementation_operator", "/api/v1/ops-alerts%", "POST").Count(&operatorWrites)
if pages != 1 || viewerWrites != 0 || operatorWrites != 3 {
t.Fatalf("unexpected RBAC page=%d viewerWrites=%d operatorWrites=%d", pages, viewerWrites, operatorWrites)
}
if !db.Migrator().HasTable(&ops_alert.Alert{}) || !db.Migrator().HasTable(&ops_alert.Transition{}) {
t.Fatal("ops alert tables missing")
}
}
+21
View File
@@ -0,0 +1,21 @@
import request from '@/utils/request'
export function listOpsAlerts(query) {
return request({ url: '/api/v1/ops-alerts', method: 'get', params: query })
}
export function getOpsAlert(id) {
return request({ url: `/api/v1/ops-alerts/${encodeURIComponent(id)}`, method: 'get' })
}
export function evaluateOpsAlerts() {
return request({ url: '/api/v1/ops-alerts/evaluate', method: 'post' })
}
export function acknowledgeOpsAlert(id, data) {
return request({ url: `/api/v1/ops-alerts/${encodeURIComponent(id)}/acknowledge`, method: 'post', data })
}
export function recoverOpsAlert(id, data) {
return request({ url: `/api/v1/ops-alerts/${encodeURIComponent(id)}/recover`, method: 'post', data })
}
@@ -0,0 +1,109 @@
<template>
<BasicLayout>
<template #wrapper>
<el-card class="box-card">
<div class="page-header">
<div>
<h3>运维告警</h3>
<p>处理设备、媒体服务和边缘节点的运行异常。</p>
</div>
<el-button v-permisaction="['sense:ops-alert:evaluate']" :icon="Refresh" :loading="refreshing" @click="refreshSources">刷新状态</el-button>
</div>
<el-alert class="boundary-alert" type="info" :closable="false" show-icon title="这里只处理系统运行问题">
运维告警不会创建本地安全事件,也不会发送到 Bell 作为业务预警;确认告警仅表示已有人处理,不代表故障已经恢复。
</el-alert>
<el-row :gutter="12" class="summary-row" aria-label="告警概览">
<el-col :xs="24" :sm="8"><div class="summary-item"><span>活动告警</span><strong>{{ summary.active }}</strong></div></el-col>
<el-col :xs="24" :sm="8"><div class="summary-item"><span>待确认</span><strong class="danger-number">{{ summary.unacknowledged }}</strong></div></el-col>
<el-col :xs="24" :sm="8"><div class="summary-item"><span>恢复观察</span><strong class="primary-number">{{ summary.recovering }}</strong></div></el-col>
</el-row>
<el-form :model="query" label-width="76px" class="filter-form" @submit.prevent="search">
<el-form-item label="告警类型">
<el-select v-model="query.alertType" clearable placeholder="全部类型">
<el-option v-for="item in typeOptions" :key="item.value" :label="item.label" :value="item.value" />
</el-select>
</el-form-item>
<el-form-item label="处理状态">
<el-select v-model="query.state" clearable placeholder="全部状态">
<el-option label="待确认" value="unacknowledged" /><el-option label="已确认" value="acknowledged" />
<el-option label="恢复观察" value="recovering" /><el-option label="已恢复" value="recovered" />
</el-select>
</el-form-item>
<el-form-item label="关键词"><el-input v-model="query.keyword" clearable placeholder="设备、位置或详情" @keyup.enter="search" /></el-form-item>
<el-form-item class="filter-actions"><el-button type="primary" :icon="Search" native-type="submit">查询</el-button><el-button :icon="RefreshLeft" @click="reset">重置</el-button></el-form-item>
</el-form>
<el-table v-loading="loading" :data="alerts" border stripe empty-text="当前筛选条件下没有运维告警">
<el-table-column label="告警" min-width="180"><template #default="scope"><strong>{{ scope.row.title }}</strong><div class="muted">{{ alertTypeLabel(scope.row.alertType) }}</div></template></el-table-column>
<el-table-column label="对象" min-width="170"><template #default="scope"><strong>{{ scope.row.objectName }}</strong><div class="muted">{{ scope.row.location || scope.row.objectId }}</div></template></el-table-column>
<el-table-column label="状态" width="110" align="center"><template #default="scope"><el-tag :type="alertStateType(scope.row.state)" size="small">{{ alertStateLabel(scope.row.state) }}</el-tag></template></el-table-column>
<el-table-column label="首次 / 最近发现" min-width="190"><template #default="scope"><div>{{ formatTime(scope.row.firstSeenAt) }}</div><small>{{ formatTime(scope.row.lastSeenAt) }}</small></template></el-table-column>
<el-table-column prop="occurrenceCount" label="重复次数" width="92" align="center" />
<el-table-column label="操作" width="184" fixed="right">
<template #default="scope">
<el-button v-permisaction="['sense:ops-alert:detail']" type="primary" link @click="openDetail(scope.row.id)">详情</el-button>
<el-button v-if="canAcknowledge(scope.row)" v-permisaction="['sense:ops-alert:acknowledge']" type="warning" link @click="openAction(scope.row, 'acknowledge')">确认</el-button>
<el-button v-if="canRecover(scope.row)" v-permisaction="['sense:ops-alert:recover']" type="success" link @click="openAction(scope.row, 'recover')">恢复</el-button>
</template>
</el-table-column>
</el-table>
<pagination v-show="total > 0" v-model:current-page="query.pageIndex" v-model:page-size="query.pageSize" :total="total" @pagination="load" />
<p class="help-text">同一对象的同类故障使用唯一指纹去重;恢复后再次发生会复用历史记录并开启新的处理周期。</p>
</el-card>
<el-dialog v-model="detailOpen" title="运维告警详情" width="min(780px, calc(100vw - 24px))">
<el-descriptions v-if="selected" :column="2" border>
<el-descriptions-item label="告警编号">{{ selected.id }}</el-descriptions-item><el-descriptions-item label="处理状态"><el-tag :type="alertStateType(selected.state)">{{ alertStateLabel(selected.state) }}</el-tag></el-descriptions-item>
<el-descriptions-item label="异常对象">{{ selected.objectName }}</el-descriptions-item><el-descriptions-item label="发现次数">{{ selected.occurrenceCount }}</el-descriptions-item>
<el-descriptions-item label="异常详情" :span="2">{{ selected.detail }}</el-descriptions-item><el-descriptions-item label="建议处理" :span="2">{{ selected.nextAction }}</el-descriptions-item>
</el-descriptions>
<el-alert class="detail-boundary" type="warning" :closable="false" show-icon title="业务隔离边界">该记录仅用于 Sense 运维,不会生成安全事件或 Bell 业务预警。</el-alert>
<el-timeline v-if="transitions.length" class="timeline"><el-timeline-item v-for="item in transitions" :key="item.id" :timestamp="formatTime(item.createdAt)"><strong>{{ actionLabel(item.action) }}</strong><div class="muted">{{ item.reason }}</div></el-timeline-item></el-timeline>
<template #footer><el-button @click="detailOpen = false">关闭</el-button></template>
</el-dialog>
<el-dialog v-model="actionOpen" :title="actionMode === 'acknowledge' ? '确认运维告警' : '确认故障恢复'" width="min(520px, calc(100vw - 24px))" :close-on-click-modal="false">
<el-alert :type="actionMode === 'acknowledge' ? 'warning' : 'success'" :closable="false" show-icon :title="actionMode === 'acknowledge' ? '确认不代表恢复' : '仅在健康状态已稳定后关闭告警'" />
<el-form ref="actionFormRef" :model="actionForm" :rules="actionRules" label-position="top" class="action-form">
<el-form-item label="处理说明" prop="reason"><el-input v-model="actionForm.reason" type="textarea" :rows="4" maxlength="256" show-word-limit placeholder="请输入至少 6 个字符,说明处理人、措施或恢复依据" /></el-form-item>
</el-form>
<template #footer><el-button @click="actionOpen = false">取消</el-button><el-button :type="actionMode === 'acknowledge' ? 'warning' : 'success'" :loading="submitting" @click="submitAction">确认提交</el-button></template>
</el-dialog>
</template>
</BasicLayout>
</template>
<script setup>
import { onMounted, reactive, ref } from 'vue'
import { Refresh, RefreshLeft, Search } from '@element-plus/icons-vue'
import { ElMessage } from 'element-plus'
import { acknowledgeOpsAlert, evaluateOpsAlerts, getOpsAlert, listOpsAlerts, recoverOpsAlert } from '@/api/sense/ops-alert'
import { alertStateLabel, alertStateType, alertTypeLabel, buildOpsAlertQuery, canAcknowledge, canRecover } from './opsAlertState'
defineOptions({ name: 'SenseOpsAlert' })
const loading = ref(false); const refreshing = ref(false); const submitting = ref(false); const detailOpen = ref(false); const actionOpen = ref(false)
const alerts = ref([]); const total = ref(0); const selected = ref(null); const transitions = ref([]); const actionMode = ref('acknowledge'); const actionFormRef = ref()
const summary = reactive({ active: 0, unacknowledged: 0, recovering: 0 })
const query = reactive({ pageIndex: 1, pageSize: 10, alertType: '', state: '', keyword: '' })
const actionForm = reactive({ reason: '', expectedVersion: 0 })
const actionRules = { reason: [{ required: true, message: '请输入处理说明', trigger: 'blur' }, { min: 6, max: 256, message: '处理说明需为 6 至 256 个字符', trigger: 'blur' }] }
const typeOptions = ['device_offline', 'authentication_failed', 'clock_drift', 'reconciliation_failed', 'media_shard_failed', 'control_tunnel_failed'].map(value => ({ value, label: alertTypeLabel(value) }))
function unwrap(response) { return response?.data?.data ?? response?.data ?? response }
function formatTime(value) { return value ? new Date(value).toLocaleString('zh-CN', { hour12: false }) : '—' }
function actionLabel(value) { return ({ detected: '发现异常', acknowledge: '人工确认', health_restored: '健康恢复', recover: '确认恢复', relapsed: '恢复失败', reopened: '再次发生' })[value] || value }
async function load() { loading.value = true; try { const payload = unwrap(await listOpsAlerts(buildOpsAlertQuery(query))) || {}; alerts.value = payload.list || []; total.value = payload.count || 0; Object.assign(summary, payload.summary || { active: 0, unacknowledged: 0, recovering: 0 }) } catch (error) { ElMessage.error(error.message || '运维告警加载失败') } finally { loading.value = false } }
async function refreshSources() { refreshing.value = true; try { await evaluateOpsAlerts(); ElMessage.success('状态已刷新'); await load() } catch (error) { ElMessage.error(error.message || '健康状态刷新失败') } finally { refreshing.value = false } }
function search() { query.pageIndex = 1; load() }
function reset() { Object.assign(query, { pageIndex: 1, alertType: '', state: '', keyword: '' }); load() }
async function openDetail(id) { try { const payload = unwrap(await getOpsAlert(id)) || {}; selected.value = payload.alert; transitions.value = payload.transitions || []; detailOpen.value = true } catch (error) { ElMessage.error(error.message || '告警详情加载失败') } }
function openAction(item, mode) { selected.value = item; actionMode.value = mode; actionForm.reason = ''; actionForm.expectedVersion = item.version; actionOpen.value = true }
async function submitAction() { if (!await actionFormRef.value.validate().catch(() => false)) return; submitting.value = true; try { const payload = { expectedVersion: actionForm.expectedVersion, reason: actionForm.reason.trim() }; if (actionMode.value === 'acknowledge') await acknowledgeOpsAlert(selected.value.id, payload); else await recoverOpsAlert(selected.value.id, payload); ElMessage.success(actionMode.value === 'acknowledge' ? '告警已确认' : '告警已恢复'); actionOpen.value = false; detailOpen.value = false; await load() } catch (error) { ElMessage.warning(error.message || '状态已变化,请刷新后重试') } finally { submitting.value = false } }
onMounted(load)
</script>
<style scoped>
.page-header{display:flex;align-items:flex-start;justify-content:space-between;gap:16px}.page-header h3{margin:0 0 6px}.page-header p{margin:0;color:#909399}.boundary-alert{margin:16px 0}.summary-row{margin-bottom:20px}.summary-item{display:flex;align-items:center;justify-content:space-between;min-height:72px;padding:12px 16px;border:1px solid #ebeef5;border-radius:4px}.summary-item span{color:#606266}.summary-item strong{font-size:22px}.danger-number{color:#f56c6c}.primary-number{color:#409eff}.filter-form{display:flex;align-items:flex-end;flex-wrap:wrap;gap:0 12px;margin:16px 0 2px}.filter-form .el-form-item{width:230px}.filter-form .filter-actions{width:auto}.filter-form :deep(.el-select){width:100%}.muted,small,.help-text{color:#909399;font-size:12px}.help-text{margin:12px 0 0}.detail-boundary{margin:16px 0}.timeline{padding-top:8px}.action-form{margin-top:16px}@media(max-width:768px){.page-header{align-items:stretch;flex-direction:column}.filter-form .el-form-item{width:100%}.summary-item{margin-bottom:8px}}
</style>
@@ -0,0 +1,30 @@
const typeLabels = {
device_offline: '设备离线',
authentication_failed: '认证失败',
clock_drift: '时间漂移',
reconciliation_failed: '状态对账失败',
media_shard_failed: '媒体分片异常',
control_tunnel_failed: '控制隧道异常'
}
const stateLabels = {
unacknowledged: '待确认',
acknowledged: '已确认',
recovering: '恢复观察',
recovered: '已恢复'
}
export function alertTypeLabel(value) { return typeLabels[value] || value || '未知类型' }
export function alertStateLabel(value) { return stateLabels[value] || value || '未知状态' }
export function alertStateType(value) { return ({ unacknowledged: 'danger', acknowledged: 'warning', recovering: 'primary', recovered: 'success' })[value] || 'info' }
export function canAcknowledge(item) { return item?.state === 'unacknowledged' && item?.version > 0 }
export function canRecover(item) { return item?.state === 'recovering' && item?.version > 0 }
export function buildOpsAlertQuery(query) {
return {
pageIndex: Number(query.pageIndex) || 1,
pageSize: Number(query.pageSize) || 10,
alertType: String(query.alertType || '').trim(),
state: String(query.state || '').trim(),
keyword: String(query.keyword || '').trim()
}
}
@@ -0,0 +1,18 @@
import request from '@/utils/request'
import { acknowledgeOpsAlert, evaluateOpsAlerts, getOpsAlert, listOpsAlerts, recoverOpsAlert } from '@/api/sense/ops-alert'
jest.mock('@/utils/request', () => jest.fn(config => config))
describe('Sense operational alert API', () => {
beforeEach(() => request.mockClear())
test('exposes read, refresh and optimistic state actions', () => {
const query = { pageIndex: 1, pageSize: 10, state: 'unacknowledged' }
const action = { expectedVersion: 3, reason: '已安排现场人员处理设备故障' }
expect(listOpsAlerts(query)).toEqual({ url: '/api/v1/ops-alerts', method: 'get', params: query })
expect(getOpsAlert('alert/a b')).toEqual({ url: '/api/v1/ops-alerts/alert%2Fa%20b', method: 'get' })
expect(evaluateOpsAlerts()).toEqual({ url: '/api/v1/ops-alerts/evaluate', method: 'post' })
expect(acknowledgeOpsAlert('alert/a b', action)).toEqual({ url: '/api/v1/ops-alerts/alert%2Fa%20b/acknowledge', method: 'post', data: action })
expect(recoverOpsAlert('alert/a b', action)).toEqual({ url: '/api/v1/ops-alerts/alert%2Fa%20b/recover', method: 'post', data: action })
})
})
@@ -0,0 +1,25 @@
import { alertStateLabel, alertStateType, alertTypeLabel, buildOpsAlertQuery, canAcknowledge, canRecover } from '@/views/sense/ops-alert/opsAlertState'
describe('Sense operational alert presentation state', () => {
test('uses ordinary operations wording and six alert types', () => {
expect(alertTypeLabel('device_offline')).toBe('设备离线')
expect(alertTypeLabel('authentication_failed')).toBe('认证失败')
expect(alertTypeLabel('clock_drift')).toBe('时间漂移')
expect(alertTypeLabel('reconciliation_failed')).toBe('状态对账失败')
expect(alertTypeLabel('media_shard_failed')).toBe('媒体分片异常')
expect(alertTypeLabel('control_tunnel_failed')).toBe('控制隧道异常')
expect(alertStateLabel('recovering')).toBe('恢复观察')
expect(alertStateType('recovered')).toBe('success')
})
test('enforces the approved state action boundary', () => {
expect(canAcknowledge({ state: 'unacknowledged', version: 1 })).toBe(true)
expect(canAcknowledge({ state: 'acknowledged', version: 2 })).toBe(false)
expect(canRecover({ state: 'recovering', version: 3 })).toBe(true)
expect(canRecover({ state: 'acknowledged', version: 3 })).toBe(false)
})
test('builds an allowlisted query', () => {
expect(buildOpsAlertQuery({ pageIndex: '2', pageSize: 20, alertType: ' clock_drift ', state: ' recovering ', keyword: ' 南门 ', ignored: true })).toEqual({ pageIndex: 2, pageSize: 20, alertType: 'clock_drift', state: 'recovering', keyword: '南门' })
})
})