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 sysprint(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。

常见的文件操作方法：

名称	功能
f.read()	读取所有内容，光标移动到文件末尾
f.readline()	读取一行内容，光标移动到第二行首部
f.readlines()	读取每一行内容，存于列表中
f.write('1111\n222\n')	针对文本模式的写，需要自己写换行符
f.write('1111\n222\n'.encode('utf-8'))	针对b模式的写，需要自己写换行符
f.writelines(['333\n','444\n'])	文件模式
f.write([bytes('333\n',encoding='utf-8'),'444\n'.encode('utf-8')])	b模式
f.readable()	文件是否可读
f.writable()	文件是否可写
f.closed	文件是否关闭
f.encoding	如果文件打开模式为b，则没有该属性
f.flush()	立刻将文件内容从内存刷到硬盘

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

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

# 基于seek实现类似Linux命令tail-f的功能import timewith 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')

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

通过read(size)增加参数，指定读取的字节数。
```
while True:block = f.read(1024)if not block:break
```

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

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

file对象常用的参数：

函数	功能
file.close()	关闭文件。关闭后不能再进行读写操作
file.flush()	刷新文件内部缓冲，直接把内部缓冲区的数据立刻写入文件，而不是被动等待输出缓冲区写入
file.fileno()	返回一个整型的文件描述符，可以用在如os模块的read方法等一些底层操作上
file.isatty()	如果文件连接到一个终端设备，则返回True，否则False
file.next()	返回文件下一行
file.read([size])	从文件读取指定的字节数，如果为给定或为负，则读取所有
file.readline([size])	读取整行，包括"\n"字符
file.readlines([sizeint])	读取所有行并返回列表，若给定sizeint>0，则返回总和为sizeint字节的行，实际读取值的可能比sizeint大，因为要填充缓冲区
file.seek(offset[,whence])	设置文件当前位置
file.tell()	返回文件当前位置
file.truncate([size])	根据size参数截取文件，size参数可选
file.write(str)	将字符串写入文件，没有返回值
file.writelines(sequence)	向文件写入一个序列字符串列表，如果需要换行，则加入每行的换行符

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类的方法：

名称	功能
characters(content)	从行开始，遇到标签之前，存在字符，content的值为这些字符串从下一个标签，遇到下一个标签之前，存在字符，content的值为这些字符串
startDocument()	文档启动时调用
endDocument()	解析器到达文档结尾时调用
startElement(name,attrs)	遇到XML开始标签时调用，name是标签的名字，attrs是标签属性值字典
endElement(name)	遇到XML结束标签时调用

语法：

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.saxclass 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 ETtree = 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 psutilpsutil.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():	# 获取所有进程的pidprint(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 SIGTERMfor p in procs:p.terminate()gone,alive = psutil.wait_procs(procs,timeout=tiemout,callback=on_terminate)if alive:# send SIGKILLfor 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 upfor p in alive:print("process {} survived SIGKILL;giving up" % p)

三、文件系统监控

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

3.1、实例

import time
from watchdog.events import *
from watchdog.observers import Observerclass 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类的对象还有其他实用方法：

名称	功能
poll()	检查是否结束，设置返回值
wait()	等待结束，设置返回值
communicate()	参数是标准输入，返回标准输出和标准出错
send_signal()	发送信号（主要在unix下有用）
terminate()	终止进程，unix对应的SIGTERM信号,windows下调用api函数TerminateProcess()
kill()	杀死进程（unix对应SIGKILL信号），windows同上
stdin stdout stderr	参数中指定PIPE时，可以使用
pid	进程ID
returncode	进程返回值