"""Discover services on this LAN and verify scope_id preservation on link-local IPv6."""

from __future__ import annotations

import asyncio
import socket
from ipaddress import IPv6Address

from zeroconf import IPVersion, ServiceInfo
from zeroconf.asyncio import (
    AsyncServiceBrowser,
    AsyncServiceInfo,
    AsyncZeroconf,
    AsyncZeroconfServiceTypes,
)


async def main() -> None:
    aiozc = AsyncZeroconf(ip_version=IPVersion.All)

    print("discovering service types for 5s...")
    types = list(await AsyncZeroconfServiceTypes.async_find(aiozc=aiozc, timeout=5))
    print(f"  found {len(types)} type(s): {types}")

    seen: dict[str, AsyncServiceInfo] = {}

    def on_service_state_change(zeroconf, service_type, name, state_change):
        if state_change.name != "Added":
            return
        seen[name] = AsyncServiceInfo(service_type, name)

    if not types:
        print("no service types found on this network")
        await aiozc.async_close()
        return

    browser = AsyncServiceBrowser(
        aiozc.zeroconf, types, handlers=[on_service_state_change]
    )
    await asyncio.sleep(6)
    await browser.async_cancel()

    print(f"\nresolving {len(seen)} service(s)...")
    link_local_with_scope = 0
    link_local_without_scope = 0
    dup_check_ok = 0
    dup_check_bad = 0

    for name, info in seen.items():
        ok = await info.async_request(aiozc.zeroconf, 3000)
        if not ok:
            continue
        scoped = info.parsed_scoped_addresses()
        v6 = info.ip_addresses_by_version(IPVersion.V6Only)
        v4 = info.ip_addresses_by_version(IPVersion.V4Only)

        ll_with_scope = []
        ll_no_scope = []
        for addr in scoped:
            if addr.startswith("fe80:") or "%" in addr:
                if "%" in addr:
                    ll_with_scope.append(addr)
                else:
                    ll_no_scope.append(addr)

        # Verify no packed-byte duplicates among v6 addresses.
        packed_seen = set()
        dup_in_packed = False
        for a in v6:
            key = int(IPv6Address(str(a).split("%", 1)[0]))
            if key in packed_seen:
                dup_in_packed = True
                break
            packed_seen.add(key)
        if dup_in_packed:
            dup_check_bad += 1
        else:
            dup_check_ok += 1

        link_local_with_scope += len(ll_with_scope)
        link_local_without_scope += len(ll_no_scope)

        marker = ""
        if ll_with_scope:
            marker = "  <-- scoped link-local"
        elif ll_no_scope:
            marker = "  <-- unscoped link-local (would be a regression)"

        print(f"  {name}{marker}")
        print(f"    v4: {[str(a) for a in v4]}")
        print(f"    v6: {[str(a) for a in v6]}")
        print(f"    parsed_scoped_addresses: {scoped}")

    print()
    print("summary:")
    print(f"  resolved: {len(seen)}")
    print(f"  link-local IPv6 with %scope:  {link_local_with_scope}")
    print(f"  link-local IPv6 without %scope: {link_local_without_scope}")
    print(f"  services with no packed-byte duplicates: {dup_check_ok}")
    print(f"  services with packed-byte duplicates (BAD): {dup_check_bad}")

    await aiozc.async_close()


if __name__ == "__main__":
    asyncio.run(main())

"""Pinpoint services where master returns a packed-byte duplicate IPv6 address."""

from __future__ import annotations

import asyncio
from ipaddress import IPv6Address

from zeroconf import IPVersion
from zeroconf.asyncio import (
    AsyncServiceBrowser,
    AsyncServiceInfo,
    AsyncZeroconf,
    AsyncZeroconfServiceTypes,
)


async def main() -> None:
    aiozc = AsyncZeroconf(ip_version=IPVersion.All)
    types = list(await AsyncZeroconfServiceTypes.async_find(aiozc=aiozc, timeout=5))
    seen: dict[str, AsyncServiceInfo] = {}

    def on_state(zeroconf, service_type, name, state_change):
        if state_change.name == "Added":
            seen[name] = AsyncServiceInfo(service_type, name)

    browser = AsyncServiceBrowser(aiozc.zeroconf, types, handlers=[on_state])
    await asyncio.sleep(8)
    await browser.async_cancel()

    print(f"scanning {len(seen)} services for v6 packed-byte duplicates...")
    for name, info in seen.items():
        if not await info.async_request(aiozc.zeroconf, 3000):
            continue
        v6 = info.ip_addresses_by_version(IPVersion.V6Only)
        packed = [int(IPv6Address(str(a).split("%", 1)[0])) for a in v6]
        if len(packed) != len(set(packed)):
            print(f"\nDUPLICATE in {name}")
            print(f"  raw v6: {[str(a) for a in v6]}")
            print(f"  parsed_scoped: {info.parsed_scoped_addresses()}")
    await aiozc.async_close()


if __name__ == "__main__":
    asyncio.run(main())