"""
ACS CWMP minimalista (TR-069) — só biblioteca padrão do Python.

Objetivo didático:
  Mostrar a conversa real CPE ↔ ACS na porta 7547:
  Inform → InformResponse → (opcional) GetParameterValues → fim de sessão,
  e Connection Request para “acordar” o roteador depois.

Não é GenieACS. Não tem Mongo, UI nem provision. É o esqueleto
para o aluno enxergar o protocolo sem ruído.
"""

from __future__ import annotations

import argparse
import base64
import hashlib
import logging
import re
import socket
import sys
import threading
import time
import urllib.error
import urllib.request
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any
from urllib.parse import urlparse

# ---------------------------------------------------------------------------
# Namespaces CWMP / SOAP (os CPE misturam 1-0, 1-1, 1-2 — tratamos todos)
# ---------------------------------------------------------------------------
NS_SOAP = "http://schemas.xmlsoap.org/soap/envelope/"
NS_CWMP_CANDIDATES = (
    "urn:dslforum-org:cwmp-1-0",
    "urn:dslforum-org:cwmp-1-1",
    "urn:dslforum-org:cwmp-1-2",
)

log = logging.getLogger("acs")


# =============================================================================
# 1) Memória do ACS — o que guardamos de cada Inform
# =============================================================================


@dataclass
class Device:
    """Snapshot do que o CPE contou no Inform (+ o que aprendemos depois)."""

    serial: str
    manufacturer: str = ""
    oui: str = ""
    product_class: str = ""
    events: list[str] = field(default_factory=list)
    params: dict[str, str] = field(default_factory=dict)
    connection_request_url: str = ""
    last_inform_at: float = 0.0
    last_peer: str = ""
    # Fila de RPCs ACS→CPE para a *próxima* conversa CWMP
    pending_rpcs: list[str] = field(default_factory=list)

    @property
    def software_version(self) -> str:
        for key in (
            "InternetGatewayDevice.DeviceInfo.SoftwareVersion",
            "Device.DeviceInfo.SoftwareVersion",
        ):
            if key in self.params:
                return self.params[key]
        return ""


