Python运维-文本处理、系统和文件信息监控、外部命令

本节主要目录如下：

一、文本处理

1.1、Python编码解码

1.2、文件操作

1.3、读写配置文件

1.4、解析XML文件

二、系统信息监控

2.1、监控CPU信息

2.2、监控内存信息

2.3、监控磁盘信息

2.4、监控网络信息

2.5、获取进程信息

2.6、实例：常见的实用方法

三、文件系统监控

3.1、实例

四、执行外部命令subprocess

4.1、subprocess.run()方法*

4.2、Popen类

4.3、其他方法

一、文本处理

日常运维工作都离不开文本，如日志分析、编码转换、ETL加工等。

1.1、Python编码解码

• 计算机只处理二进制数据，如果要处理文本，就需要将文本转换为二进制数据，再由计算机进行处理
• 将文本转换为二进制数据就是编码，将二进制数据转换为文本就是解码。
• 编码和解码要按一定的规则进行，这个规则就是字符集

常见的中文编码：

• GB2312或GB2312-80是中国国家标准简体中文字符集，共收录6763个汉字，同时收录了包括拉丁字母、希腊字母。日文平假名字母、俄语西里尔字母在内的682个字符
• GBK即汉字内码扩展规范，共收入21886个汉字和图形符号
• GB8030与GB2312-1980和GBK兼容，共收录汉字70244个，是一二四字节变长编码

可看出支持的汉字范围：GB18030>GBK>GB2312

把世界所有语言统一到一套编码中，这套编码就是Unicode编码。Unicode编码使用两个字节（16位bit）表示一个字符，比较偏僻的字符需要使用4个字节。

几乎所有的系统、编程语言都默认支持Unicode。如果一段纯英文文本，Unicode会占用比ASCII码多一倍的空间！UTF编码将一个Unicode字符编码成1-6个字节，常用的英文字母被编译成1个字节，汉字通常是3个字节，只有很生僻的字符才会被编码成4-6个字节。UTF编码有三种：

• UTF-8：使用1、2、3、4个字节表示所有符号，优先使用1个字节，若无法满足，则增加一个字节，最多4个字节 。英文占1个字节、欧洲语系占2个字节、东亚占3个字节，其他特殊字符占4个字节。
• UTF-16：使用2、4个字节表示所有符号，优先使用2个字节，否则使用4个字节表示。

• UTF-32：使用4个字节表示所有字符。

 # 汉字的“汉”，在UTF-8字符集中3个字节
 list("汉".encode("UTF-8"))
 # [230, 177, 137]

而英文无论采用哪种编码，都是一致的。如果使用纯英文编写代码，就基本不会遇到编码问题：

 print(list("a".encode("ascii")))
 # [97]
 print(list("a".encode("gbk")))
 # [97]
 print(list("a".encode("utf-8")))
 # [97]

Python语言的with...as...用法：

 with open("a.txt") as file:
     data = file.read()

默认编码可以通过sys.getdefaultencoding()来查看Python解释器会用的默认编码：

 import sys
 print(sys.getdefaultencoding())
 # utf-8

说明电脑上Python解释器默认使用的是UTF-8编码，如果不指定Python解释器以何种编码解码，则默认以UTF-8方式解码源文件，因此在保存代码源文件时确保以UTF-8编码保存。

1.2、文件操作

1.2.1、普通文件操作

Python文件操作只需要一个open函数返回一个文件句柄，无需导入任何模块。

 f=open("a.txt")     # 打开文件，得到一个文件句柄，并赋值给一个变量
 print(f.read())     # 打印读取文件的内容 
 f.close()           # 关闭文件

oepn函数：

 open(file, mode='r', buffering=None, encoding=None, errors=None, newline=None, closefd=True, opener=None)

• 参数file是一个文件名称的字符串，如果文件不在程序当前的路径下，就需要在前面加上相对路径或绝对路径

• 参数mode是一个可选参数，指示打开文件的方式，若不指定，则默认以读文本的方式打开文件：

  字符串	含义
  'r'	以读的方式打开（默认）
  'w'	以写的方式打开文件，会先清空文件
  'X'	创建一个新文件，以写方式打开
  'a'	以写的方式打开文件，如果文件已存在，就在文件最后位置追加内容
  'b'	以二进制方式打开，可以和读写命令共用
  't'	以文本方式（默认）
  '+'	以读和写方式打开文件，用于更新文件
  'U'	通用的换行模式（弃用）

• buffering是一个可选的参数，buffering=0表示关闭缓冲区（仅在二进制方式打开可用）,buffering=1表示选择缓冲区（仅在文本方式打开时可用）；buffering大于1时，其值代表固定大小的块缓冲区的大小。当不指定该参数时：二进制文件使用固定大小的块缓冲区，文本文件使用行缓冲区。

