运维人员常用脚本

 1def extract_errors(log_file, output_file):
 2    with open(log_file, 'r') as f:
 3        lines = f.readlines()
 4    
 5    errors = [line for line in lines if "ERROR" in line]
 6    
 7    with open(output_file, 'w') as f:
 8        f.writelines(errors)
 9
10# 使用示例
11extract_errors("app.log", "errors.log")

 1from collections import defaultdict
 2
 3def count_keywords(log_file, keywords):
 4    keyword_count = defaultdict(int)
 5    
 6    with open(log_file, 'r') as f:
 7        for line in f:
 8            for keyword in keywords:
 9                if keyword in line:
10                    keyword_count[keyword] += 1
11    
12    return keyword_count
13
14# 使用示例
15keywords = ["ERROR", "WARNING", "INFO"]
16result = count_keywords("app.log", keywords)
17print(result)

 1from datetime import datetime
 2
 3def filter_logs_by_time(log_file, start_time, end_time, output_file):
 4    start = datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")
 5    end = datetime.strptime(end_time, "%Y-%m-%d %H:%M:%S")
 6    
 7    withopen(log_file, 'r') as f:
 8        logs = f.readlines()
 9    
10    filtered_logs = []
11    forloginlogs:
12        log_time_str = log.split()[0] + " " + log.split()[1] # 假设时间戳在日志的前两部分
13        log_time = datetime.strptime(log_time_str, "%Y-%m-%d %H:%M:%S")
14        ifstart <= log_time <= end:
15            filtered_logs.append(log)
16    
17    withopen(output_file, 'w') as f:
18        f.writelines(filtered_logs)
19
20# 使用示例
21filter_logs_by_time("app.log", "2025-02-26 12:00:00", "2025-02-06 14:00:00", "filtered_logs.log")

 1from collections import Counter
 2import re
 3
 4def top_n_errors(log_file, n=5):
 5    error_pattern = re.compile(r"ERROR: (.+)")
 6    errors = []
 7    
 8    with open(log_file, 'r') as f:
 9        for line in f:
10            match = error_pattern.search(line)
11            if match:
12                errors.append(match.group(1))
13    
14    return Counter(errors).most_common(n)
15
16# 使用示例
17top_errors = top_n_errors("app.log", n=3)
18print(top_errors)

1def merge_log_files(log_files, output_file):
2    with open(output_file, 'w') as outfile:
3        for log_file in log_files:
4            with open(log_file, 'r') as infile:
5                outfile.write(infile.read())
6
7# 使用示例
8merge_log_files(["log1.log", "log2.log", "log3.log"], "merged_logs.log")

 1import time
 2
 3def tail_log_file(log_file):
 4    with open(log_file, 'r') as f:
 5        f.seek(0, 2) # 移动到文件末尾
 6        while True:
 7            line = f.readline()
 8            if line:
 9                print(line.strip())
10            else:
11                time.sleep(0.1)
12
13# 使用示例
14tail_log_file("app.log")

 1import psutil
 2import time
 3
 4def monitor_system(interval=1):
 5    while True:
 6        cpu_usage = psutil.cpu_percent(interval=interval)
 7        memory_usage = psutil.virtual_memory().percent
 8        print(f"CPU Usage: {cpu_usage}% | Memory Usage: {memory_usage}%")
 9        time.sleep(interval)
10
11# 使用示例
12monitor_system(interval=2)

 1import pynvml
 2
 3def monitor_gpu_usage():
 4    pynvml.nvmlInit()
 5    device_count = pynvml.nvmlDeviceGetCount()
 6    
 7    for i in range(device_count):
 8        handle = pynvml.nvmlDeviceGetHandleByIndex(i)
 9        util = pynvml.nvmlDeviceGetUtilizationRates(handle)
10        memory_info = pynvml.nvmlDeviceGetMemoryInfo(handle)
11        print(f"GPU {i}: Usage={util.gpu}%, Memory Used={memory_info.used / 1024 ** 2} MB")
12
13# 使用示例
14monitor_gpu_usage()

 1import psutil
 2import time
 3
 4def monitor_network_usage(interval=1):
 5    old_value = psutil.net_io_counters().bytes_sent + psutil.net_io_counters().bytes_recv
 6    
 7    whileTrue:
 8        new_value = psutil.net_io_counters().bytes_sent + psutil.net_io_counters().bytes_recv
 9        bandwidth = (new_value - old_value) / interval # 计算带宽（字节/秒）
