"""Linux system metric collection using /proc, /sys and statvfs.

The module intentionally depends only on the Python standard library so the
dashboard can run on a bare server. Every parser is a pure function and can be
tested with synthetic /proc data (see tests/test_metrics.py).
"""

from __future__ import annotations

import datetime as dt
import glob
import json
import os
import platform
import re
import shutil
import socket
import subprocess
import time
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple

try:  # package import
    from . import __version__
except ImportError:  # pragma: no cover - direct module execution
    __version__ = "1.0.0"


# Filesystems worth showing on a storage dashboard. Pseudo filesystems are
# deliberately excluded so the panel does not fill with tmpfs/proc entries.
REAL_FS_TYPES = {
    "ext2", "ext3", "ext4", "xfs", "btrfs", "zfs", "f2fs", "jfs", "reiserfs",
    "bcachefs", "vfat", "exfat", "ntfs", "ntfs3", "fuseblk", "overlay", "aufs",
    "nfs", "nfs4", "cifs", "smb3", "ceph", "glusterfs", "lustre", "9p",
}

VIRTUAL_IFACE_PREFIXES = (
    "lo", "veth", "docker", "br-", "virbr", "cni", "flannel", "kube", "cali",
    "tunl", "vxlan", "dummy", "ifb", "lxc", "podman",
)

# Linux /proc/stat fields, in order after the "cpu" label.
_CPU_FIELDS = ("user", "nice", "system", "idle", "iowait", "irq", "softirq", "steal")

_MEMINFO_KEYS = {
    "MemTotal": "total",
    "MemFree": "free",
    "MemAvailable": "available",
    "Buffers": "buffers",
    "Cached": "cached",
    "SReclaimable": "sreclaimable",
    "Shmem": "shmem",
    "SwapTotal": "swap_total",
    "SwapFree": "swap_free",
    "Dirty": "dirty",
    "Writeback": "writeback",
    "Slab": "slab",
}

_STAT_RE = re.compile(r"^(\d+) \((.+)\) ([A-Za-z]) (.*)$")


