Python（十三）脚本编程与系统管理

Posted 2022-11-05 HT . WANG

1.通过重定向/管道/文件接受输入

Python内置的 fileinput 模块将命令行的输出通过管道传递给该脚本、 重定向文件到该脚本，或在命令行中传递一个文件名或文件名列表给该脚本

#!/usr/bin/env python3
#-----------------
#filein.py
#-----------------
import fileinput

with fileinput.input() as f_input:
    for line in f_input:
        print(line, end='')


$ ls | ./filein.py          # 打印当前目录下文件列表进行输出显示
$ ./filein.py /etc/passwd   # 读取passwd文件内容显示
$ ./filein.py < /etc/passwd # 读取passwd文件内容重定向到终端窗口显示

2.终止程序并给出错误信息

想向标准错误打印一条消息并返回某个非零状态码来终止程序运行

import sys
sys.stderr.write('It failed!\\n') #打印错误信息
raise SystemExit(1) #返回状态码退出程序

直接将消息作为参数传给 SystemExit() ，那么你可以省略其他步骤

raise SystemExit('It failed!')

抛出一个 SystemExit 异常，使用错误消息作为参数，它会将消息在 sys.stderr 中打印，然后程序以状态码1退出

3.解析命令行选项

如何通过程序能够解析命令行选项

import argparse
parser = argparse.ArgumentParser(description='Search some files')

parser.add_argument(dest='filenames',metavar='filename', nargs='*')

parser.add_argument('-p', '--pat',metavar='pattern', required=True,
                    dest='patterns', action='append',
                    help='text pattern to search for') #允许某个参数重复出现多次，并将它们追加到一个列表中去

parser.add_argument('-v', dest='verbose', action='store_true',
                    help='verbose mode') #根据参数是否存在来设置一个 Boolean 标志

parser.add_argument('-o', dest='outfile', action='store',
                    help='output file') #接受一个单独值并将其存储为一个字符串

parser.add_argument('--speed', dest='speed', action='store',
                    choices='slow','fast', default='slow',
                    help='search speed')#接受一个值将其和可能的选择值做比较，以检测其合法性

args = parser.parse_args()

# Output the collected arguments
print(args.filenames)
print(args.patterns)
print(args.verbose)
print(args.outfile)


bash % python3 search.py -v -p spam --pat=eggs foo.txt bar.txt
filenames = ['foo.txt', 'bar.txt']
patterns  = ['spam', 'eggs']
verbose   = True
outfile   = None
speed     = slow

bash % python3 search.py -v -p spam --pat=eggs foo.txt bar.txt -o results
filenames = ['foo.txt', 'bar.txt']
patterns  = ['spam', 'eggs']
verbose   = True
outfile   = results
speed     = slow

bash % python3 search.py -v -p spam --pat=eggs foo.txt bar.txt -o results --speed=fast
filenames = ['foo.txt', 'bar.txt']
patterns  = ['spam', 'eggs']
verbose   = True
outfile   = results
speed     = fast
print(args.speed)

4.运行时弹出密码输入提示

写了个脚本，运行时需要一个密码。此脚本是交互式的，因此不能将密码在脚本中硬编码， 而是需要弹出一个密码输入提示，让用户自己输入。

import getpass

user = getpass.getuser() #该函数不会弹出用户名的输入提示。 它会根据该用户的shell环境或者会依据本地系统的密码库（支持 pwd 模块的平台）来使用当前用户的登录名，
passwd = getpass.getpass()

if svc_login(user, passwd):    # 处理密码的函数 处理过程自定义（解码 匹配）
   print('YES!')
else:
   print('NO!')

5.获取终端的大小

需要知道当前终端的大小以便正确的格式化输出。

>>> import os
>>> sz = os.get_terminal_size()
>>> sz
#结果显示 终端的行数 和 列数
os.terminal_size(columns=80, lines=24)

>>> sz.columns
80
>>> sz.lines
24
>>>

6.执行外部命令并获取它的输出

想执行一个外部命令并以Python字符串的形式获取执行结果

import subprocess
out_bytes = subprocess.check_output(['netstat','-a']) #将执行结果以一个字节字符串的形式返回

out_text = out_bytes.decode('utf-8') #将执行结果以文本形式返回