注意：

• 记得使用完毕后及时关闭文件，释放资源。推荐傻瓜式操作方式：使用with关键字来帮我们管理上下文，系统会自动为我们关闭文件和处理异常：

   with open('a.txt','w') as f:
       f.write("hello world")

• open()函数是操作系统打开文件，如果没有指定编码，那么以操作系统默认编码打开；Windos下是gbk，在Linux下使utf-8。

常见的文件操作方法：

读取文件内位置的定位方法：

• 通过read方法传输参数
• 以字节为单位定位，如seek、tell等。
  • seek(x,0)表示从起始位置即文件首行首字符开始移动x个字符。
  • seek(x,1)表示从当前位置向后移动x个字符.

  • seek(-x,2)表示从文件的结尾向前移动x个字符。

# 基于seek实现类似Linux命令tail-f的功能
 import time
 with open('tmp.txt','rb') as f:
     f.seek(0,2)         # 将光标移动至文件末尾
     while True:         # 实时显示文件新增加的内容
         line = f.read()
         if line:
             print(line.decode('utf-8'),end='')
         else:
             time.sleep(0,2) # 读取完毕后短暂的睡眠
 # 当tmp.txt追加新的内容时，新内容会被程序立即打印出来

1.2.2、大文件的读取

当文件较小时，我们可以一次性全部读入内存，对文件的内容做出任意修改，再保存至磁盘：

with open('a.txt') as read_f,open('.a.txt.swap','w') as write_f:
	data = read_f.read()	# 全部读入内存，如果文件很大，则会很卡
	data = data.replace('str1','str2')	# 在内存中完成修改
	write_f.write(data)	# 一次性写入新文件
os.remove('a.txt')
os.rename('.a.txt.swap','a.txt')

当文件很大时，如GB级的文本文件，我们需要用文件的可迭代方式将文件的内容逐行读入内存，在逐行写入新文件，最后使用新文件覆盖源文件。

with open('a.txt') as read_f,open('.a.txt.swap','w') as write_f:
    for line in read_f:	# 可迭代对象f逐行操作，防止内存溢出
        line=line.replace('str1','str2')
        write_f.write(line)
os.remove('a.txt')
os.rename('.a.txt.swap','a.txt')

处理大数据还有多种方法：

1. 通过read(size)增加参数，指定读取的字节数。

   while True:
   	block = f.read(1024)
   	if not block:
   		break

2. 通过readline()，每次只读一行。

   while True:
   	line = f.readline()
   	if not line:
   		break

file对象常用的参数：

1.2.3、序列化和反序列化

• 序列化：将数据结构或对象转换成二进制串的过程

• 反序列化：将在序列化过程中所生成的二进制串转换成数据结构或对象的过程

Python的pickle模块实现了基本的数据序列和反序列化。

# 序列化（将对象obj保存至文件中）。
pickle.dump(obj,file,[,protocal])

# 反序列化（从文件中恢复对象，并重构为原来的Python对象）
x=pickle.load(file)

序列化实例:

import pickle

# 使用pickle模块将数据对象保存到文件

# 字符串
data0 = "hello world"
# 列表
data1 = list(range(20))[1::2]
# 元组
data2 = ("x","y","z")
# 字典
data3 = {"a":data0,"b":data1,"c":data2}

print(data0)
print(data1)
print(data2)
print(data3)

output = open("data.pk1","wb")

# 使用默认的protocal
pickle.dump(data0,output)
pickle.dump(data1,output)
pickle.dump(data2,output)
pickle.dump(data3,output)
output.close()

反序列化演示：

import pickle

# 使用pickle模块从文件中重构Python对象
pkl_file = open("data.pk1","rb")

data0 = pickle.load(pkl_file)
data1 = pickle.load(pkl_file)
data2 = pickle.load(pkl_file)
data3 = pickle.load(pkl_file)

print(data0)
print(data1)
print(data2)
print(data3)

pkl_file.close()

1.3、读写配置文件

配置文件是供程序运行时读取配置信息的文件，用于将配置信息与程序分离。

Python内置的配置文件解析器模块configparser类来解析基本的配置文件。

常见的pip配置文件如下：

[global]
index-url = https://pypi.doubanio.com/simple
trusted-host = pypi.doubanio.com

读取配置文件的信息的实例：

import configparser
config = configparser.ConfigParser()    # 实例化ConfigParser类
config.read(r"lesson/pip.ini")
print("遍历配置信息")
for section in config.sections():
    print(f"section is [{section}")
    for key in config[section]:
        print(f"key is [{key}],value is [{config[section][key]}]")  # 打印键和值

