Files
tianxuan/analysis/ip_clustering.py
T

92 lines
2.4 KiB
Python

"""IP address utilities for subnet clustering and conversion.
Provides helper functions to convert IPv4 strings to integers, compute
subnet prefixes, and group IP lists by subnet.
Typical usage::
from analysis.ip_clustering import ip_to_int, get_subnet, cluster_ips
ip_int = ip_to_int('192.168.1.1') # → 3232235777
subnet = get_subnet('192.168.1.5', 24) # → '192.168.1.0/24'
groups = cluster_ips(ip_list, mask=24) # → {'192.168.1.0/24': [...]}
"""
from __future__ import annotations
import ipaddress
from collections import defaultdict
def ip_to_int(ip_str: str) -> int | None:
"""Convert an IPv4 string to its 32-bit integer representation.
Parameters
----------
ip_str:
An IPv4 address string, e.g. ``'192.168.1.1'``.
Returns
-------
int | None
The integer value of the address, or ``None`` if the input
is not a valid IPv4 address.
"""
try:
return int(ipaddress.IPv4Address(ip_str))
except (ipaddress.AddressValueError, ValueError):
return None
def get_subnet(ip_str: str, mask: int = 24) -> str | None:
"""Compute the CIDR subnet string for an IP and mask length.
Parameters
----------
ip_str:
An IPv4 address string, e.g. ``'192.168.1.5'``.
mask:
Subnet mask length (default ``24``).
Returns
-------
str | None
Subnet string like ``'192.168.1.0/24'``, or ``None`` if the IP
is not valid.
"""
try:
ip = ipaddress.IPv4Address(ip_str)
net = ipaddress.IPv4Network(f'{ip}/{mask}', strict=False)
return str(net)
except (ipaddress.AddressValueError, ValueError):
return None
def cluster_ips(
ip_list: list[str],
mask: int = 24,
) -> dict[str, list[str]]:
"""Group a list of IPv4 addresses by their subnet.
Parameters
----------
ip_list:
List of IPv4 address strings.
mask:
Subnet mask length (default ``24``).
Returns
-------
dict[str, list[str]]
Mapping of ``subnet`` → ``[ip, ip, ...]``. Invalid IPs are
grouped under the key ``'__invalid__'``.
"""
groups: dict[str, list[str]] = defaultdict(list)
for ip_str in ip_list:
subnet = get_subnet(ip_str, mask)
if subnet is not None:
groups[subnet].append(ip_str)
else:
groups['__invalid__'].append(ip_str)
# Convert to plain dict for stable serialisation
return dict(groups)