out_bytes = subprocess.check_output(['cmd','arg1','arg2'],stderr=subprocess.STDOUT) #同时收集标准输出和错误输出，使用 stderr 参数

7.复制或者移动文件和目录

import shutil

# 复制源文件到目标路径
shutil.copy(src, dst)

# 复制过程中保留源文件或目录的路径即不跟随源文件中的符号链接
shutil.copy2(src, dst)

# 递归复制 将指定目录下的所有文件与子目录一并处理
shutil.copytree(src, dst)

# 移动
shutil.move(src, dst)

注意：

使用 copytree() 复制文件夹 。在复制过程中，函数可能会碰到损坏的符号链接，因为权限无法访问文件的问题等等

如果提供关键字参数 ignore_dangling_symlinks=True ， 这时候 copytree() 会忽略掉无效符号链接

8.创建和解压归档文件

>>> import shutil
>>> shutil.unpack_archive('Python-3.3.0.tgz') #解压

>>> shutil.make_archive('py33','zip','Python-3.3.0') #压缩文件名py33  zip格式  需要被压缩文件夹python-3.3.0
'/Users/beazley/Downloads/py33.zip'
>>>

9.通过文件名查找文件

查找特定的文件名并返回所有符合条件的文件全路径

#!/usr/bin/env python3.3
import os

def findfile(start, name):指定初始查找目录 以及查找文件名字作为位置参数
    for relpath, dirs, files in os.walk(start):
        if name in files:
            full_path = os.path.join(start, relpath, name) #根据查找到信息进行目录拼接
            print(os.path.normpath(os.path.abspath(full_path)))

if __name__ == '__main__':
    findfile(sys.argv[1], sys.argv[2])

os.walk() 方法为我们遍历目录树， 每次进入一个目录，它会返回一个三元组，包含：

• 相对于查找目录的相对路径
• 该目录下的目录名列表
• 对应目录下面的文件名列表

10.读取配置文件

有如下配置文件：

; config.ini
; Sample configuration file

[installation]
library=%(prefix)s/lib
include=%(prefix)s/include
bin=%(prefix)s/bin
prefix=/usr/local

# Setting related to debug configuration
[debug]
log_errors=true
show_warnings=False

[server]
port: 8080
nworkers: 32
pid-file=/tmp/spam.pid
root=/www/root
signature:
    =================================
    Brought to you by the Python Cookbook
    =================================

configparser 模块能被用来读取配置文件

>>> from configparser import ConfigParser
>>> cfg = ConfigParser()
>>> cfg.read('config.ini')
['config.ini']
>>> cfg.sections() #获取各章节头部
['installation', 'debug', 'server'] 
>>> cfg.get('installation','library') #获取对应章节头部字段下对应字段值
'/usr/local/lib'
>>> cfg.getboolean('debug','log_errors')

True
>>> cfg.getint('server','port') #getint获取int类型数值
8080
>>> cfg.getint('server','nworkers')
32
>>> print(cfg.get('server','signature'))

\\=================================
Brought to you by the Python Cookbook
\\=================================
>>>

注意：配置文件中的名字是不区分大小写的 因此在读取配置文件信息时 对应字段可以不用区分大小写

11.给简单脚本增加日志功能

import logging

def main():
    # Configure the logging system
    logging.basicConfig(
        filename='app.log',
        level=logging.ERROR #设置错误等级的最低等级 限制错误输出
    )

    # Variables (to make the calls that follow work)
    hostname = 'www.python.org'
    item = 'spam'
    filename = 'data.csv'
    mode = 'r'

    # Example logging calls (insert into your program)
    logging.critical('Host %s unknown', hostname)#以降序方式表示不同错误的严重级别
    logging.error("Couldn't find %r", item)
    logging.warning('Feature is deprecated')
    logging.info('Opening file %r, mode=%r', filename, mode)
    logging.debug('Got here')

if __name__ == '__main__':
    main()

#结果---------------
CRITICAL:root:Host www.python.org unknown
ERROR:root:Could not find 'spam'

改变输出等级

import logging

