mysql all_ip_test局域网IP测试工具,有需要的改一改.

import threading
import subprocess
import pymysql
# threading.Lock()


class Link(object):
 
    ip = list()
    def __init__(self):

        self.connection = pymysql.connect(host='localhost', port=3306, user='**', passwd='**', db='**',
                                     charset='utf8', autocommit=True)
        
        self.cursor = self.connection.cursor()

        sql = "select ip from ip"  
        self.all = self.cursor.execute(sql)
        for i in self.cursor.fetchall():
            Link.ip.append(i[0])
        self.lock = threading.Lock()    


    def ping_ip(self,ip):

        res = subprocess.call('ping -n 4 -w 5 %s' % ip, stdout=subprocess.PIPE)
        print(res,">>>",ip,"<<<")
        if res == 0:

            sql = """UPDATE ip SET action = '在线' where ip like '%{}';"""  .format(ip)
            self.lock.acquire()     
            self.cursor.execute(sql)
            self.lock.release()     
            print('通',ip)
        else:
            print("不通",ip)


    def run(self):
        for i in Link.ip:　　# global take
            take = threading.Thread(target=self.ping_ip,args=(i,))
            take.start()

        take.join()


def main():
    run = Link()
    run.run()




if __name__ == '__main__':
    main()