gistfile1.txt

# -*- coding: utf-8 -*-
from socket import socket, AF_INET, SOCK_STREAM
from threading import Thread
from queue import Queue
import CONST


class PortChecker:
    def __init__(self, brokeCount=2, tryCount=4, timeOut=600, threads=CONST.threads):
        self.pingQueue = Queue()
        self.port_checkQueue = Queue()
        self.allOffIP = []
        self.allOnIP = []
        self.allOffPort = []
        self.allOnPort = []
        self.threads = threads
        self.tryCount = tryCount if isinstance(tryCount, str) else str(tryCount)
        self.brokeCount = brokeCount if isinstance(brokeCount, str) else str(brokeCount)
        self.timeOut = timeOut if isinstance(timeOut, str) else str(timeOut)

    # Проверка порта.
    def port_checkForThreads(self, queue, IP, log=False):
        while not queue.empty():
            port = queue.get()
            s = socket(AF_INET, SOCK_STREAM)
            result = s.connect_ex((IP, port))
            if result == 0:
                self.allOnPort.append({'port': port, 'err': CONST.port_checkErr[0]})
            else:
                for err in CONST.port_checkErr:
                    if err['id'] == result:
                        self.allOffPort.append({'port': port, 'err': err})
                        break
            if log:
                for err in CONST.port_checkErr:
                    if err['id'] == result:
                        print("Port: {0}\n'en': {1}.\n'ru': {2}.".format(port, err['en'], err['ru']))
                        break
            # print(port)
            s.close()
            queue.task_done()

    def port_check(self, IP='127.0.0.1', printOnPort=True, printOffPort=False, portEnum=False, ports=[1, 65535]):
        if ports is None:
            ports = [80]
        if portEnum:
            for port in ports:
                self.port_checkQueue.put(port)
        else:
            for port in range(ports[0], ports[1] + 1):
                self.port_checkQueue.put(port)
        for i in range(self.threads):
            worker = Thread(target=self.port_checkForThreads, args=(self.port_checkQueue, IP))
            worker.setDaemon(True)
            worker.start()
        self.port_checkQueue.join()
        if printOnPort:
            print("\n\nРабочие порты")
            for i in self.allOnPort:
                # print("Port: {0}\n'en': {1}.\n'ru': {2}.".format(i['port'], i['err']['en'], i['err']['ru']))
                print("Port: {0}".format(i['port']))
        if printOffPort:
            print("\n\nНе рабочие порты")
            for i in self.allOffPort:
                print("Port: {0}\n'en': {1}.\n'ru': {2}.".format(i['port'], i['err']['en'], i['err']['ru']))


if __name__ == '__main__':
    checker = PortChecker(threads=100)
    checker.port_check("192.168.1.1", ports=[1, 120])