"""Сравнение версий агентов (ssh-monitor / rdp-login-monitor, формат X.Y.Z-SAC).""" from __future__ import annotations import re from dataclasses import dataclass from sqlalchemy import select from sqlalchemy.orm import Session from app.models import Host _VERSION_RE = re.compile(r"^(\d+)\.(\d+)\.(\d+)(?:-SAC)?$", re.IGNORECASE) _DEFAULT_LAG_THRESHOLD = 3 @dataclass(frozen=True, order=True) class ParsedAgentVersion: major: int minor: int patch: int def parse_agent_version(raw: str | None) -> ParsedAgentVersion | None: if not raw: return None text = raw.strip() match = _VERSION_RE.match(text) if not match: return None return ParsedAgentVersion( major=int(match.group(1)), minor=int(match.group(2)), patch=int(match.group(3)), ) def agent_version_lag(host: ParsedAgentVersion, latest: ParsedAgentVersion) -> int | None: """Отставание от latest: patch при том же major.minor, иначе None (любое отставание).""" if host > latest: return 0 if host.major != latest.major or host.minor != latest.minor: return None return latest.patch - host.patch def is_agent_version_outdated( host_version: str | None, latest_version: str | None, *, lag_threshold: int = _DEFAULT_LAG_THRESHOLD, ) -> bool: host = parse_agent_version(host_version) latest = parse_agent_version(latest_version) if host is None or latest is None or host >= latest: return False lag = agent_version_lag(host, latest) if lag is None: return True return lag >= lag_threshold def latest_agent_versions_by_product(db: Session) -> dict[str, str]: rows = db.execute( select(Host.product, Host.product_version).where( Host.product_version.isnot(None), Host.product_version != "", ) ).all() best: dict[str, ParsedAgentVersion] = {} best_raw: dict[str, str] = {} for product, version in rows: parsed = parse_agent_version(version) if parsed is None: continue current = best.get(product) if current is None or parsed > current: best[product] = parsed best_raw[product] = version.strip() return best_raw