class Scan:
    def __init__(self, targets: List[str], ports: List[int]):
        self.ports = ports
        self.targets = targets
        self.loop = asyncio.get_running_loop()
        self.result_queue = asyncio.Queue()
        self.target_queue = asyncio.Queue()

    async def scan_target_port_worker(self, worker_name: str, timeout: float=1):
        while True:
            try:
                target, port = await self.target_queue.get()
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.setblocking(False)
                result = False
                while True:
                    try:
                        # await asyncio.wait_for(self.loop.sock_connect(sock, (target, port)), timeout)
                        async with asyncio.timeout(timeout):
                            await self.loop.sock_connect(sock, (target, port))
                        if target == '192.168.22.2' and port in (1, 22, 8080, 8081, 8082):
                            print(f"[{worker_name}][1] [{target},{port}] -> {sock.getpeername()}")
                    except asyncio.TimeoutError:
                        if timeout > 10:
                            print(f"[-] connect ({target}, {port}) timeout: {timeout}")
                            break
                        timeout += 0.5
                        continue
                    except (ConnectionRefusedError, ConnectionResetError, ConnectionAbortedError) as e:
                        if target == '192.168.22.2' and port in (1, 22, 8080, 8081, 8082):
                            print(f"[{worker_name}][2] [{target},{port}] -> {str(e)}")
                            raise
                        pass
                    except Exception as e:
                        if target == '192.168.22.2' and port in (1, 22, 8080, 8081, 8082):
                            print(f"[{worker_name}][3] [{target},{port}] -> {str(e)}")
                            raise
                        print(f"[-] connect ({target},{port}) errno: {e.errno}, error: {str(e)}")
                    else:
                        result = True
                        if target == '192.168.22.2' and port in (1, 22, 8080, 8081, 8082):
                            print(f"[{worker_name}][5] [{target},{port}] -> {result}")
                    finally:
                        if target == '192.168.22.2' and port in (1, 22, 8080, 8081, 8082):
                            print(f"[{worker_name}][6] [{target},{port},{timeout}] -> result: {result} ---> {sock}\n")
                        await self.result_queue.put((target, port, result, worker_name, timeout))
                        sock.close()
                        self.target_queue.task_done()
                        break
            except asyncio.CancelledError:
                break
            except Exception as e:
                print("[-] open tcp port other error:", str(e))
                break
    
    async def result_consumer(self, total: int):
        response = {}
        while True:
            try:
                host, port, is_open, work_name, timeout = await asyncio.wait_for(self.result_queue.get(), 5)
            except asyncio.TimeoutError:
                print(f"[-] wait result timeout, except {total} results")
                break
            except asyncio.CancelledError:
                break
            else:
                if is_open:
                    print(f"[consumer][{work_name}] [{host},{port},{timeout}] -> {is_open}")
                    if host not in response:
                        response[host] = []
                    response[host].append(port)
                total -= 1
                if total == 0:
                    break
                if (total % 10000) == 0:
                    print(f"[consumer] total:", total)
        return response
    
    async def scan_port(self, workers: int=5000):
        start_time = time.time()
        tasks = [asyncio.create_task(self.scan_target_port_worker(worker_name=f"worker {i}")) for i in range(workers)]
        total= len(self.targets) * len(self.ports)
        consumer_task = asyncio.create_task(self.result_consumer(total))

        for target in self.targets:
            for port in self.ports:
                self.target_queue.put_nowait((target, port))
        print("[.] all targets are to queue, total: %d size: %d" % (total, self.target_queue.qsize()))
        await self.target_queue.join()
        print("[.] target join finish")
        try:
            await consumer_task
        except Exception as e:
            print("[-] await consumer error:", str(e))
            raise
        else:
            response = consumer_task.result()
            if not consumer_task.cancelled():
                consumer_task.cancel()
        
        for task in tasks:
            task.cancel()

        count = int(time.time() - start_time)
        print(f"[.] total {count} seconds")
        return response