def _read_text(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8", errors="replace")
    except (OSError, ValueError):
        return ""


def _as_int(value: str, default: int = 0) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def _pct(used: float, total: float) -> float:
    if total <= 0:
        return 0.0
    return max(0.0, min(100.0, used / total * 100.0))


def _round(value: Optional[float], digits: int = 1) -> Optional[float]:
    if value is None:
        return None
    return round(float(value), digits)


def parse_cpu_stat(text: str) -> Tuple[Optional[Dict[str, int]], List[Dict[str, int]]]:
    """Parse /proc/stat CPU times into aggregate and per-core dictionaries."""

    aggregate: Optional[Dict[str, int]] = None
    cores: List[Dict[str, int]] = []
    for line in text.splitlines():
        if not line.startswith("cpu"):
            continue
        parts = line.split()
        if len(parts) < 5:
            continue
        label = parts[0]
        try:
            values = [int(v) for v in parts[1:9]]
        except ValueError:
            continue
        values += [0] * (8 - len(values))
        item = dict(zip(_CPU_FIELDS, values))
        item["total"] = sum(values)
        item["idle_all"] = item["idle"] + item["iowait"]
        if label == "cpu":
            aggregate = item
        elif label[3:].isdigit():
            item["index"] = int(label[3:])
            cores.append(item)
    cores.sort(key=lambda c: c["index"])
    return aggregate, cores


def parse_cpuinfo(text: str) -> Dict[str, Any]:
    """Extract model name, logical cores and physical core count from cpuinfo."""

    model = ""
    processors = 0
    physical = set()
    physical_id: Optional[str] = None
    core_id: Optional[str] = None
    for line in text.splitlines() + [""]:
        if not line.strip():
            if physical_id is not None and core_id is not None:
                physical.add((physical_id, core_id))
            physical_id = core_id = None
            continue
        if ":" not in line:
            continue
        key, value = line.split(":", 1)
        key = key.strip().lower()
        value = value.strip()
        if key == "processor":
            processors += 1
        elif key in ("model name", "hardware", "processor model") and not model:
            model = value
        elif key == "physical id":
            physical_id = value
        elif key == "core id":
            core_id = value
    return {
        "model": model,
        "logical_cores": processors,
        "physical_cores": len(physical) if physical else processors,
    }


def parse_meminfo(text: str) -> Dict[str, int]:
    """Parse /proc/meminfo into a canonical byte-valued dictionary."""

    data: Dict[str, int] = {}
    for line in text.splitlines():
        if ":" not in line:
            continue
        key, raw = line.split(":", 1)
        key = key.strip()
        target = _MEMINFO_KEYS.get(key)
        if not target:
            continue
        match = re.search(r"(\d+)", raw)
        if not match:
            continue
        # All tracked fields are reported in kB by the kernel.
        data[target] = int(match.group(1)) * 1024
    return data


def parse_loadavg(text: str) -> Dict[str, Any]:
    parts = text.split()
    if len(parts) < 5:
        return {"load1": 0.0, "load5": 0.0, "load15": 0.0, "running": 0, "threads": 0}
    running = threads = 0
    if "/" in parts[3]:
        left, right = parts[3].split("/", 1)
        running, threads = _as_int(left), _as_int(right)
    return {
        "load1": float(parts[0]),
        "load5": float(parts[1]),
        "load15": float(parts[2]),
        "running": running,
        "threads": threads,
    }


def parse_uptime(text: str) -> float:
    parts = text.split()
    if not parts:
        return 0.0
    try:
        return max(0.0, float(parts[0]))
    except ValueError:
        return 0.0


def parse_net_dev(text: str) -> Dict[str, Dict[str, int]]:
    """Parse /proc/net/dev, returning cumulative counters by interface."""

    result: Dict[str, Dict[str, int]] = {}
    for line in text.splitlines():
        if ":" not in line:
            continue
        name, raw = line.split(":", 1)
        name = name.strip()
        fields = raw.split()
        if not name or len(fields) < 16:
            continue
        try:
            values = [int(v) for v in fields[:16]]
        except ValueError:
            continue
        result[name] = {
            "rx_bytes": values[0],
            "rx_packets": values[1],
            "rx_errors": values[2],
            "rx_drops": values[3],
            "tx_bytes": values[8],
            "tx_packets": values[9],
            "tx_errors": values[10],
            "tx_drops": values[11],
        }
    return result


def parse_diskstats(text: str) -> Dict[str, Dict[str, int]]:
    """Parse /proc/diskstats into cumulative counters by block device."""

    result: Dict[str, Dict[str, int]] = {}
    for line in text.splitlines():
        parts = line.split()
        if len(parts) < 14:
            continue
        name = parts[2]
        try:
            values = [int(v) for v in parts[3:14]]
        except ValueError:
            continue
        result[name] = {
            "reads": values[0],
            "reads_merged": values[1],
            "sectors_read": values[2],
            "read_ms": values[3],
            "writes": values[4],
            "writes_merged": values[5],
            "sectors_written": values[6],
            "write_ms": values[7],
            "io_in_progress": values[8],
            "io_ms": values[9],
            "weighted_io_ms": values[10],
        }
    return result


def _unescape_mount(value: str) -> str:
    def replace(match: re.Match[str]) -> str:
        return chr(int(match.group(1), 8))

    return re.sub(r"\\([0-7]{3})", replace, value)


def parse_mounts(text: str) -> List[Dict[str, str]]:
    mounts = []
    for line in text.splitlines():
        parts = line.split()
        if len(parts) < 4:
            continue
        mounts.append({
            "device": _unescape_mount(parts[0]),
            "mount": _unescape_mount(parts[1]),
            "fstype": parts[2],
            "options": parts[3],
        })
    return mounts


def build_alerts(
    cpu_percent: float,
    memory_percent: float,
    swap_percent: float,
    load1: float,
    cores: int,
    disks: Sequence[Dict[str, Any]],
    temp_c: Optional[float],
) -> List[Dict[str, str]]:
    """Build compact, user-facing alert messages for the status ticker."""

    alerts: List[Dict[str, str]] = []

    def add(severity: str, message: str) -> None:
        alerts.append({"severity": severity, "message": message})

    if cpu_percent >= 95:
        add("crit", f"CPU 使用率 {cpu_percent:.0f}%")
    elif cpu_percent >= 85:
        add("warn", f"CPU 使用率 {cpu_percent:.0f}%")

    if memory_percent >= 95:
        add("crit", f"内存使用率 {memory_percent:.0f}%")
    elif memory_percent >= 85:
        add("warn", f"内存使用率 {memory_percent:.0f}%")

    if swap_percent >= 80:
        add("crit", f"Swap 已用 {swap_percent:.0f}%")
    elif swap_percent >= 50:
        add("warn", f"Swap 已用 {swap_percent:.0f}%")

    if cores > 0 and load1 >= cores * 3:
        add("crit", f"负载 {load1:.2f} / {cores} 核")
    elif cores > 0 and load1 >= cores * 1.5:
        add("warn", f"负载 {load1:.2f} / {cores} 核")

    for disk in sorted(disks, key=lambda d: d.get("percent", 0), reverse=True):
        pct = float(disk.get("percent", 0))
        if pct >= 95:
            add("crit", f"磁盘 {disk.get('mount', '?')} 已用 {pct:.0f}%")
        elif pct >= 85:
            add("warn", f"磁盘 {disk.get('mount', '?')} 已用 {pct:.0f}%")
        if len([a for a in alerts if a["message"].startswith("磁盘")]) >= 2:
            break

    if temp_c is not None:
        if temp_c >= 90:
            add("crit", f"CPU 温度 {temp_c:.0f}°C")
        elif temp_c >= 80:
            add("warn", f"CPU 温度 {temp_c:.0f}°C")

    order = {"crit": 0, "warn": 1, "info": 2}
    alerts.sort(key=lambda a: order.get(a["severity"], 3))
    return alerts


class SystemCollector:
    """Collect one coherent snapshot from a Linux-like /proc and /sys tree."""

    def __init__(
        self,
        proc_root: str = "/proc",
        sys_root: str = "/sys",
        clock: Callable[[], float] = time.monotonic,
        wall_clock: Callable[[], float] = time.time,
    ) -> None:
        self.proc_root = Path(proc_root)
        self.sys_root = Path(sys_root)
        self.clock = clock
        self.wall_clock = wall_clock
        self._ncpu = os.cpu_count() or 1
        self._page_size = self._get_page_size()
        self._static = self._read_static()

        self._prev_cpu: Optional[Dict[str, int]] = None
        self._prev_cpu_cores: List[Dict[str, int]] = []
        self._cpu_total_now = 0
        self._prev_net: Dict[str, Dict[str, int]] = {}
        self._prev_net_ts: Optional[float] = None
        self._prev_disk: Dict[str, Dict[str, int]] = {}
        self._prev_disk_ts: Optional[float] = None
        self._proc_ticks: Dict[int, int] = {}
        self._prev_proc_total = 0
        self._proc_cache: List[Dict[str, Any]] = []
        self._proc_cache_ts = 0.0
        self._cpu_meta: Dict[str, Any] = {"freq_mhz": None, "temp_c": None, "temp_label": None}
        self._cpu_meta_ts = 0.0
        self._disk_cache: List[Dict[str, Any]] = []
        self._disk_cache_ts = 0.0
        self._addr_cache: Dict[str, Any] = {"interfaces": {}, "primary": None, "all": []}
        self._addr_cache_ts = 0.0

    def _get_page_size(self) -> int:
        try:
            return int(os.sysconf("SC_PAGE_SIZE"))
        except (AttributeError, ValueError, OSError):
            return 4096

    def _read_static(self) -> Dict[str, Any]:
        cpuinfo = parse_cpuinfo(_read_text(self.proc_root / "cpuinfo"))
        osr: Dict[str, str] = {}
        for line in _read_text(Path("/etc/os-release")).splitlines():
            if "=" in line:
                key, value = line.split("=", 1)
                osr[key] = value.strip().strip('"')
        if not osr:
            for candidate in (self.proc_root.parent / "etc" / "os-release",):
                for line in _read_text(candidate).splitlines():
                    if "=" in line:
                        key, value = line.split("=", 1)
                        osr[key] = value.strip().strip('"')
        os_name = osr.get("PRETTY_NAME") or osr.get("NAME") or platform.platform()
        if osr.get("VERSION") and osr.get("VERSION") not in os_name:
            os_name = f"{os_name} {osr['VERSION']}"

        cores = len(self._per_core_names()) or self._ncpu
        physical = cpuinfo.get("physical_cores") or cores
        now = dt.datetime.now().astimezone()
        offset = now.utcoffset()
        tz_offset = int(offset.total_seconds()) if offset else -time.timezone
        tz_name = now.tzname() or time.tzname[0] or "LOCAL"

        return {
            "hostname": socket.gethostname() or "unknown",
            "os": os_name,
            "kernel": platform.release(),
            "arch": platform.machine() or "unknown",
            "cpu_model": cpuinfo.get("model") or platform.processor() or "Unknown CPU",
            "cores": cores,
            "physical_cores": physical,
            "tz_offset": tz_offset,
            "tz_name": tz_name,
            "version": __version__,
        }

    def _per_core_names(self) -> List[str]:
        return sorted(p.name for p in self.sys_root.glob("devices/system/cpu/cpu[0-9]*") if p.is_dir())

    # ------------------------------------------------------------------
    # Host / address helpers
    # ------------------------------------------------------------------
    def _default_interface(self) -> Optional[str]:
        text = _read_text(self.proc_root / "net" / "route")
        best: Optional[Tuple[int, str]] = None
        for line in text.splitlines()[1:]:
            fields = line.split()
            if len(fields) < 8:
                continue
            iface, dest, _gw, flags, _ref, _use, metric, _mask = fields[:8]
            try:
                flag_value = int(flags, 16)
            except ValueError:
                continue
            if dest == "00000000" and flag_value & 0x2:
                try:
                    value = int(metric)
                except ValueError:
                    value = 0
                if best is None or value < best[0]:
                    best = (value, iface)
        return best[1] if best else None

    def _read_addresses(self) -> Dict[str, Any]:
        mapping: Dict[str, List[str]] = {}
        ip_bin = shutil.which("ip")
        if ip_bin:
            try:
                proc = subprocess.run(
                    [ip_bin, "-j", "addr", "show"],
                    capture_output=True,
                    text=True,
                    timeout=1.5,
                    check=False,
                )
                if proc.returncode == 0 and proc.stdout.strip():
                    for item in json.loads(proc.stdout):
                        ifname = item.get("ifname")
                        if not ifname:
                            continue
                        addresses = []
                        for addr in item.get("addr_info", []):
                            if addr.get("family") == "inet" and addr.get("local"):
                                addresses.append(addr["local"])
                        if addresses:
                            mapping[ifname] = addresses
            except (OSError, ValueError, subprocess.SubprocessError):
                mapping = {}

        primary: Optional[str] = None
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
                sock.settimeout(0.2)
                sock.connect(("1.1.1.1", 80))
                primary = sock.getsockname()[0]
        except OSError:
            primary = None

        default_iface = self._default_interface()
        if primary and default_iface and primary not in mapping.get(default_iface, []):
            mapping.setdefault(default_iface, []).insert(0, primary)

        all_ips = sorted({ip for values in mapping.values() for ip in values})
        if not primary and all_ips:
            primary = all_ips[0]
        return {"interfaces": mapping, "primary": primary, "all": all_ips}

    def _addresses(self, now: float) -> Dict[str, Any]:
        if now - self._addr_cache_ts >= 30.0 or not self._addr_cache_ts:
            self._addr_cache = self._read_addresses()
            self._addr_cache_ts = now
        return self._addr_cache

    # ------------------------------------------------------------------
    # CPU / memory / load
    # ------------------------------------------------------------------
    def _read_cpu(self, now: float) -> Dict[str, Any]:
        aggregate, cores = parse_cpu_stat(_read_text(self.proc_root / "stat"))
        if aggregate is None:
            return {
                "percent": 0.0,
                "per_core": [],
                "breakdown": {},
                "freq_mhz": None,
                "temp_c": None,
                "temp_label": None,
                "model": self._static["cpu_model"],
                "cores": self._static["cores"],
                "physical_cores": self._static["physical_cores"],
            }

        self._cpu_total_now = aggregate["total"]
        first = self._prev_cpu is None
        if first:
            used = aggregate["total"] - aggregate["idle_all"]
            total = aggregate["total"]
            percent = _pct(used, total)
            breakdown = {
                key: _pct(aggregate.get(key, 0), total)
                for key in ("user", "nice", "system", "iowait", "irq", "softirq", "steal")
            }
            per_core = [
                _pct(core["total"] - core["idle_all"], core["total"]) for core in cores
            ]
        else:
            prev = self._prev_cpu
            total_delta = max(0, aggregate["total"] - prev["total"])
            idle_delta = max(0, aggregate["idle_all"] - prev["idle_all"])
            percent = _pct(total_delta - idle_delta, total_delta)
            breakdown = {}
            for key in ("user", "nice", "system", "iowait", "irq", "softirq", "steal"):
                breakdown[key] = _pct(
                    max(0, aggregate.get(key, 0) - prev.get(key, 0)), total_delta
                )
            per_core = []
            prev_cores = {c["index"]: c for c in self._prev_cpu_cores}
            for core in cores:
                old = prev_cores.get(core["index"])
                if not old:
                    per_core.append(0.0)
                    continue
                dt = max(0, core["total"] - old["total"])
                di = max(0, core["idle_all"] - old["idle_all"])
                per_core.append(_pct(dt - di, dt))

        self._prev_cpu = aggregate
        self._prev_cpu_cores = cores
        meta = self._read_cpu_meta(now)
        return {
            "percent": _round(percent, 1),
            "per_core": [_round(v, 1) for v in per_core],
            "breakdown": {k: _round(v, 1) for k, v in breakdown.items()},
            "freq_mhz": meta["freq_mhz"],
            "temp_c": meta["temp_c"],
            "temp_label": meta["temp_label"],
            "model": self._static["cpu_model"],
            "cores": len(cores) or self._static["cores"],
            "physical_cores": self._static["physical_cores"],
        }

    def _read_cpu_meta(self, now: float) -> Dict[str, Any]:
        if now - self._cpu_meta_ts < 3.0 and self._cpu_meta_ts:
            return self._cpu_meta

        freqs: List[float] = []
        for path in glob.glob(str(self.sys_root / "devices/system/cpu/cpu*/cpufreq/scaling_cur_freq")):
            try:
                freqs.append(int(Path(path).read_text().strip()) / 1000.0)
            except (OSError, ValueError):
                continue
        if not freqs:
            for line in _read_text(self.proc_root / "cpuinfo").splitlines():
                if line.lower().startswith("cpu mhz"):
                    try:
                        freqs.append(float(line.split(":", 1)[1].strip()))
                    except (IndexError, ValueError):
                        pass
        freq = round(sum(freqs) / len(freqs)) if freqs else None

        zones: List[Tuple[float, str]] = []
        for path in glob.glob(str(self.sys_root / "class/thermal/thermal_zone*")):
            zone = Path(path)
            raw_text = _read_text(zone / "temp").strip()
            try:
                raw = float(raw_text)
            except ValueError:
                continue
            celsius = raw / 1000.0 if raw > 1000 else raw
            if not -20.0 <= celsius <= 150.0:
                continue
            label = _read_text(zone / "type").strip() or zone.name
            zones.append((celsius, label))
        if not zones:
            for path in glob.glob(str(self.sys_root / "class/hwmon/hwmon*/temp*_input")):
                pw = Path(path)
                name = _read_text(pw.parent / "name").strip()
                label = _read_text(pw.with_name(pw.stem.replace("_input", "_label"))).strip() or name
                if not any(token in f"{name} {label}".lower() for token in ("cpu", "package", "tctl", "tdie", "core")):
                    continue
                try:
                    raw = float(_read_text(pw).strip())
                except ValueError:
                    continue
                celsius = raw / 1000.0 if raw > 1000 else raw
                if -20.0 <= celsius <= 150.0:
                    zones.append((celsius, label))
        if zones:
            preferred = [z for z in zones if any(token in z[1].lower() for token in ("cpu", "package", "tctl", "tdie", "core"))]
            temp_c, temp_label = max(preferred or zones, key=lambda z: z[0])
            temp_c = round(temp_c, 1)
        else:
            temp_c, temp_label = None, None

        self._cpu_meta = {"freq_mhz": freq, "temp_c": temp_c, "temp_label": temp_label}
        self._cpu_meta_ts = now
        return self._cpu_meta

    def _read_memory(self) -> Dict[str, Any]:
        info = parse_meminfo(_read_text(self.proc_root / "meminfo"))
        total = info.get("total", 0)
        free = info.get("free", 0)
        available = info.get("available") or free
        buffers = info.get("buffers", 0)
        cached = info.get("cached", 0) + info.get("sreclaimable", 0)
        swap_total = info.get("swap_total", 0)
        swap_free = info.get("swap_free", 0)

        used = max(0, total - available) if total else 0
        percent = _pct(used, total)
        swap_used = max(0, swap_total - swap_free)
        swap_percent = _pct(swap_used, swap_total)

        apps = max(0, total - free - buffers - cached)
        cache_value = min(cached, max(0, total - free - buffers - apps))
        composition = {
            "apps": apps,
            "cache": max(0, cache_value),
            "buffers": buffers,
            "free": free,
        }
        comp_total = sum(composition.values())
        if comp_total > total and comp_total > 0:
            scale = total / comp_total
            composition = {k: int(v * scale) for k, v in composition.items()}

        return {
            "total": total,
            "used": used,
            "free": free,
            "available": available,
            "buffers": buffers,
            "cached": cached,
            "apps": apps,
            "percent": _round(percent, 1),
            "composition": composition,
            "swap_total": swap_total,
            "swap_used": swap_used,
            "swap_free": swap_free,
            "swap_percent": _round(swap_percent, 1),
            "dirty": info.get("dirty", 0),
            "writeback": info.get("writeback", 0),
        }

    def _read_load(self, cores: int) -> Dict[str, Any]:
        load = parse_loadavg(_read_text(self.proc_root / "loadavg"))
        load["percent"] = _round(_pct(load["load1"], max(1, cores)), 1)
        return load

    # ------------------------------------------------------------------
    # Disks
    # ------------------------------------------------------------------
    def _stat_disk(self, mount: Dict[str, str]) -> Optional[Dict[str, Any]]:
        mountpoint = mount["mount"]
        try:
            stat = os.statvfs(mountpoint)
        except OSError:
            return None
        frsize = stat.f_frsize or stat.f_bsize
        total = frsize * stat.f_blocks
        free = frsize * stat.f_bavail
        used = total - frsize * stat.f_bfree
        if total <= 0:
            return None
        if total < 1024 * 1024:
            return None
        percent = _pct(used, used + free)
        inodes_total = stat.f_files
        inodes_free = stat.f_ffree
        return {
            "device": mount["device"],
            "mount": mountpoint,
            "fstype": mount["fstype"],
            "total": int(total),
            "used": int(max(0, used)),
            "free": int(max(0, free)),
            "percent": _round(percent, 1),
            "readonly": "ro" in mount["options"].split(","),
            "inodes_total": int(inodes_total),
            "inodes_free": int(inodes_free),
        }

    def _read_disks(self, now: float) -> List[Dict[str, Any]]:
        if now - self._disk_cache_ts < 5.0 and self._disk_cache_ts:
            return self._disk_cache

        mounts = parse_mounts(_read_text(self.proc_root / "mounts"))
        disks: List[Dict[str, Any]] = []
        seen: set = set()
        for mount in mounts:
            fstype = mount["fstype"].lower()
            device = mount["device"]
            is_real = (
                fstype in REAL_FS_TYPES
                or device.startswith("/dev/")
                or (fstype.startswith("fuse.") and fstype not in ("fuse.gvfsd-fuse", "fuse.lxcfs"))
            )
            if not is_real:
                continue
            if mount["mount"] in ("/dev", "/proc", "/sys", "/run"):
                continue
            disk = self._stat_disk(mount)
            if not disk:
                continue
            key = (disk["device"], disk["total"], disk["used"])
            if key in seen:
                continue
            seen.add(key)
            disks.append(disk)

        disks.sort(key=lambda d: (0 if d["mount"] == "/" else 1, -d["percent"]))
        self._disk_cache = disks
        self._disk_cache_ts = now
        return disks

    def _is_partition(self, name: str) -> bool:
        try:
            if (self.sys_root / "class" / "block" / name / "partition").exists():
                return True
        except OSError:
            pass
        patterns = (
            r"^(sd|hd|vd|xvd)[a-z]+\d+$",
            r"^nvme\d+n\d+p\d+$",
            r"^mmcblk\d+p\d+$",
            r"^md\d+p\d+$",
        )
        return any(re.match(pattern, name) for pattern in patterns)

    def _read_disk_io(self, now: float) -> Dict[str, Any]:
        stats = parse_diskstats(_read_text(self.proc_root / "diskstats"))
        dt = (now - self._prev_disk_ts) if self._prev_disk_ts is not None else 0.0
        devices: List[Dict[str, Any]] = []
        new_prev: Dict[str, Dict[str, int]] = {}

        for name, current in stats.items():
            if name.startswith(("loop", "ram", "fd", "sr", "zram")):
                continue
            if self._is_partition(name):
                continue
            new_prev[name] = current
            old = self._prev_disk.get(name)
            if old and dt > 0:
                read_sectors = max(0, current["sectors_read"] - old["sectors_read"])
                write_sectors = max(0, current["sectors_written"] - old["sectors_written"])
                reads = max(0, current["reads"] - old["reads"])
                writes = max(0, current["writes"] - old["writes"])
                read_ms = max(0, current["read_ms"] - old["read_ms"])
                write_ms = max(0, current["write_ms"] - old["write_ms"])
                io_ms = max(0, current["io_ms"] - old["io_ms"])
                devices.append({
                    "device": name,
                    "read_bps": read_sectors * 512.0 / dt,
                    "write_bps": write_sectors * 512.0 / dt,
                    "read_iops": reads / dt,
                    "write_iops": writes / dt,
                    "util_percent": min(100.0, io_ms / (dt * 1000.0) * 100.0),
                    "read_await_ms": read_ms / reads if reads else 0.0,
                    "write_await_ms": write_ms / writes if writes else 0.0,
                })
            elif current["sectors_read"] or current["sectors_written"]:
                devices.append({
                    "device": name,
                    "read_bps": 0.0,
                    "write_bps": 0.0,
                    "read_iops": 0.0,
                    "write_iops": 0.0,
                    "util_percent": 0.0,
                    "read_await_ms": 0.0,
                    "write_await_ms": 0.0,
                })

        self._prev_disk = new_prev
        self._prev_disk_ts = now
        devices.sort(key=lambda d: d["read_bps"] + d["write_bps"], reverse=True)
        devices = devices[:8]
        for item in devices:
            item["read_bps"] = round(item["read_bps"])
            item["write_bps"] = round(item["write_bps"])
            item["read_iops"] = round(item["read_iops"], 1)
            item["write_iops"] = round(item["write_iops"], 1)
            item["util_percent"] = round(item["util_percent"], 1)
            item["read_await_ms"] = round(item["read_await_ms"], 2)
            item["write_await_ms"] = round(item["write_await_ms"], 2)
        return {
            "devices": devices,
            "read_bps": sum(d["read_bps"] for d in devices),
            "write_bps": sum(d["write_bps"] for d in devices),
        }

    # ------------------------------------------------------------------
    # Network
    # ------------------------------------------------------------------
    def _iface_speed(self, name: str) -> Optional[int]:
        try:
            raw = _read_text(self.sys_root / "class" / "net" / name / "speed").strip()
            value = int(raw)
            return value if value > 0 else None
        except (ValueError, OSError):
            return None

    def _read_network(self, now: float) -> Dict[str, Any]:
        stats = parse_net_dev(_read_text(self.proc_root / "net" / "dev"))
        addresses = self._addresses(now)
        dt = (now - self._prev_net_ts) if self._prev_net_ts is not None else 0.0
        interfaces: List[Dict[str, Any]] = []
        default_iface = self._default_interface()

        for name, current in stats.items():
            if name == "lo":
                continue
            old = self._prev_net.get(name)
            rx_bps = (current["rx_bytes"] - old["rx_bytes"]) / dt if old and dt > 0 else 0.0
            tx_bps = (current["tx_bytes"] - old["tx_bytes"]) / dt if old and dt > 0 else 0.0
            rx_bps = max(0.0, rx_bps)
            tx_bps = max(0.0, tx_bps)
            is_virtual = name.startswith(VIRTUAL_IFACE_PREFIXES)
            ips = addresses["interfaces"].get(name, [])
            if not ips and name == default_iface and addresses.get("primary"):
                ips = [addresses["primary"]]
            interfaces.append({
                "name": name,
                "ip": ips[0] if ips else "",
                "virtual": is_virtual,
                "primary": name == default_iface,
                "rx_bps": rx_bps,
                "tx_bps": tx_bps,
                "rx_total": current["rx_bytes"],
                "tx_total": current["tx_bytes"],
                "errors": current["rx_errors"] + current["tx_errors"],
                "drops": current["rx_drops"] + current["tx_drops"],
                "speed_mbps": self._iface_speed(name),
                "link": bool(ips) or rx_bps > 0 or tx_bps > 0,
            })

        physical = [i for i in interfaces if not i["virtual"]]
        aggregate_source = physical or interfaces
        total_rx = sum(i["rx_bps"] for i in aggregate_source)
        total_tx = sum(i["tx_bps"] for i in aggregate_source)

        for item in interfaces:
            item["rx_bps"] = round(item["rx_bps"])
            item["tx_bps"] = round(item["tx_bps"])
        interfaces.sort(key=lambda i: (not i["primary"], i["virtual"], -(i["rx_bps"] + i["tx_bps"])))
        self._prev_net = stats
        self._prev_net_ts = now

        return {
            "interfaces": interfaces[:8],
            "interface_count": len(interfaces),
            "total_rx_bps": round(total_rx),
            "total_tx_bps": round(total_tx),
            "primary_iface": default_iface,
            "primary_ip": addresses.get("primary"),
            "all_ips": addresses.get("all", []),
            "errors": sum(i["errors"] for i in interfaces),
            "drops": sum(i["drops"] for i in interfaces),
        }

    # ------------------------------------------------------------------
    # Processes
    # ------------------------------------------------------------------
    def _read_processes(self, now: float, mem_total: int) -> List[Dict[str, Any]]:
        if self._proc_cache_ts and now - self._proc_cache_ts < 2.0:
            return self._proc_cache

        pids: List[int] = []
        try:
            with os.scandir(self.proc_root) as entries:
                for entry in entries:
                    if entry.is_dir(follow_symlinks=False) and entry.name.isdigit():
                        pids.append(int(entry.name))
        except OSError:
            return self._proc_cache

        total_jiffies = self._cpu_total_now
        delta_total = max(0, total_jiffies - self._prev_proc_total)
        previous = self._proc_ticks
        new_ticks: Dict[int, int] = {}
        candidates: List[Dict[str, Any]] = []

        for pid in pids:
            text = _read_text(self.proc_root / str(pid) / "stat")
            match = _STAT_RE.match(text)
            if not match:
                continue
            try:
                rest = match.group(4).split()
                utime = int(rest[10])
                stime = int(rest[11])
                rss_pages = int(rest[20])
                threads = int(rest[16])
            except (IndexError, ValueError):
                continue
            ticks = utime + stime
            new_ticks[pid] = ticks
            old_ticks = previous.get(pid)
            cpu_percent = 0.0
            if old_ticks is not None and delta_total > 0:
                cpu_percent = max(0.0, (ticks - old_ticks) / delta_total * self._ncpu * 100.0)
            rss = max(0, rss_pages) * self._page_size
            candidates.append({
                "pid": pid,
                "name": match.group(2).strip(),
                "cmd": "",
                "cpu_percent": cpu_percent,
                "mem_percent": _pct(rss, mem_total),
                "rss_bytes": rss,
                "threads": threads,
            })

        self._proc_ticks = new_ticks
        self._prev_proc_total = total_jiffies

        by_cpu = sorted(candidates, key=lambda p: p["cpu_percent"], reverse=True)[:8]
        by_mem = sorted(candidates, key=lambda p: p["rss_bytes"], reverse=True)[:8]
        selected: List[Dict[str, Any]] = []
        seen = set()
        for item in by_cpu + by_mem:
            if item["pid"] in seen:
                continue
            seen.add(item["pid"])
            selected.append(item)
            if len(selected) >= 12:
                break

        for item in selected:
            raw = _read_text(self.proc_root / str(item["pid"]) / "cmdline")
            cmd = " ".join(part for part in raw.split("\x00") if part).strip()
            if not cmd:
                cmd = item["name"]
            item["cmd"] = cmd[:90]
            item["cpu_percent"] = round(item["cpu_percent"], 1)
            item["mem_percent"] = round(item["mem_percent"], 2)

        selected.sort(key=lambda p: p["cpu_percent"], reverse=True)
        self._proc_cache = selected
        self._proc_cache_ts = now
        return selected

    # ------------------------------------------------------------------
    # Snapshot
    # ------------------------------------------------------------------
    def snapshot(self) -> Dict[str, Any]:
        now = self.clock()
        wall = self.wall_clock()
        cpu = self._read_cpu(now)
        memory = self._read_memory()
        load = self._read_load(cpu["cores"] or self._static["cores"])
        uptime = parse_uptime(_read_text(self.proc_root / "uptime"))
        disks = self._read_disks(now)
        disk_io = self._read_disk_io(now)
        network = self._read_network(now)
        addresses = self._addresses(now)
        processes = self._read_processes(now, memory["total"])

        alerts = build_alerts(
            cpu_percent=float(cpu["percent"]),
            memory_percent=float(memory["percent"]),
            swap_percent=float(memory["swap_percent"]),
            load1=float(load["load1"]),
            cores=int(cpu["cores"] or self._static["cores"]),
            disks=disks,
            temp_c=cpu["temp_c"],
        )

        host = dict(self._static)
        local_now = dt.datetime.now().astimezone()
        local_offset = local_now.utcoffset()
        host.update({
            "tz_offset": int(local_offset.total_seconds()) if local_offset else host["tz_offset"],
            "tz_name": local_now.tzname() or host["tz_name"],
            "boot_epoch": wall - uptime,
            "uptime_seconds": round(uptime, 1),
            "primary_ip": addresses.get("primary"),
            "ips": addresses.get("all", []),
            "process_count": len(self._proc_ticks),
            "thread_count": load.get("threads", 0),
        })

        return {
            "ts": wall,
            "version": __version__,
            "host": host,
            "cpu": cpu,
            "memory": memory,
            "load": load,
            "disks": disks,
            "disk_io": disk_io,
            "network": network,
            "processes": processes,
            "alerts": alerts,
        }