python多进程检查端口并写日志

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: 版权声明:本文可能为博主原创文章,若标明出处可随便转载。 https://blog.
版权声明:本文可能为博主原创文章,若标明出处可随便转载。 https://blog.csdn.net/Jailman/article/details/77573916
#!C:\Python27\python.exe
#coding: utf-8

from multiprocessing import Pool
import socket
socket.setdefaulttimeout(10)


#################################################################
import ctypes

STD_INPUT_HANDLE = -10
STD_OUTPUT_HANDLE = -11
STD_ERROR_HANDLE = -12

FOREGROUND_BLACK = 0x0
FOREGROUND_BLUE = 0x01  # text color contains blue.
FOREGROUND_GREEN = 0x02  # text color contains green.
FOREGROUND_RED = 0x04  # text color contains red.
FOREGROUND_INTENSITY = 0x08  # text color is intensified.

BACKGROUND_BLUE = 0x10  # background color contains blue.
BACKGROUND_GREEN = 0x20  # background color contains green.
BACKGROUND_RED = 0x40  # background color contains red.
BACKGROUND_INTENSITY = 0x80  # background color is intensified.


class Color:
    ''''' See http://msdn.microsoft.com/library/default.asp?url=/library/en-us/winprog/winprog/windows_api_reference.asp
    for information on Windows APIs.'''
    std_out_handle = ctypes.windll.kernel32.GetStdHandle(STD_OUTPUT_HANDLE)

    def set_cmd_color(self, color, handle=std_out_handle):
        """(color) -> bit
        Example: set_cmd_color(FOREGROUND_RED | FOREGROUND_GREEN | FOREGROUND_BLUE | FOREGROUND_INTENSITY)
        """
        bool = ctypes.windll.kernel32.SetConsoleTextAttribute(handle, color)
        return bool

    def reset_color(self):
        self.set_cmd_color(FOREGROUND_RED | FOREGROUND_GREEN | FOREGROUND_BLUE)

    def print_red_text(self, print_text):
        self.set_cmd_color(FOREGROUND_RED | FOREGROUND_INTENSITY)
        print print_text
        self.reset_color()

    def print_green_text(self, print_text):
        self.set_cmd_color(FOREGROUND_GREEN | FOREGROUND_INTENSITY)
        print print_text
        self.reset_color()

    def print_blue_text(self, print_text):
        self.set_cmd_color(FOREGROUND_BLUE | FOREGROUND_INTENSITY)
        print print_text
        self.reset_color()

    def print_red_text_with_blue_bg(self, print_text):
        self.set_cmd_color(FOREGROUND_RED | FOREGROUND_INTENSITY | BACKGROUND_BLUE | BACKGROUND_INTENSITY)
        print print_text
        self.reset_color()

clr = Color()
# clr.print_red_text('red')
# clr.print_green_text('green')
# clr.print_blue_text('blue')
# clr.print_red_text_with_blue_bg('background')
#################################################################

def check_server(address, port):
    s = socket.socket()
    # print "Attempting to connect to %s on port %s" % (address, port)
    try:
        port = int(port)
        s.connect((address, port))
        # print "Connected to %s on port %s" % (address, port)
        return True
    except socket.error, e:
        clr.print_red_text("Connection to %s on port %s failed: %s" % (address, port, e))
        return False


def single_func(i):
    with open('proxy.txt') as f:
        lines = f.readlines()
        if lines[i] != '':
            ip = lines[i].split(':')[0].strip()
            port = lines[i].split(':')[1].strip()
            if check_server(ip, port):
                msg = "Connected to %s on port %s" % (ip, port)
                clr.print_green_text(msg)
                return msg
            else:
                return "Connection to %s on port %s failed" % (ip, port)
        else:
            msg = "Empty line!"
            return msg
    # f.close()


def logger(msg):
    with open('log.txt', 'a+') as f:
        f.write(str(msg) + "\n")
        f.flush()


if __name__ == '__main__':
    try:
        pool = Pool(processes=4)
        with open('proxy.txt') as f:
            lines = f.readlines()
            for i in range(1, lines.__len__() + 1):
                result = pool.apply_async(single_func, (i,), callback=logger).get(99999)
            pool.close()
            pool.join()
            # f.close()
    except KeyboardInterrupt as e:
        clr.print_red_text("Interrupt Reason: %s" % str(e))




相关实践学习
【涂鸦即艺术】基于云应用开发平台CAP部署AI实时生图绘板
【涂鸦即艺术】基于云应用开发平台CAP部署AI实时生图绘板
目录
相关文章
|
网络协议 Linux
Linux查看端口监听情况,以及Linux查看某个端口对应的进程号和程序
Linux查看端口监听情况,以及Linux查看某个端口对应的进程号和程序
1317 2
|
运维 网络性能优化 网络虚拟化
|
11月前
|
存储 Prometheus 监控
Docker容器内进行应用调试与故障排除的方法与技巧,包括使用日志、进入容器检查、利用监控工具及检查配置等,旨在帮助用户有效应对应用部署中的挑战,确保应用稳定运行
本文深入探讨了在Docker容器内进行应用调试与故障排除的方法与技巧,包括使用日志、进入容器检查、利用监控工具及检查配置等,旨在帮助用户有效应对应用部署中的挑战,确保应用稳定运行。
460 5
|
运维 Linux
Linux查找占用的端口,并杀死进程的简单方法
通过上述步骤和命令,您能够迅速识别并根据实际情况管理Linux系统中占用特定端口的进程。为了获得更全面的服务器管理技巧和解决方案,提供了丰富的资源和专业服务,是您提升运维技能的理想选择。
821 1
|
Linux Python
用python扫描linux开放的端口(3种方式)
这篇文章介绍了三种使用Python实现Linux端口扫描的方法,包括基础版端口扫描、全端口扫描和多线程扫描技术。
456 16
|
网络协议 安全 Linux
阿里云国际上如何检查 Linux 上正在使用哪些端口
阿里云国际上如何检查 Linux 上正在使用哪些端口
|
Python
Python编程--使用NMAP端口扫描
Python编程--使用NMAP端口扫描
197 1
|
网络安全 Python
Python编程--目标IP地址段主机指定端口状态扫描
Python编程--目标IP地址段主机指定端口状态扫描
179 1
|
监控 网络协议 数据库连接
Python3 监控端口:使用 socket 库
Python3 监控端口:使用 socket 库
226 1
|
网络协议
Mac根据端口查询进程id的命令
这篇文章介绍了在Mac操作系统上如何使用两种命令来查询监听特定端口的进程ID。第一种方法是使用`netstat -anp tcp -v | grep 端口号`,例如`netstat -anp tcp -v | grep 80`,这将列出所有使用端口80的TCP连接及其相关信息。第二种方法是使用`lsof -P -n -i:端口号`,例如`lsof -P -n -i:8080`,这将显示使用指定端口的进程列表,包括进程ID、用户、文件描述符等信息。文章通过示例展示了如何使用这些命令,并提供了输出结果的截图。
965 2

推荐镜像

更多