You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
122 lines
3.9 KiB
Python
122 lines
3.9 KiB
Python
import logging
|
|
import os
|
|
import socket
|
|
import yaml
|
|
|
|
def get_ip_address(hostname: str) -> str:
|
|
"""
|
|
Resolve a hostname to an IP address (IPv4 or IPv6).
|
|
|
|
Returns:
|
|
str: The resolved IP address, or the original hostname if resolution fails.
|
|
"""
|
|
if not hostname:
|
|
return hostname
|
|
|
|
try:
|
|
# Try both IPv4 and IPv6 (AF_UNSPEC)
|
|
infos = socket.getaddrinfo(
|
|
hostname, None, family=socket.AF_UNSPEC, type=socket.SOCK_STREAM
|
|
)
|
|
# Prefer IPv4 addresses if available
|
|
for family, _, _, _, sockaddr in infos:
|
|
if family == socket.AF_INET:
|
|
return sockaddr[0]
|
|
# Otherwise, fallback to first valid IPv6
|
|
return infos[0][4][0] if infos else hostname
|
|
except socket.gaierror as e:
|
|
logging.debug(f"DNS lookup failed for {hostname}: {e}")
|
|
except Exception as e:
|
|
logging.debug(f"Unexpected error resolving {hostname}: {e}")
|
|
|
|
return hostname
|
|
|
|
|
|
def to_gb(bytes_value):
|
|
"""Convert bytes to a rounded string in gigabytes."""
|
|
try:
|
|
gb = float(bytes_value) / (1024**3)
|
|
return f"{gb:.2f} GB"
|
|
except Exception:
|
|
return "0.00 GB"
|
|
|
|
|
|
def read_file(file_name, strip_newlines=True, default=None, encoding="utf-8"):
|
|
try:
|
|
with open(file_name, "r", encoding=encoding) as f:
|
|
data = f.read()
|
|
return data.replace("\n", "") if strip_newlines else data
|
|
except FileNotFoundError:
|
|
if default is not None:
|
|
return default
|
|
raise
|
|
|
|
|
|
def read_version():
|
|
base_dir = os.path.dirname(os.path.abspath(__file__))
|
|
version_path = os.path.join(base_dir, "VERSION")
|
|
try:
|
|
with open(version_path, "r") as f:
|
|
return f.read().strip() or "unknown"
|
|
except FileNotFoundError:
|
|
env_version = os.getenv("APP_VERSION")
|
|
return env_version.strip() if env_version else "unknown"
|
|
|
|
|
|
def load_config(path=None):
|
|
"""Load and normalize configuration from YAML file or directory."""
|
|
|
|
logger = logging.getLogger(__name__)
|
|
default_path = "/config/config.yaml"
|
|
|
|
# Resolve config path
|
|
config_path = path or default_path
|
|
if os.path.isdir(config_path):
|
|
config_path = os.path.join(config_path, "config.yaml")
|
|
|
|
if not os.path.exists(config_path):
|
|
raise FileNotFoundError(f"Config file not found: {config_path}")
|
|
|
|
with open(config_path, "r") as f:
|
|
config = yaml.safe_load(f) or {}
|
|
|
|
# --- Normalization helpers ------------------------------------------------
|
|
def to_bool(value):
|
|
"""Coerce common truthy/falsey forms to proper bool."""
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, (int, float)):
|
|
return bool(value)
|
|
if value is None:
|
|
return False
|
|
return str(value).strip().lower() in ("true", "1", "yes", "on")
|
|
|
|
def normalize_section(section, bool_keys):
|
|
"""Normalize booleans within a nested section."""
|
|
if not isinstance(config.get(section), dict):
|
|
return
|
|
for key in bool_keys:
|
|
if key in config[section]:
|
|
config[section][key] = to_bool(config[section][key])
|
|
|
|
# --- Global defaults + normalization -------------------------------------
|
|
config.setdefault("version", "1.0.0")
|
|
config.setdefault("debug", False)
|
|
config.setdefault("hide_ts", False)
|
|
config.setdefault("timezone", "UTC")
|
|
|
|
# normalize top-level flags
|
|
config["debug"] = to_bool(config.get("debug"))
|
|
config["hide_ts"] = to_bool(config.get("hide_ts"))
|
|
|
|
# Example: normalize booleans within sections
|
|
normalize_section("mqtt", ["tls", "retain", "clean_session"])
|
|
normalize_section("amcrest", ["webrtc", "verify_ssl"])
|
|
normalize_section("service", ["enabled", "auto_restart"])
|
|
|
|
# Add metadata for debugging/logging
|
|
config["config_path"] = os.path.abspath(config_path)
|
|
config["config_from"] = "file" if os.path.exists(config_path) else "defaults"
|
|
|
|
return config
|