socket_setup.py

# socket_setup.py
import socket

from utils.network_tools import get_default_interface_ipv4


def discover_local_ipv4_candidates():
    """Return extra local IPv4 candidates that may still be bindable after route churn."""
    candidates = []
    try:
        _host_name, _aliases, host_ips = socket.gethostbyname_ex(socket.gethostname())
    except OSError:
        host_ips = []

    for candidate in host_ips:
        normalized_candidate = (candidate or "").strip()
        if not normalized_candidate or normalized_candidate.startswith("127."):
            continue
        if normalized_candidate not in candidates:
            candidates.append(normalized_candidate)
    return candidates


def get_bind_candidates(preferred_ip, destination_ip):
    """Return the ordered list of local IPv4 candidates for an outbound socket bind."""
    candidates = []
    for candidate in [
        preferred_ip,
        get_default_interface_ipv4(destination_ip),
        get_default_interface_ipv4("1.1.1.1"),
        get_default_interface_ipv4("8.8.8.8"),
        *discover_local_ipv4_candidates(),
    ]:
        if candidate and candidate not in candidates:
            candidates.append(candidate)
    return candidates


def bind_socket_to_available_interface(sock, preferred_ip, destination_ip):
    """Bind to the first usable local IPv4 and report whether a fallback was needed."""
    last_error = None
    candidates = get_bind_candidates(preferred_ip, destination_ip)

    for candidate in candidates:
        try:
            sock.bind((candidate, 0))
            warning_message = None
            if preferred_ip and candidate != preferred_ip:
                warning_message = (
                    f"آدرس شبکه محلی {preferred_ip} دیگر روی این سیستم قابل استفاده نبود؛ "
                    f"برنامه از آدرس {candidate} برای ادامه کار استفاده می‌کند."
                )
            return candidate, warning_message
        except OSError as error:
            last_error = error
            if getattr(error, "winerror", None) != 10049:
                raise

    if last_error is not None:
        raise last_error

    raise OSError(10049, "No usable local IPv4 address was found for socket bind")


def bind_socket_with_refresh(sock, preferred_ip, destination_ip):
    """Retry socket bind after clearing a stale preferred interface hint."""
    normalized_preferred_ip = (preferred_ip or "").strip() or None
    try:
        return bind_socket_to_available_interface(sock, normalized_preferred_ip, destination_ip)
    except OSError as error:
        if getattr(error, "winerror", None) != 10049 or not normalized_preferred_ip:
            raise

    bound_ip, warning_message = bind_socket_to_available_interface(sock, None, destination_ip)
    recovery_message = (
        f"The cached local interface {normalized_preferred_ip} was cleared after a bind failure; "
        f"the app recovered with {bound_ip}."
    )
    if warning_message:
        return bound_ip, f"{warning_message} {recovery_message}"
    return bound_ip, recovery_message