class DeviceStore:
    """Registro em memória (dict por serial). Some se o processo cair — de propósito."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._by_serial: dict[str, Device] = {}

    def upsert_from_inform(
        self,
        *,
        serial: str,
        manufacturer: str,
        oui: str,
        product_class: str,
        events: list[str],
        params: dict[str, str],
        peer: str,
    ) -> Device:
        with self._lock:
            dev = self._by_serial.get(serial) or Device(serial=serial)
            dev.manufacturer = manufacturer or dev.manufacturer
            dev.oui = oui or dev.oui
            dev.product_class = product_class or dev.product_class
            dev.events = events
            dev.params.update(params)
            cr = (
                params.get("InternetGatewayDevice.ManagementServer.ConnectionRequestURL")
                or params.get("Device.ManagementServer.ConnectionRequestURL")
                or ""
            )
            if cr:
                dev.connection_request_url = cr
            dev.last_inform_at = time.time()
            dev.last_peer = peer
            self._by_serial[serial] = dev
            return dev

    def get(self, serial: str) -> Device | None:
        with self._lock:
            return self._by_serial.get(serial)

    def list(self) -> list[Device]:
        with self._lock:
            return list(self._by_serial.values())

    def enqueue_rpc(self, serial: str, soap_body_inner: str) -> bool:
        with self._lock:
            dev = self._by_serial.get(serial)
            if not dev:
                return False
            dev.pending_rpcs.append(soap_body_inner)
            return True

    def pop_rpc(self, serial: str) -> str | None:
        with self._lock:
            dev = self._by_serial.get(serial)
            if not dev or not dev.pending_rpcs:
                return None
            return dev.pending_rpcs.pop(0)


STORE = DeviceStore()

# serial CWMP associado à conexão TCP (keep-alive). Em 3.14 o socket
# não aceita atributo dinâmico — usamos id(connection).
SESSION_SERIAL: dict[int, str] = {}


# =============================================================================
# 2) SOAP — ler o que o CPE manda / montar o que o ACS responde
# =============================================================================


def _local(tag: str) -> str:
    """'{ns}Inform' → 'Inform'."""
    if "}" in tag:
        return tag.rsplit("}", 1)[-1]
    return tag


def _find_cwmp_ns(root: ET.Element) -> str:
    for el in root.iter():
        if el.tag.startswith("{"):
            ns = el.tag[1:].split("}", 1)[0]
            if ns.startswith("urn:dslforum-org:cwmp"):
                return ns
    return NS_CWMP_CANDIDATES[0]


def _text(el: ET.Element | None) -> str:
    if el is None or el.text is None:
        return ""
    return el.text.strip()


def parse_soap(xml_bytes: bytes) -> dict[str, Any]:
    """
    Extrai o método CWMP do Body e campos úteis do Inform.

    Retorno típico:
      {"method": "Inform", "cwmp_id": "...", "cwmp_ns": "...", ...}
    """
    if not xml_bytes.strip():
        return {"method": None, "empty": True}

    # Alguns CPE mandam encoding declarado estranho; ElementTree tolera utf-8.
    root = ET.fromstring(xml_bytes)
    cwmp_ns = _find_cwmp_ns(root)

    cwmp_id = ""
    for el in root.iter():
        if _local(el.tag) == "ID":
            cwmp_id = _text(el)
            break

    body = None
    for el in root.iter():
        if _local(el.tag) == "Body":
            body = el
            break
    if body is None:
        return {"method": None, "error": "SOAP Body ausente", "cwmp_id": cwmp_id}

    method_el = None
    for child in list(body):
        if _local(child.tag) not in ("Fault",):
            method_el = child
            break
        method_el = child
        break

    if method_el is None:
        return {"method": None, "empty": True, "cwmp_id": cwmp_id, "cwmp_ns": cwmp_ns}

    method = _local(method_el.tag)
    out: dict[str, Any] = {
        "method": method,
        "cwmp_id": cwmp_id,
        "cwmp_ns": cwmp_ns,
        "empty": False,
    }

    if method == "Inform":
        out.update(_parse_inform(method_el))
    elif method in ("GetParameterValuesResponse", "GetRPCMethodsResponse"):
        out["raw_method_xml"] = ET.tostring(method_el, encoding="unicode")
        if method == "GetParameterValuesResponse":
            out["params"] = _parse_parameter_value_list(method_el)

    return out


def _parse_inform(inform_el: ET.Element) -> dict[str, Any]:
    device_id = {}
    events: list[str] = []
    params: dict[str, str] = {}

    for el in inform_el:
        name = _local(el.tag)
        if name == "DeviceId":
            for child in el:
                device_id[_local(child.tag)] = _text(child)
        elif name == "Event":
            for ev in el:
                if _local(ev.tag) == "EventStruct":
                    code = ""
                    for c in ev:
                        if _local(c.tag) == "EventCode":
                            code = _text(c)
                    if code:
                        events.append(code)
        elif name == "ParameterList":
            params = _parse_parameter_value_list(el)

    return {
        "device_id": device_id,
        "events": events,
        "params": params,
        "serial": device_id.get("SerialNumber", ""),
        "manufacturer": device_id.get("Manufacturer", ""),
        "oui": device_id.get("OUI", ""),
        "product_class": device_id.get("ProductClass", ""),
    }


def _parse_parameter_value_list(parent: ET.Element) -> dict[str, str]:
    params: dict[str, str] = {}
    for el in parent.iter():
        if _local(el.tag) != "ParameterValueStruct":
            continue
        pname, pval = "", ""
        for c in el:
            if _local(c.tag) == "Name":
                pname = _text(c)
            elif _local(c.tag) == "Value":
                pval = _text(c)
        if pname:
            params[pname] = pval
    return params


def build_envelope(*, cwmp_ns: str, cwmp_id: str, body_inner_xml: str) -> bytes:
    """Monta Envelope SOAP com Header/ID + Body (miolo CWMP já em XML)."""
    # Escapar só o ID; o body_inner já é XML controlado por nós.
    cid = (
        cwmp_id.replace("&", "&")
        .replace("<", "&lt;")
        .replace(">", "&gt;")
        .replace('"', "&quot;")
    )
    xml = f"""<?xml version="1.0" encoding="UTF-8"?>
