Python技能脚本实战：从零构建自动化任务处理系统

6次阅读

共计 4053 个字符，预计需要花费 11 分钟才能阅读完成。

刚开始学习 Python 脚本编写时，我经常遇到这些问题，相信你也深有体会：

硬编码路径：在脚本里直接写死文件路径，换个环境就跑不通
缺乏错误处理：程序遇到异常直接崩溃，没有任何友好提示
打印调试信息 ：到处用 print() 输出日志，调试完又得手动删除
参数不灵活：每次修改参数都要改源代码，非常不方便
配置混杂：各种设置参数和业务代码混在一起，难以维护

Python 标准库中的 argparse 模块可以轻松处理命令行参数。相比直接读取 sys.argv，它能自动生成帮助信息并支持参数验证。

import argparse

parser = argparse.ArgumentParser(description='文件处理脚本')
parser.add_argument('-i', '--input', required=True, help='输入目录路径')
parser.add_argument('-o', '--output', help='输出目录路径')
parser.add_argument('-v', '--verbose', action='store_true', help='显示详细日志')
args = parser.parse_args()

logging 模块提供了灵活的日志记录功能，可以设置不同级别（DEBUG、INFO、WARNING 等）并输出到不同位置。

import logging

logging.basicConfig(
    level=logging.DEBUG if args.verbose else logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.FileHandler('process.log'),
        logging.StreamHandler()]
)

合理的异常处理能让脚本在遇到问题时优雅降级，而不是直接崩溃。

try:
    with open('config.ini', 'r') as f:
        config = f.read()
except FileNotFoundError:
    logging.error('配置文件不存在')
    sys.exit(1)
except PermissionError:
    logging.error('没有权限读取配置文件')
    sys.exit(1)

将配置参数分离到配置文件中，避免硬编码。

config.ini 示例：

[DEFAULT]
max_files = 100
exclude_extensions = .tmp, .bak

读取配置的代码：

import configparser

config = configparser.ConfigParser()
config.read('config.ini')
max_files = config.getint('DEFAULT', 'max_files')

下面是一个整合了上述所有技术的完整脚本示例：

#!/usr/bin/env python3
"""
文件批量处理脚本
功能：扫描指定目录，按规则处理文件
"""
import os
import sys
import argparse
import logging
import configparser
from datetime import datetime

# 初始化参数解析
def init_args():
    parser = argparse.ArgumentParser(description='文件批量处理工具')
    parser.add_argument('-i', '--input', required=True, help='输入目录路径')
    parser.add_argument('-o', '--output', help='输出目录路径')
    parser.add_argument('-c', '--config', default='config.ini', help='配置文件路径')
    parser.add_argument('-v', '--verbose', action='store_true', help='详细日志模式')
    return parser.parse_args()

# 初始化日志配置
def init_logging(verbose=False):
    level = logging.DEBUG if verbose else logging.INFO
    logging.basicConfig(
        level=level,
        format='%(asctime)s - %(levelname)s - %(message)s',
        handlers=[logging.StreamHandler()]
    )

# 加载配置文件
def load_config(config_path):
    config = configparser.ConfigParser()
    try:
        config.read(config_path)
        return config
    except Exception as e:
        logging.error(f'加载配置文件失败: {e}')
        sys.exit(1)

# 主处理函数
def process_files(input_dir, output_dir, config):
    try:
        if not os.path.exists(input_dir):
            raise FileNotFoundError(f'输入目录不存在: {input_dir}')

        if output_dir and not os.path.exists(output_dir):
            os.makedirs(output_dir)

        max_files = config.getint('DEFAULT', 'max_files', fallback=100)
        exclude = config.get('DEFAULT', 'exclude_extensions', fallback='').split(',')

        processed = 0
        for root, _, files in os.walk(input_dir):
            for file in files:
                if any(file.endswith(ext.strip()) for ext in exclude):
                    continue

                src = os.path.join(root, file)
                if output_dir:
                    dst = os.path.join(output_dir, file)
                    os.rename(src, dst)

                processed += 1
                if processed >= max_files:
                    logging.warning(f'达到最大处理文件数限制: {max_files}')
                    return

    except Exception as e:
        logging.error(f'处理文件时出错: {e}', exc_info=True)
        sys.exit(1)

if __name__ == '__main__':
    args = init_args()
    init_logging(args.verbose)
    config = load_config(args.config)
    process_files(args.input, args.output, config)

使用 PyInstaller 可以将脚本打包成独立的可执行文件：

安装 PyInstaller：pip install pyinstaller
执行打包命令：pyinstaller --onefile your_script.py
生成的可执行文件在 dist 目录下

使用 unittest 模块为脚本添加测试：

import unittest
import tempfile
import shutil
from your_script import process_files

class TestFileProcessing(unittest.TestCase):
    def setUp(self):
        self.test_dir = tempfile.mkdtemp()
        self.input_dir = os.path.join(self.test_dir, 'input')
        self.output_dir = os.path.join(self.test_dir, 'output')
        os.makedirs(self.input_dir)

        # 创建测试文件
        with open(os.path.join(self.input_dir, 'test.txt'), 'w') as f:
            f.write('test content')

    def tearDown(self):
        shutil.rmtree(self.test_dir)

    def test_file_processing(self):
        process_files(self.input_dir, self.output_dir, {})
        self.assertTrue(os.path.exists(os.path.join(self.output_dir, 'test.txt')))