Python办公自动化之网络监控和压缩文件处理

# 使⽤ ping3 库进⾏Ping测试
import ping3

# 创建Ping对象
pinger = ping3.Ping()

# Ping主机并获取延迟（如果主机可达）
host = "example.com"
response_time = pinger.ping(host)
if response_time is not None:
	print(f"{host} 可达，延迟为 {response_time} 毫秒")
else:
	print(f"{host} 不可达")

# 使⽤ socket 库检查端⼝状态
import socket

# 检查端⼝是否开放
def is_port_open(host, port):
	sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
	sock.settimeout(1)
	result = sock.connect_ex((host, port))
	sock.close()
	return result == 0

host = "example.com"
port = 80
if is_port_open(host, port):
	print(f"端⼝ {port} 开放")
else:
	print(f"端⼝ {port} 未开放")

#!/usr/bin/env python3  
import sys  
import subprocess  
  
def check_ssh(host, port):  
    try:  
        result = subprocess.run(['ssh', f'{host}:{port}', 'echo', 'OK'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)  
        if result.stdout == b'OK\n':  
            return 0  # 正常状态  
        else:  
            return 2  # 警告状态  
    except Exception as e:  
        return 3  # 未知状态  
  
if __name__ == '__main__':  
    host = sys.argv[1]  # 获取主机名或IP地址  
    port = int(sys.argv[2])  # 获取端口号  
    status = check_ssh(host, port)  
    if status == 0:  
        print('SSH服务正常')  
    elif status == 2:  
        print('SSH服务警告')  
    else:  
        print('SSH服务未知')

define host {  
    host_name           my_server  
    alias               My Server  
    address             192.168.1.100  
    check_command        check-host-alive  
    check_period         24x7  
    retry_interval       10  
    max_check_attempts    10  
}  
  
define service {  
    service_description   SSH Service  
    host_name             my_server  
    check_command         check_ssh!12345  # 替换为实际的主机名或IP地址和端口号  
    check_period          24x7  
    retry_interval         5  
    max_check_attempts     3  
}

import pyshark

# 捕获⽹络流量
cap = pyshark.LiveCapture(interface='eth0')

# 分析数据包
for packet in cap.sniff_continuously():
	print(packet)

import pcapy  
from dpkt import ethhdr, ip, tcp  
  
# 打开网络接口进行监听  
dev = pcapy.findalldevs()[0]  # 通常使用第一个可用的网络接口  
cap = pcapy.open_live(dev, 65536, 1, 0)  # 65536是缓冲区大小，1表示只捕获进入的数据包，0表示不进行混杂模式  
  
try:  
    while True:  
        (header, payload) = cap.next()  # 读取数据包  
        if header is None:  
            break  
        eth = ethhdr(header)  # 解析以太网头  
        if eth.type == 0x800:  # 判断是否为IP数据包  
            ip_packet = ip.ip(header[14:])  # 解析IP头  
            if ip_packet.p == 0x11:  # 判断是否为TCP数据包  
                tcp_packet = tcp.tcp(payload)  # 解析TCP头  
                print(f"Source IP: {ip_packet.src}, Destination IP: {ip_packet.dst}, Source Port: {tcp_packet.sport}, Destination Port: {tcp_packet.dport}")  
except KeyboardInterrupt:  
    pass  # 可以通过Ctrl+C退出程序  
finally:  
    cap.close()  # 关闭捕获接口

## 压缩⽂件：
import zipfile

# 创建⼀个ZIP⽂件并添加⽂件
with zipfile.ZipFile('example.zip', 'w') as zipf:
	zipf.write('file1.txt')
	zipf.write('file2.txt')
 
 
## 解压⽂件：
import zipfile

# 解压ZIP⽂件
with zipfile.ZipFile('example.zip', 'r') as zipf:
	zipf.extractall('extracted_files/')
 
 
## 列出ZIP⽂件中的⽂件：
import zipfile

# 列出ZIP⽂件中的⽂件
with zipfile.ZipFile('example.zip', 'r') as zipf:
	file_list = zipf.namelist()
	print(file_list)

# 压缩⽂件：
import rarfile

# 创建⼀个RAR⽂件并添加⽂件
with rarfile.RarFile('example.rar', 'w') as rar:
	rar.write('file1.txt')
	rar.write('file2.txt')


# 解压⽂件：
import rarfile

# 解压RAR⽂件
with rarfile.RarFile('example.rar', 'r') as rar:
	rar.extractall('extracted_files/')
 
 
# 列出RAR⽂件中的⽂件：
import rarfile

# 列出RAR⽂件中的⽂件
with rarfile.RarFile('example.rar', 'r') as rar:
	file_list = rar.namelist()
	print(file_list)

# 从 tar.gz 文件中删除一个文件
with tarfile.open('example.tar.gz', 'r:gz') as tar:  
    for member in tar.getmembers():  
        if member.name == 'file_to_add.txt':  
            tar.remove(member)  
            break  
    tar.write("new_example.tar.gz", members=tar.getmembers())

# 创建时为其设置密码：
with zipfile.ZipFile('example_password_protected.zip', 'w', zipfile.ZIP_DEFLATED) as zf:  
    zf.setpassword(b'my_password'.encode('utf-8'))  # 设置密码为 "my_password"  
    zf.write('file_to_add.txt')