<soap-env:Envelope
  xmlns:soap-env="{NS_SOAP}"
  xmlns:soap-enc="http://schemas.xmlsoap.org/soap/encoding/"
  xmlns:xsd="http://www.w3.org/2001/XMLSchema"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xmlns:cwmp="{cwmp_ns}">
  <soap-env:Header>
    <cwmp:ID soap-env:mustUnderstand="1">{cid}</cwmp:ID>
  </soap-env:Header>
  <soap-env:Body>
    {body_inner_xml}
  </soap-env:Body>
</soap-env:Envelope>
"""
    return xml.encode("utf-8")


def soap_inform_response(cwmp_ns: str, cwmp_id: str) -> bytes:
    inner = """<cwmp:InformResponse>
      <MaxEnvelopes>1</MaxEnvelopes>
    </cwmp:InformResponse>"""
    return build_envelope(cwmp_ns=cwmp_ns, cwmp_id=cwmp_id or "1", body_inner_xml=inner)


def soap_get_rpc_methods_response(cwmp_ns: str, cwmp_id: str) -> bytes:
    # Lista enxuta — o suficiente para o CPE achar o ACS “vivo”.
    methods = [
        "GetRPCMethods",
        "Inform",
        "GetParameterNames",
        "GetParameterValues",
        "SetParameterValues",
        "Reboot",
    ]
    items = "\n".join(f"        <string>{m}</string>" for m in methods)
    inner = f"""<cwmp:GetRPCMethodsResponse>
      <MethodList soap-enc:arrayType="xsd:string[{len(methods)}]">
{items}
      </MethodList>
    </cwmp:GetRPCMethodsResponse>"""
    return build_envelope(cwmp_ns=cwmp_ns, cwmp_id=cwmp_id or "1", body_inner_xml=inner)


def soap_get_parameter_values(cwmp_ns: str, cwmp_id: str, names: list[str]) -> bytes:
    items = "\n".join(f"        <string>{n}</string>" for n in names)
    inner = f"""<cwmp:GetParameterValues>
      <ParameterNames soap-enc:arrayType="xsd:string[{len(names)}]">
{items}
      </ParameterNames>
    </cwmp:GetParameterValues>"""
    return build_envelope(cwmp_ns=cwmp_ns, cwmp_id=cwmp_id or "1", body_inner_xml=inner)


# =============================================================================
# 3) Connection Request — ACS “bate na porta” do CPE
# =============================================================================


def connection_request(
    url: str,
    *,
    username: str = "",
    password: str = "",
    timeout: float = 8.0,
) -> tuple[bool, str]:
    """
    HTTP GET (ou empty POST) no ConnectionRequestURL.

    ZTE/Huawei costumam exigir Digest. Tentamos Digest se vier 401;
    se o user/pass estiverem vazios, ainda assim tentamos o GET limpo
    (alguns labs abrem CR sem auth).
    """
    if not url:
        return False, "ConnectionRequestURL vazio (CPE ainda não informou)"

    req = urllib.request.Request(url, method="GET")
    req.add_header("User-Agent", "mauro-acs-python/0.1")

    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return True, f"HTTP {resp.status} (sem auth)"
    except urllib.error.HTTPError as e:
        if e.code != 401:
            return False, f"HTTP {e.code}: {e.reason}"
        www = e.headers.get("WWW-Authenticate", "")
        if "digest" in www.lower() and username:
            return _digest_get(url, www, username, password, timeout)
        if username:
            # Fallback Basic
            token = base64.b64encode(f"{username}:{password}".encode()).decode()
            req2 = urllib.request.Request(url, method="GET")
            req2.add_header("Authorization", f"Basic {token}")
            try:
                with urllib.request.urlopen(req2, timeout=timeout) as resp:
                    return True, f"HTTP {resp.status} (Basic)"
            except Exception as ex:  # noqa: BLE001
                return False, f"Basic falhou: {ex}"
        return False, f"401 e sem credencial CR configurada. WWW-Authenticate={www!r}"
    except Exception as ex:  # noqa: BLE001
        return False, str(ex)


def _digest_get(
    url: str,
    www_authenticate: str,
    username: str,
    password: str,
    timeout: float,
) -> tuple[bool, str]:
    # Parser bem enxuto de Digest challenge
    realm = _auth_param(www_authenticate, "realm") or ""
    nonce = _auth_param(www_authenticate, "nonce") or ""
    qop = _auth_param(www_authenticate, "qop") or ""
    opaque = _auth_param(www_authenticate, "opaque")
    algorithm = (_auth_param(www_authenticate, "algorithm") or "MD5").upper()

    path = urlparse(url).path or "/"
    if urlparse(url).query:
        path = path + "?" + urlparse(url).query

    ha1 = hashlib.md5(f"{username}:{realm}:{password}".encode()).hexdigest()
    ha2 = hashlib.md5(f"GET:{path}".encode()).hexdigest()
    nc = "00000001"
    cnonce = hashlib.md5(str(time.time()).encode()).hexdigest()[:16]

    if qop:
        response = hashlib.md5(
            f"{ha1}:{nonce}:{nc}:{cnonce}:{qop}:{ha2}".encode()
        ).hexdigest()
    else:
        response = hashlib.md5(f"{ha1}:{nonce}:{ha2}".encode()).hexdigest()

    parts = [
        f'username="{username}"',
        f'realm="{realm}"',
        f'nonce="{nonce}"',
        f'uri="{path}"',
        f'response="{response}"',
        f'algorithm={algorithm}',
    ]
    if opaque:
        parts.append(f'opaque="{opaque}"')
    if qop:
        parts.append(f'qop={qop}')
        parts.append(f'nc={nc}')
        parts.append(f'cnonce="{cnonce}"')

    req = urllib.request.Request(url, method="GET")
    req.add_header("Authorization", "Digest " + ", ".join(parts))
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return True, f"HTTP {resp.status} (Digest)"
    except Exception as ex:  # noqa: BLE001
        return False, f"Digest falhou: {ex}"


def _auth_param(header: str, name: str) -> str | None:
    m = re.search(rf'{name}="([^"]*)"', header, flags=re.I)
    if m:
        return m.group(1)
    m = re.search(rf"{name}=([^,\s]+)", header, flags=re.I)
    return m.group(1) if m else None


# =============================================================================
# 4) HTTP server — a porta 7547
# =============================================================================


class AcsHandler(BaseHTTPRequestHandler):
    """
    Cada POST do CPE é um passo da sessão CWMP.

    Fluxo típico ZTE/Genie:
      1) POST Inform          → 200 InformResponse
      2) POST (vazio)         → 200 com GetParameterValues (se houver fila) ou 204
      3) POST GPVResponse     → 204 (fim)
    Connection Request fora da sessão faz o CPE recomeçar no passo 1.
    """

    server_version = "MauroACS/0.1"

    def log_message(self, fmt: str, *args: Any) -> None:
        log.info("HTTP " + fmt, *args)

    def do_GET(self) -> None:  # noqa: N802
        # Alguns testes batem GET; respondemos health simples.
        if self.path in ("/", "/health"):
            body = b"ACS CWMP up\n"
            self.send_response(200)
            self.send_header("Content-Type", "text/plain; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        self.send_error(404)

    def do_POST(self) -> None:  # noqa: N802
        length = int(self.headers.get("Content-Length", "0") or 0)
        raw = self.rfile.read(length) if length > 0 else b""
        peer = self.client_address[0]

        log.info("=" * 60)
        log.info("POST de %s  path=%s  bytes=%s", peer, self.path, length)
        if raw:
            preview = raw[:800].decode("utf-8", errors="replace")
            log.info("SOAP recebido (preview):\n%s", preview)

        try:
            parsed = parse_soap(raw) if raw else {"method": None, "empty": True}
        except ET.ParseError as e:
            log.error("XML inválido: %s", e)
            self._reply(400, b"bad xml")
            return

        method = parsed.get("method")
        cwmp_ns = parsed.get("cwmp_ns") or NS_CWMP_CANDIDATES[0]
        cwmp_id = parsed.get("cwmp_id") or "1"

        # --- Inform: o CPE se apresenta ------------------------------------
        if method == "Inform":
            serial = parsed.get("serial") or "UNKNOWN"
            events = parsed.get("events") or []
            params = parsed.get("params") or {}
            log.info(
                "→ Inform serial=%s events=%s manufacturer=%s",
                serial,
                events,
                parsed.get("manufacturer"),
            )
            for k, v in list(params.items())[:12]:
                log.info("   param %s = %s", k, v)
            if len(params) > 12:
                log.info("   … +%s params", len(params) - 12)

            STORE.upsert_from_inform(
                serial=serial,
                manufacturer=parsed.get("manufacturer") or "",
                oui=parsed.get("oui") or "",
                product_class=parsed.get("product_class") or "",
                events=events,
                params=params,
                peer=peer,
            )
            # Lembra qual serial está nesta conexão TCP (sessão keep-alive)
            SESSION_SERIAL[id(self.connection)] = serial

            resp = soap_inform_response(cwmp_ns, cwmp_id)
            log.info("← InformResponse (MaxEnvelopes=1)")
            self._reply(200, resp, content_type="text/xml; charset=utf-8")
            return

        # --- GetRPCMethods: CPE pergunta o que o ACS sabe ------------------
        if method == "GetRPCMethods":
            log.info("→ GetRPCMethods")
            resp = soap_get_rpc_methods_response(cwmp_ns, cwmp_id)
            log.info("← GetRPCMethodsResponse")
            self._reply(200, resp, content_type="text/xml; charset=utf-8")
            return

        # --- Resposta a um GetParameterValues nosso ------------------------
        if method == "GetParameterValuesResponse":
            serial = SESSION_SERIAL.get(id(self.connection))
            params = parsed.get("params") or {}
            log.info("→ GetParameterValuesResponse (%s params) serial=%s", len(params), serial)
            for k, v in params.items():
                log.info("   %s = %s", k, v)
            if serial:
                dev = STORE.get(serial)
                if dev:
                    dev.params.update(params)
            # Sessão pode acabar aqui
            log.info("← 204 No Content (fim de sessão / sem mais RPC)")
            self._reply(204, b"")
            return

        # --- POST vazio: nossa vez de mandar RPC ou encerrar ---------------
        if parsed.get("empty") or method is None:
            serial = SESSION_SERIAL.get(id(self.connection))
            # Alguns CPE fecham o TCP após InformResponse e abrem outro.
            # Sem serial na conexão, usamos o Inform mais recente.
            if not serial:
                devices = STORE.list()
                if devices:
                    devices.sort(key=lambda d: d.last_inform_at, reverse=True)
                    serial = devices[0].serial
            log.info("→ POST vazio (ACS pode falar). serial=%s", serial)
            if serial:
                pending = STORE.pop_rpc(serial)
                if pending:
                    resp = build_envelope(
                        cwmp_ns=cwmp_ns, cwmp_id=cwmp_id, body_inner_xml=pending
                    )
                    log.info("← RPC pendente da fila")
                    self._reply(200, resp, content_type="text/xml; charset=utf-8")
                    return
            log.info("← 204 No Content (nada a pedir — sessão encerra)")
            self._reply(204, b"")
            return

        log.warning("Método não tratado: %s — encerrando com 204", method)
        self._reply(204, b"")

    def _reply(
        self,
        code: int,
        body: bytes,
        content_type: str = "text/plain",
    ) -> None:
        self.send_response(code)
        if code != 204:
            self.send_header("Content-Type", content_type)
            self.send_header("Content-Length", str(len(body)))
        self.send_header("Connection", "keep-alive")
        self.end_headers()
        if code != 204 and body:
            self.wfile.write(body)


# =============================================================================
# 5) CLI interativa — listar devices e disparar CR / GPV
# =============================================================================


def cli_loop(cr_user: str, cr_pass: str) -> None:
    help_txt = """