10        print(f"Network Bandwidth: {bandwidth} B/s")
11        
12        old_value = new_value
13        time.sleep(interval)
14
15# 使用示例
16monitor_network_usage(interval=2)

 1import psutil
 2import time
 3
 4def monitor_disk_io(interval=1):
 5    old_read = psutil.disk_io_counters().read_bytes
 6    old_write = psutil.disk_io_counters().write_bytes
 7    
 8    whileTrue:
 9        new_read = psutil.disk_io_counters().read_bytes
10        new_write = psutil.disk_io_counters().write_bytes
11        
12        read_speed = (new_read - old_read) / interval
13        write_speed = (new_write - old_write) / interval
14        
15        print(f"Read Speed: {read_speed / 1024} KB/s | Write Speed: {write_speed / 1024} KB/s")
16        
17        old_read = new_read
18        old_write = new_write
19        time.sleep(interval)
20
21# 使用示例
22monitor_disk_io(interval=2)

 1import psutil
 2
 3def monitor_process(pid):
 4    process = psutil.Process(pid)
 5    
 6    while True:
 7        cpu_usage = process.cpu_percent(interval=1)
 8        memory_usage = process.memory_info().rss / 1024 ** 2  # 转换为MB
 9        print(f"PID {pid}: CPU={cpu_usage}%, Memory={memory_usage} MB")
10
11# 使用示例
12monitor_process(1234) # 替换为目标进程的PID

 1import psutil
 2
 3def monitor_temperature():
 4    temps = psutil.sensors_temperatures()
 5    for name, entries in temps.items():
 6        for entry in entries:
 7            print(f"{name}: {entry.label} = {entry.current}°C")
 8
 9# 使用示例
10monitor_temperature()

 1import pandas as pd
 2
 3def clean_data(input_file, output_file):
 4    df = pd.read_csv(input_file)
 5    df.dropna(inplace=True) # 删除空值
 6    df.drop_duplicates(inplace=True) # 删除重复值
 7    df.to_csv(output_file, index=False)
 8
 9# 使用示例
10clean_data("data.csv", "cleaned_data.csv")

 1import pandas as pd
 2
 3def compare_data(file1, file2):
 4    df1 = pd.read_csv(file1)
 5    df2 = pd.read_csv(file2)
 6    diff = df1.compare(df2)
 7    return diff
 8
 9# 使用示例
10result = compare_data("file1.csv", "file2.csv")
11print(result)

1import os
2
3def batch_rename(directory, prefix):
4    for count, filename in enumerate(os.listdir(directory)):
5        new_name = f"{prefix}_{count}.txt"
6        os.rename(os.path.join(directory, filename), os.path.join(directory, new_name))
7
8# 使用示例
9batch_rename("/path/to/files", "file")

 1import os
 2
 3def find_large_files(directory, size_limit_mb):
 4    size_limit = size_limit_mb * 1024 * 1024# 转换为字节
 5    large_files = []
 6    
 7    for root, dirs, files in os.walk(directory):
 8        for file in files:
 9            file_path = os.path.join(root, file)
10            if os.path.getsize(file_path) > size_limit:
11                large_files.append(file_path)
12    
13    return large_files
14
15# 使用示例
16large_files = find_large_files("/path/to/directory", 100) # 查找大于100MB的文件
17print(large_files)

 1import socket
 2
 3def check_port(host, port):
 4    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 5    result = sock.connect_ex((host, port))
 6    sock.close()
 7    return result == 0
 8
 9# 使用示例
10if check_port("example.com", 80):
11    print("Port 80 is open")
12else:
13    print("Port 80 is closed")

 1import os
 2
 3def ping_hosts(hosts):
 4    for host in hosts:
 5        response = os.system(f"ping -c 1 {host}")
 6        if response == 0:
 7            print(f"{host} is up")
 8        else:
 9            print(f"{host} is down")
10
11# 使用示例
12hosts = ["google.com", "example.com", "localhost"]
13ping_hosts(hosts)

 1import time
 2from datetime import datetime
 3
 4def scheduled_task(interval):
 5    while True:
 6        print(f"Task executed at {datetime.now()}")
 7        time.sleep(interval)
 8
 9# 使用示例
10scheduled_task(60) # 每60秒执行一次