print("通过键获取相对应的值：")
print(f"index-url is [{config['global']['index-url']}]")
print(f"trusted-host is [{config['global']['trusted-host']}]")

# 上述代码通过实例化ConfigParser类读取配置文件，遍历配置文件中的section信息及键值信息，通过索引获取值信息。

将相关信息写入配置文件实例：

import configparser
config = configparser.ConfigParser()

config["DEFAULT"] = {
    "ServerAliveInterval":"45",
    "Compression":"yes",
    "CompressionLevel":"9"
}

config["bitbucket.org"] = {}
config["bitbucket.org"]["User"] = "hg"

config["topsecret.server.com"] = {}
topsecret = config["topsecret.server.com"]
topsecret["Port"] = "50022"
topsecret["ForwardX11"] = "no"

config["DEFAULT"]["ForwaldX11"] = "yes"

with open("example.ini","w") as configfile:
    config.write(configfile)
with open("example.ini","r") as f:
    print(f.read())

configparser模块的接口非常直接、明确。注意以下几点：

• section名称是区分大小写的
• section下的键值对中键是不区分大小写的，config["bitbucket.org"]在写入时会统一变成小写user保存在文件中
• section下的键值对中的值是不区分类型的，都是字符串，具体使用时需要转换成想要的数据类型
• section的名称是[DEFAULT]时，其他section的键值会继承[DEFAULT]的键值信息

1.4、解析XML文件*

XML的全称是eXtensible Markup Language，意为可扩展的标记语言，是一种用于标记电子文件使其具有结构性的标记语言。被设计用来传输和存储数据，例如：

<note>
    <to>George</to>
    <form>John</form>
    <heading>Reminder</heading>
    <body>Don't forget the meeting!</body>
</note>

Python有三种方法解析XML：SAX、DOM、ElementTree。

1.4.1、SAX（simple API for XML）

SAX是一种基于事件驱动的API，使用时涉及两个部分：解析器和事件处理器。解析器负责读取XML文件，并向事件处理器发送相应的事件。事件处理器对相应的事件做出相应，对数据做出处理。

• 创建一个新的XMLReader对象，parser_list是可选参数，是解析器列表xml.sax.make_parser([parser_list])

• 自定义事件处理器，继承ContentHandler类

ContenHandler类的方法：

语法：

xml.sax.parse(xmlfile,contenthandler[,errorhandler])
# 参数说明：
xmlstring：xml字符串
contenthandler：必须是一个ContentHandler的对象
errorhandler：如果指定该参数，则errorhandler必须是一个SAXErrorHandler对象

解析XML的例子：首先有一个example.xml：

<?xml version="1.0" encoding="UTF-8"?>
<library>
    <book id="1">
        <title>Book Title One</title>
        <author>Author A</author>
        <year>2021</year>
    </book>
    <book id="2">
        <title>Book Title Two</title>
        <author>Author B</author>
        <year>2022</year>
    </book>
    <!-- More books can be added here -->
</library>

解析代码实例：

import xml.sax

class MySAXHandler(xml.sax.ContentHandler):
    def __init__(self):
        self.current_data = ""

    def startElement(self, tag, attributes):
        print(f"Start Element: {tag}")

    def endElement(self, tag):
        print(f"End Element: {tag}")
        if tag == "title":
            print(f"Title: {self.current_data}")
            self.current_data = ""
        elif tag == "author":
            print(f"Author: {self.current_data}")
            self.current_data = ""
        elif tag == "year":
            print(f"Year: {self.current_data}")
            self.current_data = ""

    def characters(self, content):
        self.current_data += content.strip()

# 创建SAX解析器
parser = xml.sax.make_parser()
# 将我们自定义的处理类注册到解析器
parser.setContentHandler(MySAXHandler())

# 解析XML文档
parser.parse("example.xml")

SAX用事件驱动模型，通过在解析XML的过程中触发一个个的事件并调用用户定义的回调函数来处理XML文件，一次处理一个标签，无需事先全部读取整个XML文档，处理效率较高。适用场景：

• 对大型文件进行处理
• 只需要文件的部分内容，或者只需从文件中得到特定信息

• 想建立自己的对象模型时

1.4.2、DOM（Document Object Model）

文件对象模型是W3C组织推荐的处理可扩展置标语言的标准编程接口。

实例：使用xml.dom.minidom解析xml文件：

import xml.dom.minidom
from xml.dom.minidom import parse

