python获取机器信息脚本(网上寻找的)

获取机器信息(待测试)

python;gutter:true;</p> <h1>-<em>- coding: UTF-8 -</em>-</h1> <p>import psutil import json import os import socket import struct import time import sys</p> <h1>当前时间函数</h1> <p>def now_time(): now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time())) return now_time</p> <h1>获取硬盘信息函数</h1> <p>def disk_info(): disk_dict = {} disk_name = [] disk_parttions = psutil.disk_partitions() for i in range(len(disk_parttions)): parttions_name = disk_parttions[i].device mountpoint = disk_parttions[i].mountpoint mountfstype = disk_parttions[i].fstype disk_info = psutil.disk_usage(mountpoint) total = disk_info.total used = disk_info.used free = disk_info.free disk_io = psutil.disk_io_counters(perdisk=True) disk_name.append(str(parttions_name.split('/', parttions_name.count('/'))[-1])) for key1 in disk_name: tmp_disk_list = [] for key2 in disk_io.keys(): tmp_disk_list.append(key2) if key1 in tmp_disk_list: disk_dict[parttions_name] = { "read_bytes": disk_io[key1].read_bytes, "write_bytes": disk_io[key1].write_bytes, "read_count": disk_io[key1].read_count, "write_count": disk_io[key1].write_count, "mountpoint": mountpoint, "mountfstype": mountfstype, "total": total, "used": used, "free": free } else: pass return disk_dict</p> <h1>获取内存信息函数</h1> <p>def mem_info(): mem_dict_list = psutil.virtual_memory() mem_total = mem_dict_list.total mem_used = mem_dict_list.used mem_free = mem_dict_list.free cpu_use = psutil.cpu_percent(interval=True) host_runtime = psutil.boot_time() mem_list_info = { "mem_total": mem_total, "mem_used": mem_used, "mem_free": mem_free, "cpu_used": cpu_use, "host_run": host_runtime } return mem_list_info</p> <h1>获取CPU信息函数</h1> <p>def cpu_info(): cpu_use = psutil.cpu_percent(interval=True) cpu_usage = { "cpu_used": cpu_use } return cpu_usage</p> <h1>获取网卡信息函数</h1> <p>def net_info(): net_dict_list = {} net_dev = psutil.net_io_counters(pernic=True) for dev_name in net_dev.keys(): net_dict_list[dev_name] = { "RX_b": net_dev[dev_name].bytes_recv, "TX_b": net_dev[dev_name].bytes_sent, "Dropout": net_dev[dev_name].dropout, "Dropint": net_dev[dev_name].dropin } return net_dict_list</p> <h1>获取用户信息函数</h1> <p>def user_info(): user_dict_list = [] user_dict = psutil.users() for i in range(len(user_dict)): user_dict_tmp = {} user_dict_tmp[user_dict[i].name] = { "terminal": user_dict[i].terminal, "host": user_dict[i].host, "start_time": user_dict[i].started } user_dict_list.append(user_dict_tmp) return user_dict_list</p> <h1>获取内存消耗前10的进程函数</h1> <p>def process_info(): process_dict_list_tmp = [] pid_dict = psutil.pids() for i in pid_dict: pid = psutil.Process(i) pid_tuple = (i, pid.name(), pid.memory_percent(), pid.create_time(), pid.status(), pid.cwd(), pid.exe()) process_dict_list_tmp.append(pid_tuple) process_dict_list_tmp.sort(key=lambda mem1: mem1[2], reverse=True) process_dict_list = process_dict_list_tmp[0:10] return process_dict_list</p> <h1>获取开机时间函数</h1> <p>def runtime_info(): host_runtime = psutil.boot_time() host_run_time = { "host_runtime": host_runtime } return host_run_time</p> <h1>没用的函数</h1> <p>def get_info(): host_info = { 'disk_info': disk_info(), 'mem_info': mem_info(), 'cpu_info': cpu_info(), 'net_info': net_info(), 'process_info': process_info(), 'user_info': user_info(), 'runtime_info': runtime_info() } return repr(host_info)</p> <h1>socket数据发送函数</h1> <p>def socket_send(socket_server=None, socket_port=None, cammds=None, type=None, PackName=None): if type == 'regs': s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.connect((socket_server, socket_port)) s.sendall(cammds) request = str(s.recv(18)) s.close() return request elif type == 'data': s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.connect((socket_server, socket_port)) s.sendall(cammds) request = str(s.recv(128)) if request == 'OK': Rtime = now_time() PackSize = sys.getsizeof(cammds) print(" '%s' : transfer TCP Package '%s' Size '%s' sucessful ") % (Rtime, PackName, PackSize) s.close()</p> <h1>构架TCP格式数据包函数</h1> <p>def trans_pack(commands=None, type=None): Net_type = 4 # package versions Mem_type = 1 Cpu_type = 2 Disk_type = 3 Runtime_type = 5 if type == 'Mem': machine_number = json_conf()["machine_number"] # 获取机器码 body = json.dumps(dict(commands)) # 生成JSON格式包体 header = [Mem_type, body.<strong>len</strong>(), machine_number] # 组成包头格式为:版本,包体长度,机器码 headPack = struct.pack("!3I", <em>header) # 生成包头 sendPack = headPack + body.encode() # 构建TCP数据包 return sendPack elif type == 'Net': machine_number = json_conf()["machine_number"] # 获取机器码 body = json.dumps(dict(commands)) # 生成JSON格式包体 header = [Net_type, body.<strong>len</strong>(), machine_number] # 组成包头格式为:版本,包体长度,机器码 headPack = struct.pack("!3I", </em>header) # 生成包头 sendPack = headPack + body.encode() # 构建TCP数据包 return sendPack elif type == 'Cpu': machine_number = json_conf()["machine_number"] # 获取机器码 body = json.dumps(dict(commands)) # 生成JSON格式包体 header = [Cpu_type, body.<strong>len</strong>(), machine_number] # 组成包头格式为:版本,包体长度,机器码 headPack = struct.pack("!3I", <em>header) # 生成包头 sendPack = headPack + body.encode() # 构建TCP数据包 return sendPack elif type == 'Disk': machine_number = json_conf()["machine_number"] # 获取机器码 body = json.dumps(dict(commands)) # 生成JSON格式包体 header = [Disk_type, body.<strong>len</strong>(), machine_number] # 组成包头格式为:版本,包体长度,机器码 headPack = struct.pack("!3I", </em>header) # 生成包头 sendPack = headPack + body.encode() # 构建TCP数据包 return sendPack elif type == 'Run': machine_number = json_conf()["machine_number"] # 获取机器码 body = json.dumps(dict(commands)) # 生成JSON格式包体 header = [Runtime_type, body.<strong>len</strong>(), machine_number] # 组成包头格式为:版本,包体长度,机器码 headPack = struct.pack("!3I", *header) # 生成包头 sendPack = headPack + body.encode() # 构建TCP数据包 return sendPack</p> <h1>读取JSON配置函数</h1> <p>def json_conf(): a = os.getcwd() # os.chdir("%s/client/conf" % a) # 改变工作目录为JSON文件目录 json_list = file("%s/client/conf/client_conf.json" % a) # 读取JSON配置文件 m = json_list.read() json_list.seek(0) json_conf = json.loads(json_list.read()) # 取出JSON配置文件 json_list.close() return json_conf</p> <p>def json_write(command=None): a = os.getcwd() os.chdir("%s/client/conf" % a) print(command) b = type(command) m = open("client_conf.json", 'w')</p> <h1>agent第一次注册函数</h1> <p>def client_regs(): host_info = json.dumps({'ip': str(json_conf()['local_adder']), 'mac': str(json_conf()['local_mac'])}) regs_addr = json_conf()['send_server'] regs_port = json_conf()['server_regs_port'] server_request = str(socket_send(socket_server=regs_addr, socket_port=regs_port, cammds=host_info, type='regs')) conf = json_conf() conf['machine_number'] = server_request conf = json.dumps(conf) json_write(command=conf)</p> <h1>数据包发送函数</h1> <p>def send_data(): DiskPack = trans_pack(commands=disk_info(), type='Disk') # 生成TCP数据包 MemPack = trans_pack(commands=mem_info(), type='Mem') NetPack = trans_pack(commands=net_info(), type='Net') regs_addr = json_conf()['send_server'] # 获取服务器地址 regs_port = json_conf()['send_port'] # 获取服务器端口 socket_send(socket_server=regs_addr, socket_port=regs_port, cammds=DiskPack, type='data', PackName='DiskPack') # 向服务器发送数据包 socket_send(socket_server=regs_addr, socket_port=regs_port, cammds=MemPack, type='data', PackName='MemPack') socket_send(socket_server=regs_addr, socket_port=regs_port, cammds=NetPack, type='data', PackName='NetPack') time.sleep(json_conf()['sendoff_time']) # 等待延时</p> <p>def check_client(): print('start get client info....') machine_number = json_conf()['machine_number'] if machine_number: print(machine_number) print(11111) else: print(machine_number) print(22222)</p> <p>if <strong>name</strong> == '<strong>main</strong>': while True: send_data()

Original: https://www.cnblogs.com/nmap/p/10474231.html
Author: nmap
Title: python获取机器信息脚本(网上寻找的)

原创文章受到原创版权保护。转载请注明出处:https://www.johngo689.com/10825/

转载文章受原作者版权保护。转载请注明原作者出处!

(0)

大家都在看

免费咨询
免费咨询
扫码关注
扫码关注
联系站长

站长Johngo!

大数据和算法重度研究者!

持续产出大数据、算法、LeetCode干货,以及业界好资源!

2022012703491714

微信来撩,免费咨询:xiaozhu_tec

分享本页
返回顶部