178 lines
6.4 KiB
Python
178 lines
6.4 KiB
Python
"""
|
|
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
|