# 假设我们有一个XML字符串，也可以从文件中读取
xml_string = """
<?xml version="1.0" encoding="UTF-8"?>
<library>
    <book id="1">
        <title>Book Title One</title>
        <author>Author A</author>
        <year>2021</year>
    </book>
    <book id="2">
        <title>Book Title Two</title>
        <author>Author B</author>
        <year>2022</year>
    </book>
</library>
"""

# 解析XML字符串
dom = xml.dom.minidom.parseString(xml_string)
# 文档方式：
dom = xml.dom.minidom.parse("example.xml")

# 获取<library>元素
library = dom.documentElement

# 遍历所有的<book>元素
for book in library.getElementsByTagName('book'):
    # 获取<book>的id属性
    book_id = book.getAttribute('id')
    # 获取并打印<title>元素的文本
    title = book.getElementsByTagName('title')[0].firstChild.data
    # 获取并打印<author>元素的文本
    author = book.getElementsByTagName('author')[0].firstChild.data
    # 获取并打印<year>元素的文本
    year = book.getElementsByTagName('year')[0].firstChild.data

    # 打印信息
    print(f"Book ID: {book_id}")
    print(f"Title: {title}")
    print(f"Author: {author}")
    print(f"Year: {year}")
    print()  # 打印空行以便区分每本书的信息
    
# 代码使用minidom解析器打开文档，使用getElementsByTagName方法获取所有标签并遍历子标签，逻辑上比SAX直观

1.4.3、ElementTree

ElementTree将XML数据在内存中解析成树，通过树来操作XML。

import xml.etree.ElementTree as ET

tree = ET.parse("example.xml")
root = tree.getroot()
print(f"这是一个早餐菜单\n{root.attrib['year']}")

for child in root:
	print("Book:",child[0].text)
	print("Title:",child[1].text)

二、系统信息监控

Python获取系统信息的模块是psutil（process and system utilities）。

2.1、监控CPU信息

import psutil

psutil.cpu_times()				# 获取CPU（逻辑CPU的平均）占用时间的详细信息
psutil.cpu_times(percpu=True)   # 获取每个CPU占用时间的详细信息
psutil.cpu_count()      		# CPU逻辑数量
psutil.cpu_count(logical=False) # CPU物理数量
psutil.cpu_percent()        	# CPU占比
psutil.cpu_percent()        	# 每个CPU的占比

2.2、监控内存信息

psutil.virtual_memory()     # 数值以字节为单位显示，自行转换

2.3、监控磁盘信息

psutil.disk_partitions()
psutil.disk_usage('/')      # 磁盘使用情况

2.4、监控网络信息

psutil.net_io_counters()    # 获取网络读写字节数/包的个数
psutil.net_if_addrs()       # 获取网络接口信息
psutil.net_if_stats()       # 获取网络接口状态
psutil.net_connections()    # 获取当前网络连接信息

2.5、获取进程信息

for pid in psutil.pids():	# 获取所有进程的pid
    print(pid,end=',')
for proc in psutil.process_iter(attrs=['pid','name','username']):
    if proc.info['name'].startswith('Wechat'):	# 查找微信程序的相关信息
        print(proc.info)
        
# psutil.process_iter返回的是一个可迭代对象，每个元素的info是一个字典，通过其可以获取我们关心的信息：
psutil.Process(12476).cpu_times()   # 获取CPU占用
psutil.Process(12476).memory_info() # 获取内存占用，rss就是实际占用的内存
psutil.Process(12476).num_threads() # 获取线程数
psutil.Process(12476).memory_percent()  # 获取内存占比

2.6、实例：常见的实用方法

import import os
import psutil
import signal

# 按名称查找进程相关信息 1
def find_procs_by_name1(name):
    "Return a list of processes matching 'name'."
    ls = []
    for p in psutil.process_iter(attrs=['name']):
        if p.info['name'] == name:
            ls.append(p)
    return ls

# 按名称查找进程相关信息 2
def find_procs_by_name2(name):
    ls = []
    for p in psutil.process_iter(attrs=["name","exe","cmdline"]):
        if name == p.info['name'] or \
            p.info['exe'] and os.path.basename(p.info['exe']) == name or \
            p.info['cmdline'] and p.info['cmdline'][0] == name:
            ls.append(p)
    return ls

# 杀掉进程树
def kill_proc_tree(pid,sig=signal.SIGTERM,include_parent=True,timeout=None,on_terminate=None):
    if pid == os.getpid():
        raise RuntimeError("I refuse to kill myself")
    parent = psutil.Process(pid)
    children = parent.children(recursive=True)
    if include_parent:
        children.append(parent)
    for p in children:
        p.send_signal(sig)
    gone,alive = psutil.wait_procs(children,timeout=timeout,callback=on_terminate)
    return (gone,alive)

