mdns-termux/lib/mdns_tools/service.py

"""Long-running publisher: singleton PID lock, clean SIGTERM/SIGINT goodbye,
and automatic re-announce when the device's IP changes (WiFi<->data, roaming).
"""
import json
import os
import signal
import time
from pathlib import Path

from zeroconf import Zeroconf

from . import publish as _p

PID_FILE = Path.home() / ".cache" / "mdns" / "publisher.pid"
_STOP = False


def _read_pid() -> dict | None:
    if not PID_FILE.exists():
        return None
    try:
        d = json.loads(PID_FILE.read_text())
    except Exception:
        return None
    try:
        os.kill(int(d["pid"]), 0)  # probe liveness
    except (OSError, ValueError, KeyError):
        return None
    return d


def _write_pid(host: str, ip: str, n: int) -> None:
    PID_FILE.parent.mkdir(parents=True, exist_ok=True)
    PID_FILE.write_text(json.dumps({
        "pid": os.getpid(), "host": host, "ip": ip, "services": n,
        "since": time.strftime("%Y-%m-%d %H:%M:%S"),
    }))


def status() -> int:
    d = _read_pid()
    if not d:
        print("publisher: stopped")
        return 1
    print(f"publisher: running (pid {d['pid']})\n"
          f"  host:     {d.get('host')}.local\n"
          f"  ip:       {d.get('ip')}\n"
          f"  services: {d.get('services', 0)}\n"
          f"  since:    {d.get('since')}")
    return 0


def _handle_stop(signum, frame) -> None:
    global _STOP
    _STOP = True


def run_daemon(config_path: str | None = None, hostname: str | None = None,
               interval: int = 15, verbose: bool = True) -> int:
    existing = _read_pid()
    if existing:
        print(f"error: publisher already running (pid {existing['pid']}). "
              f"Use `mdns stop` first.")
        return 1
    signal.signal(signal.SIGTERM, _handle_stop)
    signal.signal(signal.SIGINT, _handle_stop)

    cfg = _p.load_config(config_path)
    host = hostname or cfg.get("hostname", "movil")
    ip = _p.local_ipv4()
    zc = Zeroconf()
    infos = _p.build_infos(cfg, host, ip)
    _write_pid(host, ip, len(infos))  # mark alive before the slow probing register
    _p.register_all(zc, infos)
    if verbose:
        print(f"[mdns] daemon publishing {host}.local ({ip}), {len(infos)} record(s)")
    try:
        elapsed = 0
        while not _STOP:
            time.sleep(1)  # short tick so SIGTERM is honored within ~1s
            elapsed += 1
            if elapsed < interval:
                continue
            elapsed = 0
            new_ip = _p.local_ipv4()
            if new_ip != ip and new_ip != "127.0.0.1":
                if verbose:
                    print(f"[mdns] IP changed {ip} -> {new_ip}, re-announcing")
                _p.unregister_all(zc, infos)
                ip = new_ip
                infos = _p.build_infos(cfg, host, ip)
                _p.register_all(zc, infos)
                _write_pid(host, ip, len(infos))
    finally:
        _p.unregister_all(zc, infos)
        zc.close()
        PID_FILE.unlink(missing_ok=True)
        if verbose:
            print("[mdns] daemon stopped, goodbye packets sent.")
    return 0


def stop() -> int:
    d = _read_pid()
    if not d:
        print("publisher: not running")
        return 1
    os.kill(int(d["pid"]), signal.SIGTERM)
    print(f"sent stop to pid {d['pid']}")
    return 0