разбил ssh discovery

.claude/settings.local.json

@@ -16,7 +16,8 @@
       "Bash(docker compose:*)",
       "Bash(grep -r 8000 /c/Users/Professional/Documents/VSCode/utp_service/backend/ --include=*.py --include=*.sh)",
       "Bash(grep -n \"rack_type\" /c/Users/Professional/Documents/VSCode/utp_service/backend/alembic/versions/*.py)",
-      "Bash(python3 -c \":*)"
+      "Bash(python3 -c \":*)",
+      "Bash(python -c ':*)"
     ]
   }
 }

backend/app/services/database.py

@@ -6,13 +6,26 @@ from app.config import settings
 
 engine = create_async_engine(
     settings.DATABASE_URL,
-    echo=settings.DEBUG,
+    #echo=settings.DEBUG,
+    echo=False
     pool_pre_ping=True,
 )
 
+
+#if not settings.SQLALCHEMY_LOG:
+#    for _log_name in ("sqlalchemy.engine", "sqlalchemy.pool", "sqlalchemy.dialects", "sqlalchemy.orm"):
+#        logging.getLogger(_log_name).setLevel(logging.WARNING)
+
 if not settings.SQLALCHEMY_LOG:
-    for _log_name in ("sqlalchemy.engine", "sqlalchemy.pool", "sqlalchemy.dialects", "sqlalchemy.orm"):
+    for name in (
-        logging.getLogger(_log_name).setLevel(logging.WARNING)
+        "sqlalchemy.engine",
+        "sqlalchemy.pool",
+        "sqlalchemy.dialects",
+        "sqlalchemy.orm",
+    ):
+        logger = logging.getLogger(name)
+        logger.setLevel(logging.ERROR)
+        logger.propagate = False
 
 AsyncSessionLocal = async_sessionmaker(
     bind=engine,

backend/app/services/discovery/__init__.py

@@ -0,0 +1,20 @@
+"""
+SSH-based device discovery package.
+
+Public API:
+    discover_via_ssh   — run a full discovery pass for one Object
+    DiscoveryResult    — summary of what was created/updated/skipped
+    DeviceAction       — per-device action record
+    DiscoveredDevice   — device parsed from a CAPS config plugin section
+"""
+
+from .orchestrator import discover_via_ssh
+from .types import DeviceAction, DiscoveredDevice, DiscoveryResult, ProgressCallback
+
+__all__ = [
+    "discover_via_ssh",
+    "DiscoveryResult",
+    "DeviceAction",
+    "DiscoveredDevice",
+    "ProgressCallback",
+]

backend/app/services/discovery/config_parser.py

@@ -0,0 +1,178 @@
+"""
+CAPS INI config parsing utilities.
+All functions are pure (no network I/O, no DB).
+"""
+
+import logging
+import re
+from configparser import RawConfigParser
+from urllib.parse import urlparse
+
+from .types import DiscoveredDevice
+from .uri import (
+    _SKIP_PLUGIN_PREFIXES,
+    extract_device_info,
+    extract_ip,
+    cls_to_category,
+    cls_to_role,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def parse_config(
+    config_text: str,
+    rack_host: str,
+    all_rack_hosts: list[str] | None = None,
+) -> list[DiscoveredDevice]:
+    """Extract discovered network devices from a CAPS config file.
+
+    Args:
+        config_text: Config file content.
+        rack_host: IP of the current rack PC (excluded from results).
+        all_rack_hosts: All rack IPs (excluded so rack servers aren't mistaken for devices).
+    """
+    if all_rack_hosts is None:
+        all_rack_hosts = [rack_host]
+
+    parser = RawConfigParser(strict=False)
+    parser.read_string(config_text)
+
+    discovered: list[DiscoveredDevice] = []
+    seen_ips: set[str] = set()
+    all_sections = parser.sections()
+    plugin_sections = [s for s in all_sections if s.startswith("plugin:")]
+
+    logger.info(
+        "[discovery] Config parsing: %d sections total, %d plugins",
+        len(all_sections), len(plugin_sections),
+    )
+
+    for section in all_sections:
+        if not section.startswith("plugin:"):
+            continue
+
+        plugin_name = section[len("plugin:"):]
+        if any(plugin_name.startswith(prefix) for prefix in _SKIP_PLUGIN_PREFIXES):
+            logger.info("[discovery] Plugin '%s' - IGNORED (skip list)", plugin_name)
+            continue
+
+        cls = parser.get(section, "cls", fallback="")
+        if not cls:
+            logger.info("[discovery] Plugin '%s' - IGNORED (no 'cls' field)", plugin_name)
+            continue
+
+        logger.info("[discovery] Plugin '%s' (cls=%s)", plugin_name, cls)
+
+        # uri / frame_uri / host → actual device (camera, io_board, etc.)
+        device_uri_fields = ("uri", "frame_uri", "host")
+        found_ip_in_plugin = False
+        for field_name in device_uri_fields:
+            raw = parser.get(section, field_name, fallback=None)
+            if not raw:
+                logger.info("[discovery]   %s: (not set)", field_name)
+                continue
+            logger.info("[discovery]   %s='%s'", field_name, raw)
+            info = extract_device_info(raw)
+            if not info:
+                logger.info("[discovery]     → IGNORED (not a valid network IP)")
+                continue
+            ip, http_user, http_pass = info
+            if ip in all_rack_hosts:
+                logger.info("[discovery]     → IGNORED (is a rack server IP: %s)", ip)
+                continue
+            if ip in seen_ips:
+                logger.info("[discovery]     → IGNORED (duplicate)")
+                continue
+
+            seen_ips.add(ip)
+            cred_str = f", http_user={http_user}" if http_user else ""
+            logger.info(
+                "[discovery]     → EXTRACTED %s (category=%s, role=%s%s)",
+                ip, cls_to_category(cls), cls_to_role(cls, plugin_name), cred_str,
+            )
+            discovered.append(DiscoveredDevice(
+                ip=ip,
+                category=cls_to_category(cls),
+                role=cls_to_role(cls, plugin_name),
+                plugin_name=plugin_name,
+                cls=cls,
+                http_username=http_user,
+                http_password=http_pass,
+            ))
+            found_ip_in_plugin = True
+
+        # stream_uri → RTSP server (a separate VM, not the camera itself)
+        stream_raw = parser.get(section, "stream_uri", fallback=None)
+        if stream_raw:
+            logger.info("[discovery]   stream_uri='%s'", stream_raw)
+            stream_ip = extract_ip(stream_raw)
+            if stream_ip and stream_ip not in all_rack_hosts and stream_ip not in seen_ips:
+                seen_ips.add(stream_ip)
+                logger.info("[discovery]     → EXTRACTED %s as RTSP server (category=vm)", stream_ip)
+                discovered.append(DiscoveredDevice(
+                    ip=stream_ip,
+                    category="vm",
+                    role="other",
+                    plugin_name=f"{plugin_name}_rtsp",
+                    cls="rtsp_server",
+                ))
+                found_ip_in_plugin = True
+
+        if not found_ip_in_plugin:
+            logger.info("[discovery]   → No network devices found in this plugin")
+
+    # [caps.voice] section → Asterisk server
+    voice_addr = parser.get("caps.voice", "server_addr", fallback=None)
+    if voice_addr:
+        logger.info("[discovery] [caps.voice] server_addr='%s'", voice_addr)
+        voice_ip = extract_ip(f"sip://{voice_addr}") or (
+            voice_addr.strip()
+            if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", voice_addr.strip())
+            else None
+        )
+        if voice_ip and voice_ip not in all_rack_hosts and voice_ip not in seen_ips:
+            seen_ips.add(voice_ip)
+            logger.info("[discovery]     → EXTRACTED %s as Asterisk (category=vm)", voice_ip)
+            discovered.append(DiscoveredDevice(
+                ip=voice_ip,
+                category="vm",
+                role="other",
+                plugin_name="asterisk",
+                cls="asterisk",
+            ))
+
+    return discovered
+
+
+def parse_slave_check(config_text: str, master_ip: str, host: str) -> tuple[bool, str, str]:
+    """Check a parsed config against one master IP. No network I/O.
+
+    Returns:
+        (is_slave, identifier, config_text) — config_text is passed through unchanged.
+    """
+    parser = RawConfigParser(strict=False)
+    parser.read_string(config_text)
+    linking_cls = parser.get("plugin:linking", "cls", fallback="")
+    linking_uri = parser.get("plugin:linking", "uri", fallback="")
+    identifier = parser.get("caps.rack", "identifier", fallback="").strip()
+    if "secondary" in linking_cls.lower() and master_ip in linking_uri:
+        return True, identifier or host, config_text
+    return False, "", ""
+
+
+def parse_db_uri(uri: str) -> dict | None:
+    """Parse postgresql+psycopg2://user:pass@host:port/dbname → dict."""
+    try:
+        parsed = urlparse(uri)
+        if not parsed.hostname:
+            return None
+        return {
+            "host": parsed.hostname,
+            "port": parsed.port or 5432,
+            "user": parsed.username or "",
+            "password": parsed.password or "",
+            "dbname": (parsed.path or "").lstrip("/"),
+        }
+    except Exception:
+        return None

backend/app/services/discovery/db_resolver.py

@@ -0,0 +1,103 @@
+"""
+DB-related helpers for device discovery:
+  - Auto-detect the CAPS DB connection from a remote config file.
+  - Fetch rack host IPs from the CAPS PostgreSQL database.
+"""
+
+import logging
+
+import psycopg
+
+from .config_parser import parse_db_uri
+from .ssh_client import ssh_run_multi, open_ssh_tunnel
+
+logger = logging.getLogger(__name__)
+
+
+async def autodetect_db_from_server(
+    server_ip: str,
+    ssh_port: int,
+    ssh_options: list,
+    server_config_path: str = "/etc/caps/config.ini",
+) -> dict | None:
+    """SSH into server, read its caps config, extract DB connection params.
+
+    Returns a dict with keys: host, port, user, password, dbname.
+    Returns None if no db_uri is found.
+    """
+    from configparser import RawConfigParser
+
+    try:
+        config_text = await ssh_run_multi(
+            server_ip, ssh_port, ssh_options, f"cat {server_config_path}"
+        )
+    except Exception:
+        return None
+
+    parser = RawConfigParser(strict=False)
+    parser.read_string(config_text)
+
+    for section in list(parser.sections()) + ["DEFAULT"]:
+        try:
+            db_uri = parser.get(section, "db_uri", fallback=None)
+            if db_uri and db_uri.startswith("postgresql"):
+                return parse_db_uri(db_uri)
+        except Exception:
+            continue
+
+    db_uri = parser.defaults().get("db_uri")
+    if db_uri and db_uri.startswith("postgresql"):
+        return parse_db_uri(db_uri)
+
+    return None
+
+
+async def get_rack_hosts(
+    host: str,
+    port: int,
+    user: str,
+    password: str,
+    dbname: str,
+    table: str,
+    via_tunnel: bool = False,
+    tunnel_host: str | None = None,
+    tunnel_ssh_port: int = 22,
+    ssh_options: list | None = None,
+) -> list[tuple[str, str, str, str | None]]:
+    """Return list of (external_id, host_ip, rack_name, rack_type) from the racks table.
+
+    Supports an optional SSH tunnel when the DB is not directly reachable.
+    """
+    if via_tunnel and tunnel_host and ssh_options:
+        conn = await open_ssh_tunnel(tunnel_host, tunnel_ssh_port, ssh_options, host, port)
+        async with conn:
+            listener = await conn.forward_local_port("127.0.0.1", 0, host, port)
+            local_port = listener.get_port()
+            dsn = (
+                f"host=127.0.0.1 port={local_port} dbname={dbname} "
+                f"user={user} password={password} connect_timeout=10"
+            )
+            return await _fetch_rack_rows(dsn, table)
+    else:
+        dsn = f"host={host} port={port} dbname={dbname} user={user} password={password} connect_timeout=10"
+        return await _fetch_rack_rows(dsn, table)
+
+
+async def _fetch_rack_rows(dsn: str, table: str) -> list[tuple[str, str, str, str | None]]:
+    async with await psycopg.AsyncConnection.connect(dsn) as conn:
+        cursor = await conn.execute(
+            f"SELECT id, data->>'host' AS host, name, data->>'rack_type' AS rack_type "
+            f"FROM \"{table}\" WHERE data->>'host' IS NOT NULL"
+        )
+        rows = await cursor.fetchall()
+
+    result = []
+    for r in rows:
+        if not r[1]:
+            continue
+        ext_id = str(r[0])
+        short_name = str(r[2]) if r[2] else None
+        # Combine CAPS id with short name: "lane-1-1 (01)"
+        rack_name = f"{ext_id} ({short_name})" if short_name and short_name != ext_id else ext_id
+        result.append((ext_id, str(r[1]), rack_name, r[3] or None))
+    return result

backend/app/services/discovery/orchestrator.py

@@ -0,0 +1,927 @@
+"""
+Main discovery orchestrator.
+
+Coordinates all phases of SSH-based device discovery:
+  0.  Resolve DB connection (explicit config or autodetect via SSH).
+  0.5 Pre-upsert the DB server device (so users can configure tunnels on failure).
+  1.  Fetch rack hosts from the CAPS DB; sync Rack table entries.
+  1b. Upsert infrastructure devices (main_server).
+  2.  SSH into each rack → parse CAPS config → upsert plugin devices.
+  2.5 Discover slave racks via ARP neighbor tables.
+  3.  Discover MikroTik via SSH_CLIENT env var.
+  4.  Discover Proxmox via port 8006 scan.
+"""
+
+import logging
+from configparser import RawConfigParser
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.device import Device
+from app.models.object import Object
+from app.models.rack import Rack
+from app.services.crypto import decrypt, encrypt
+
+from .config_parser import parse_config, parse_slave_check
+from .db_resolver import autodetect_db_from_server, get_rack_hosts
+from .ssh_client import (
+    discover_mikrotik_multi,
+    discover_proxmox_via_port_scan,
+    fetch_slave_config,
+    get_ip_neighbors_multi,
+    ssh_run_multi,
+)
+from .types import DeviceAction, DiscoveredDevice, DiscoveryResult, ProgressCallback
+
+logger = logging.getLogger(__name__)
+
+
+# ── Public entry point ────────────────────────────────────────────────────────
+
+async def discover_via_ssh(
+    db: AsyncSession,
+    obj: Object,
+    config_path: str = "/etc/caps/config.ini",
+    progress_cb: ProgressCallback | None = None,
+) -> DiscoveryResult:
+    result = DiscoveryResult()
+
+    async def _emit(event_type: str, data: dict) -> None:
+        if progress_cb is not None:
+            await progress_cb(event_type, data)
+
+    from app.services.ssh import build_object_auth_options
+    ssh_options = await build_object_auth_options(obj, db)
+
+    if not ssh_options:
+        result.errors.append("SSH credentials not configured on this object")
+        return result
+
+    ssh_port = obj.ssh_port or 22
+    # First option's password is used by single-auth Proxmox scan (legacy).
+    _first = ssh_options[0]
+    ssh_password = _first.password if _first.type == "password" else ""
+
+    # ── Phase 0: Resolve DB connection ────────────────────────────────────────
+    db_params = await _resolve_db_connection(obj, ssh_port, ssh_options, result, _emit)
+    if db_params is None:
+        return result
+
+    db_host, db_port, db_name, db_user, db_password, db_table, server_ip_ssh_verified = db_params
+
+    # ── Phase 0.5: Pre-upsert DB server device ────────────────────────────────
+    now = datetime.now(timezone.utc)
+    db_device_obj, use_tunnel = await _pre_upsert_db_device(
+        db, obj, db_host, result, now
+    )
+
+    # ── Phase 1: Fetch rack hosts ──────────────────────────────────────────────
+    await _emit("disc_action", {"message": "Получение списка стоек из БД..."})
+    logger.info(
+        "[discovery] Connecting to object DB: host=%s port=%s dbname=%s user=%s table=%s via_tunnel=%s",
+        db_host, db_port, db_name, db_user, db_table, use_tunnel,
+    )
+    try:
+        rack_hosts = await get_rack_hosts(
+            host=db_host, port=db_port, user=db_user,
+            password=db_password or "", dbname=db_name, table=db_table,
+            via_tunnel=use_tunnel,
+            tunnel_host=obj.server_ip, tunnel_ssh_port=ssh_port,
+            ssh_options=ssh_options,
+        )
+        if db_device_obj is not None:
+            db_device_obj.status = "online"
+            db_device_obj.last_seen = now
+    except Exception as exc:
+        if db_device_obj is not None:
+            db_device_obj.status = "offline"
+        err_msg = f"Cannot fetch rack hosts from object DB: {exc}"
+        if "timeout" in str(exc).lower():
+            err_msg += ". Попробуйте настроить SSH-туннель (отредактируйте устройство БД и включите опцию)"
+        result.errors.append(err_msg)
+        return result
+
+    if not rack_hosts:
+        result.errors.append("No racks with host IPs found in object DB")
+        return result
+
+    await _emit("disc_server", {"rack_count": len(rack_hosts)})
+
+    # ── Phase 1a: Sync Rack table entries ──────────────────────────────────────
+    rack_id_mapping = await _sync_rack_entries(db, obj, rack_hosts)
+
+    # ── Phase 1b: Upsert infrastructure devices ───────────────────────────────
+    infra_devices = []
+    if obj.server_ip:
+        infra_devices.append((obj.server_ip, "main_server", "caps_server", server_ip_ssh_verified))
+    await _upsert_infra_devices(db, obj, infra_devices, result, now)
+
+    # ── Phase 2: Process racks ────────────────────────────────────────────────
+    neighbor_candidates: set[str] = set()
+    discovered_device_ips: set[str] = set()
+
+    for rack_idx, (external_id, rack_host, rack_name, _rack_type_db) in enumerate(rack_hosts):
+        await _emit("disc_rack_start", {
+            "rack_name": rack_name, "rack_host": rack_host,
+            "index": rack_idx, "total": len(rack_hosts),
+        })
+
+        # Allow per-device config_path override stored in connection_params.
+        effective_config_path = await _resolve_rack_config_path(
+            db, obj, rack_host, config_path
+        )
+
+        rack_ssh_ok, config_text = await _ssh_fetch_rack_config(
+            rack_host, ssh_port, ssh_options, effective_config_path, rack_name, result, _emit
+        )
+
+        await _upsert_rack_device(
+            db, obj, external_id, rack_host, rack_name,
+            rack_id_mapping, rack_ssh_ok, now, result
+        )
+
+        if not rack_ssh_ok:
+            try:
+                await db.flush()
+            except Exception as exc:
+                result.errors.append(f"Flush error for offline rack {rack_host}: {exc}")
+            await _emit("disc_rack_done", {
+                "rack_name": rack_name, "rack_host": rack_host, "found": 0, "ssh_ok": False
+            })
+            continue
+
+        neigh_ips = await get_ip_neighbors_multi(rack_host, ssh_port, ssh_options)
+        neighbor_candidates.update(neigh_ips)
+        discovered_device_ips.add(rack_host)
+
+        all_rack_ips = [h for _, h, _, _ in rack_hosts]
+        await _emit("disc_action", {"message": f"Парсинг конфигурации {rack_name}..."})
+        devices_found = parse_config(config_text, rack_host, all_rack_ips)
+        logger.info("[discovery] Found %d devices from plugins in %s", len(devices_found), rack_host)
+
+        for found in devices_found:
+            await _upsert_plugin_device(
+                db, obj, found, external_id, rack_id_mapping, result
+            )
+            discovered_device_ips.add(found.ip)
+
+        try:
+            await db.flush()
+        except Exception as exc:
+            result.errors.append(f"Flush error after rack {rack_host}: {exc}")
+        await _emit("disc_rack_done", {
+            "rack_name": rack_name, "rack_host": rack_host,
+            "found": len(devices_found), "ssh_ok": True,
+        })
+
+    # ── Phase 2.5: Slave rack discovery via ARP ───────────────────────────────
+    all_rack_ips_set = {h for _, h, _, _ in rack_hosts}
+    infra_ips = {ip for ip in [obj.server_ip, db_host] if ip}
+    common_gateways = {ip for ip in neighbor_candidates if ip.endswith(".1") or ip.endswith(".254")}
+    slave_candidates = (
+        neighbor_candidates - all_rack_ips_set - discovered_device_ips - infra_ips - common_gateways
+    )
+
+    await _emit("disc_phase", {"phase": "slaves", "candidate_count": len(slave_candidates)})
+    if slave_candidates:
+        logger.info("[discovery] Phase 2.5: Slave discovery — %d candidate(s)", len(slave_candidates))
+        await _emit("disc_action", {
+            "message": f"Поиск слейв-стоек: проверяем {len(slave_candidates)} ARP-кандидатов..."
+        })
+    else:
+        await _emit("disc_action", {"message": "Слейв-стойки: ARP-кандидатов не найдено"})
+
+    await _discover_slave_racks(
+        db, obj, slave_candidates, rack_hosts, ssh_port, ssh_options,
+        config_path, all_rack_ips_set, now, result, _emit
+    )
+
+    # ── Phase 3: MikroTik via SSH_CLIENT ──────────────────────────────────────
+    await _emit("disc_phase", {"phase": "mikrotik"})
+    await _emit("disc_action", {"message": "Поиск MikroTik (SSH_CLIENT)..."})
+    await _discover_mikrotik(db, obj, rack_hosts, ssh_port, ssh_options, result, _emit)
+
+    # ── Phase 4: Proxmox via port 8006 scan ───────────────────────────────────
+    await _emit("disc_phase", {"phase": "proxmox"})
+    await _emit("disc_action", {"message": "Сканирование порта 8006 (Proxmox)..."})
+    await _discover_proxmox(db, obj, ssh_port, ssh_password, rack_hosts, result, _emit)
+
+    # ── Summary ───────────────────────────────────────────────────────────────
+    _log_summary(result)
+    return result
+
+
+# ── Phase helpers ─────────────────────────────────────────────────────────────
+
+async def _resolve_db_connection(
+    obj: Object,
+    ssh_port: int,
+    ssh_options: list,
+    result: DiscoveryResult,
+    _emit,
+) -> tuple | None:
+    """Determine the DB connection parameters.
+
+    Returns (db_host, db_port, db_name, db_user, db_password, db_table, server_ip_ssh_verified)
+    or None if resolution failed (errors added to result).
+    """
+    db_host = obj.db_host
+    db_port = obj.db_port or 5432
+    db_name = obj.db_name
+    db_user = obj.db_user
+    db_password: str | None = None
+    db_table = obj.db_table
+    server_ip_ssh_verified = False
+
+    if db_host and db_name and db_user and obj.db_pass_enc and db_table:
+        try:
+            db_password = decrypt(obj.db_pass_enc)
+        except Exception:
+            result.errors.append("Failed to decrypt DB password")
+            return None
+    elif obj.server_ip:
+        await _emit("disc_action", {"message": f"Чтение конфигурации сервера {obj.server_ip}..."})
+        detected = await autodetect_db_from_server(
+            server_ip=obj.server_ip, ssh_port=ssh_port, ssh_options=ssh_options,
+        )
+        if not detected or not all([detected.get("host"), detected.get("user"), detected.get("dbname")]):
+            await _emit_db_autodetect_error(obj, ssh_port, ssh_options, result)
+            return None
+
+        db_host = detected["host"]
+        db_port = detected["port"]
+        db_user = detected["user"]
+        db_password = detected["password"]
+        db_name = detected["dbname"]
+        db_table = db_table or obj.db_table or "racks"
+        server_ip_ssh_verified = True
+    else:
+        result.errors.append(
+            "DB connection not configured. Either fill in the DB fields on the object "
+            "or set server_ip so the DB can be auto-detected via SSH."
+        )
+        return None
+
+    return db_host, db_port, db_name, db_user, db_password, db_table, server_ip_ssh_verified
+
+
+async def _emit_db_autodetect_error(obj, ssh_port, ssh_options, result: DiscoveryResult) -> None:
+    try:
+        raw = await ssh_run_multi(obj.server_ip, ssh_port, ssh_options, "cat /etc/caps/config.ini")
+        diag_parser = RawConfigParser(strict=False)
+        diag_parser.read_string(raw)
+        found_sections = diag_parser.sections()
+        found_keys = {s: list(diag_parser.options(s)) for s in found_sections}
+        result.errors.append(
+            f"db_uri not found in server config. "
+            f"Sections found: {found_sections}. "
+            f"Keys per section: {found_keys}. "
+            f"Fill in DB fields manually on the object."
+        )
+    except Exception as diag_exc:
+        result.errors.append(
+            f"Could not read server config for auto-detection: {diag_exc}. "
+            f"Fill in DB fields manually on the object."
+        )
+
+
+async def _pre_upsert_db_device(
+    db: AsyncSession,
+    obj: Object,
+    db_host: str | None,
+    result: DiscoveryResult,
+    now: datetime,
+) -> tuple:
+    """Upsert the CAPS DB server as a device before the DB connection attempt.
+
+    Returns (db_device_obj, use_tunnel).
+    """
+    db_device_obj: Device | None = None
+    if db_host and db_host != obj.server_ip:
+        try:
+            db_device_obj = (await db.execute(
+                select(Device).where(Device.object_id == obj.id, Device.ip == db_host)
+            )).scalar_one_or_none()
+            if db_device_obj is None:
+                db_device_obj = Device(
+                    object_id=obj.id, ip=db_host, hostname="db_server",
+                    category="vm", role="other", source="auto_ssh",
+                    location="server_room", status="unknown",
+                    device_meta={"is_caps_db": True},
+                )
+                db.add(db_device_obj)
+                await db.flush()
+                logger.info("[discovery] %s (db_server) - PRE-ADDED before connection attempt", db_host)
+                result.created += 1
+                result.actions.append(DeviceAction(
+                    ip=db_host, hostname="db_server", action="ADDED",
+                    reason="object infrastructure device (vm/db_server)",
+                    category="vm", role="other",
+                ))
+        except Exception as exc:
+            logger.warning("[discovery] Failed to pre-upsert db_server %s: %s", db_host, exc)
+
+    device_via_tunnel = bool(
+        (db_device_obj.connection_params or {}).get("db_via_tunnel")
+    ) if db_device_obj else False
+    use_tunnel = device_via_tunnel or obj.db_via_tunnel
+
+    return db_device_obj, use_tunnel
+
+
+async def _sync_rack_entries(
+    db: AsyncSession,
+    obj: Object,
+    rack_hosts: list[tuple],
+) -> dict[str, int]:
+    """Ensure a Rack row exists for every rack returned from the CAPS DB.
+
+    Returns a mapping of external_id → Rack.id.
+    """
+    rack_id_mapping: dict[str, int] = {}
+    for ext_id, _, rack_name, rack_type_db in rack_hosts:
+        existing = (await db.execute(
+            select(Rack).where(
+                Rack.object_id == obj.id,
+                Rack.name.in_([ext_id, rack_name]),
+            )
+        )).scalar_one_or_none()
+        if existing:
+            rack_id_mapping[ext_id] = existing.id
+            if existing.name != rack_name:
+                logger.info("[discovery] Renamed rack '%s' → '%s'", existing.name, rack_name)
+                existing.name = rack_name
+            if rack_type_db and not existing.rack_type:
+                existing.rack_type = rack_type_db
+        else:
+            new_rack = Rack(object_id=obj.id, name=rack_name, rack_type=rack_type_db)
+            db.add(new_rack)
+            await db.flush()
+            rack_id_mapping[ext_id] = new_rack.id
+            logger.info("[discovery] Created rack '%s' rack_type=%s (id=%d)", rack_name, rack_type_db, new_rack.id)
+    return rack_id_mapping
+
+
+async def _upsert_infra_devices(
+    db: AsyncSession,
+    obj: Object,
+    infra_devices: list[tuple],
+    result: DiscoveryResult,
+    now: datetime,
+) -> None:
+    for infra_ip, infra_category, infra_hostname, infra_verified in infra_devices:
+        try:
+            existing = (await db.execute(
+                select(Device).where(Device.object_id == obj.id, Device.ip == infra_ip)
+            )).scalar_one_or_none()
+
+            if existing is None:
+                db.add(Device(
+                    object_id=obj.id, ip=infra_ip, hostname=infra_hostname,
+                    category=infra_category, role="other", source="auto_ssh",
+                    location="server_room",
+                    status="online" if infra_verified else "unknown",
+                    last_seen=now if infra_verified else None,
+                ))
+                logger.info(
+                    "[discovery] %s (%s) - ADDED: category=%s, status=%s",
+                    infra_ip, infra_hostname, infra_category,
+                    "online" if infra_verified else "unknown",
+                )
+                result.created += 1
+                result.actions.append(DeviceAction(
+                    ip=infra_ip, hostname=infra_hostname, action="ADDED",
+                    reason=f"object infrastructure device ({infra_category})",
+                    category=infra_category, role="other",
+                ))
+            else:
+                changes = []
+                if infra_verified:
+                    if existing.status != "online":
+                        existing.status = "online"
+                        changes.append("status: →online")
+                    existing.last_seen = now
+                    changes.append(f"last_seen: →{now.isoformat()}")
+                if changes:
+                    logger.info("[discovery] %s (%s) - UPDATED: %s", infra_ip, infra_hostname, ", ".join(changes))
+                    result.updated += 1
+                    result.actions.append(DeviceAction(
+                        ip=infra_ip, hostname=existing.hostname or infra_hostname,
+                        action="UPDATED", reason="; ".join(changes),
+                        category=existing.category, role=existing.role,
+                    ))
+                else:
+                    logger.info("[discovery] %s (%s) - SKIPPED: already up to date", infra_ip, infra_hostname)
+                    result.skipped += 1
+                    result.actions.append(DeviceAction(
+                        ip=infra_ip, hostname=existing.hostname or infra_hostname,
+                        action="SKIPPED", reason=f"already exists (category={existing.category})",
+                        category=existing.category, role=existing.role,
+                    ))
+        except Exception as exc:
+            logger.error("[discovery] %s (%s) - ERROR: %s", infra_ip, infra_hostname, exc)
+            result.errors.append(f"Error upserting infra device {infra_ip}: {exc}")
+
+    if infra_devices:
+        await db.flush()
+
+
+async def _resolve_rack_config_path(
+    db: AsyncSession, obj: Object, rack_host: str, default_path: str
+) -> str:
+    """Return the config path to use for this rack, honouring any per-device override."""
+    existing = (await db.execute(
+        select(Device).where(Device.object_id == obj.id, Device.ip == rack_host)
+    )).scalar_one_or_none()
+    if existing and existing.connection_params:
+        return existing.connection_params.get("config_path", default_path)
+    return default_path
+
+
+async def _ssh_fetch_rack_config(
+    rack_host: str,
+    ssh_port: int,
+    ssh_options: list,
+    config_path: str,
+    rack_name: str,
+    result: DiscoveryResult,
+    _emit,
+) -> tuple[bool, str | None]:
+    """Attempt to SSH into rack_host and read its config file.
+
+    Returns (ssh_ok, config_text).
+    """
+    await _emit("disc_action", {"message": f"SSH → {rack_host} ({rack_name})"})
+    try:
+        config_text = await ssh_run_multi(rack_host, ssh_port, ssh_options, f"cat {config_path}")
+        return True, config_text
+    except Exception as exc:
+        result.errors.append(f"SSH to {rack_host} failed: {exc}")
+        return False, None
+
+
+async def _upsert_rack_device(
+    db: AsyncSession,
+    obj: Object,
+    external_id: str,
+    rack_host: str,
+    rack_name: str,
+    rack_id_mapping: dict,
+    rack_ssh_ok: bool,
+    now: datetime,
+    result: DiscoveryResult,
+) -> None:
+    """Upsert the rack PC itself as an embedded device."""
+    rack_hostname = external_id
+    logger.info(
+        "[discovery] Processing rack %s (%s), ssh_ok=%s",
+        rack_host, rack_hostname, rack_ssh_ok,
+    )
+    try:
+        existing = (await db.execute(
+            select(Device).where(Device.object_id == obj.id, Device.ip == rack_host)
+        )).scalar_one_or_none()
+
+        if existing is None:
+            status = "online" if rack_ssh_ok else "unknown"
+            db.add(Device(
+                object_id=obj.id, ip=rack_host, hostname=rack_hostname,
+                category="embedded", role="other", source="auto_ssh",
+                rack_id=rack_id_mapping.get(external_id),
+                status=status, last_seen=now if rack_ssh_ok else None,
+            ))
+            logger.info("[discovery] %s (%s) - ADDED: status=%s", rack_host, rack_hostname, status)
+            result.created += 1
+            result.actions.append(DeviceAction(
+                ip=rack_host, hostname=rack_hostname, action="ADDED",
+                reason="rack PC from racks table" + (" (SSH ok)" if rack_ssh_ok else " (SSH failed, added as unknown)"),
+                category="embedded", role="other",
+            ))
+        else:
+            changes = []
+            if rack_ssh_ok:
+                if existing.status != "online":
+                    existing.status = "online"
+                    changes.append("status: →online")
+                existing.last_seen = now
+                changes.append("last_seen: updated")
+            if existing.rack_id is None and rack_id_mapping.get(external_id) is not None:
+                existing.rack_id = rack_id_mapping[external_id]
+                changes.append(f"rack_id: None→{existing.rack_id}")
+            if changes:
+                logger.info("[discovery] %s (%s) - UPDATED: %s", rack_host, existing.hostname, ", ".join(changes))
+                result.updated += 1
+                result.actions.append(DeviceAction(
+                    ip=rack_host, hostname=existing.hostname, action="UPDATED",
+                    reason="; ".join(changes), category=existing.category, role=existing.role,
+                ))
+            else:
+                reason = "SSH failed, rack already in DB" if not rack_ssh_ok else "rack already up to date"
+                logger.info("[discovery] %s (%s) - SKIPPED: %s", rack_host, existing.hostname, reason)
+                result.skipped += 1
+                result.actions.append(DeviceAction(
+                    ip=rack_host, hostname=existing.hostname, action="SKIPPED",
+                    reason=reason, category=existing.category, role=existing.role,
+                ))
+    except Exception as exc:
+        logger.error("[discovery] %s - ERROR: %s", rack_host, exc)
+        result.errors.append(f"Error processing rack {rack_host}: {exc}")
+        result.actions.append(DeviceAction(
+            ip=rack_host, hostname=rack_hostname, action="ERROR", reason=str(exc),
+            category="embedded", role="other",
+        ))
+
+
+async def _upsert_plugin_device(
+    db: AsyncSession,
+    obj: Object,
+    found: DiscoveredDevice,
+    external_id: str,
+    rack_id_mapping: dict,
+    result: DiscoveryResult,
+    slave_rack_id: int | None = None,
+    slave_identifier: str | None = None,
+) -> None:
+    """Upsert a single device discovered from a plugin section.
+
+    Works for both primary rack devices and slave rack devices.
+    When slave_rack_id / slave_identifier are given, the device is treated as
+    belonging to that slave rack instead of the primary rack keyed by external_id.
+    """
+    is_slave = slave_identifier is not None
+    is_shared = found.category == "vm"
+
+    if is_shared:
+        hostname = found.plugin_name
+        rack_id: int | None = None
+        device_location: str | None = "server_room"
+    elif is_slave:
+        hostname = f"{slave_identifier}-{found.plugin_name}"
+        rack_id = slave_rack_id
+        device_location = None
+    else:
+        hostname = f"{external_id}-{found.plugin_name}"
+        rack_id = rack_id_mapping.get(external_id)
+        device_location = None
+
+    try:
+        existing = (await db.execute(
+            select(Device).where(Device.object_id == obj.id, Device.ip == found.ip)
+        )).scalar_one_or_none()
+
+        if existing is not None:
+            # Protect embedded (rack PC) devices from being overwritten by plugin devices.
+            if existing.category == "embedded":
+                logger.info(
+                    "[discovery] %s (%s) - SKIPPED: device is rack server (embedded)",
+                    found.ip, found.plugin_name,
+                )
+                result.skipped += 1
+                result.actions.append(DeviceAction(
+                    ip=found.ip, hostname=existing.hostname, action="SKIPPED",
+                    reason="device is rack server (embedded), cannot overwrite with plugin device",
+                    category=existing.category, role=existing.role,
+                ))
+                return
+
+            changes = []
+            if existing.category != found.category and existing.source == "auto_ssh":
+                changes.append(f"category: {existing.category}→{found.category}")
+                existing.category = found.category
+            if existing.role == "other" and found.role != "other":
+                changes.append(f"role: other→{found.role}")
+                existing.role = found.role
+            if not is_shared:
+                if not existing.hostname or existing.hostname.startswith("rack"):
+                    if existing.hostname != hostname:
+                        changes.append(f"hostname: '{existing.hostname}'→'{hostname}'")
+                        existing.hostname = hostname
+                if existing.rack_id is None and rack_id is not None:
+                    existing.rack_id = rack_id
+                    changes.append(f"rack_id: None→{rack_id}")
+
+            if changes:
+                logger.info("[discovery] %s (%s) - UPDATED: %s", found.ip, found.plugin_name, ", ".join(changes))
+                result.updated += 1
+                result.actions.append(DeviceAction(
+                    ip=found.ip, hostname=existing.hostname, action="UPDATED",
+                    reason="; ".join(changes), category=found.category, role=found.role,
+                ))
+            else:
+                logger.info(
+                    "[discovery] %s (%s) - SKIPPED: already exists with same values",
+                    found.ip, found.plugin_name,
+                )
+                result.skipped += 1
+                result.actions.append(DeviceAction(
+                    ip=found.ip, hostname=existing.hostname, action="SKIPPED",
+                    reason=f"already exists (source={existing.source}, category={existing.category})",
+                    category=existing.category, role=existing.role,
+                ))
+        else:
+            conn_params: dict = {}
+            if found.http_username:
+                conn_params["username"] = found.http_username
+                conn_params["password_enc"] = encrypt(found.http_password) if found.http_password else ""
+            db.add(Device(
+                object_id=obj.id, ip=found.ip, hostname=hostname,
+                category=found.category, role=found.role, source="auto_ssh",
+                rack_id=rack_id, location=device_location,
+                device_meta={"plugin": found.plugin_name, "cls": found.cls},
+                connection_params=conn_params,
+            ))
+            cred_note = f", http_user={found.http_username}" if found.http_username else ""
+            logger.info(
+                "[discovery] %s (%s) - ADDED: category=%s, role=%s, rack_id=%s%s",
+                found.ip, found.plugin_name, found.category, found.role, rack_id, cred_note,
+            )
+            result.created += 1
+            result.actions.append(DeviceAction(
+                ip=found.ip, hostname=hostname, action="ADDED",
+                reason=f"new device from plugin {found.plugin_name}",
+                category=found.category, role=found.role,
+            ))
+    except Exception as exc:
+        logger.error("[discovery] %s (%s) - ERROR: %s", found.ip, found.plugin_name, exc)
+        result.errors.append(f"Error upserting {found.ip}: {exc}")
+        result.actions.append(DeviceAction(
+            ip=found.ip, hostname=hostname, action="ERROR", reason=str(exc),
+            category=found.category, role=found.role,
+        ))
+
+
+async def _discover_slave_racks(
+    db: AsyncSession,
+    obj: Object,
+    slave_candidates: set[str],
+    rack_hosts: list[tuple],
+    ssh_port: int,
+    ssh_options: list,
+    config_path: str,
+    all_rack_ips_set: set[str],
+    now: datetime,
+    result: DiscoveryResult,
+    _emit,
+) -> None:
+    for candidate_ip in slave_candidates:
+        logger.info("[discovery] Checking slave candidate %s...", candidate_ip)
+        await _emit("disc_action", {"message": f"Проверяем слейв-кандидата {candidate_ip}..."})
+
+        slave_config = await fetch_slave_config(candidate_ip, ssh_port, ssh_options, config_path)
+        is_slave = False
+        slave_identifier = ""
+        master_ip_for_slave = ""
+
+        if slave_config is None:
+            logger.info("[discovery] %s — could not read config, skipping", candidate_ip)
+        else:
+            for _, master_host, _, _ in rack_hosts:
+                ok, _, _ = parse_slave_check(slave_config, master_host, candidate_ip)
+                if ok:
+                    is_slave = True
+                    master_ip_for_slave = master_host
+                    master_ext_id = next(
+                        (ext_id for ext_id, h, _, _ in rack_hosts if h == master_host),
+                        master_host,
+                    )
+                    slave_identifier = f"{master_ext_id}-slave"
+                    break
+
+        if not is_slave:
+            logger.info("[discovery] %s — not a slave rack", candidate_ip)
+            continue
+
+        logger.info("[discovery] %s — SLAVE confirmed (name=%s, master=%s)", candidate_ip, slave_identifier, master_ip_for_slave)
+        await _emit("disc_action", {"message": f"Слейв подтверждён: {slave_identifier} ({candidate_ip})"})
+
+        slave_rack_id = await _ensure_slave_rack(db, obj, slave_identifier, result)
+
+        await _upsert_slave_device(db, obj, candidate_ip, slave_identifier, slave_rack_id, master_ip_for_slave, now, result)
+
+        all_rack_ips_with_slave = list(all_rack_ips_set) + [candidate_ip]
+        slave_devices = parse_config(slave_config, candidate_ip, all_rack_ips_with_slave)
+        logger.info("[discovery] Slave %s: found %d plugin devices", candidate_ip, len(slave_devices))
+        await _emit("disc_slave_found", {
+            "rack_name": slave_identifier, "ip": candidate_ip, "found": len(slave_devices)
+        })
+
+        for found in slave_devices:
+            await _upsert_plugin_device(
+                db, obj, found,
+                external_id="",  # not used for slaves (is_slave=True path)
+                rack_id_mapping={},
+                result=result,
+                slave_rack_id=slave_rack_id,
+                slave_identifier=slave_identifier,
+            )
+
+        try:
+            await db.flush()
+        except Exception as exc:
+            result.errors.append(f"Flush error after slave {candidate_ip}: {exc}")
+
+
+async def _ensure_slave_rack(
+    db: AsyncSession, obj: Object, slave_identifier: str, result: DiscoveryResult
+) -> int | None:
+    """Ensure a Rack entry exists for the slave and return its ID."""
+    try:
+        existing = (await db.execute(
+            select(Rack).where(Rack.object_id == obj.id, Rack.name == slave_identifier)
+        )).scalar_one_or_none()
+        if existing:
+            return existing.id
+        new_rack = Rack(
+            object_id=obj.id, name=slave_identifier,
+            description="slave (auto-discovered via ARP)",
+        )
+        db.add(new_rack)
+        await db.flush()
+        logger.info("[discovery] Created rack '%s' (id=%d) for slave", slave_identifier, new_rack.id)
+        return new_rack.id
+    except Exception as exc:
+        result.errors.append(f"Error creating rack entry for slave: {exc}")
+        return None
+
+
+async def _upsert_slave_device(
+    db: AsyncSession,
+    obj: Object,
+    candidate_ip: str,
+    slave_identifier: str,
+    slave_rack_id: int | None,
+    master_ip: str,
+    now: datetime,
+    result: DiscoveryResult,
+) -> None:
+    try:
+        existing = (await db.execute(
+            select(Device).where(Device.object_id == obj.id, Device.ip == candidate_ip)
+        )).scalar_one_or_none()
+        if existing is None:
+            db.add(Device(
+                object_id=obj.id, ip=candidate_ip, hostname=slave_identifier,
+                category="embedded", role="other", source="auto_ssh",
+                rack_id=slave_rack_id, status="online", last_seen=now,
+            ))
+            logger.info("[discovery] %s (%s) - ADDED: slave rack PC", candidate_ip, slave_identifier)
+            result.created += 1
+            result.actions.append(DeviceAction(
+                ip=candidate_ip, hostname=slave_identifier, action="ADDED",
+                reason=f"slave rack PC (master={master_ip})",
+                category="embedded", role="other",
+            ))
+        else:
+            changes = []
+            if existing.status != "online":
+                existing.status = "online"
+                changes.append("status: →online")
+            existing.last_seen = now
+            changes.append("last_seen: updated")
+            if existing.rack_id is None and slave_rack_id:
+                existing.rack_id = slave_rack_id
+                changes.append(f"rack_id: →{slave_rack_id}")
+            result.updated += 1
+            result.actions.append(DeviceAction(
+                ip=candidate_ip, hostname=existing.hostname or slave_identifier,
+                action="UPDATED", reason="; ".join(changes),
+                category="embedded", role="other",
+            ))
+        await db.flush()
+    except Exception as exc:
+        result.errors.append(f"Error upserting slave device {candidate_ip}: {exc}")
+
+
+async def _discover_mikrotik(
+    db: AsyncSession,
+    obj: Object,
+    rack_hosts: list[tuple],
+    ssh_port: int,
+    ssh_options: list,
+    result: DiscoveryResult,
+    _emit,
+) -> None:
+    target_host = obj.server_ip or rack_hosts[0][1]
+    logger.info("[discovery] Phase 3: MikroTik via SSH_CLIENT on %s", target_host)
+    mikrotik_ip = await discover_mikrotik_multi(target_host, ssh_port, ssh_options)
+
+    if not mikrotik_ip:
+        logger.info("[discovery] MikroTik: no IP found via SSH_CLIENT")
+        return
+
+    logger.info("[discovery] MikroTik found at %s", mikrotik_ip)
+    await _emit("disc_infra_found", {"type": "mikrotik", "ip": mikrotik_ip})
+    try:
+        existing = (await db.execute(
+            select(Device).where(Device.object_id == obj.id, Device.ip == mikrotik_ip)
+        )).scalar_one_or_none()
+        if existing is not None:
+            logger.info("[discovery] %s (mikrotik) - SKIPPED: already exists", mikrotik_ip)
+            result.skipped += 1
+            result.actions.append(DeviceAction(
+                ip=mikrotik_ip, hostname="mikrotik", action="SKIPPED",
+                reason=f"already exists (source={existing.source}, category={existing.category})",
+                category=existing.category, role=existing.role,
+            ))
+        else:
+            db.add(Device(
+                object_id=obj.id, ip=mikrotik_ip, hostname="mikrotik",
+                category="router", role="other", source="auto_ssh",
+            ))
+            logger.info("[discovery] %s (mikrotik) - ADDED: category=router", mikrotik_ip)
+            result.created += 1
+            result.actions.append(DeviceAction(
+                ip=mikrotik_ip, hostname="mikrotik", action="ADDED",
+                reason="discovered via SSH_CLIENT",
+                category="router", role="other",
+            ))
+    except Exception as exc:
+        logger.error("[discovery] %s (mikrotik) - ERROR: %s", mikrotik_ip, exc)
+        result.errors.append(f"Error upserting {mikrotik_ip}: {exc}")
+        result.actions.append(DeviceAction(
+            ip=mikrotik_ip, hostname="mikrotik", action="ERROR", reason=str(exc),
+            category="router", role="other",
+        ))
+    await db.flush()
+
+
+async def _discover_proxmox(
+    db: AsyncSession,
+    obj: Object,
+    ssh_port: int,
+    ssh_password: str,
+    rack_hosts: list[tuple],
+    result: DiscoveryResult,
+    _emit,
+) -> None:
+    target_host = obj.server_ip or rack_hosts[0][1]
+    logger.info("[discovery] Phase 4: Proxmox port 8006 scan on %s", target_host)
+    proxmox_ips = await discover_proxmox_via_port_scan(
+        host=target_host, port=obj.ssh_port or 22,
+        username=obj.ssh_user, password=ssh_password,
+    )
+    if not proxmox_ips:
+        logger.info("[discovery] Proxmox: no servers found on port 8006")
+        return
+
+    logger.info("[discovery] Proxmox: found %d server(s): %s", len(proxmox_ips), proxmox_ips)
+    for proxmox_ip in proxmox_ips:
+        await _emit("disc_infra_found", {"type": "proxmox", "ip": proxmox_ip})
+
+    for idx, proxmox_ip in enumerate(proxmox_ips, 1):
+        hostname = f"proxmox_{idx}"
+        try:
+            existing = (await db.execute(
+                select(Device).where(Device.object_id == obj.id, Device.ip == proxmox_ip)
+            )).scalar_one_or_none()
+            if existing is not None:
+                logger.info("[discovery] %s (%s) - SKIPPED: already exists", proxmox_ip, hostname)
+                result.skipped += 1
+                result.actions.append(DeviceAction(
+                    ip=proxmox_ip, hostname=hostname, action="SKIPPED",
+                    reason=f"already exists (source={existing.source}, category={existing.category})",
+                    category=existing.category, role=existing.role,
+                ))
+            else:
+                db.add(Device(
+                    object_id=obj.id, ip=proxmox_ip, hostname=hostname,
+                    category="vm", role="other", source="auto_ssh",
+                ))
+                logger.info("[discovery] %s (%s) - ADDED: category=vm", proxmox_ip, hostname)
+                result.created += 1
+                result.actions.append(DeviceAction(
+                    ip=proxmox_ip, hostname=hostname, action="ADDED",
+                    reason="discovered via port 8006 scan",
+                    category="vm", role="other",
+                ))
+        except Exception as exc:
+            logger.error("[discovery] %s (%s) - ERROR: %s", proxmox_ip, hostname, exc)
+            result.errors.append(f"Error upserting {proxmox_ip}: {exc}")
+            result.actions.append(DeviceAction(
+                ip=proxmox_ip, hostname=hostname, action="ERROR", reason=str(exc),
+                category="vm", role="other",
+            ))
+    await db.flush()
+
+
+# ── Summary logging ───────────────────────────────────────────────────────────
+
+def _log_summary(result: DiscoveryResult) -> None:
+    sep = "=" * 100
+    logger.info(sep)
+    logger.info(
+        "[discovery] SUMMARY: ADDED=%d, UPDATED=%d, SKIPPED=%d, ERRORS=%d",
+        result.created, result.updated, result.skipped, len(result.errors),
+    )
+    by_action: dict[str, list[DeviceAction]] = {}
+    for action in result.actions:
+        by_action.setdefault(action.action, []).append(action)
+
+    for label, actions in by_action.items():
+        log_fn = logger.error if label == "ERROR" else logger.info
+        log_fn("[discovery] ─── %s devices ───", label)
+        for a in actions:
+            log_fn("  %-20s (%-30s) | %-15s | %-10s | %s", a.ip, a.hostname, a.category, a.role, a.reason)
+    logger.info(sep)

backend/app/services/discovery/ssh_client.py

@@ -0,0 +1,222 @@
+"""
+Low-level SSH operations for device discovery.
+No SQLAlchemy, no business logic — pure network I/O.
+"""
+
+import asyncio
+import logging
+import re
+
+import asyncssh
+
+logger = logging.getLogger(__name__)
+
+
+# ── Single connection helpers (used internally and for legacy paths) ──────────
+
+async def read_remote_file(
+    host: str,
+    port: int,
+    username: str,
+    password: str,
+    remote_path: str,
+) -> str:
+    """SSH into host and return the contents of remote_path."""
+    async with asyncssh.connect(
+        host, port=port, username=username, password=password,
+        known_hosts=None, connect_timeout=10,
+    ) as conn:
+        result = await conn.run(f"cat {remote_path}", check=True)
+        return result.stdout
+
+
+async def get_ip_neighbors(
+    host: str, port: int, username: str, password: str,
+) -> list[str]:
+    """Run `ip neigh show` and return REACHABLE/DELAY neighbour IPs."""
+    try:
+        async with asyncssh.connect(
+            host, port=port, username=username, password=password,
+            known_hosts=None, connect_timeout=10,
+        ) as conn:
+            result = await conn.run("ip neigh show", check=False)
+            return _parse_neigh_output(result.stdout)
+    except Exception as exc:
+        logger.warning("[discovery] ip neigh failed on %s: %s", host, exc)
+        return []
+
+
+# ── Multi-auth helpers ────────────────────────────────────────────────────────
+
+async def ssh_run_multi(
+    host: str, port: int, options: list, command: str, timeout: int = 15,
+) -> str:
+    """Try each SSH auth option in order; return stdout on first success or raise."""
+    last_exc: Exception = Exception("No auth options provided")
+    for i, opt in enumerate(options):
+        _log_attempt(i, len(options), host, opt)
+        try:
+            connect_kwargs = _build_connect_kwargs(host, port, opt)
+            async with asyncssh.connect(**connect_kwargs) as conn:
+                result = await asyncio.wait_for(
+                    conn.run(command, check=True), timeout=timeout
+                )
+                logger.info(
+                    "[discovery] SSH auth succeeded for %s with %s(%s)",
+                    host, opt.type, opt.username,
+                )
+                return result.stdout
+        except Exception as exc:
+            last_exc = exc
+            logger.info(
+                "[discovery] SSH auth failed for %s with %s(%s): %s",
+                host, opt.type, opt.username, exc,
+            )
+    raise last_exc
+
+
+async def get_ip_neighbors_multi(host: str, port: int, options: list) -> list[str]:
+    """Multi-auth version of get_ip_neighbors."""
+    try:
+        stdout = await ssh_run_multi(host, port, options, "ip neigh show", timeout=15)
+        return _parse_neigh_output(stdout)
+    except Exception as exc:
+        logger.warning("[discovery] ip neigh failed on %s: %s", host, exc)
+        return []
+
+
+async def discover_mikrotik_multi(host: str, port: int, options: list) -> str | None:
+    """Discover MikroTik IP via SSH_CLIENT environment variable (multi-auth)."""
+    try:
+        stdout = await ssh_run_multi(host, port, options, "echo $SSH_CLIENT", timeout=15)
+        parts = stdout.strip().split()
+        return parts[0] if parts else None
+    except Exception as exc:
+        logger.warning("[discovery] Failed to discover MikroTik via SSH_CLIENT: %s", exc)
+        return None
+
+
+async def fetch_slave_config(
+    host: str, port: int, options: list, config_path: str,
+) -> str | None:
+    """Read a remote config file from a slave candidate. Returns None on failure."""
+    try:
+        return await ssh_run_multi(host, port, options, f"cat {config_path}", timeout=15)
+    except Exception:
+        return None
+
+
+async def discover_proxmox_via_port_scan(
+    host: str, port: int, username: str, password: str,
+) -> list[str]:
+    """Discover Proxmox servers on the local subnet by scanning port 8006.
+
+    Note: uses single-auth (password) because it's called from the legacy path.
+    Migrate to multi-auth when needed.
+    """
+    try:
+        async with asyncssh.connect(
+            host, port=port, username=username, password=password,
+            known_hosts=None, connect_timeout=10,
+        ) as conn:
+            scan_script = (
+                "subnet=$(ip -o -f inet addr show | awk '!/ lo |docker|br-|wg/ {print $4; exit}' | "
+                "cut -d/ -f1 | awk -F. '{print $1\".\"$2\".\"$3}') && "
+                "(set +m; for ip in $subnet.{1..254}; do "
+                "timeout 0.3 bash -c \"echo > /dev/tcp/$ip/8006\" 2>/dev/null && echo \"$ip\" & "
+                "done; wait)"
+            )
+            result = await conn.run(scan_script, check=False)
+            return [ln.strip() for ln in result.stdout.strip().split("\n") if ln.strip()]
+    except Exception as exc:
+        logger.warning("[discovery] Proxmox port scan failed on %s: %s", host, exc)
+        return []
+
+
+# ── Tunnel ────────────────────────────────────────────────────────────────────
+
+async def open_ssh_tunnel(
+    tunnel_host: str,
+    tunnel_port: int,
+    ssh_options: list,
+    target_host: str,
+    target_port: int,
+):
+    """Open an asyncssh connection, trying each option in order.
+
+    Returns an open asyncssh connection on first success, or raises.
+    The caller is responsible for closing the connection.
+    """
+    last_exc: Exception = Exception("No SSH options provided")
+    for opt in ssh_options:
+        try:
+            connect_kwargs = _build_connect_kwargs(tunnel_host, tunnel_port, opt)
+            if opt.type == "key":
+                logger.info(
+                    "[discovery] Tunnel attempt for %s — key: user=%s key_path=%s",
+                    tunnel_host, opt.username, opt.key_path,
+                )
+            else:
+                masked = _mask(opt.password)
+                logger.info(
+                    "[discovery] Tunnel attempt for %s — password: user=%s password=%s",
+                    tunnel_host, opt.username, masked,
+                )
+            conn = await asyncssh.connect(**connect_kwargs)
+            logger.info(
+                "[discovery] Tunnel auth succeeded for %s with %s(%s)",
+                tunnel_host, opt.type, opt.username,
+            )
+            return conn
+        except Exception as exc:
+            last_exc = exc
+            logger.info(
+                "[discovery] Tunnel auth failed for %s with %s(%s): %s",
+                tunnel_host, opt.type, opt.username, exc,
+            )
+    raise last_exc
+
+
+# ── Internal helpers ──────────────────────────────────────────────────────────
+
+def _parse_neigh_output(stdout: str) -> list[str]:
+    ips = []
+    for line in stdout.strip().splitlines():
+        parts = line.split()
+        if not parts:
+            continue
+        ip = parts[0]
+        state = parts[-1]
+        if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", ip) and state in ("REACHABLE", "DELAY"):
+            ips.append(ip)
+    return ips
+
+
+def _build_connect_kwargs(host: str, port: int, opt) -> dict:
+    kwargs: dict = dict(
+        host=host, port=port, username=opt.username,
+        known_hosts=None, connect_timeout=10,
+    )
+    if opt.type == "key":
+        kwargs["client_keys"] = [opt.key_path]
+        kwargs["passphrase"] = None
+    else:
+        kwargs["password"] = opt.password
+    return kwargs
+
+
+def _mask(password: str) -> str:
+    return (password[:1] + "***" + password[-1:]) if len(password) > 2 else "***"
+
+
+def _log_attempt(idx: int, total: int, host: str, opt) -> None:
+    if opt.type == "key":
+        logger.info(
+            "[discovery] SSH attempt %d/%d for %s — key: user=%s key_path=%s",
+            idx + 1, total, host, opt.username, opt.key_path,
+        )
+    else:
+        logger.info(
+            "[discovery] SSH attempt %d/%d for %s — password: user=%s password=%s",
+            idx + 1, total, host, opt.username, _mask(opt.password),
+        )

backend/app/services/discovery/types.py

@@ -0,0 +1,42 @@
+"""
+Shared dataclasses and type aliases for the discovery subsystem.
+"""
+
+from collections.abc import Callable, Awaitable
+from dataclasses import dataclass, field
+
+
+@dataclass
+class DiscoveredDevice:
+    """A network device extracted from a CAPS config plugin section."""
+    ip: str
+    category: str
+    role: str
+    plugin_name: str
+    cls: str
+    http_username: str = ""  # from URI credentials (e.g. http://user:pass@ip)
+    http_password: str = ""
+
+
+@dataclass
+class DeviceAction:
+    """Records what happened to one device during a discovery run."""
+    ip: str
+    hostname: str
+    action: str  # "ADDED", "UPDATED", "SKIPPED", "IGNORED", "ERROR"
+    reason: str
+    category: str = ""
+    role: str = ""
+
+
+@dataclass
+class DiscoveryResult:
+    created: int = 0
+    updated: int = 0
+    skipped: int = 0
+    errors: list[str] = field(default_factory=list)
+    actions: list[DeviceAction] = field(default_factory=list)
+
+
+# Async callback: (event_type, data_dict) → None
+ProgressCallback = Callable[[str, dict], Awaitable[None]]

backend/app/services/discovery/uri.py

@@ -0,0 +1,68 @@
+"""
+URI parsing and device classification utilities.
+All functions are pure (no I/O, no DB).
+"""
+
+import re
+from urllib.parse import urlparse
+
+
+_SERIAL_SCHEMES = {"serial", "com"}
+_SERVICE_HOSTS = {"localhost", "127.0.0.1"}
+
+# Plugin name prefixes whose URIs point at the CAPS server itself, not separate devices.
+_SKIP_PLUGIN_PREFIXES = ("mnemo", "processor", "recognition", "photo", "voice")
+
+
+def extract_device_info(uri: str) -> tuple[str, str, str] | None:
+    """Return (ip, username, password) from a URI, or None if not a network device.
+
+    Username and password are extracted from URI credentials (http://user:pass@ip).
+    They may be empty strings if not present in the URI.
+    """
+    if not uri:
+        return None
+    try:
+        parsed = urlparse(uri)
+        scheme = (parsed.scheme or "").lower()
+        if scheme in _SERIAL_SCHEMES:
+            return None
+        hostname = parsed.hostname
+        if not hostname or hostname in _SERVICE_HOSTS:
+            return None
+        if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", hostname):
+            return hostname, parsed.username or "", parsed.password or ""
+    except Exception:
+        pass
+    return None
+
+
+def extract_ip(uri: str) -> str | None:
+    """Return IP from a URI string, or None if not a network device."""
+    info = extract_device_info(uri)
+    return info[0] if info else None
+
+
+def cls_to_category(cls: str) -> str:
+    cls_lower = cls.lower()
+    if "camera" in cls_lower or "facedetect" in cls_lower:
+        return "camera"
+    if "controller" in cls_lower or "ovenmk" in cls_lower or "rodos" in cls_lower:
+        return "io_board"
+    if "payment" in cls_lower or "sberpilot" in cls_lower or "payx" in cls_lower:
+        return "bank_terminal"
+    if "payonline" in cls_lower or "kkt" in cls_lower or "fiscal" in cls_lower:
+        return "cash_register"
+    return "other"
+
+
+def cls_to_role(cls: str, plugin_name: str) -> str:
+    cls_lower = cls.lower()
+    name_lower = plugin_name.lower()
+    if "camera" in cls_lower or "camera" in name_lower:
+        if "grz" in name_lower or "plate" in name_lower or "lpr" in name_lower or "frame" in name_lower:
+            return "plate"
+        if "face" in name_lower or "lico" in name_lower:
+            return "face"
+        return "overview"
+    return "other"