# 杀掉子进程
def reap_children(tiemout=3):
    def on_terminate(proc):
        print("process {} terminated with exit code {}".format(proc,proc.returncode))
        procs = psutil.Process().children()
        # send SIGTERM
        for p in procs:
            p.terminate()
        gone,alive = psutil.wait_procs(procs,timeout=tiemout,callback=on_terminate)
        if alive:
            # send SIGKILL
            for p in alive:
                print("process {} survived SIGTERM;trying SIGKILL" % p)
                p.kill()
            gone,alive = psutil.wait_procs(alive,timeout=tiemout,callback=on_terminate)
            if alive:
                # give up
                for p in alive:
                    print("process {} survived SIGKILL;giving up" % p)

三、文件系统监控

如某个目录被删除，或者某个文件被修改、移动、删除时需要执行一定的操作或发出警报。Python使用watchdog库来实现文件系统监控，其原理是通过操作系统的事件触发。

3.1、实例

import time
from watchdog.events import *
from watchdog.observers import Observer

class FileEventHandler(FileSystemEventHandler):
    def __init__(self):
        FileSystemEventHandler.__init__(self)

    def on_moved(self, event):
        now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        if event.is_directory:
            print(f"{now}文件夹由{event.src_path}移动至{event.dest_path}")
        else:
            print(f"{now}文件由{event.src_path}移动至{event.dest_path}")

    def on_created(self, event):
        now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        if event.is_directory:
            print(f"{now}文件夹{event.src_path} 创建")
        else:
            print(f"{now}文件{event.src_path} 创建")

    def on_deleted(self, event):
        now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        if event.is_directory:
            print(f"{now}文件夹{event.src_path} 删除")
        else:
            print(f"{now}文件{event.src_path} 删除")

    def on_modified(self, event):
        now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        if event.is_directory:
            print(f"{now}文件夹{event.src_path} 修改")
        else:
            print(f"{now}文件{event.src_path} 修改")

if __name__ == '__main__':
    observer = Observer()
    path = r"d:/test"
    event_handler = FileEventHandler()
    observer.schedule(event_handler, path, True)  # True表示递归子目录
    print(f"监控目录{path}")
    observer.start()
    observer.join()

适用场景：

• 监控文件系统中文件或目录的增、删、改情况

• 当特定的文件被创建、删除、修改、移动时执行相应的任务

四、执行外部命令subprocess

subprocess是Python自带的模块，主要用来取代一些旧的模块或方法，如os.system等。适用subprocess模块更方便地执行操作系统支持的命令，可与其他应用程序结合适用

4.1、subprocess.run()方法*

这是官方推荐使用的方法，查看其原型：

subprocess.run(args,*,stdin=None,input=None,stdout=None,stderr=None,shell=False,cwd=None,timeout=None,check=False,encoding=None,errors=None)

该函数返回一个CompletedProcess类的实例，其常用参数：

• args代表需要在操作系统中执行的命令。可以是字符串形式（要去shell=True），也可以是列表list类型
• *代表可变参数，一般是列或字典形式
• stdin、stdout、stderr指定了可执行程序的标准输入、标准输出、标准错误文件句柄。
• shell代表程序是否需要在shell上执行
• check设置为True表示检车命令的返回值，当返回值为非0时，就抛出CalledProcessError异常
• timeout设置超时时间，如果超时，则强制kill掉子进程

在Linux系统中操作，实例自行查找

4.2、Popen类

Popen类的构造函数：

class subprocess.Popen(args,
	bufsize=-1, 	# 0无缓冲、1行缓冲，其他正值，缓冲区大小，负值，默认系统缓冲
	executable=None,# 一般不用
    stdin=None, 	# None没有任何重定向，继承父进程。PIPE创建管道...
    stdout=None,
    stderr=None,
    preexec_fn=None, 	# 钩子函数
    close_fds=True,		
    shell=False, 
    cwd=None, 
    env=None, 
    universal_newlines=None,
    startupinfo=None, 
    creationflags=0)

使用方法：

subprocess.Popen(["getit","abc.txt"])
subprocess.Popen("getit abc.txt")

Popen类的对象还有其他实用方法：

4.3、其他方法

• subprocess.call(*popenargs，**kwargs)：call方法调用Popen()执行程序，并等待它完成
• subprocess.check_call(*popenargs，**kwargs)：调用前面的call()，如果返回值非0，则抛出异常
• subprocess.check_output(*popenargs，**kwargs)：调用Popen()执行程序，并返回其标准输出