1. Python输入输出基础回顾与进阶场景
在Python编程中,输入输出(I/O)操作是与用户交互和数据持久化的基础通道。上篇我们已经探讨了基本的input()和print()用法,本篇将深入更多实际开发中会遇到的核心场景和技巧。
注意:本文所有代码示例基于Python 3.8+环境,部分特性在早期版本可能不兼容
1.1 输入处理的安全边界
用户输入永远是不可信的,这是编程的第一原则。来看一个典型的危险案例:
user_input = input("请输入数字:") result = 100 / int(user_input) # 用户输入0会导致崩溃正确处理方式应包含类型检查和异常捕获:
while True: try: num = float(input("请输入非零数字:")) print(f"100除以该数的结果是:{100/num:.2f}") break except ValueError: print("请输入有效的数字!") except ZeroDivisionError: print("除数不能为零!")1.2 格式化输出的现代实践
Python 3.6+推荐使用f-string进行格式化,它比传统的%和format()更直观高效:
name = "张三" age = 25 score = 89.456 # 传统方式 print("姓名:%s,年龄:%d,分数:%.1f" % (name, age, score)) # format方法 print("姓名:{},年龄:{},分数:{:.1f}".format(name, age, score)) # f-string(推荐) print(f"姓名:{name},年龄:{age},分数:{score:.1f}")f-string的高级用法包括表达式计算和格式规范:
import datetime now = datetime.datetime.now() print(f"当前时间:{now:%Y-%m-%d %H:%M:%S}") print(f"计算结果:{3**2 + 4**2 = }") # 输出:计算结果:3**2 + 4**2 = 252. 文件I/O的高效处理方案
2.1 上下文管理器与异常处理
文件操作必须使用with语句确保资源释放:
# 危险写法(可能造成文件未关闭) f = open('data.txt', 'r') content = f.read() f.close() # 正确写法 try: with open('data.txt', 'r', encoding='utf-8') as f: content = f.read() except FileNotFoundError: print("文件不存在!") except UnicodeDecodeError: print("文件编码错误!")2.2 大文件处理策略
对于GB级别的大文件,应该使用迭代方式读取:
def process_large_file(filename): with open(filename, 'r', encoding='utf-8') as f: for line_number, line in enumerate(f, 1): # 逐行处理,避免内存溢出 process_line(line) # 每处理10000行输出进度 if line_number % 10000 == 0: print(f"已处理 {line_number} 行")2.3 二进制文件的读写技巧
处理图片、音频等二进制文件时:
# 复制二进制文件 with open('source.jpg', 'rb') as src, open('copy.jpg', 'wb') as dst: while chunk := src.read(4096): # 4KB分块读取 dst.write(chunk)3. 高级输出控制技巧
3.1 动态进度显示
import time import sys def progress_bar(total): for i in range(total+1): percent = i / total * 100 sys.stdout.write(f"\r进度:[{'#'*int(percent//2):<50}] {percent:.1f}%") sys.stdout.flush() time.sleep(0.1) progress_bar(100)3.2 颜色输出方案
在终端中输出彩色文字(跨平台方案):
from colorama import init, Fore, Back, Style init(autoreset=True) # 自动重置样式 print(Fore.RED + "错误信息") print(Back.GREEN + "成功背景") print(Style.BRIGHT + "加粗文本")3.3 日志系统与输出重定向
生产环境应该使用logging模块:
import logging logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('app.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) logger.info("系统启动") logger.error("发生错误")4. 输入验证与清洗框架
4.1 通用验证装饰器
from functools import wraps def validate_input(*validators): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): for i, (arg, validator) in enumerate(zip(args, validators)): if not validator(arg): raise ValueError(f"参数{i}验证失败: {arg}") return func(*args, **kwargs) return wrapper return decorator # 使用示例 @validate_input(lambda x: x > 0, lambda x: isinstance(x, str)) def calculate_sqrt(num, name): return f"{name}的平方根是:{num**0.5}" print(calculate_sqrt(16, "数字")) # 正常 print(calculate_sqrt(-1, "数字")) # 触发异常4.2 正则表达式验证模板
import re def validate_email(email): pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return re.match(pattern, email) is not None def validate_phone(phone): pattern = r'^1[3-9]\d{9}$' # 中国大陆手机号 return re.match(pattern, phone) is not None5. 流式处理与管道技术
5.1 生成器管道
def read_lines(filename): with open(filename) as f: for line in f: yield line.strip() def filter_comments(lines): for line in lines: if not line.startswith('#'): yield line def to_uppercase(lines): for line in lines: yield line.upper() # 构建处理管道 lines = read_lines('config.ini') lines = filter_comments(lines) lines = to_uppercase(lines) for processed_line in lines: print(processed_line)5.2 标准流重定向
import sys from io import StringIO # 捕获标准输出 old_stdout = sys.stdout sys.stdout = buffer = StringIO() print("这段文字会被捕获") print("而不是显示在屏幕上") sys.stdout = old_stdout # 恢复 captured_output = buffer.getvalue()6. 性能优化与调试技巧
6.1 I/O性能对比
import timeit def test_concat(): with open('output.txt', 'w') as f: for i in range(10000): f.write(str(i) + '\n') def test_join(): lines = [str(i)+'\n' for i in range(10000)] with open('output.txt', 'w') as f: f.writelines(lines) print("直接拼接:", timeit.timeit(test_concat, number=10)) print("批量写入:", timeit.timeit(test_join, number=10))6.2 内存分析工具
使用memory_profiler检测内存使用:
from memory_profiler import profile @profile def process_data(): data = [open(f'temp_{i}.txt').read() for i in range(100)] # 危险操作 return ''.join(data) process_data()7. 实战:构建CLI应用程序
7.1 使用argparse创建命令行工具
import argparse def main(): parser = argparse.ArgumentParser(description="文件处理工具") parser.add_argument('input', help="输入文件路径") parser.add_argument('-o', '--output', help="输出文件路径") parser.add_argument('-v', '--verbose', action='store_true', help="详细模式") args = parser.parse_args() if args.verbose: print(f"正在处理文件:{args.input}") # 处理逻辑... if args.output: with open(args.output, 'w') as f: f.write("处理结果") if __name__ == '__main__': main()7.2 交互式命令行界面
import cmd class MyCLI(cmd.Cmd): prompt = '(myapp) ' def do_greet(self, arg): """问候用户""" print(f"你好,{arg or '陌生人'}!") def do_exit(self, arg): """退出程序""" print("再见!") return True if __name__ == '__main__': MyCLI().cmdloop()8. 跨平台兼容性问题
8.1 路径处理最佳实践
from pathlib import Path # 创建跨平台路径 config_path = Path('config') / 'settings.ini' # 安全读取 if config_path.exists(): content = config_path.read_text(encoding='utf-8')8.2 编码问题解决方案
import locale def get_system_encoding(): try: return locale.getpreferredencoding() except: return 'utf-8' with open('data.txt', 'r', encoding=get_system_encoding()) as f: content = f.read()9. 测试驱动开发实践
9.1 输入处理单元测试
import unittest from io import StringIO from unittest.mock import patch class TestInputProcessing(unittest.TestCase): def test_valid_input(self): with patch('sys.stdin', StringIO('42\n')): result = input("请输入数字:") self.assertEqual(result, "42") def test_invalid_input(self): with patch('sys.stdin', StringIO('abc\n42\n')): with self.assertRaises(ValueError): num = int(input("请输入数字:")) if __name__ == '__main__': unittest.main()9.2 输出捕获测试
from io import StringIO import sys def test_output_redirect(): buffer = StringIO() sys.stdout = buffer print("测试输出") output = buffer.getvalue() sys.stdout = sys.__stdout__ # 恢复 assert "测试输出" in output10. 安全防护与异常处理
10.1 注入攻击防护
import ast def safe_eval(input_str): try: return ast.literal_eval(input_str) except (ValueError, SyntaxError): return input_str # 安全处理用户输入 user_input = input("请输入数据:") processed = safe_eval(user_input)10.2 资源耗尽防护
import resource import signal def set_memory_limit(mb): soft, hard = resource.getrlimit(resource.RLIMIT_AS) resource.setrlimit(resource.RLIMIT_AS, (mb * 1024 * 1024, hard)) def handle_timeout(signum, frame): raise TimeoutError("操作超时") # 设置超时和内存限制 signal.signal(signal.SIGALRM, handle_timeout) signal.alarm(5) # 5秒超时 set_memory_limit(100) # 限制100MB内存