mdns-termux/lib/mdns_tools/publish.py

"""Publish on the LAN: services (JSON config) or a bare hostname->IP.
Public helpers reused by service.py: local_ipv4, load_config, build_infos,
register_all, unregister_all."""
import json
import socket
import time
from pathlib import Path

from zeroconf import Zeroconf, ServiceInfo, NonUniqueNameException

DEFAULT_CONFIG = Path.home() / ".config" / "mdns" / "services.json"


def local_ipv4() -> str:
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        s.connect(("8.8.8.8", 80))
        return s.getsockname()[0]
    except OSError:
        return "127.0.0.1"
    finally:
        s.close()


def _addr_bytes(ip: str) -> bytes:
    if ":" in ip:
        return socket.inet_pton(socket.AF_INET6, ip)
    return socket.inet_aton(ip)


def load_config(path: str | None) -> dict:
    p = Path(path).expanduser() if path else DEFAULT_CONFIG
    if not p.exists():
        return {"hostname": "movil", "services": []}
    try:
        return json.loads(p.read_text())
    except Exception as e:
        print(f"warning: bad config {p}: {e}")
        return {"hostname": "movil", "services": []}


def _workstation(host: str, ip: str) -> ServiceInfo:
    return ServiceInfo(
        type_="_workstation._tcp.local.",
        name=f"{host} [{ip.replace('.', '-')}]._workstation._tcp.local.",
        addresses=[_addr_bytes(ip)], port=9, server=f"{host}.local.",
    )


def build_infos(cfg: dict, host: str, ip: str) -> list[ServiceInfo]:
    infos = [_workstation(host, ip)]
    for svc in (cfg.get("services") or []):
        try:
            t = svc["type"].rstrip(".") + ".local."
            infos.append(ServiceInfo(
                type_=t, name=f'{svc.get("name", host)}.{t}',
                addresses=[_addr_bytes(ip)], port=int(svc["port"]),
                server=f"{host}.local.", properties=svc.get("txt", {}),
            ))
        except Exception as e:
            print(f"warning: could not register {svc}: {e}")
    return infos


def register_all(zc: Zeroconf, infos: list[ServiceInfo]) -> None:
    for i in infos:
        try:  # tolerate name conflicts per RFC 6762 (probe + rename)
            zc.register_service(i)
        except NonUniqueNameException:
            zc.register_service(i, allow_name_change=True)


def unregister_all(zc: Zeroconf, infos: list[ServiceInfo]) -> None:
    for i in infos:
        try:
            zc.unregister_service(i)
        except Exception:
            pass


def _serve(infos: list[ServiceInfo], verbose: bool, label: str) -> None:
    zc = Zeroconf()
    register_all(zc, infos)
    if verbose:
        print(f"[mdns] {label} ({len(infos)} record(s)). Ctrl+C to stop.")
    try:
        while True:
            time.sleep(60)
    except KeyboardInterrupt:
        pass
    finally:
        unregister_all(zc, infos)
        zc.close()
        if verbose:
            print("[mdns] stopped, goodbye packets sent.")


def run(config_path: str | None = None, hostname: str | None = None,
        verbose: bool = True) -> None:
    cfg = load_config(config_path)
    host = hostname or cfg.get("hostname", "movil")
    ip = local_ipv4()
    _serve(build_infos(cfg, host, ip), verbose, f"published {host}.local ({ip})")


def publish_address(name: str, ip: str, verbose: bool = True) -> None:
    """Publish a bare hostname -> IP (like `avahi-publish -a`)."""
    host = name.rstrip(".").removesuffix(".local")
    _serve([_workstation(host, ip)], verbose, f"published {host}.local -> {ip}")