netadm/scripts/check_available_ports.py

import socket
import re
from typing import List, Dict, Union
import logging


def check_conn(ip: str, port: int) -> bool:
    logging.info("")
    conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    conn.settimeout(3)
    try:
        conn.connect((ip, port))
        logging.info(f"{ip}:{port} - True")
        return True
    except TimeoutError:
        logging.warning(f"{ip}:{port} - Timeout")
        return False


def parse_input(text: str) -> List[Dict[str, Union[int, str]]]:
    logging.info("")
    pattern = r"Error;.*?(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}):(\d+)/(\w+)"
    matches = re.findall(pattern, text)
    result = [
        {"ip": match[0], "port": int(match[1]), "protocol": match[2]}
        for match in matches
    ]
    return result


def start_check(
    text: str,
    result_type: str = "json",
) -> List[Dict[str, Union[int, str, bool]]] | List[str]:
    logging.info("")
    result = []
    for i in parse_input(text):
        if i["protocol"] == "tcp":
            if check_conn(i["ip"], i["port"]):
                i["result"] = True
                result.append(i)
            else:
                i["result"] = False
                result.append(i)

    if result_type == "text":
        results = []
        for i in result:
            if i["result"]:
                results.append(f"{i['ip']}:{i['port']}/{i['protocol']} - OK")
            else:
                results.append(f"{i['ip']}:{i['port']}/{i['protocol']} - FAIL")
        return results
    return result