Comandos:
  list                         — devices que já mandaram Inform
  show <serial>                — params guardados
  cr <serial>                  — Connection Request (acorda o CPE)
  gpv <serial> <path> [path…]  — enfileira GetParameterValues + dispara CR
  help                         — esta ajuda
  quit                         — sai (Ctrl+C também)
""".strip()
    print(help_txt)
    while True:
        try:
            line = input("acs> ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            return
        if not line:
            continue
        parts = line.split()
        cmd = parts[0].lower()

        if cmd in ("quit", "exit", "q"):
            return
        if cmd == "help":
            print(help_txt)
            continue
        if cmd == "list":
            devices = STORE.list()
            if not devices:
                print("(nenhum Inform ainda — configure o ZTE apontando pra este ACS)")
                continue
            for d in devices:
                age = int(time.time() - d.last_inform_at) if d.last_inform_at else -1
                print(
                    f"- {d.serial}  {d.manufacturer} {d.product_class}  "
                    f"sw={d.software_version or '?'}  "
                    f"last_inform={age}s  cr={d.connection_request_url or '-'}"
                )
            continue
        if cmd == "show" and len(parts) >= 2:
            d = STORE.get(parts[1])
            if not d:
                print("serial não encontrado")
                continue
            print(d)
            for k, v in sorted(d.params.items()):
                print(f"  {k} = {v}")
            continue
        if cmd == "cr" and len(parts) >= 2:
            d = STORE.get(parts[1])
            if not d:
                print("serial não encontrado")
                continue
            ok, msg = connection_request(
                d.connection_request_url, username=cr_user, password=cr_pass
            )
            print(("OK" if ok else "FAIL"), msg)
            continue
        if cmd == "gpv" and len(parts) >= 3:
            serial = parts[1]
            paths = parts[2:]
            # Miolo sem envelope — a fila monta o envelope na hora do POST vazio
            items = "\n".join(f"        <string>{p}</string>" for p in paths)
            inner = f"""<cwmp:GetParameterValues>
      <ParameterNames soap-enc:arrayType="xsd:string[{len(paths)}]">
{items}
      </ParameterNames>
    </cwmp:GetParameterValues>"""
            if not STORE.enqueue_rpc(serial, inner):
                print("serial não encontrado — espere o primeiro Inform")
                continue
            d = STORE.get(serial)
            assert d
            print(f"GPV enfileirado ({len(paths)} paths). Disparando Connection Request…")
            ok, msg = connection_request(
                d.connection_request_url, username=cr_user, password=cr_pass
            )
            print(("OK" if ok else "FAIL"), msg)
            print("Aguarde o próximo Inform + sessão (logs acima).")
            continue

        print("comando inválido — digite help")


# =============================================================================
# main
# =============================================================================


def main() -> None:
    parser = argparse.ArgumentParser(description="ACS CWMP didático (porta 7547)")
    parser.add_argument("--host", default="0.0.0.0", help="Bind (default 0.0.0.0)")
    parser.add_argument("--port", type=int, default=7547, help="Porta CWMP (default 7547)")
    parser.add_argument(
        "--cr-user",
        default="",
        help="Usuário Connection Request (no ZTE: Management → TR069)",
    )
    parser.add_argument("--cr-pass", default="", help="Senha Connection Request")
    parser.add_argument("-v", "--verbose", action="store_true")
    args = parser.parse_args()

    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
        datefmt="%H:%M:%S",
    )

    httpd = ThreadingHTTPServer((args.host, args.port), AcsHandler)
    # Reuso de porta ajuda no lab (reinício rápido)
    httpd.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    lan_hint = _guess_lan_ip()
    log.info("ACS ouvindo em http://%s:%s/", args.host, args.port)
    if lan_hint:
        log.info(
            "No ZTE, URL do ACS costuma ser: http://%s:%s/",
            lan_hint,
            args.port,
        )
    log.info("Periodic Inform Enable = ON; intervalo curto (60s) ajuda no lab.")

    if sys.stdin.isatty():
        t = threading.Thread(
            target=cli_loop, args=(args.cr_user, args.cr_pass), daemon=True
        )
        t.start()
    else:
        log.info("stdin não é TTY — CLI interativa desligada (só logs HTTP).")

    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        log.info("Encerrando…")
    finally:
        httpd.server_close()


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


if __name__ == "__main__":
    main()