def main():
    # Configure the logging system
    logging.basicConfig(
     filename='app.log',
     level=logging.WARNING,#更改错误等级的最低等级为WARNING 限制错误输出
     format='%(levelname)s:%(asctime)s:%(message)s') 
    )

    # Variables (to make the calls that follow work)
    hostname = 'www.python.org'
    item = 'spam'
    filename = 'data.csv'
    mode = 'r'

    # Example logging calls (insert into your program)
    logging.critical('Host %s unknown', hostname)#以降序方式表示不同错误的严重级别
    logging.error("Couldn't find %r", item)
    logging.warning('Feature is deprecated')
    logging.info('Opening file %r, mode=%r', filename, mode)
    logging.debug('Got here')

if __name__ == '__main__':
    main()

#结果---------------
CRITICAL:2012-11-20 12:27:13,595:Host www.python.org unknown
ERROR:2012-11-20 12:27:13,595:Could not find 'spam'
WARNING:2012-11-20 12:27:13,595:Feature is deprecated

12.给函数库增加日志功能

想给某个函数库增加日志功能，但是又不能影响到那些不使用日志功能的程序

import logging
log = logging.getLogger(__name__)
log.addHandler(logging.NullHandler())

# Example function (for testing)
def func():
    log.critical('A Critical Error!')
    log.debug('A debug message')

>>> import logging
>>> logging.basicConfig() #配置日志系统
>>> somelib.func() #调用打印日志信息
CRITICAL:somelib:A Critical Error!
>>>

13.实现一个计时器

记录程序执行多个任务所花费的时间

import time

class Timer:
    def __init__(self, func=time.perf_counter):
        self.elapsed = 0.0
        self._func = func
        self._start = None

    def start(self):
        if self._start is not None:
            raise RuntimeError('Already started')
        self._start = self._func()

    def stop(self):
        if self._start is None:
            raise RuntimeError('Not started')
        end = self._func()
        self.elapsed += end - self._start  #在 elapsed 属性中记录整个消耗时间
        self._start = None

    def reset(self):
        self.elapsed = 0.0

    @property
    def running(self):
        return self._start is not None

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, *args):
        self.stop()

def countdown(n):
    while n > 0:
        n -= 1

# Use 1: Explicit start/stop
t = Timer() #构建定时器对象
t.start() #定时器启动
countdown(1000000)
t.stop()  #定时器关闭
print(t.elapsed)

# Use 2: As a context manager
with t: #开启会话 会自动调用enter和exit接口 实现 进行会话后调用开启函数 执行完成离开会话后调用关闭函数
    countdown(1000000)

print(t.elapsed)

with Timer() as t2:
    countdown(1000000)
print(t2.elapsed)

注意：述代码中由 Timer 类记录的时间是钟表时间，并包含了所有休眠时间。 如果你只想计算该进程所花费的CPU时间，应该使用 time.process_time() 来代替

t = Timer(time.process_time)
with t:
    countdown(1000000)
print(t.elapsed)

14.限制内存和CPU的使用量

对在Unix系统上面运行的程序设置内存或CPU的使用限制

import signal
import resource
import os

def time_exceeded(signo, frame):
    print("Time's up!")
    raise SystemExit(1)

def set_max_runtime(seconds): #限制运行时间
    soft, hard = resource.getrlimit(resource.RLIMIT_CPU)
    resource.setrlimit(resource.RLIMIT_CPU, (seconds, hard))
    signal.signal(signal.SIGXCPU, time_exceeded)

def limit_memory(maxsize): #限制内存 设置了内存限制后，程序运行到没有多余内存时会抛出 MemoryError 异常
    soft, hard = resource.getrlimit(resource.RLIMIT_AS)
    resource.setrlimit(resource.RLIMIT_AS, (maxsize, hard))

if __name__ == '__main__':
    set_max_runtime(15)
    while True:
        pass

15.启动一个WEB浏览器

通过脚本启动浏览器并打开指定的URL网页

>>> import webbrowser
>>> webbrowser.open('http://www.python.org') #启动一个浏览器，并且与平台无关
True
>>>

>>> # Open the page in a new browser window
>>> webbrowser.open_new('http://www.python.org') #打开一个新的浏览器窗口
True
>>>

>>> # Open the page in a new browser tab
>>> webbrowser.open_new_tab('http://www.python.org') #打开一个新标签页
>>>

>>> c = webbrowser.get('firefox') #通过指定浏览器平台 打开网页
>>> c.open('http://www.python.org')
True
>>> c.open_new_tab('http://docs.python.org')
True
>>>