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(f100除以该数的结果是{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, encodingutf-8) as f: content f.read() except FileNotFoundError: print(文件不存在) except UnicodeDecodeError: print(文件编码错误)2.2 大文件处理策略对于GB级别的大文件应该使用迭代方式读取def process_large_file(filename): with open(filename, r, encodingutf-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(total1): 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(autoresetTrue) # 自动重置样式 print(Fore.RED 错误信息) print(Back.GREEN 成功背景) print(Style.BRIGHT 加粗文本)3.3 日志系统与输出重定向生产环境应该使用logging模块import logging logging.basicConfig( levellogging.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, number10)) print(批量写入, timeit.timeit(test_join, number10))6.2 内存分析工具使用memory_profiler检测内存使用from memory_profiler import profile profile def process_data(): data [open(ftemp_{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, actionstore_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(encodingutf-8)8.2 编码问题解决方案import locale def get_system_encoding(): try: return locale.getpreferredencoding() except: return utf-8 with open(data.txt, r, encodingget_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内存
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(f100除以该数的结果是{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, encodingutf-8) as f: content f.read() except FileNotFoundError: print(文件不存在) except UnicodeDecodeError: print(文件编码错误)2.2 大文件处理策略对于GB级别的大文件应该使用迭代方式读取def process_large_file(filename): with open(filename, r, encodingutf-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(total1): 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(autoresetTrue) # 自动重置样式 print(Fore.RED 错误信息) print(Back.GREEN 成功背景) print(Style.BRIGHT 加粗文本)3.3 日志系统与输出重定向生产环境应该使用logging模块import logging logging.basicConfig( levellogging.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, number10)) print(批量写入, timeit.timeit(test_join, number10))6.2 内存分析工具使用memory_profiler检测内存使用from memory_profiler import profile profile def process_data(): data [open(ftemp_{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, actionstore_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(encodingutf-8)8.2 编码问题解决方案import locale def get_system_encoding(): try: return locale.getpreferredencoding() except: return utf-8 with open(data.txt, r, encodingget_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内存
