Python技能脚本实战：从自动化到生产级代码的避坑指南

5次阅读

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

很多 Python 开发者都经历过这样的阶段：写了一个能跑通的脚本，过几个月再来看，发现完全看不懂当时的逻辑。或者脚本在小数据量下运行良好，一旦处理大规模数据就崩溃。以下是几个最常见的坑：

全局变量滥用：直接在脚本顶层定义变量，导致函数间隐式依赖
裸奔的异常处理 ：只有try...except 但没有具体异常类型捕获
硬编码配置：数据库连接字符串、API 密钥直接写在代码里
缺乏日志记录 ：出错时只能靠print 调试
单线程阻塞：没有利用多核 CPU 处理耗时任务

# 典型问题脚本示例
config = 'mysql://root:123456@localhost'  # 硬编码

def query_data():
    # 直接使用全局变量
    conn = pymysql.connect(config)  # 没有异常处理
    return conn.cursor().execute('SELECT * FROM big_table')

通过对比表格理解架构设计的本质区别：

特性	普通脚本	生产级脚本
配置管理	硬编码	环境变量 / 配置文件
错误处理	简单 try-catch	分级异常捕获 + 日志
接口设计	直接执行	CLI 参数解析 + 入口函数
性能考虑	单线程	多进程 / 异步 IO
可测试性	难以单元测试	模块化 + 依赖注入

import argparse

def create_parser():
    parser = argparse.ArgumentParser(description='文件处理器')
    parser.add_argument('input_dir', help='输入目录路径')
    parser.add_argument('--output', '-o', default='./output', 
                       help='输出目录路径（默认./output）')
    parser.add_argument('--workers', type=int, default=4,
                       help='并行工作进程数（默认 4）')
    return parser

if __name__ == '__main__':
    parser = create_parser()
    args = parser.parse_args()
    print(f'处理目录: {args.input_dir}')

import logging
import sys

def setup_logger(name):
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)

    # 控制台 Handler
    ch = logging.StreamHandler(sys.stdout)
    ch.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
    logger.addHandler(ch)

    # 文件 Handler（按天滚动）fh = logging.handlers.TimedRotatingFileHandler('app.log', when='midnight')
    fh.setLevel(logging.INFO)
    logger.addHandler(fh)
    return logger

logger = setup_logger(__name__)
logger.info('系统初始化完成')

from functools import wraps

def require_role(role):
    """检查用户角色的装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapped(user, *args, **kwargs):
            if user.get('role') != role:
                raise PermissionError(f'需要 {role} 权限')
            return func(user, *args, **kwargs)
        return wrapped
    return decorator

@require_role('admin')
def delete_file(user, filename):
    """只有管理员能删除文件"""
    os.remove(filename)

#!/usr/bin/env python3
"""
文件批量处理器 - 支持并行转换和压缩
Features:
1. 使用 with 语句确保文件描述符释放
2. 多进程池处理 CPU 密集型任务
3. 完整的类型注解和文档字符串
"""
import os
import argparse
from multiprocessing import Pool
from typing import List, Optional

# 类型别名
FilePath = str

def process_file(input_path: FilePath, 
                 output_dir: FilePath) -> Optional[FilePath]:
    """处理单个文件的核心逻辑"""
    try:
        with open(input_path, 'r') as f:
            content = transform_content(f.read())  # 假设的转换函数

        output_path = os.path.join(output_dir, os.path.basename(input_path))
        with open(output_path, 'w') as f:
            f.write(content)

        return output_path
    except Exception as e:
        print(f'处理失败 {input_path}: {str(e)}')
        return None

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('input_dir', help='输入目录')
    parser.add_argument('--output', default='./output')
    parser.add_argument('--workers', type=int, default=4)
    args = parser.parse_args()

    if not os.path.exists(args.output):
        os.makedirs(args.output)

    input_files = [os.path.join(args.input_dir, f) 
        for f in os.listdir(args.input_dir)
        if f.endswith('.txt')
    ]

    with Pool(args.workers) as pool:
        results = pool.starmap(
            process_file,
            [(f, args.output) for f in input_files]
        )

    print(f'处理完成，成功 {len([r for r in results if r])} 个文件')

if __name__ == '__main__':
    main()

安装：pip install memory_profiler

from memory_profiler import profile

@profile
def process_large_file():
    # 会显示每行代码的内存变化
    data = [i**2 for i in range(100000)]
    return sum(data)

CPU 密集型：用 multiprocessing 替代threading
IO 密集型：使用 asyncio 协程

# 多进程示例
with Pool(processes=4) as pool:
    results = pool.map(heavy_computation, big_data)

# 协程示例
import aiohttp

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()