54. 日志与子进程

程序运行时需要记录状态、排查问题；有时还需要调用外部程序（比如系统命令、其他脚本）。logging 和 subprocess 就是解决这两个需求的标准工具。

logging：生产级日志系统

40 章初步介绍了 logging，这里深入它的配置和高级用法。

基础回顾

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    filename='app.log',
    encoding='utf-8'
)

logging.info('程序启动')
logging.warning('磁盘空间不足')

日志级别

从低到高：

logging.debug('调试信息')     # 开发排错用
logging.info('一般信息')       # 正常流程记录
logging.warning('警告')        # 需要注意但不致命
logging.error('错误')          # 功能受损
logging.critical('严重错误')   # 程序可能崩溃

设置 level=logging.INFO 后，debug 级别的日志不会输出。生产环境通常设为 INFO 或 WARNING。

同时输出到文件和屏幕

basicConfig 只能配置一个目标。如果需要「文件 + 屏幕」双输出，用 Handler：

import logging

# 创建 logger
logger = logging.getLogger('myapp')
logger.setLevel(logging.DEBUG)

# 文件处理器
file_handler = logging.FileHandler('app.log', encoding='utf-8')
file_handler.setLevel(logging.INFO)

# 控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)

# 设置格式
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)

# 添加处理器
logger.addHandler(file_handler)
logger.addHandler(console_handler)

logger.debug('这条只在屏幕显示')
logger.info('这条屏幕和文件都有')

日志轮转

日志文件无限增长会撑爆磁盘。用 RotatingFileHandler 按大小切割：

from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler(
    'app.log',
    maxBytes=10 * 1024 * 1024,  # 10 MB
    backupCount=5,               # 保留 5 个备份
    encoding='utf-8'
)

或者按日期切割，用 TimedRotatingFileHandler：

from logging.handlers import TimedRotatingFileHandler

handler = TimedRotatingFileHandler(
    'app.log',
    when='midnight',   # 每天午夜切割
    interval=1,
    backupCount=7,     # 保留 7 天
    encoding='utf-8'
)

用 Logger 分层

大型项目里，不同模块用不同的 logger，便于控制输出：

# database.py
logger = logging.getLogger('myapp.database')

# api.py
logger = logging.getLogger('myapp.api')

配置时可以分别控制：

logging.getLogger('myapp.database').setLevel(logging.DEBUG)
logging.getLogger('myapp.api').setLevel(logging.WARNING)

subprocess：调用外部命令

Python 里执行系统命令，用 subprocess 模块。它替代了旧版的 os.system() 和 os.popen()。

简单执行

import subprocess

# 执行命令，等待完成
result = subprocess.run(['python', '--version'], capture_output=True, text=True)
print(result.stdout)   # Python 3.x.x
print(result.returncode)  # 0 表示成功

检查返回码

result = subprocess.run(['ls', '不存在的目录'], capture_output=True, text=True)
if result.returncode != 0:
    print(f"错误: {result.stderr}")

# 或者让 subprocess 自动抛异常
result = subprocess.run(
    ['ls', '不存在的目录'],
    capture_output=True,
    text=True,
    check=True  # 非零返回码会抛 CalledProcessError
)

获取实时输出

run() 会等命令完全结束才返回。如果需要实时看到输出，用 Popen：

import subprocess

process = subprocess.Popen(
    ['ping', '127.0.0.1', '-n', '4'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

for line in process.stdout:
    print(line.strip())

process.wait()
print(f"返回码: {process.returncode}")

管道和重定向

# 把命令 A 的输出传给命令 B
p1 = subprocess.Popen(['cat', 'data.txt'], stdout=subprocess.PIPE)
p2 = subprocess.Popen(['grep', 'error'], stdin=p1.stdout, stdout=subprocess.PIPE)
p1.stdout.close()
output = p2.communicate()[0]

或者用 shell 管道（注意 shell=True 有注入风险）：

result = subprocess.run(
    'cat data.txt | grep error',
    shell=True,
    capture_output=True,
    text=True
)

超时控制

try:
    result = subprocess.run(['sleep', '10'], timeout=3)
except subprocess.TimeoutExpired:
    print('命令执行超时')

实战：批量处理视频

结合 subprocess 和 logging，写一个调用 ffmpeg 转码的脚本：

import subprocess
import logging
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('video_converter')

def convert_video(src: Path, dst: Path):
    cmd = [
        'ffmpeg', '-i', str(src),
        '-c:v', 'libx264',
        '-crf', '23',
        str(dst)
    ]
    logger.info(f"开始转换: {src.name}")
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
        logger.info(f"转换完成: {dst.name}")
    except subprocess.CalledProcessError as e:
        logger.error(f"转换失败: {e.stderr}")

for video in Path('raw').glob('*.mov'):
    output = Path('converted') / (video.stem + '.mp4')
    convert_video(video, output)

这个例子展示了 subprocess 调用外部工具、logging 记录流程、以及 pathlib 处理路径的组合。

小结

• logging 是生产环境的标配，用 Handler 实现多目标输出，用 RotatingFileHandler 控制文件大小。
• Logger 按模块分层，便于灵活控制日志级别。
• subprocess.run() 执行命令并等待完成，Popen 适合实时处理输出。
• 命令用列表传入更安全，避免不必要的 shell=True。
• 所有外部命令调用都要设超时，处理非零返回码。

日志让你「看见」程序的运行状态，子进程让你「利用」系统已有的工具。两者结合，Python 脚本就能胜任复杂的自动化任务。

---