# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

import re
import time as _time
from typing import Any


class OpenWrtSystemMixin:
    """Mixin providing system-level NAPALM getters (environment, NTP, SNMP, users, services, updates, device actions)."""

    def get_environment(self) -> dict[str, Any]:
        """Return device environment data (CPU, memory).

        CPU usage from ``/proc/stat`` (two samples 1 second apart via ``awk``).
        Memory from ``/proc/meminfo``.
        """
        cpu_out = self._send_command(
            "awk '/^cpu /{idle1=$5; total1=$2+$3+$4+$5+$6+$7+$8} END{print (1-(idle1/total1))*100}' /proc/stat"
        )
        mem_out = self._send_command("cat /proc/meminfo")

        cpu_pct = 0.0
        try:
            cpu_pct = float(cpu_out.strip())
        except (ValueError, AttributeError):
            pass

        mem_total = 0
        mem_available = 0
        for line in mem_out.splitlines():
            if line.startswith("MemTotal:"):
                try:
                    mem_total = int(line.split()[1])
                except (IndexError, ValueError):
                    pass
            elif line.startswith("MemAvailable:"):
                try:
                    mem_available = int(line.split()[1])
                except (IndexError, ValueError):
                    pass

        return {
            "fans": {},
            "temperature": {},
            "power": {},
            "cpu": {0: {"%usage": round(cpu_pct, 1)}},
            "memory": {
                "available_ram": mem_available * 1024,
                "used_ram": (mem_total - mem_available) * 1024,
            },
        }

    def get_system_config(self) -> dict[str, Any]:
        """Return system-level configuration from UCI.

        Reads ``uci show system`` and ``uci show dropbear`` to collect:

        * hostname (str)
        * timezone (str) — POSIX TZ string, e.g. ``"CET-1CEST,M3.5.0,M10.5.0/3"``
        * zonename (str) — human-readable name, e.g. ``"Europe/Berlin"``
        * ntp_servers (list[str])
        * dropbear_port (int) — SSH port
        * dropbear_password_auth (bool) — whether password login is allowed
        * dropbear_root_password_auth (bool)
        """
        sys_out = self._send_command("uci show system 2>/dev/null || true")
        db_out = self._send_command("uci show dropbear 2>/dev/null || true")

        sys_cfg: dict[str, str] = {}
        for line in sys_out.splitlines():
            m = re.match(r"system\.@system\[0\]\.(\w+)='([^']*)'", line.strip())
            if m:
                sys_cfg[m.group(1)] = m.group(2)

        # NTP server list: all servers on a single line, space-separated quoted values
        # e.g. system.ntp.server='0.openwrt.pool.ntp.org' '1.openwrt.pool.ntp.org' ...
        ntp_servers: list[str] = []
        for line in sys_out.splitlines():
            if re.match(r"system\.ntp\.server=", line.strip()):
                ntp_servers = re.findall(r"'([^']+)'", line)
                break

        # Dropbear settings
        db_cfg: dict[str, str] = {}
        for line in db_out.splitlines():
            # May be @dropbear[0] or named section
            m = re.match(r"dropbear\.[@\w]+\.(\w+)='([^']*)'", line.strip())
            if m:
                db_cfg.setdefault(m.group(1), m.group(2))

        try:
            ssh_port = int(db_cfg.get("Port", "22"))
        except (ValueError, TypeError):
            ssh_port = 22

        def _bool_uci(val: str, default: bool = True) -> bool:
            return val.lower() not in ("0", "off", "false", "no") if val else default

        return {
            "hostname": sys_cfg.get("hostname", ""),
            "timezone": sys_cfg.get("timezone", ""),
            "zonename": sys_cfg.get("zonename", ""),
            "ntp_servers": ntp_servers,
            "dropbear_port": ssh_port,
            "dropbear_password_auth": _bool_uci(db_cfg.get("PasswordAuth", "on")),
            "dropbear_root_password_auth": _bool_uci(db_cfg.get("RootPasswordAuth", "on")),
        }

    def get_snmp_information(self) -> dict[str, Any]:
        """Return SNMP configuration from ``uci show snmpd``.

        UCI example::

            snmpd.@com2sec[0].community='public'
            snmpd.@com2sec[0].secname='public'
            snmpd.@system[0].sysContact='root@localhost'
            snmpd.@system[0].sysLocation='Unknown'
        """
        uci_out = self._send_command("uci show snmpd")

        contact = ""
        location = ""
        chassis_id = ""
        community: dict[str, Any] = {}

        # Track com2sec entries by index
        com2sec: dict[str, dict[str, str]] = {}

        for line in uci_out.splitlines():
            line_s = line.strip()
            m = re.match(r"snmpd\.@com2sec\[(\d+)\]\.(\w+)='([^']*)'", line_s)
            if m:
                idx, key, val = m.group(1), m.group(2), m.group(3)
                com2sec.setdefault(idx, {})[key] = val
                continue
            m = re.match(r"snmpd\.@system\[0\]\.sys(\w+)='([^']*)'", line_s)
            if m:
                key, val = m.group(1).lower(), m.group(2)
                if key == "contact":
                    contact = val
                elif key == "location":
                    location = val
                elif key == "name":
                    chassis_id = val

        for entry in com2sec.values():
            name = entry.get("community", entry.get("secname", ""))
            if not name:
                continue
            # OpenWrt snmpd doesn't distinguish rw/ro per community via UCI by default
            mode = "ro"
            if entry.get("secname", "").lower() in ("private", "readwrite", "rw"):
                mode = "rw"
            community[name] = {
                "acl": entry.get("source", "N/A"),
                "mode": mode,
            }

        return {
            "chassis_id": chassis_id,
            "community": community,
            "contact": contact,
            "location": location,
        }

    def get_snmp_config(self):
        """Return SNMP agent config if snmpd is installed and running on OpenWrt."""
        try:
            from napalm_device_types.models import SNMPConfigDict
        except ImportError:
            return None

        running = (
            self._send_command(
                "/etc/init.d/snmpd running 2>/dev/null && echo active || echo inactive"
            ).strip() == "active"
        )
        if not running:
            return None

        community = "public"
        try:
            # UCI config (set by luci-app-snmpd)
            uci_comm = self._send_command(
                "uci -q get snmpd.public.community 2>/dev/null || "
                "uci -q get snmpd.@com2sec[0].community 2>/dev/null || echo ''"
            ).strip()
            if uci_comm:
                community = uci_comm
        except Exception:
            pass

        return SNMPConfigDict(running=True, community=community, port=161, version="2c")

    def _action_fix_snmp(self) -> dict[str, Any]:
        """Install and configure snmpd on OpenWrt.

        Installs snmpd-nossl (the daemon) and luci-app-snmpd (UCI schema +
        proper procd init script).  Configures community 'public' via UCI.
        """
        lines: list = []

        # 1. Install packages — snmpd-nossl (daemon) + luci-app-snmpd (UCI init)
        pm = self._pm_type()
        if pm == "apk":
            raw = self._send_command("apk add snmpd-nossl luci-app-snmpd 2>&1")
        else:
            self._send_command("opkg update 2>/dev/null || true")
            raw = self._send_command("opkg install snmpd-nossl luci-app-snmpd 2>&1")
        out = self._clean_pkg_output(raw)
        low = out.lower()
        installed = not any(kw in low for kw in ("error:", "failed"))
        lines.append(f"[install] {out[-300:]}")

        if not installed and "already installed" not in low:
            return {"success": False, "output": "\n".join(lines)}

        # 2. Configure via UCI — modify the existing default sections only.
        #    Do NOT create new named sections (causes duplicate directives in
        #    the generated /var/run/snmpd.conf which crashes snmpd).
        #    Also remove any stale named sections from previous fix attempts.
        # The init script reads these UCI field names to generate /var/run/snmpd.conf:
        #   agent:  agentaddress
        #   com2sec: secname, source, community
        #   group:  group (name!), version, secname
        #   view:   viewname (not name!), type, oid
        #   access: group, version, level, prefix, read, write, notify
        # Default luci-app-snmpd schema uses different field names for group/view/access,
        # so we patch all required fields explicitly.
        uci_cmds = [
            # Remove any stale named sections from previous runs
            "uci -q delete snmpd.agent",
            "uci -q delete snmpd.public",
            # agent
            "uci set snmpd.@agent[0].agentaddress='161'",
            # com2sec
            "uci set snmpd.@com2sec[0].secname='ro'",
            "uci set snmpd.@com2sec[0].source='0.0.0.0/0'",
            "uci set snmpd.@com2sec[0].community='public'",
            # group — init script reads field 'group' (not 'name')
            "uci set snmpd.@group[0].group='rogroup'",
            "uci set snmpd.@group[0].version='v2c'",
            "uci set snmpd.@group[0].secname='ro'",
            # view — init script reads field 'viewname' (not 'name')
            "uci set snmpd.@view[0].viewname='all'",
            "uci set snmpd.@view[0].type='included'",
            "uci set snmpd.@view[0].oid='.1'",
            # access — init script needs write + notify or it returns early
            "uci set snmpd.@access[0].group='rogroup'",
            "uci set snmpd.@access[0].context='none'",
            "uci set snmpd.@access[0].version='v2c'",
            "uci set snmpd.@access[0].level='noAuthNoPriv'",
            "uci set snmpd.@access[0].prefix='exact'",
            "uci set snmpd.@access[0].read='all'",
            "uci set snmpd.@access[0].write='none'",
            "uci set snmpd.@access[0].notify='none'",
            "uci commit snmpd",
        ]
        for cmd in uci_cmds:
            self._send_command(f"{cmd} 2>/dev/null || true")
        lines.append("[config] Configured snmpd via UCI (all required fields set).")

        # 3. Firewall: allow UDP 161 from the management zone
        try:
            # Use `ss` (always available on OpenWrt) to find the IP of the current SSH client
            raw_conn = self._send_command(
                "ss -tnp 2>/dev/null | awk '/ESTAB.*:22/{print $5}' | head -1 | sed 's/:[0-9]*$//'"
            ).strip()
            # Strip IPv6-mapped prefix if present
            if raw_conn.startswith("::ffff:"):
                raw_conn = raw_conn[7:]
            peer_ip = raw_conn.splitlines()[-1].strip() if raw_conn else ""

            if peer_ip and peer_ip not in ("", "0.0.0.0", "::"):
                # Determine which firewall zone owns the interface that routes to peer_ip.
                # Walk uci firewall zones: find the zone whose associated network interface
                # has an address in the same /24 as peer_ip.
                peer_prefix = peer_ip.rsplit(".", 1)[0] if "." in peer_ip else ""
                fw_out = self._send_command("uci show firewall 2>/dev/null || true")
                src_zone = "*"
                # Build map zone_name → zone_uci_key
                zone_map: dict[str, str] = {}
                for fw_line in fw_out.splitlines():
                    m = re.match(r"firewall\.(\w+)\.name='([^']+)'", fw_line.strip())
                    if m:
                        zone_map[m.group(2)] = m.group(1)
                # For each zone find its network, then the interface IP
                for zone_name, uci_key in zone_map.items():
                    net_line = self._send_command(
                        f"uci -q get firewall.{uci_key}.network 2>/dev/null || true"
                    ).strip()
                    if not net_line:
                        continue
                    for net_name in net_line.split():
                        iface_ip = self._send_command(
                            f"uci -q get network.{net_name}.ipaddr 2>/dev/null || true"
                        ).strip()
                        if iface_ip and "." in iface_ip:
                            iface_prefix = iface_ip.split("/")[0].rsplit(".", 1)[0]
                            if peer_prefix and iface_prefix == peer_prefix:
                                src_zone = zone_name
                                break
                    if src_zone != "*":
                        break

                self._send_command(
                    f"uci -q delete firewall.snmp_netork 2>/dev/null; "
                    f"uci set firewall.snmp_netork=rule; "
                    f"uci set firewall.snmp_netork.name='Allow-SNMP-from-mgmt'; "
                    f"uci set firewall.snmp_netork.src='{src_zone}'; "
                    f"uci set firewall.snmp_netork.dest_port='161'; "
                    f"uci set firewall.snmp_netork.proto='udp'; "
                    f"uci set firewall.snmp_netork.target='ACCEPT'; "
                    f"uci commit firewall; "
                    f"/etc/init.d/firewall reload 2>/dev/null || true"
                )
                lines.append(f"[firewall] Added UDP:161 allow rule (src zone: {src_zone}).")
            else:
                lines.append("[firewall] Could not detect peer IP via ss — skipping firewall step.")
        except Exception as exc:
            lines.append(f"[firewall] skipped — {exc}")

        # 4. Break any crash-loop, then start cleanly
        self._send_command("/etc/init.d/snmpd stop 2>/dev/null; true")
        _time.sleep(2)
        self._send_command("pkill -9 snmpd 2>/dev/null; true")  # kill crash-loop zombie
        _time.sleep(3)
        self._send_command("/etc/init.d/snmpd enable 2>/dev/null; true")
        self._send_command("/etc/init.d/snmpd start 2>/dev/null; true")
        _time.sleep(4)
        lines.append("[service] snmpd started via procd.")

        # 5. Check if snmpd is now active (no local snmpget on OpenWrt by default)
        status = self._send_command(
            "/etc/init.d/snmpd running 2>/dev/null && echo active || echo inactive"
        ).strip()
        success = status == "active"
        if success:
            lines.append("[ok] snmpd is active.")
        else:
            lines.append(f"[warn] snmpd status: {status}")

        return {"success": success, "output": "\n".join(lines)}

    def get_users(self) -> dict[str, Any]:
        """Return users configured on the device.

        Parses ``/etc/passwd`` for accounts with a valid login shell.
        SSH public keys are read from ``~<user>/.ssh/authorized_keys``
        (Dropbear also stores root keys at ``/etc/dropbear/authorized_keys``).

        Level mapping:
        - UID 0 (root) → 15 (full access)
        - all other users → 1
        """
        passwd_out = self._send_command("cat /etc/passwd")
        # Root authorized_keys locations on OpenWrt
        root_keys_out = self._send_command(
            ["cat /root/.ssh/authorized_keys", "cat /etc/dropbear/authorized_keys"]
        )

        users: dict[str, Any] = {}
        valid_shells = {"/bin/sh", "/bin/ash", "/bin/bash", "/usr/bin/fish"}

        for line in passwd_out.splitlines():
            parts = line.strip().split(":")
            if len(parts) < 7:
                continue
            username, password_hash, uid_str, _, _, home, shell = (
                parts[0], parts[1], parts[2], parts[3], parts[4], parts[5], parts[6],
            )
            if shell not in valid_shells:
                continue
            try:
                uid = int(uid_str)
            except ValueError:
                continue

            level = 15 if uid == 0 else 1

            # Collect SSH keys for this user
            sshkeys: list[str] = []
            if uid == 0:
                for line_k in root_keys_out.splitlines():
                    line_k = line_k.strip()
                    if line_k and not line_k.startswith("#"):
                        sshkeys.append(line_k)
            else:
                # Try reading per-user authorized_keys
                keys_out = self._send_command(f"cat {home}/.ssh/authorized_keys 2>/dev/null")
                for line_k in keys_out.splitlines():
                    line_k = line_k.strip()
                    if line_k and not line_k.startswith("#"):
                        sshkeys.append(line_k)

            users[username] = {
                "level": level,
                "password": password_hash,
                "sshkeys": sshkeys,
            }

        return users

    def get_services(self) -> list[dict[str, Any]]:
        """Return all system services with their running and enabled state.

        Uses ``ubus call service list`` for running/PID info and
        ``/etc/rc.d/S*`` symlinks for enabled-at-boot state.
        """
        import json as _json

        # -- enabled set: names from /etc/rc.d/S<priority><name> symlinks ----
        rc_out = self._send_command(
            "ls /etc/rc.d/ 2>/dev/null | grep '^S' | sed 's/^S[0-9]*//'"
        )
        enabled: set[str] = {s.strip() for s in rc_out.splitlines() if s.strip()}

        # -- running info from procd via ubus ---------------------------------
        ubus_raw = self._send_command("ubus call service list 2>/dev/null")
        ubus_data: dict = {}
        try:
            ubus_data = _json.loads(ubus_raw)
        except (ValueError, TypeError):
            pass

        # Build index from ubus data
        service_map: dict[str, dict] = {}
        for svc_name, svc_info in ubus_data.items():
            if not isinstance(svc_info, dict):
                continue
            instances = svc_info.get("instances", {})
            running = any(
                inst.get("running", False)
                for inst in instances.values()
                if isinstance(inst, dict)
            )
            pid = next(
                (
                    inst.get("pid", 0)
                    for inst in instances.values()
                    if isinstance(inst, dict) and inst.get("running")
                ),
                0,
            )
            service_map[svc_name] = {"running": running, "pid": pid}

        # -- all init scripts -------------------------------------------------
        init_raw = self._send_command("ls -1 /etc/init.d/ 2>/dev/null")
        init_scripts: set[str] = {s.strip() for s in init_raw.splitlines() if s.strip()}

        # Merge: all known services (from init.d + ubus)
        all_names = init_scripts | set(service_map.keys())
        # Exclude procd internal pseudo-service
        all_names.discard("")

        result: list[dict[str, Any]] = []
        for name in sorted(all_names):
            info = service_map.get(name, {})
            result.append({
                "name": name,
                "running": info.get("running", False),
                "enabled": name in enabled,
                "pid": info.get("pid", 0),
            })

        return result

    def manage_service(self, name: str, action: str) -> dict[str, Any]:
        """Execute a lifecycle action (start/stop/restart/enable/disable) on a service."""
        import re as _re
        if not _re.match(r'^[a-zA-Z0-9_\-]+$', name):
            raise ValueError(f"Invalid service name: {name!r}")
        if action not in ('start', 'stop', 'restart', 'enable', 'disable'):
            raise ValueError(f"Invalid action: {action!r}")
        output = self._send_command(f"/etc/init.d/{name} {action} 2>&1")
        return {"success": True, "output": output}

    def get_device_warnings(self) -> list[dict[str, Any]]:
        """Return a list of warning dicts for issues detected on this device.

        Currently detects:
        - lldpd not installed (LLDP neighbor discovery unavailable)
        - package updates available (uses local package cache, no network call)
        - update notifications not configured (auc not installed, opkg only)
        """
        warnings: list[dict[str, Any]] = []

        # 1. LLDP daemon
        lldpd_path = self._send_command("which lldpd 2>/dev/null").strip()
        if not lldpd_path:
            warnings.append({
                "code": "lldpd_not_installed",
                "severity": "warning",
                "action": "install_lldpd",
            })

        pm = self._pm_type()

        # 2. Package updates available (local cache only – no opkg update)
        def _pkg_name(line: str, pm_type: str) -> str:
            """Extract just the package name from an upgradable line.

            apk:  'luci-app-firewall-26.143~abc < 26.151~def' → 'luci-app-firewall'
            opkg: 'luci-app-firewall - 1.0 - 1.1'             → 'luci-app-firewall'
            """
            import re as _re
            if pm_type == "apk":
                # Strip trailing ' < ...' then remove the version suffix (-\d...)
                name_ver = line.split(" ")[0]
                m = _re.match(r'^(.*?)-\d', name_ver)
                return m.group(1) if m else name_ver
            else:
                return line.split(" - ")[0].strip()

        try:
            if pm == "apk":
                raw_upg = self._send_command("apk version 2>/dev/null | grep '<'")
            else:
                raw_upg = self._send_command("opkg list-upgradable 2>/dev/null")
            upgradable = [ln.strip() for ln in raw_upg.splitlines() if ln.strip()]
        except Exception:
            upgradable = []

        if upgradable:
            pkg_names = [_pkg_name(ln, pm) for ln in upgradable]
            warnings.append({
                "code": "updates_available",
                "severity": "info",
                "action": None,
                "meta": {
                    "count": len(upgradable),
                    "packages": pkg_names[:10],
                },
            })

        # 3. Attended sysupgrade client not installed (opkg systems only)
        if pm == "opkg":
            auc_path = self._send_command("which auc 2>/dev/null").strip()
            if not auc_path:
                warnings.append({
                    "code": "update_notifications_disabled",
                    "severity": "warning",
                    "action": "install_auc",
                })

        # 4. base64 not available — needed for efficient config apply
        b64_path = self._send_command("command -v base64 2>/dev/null").strip()
        if not b64_path:
            warnings.append({
                "code": "no_base64",
                "severity": "warning",
                "action": "install_coreutils_base64",
            })

        return warnings

    def get_available_updates(self) -> list[dict[str, Any]]:
        """Return list of upgradable packages from the local package manager cache."""
        import re as _re
        pm = self._pm_type()
        updates: list[dict[str, Any]] = []

        if pm == "apk":
            # Output format: "pkgname-current_ver < new_ver"
            raw = self._send_command("apk version 2>/dev/null | grep '<'")
            for line in raw.splitlines():
                line = line.strip()
                m = _re.match(r'^(.+)-(\d\S*)\s+<\s+(\S+)', line)
                if m:
                    updates.append({
                        "name": m.group(1),
                        "current_version": m.group(2),
                        "new_version": m.group(3),
                    })
        else:
            # opkg output: "pkgname - current_ver - new_ver"
            raw = self._send_command("opkg list-upgradable 2>/dev/null")
            for line in raw.splitlines():
                parts = [p.strip() for p in line.split(" - ")]
                if len(parts) == 3:
                    updates.append({
                        "name": parts[0],
                        "current_version": parts[1],
                        "new_version": parts[2],
                    })

        return sorted(updates, key=lambda u: u["name"])

    def apply_updates(self, packages: list[str]) -> dict[str, Any]:
        """Upgrade the given packages using the device's package manager."""
        import re as _re
        for pkg in packages:
            if not _re.match(r'^[a-zA-Z0-9_\-\+\.]+$', pkg):
                raise ValueError(f"Invalid package name: {pkg!r}")
        pm = self._pm_type()
        pkg_args = " ".join(packages)
        if pm == "apk":
            cmd = f"apk upgrade {pkg_args} 2>&1"
        else:
            cmd = f"opkg upgrade {pkg_args} 2>&1"
        output = self._send_command(cmd)
        return {"success": True, "output": output}

    # ------------------------------------------------------------------
    # NTP
    # ------------------------------------------------------------------

    def get_ntp_servers(self) -> dict[str, Any]:
        """Return configured NTP servers from ``uci show system``.

        UCI example::

            system.ntp.server='0.openwrt.pool.ntp.org 1.openwrt.pool.ntp.org'
        """
        uci_out = self._send_command("uci show system")
        servers: dict[str, Any] = {}

        for line in uci_out.splitlines():
            # Handles both list and single-value UCI representations
            m = re.match(r"system\.ntp\.server(?:\[\d+\])?='([^']*)'", line.strip())
            if m:
                for srv in m.group(1).split():
                    srv = srv.strip()
                    if srv:
                        servers[srv] = {}

        return servers

    def get_ntp_peers(self) -> dict[str, Any]:
        """Return NTP peers from ``uci show system``.

        OpenWrt's busybox ntpd does not differentiate peers from servers;
        the same UCI ``ntp.server`` list is returned.
        """
        return self.get_ntp_servers()

    def get_ntp_stats(self) -> list[dict[str, Any]]:
        """Return NTP synchronisation statistics.

        Tries ``ntpq -pn`` first (ntpd), then ``chronyc sources -v`` (chrony).
        Returns an empty list when neither tool is available.

        ``ntpq -pn`` example line::

            *188.114.101.4   188.114.100.1    4 u  107  256  377  164.228 -13.866   2.695

        ``chronyc sources -v`` example line::

            ^* 192.168.1.1                   2   6    17     8  +2345us[  0ns] +/-   15ms
        """
        ntpq_out = self._send_command("ntpq -pn")
        if ntpq_out and not ntpq_out.startswith(("ntpq: ", "sh: ", "ash: ", "command not found")):
            return self._parse_ntpq(ntpq_out)

        chrony_out = self._send_command("chronyc sources -v")
        if chrony_out and not chrony_out.startswith(("sh: ", "ash: ", "command not found")):
            return self._parse_chronyc(chrony_out)

        return []

    @staticmethod
    def _parse_ntpq(output: str) -> list[dict[str, Any]]:
        """Parse ``ntpq -pn`` tabular output."""
        stats = []
        for line in output.splitlines():
            line_s = line.strip()
            if not line_s or line_s.startswith(("remote", "=")):
                continue
            # First char is the tally code (* = synchronized, + = candidate, etc.)
            tally = line_s[0] if line_s[0] in "* +-x.o#" else " "
            parts = line_s[1:].split()
            if len(parts) < 10:
                continue
            try:
                stats.append({
                    "remote": parts[0],
                    "referenceid": parts[1],
                    "synchronized": tally == "*",
                    "stratum": int(parts[2]),
                    "type": parts[3],
                    "when": parts[4],
                    "hostpoll": int(parts[5]),
                    "reachability": int(parts[6], 8),  # octal
                    "delay": float(parts[7]),
                    "offset": float(parts[8]),
                    "jitter": float(parts[9]),
                })
            except (ValueError, IndexError):
                continue
        return stats

    @staticmethod
    def _parse_chronyc(output: str) -> list[dict[str, Any]]:
        """Parse ``chronyc sources -v`` tabular output."""
        stats = []
        for line in output.splitlines():
            line_s = line.strip()
            # Data lines start with ^* ^+ ^- ^?
            m = re.match(r"^(\^[*+\-?])\s+(\S+)\s+(\d+)\s+(\d+)\s+(\d+)\s+(\S+)\s+(.*)", line_s)
            if not m:
                continue
            tally = m.group(1)
            try:
                stats.append({
                    "remote": m.group(2),
                    "referenceid": "",
                    "synchronized": tally == "^*",
                    "stratum": int(m.group(3)),
                    "type": "u",
                    "when": m.group(6),
                    "hostpoll": int(m.group(4)),
                    "reachability": int(m.group(5), 8) if re.match(r"^[0-7]+$", m.group(5)) else 0,
                    "delay": 0.0,
                    "offset": 0.0,
                    "jitter": 0.0,
                })
            except (ValueError, IndexError):
                continue
        return stats