1. 文本处理
1.1 string字符串操作
01.常用字符串方法
a.大小写转换
Python提供了多种字符串大小写转换方法，包括upper转大写、lower转小写、capitalize首字母大写、title标题化、swapcase大小写互换等。
b.代码示例
---
# 字符串大小写转换示例
text = "Hello World Python"
# 转换为大写
print(text.upper()) # HELLO WORLD PYTHON
# 转换为小写
print(text.lower()) # hello world python
# 首字母大写
print(text.capitalize()) # Hello world python
# 标题化（每个单词首字母大写）
print(text.title()) # Hello World Python
# 大小写互换
print(text.swapcase()) # hELLO wORLD pYTHON
---
c.查找与替换
字符串查找方法包括find查找子串位置、index查找子串索引、count统计出现次数、replace替换子串、startswith检查前缀、endswith检查后缀等。
d.代码示例
---
# 字符串查找与替换示例
text = "Python is powerful. Python is easy."
# 查找子串位置（找不到返回-1）
pos = text.find("Python")
print(f"First 'Python' at: {pos}") # 0
# 查找子串索引（找不到抛出异常）
try:
idx = text.index("Java")
except ValueError:
print("'Java' not found")
# 统计子串出现次数
count = text.count("Python")
print(f"'Python' appears {count} times") # 2
# 替换子串
new_text = text.replace("Python", "Go")
print(new_text) # Go is powerful. Go is easy.
# 检查前缀和后缀
print(text.startswith("Python")) # True
print(text.endswith("easy.")) # True
---
02.字符串分割与连接
a.分割方法
split方法按分隔符分割字符串返回列表，rsplit从右侧分割，splitlines按行分割，partition分割为三元组。
b.代码示例
---
# 字符串分割示例
text = "apple,banana,orange,grape"
# 按逗号分割
fruits = text.split(",")
print(fruits) # ['apple', 'banana', 'orange', 'grape']
# 限制分割次数
parts = text.split(",", 2)
print(parts) # ['apple', 'banana', 'orange,grape']
# 从右侧分割
parts = text.rsplit(",", 1)
print(parts) # ['apple,banana,orange', 'grape']
# 按行分割
multiline = "line1\nline2\nline3"
lines = multiline.splitlines()
print(lines) # ['line1', 'line2', 'line3']
# 分割为三元组（前、分隔符、后）
before, sep, after = "key=value".partition("=")
print(f"{before} | {sep} | {after}") # key | = | value
---
c.连接方法
join方法将序列中的字符串连接为一个字符串，可以指定连接符。
d.代码示例
---
# 字符串连接示例
words = ["Python", "is", "awesome"]
# 用空格连接
sentence = " ".join(words)
print(sentence) # Python is awesome
# 用逗号连接
csv_line = ",".join(words)
print(csv_line) # Python,is,awesome
# 用换行符连接
multiline = "\n".join(words)
print(multiline)
# Python
# is
# awesome
# 连接路径
path_parts = ["home", "user", "documents"]
path = "/".join(path_parts)
print(path) # home/user/documents
---
03.字符串格式化
a.格式化方法
Python提供多种字符串格式化方式，包括百分号格式化、format方法、f-string字面量格式化等。
b.百分号格式化
---
# 百分号格式化示例
name = "Alice"
age = 25
score = 95.5
# 基本格式化
msg = "Name: %s, Age: %d" % (name, age)
print(msg) # Name: Alice, Age: 25
# 浮点数格式化
msg = "Score: %.2f" % score
print(msg) # Score: 95.50
# 字典格式化
data = {"name": "Bob", "age": 30}
msg = "Name: %(name)s, Age: %(age)d" % data
print(msg) # Name: Bob, Age: 30
---
c.format方法
---
# format方法格式化示例
name = "Charlie"
age = 28
# 位置参数
msg = "Name: {}, Age: {}".format(name, age)
print(msg) # Name: Charlie, Age: 28
# 索引参数
msg = "Age: {1}, Name: {0}".format(name, age)
print(msg) # Age: 28, Name: Charlie
# 命名参数
msg = "Name: {n}, Age: {a}".format(n=name, a=age)
print(msg) # Name: Charlie, Age: 28
# 格式说明符
pi = 3.14159
msg = "Pi: {:.2f}".format(pi)
print(msg) # Pi: 3.14
# 对齐和填充
msg = "{:>10}".format("right") # 右对齐，宽度10
print(msg) # right
msg = "{:<10}".format("left") # 左对齐
print(msg) # left
msg = "{:^10}".format("center") # 居中
print(msg) # center
---
d.f-string格式化
---
# f-string格式化示例（Python 3.6+）
name = "David"
age = 32
balance = 1234.567
# 基本用法
msg = f"Name: {name}, Age: {age}"
print(msg) # Name: David, Age: 32
# 表达式计算
msg = f"Next year: {age + 1}"
print(msg) # Next year: 33
# 格式说明符
msg = f"Balance: ${balance:.2f}"
print(msg) # Balance: $1234.57
# 调用方法
msg = f"Uppercase: {name.upper()}"
print(msg) # Uppercase: DAVID
# 对齐和填充
msg = f"{name:>10}" # 右对齐
print(msg) # David
# 日期格式化
from datetime import datetime
now = datetime.now()
msg = f"Date: {now:%Y-%m-%d %H:%M:%S}"
print(msg) # Date: 2025-01-24 10:30:45
---
1.2 re正则表达式
01.正则表达式基础
a.模式匹配
re模块提供正则表达式匹配功能，支持search查找第一个匹配、match从开头匹配、findall查找所有匹配、finditer返回迭代器等方法。
b.代码示例
---
import re
text = "Contact: alice@example.com or bob@test.org"
# search查找第一个匹配
match = re.search(r'\w+@\w+\.\w+', text)
if match:
print(f"Found: {match.group()}") # alice@example.com
# match从字符串开头匹配
result = re.match(r'Contact', text)
print(f"Match: {result.group()}") # Contact
# findall查找所有匹配
emails = re.findall(r'\w+@\w+\.\w+', text)
print(f"All emails: {emails}") # ['alice@example.com', 'bob@test.org']
# finditer返回迭代器
for match in re.finditer(r'\w+@\w+\.\w+', text):
print(f"Email at {match.start()}: {match.group()}")
---
c.常用元字符
正则表达式元字符包括点号匹配任意字符、星号零次或多次、加号一次或多次、问号零次或一次、方括号字符集、圆括号分组等。
d.代码示例
---
import re
# 点号匹配任意字符（除换行符）
print(re.findall(r'a.c', 'abc adc a c')) # ['abc', 'adc']
# 星号：零次或多次
print(re.findall(r'ab*c', 'ac abc abbc')) # ['ac', 'abc', 'abbc']
# 加号：一次或多次
print(re.findall(r'ab+c', 'ac abc abbc')) # ['abc', 'abbc']
# 问号：零次或一次
print(re.findall(r'ab?c', 'ac abc abbc')) # ['ac', 'abc']
# 方括号：字符集
print(re.findall(r'[aeiou]', 'hello')) # ['e', 'o']
# 圆括号：分组
match = re.search(r'(\w+)@(\w+)\.(\w+)', 'user@example.com')
print(match.groups()) # ('user', 'example', 'com')
---
02.正则表达式替换
a.sub方法
sub方法用于替换匹配的字符串，subn方法返回替换后的字符串和替换次数。
b.代码示例
---
import re
text = "Phone: 123-456-7890, Mobile: 098-765-4321"
# 替换所有匹配
result = re.sub(r'\d{3}-\d{3}-\d{4}', 'XXX-XXX-XXXX', text)
print(result) # Phone: XXX-XXX-XXXX, Mobile: XXX-XXX-XXXX
# 限制替换次数
result = re.sub(r'\d{3}-\d{3}-\d{4}', 'XXX-XXX-XXXX', text, count=1)
print(result) # Phone: XXX-XXX-XXXX, Mobile: 098-765-4321
# subn返回替换次数
result, count = re.subn(r'\d{3}-\d{3}-\d{4}', 'XXX-XXX-XXXX', text)
print(f"Replaced {count} times: {result}")
# 使用函数进行替换
def mask_phone(match):
phone = match.group()
return phone[:3] + '-XXX-' + phone[-4:]
result = re.sub(r'\d{3}-\d{3}-\d{4}', mask_phone, text)
print(result) # Phone: 123-XXX-7890, Mobile: 098-XXX-4321
---
c.分组引用
在替换字符串中可以使用\1、\2等引用捕获组，或使用\g<name>引用命名组。
d.代码示例
---
import re
# 使用数字引用分组
text = "2025-01-24"
result = re.sub(r'(\d{4})-(\d{2})-(\d{2})', r'\3/\2/\1', text)
print(result) # 24/01/2025
# 使用命名组
pattern = r'(?P<year>\d{4})-(?P<month>\d{2})-(?P<day>\d{2})'
result = re.sub(pattern, r'\g<day>/\g<month>/\g<year>', text)
print(result) # 24/01/2025
# 交换单词
text = "First Last"
result = re.sub(r'(\w+) (\w+)', r'\2, \1', text)
print(result) # Last, First
---
03.编译正则表达式
a.compile方法
使用compile方法预编译正则表达式可以提高性能，特别是需要多次使用同一模式时。
b.代码示例
---
import re
# 编译正则表达式
email_pattern = re.compile(r'\w+@\w+\.\w+')
text1 = "Contact: alice@example.com"
text2 = "Email: bob@test.org"
# 使用编译后的模式
match1 = email_pattern.search(text1)
print(match1.group()) # alice@example.com
match2 = email_pattern.search(text2)
print(match2.group()) # bob@test.org
# 使用标志
pattern = re.compile(r'hello', re.IGNORECASE)
print(pattern.findall('Hello HELLO hello')) # ['Hello', 'HELLO', 'hello']
# 多行模式
pattern = re.compile(r'^line', re.MULTILINE)
text = "line1\nline2\nline3"
print(pattern.findall(text)) # ['line', 'line', 'line']
---
c.常用标志
正则表达式标志包括IGNORECASE忽略大小写、MULTILINE多行模式、DOTALL点号匹配换行符、VERBOSE详细模式等。
d.代码示例
---
import re
# IGNORECASE：忽略大小写
pattern = re.compile(r'python', re.I)
print(pattern.findall('Python PYTHON python')) # ['Python', 'PYTHON', 'python']
# MULTILINE：多行模式（^和$匹配每行开头和结尾）
text = "start\nmiddle\nend"
pattern = re.compile(r'^middle$', re.M)
print(pattern.search(text).group()) # middle
# DOTALL：点号匹配换行符
text = "line1\nline2"
pattern = re.compile(r'line1.line2', re.S)
print(pattern.search(text).group()) # line1\nline2
# VERBOSE：详细模式（允许注释和空白）
pattern = re.compile(r'''
\d{3} # 区号
- # 分隔符
\d{4} # 号码
''', re.X)
print(pattern.search('123-4567').group()) # 123-4567
# 组合多个标志
pattern = re.compile(r'hello', re.I | re.M)
---
1.3 textwrap文本包装
01.文本包装功能
a.wrap方法
wrap方法将长文本按指定宽度分割成多行，返回字符串列表，fill方法返回单个字符串。
b.代码示例
---
import textwrap
text = "Python is a high-level, interpreted programming language with dynamic semantics and simple syntax."
# 包装为指定宽度
wrapped = textwrap.wrap(text, width=30)
for line in wrapped:
print(line)
# Python is a high-level,
# interpreted programming
# language with dynamic
# semantics and simple syntax.
# fill方法返回单个字符串
filled = textwrap.fill(text, width=30)
print(filled)
# 设置后续行缩进
wrapped = textwrap.fill(text, width=40, initial_indent=" ", subsequent_indent=" ")
print(wrapped)
---
02.文本缩进
a.indent方法
indent方法为文本的每一行添加指定的前缀，可以选择性地跳过空行。
b.代码示例
---
import textwrap
text = """Line 1
Line 2
Line 4"""
# 为所有行添加前缀
indented = textwrap.indent(text, '> ')
print(indented)
# > Line 1
# > Line 2
# >
# > Line 4
# 跳过空行
indented = textwrap.indent(text, '> ', lambda line: line.strip())
print(indented)
# 代码块缩进
code = "def hello():\n print('Hello')"
indented = textwrap.indent(code, ' ')
print(indented)
---
03.文本清理
a.dedent方法
dedent方法移除文本每行开头的公共空白字符，常用于处理多行字符串。
b.代码示例
---
import textwrap
# 清理缩进
text = """
Hello World
Python Programming
Text Processing
"""
cleaned = textwrap.dedent(text)
print(cleaned)
# Hello World
# Python Programming
# Text Processing
# 结合strip使用
text = """
def example():
print("Hello")
return True
"""
code = textwrap.dedent(text).strip()
print(code)
---
1.4 difflib差异比较
01.序列差异比较
a.SequenceMatcher类
SequenceMatcher类用于比较任意类型的序列对，计算相似度并找出差异部分。
b.代码示例
---
from difflib import SequenceMatcher
# 比较两个字符串
s1 = "apple"
s2 = "apply"
matcher = SequenceMatcher(None, s1, s2)
# 计算相似度
ratio = matcher.ratio()
print(f"Similarity: {ratio:.2%}") # 80.00%
# 获取匹配块
for block in matcher.get_matching_blocks():
print(f"Match: a[{block.a}:{block.a+block.size}] == b[{block.b}:{block.b+block.size}]")
# 获取操作码
for tag, i1, i2, j1, j2 in matcher.get_opcodes():
print(f"{tag}: a[{i1}:{i2}] -> b[{j1}:{j2}]")
---
02.文本差异对比
a.unified_diff方法
unified_diff生成统一格式的差异输出，类似于Unix diff命令的-u选项。
b.代码示例
---
from difflib import unified_diff
text1 = """Line 1
Line 2
Line 3
Line 4""".splitlines(keepends=True)
text2 = """Line 1
Line 2 modified
Line 3
Line 5""".splitlines(keepends=True)
# 生成统一差异
diff = unified_diff(text1, text2, fromfile='old.txt', tofile='new.txt', lineterm='')
for line in diff:
print(line)
---
03.HTML差异报告
a.HtmlDiff类
HtmlDiff类生成HTML格式的差异对比报告，支持并排和行内两种显示模式。
b.代码示例
---
from difflib import HtmlDiff
text1 = ['Line 1\n', 'Line 2\n', 'Line 3\n']
text2 = ['Line 1\n', 'Line 2 modified\n', 'Line 4\n']
# 创建HTML差异
differ = HtmlDiff()
html = differ.make_file(text1, text2, fromdesc='Version 1', todesc='Version 2')
# 保存为HTML文件
with open('diff.html', 'w', encoding='utf-8') as f:
f.write(html)
---
1.5 unicodedata Unicode数据库
01.字符属性查询
a.name和category方法
name方法返回Unicode字符的名称，category方法返回字符的类别代码。
b.代码示例
---
import unicodedata
# 获取字符名称
char = 'A'
name = unicodedata.name(char)
print(f"Name of '{char}': {name}") # LATIN CAPITAL LETTER A
# 获取字符类别
category = unicodedata.category(char)
print(f"Category: {category}") # Lu (Letter, uppercase)
# 中文字符
char = '中'
print(unicodedata.name(char)) # CJK UNIFIED IDEOGRAPH-4E2D
print(unicodedata.category(char)) # Lo (Letter, other)
# 数字字符
char = '5'
print(unicodedata.category(char)) # Nd (Number, decimal digit)
---
02.字符规范化
a.normalize方法
normalize方法将Unicode字符串转换为标准形式，支持NFC、NFD、NFKC、NFKD四种规范化形式。
b.代码示例
---
import unicodedata
# 组合字符和分解字符
s1 = 'café' # e + 组合重音符
s2 = 'café' # 预组合字符
print(f"Length: {len(s1)} vs {len(s2)}")
# NFC规范化（组合形式）
nfc1 = unicodedata.normalize('NFC', s1)
nfc2 = unicodedata.normalize('NFC', s2)
print(f"NFC equal: {nfc1 == nfc2}") # True
# NFD规范化（分解形式）
nfd = unicodedata.normalize('NFD', s1)
print(f"NFD length: {len(nfd)}")
# NFKC和NFKD（兼容性规范化）
s = '①' # 带圈数字
nfkc = unicodedata.normalize('NFKC', s)
print(f"NFKC: {nfkc}") # 1
---
03.数字转换
a.numeric和digit方法
numeric方法返回字符的数值，digit方法返回数字字符的值。
b.代码示例
---
import unicodedata
# 获取数字值
print(unicodedata.numeric('5')) # 5.0
print(unicodedata.numeric('½')) # 0.5
print(unicodedata.numeric('Ⅳ')) # 4.0
# digit方法（仅用于0-9）
print(unicodedata.digit('7')) # 7
# 中文数字
print(unicodedata.numeric('三')) # 3.0
---
2. 数据类型
2.1 datetime日期时间
01.日期时间对象
a.创建日期时间
datetime模块提供date日期类、time时间类、datetime日期时间类、timedelta时间差类等。
b.代码示例
---
from datetime import date, time, datetime, timedelta
# 创建日期对象
d = date(2025, 1, 24)
print(d) # 2025-01-24
# 创建时间对象
t = time(14, 30, 45)
print(t) # 14:30:45
# 创建日期时间对象
dt = datetime(2025, 1, 24, 14, 30, 45)
print(dt) # 2025-01-24 14:30:45
# 获取当前日期时间
now = datetime.now()
print(f"Now: {now}")
# 获取今天日期
today = date.today()
print(f"Today: {today}")
# 从时间戳创建
timestamp = 1706097045
dt = datetime.fromtimestamp(timestamp)
print(f"From timestamp: {dt}")
---
02.日期时间格式化
a.strftime和strptime
strftime将日期时间格式化为字符串，strptime将字符串解析为日期时间对象。
b.代码示例
---
from datetime import datetime
now = datetime.now()
# 格式化为字符串
s1 = now.strftime("%Y-%m-%d %H:%M:%S")
print(s1) # 2025-01-24 14:30:45
s2 = now.strftime("%Y年%m月%d日")
print(s2) # 2025年01月24日
s3 = now.strftime("%A, %B %d, %Y")
print(s3) # Friday, January 24, 2025
# 解析字符串
dt1 = datetime.strptime("2025-01-24", "%Y-%m-%d")
print(dt1) # 2025-01-24 00:00:00
dt2 = datetime.strptime("24/01/2025 14:30", "%d/%m/%Y %H:%M")
print(dt2) # 2025-01-24 14:30:00
---
03.时间差计算
a.timedelta对象
timedelta表示两个日期或时间之间的差值，支持天、秒、微秒等单位。
b.代码示例
---
from datetime import datetime, timedelta
now = datetime.now()
# 创建时间差
delta = timedelta(days=7, hours=3, minutes=30)
print(f"Delta: {delta}")
# 日期加减
future = now + timedelta(days=30)
print(f"30 days later: {future}")
past = now - timedelta(weeks=2)
print(f"2 weeks ago: {past}")
# 计算两个日期的差值
date1 = datetime(2025, 1, 1)
date2 = datetime(2025, 12, 31)
diff = date2 - date1
print(f"Days between: {diff.days}") # 364
# 获取时间差的各个部分
delta = timedelta(days=5, hours=3, minutes=30, seconds=45)
print(f"Total seconds: {delta.total_seconds()}")
print(f"Days: {delta.days}")
print(f"Seconds: {delta.seconds}")
---
2.2 collections容器数据类型
01.Counter计数器
a.元素计数
Counter是dict的子类，用于计数可哈希对象，提供most_common等便捷方法。
b.代码示例
---
from collections import Counter
# 创建计数器
words = ['apple', 'banana', 'apple', 'orange', 'banana', 'apple']
counter = Counter(words)
print(counter) # Counter({'apple': 3, 'banana': 2, 'orange': 1})
# 最常见的元素
print(counter.most_common(2)) # [('apple', 3), ('banana', 2)]
# 访问计数
print(counter['apple']) # 3
print(counter['grape']) # 0（不存在返回0）
# 更新计数
counter.update(['apple', 'grape'])
print(counter)
# 计数器运算
c1 = Counter(['a', 'b', 'c'])
c2 = Counter(['b', 'c', 'd'])
print(c1 + c2) # Counter({'b': 2, 'c': 2, 'a': 1, 'd': 1})
print(c1 - c2) # Counter({'a': 1})
---
02.deque双端队列
a.队列操作
deque提供线程安全的双端队列，支持从两端快速添加和删除元素。
b.代码示例
---
from collections import deque
# 创建双端队列
dq = deque([1, 2, 3])
# 右端操作
dq.append(4)
print(dq) # deque([1, 2, 3, 4])
dq.pop()
print(dq) # deque([1, 2, 3])
# 左端操作
dq.appendleft(0)
print(dq) # deque([0, 1, 2, 3])
dq.popleft()
print(dq) # deque([1, 2, 3])
# 扩展
dq.extend([4, 5])
dq.extendleft([0, -1])
print(dq) # deque([-1, 0, 1, 2, 3, 4, 5])
# 旋转
dq.rotate(2) # 向右旋转2位
print(dq)
# 限制长度
limited = deque(maxlen=3)
for i in range(5):
limited.append(i)
print(limited) # deque([2, 3, 4], maxlen=3)
---
03.defaultdict默认字典
a.默认值字典
defaultdict在访问不存在的键时自动创建默认值，避免KeyError异常。
b.代码示例
---
from collections import defaultdict
# 创建默认字典
dd = defaultdict(int)
dd['a'] += 1
dd['b'] += 2
print(dd) # defaultdict(<class 'int'>, {'a': 1, 'b': 2})
# 列表作为默认值
dd_list = defaultdict(list)
dd_list['fruits'].append('apple')
dd_list['fruits'].append('banana')
dd_list['colors'].append('red')
print(dd_list)
# 分组示例
words = ['apple', 'ant', 'banana', 'bear', 'cat']
grouped = defaultdict(list)
for word in words:
grouped[word[0]].append(word)
print(dict(grouped)) # {'a': ['apple', 'ant'], 'b': ['banana', 'bear'], 'c': ['cat']}
---
2.3 array数组
01.数组基础
a.创建数组
array模块提供高效的数值数组，所有元素必须是相同类型，比列表更节省内存。
b.代码示例
---
import array
# 创建整数数组
arr = array.array('i', [1, 2, 3, 4, 5])
print(arr) # array('i', [1, 2, 3, 4, 5])
# 创建浮点数组
farr = array.array('f', [1.1, 2.2, 3.3])
print(farr)
# 从字节创建
barr = array.array('i')
barr.frombytes(b'\x01\x00\x00\x00\x02\x00\x00\x00')
print(barr)
# 数组操作
arr.append(6)
arr.extend([7, 8])
arr.insert(0, 0)
print(arr)
---
02.数组方法
a.常用操作
数组支持append、extend、insert、remove、pop等方法，以及tobytes、tolist等转换方法。
b.代码示例
---
import array
arr = array.array('i', [3, 1, 4, 1, 5, 9, 2, 6])
# 查找和计数
print(arr.index(4)) # 2
print(arr.count(1)) # 2
# 反转
arr.reverse()
print(arr)
# 转换为列表
lst = arr.tolist()
print(lst)
# 转换为字节
bytes_data = arr.tobytes()
print(bytes_data)
# 从文件读写
with open('array.dat', 'wb') as f:
arr.tofile(f)
new_arr = array.array('i')
with open('array.dat', 'rb') as f:
new_arr.fromfile(f, len(arr))
print(new_arr)
---
2.4 heapq堆队列
01.堆操作
a.创建和维护堆
heapq模块实现最小堆算法，提供heappush、heappop、heapify等操作。
b.代码示例
---
import heapq
# 创建空堆
heap = []
# 添加元素
heapq.heappush(heap, 5)
heapq.heappush(heap, 3)
heapq.heappush(heap, 7)
heapq.heappush(heap, 1)
print(heap) # [1, 3, 7, 5]
# 弹出最小元素
min_val = heapq.heappop(heap)
print(f"Min: {min_val}") # 1
print(heap) # [3, 5, 7]
# 将列表转换为堆
nums = [5, 2, 8, 1, 9, 3]
heapq.heapify(nums)
print(nums) # [1, 2, 3, 5, 9, 8]
# 替换堆顶元素
heapq.heapreplace(nums, 4)
print(nums)
---
02.堆应用
a.Top K问题
使用nlargest和nsmallest快速找出最大或最小的K个元素。
b.代码示例
---
import heapq
nums = [3, 1, 4, 1, 5, 9, 2, 6, 5, 3]
# 找出最大的3个元素
largest = heapq.nlargest(3, nums)
print(f"Largest 3: {largest}") # [9, 6, 5]
# 找出最小的3个元素
smallest = heapq.nsmallest(3, nums)
print(f"Smallest 3: {smallest}") # [1, 1, 2]
# 使用key参数
words = ['apple', 'pie', 'a', 'banana']
longest = heapq.nlargest(2, words, key=len)
print(f"Longest: {longest}") # ['banana', 'apple']
# 合并多个有序序列
seq1 = [1, 3, 5, 7]
seq2 = [2, 4, 6, 8]
merged = list(heapq.merge(seq1, seq2))
print(f"Merged: {merged}") # [1, 2, 3, 4, 5, 6, 7, 8]
---
2.5 bisect数组二分查找
01.二分查找
a.bisect方法
bisect模块提供有序序列的二分查找和插入功能，保持序列有序。
b.代码示例
---
import bisect
# 有序列表
nums = [1, 3, 5, 7, 9]
# 查找插入位置（右侧）
pos = bisect.bisect(nums, 6)
print(f"Insert position: {pos}") # 3
# 查找插入位置（左侧）
pos_left = bisect.bisect_left(nums, 5)
print(f"Left position: {pos_left}") # 2
pos_right = bisect.bisect_right(nums, 5)
print(f"Right position: {pos_right}") # 3
# 插入元素并保持有序
bisect.insort(nums, 6)
print(nums) # [1, 3, 5, 6, 7, 9]
bisect.insort_left(nums, 5)
print(nums) # [1, 3, 5, 5, 6, 7, 9]
---
02.应用示例
a.成绩等级划分
使用bisect实现成绩等级划分等实际应用场景。
b.代码示例
---
import bisect
# 成绩等级划分
breakpoints = [60, 70, 80, 90]
grades = 'FDCBA'
def get_grade(score):
i = bisect.bisect(breakpoints, score)
return grades[i]
scores = [55, 65, 75, 85, 95]
for score in scores:
print(f"Score {score}: Grade {get_grade(score)}")
# 查找范围
def find_range(arr, target):
left = bisect.bisect_left(arr, target)
right = bisect.bisect_right(arr, target)
return left, right
nums = [1, 2, 2, 2, 3, 4, 5]
left, right = find_range(nums, 2)
print(f"Range of 2: [{left}, {right})") # [1, 4)
---
3. 数学运算
3.1 math数学函数
01.基本数学运算
a.常用数学函数
math模块提供了基本的数学运算函数，包括幂运算pow、平方根sqrt、绝对值fabs、向上取整ceil、向下取整floor、四舍五入round等。
b.代码示例
---
# math模块基本运算示例
import math
# 幂运算和平方根
print(math.pow(2, 3)) # 8.0
print(math.sqrt(16)) # 4.0
# 绝对值
print(math.fabs(-5.5)) # 5.5
# 取整运算
print(math.ceil(3.2)) # 4 向上取整
print(math.floor(3.8)) # 3 向下取整
# 最大公约数和最小公倍数
print(math.gcd(48, 18)) # 6
# 阶乘
print(math.factorial(5)) # 120
# 对数运算
print(math.log(100, 10)) # 2.0 以10为底
print(math.log2(8)) # 3.0 以2为底
print(math.log10(1000)) # 3.0 以10为底
---
c.三角函数
math模块提供了完整的三角函数支持，包括sin正弦、cos余弦、tan正切及其反函数，角度与弧度转换等。
d.代码示例
---
# 三角函数示例
import math
# 角度转弧度
angle = 45
radians = math.radians(angle)
print(f"{angle}度 = {radians}弧度") # 0.7853981633974483
# 三角函数计算
print(f"sin(45°) = {math.sin(radians):.4f}") # 0.7071
print(f"cos(45°) = {math.cos(radians):.4f}") # 0.7071
print(f"tan(45°) = {math.tan(radians):.4f}") # 1.0000
# 反三角函数
print(f"arcsin(0.5) = {math.degrees(math.asin(0.5))}°") # 30.0
print(f"arccos(0.5) = {math.degrees(math.acos(0.5))}°") # 60.0
# 弧度转角度
rad = math.pi / 3
deg = math.degrees(rad)
print(f"{rad}弧度 = {deg}度") # 60.0度
---
02.数学常量
a.常用常量
math模块定义了常用的数学常量，包括圆周率pi、自然对数底e、无穷大inf、非数字nan等。
b.代码示例
---
# 数学常量示例
import math
# 圆周率和自然对数底
print(f"π = {math.pi}") # 3.141592653589793
print(f"e = {math.e}") # 2.718281828459045
# 无穷大和非数字
print(math.inf) # inf
print(-math.inf) # -inf
print(math.nan) # nan
# 判断特殊值
print(math.isinf(math.inf)) # True
print(math.isnan(math.nan)) # True
print(math.isfinite(100)) # True
print(math.isfinite(math.inf)) # False
# 使用常量计算
radius = 5
area = math.pi * radius ** 2
print(f"半径{radius}的圆面积: {area:.2f}") # 78.54
---
3.2 decimal十进制定点数
01.Decimal基本使用
a.精确十进制运算
Decimal类型提供精确的十进制运算，避免浮点数精度问题，适用于金融计算等需要精���结果的场景。
b.代码示例
---
# Decimal精确运算示例
from decimal import Decimal, getcontext
# 浮点数精度问题
print(0.1 + 0.2) # 0.30000000000000004
# Decimal精确计算
d1 = Decimal('0.1')
d2 = Decimal('0.2')
print(d1 + d2) # 0.3
# 金融计算示例
price = Decimal('19.99')
quantity = Decimal('3')
total = price * quantity
print(f"总价: ${total}") # 总价: $59.97
# 设置精度
getcontext().prec = 6
result = Decimal('1') / Decimal('3')
print(result) # 0.333333
---
c.精度控制
可以通过getcontext设置全局精度，或使用quantize方法控制单个结果的精度和舍入方式。
d.代码示例
---
# Decimal精度控制示例
from decimal import Decimal, ROUND_HALF_UP, ROUND_DOWN
# 使用quantize控制精度
value = Decimal('3.14159')
rounded = value.quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)
print(rounded) # 3.14
# 不同舍入模式
value = Decimal('2.5')
print(value.quantize(Decimal('1'), rounding=ROUND_HALF_UP)) # 3
print(value.quantize(Decimal('1'), rounding=ROUND_DOWN)) # 2
# 货币格式化
amount = Decimal('1234.567')
formatted = amount.quantize(Decimal('0.01'))
print(f"${formatted}") # $1234.57
---
02.Decimal运算
a.算术运算
Decimal支持所有基本算术运算，包括加减乘除、幂运算、平方根等，并保持精确性。
b.代码示例
---
# Decimal算术运算示例
from decimal import Decimal
a = Decimal('10.5')
b = Decimal('3.2')
# 基本运算
print(f"加法: {a + b}") # 13.7
print(f"减法: {a - b}") # 7.3
print(f"乘法: {a * b}") # 33.60
print(f"除法: {a / b}") # 3.28125
# 幂运算
print(f"平方: {a ** 2}") # 110.25
# 平方根
print(f"平方根: {a.sqrt()}") # 3.240370349
# 比较运算
print(a > b) # True
print(a == Decimal('10.50')) # True
# 取整运算
print(f"向下取整: {a // b}") # 3
print(f"取余: {a % b}") # 1.1
---
3.3 fractions分数
01.Fraction基本使用
a.分数表示
Fraction类提供有理数的精确表示，自动化简分数，支持从整数、浮点数、字符串创建分数对象。
b.代码示例
---
# Fraction基本使用示例
from fractions import Fraction
# 创建分数
f1 = Fraction(3, 4) # 3/4
f2 = Fraction(5, 6) # 5/6
f3 = Fraction('0.5') # 1/2
f4 = Fraction(0.25) # 1/4
print(f1) # 3/4
print(f3) # 1/2
# 自动化简
f5 = Fraction(6, 8)
print(f5) # 3/4
# 访问分子分母
print(f"分子: {f1.numerator}") # 3
print(f"分母: {f1.denominator}") # 4
# 转换为浮点数
print(float(f1)) # 0.75
---
c.分数运算
Fraction支持所有算术运算，结果自动化简为最简分数形式。
d.代码示例
---
# Fraction运算示例
from fractions import Fraction
a = Fraction(1, 3)
b = Fraction(1, 6)
# 四则运算
print(f"{a} + {b} = {a + b}") # 1/3 + 1/6 = 1/2
print(f"{a} - {b} = {a - b}") # 1/3 - 1/6 = 1/6
print(f"{a} * {b} = {a * b}") # 1/3 * 1/6 = 1/18
print(f"{a} / {b} = {a / b}") # 1/3 / 1/6 = 2
# 幂运算
print(f"{a} ** 2 = {a ** 2}") # 1/9
# 比较运算
print(a > b) # True
print(a == Fraction(2, 6)) # True
---
02.分数应用
a.精确计算
使用Fraction进行需要精确结果的数学计算，避免浮点数误差。
b.代码示例
---
# Fraction精确计算示例
from fractions import Fraction
# 计算平均值
scores = [Fraction(85, 100), Fraction(90, 100), Fraction(88, 100)]
average = sum(scores) / len(scores)
print(f"平均分: {average} = {float(average):.2f}") # 263/300 = 0.88
# 比例计算
total = 100
part1 = Fraction(1, 3)
part2 = Fraction(1, 4)
remaining = 1 - part1 - part2
print(f"第一部分: {part1 * total}") # 100/3
print(f"第二部分: {part2 * total}") # 25
print(f"剩余: {remaining * total}") # 125/3
# 混合运算
result = Fraction(1, 2) + Fraction(1, 3) + Fraction(1, 6)
print(f"1/2 + 1/3 + 1/6 = {result}") # 1
---
3.4 random随机数生成
01.随机数生成
a.基本随机函数
random模块提供各种随机数生成函数，包括random生成0-1浮点数、randint生成整数、uniform生成指定范围浮点数等。
b.代码示例
---
# 随机数生成示例
import random
# 生成0-1之间的随机浮点数
print(random.random()) # 0.8444218515250481
# 生成指定范围的随机整数（包含两端）
dice = random.randint(1, 6)
print(f"骰子点数: {dice}")
# 生成指定范围的随机浮点数
temperature = random.uniform(20.0, 30.0)
print(f"温度: {temperature:.2f}°C")
# 生成指定步长的随机数
even = random.randrange(0, 10, 2) # 0,2,4,6,8
print(f"随机偶数: {even}")
# 设置随机种子（可重现结果）
random.seed(42)
print(random.random()) # 每次运行结果相同
---
c.序列随机操作
random模块提供对序列的随机操作，包括choice随机选择、shuffle打乱顺序、sample随机抽样等。
d.代码示例
---
# 序列随机操作示例
import random
# 从序列中随机选择一个元素
colors = ['red', 'green', 'blue', 'yellow']
color = random.choice(colors)
print(f"随机颜色: {color}")
# 随机抽样（不重复）
lottery = random.sample(range(1, 50), 6)
print(f"彩票号码: {sorted(lottery)}")
# 随机抽样（可重复）
samples = random.choices(['A', 'B', 'C'], k=5)
print(f"随机样本: {samples}")
# 打乱列表顺序
cards = list(range(1, 14))
random.shuffle(cards)
print(f"洗牌后: {cards}")
---
02.随机分布
a.概率分布
random模块支持多种概率分布，包括正态分布、指数分布、伽马分布等。
b.代码示例
---
# 概率分布示例
import random
# 正态分布（高斯分布）
# mu=平均值, sigma=标准差
height = random.gauss(170, 10)
print(f"身高: {height:.1f}cm")
# 指数分布
# lambd=1/平均值
wait_time = random.expovariate(1/5)
print(f"等待时间: {wait_time:.2f}分钟")
# 三角分布
# low=最小值, high=最大值, mode=众数
score = random.triangular(60, 100, 85)
print(f"考试分数: {score:.1f}")
# 贝塔分布
alpha, beta = 2, 5
value = random.betavariate(alpha, beta)
print(f"贝塔分布值: {value:.3f}")
---
3.5 statistics统计函数
01.基本统计量
a.集中趋势
statistics模块提供计算数据集中趋势的函数，包括mean平均值、median中位数、mode众数等。
b.代码示例
---
# 集中趋势统计示例
import statistics
data = [85, 90, 78, 92, 88, 85, 95, 82]
# 算术平均值
avg = statistics.mean(data)
print(f"平均分: {avg:.2f}") # 86.88
# 中位数
med = statistics.median(data)
print(f"中位数: {med}") # 86.5
# 众数（出现最多的值）
mod = statistics.mode(data)
print(f"众数: {mod}") # 85
# 几何平均数
geo_mean = statistics.geometric_mean([2, 8])
print(f"几何平均数: {geo_mean}") # 4.0
# 调和平均数
har_mean = statistics.harmonic_mean([40, 60])
print(f"调和平均数: {har_mean:.2f}") # 48.00
---
c.离散程度
statistics模块提供计算数据离散程度的函数，包括方差variance、标准差stdev等。
d.代码示例
---
# 离散程度统计示例
import statistics
scores = [85, 90, 78, 92, 88, 85, 95, 82]
# 总体方差
var = statistics.pvariance(scores)
print(f"总体方差: {var:.2f}") # 28.86
# 样本方差
sample_var = statistics.variance(scores)
print(f"样本方差: {sample_var:.2f}") # 33.27
# 总体标准差
std = statistics.pstdev(scores)
print(f"总体标准差: {std:.2f}") # 5.37
# 样本标准差
sample_std = statistics.stdev(scores)
print(f"样本标准差: {sample_std:.2f}") # 5.77
---
02.高级统计
a.分位数和相关性
statistics模块提供分���数计算和数据分析功能。
b.代码示例
---
# 高级统计示例
import statistics
data = [15, 20, 35, 40, 50, 55, 60, 70, 85, 90]
# 四分位数
q1 = statistics.quantiles(data, n=4)[0] # 第一四分位数
q2 = statistics.quantiles(data, n=4)[1] # 中位数
q3 = statistics.quantiles(data, n=4)[2] # 第三四分位数
print(f"Q1: {q1}, Q2: {q2}, Q3: {q3}")
# 十分位数
deciles = statistics.quantiles(data, n=10)
print(f"十分位数: {deciles}")
# 数据范围
data_range = max(data) - min(data)
print(f"数据范围: {data_range}")
# 四分位距
iqr = q3 - q1
print(f"四分位距: {iqr}")
---
4. 文件与目录
4.1 os操作系统接口
01.文件和目录操作
a.基本操作
os模块提供与操作系统交互的功能，包括文件目录操作、进程管理、环境变量访问等。
b.代码示例
---
# os模块文件目录操作示例
import os
# 获取当前工作目录
cwd = os.getcwd()
print(f"当前目录: {cwd}")
# 列出目录内容
files = os.listdir('.')
print(f"文件列表: {files[:5]}") # 显示前5个
# 创建目录
if not os.path.exists('test_dir'):
os.mkdir('test_dir')
print("目录已创建")
# 创建多级目录
os.makedirs('parent/child/grandchild', exist_ok=True)
# 删除目录
os.rmdir('test_dir')
# 重命名文件或目录
# os.rename('old_name.txt', 'new_name.txt')
# 删除文件
# os.remove('file.txt')
---
c.路径操作
os.path模块提供路径相关的操作函数，包括路径拼接、分割、判断等。
d.代码示例
---
# os.path路径操作示例
import os
# 路径拼接
path = os.path.join('folder', 'subfolder', 'file.txt')
print(f"拼接路径: {path}")
# 获取绝对路径
abs_path = os.path.abspath('.')
print(f"绝对路径: {abs_path}")
# 分割路径
dir_name = os.path.dirname('/home/user/file.txt')
base_name = os.path.basename('/home/user/file.txt')
print(f"目录: {dir_name}, 文件名: {base_name}")
# 分割扩展名
name, ext = os.path.splitext('document.pdf')
print(f"文件名: {name}, 扩展名: {ext}")
# 判断路径
print(f"是否存在: {os.path.exists('.')}")
print(f"是否为文件: {os.path.isfile('.')}")
print(f"是否为目录: {os.path.isdir('.')}")
---
02.环境和进程
a.环境变量
os模块提供访问和修改环境变量的功能。
b.代码示例
---
# 环境变量和进程操作示例
import os
# 获取环境变量
home = os.environ.get('HOME') or os.environ.get('USERPROFILE')
print(f"用户目录: {home}")
# 设置环境变量
os.environ['MY_VAR'] = 'test_value'
print(f"MY_VAR: {os.environ.get('MY_VAR')}")
# 获取所有环境变量
env_count = len(os.environ)
print(f"环境变量数量: {env_count}")
# 获取进程ID
pid = os.getpid()
print(f"当前进程ID: {pid}")
# 获取父进程ID
ppid = os.getppid()
print(f"父进程ID: {ppid}")
# 执行系统命令
# result = os.system('echo Hello')
---
4.2 pathlib面向对象的路径
01.Path对象基础
a.创建和操作Path对象
pathlib提供面向对象的路径操作接口，Path对象支持路径拼接、属性访问、文件操作等。
b.代码示例
---
# Path对象基础示例
from pathlib import Path
# 创建Path对象
p = Path('.')
print(f"当前路径: {p}")
# 获取绝对路径
abs_p = p.absolute()
print(f"绝对路径: {abs_p}")
# 路径拼接（使用/运算符）
file_path = p / 'folder' / 'file.txt'
print(f"拼接路径: {file_path}")
# 获取路径组成部分
print(f"父目录: {file_path.parent}")
print(f"文件名: {file_path.name}")
print(f"扩展名: {file_path.suffix}")
print(f"不含扩展名: {file_path.stem}")
# 获取所有父目录
for parent in file_path.parents:
print(f"父级: {parent}")
---
c.路径判断和属性
Path对象提供多种方法判断路径类型和获取路径属性。
d.代码示例
---
# Path判断和属性示例
from pathlib import Path
p = Path('.')
# 判断路径类型
print(f"是否存在: {p.exists()}")
print(f"是否为文件: {p.is_file()}")
print(f"是否为目录: {p.is_dir()}")
print(f"是否为符号链接: {p.is_symlink()}")
print(f"是否为绝对路径: {p.is_absolute()}")
# 获取文件信息
if p.exists():
stat = p.stat()
print(f"文件大小: {stat.st_size}字节")
print(f"修改时间: {stat.st_mtime}")
# 遍历目录
for item in p.iterdir():
print(f"{'[DIR]' if item.is_dir() else '[FILE]'} {item.name}")
if len(list(p.iterdir())) > 5:
break
---
02.文件操作
a.读写文件
Path对象提供便捷的文件读写方法。
b.代码示例
---
# Path文件操作示例
from pathlib import Path
# 创建测试文件
file_path = Path('test.txt')
# 写入文本
file_path.write_text('Hello, World!\n你好，世界！', encoding='utf-8')
print("文件已写入")
# 读取文本
content = file_path.read_text(encoding='utf-8')
print(f"文件内容:\n{content}")
# 写入字节
file_path.write_bytes(b'Binary data')
# 读取字节
data = file_path.read_bytes()
print(f"字节数据: {data}")
# 追加内容（使用open方法）
with file_path.open('a', encoding='utf-8') as f:
f.write('追加的内容\n')
# 删除文件
file_path.unlink()
print("文件已删除")
---
4.3 shutil高级文件操作
01.文件复制
a.复制方法
shutil模块提供高级文件操作功能，包括文件复制、移动、删除等，支持保留元数据。
b.代码示例
---
# shutil文件复制示例
import shutil
from pathlib import Path
# 创建测试文件
src = Path('source.txt')
src.write_text('测试内容')
# 复制文件（不保留元数据）
shutil.copy('source.txt', 'dest1.txt')
print("copy: 文件已复制")
# 复制文件（保留元数据）
shutil.copy2('source.txt', 'dest2.txt')
print("copy2: 文件已复制（保留元数据）")
# 复制文件对象
with open('source.txt', 'rb') as fsrc:
with open('dest3.txt', 'wb') as fdst:
shutil.copyfileobj(fsrc, fdst)
print("copyfileobj: 文件对象已复制")
# 清理测试文件
for f in ['source.txt', 'dest1.txt', 'dest2.txt', 'dest3.txt']:
Path(f).unlink(missing_ok=True)
---
c.目录操作
shutil提供目录树的复制、移动和删除功能。
d.代码示例
---
# shutil目录操作示例
import shutil
from pathlib import Path
# 创建测试目录结构
test_dir = Path('test_source')
test_dir.mkdir(exist_ok=True)
(test_dir / 'file.txt').write_text('内容')
# 复制整个目录树
shutil.copytree('test_source', 'test_dest')
print("目录树已复制")
# 移动文件或目录
shutil.move('test_dest', 'test_moved')
print("目录已移动")
# 删除整个目录树
shutil.rmtree('test_source')
shutil.rmtree('test_moved')
print("目录树已删除")
---
02.磁盘操作
a.磁盘使用情况
shutil提供查询磁盘空间和文件操作的实用函数。
b.代码示例
---
# shutil磁盘操作示例
import shutil
# 获取磁盘使用情况
usage = shutil.disk_usage('.')
print(f"总空间: {usage.total / (1024**3):.2f} GB")
print(f"已使用: {usage.used / (1024**3):.2f} GB")
print(f"可用空间: {usage.free / (1024**3):.2f} GB")
# 查找可执行文件
python_path = shutil.which('python')
print(f"Python路径: {python_path}")
# 获取终端大小
size = shutil.get_terminal_size()
print(f"终端大小: {size.columns}x{size.lines}")
# 修改文件权限（Unix）
# shutil.chown('file.txt', user='username', group='groupname')
---
4.4 glob文件名模式匹配
01.基本模式匹配
a.通配符使用
glob模块使用Unix shell风格的通配符匹配文件名，支持*、?、[]等模式。
b.代码示例
---
# glob基本模式匹配示例
import glob
from pathlib import Path
# 创建测试文件
for i in range(3):
Path(f'test{i}.txt').write_text('test')
Path(f'data{i}.csv').write_text('data')
# 匹配所有txt文件
txt_files = glob.glob('*.txt')
print(f"txt文件: {txt_files}")
# 匹配所有csv文件
csv_files = glob.glob('*.csv')
print(f"csv文件: {csv_files}")
# 使用?匹配单个字符
files = glob.glob('test?.txt')
print(f"test?.txt: {files}")
# 使用[]匹配字符集
files = glob.glob('test[01].txt')
print(f"test[01].txt: {files}")
# 清理测试文件
for f in glob.glob('test*.txt') + glob.glob('data*.csv'):
Path(f).unlink()
---
c.递归匹配
glob支持递归搜索子目录中的文件。
d.代码示例
---
# glob递归匹配示例
import glob
from pathlib import Path
# 创建测试目录结构
Path('dir1/sub1').mkdir(parents=True, exist_ok=True)
Path('dir1/file1.py').write_text('code')
Path('dir1/sub1/file2.py').write_text('code')
# 递归匹配所有py文件
py_files = glob.glob('**/*.py', recursive=True)
print(f"所有py文件: {py_files}")
# 递归匹配特定目录
files = glob.glob('dir1/**/*.py', recursive=True)
print(f"dir1中的py文件: {files}")
# 使用iglob返回迭代器（节省内存）
for file in glob.iglob('**/*.py', recursive=True):
print(f"找到: {file}")
# 清理
import shutil
shutil.rmtree('dir1', ignore_errors=True)
---
02.Path对象匹配
a.pathlib集成
pathlib的Path对象也支持glob模式匹配。
b.代码示例
---
# pathlib glob匹配示例
from pathlib import Path
# 创建测试文件
for i in range(3):
Path(f'item{i}.md').write_text('content')
# 使用Path.glob()
p = Path('.')
md_files = list(p.glob('*.md'))
print(f"md文件: {[f.name for f in md_files]}")
# 递归匹配
all_md = list(p.glob('**/*.md'))
print(f"所有md文件: {len(all_md)}个")
# 使用rglob（等同于**/pattern）
all_md = list(p.rglob('*.md'))
print(f"rglob找到: {len(all_md)}个")
# 匹配多种扩展名
for ext in ['*.md', '*.txt']:
for file in p.glob(ext):
print(f"找到: {file.name}")
# 清理
for f in p.glob('item*.md'):
f.unlink()
---
4.5 tempfile临时文件
01.临时文件创建
a.临时文件对象
tempfile模块提供创建临时文件和目录的功能，自动管理临时资源的生命周期。
b.代码示例
---
# tempfile临时文件示例
import tempfile
# 创建临时文件（自动删除）
with tempfile.TemporaryFile(mode='w+') as tf:
tf.write('临时数据\n')
tf.write('更多内容\n')
tf.seek(0)
content = tf.read()
print(f"临时文件内容: {content}")
# 文件已自动删除
# 创建命名临时文件
with tempfile.NamedTemporaryFile(mode='w+', delete=False) as ntf:
print(f"临时文件路径: {ntf.name}")
ntf.write('命名临时文件')
temp_path = ntf.name
# 手动删除
import os
os.unlink(temp_path)
print("临时文件已删除")
---
c.临时目录
tempfile可以创建临时目录，用于存放多个临时文件。
d.代码示例
---
# tempfile临时目录示例
import tempfile
from pathlib import Path
# 创建临时目录
with tempfile.TemporaryDirectory() as tmpdir:
print(f"临时目录: {tmpdir}")
# 在临时目录中创建文件
temp_file = Path(tmpdir) / 'data.txt'
temp_file.write_text('临时数据')
# 创建子目录
sub_dir = Path(tmpdir) / 'subdir'
sub_dir.mkdir()
# 列出临时目录内容
for item in Path(tmpdir).iterdir():
print(f" {item.name}")
# 目录及其内容已自动删除
print("临时目录已清理")
---
02.临时文件配置
a.自定义临时文件
可以自定义临时文件的位置、前缀、后缀等属性。
b.代码示例
---
# tempfile配置示例
import tempfile
# 获取临时目录位置
temp_dir = tempfile.gettempdir()
print(f"系统临时目录: {temp_dir}")
# 创建带前缀和后缀的临时文件
with tempfile.NamedTemporaryFile(
prefix='myapp_',
suffix='.log',
mode='w+'
) as tf:
print(f"临时文件名: {tf.name}")
tf.write('日志内容\n')
# 在指定目录创建临时文件
with tempfile.NamedTemporaryFile(dir='.', delete=True) as tf:
print(f"当前目录临时文件: {tf.name}")
# 生成临时文件名（不创建文件）
temp_name = tempfile.mktemp(suffix='.tmp')
print(f"临时文件名: {temp_name}")
# 创建唯一临时目录
temp_dir = tempfile.mkdtemp(prefix='test_')
print(f"临时目录: {temp_dir}")
import shutil
shutil.rmtree(temp_dir)
---
5. 数据持久化
5.1 pickle对象序列化
01.基本序列化
a.dump和load
pickle模块实现Python对象的序列化和反序列化，可以保存和恢复复杂的Python对象。
b.代码示例
---
# pickle基本序列化示例
import pickle
# 准备数据
data = {
'name': 'Alice',
'age': 25,
'scores': [85, 90, 92],
'info': {'city': 'Beijing', 'job': 'Engineer'}
}
# 序列化到文件
with open('data.pkl', 'wb') as f:
pickle.dump(data, f)
print("数据已序列化")
# 从文件反序列化
with open('data.pkl', 'rb') as f:
loaded_data = pickle.load(f)
print(f"加载的数据: {loaded_data}")
# 清理
import os
os.unlink('data.pkl')
---
c.dumps和loads
pickle提供字符串形式的序列化和反序列化。
d.代码示例
---
# pickle字符串序列化示例
import pickle
# 创建复杂对象
class Person:
def __init__(self, name, age):
self.name = name
self.age = age
person = Person('Bob', 30)
# 序列化为字节串
serialized = pickle.dumps(person)
print(f"序列化结果: {serialized[:50]}...")
# 从字节串反序列化
restored = pickle.loads(serialized)
print(f"恢复的对象: {restored.name}, {restored.age}")
# 序列化多个对象
data_list = [1, 2, 3, 'test', {'key': 'value'}]
serialized = pickle.dumps(data_list)
restored = pickle.loads(serialized)
print(f"恢复的列表: {restored}")
---
02.高级用法
a.协议版本
pickle支持多个协议版本，版本越高压缩率越好但兼容性越差。
b.代码示例
---
# pickle协议版本示例
import pickle
data = {'numbers': list(range(100)), 'text': 'Hello' * 20}
# 使用不同协议版本
for protocol in range(pickle.HIGHEST_PROTOCOL + 1):
serialized = pickle.dumps(data, protocol=protocol)
print(f"协议{protocol}: {len(serialized)}字节")
# 使用最高协议（推荐）
serialized = pickle.dumps(data, protocol=pickle.HIGHEST_PROTOCOL)
restored = pickle.loads(serialized)
print(f"数据已恢复: {len(restored['numbers'])}个数字")
# 序列化自定义类
class Config:
def __init__(self):
self.settings = {'debug': True, 'port': 8080}
config = Config()
serialized = pickle.dumps(config)
restored_config = pickle.loads(serialized)
print(f"配置: {restored_config.settings}")
---
5.2 shelve对象持久化
01.shelve基本使用
a.字典式接口
shelve提供类似字典的持久化存储接口，基于pickle实现，支持任意Python对象。
b.代码示例
---
# shelve基本使用示例
import shelve
# 打开shelve数据库
with shelve.open('mydata') as db:
# 存储数据（类似字典）
db['user'] = {'name': 'Alice', 'age': 25}
db['scores'] = [85, 90, 92, 88]
db['config'] = {'debug': True, 'port': 8080}
print("数据已存储")
# 读取数据
with shelve.open('mydata') as db:
user = db['user']
scores = db['scores']
print(f"用户: {user}")
print(f"分数: {scores}")
# 列出所有键
print(f"所有键: {list(db.keys())}")
# 清理
import os
for ext in ['', '.db', '.dat', '.bak', '.dir']:
try:
os.unlink(f'mydata{ext}')
except:
pass
---
c.数据操作
shelve支持字典的所有基本操作，包括增删改查。
d.代码示例
---
# shelve数据操作示例
import shelve
with shelve.open('database') as db:
# 添加数据
db['item1'] = 'value1'
db['item2'] = [1, 2, 3]
# 检查键是否存在
if 'item1' in db:
print(f"item1存在: {db['item1']}")
# 更新数据
db['item1'] = 'new_value'
# 删除数据
del db['item2']
# 获取所有键值对
for key in db:
print(f"{key}: {db[key]}")
# 清理
import os
for ext in ['', '.db', '.dat', '.bak', '.dir']:
try:
os.unlink(f'database{ext}')
except:
pass
---
02.高级特性
a.同步和回写
shelve支持同步模式和回写模式，确保数据一致性。
b.代码示例
---
# shelve高级特性示例
import shelve
# 使用writeback模式（自动同步修改）
with shelve.open('cache', writeback=True) as db:
# 存储可变对象
db['list'] = [1, 2, 3]
db['dict'] = {'a': 1, 'b': 2}
# 直接修改（writeback=True时有效）
db['list'].append(4)
db['dict']['c'] = 3
print(f"修改后的列表: {db['list']}")
print(f"修改后的字典: {db['dict']}")
# 重新打开验证
with shelve.open('cache') as db:
print(f"持久化的列表: {db['list']}")
print(f"持久化的字典: {db['dict']}")
# 清理
import os
for ext in ['', '.db', '.dat', '.bak', '.dir']:
try:
os.unlink(f'cache{ext}')
except:
pass
---
5.3 sqlite3 SQLite数据库
01.数据库基础操作
a.连接和创建
sqlite3模块提供轻量级的SQL数据库支持，无需独立服务器，适合嵌入式应用。
b.代码示例
---
# sqlite3基础操作示例
import sqlite3
# 连接数据库（不存在则创建）
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
# 创建表
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
age INTEGER,
email TEXT
)
''')
# 插入数据
cursor.execute("INSERT INTO users (name, age, email) VALUES (?, ?, ?)",
('Alice', 25, 'alice@example.com'))
# 插入多条数据
users = [
('Bob', 30, 'bob@example.com'),
('Charlie', 28, 'charlie@example.com')
]
cursor.executemany("INSERT INTO users (name, age, email) VALUES (?, ?, ?)", users)
conn.commit()
conn.close()
print("数据库操作完成")
---
c.查询数据
sqlite3支持标准SQL查询语句，可以灵活检索数据。
d.代码示例
---
# sqlite3查询示例
import sqlite3
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
# 查询所有数据
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()
for row in rows:
print(f"ID: {row[0]}, Name: {row[1]}, Age: {row[2]}")
# 条件查询
cursor.execute("SELECT * FROM users WHERE age > ?", (25,))
print(f"\n年龄>25的用户: {cursor.fetchall()}")
# 使用Row工厂（字典式访问）
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM users LIMIT 1")
row = cursor.fetchone()
print(f"\n用户名: {row['name']}, 邮箱: {row['email']}")
conn.close()
import os
os.unlink('test.db')
---
02.高级操作
a.事务和优化
sqlite3支持事务管理和性能优化。
b.代码示例
---
# sqlite3高级操作示例
import sqlite3
conn = sqlite3.connect('advanced.db')
cursor = conn.cursor()
# 创建表
cursor.execute('''CREATE TABLE products (
id INTEGER PRIMARY KEY,
name TEXT,
price REAL
)''')
# 使用事务
try:
cursor.execute("INSERT INTO products VALUES (1, 'Book', 29.99)")
cursor.execute("INSERT INTO products VALUES (2, 'Pen', 5.99)")
conn.commit()
print("事务提交成功")
except Exception as e:
conn.rollback()
print(f"事务回滚: {e}")
# 聚合查询
cursor.execute("SELECT COUNT(*), AVG(price) FROM products")
count, avg_price = cursor.fetchone()
print(f"商品数: {count}, 平均价格: {avg_price:.2f}")
conn.close()
import os
os.unlink('advanced.db')
---
5.4 dbm数据库接口
01.dbm基础操作
a.打开数据库
dbm模块提供简单的键值对数据库接口，支持持久化存储字符串数据。
b.代码示例
---
import dbm
# 创建或打开数据库
with dbm.open('mydb', 'c') as db:
# 存储数据
db['name'] = 'Alice'
db['age'] = '25'
db['city'] = 'Beijing'
# 读取数据
print(db['name']) # b'Alice'
# 检查键是否存在
if 'name' in db:
print('Name exists')
# 删除键
del db['age']
# 获取所有键
keys = list(db.keys())
print(keys)
---
02.dbm应用
a.配置存储
使用dbm存储应用配置和简单的键值数据。
b.代码示例
---
import dbm
import json
# 存储复杂数据
with dbm.open('config', 'c') as db:
config = {'host': 'localhost', 'port': 8080}
db['server'] = json.dumps(config)
# 读取复杂数据
data = json.loads(db['server'])
print(data) # {'host': 'localhost', 'port': 8080}
---
5.5 csv CSV文件读写
01.CSV读取
a.reader方法
csv模块提供读写CSV文件的功能，支持自定义分隔符和引号字符。
b.代码示例
---
import csv
# 写入CSV文件
with open('data.csv', 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(['Name', 'Age', 'City'])
writer.writerow(['Alice', 25, 'Beijing'])
writer.writerow(['Bob', 30, 'Shanghai'])
# 读取CSV文件
with open('data.csv', 'r', encoding='utf-8') as f:
reader = csv.reader(f)
for row in reader:
print(row)
# 使用DictReader
with open('data.csv', 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
print(f"{row['Name']}: {row['Age']}")
---
02.CSV写入
a.DictWriter方法
DictWriter允许使用字典写入CSV文件，更加直观和易用。
b.代码示例
---
import csv
# 使用DictWriter写入
data = [
{'name': 'Alice', 'age': 25, 'city': 'Beijing'},
{'name': 'Bob', 'age': 30, 'city': 'Shanghai'}
]
with open('output.csv', 'w', newline='', encoding='utf-8') as f:
fieldnames = ['name', 'age', 'city']
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(data)
---
6. 数据压缩与归档
6.1 zlib压缩
01.数据压缩
a.压缩和解压
zlib模块提供数据压缩和解压缩功能，支持不同的压缩级别和格式。
b.代码示例
---
import zlib
# 原始数据
data = b"Python zlib compression example with repeated text." * 100
print(f"Original size: {len(data)} bytes")
# 压缩数据
compressed = zlib.compress(data, level=9)
print(f"Compressed size: {len(compressed)} bytes")
print(f"Compression ratio: {len(compressed)/len(data):.2%}")
# 解压数据
decompressed = zlib.decompress(compressed)
print(f"Decompressed size: {len(decompressed)} bytes")
# 验证数据完整性
print(f"Data matches: {data == decompressed}")
# 逐步压缩
compressor = zlib.compressobj(level=6)
chunks = []
for i in range(0, len(data), 1024):
chunk = data[i:i+1024]
compressed_chunk = compressor.compress(chunk)
if compressed_chunk:
chunks.append(compressed_chunk)
# 完成压缩
chunks.append(compressor.flush())
result = b''.join(chunks)
print(f"Streaming compression size: {len(result)} bytes")
---
02.数据流处理
a.流式压缩
使用compressobj和decompressobj进行流式数据压缩和解压缩。
b.代码示例
---
import zlib
# 大数据流式压缩
def compress_stream(data, chunk_size=1024):
compressor = zlib.compressobj()
compressed_chunks = []
for i in range(0, len(data), chunk_size):
chunk = data[i:i+chunk_size]
compressed_chunk = compressor.compress(chunk)
if compressed_chunk:
compressed_chunks.append(compressed_chunk)
# 添加压缩结束标记
compressed_chunks.append(compressor.flush())
return b''.join(compressed_chunks)
def decompress_stream(compressed_data, chunk_size=1024):
decompressor = zlib.decompressobj()
decompressed_chunks = []
for i in range(0, len(compressed_data), chunk_size):
chunk = compressed_data[i:i+chunk_size]
try:
decompressed_chunk = decompressor.decompress(chunk)
decompressed_chunks.append(decompressed_chunk)
except zlib.error:
pass
# 处理剩余数据
decompressed_chunks.append(decompressor.flush())
return b''.join(decompressed_chunks)
# 测试流式压缩
large_data = b"Large data for compression testing." * 1000
compressed = compress_stream(large_data)
decompressed = decompress_stream(compressed)
print(f"Stream test passed: {large_data == decompressed}")
---
6.2 gzip文件压缩
01.文件压缩
a.GzipFile对象
gzip模块提供文件级别的压缩功能，可以直接读写压缩文件。
b.代码示例
---
import gzip
os
# 创建原始文本文件
text_data = "Python gzip compression example.\n" * 100
with open('original.txt', 'w', encoding='utf-8') as f:
f.write(text_data)
# 压缩文件
with open('original.txt', 'rb') as f_in:
with gzip.open('compressed.gz', 'wb') as f_out:
f_out.write(f_in.read())
# 检查文件大小
original_size = os.path.getsize('original.txt')
compressed_size = os.path.getsize('compressed.gz')
print(f"Original: {original_size} bytes")
print(f"Compressed: {compressed_size} bytes")
print(f"Ratio: {compressed_size/original_size:.2%}")
# 读取压缩文件
with gzip.open('compressed.gz', 'rt', encoding='utf-8') as f:
content = f.read()
print(f"Content length: {len(content)}")
# 追加到压缩文件
with gzip.open('compressed.gz', 'at', encoding='utf-8') as f:
f.write("\nAdditional line appended to gz file.")
# 读取追加后的内容
with gzip.open('compressed.gz', 'rt', encoding='utf-8') as f:
lines = f.readlines()
print(f"Total lines: {len(lines)}")
# 清理文件
os.remove('original.txt')
os.remove('compressed.gz')
---
02.流式压缩
a.Gzip压缩器
使用GzipFile对象进行流式压缩，适合处理大数据。
b.代码示例
---
import gzip
io
# 内存流压缩
buffer = io.BytesIO()
with gzip.GzipFile(fileobj=buffer, mode='wb') as gz_file:
gz_file.write(b"Large data for compression " * 1000)
compressed_data = buffer.getvalue()
print(f"Compressed in memory: {len(compressed_data)} bytes")
# 内存解压
buffer.seek(0)
with gzip.GzipFile(fileobj=buffer, mode='rb') as gz_file:
decompressed = gz_file.read()
print(f"Decompressed size: {len(decompressed)} bytes")
# 批量文件压缩
import tempfile
import shutil
def compress_file_list(file_paths, output_path):
"""压缩多个文件到一个gzip文件"""
with gzip.open(output_path, 'wb') as gz_out:
for file_path in file_paths:
if os.path.exists(file_path):
with open(file_path, 'rb') as f_in:
shutil.copyfileobj(f_in, gz_out)
# 添加分隔符
gz_out.write(b'\n--- FILE SEPARATOR ---\n')
# 创建临时文件测试
temp_files = []
for i in range(3):
temp_file = f"temp_{i}.txt"
with open(temp_file, 'w') as f:
f.write(f"Content of file {i}\n" * 10)
temp_files.append(temp_file)
# 压缩所有临时文件
compress_file_list(temp_files, 'all_files.gz')
# 读取压缩文件并显示信息
with gzip.open('all_files.gz', 'rt') as f:
content = f.read()
print(f"Compressed content contains {content.count('FILE SEPARATOR')} files")
# 清理临时文件
for f in temp_files:
os.remove(f)
os.remove('all_files.gz')
---
6.3 bz2 bzip2压缩
01.基本压缩与解压缩
a.使用bz2模块进行字符串压缩 - 提供高效的bzip2压缩算法实现
b.代码示例
---
import bz2
import os
# 原始数据
original_data = "这是一个需要压缩的长字符串。" * 50
print(f"原始数据长度: {len(original_data)} 字符")
# 压缩数据
compressed_data = bz2.compress(original_data.encode('utf-8'))
print(f"压缩后长度: {len(compressed_data)} 字节")
print(f"压缩率: {len(compressed_data)/len(original_data.encode('utf-8')):.2%}")
# 解压缩数据
decompressed_data = bz2.decompress(compressed_data).decode('utf-8')
print(f"解压后长度: {len(decompressed_data)} 字符")
print(f"数据完整性验证: {original_data == decompressed_data}")
# 使用不同的压缩级别
for level in [1, 5, 9]:
compressed = bz2.compress(original_data.encode('utf-8'), compresslevel=level)
print(f"压缩级别 {level}: {len(compressed)} 字节")
---
c.文件压缩与解压缩
直接操作文件进行bzip2压缩
d.代码示例
---
import bz2
import tempfile
import shutil
# 创建临时文本文件
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as f:
f.write("Python bzip2文件压缩示例\n" * 100)
original_file = f.name
# 压缩文件
compressed_file = original_file + '.bz2'
with open(original_file, 'rb') as f_in:
with bz2.open(compressed_file, 'wb') as f_out:
shutil.copyfileobj(f_in, f_out)
# 检查文件大小
original_size = os.path.getsize(original_file)
compressed_size = os.path.getsize(compressed_file)
print(f"原始文件大小: {original_size} 字节")
print(f"压缩文件大小: {compressed_size} 字节")
print(f"压缩率: {compressed_size/original_size:.2%}")
# 读取压缩文件内容
with bz2.open(compressed_file, 'rt', encoding='utf-8') as f:
content = f.read()
print(f"读取内容长度: {len(content)} 字符")
# 清理临时文件
os.unlink(original_file)
os.unlink(compressed_file)
---
02.流式压缩处理
a.增量式压缩处理
处理大文件时使用流式压缩避免内存溢出
b.代码示例
---
import bz2
import tempfile
import os
# 创建大文本文件用于演示
large_text = "这是一行文本内容\n" * 10000
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as f:
f.write(large_text)
large_file = f.name
# 流式压缩处理
compressed_file = large_file + '.bz2'
buffer_size = 1024 # 1KB缓冲区
with open(large_file, 'rb') as f_in:
with bz2.open(compressed_file, 'wb') as f_out:
while True:
chunk = f_in.read(buffer_size)
if not chunk:
break
f_out.write(chunk)
print(f"原始文件大小: {os.path.getsize(large_file)} 字节")
print(f"压缩文件大小: {os.path.getsize(compressed_file)} 字节")
# 流式解压缩验证
decompressed_file = large_file + '_decompressed.txt'
with bz2.open(compressed_file, 'rb') as f_in:
with open(decompressed_file, 'wb') as f_out:
while True:
chunk = f_in.read(buffer_size)
if not chunk:
break
f_out.write(chunk)
# 验证文件内容
with open(large_file, 'r', encoding='utf-8') as f1, \
open(decompressed_file, 'r', encoding='utf-8') as f2:
content1 = f1.read()
content2 = f2.read()
print(f"内容验证: {content1 == content2}")
# 清理临时文件
os.unlink(large_file)
os.unlink(compressed_file)
os.unlink(decompressed_file)
---
c.BZ2File类使用
面向对象的方式处理bzip2压缩
d.代码示例
---
import bz2
import tempfile
import io
# 使用BZ2File进行内存压缩
original_data = "使用BZ2File类进行压缩处理的示例数据" * 100
# 创建内存中的压缩文件
compressed_buffer = io.BytesIO()
with bz2.BZ2File(compressed_buffer, 'wb', compresslevel=9) as bz2_file:
bz2_file.write(original_data.encode('utf-8'))
compressed_data = compressed_buffer.getvalue()
print(f"压缩数据大小: {len(compressed_data)} 字节")
# 解压缩数据
compressed_buffer.seek(0) # 重置缓冲区位置
with bz2.BZ2File(compressed_buffer, 'rb') as bz2_file:
decompressed_data = bz2_file.read().decode('utf-8')
print(f"解压数据长度: {len(decompressed_data)} 字符")
print(f"数据完整性: {original_data == decompressed_data}")
# 使用BZ2File处理文件
with tempfile.NamedTemporaryFile(suffix='.bz2', delete=False) as f:
temp_compressed = f.name
# 写入压缩数据
with bz2.BZ2File(temp_compressed, 'wb', compresslevel=5) as bz2_file:
bz2_file.write(original_data.encode('utf-8'))
# 读取压缩数据
with bz2.BZ2File(temp_compressed, 'rb') as bz2_file:
read_data = bz2_file.read().decode('utf-8')
print(f"文件压缩验证: {original_data == read_data}")
# 清理临时文件
os.unlink(temp_compressed)
---
6.4 zipfile ZIP归档
01.基本ZIP操作
a.创建ZIP文件
将多个文件和目录打包成ZIP格式归档
b.代码示例
---
import zipfile
import os
import tempfile
from datetime import datetime
# 创建临时文件用于演示
with tempfile.TemporaryDirectory() as temp_dir:
# 创建一些测试文件
test_files = []
for i, content in enumerate(["文件内容1", "文件内容2", "文件内容3"]):
file_path = os.path.join(temp_dir, f"test_file_{i+1}.txt")
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content * 50)
test_files.append(file_path)
# 创建ZIP文件
zip_path = os.path.join(temp_dir, "archive.zip")
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
# 添加文件到ZIP
for file_path in test_files:
arcname = os.path.basename(file_path)
zipf.write(file_path, arcname)
print(f"已添加: {arcname}")
# 显示ZIP文件信息
print(f"ZIP文件大小: {os.path.getsize(zip_path)} 字节")
# 读取ZIP文件内容列表
with zipfile.ZipFile(zip_path, 'r') as zipf:
print("\nZIP文件内容:")
for info in zipf.infolist():
print(f" {info.filename} - {info.file_size} 字节 - 压缩率: {(1-info.compress_size/info.file_size):.1%}")
# 计算原始文件总大小
total_original = sum(os.path.getsize(f) for f in test_files)
total_compressed = os.path.getsize(zip_path)
print(f"\n总体压缩率: {(1-total_compressed/total_original):.1%}")
---
c.解压ZIP文件
从ZIP归档中提取文件和目录
d.代码示例
---
import zipfile
import os
import tempfile
# 创建包含多个文件的ZIP
with tempfile.TemporaryDirectory() as temp_dir:
zip_path = os.path.join(temp_dir, "test_archive.zip")
# 创建ZIP文件并添加内容
with zipfile.ZipFile(zip_path, 'w') as zipf:
# 添加文本文件
zipf.writestr("readme.txt", "这是一个说明文件\n包含重要信息")
# 添加配置文件
config_content = """[settings]
debug=true
port=8080
host=localhost
"""
zipf.writestr("config/settings.ini", config_content)
# 添加数据文件
data = ["数据行1", "数据行2", "数据行3", "数据行4", "数据行5"]
zipf.writestr("data/sample.csv", "\n".join(data))
print(f"创建的ZIP文件: {zip_path}")
# 解压所有文件
extract_dir = os.path.join(temp_dir, "extracted")
with zipfile.ZipFile(zip_path, 'r') as zipf:
zipf.extractall(extract_dir)
print(f"已解压到: {extract_dir}")
# 验证解压的文件
print("\n解压后的文件:")
for root, dirs, files in os.walk(extract_dir):
for file in files:
file_path = os.path.join(root, file)
rel_path = os.path.relpath(file_path, extract_dir)
print(f" {rel_path} ({os.path.getsize(file_path)} 字节)")
# 选择性解压特定文件
selective_dir = os.path.join(temp_dir, "selective")
os.makedirs(selective_dir, exist_ok=True)
with zipfile.ZipFile(zip_path, 'r') as zipf:
# 只解压配置文件
if "config/settings.ini" in zipf.namelist():
config_data = zipf.read("config/settings.ini")
with open(os.path.join(selective_dir, "settings.ini"), 'wb') as f:
f.write(config_data)
print(f"\n已提取配置文件到: {selective_dir}")
---
02.高级ZIP功能
a.密码保护和压缩设置
为ZIP文件添加密码和配置压缩参数
b.代码示例
---
import zipfile
import tempfile
import os
# 创建带密码保护的ZIP文件
with tempfile.TemporaryDirectory() as temp_dir:
# 创建测试文件
files_data = {
"secret.txt": "这是机密文件内容\n需要密码保护",
"public.txt": "这是公开文件内容\n无需密码",
"docs/manual.txt": "用户手册\n版本1.0"
}
# 创建普通ZIP文件
normal_zip = os.path.join(temp_dir, "normal.zip")
with zipfile.ZipFile(normal_zip, 'w', zipfile.ZIP_DEFLATED, compresslevel=6) as zipf:
for filename, content in files_data.items():
zipf.writestr(filename, content)
# 创建密码保护的ZIP文件（需要使用第三方库如pyzipper）
# 这里演示基本的ZIP创建
protected_zip = os.path.join(temp_dir, "protected.zip")
with zipfile.ZipFile(protected_zip, 'w', zipfile.ZIP_DEFLATED, compresslevel=9) as zipf:
for filename, content in files_data.items():
if "secret" in filename:
# 对于敏感文件，使用更高压缩级别
zipf.writestr(filename, content, compress_type=zipfile.ZIP_DEFLATED, compresslevel=9)
else:
zipf.writestr(filename, content)
# 比较不同压缩设置的效果
print("压缩效果对比:")
for zip_file, description in [(normal_zip, "普通压缩"), (protected_zip, "高级压缩")]:
size = os.path.getsize(zip_file)
print(f"{description}: {size} 字节")
# 读取ZIP文件详细信息
print("\n文件详细信息:")
with zipfile.ZipFile(protected_zip, 'r') as zipf:
for info in zipf.infolist():
ratio = (1 - info.compress_size / info.file_size) * 100
print(f"{info.filename}: {info.file_size} -> {info.compress_size} 字节 (压缩率: {ratio:.1f}%)")
print(f" 修改时间: {datetime.fromtimestamp(info.date_time.timestamp()):%Y-%m-%d %H:%M:%S}")
print(f" CRC32: {info.CRC:08X}")
---
c.追加和更新ZIP内容
向现有ZIP文件添加新文件或替换现有文件
d.代码示例
---
import zipfile
import tempfile
import os
import shutil
# 演示ZIP文件的追加和更新
with tempfile.TemporaryDirectory() as temp_dir:
initial_zip = os.path.join(temp_dir, "initial.zip")
# 创建初始ZIP文件
with zipfile.ZipFile(initial_zip, 'w', zipfile.ZIP_DEFLATED) as zipf:
zipf.writestr("version.txt", "版本: 1.0\n创建时间: 2024-01-01")
zipf.writestr("data/initial.dat", "初始数据内容" * 10)
print(f"初始ZIP文件: {os.path.getsize(initial_zip)} 字节")
# 追加新文件到ZIP
updated_zip = os.path.join(temp_dir, "updated.zip")
# 复制原ZIP内容并添加新文件
with zipfile.ZipFile(initial_zip, 'r') as source_zip:
with zipfile.ZipFile(updated_zip, 'w', zipfile.ZIP_DEFLATED) as target_zip:
# 复制所有原有文件
for item in source_zip.infolist():
data = source_zip.read(item.filename)
target_zip.writestr(item, data)
# 添加新文件
target_zip.writestr("changelog.txt", "版本更新日志:\nv1.1: 添加新功能\nv1.2: 修复bug")
target_zip.writestr("data/new_feature.dat", "新功能数据" * 15)
print(f"更新后ZIP文件: {os.path.getsize(updated_zip)} 字节")
# 比较两个ZIP文件的内容
print("\n文件内容对比:")
for zip_file, label in [(initial_zip, "初始版本"), (updated_zip, "更新版本")]:
print(f"\n{label}:")
with zipfile.ZipFile(zip_file, 'r') as zipf:
for info in zipf.infolist():
print(f" {info.filename} ({info.file_size} 字节)")
# 创建增量更新ZIP
incremental_zip = os.path.join(temp_dir, "incremental.zip")
with zipfile.ZipFile(incremental_zip, 'w', zipfile.ZIP_DEFLATED) as zipf:
# 只添加新增或修改的文件
zipf.writestr("updates/patch_v1.3.dat", "补丁数据内容")
zipf.writestr("version.txt", "版本: 1.3\n更新时间: 2024-01-15")
print(f"\n增量更新ZIP: {os.path.getsize(incremental_zip)} 字节")
---
6.5 tarfile TAR归档
01.基本TAR操作
a.创建TAR文件
将多个文件和目录打包成TAR格式归档
b.代码示例
---
import tarfile
import os
import tempfile
from datetime import datetime
# 创建临时文件和目录用于演示
with tempfile.TemporaryDirectory() as temp_dir:
# 创建目录结构
project_dir = os.path.join(temp_dir, "project")
os.makedirs(os.path.join(project_dir, "src"))
os.makedirs(os.path.join(project_dir, "docs"))
os.makedirs(os.path.join(project_dir, "data"))
# 创建项目文件
files_to_create = [
("README.md", "# 项目说明\n这是一个示例项目"),
("src/main.py", "def main():\n print('Hello World')\nif __name__ == '__main__':\n main()"),
("src/utils.py", "def helper():\n return '辅助函数'"),
("docs/manual.txt", "用户手册\n版本1.0"),
("data/config.json", '{"debug": true, "port": 8080}'),
("data/sample.txt", "示例数据\n第一行\n第二行\n第三行")
]
for file_path, content in files_to_create:
full_path = os.path.join(project_dir, file_path)
with open(full_path, 'w', encoding='utf-8') as f:
f.write(content)
# 创建未压缩的TAR文件
tar_path = os.path.join(temp_dir, "project.tar")
with tarfile.open(tar_path, 'w') as tar:
tar.add(project_dir, arcname=os.path.basename(project_dir))
print(f"创建TAR文件: {tar_path}")
print(f"TAR文件大小: {os.path.getsize(tar_path)} 字节")
# 列出TAR文件内容
with tarfile.open(tar_path, 'r') as tar:
print("\nTAR文件内容:")
for member in tar.getmembers():
size_str = f"{member.size} 字节" if member.isreg() else "目录"
print(f" {member.name} ({size_str})")
# 计算原始目录大小
total_size = 0
for root, dirs, files in os.walk(project_dir):
for file in files:
file_path = os.path.join(root, file)
total_size += os.path.getsize(file_path)
print(f"\n原始目录总大小: {total_size} 字节")
print(f"TAR文件大小: {os.path.getsize(tar_path)} 字节")
print(f"存储效率: {(1-os.path.getsize(tar_path)/total_size)*100:.1f}%")
---
c.解压TAR文件
从TAR归档中提取文件和目录结构
d.代码示例
---
import tarfile
import os
import tempfile
# 创建包含多种文件类型的TAR
with tempfile.TemporaryDirectory() as temp_dir:
# 创建源文件
source_dir = os.path.join(temp_dir, "source")
os.makedirs(source_dir)
source_files = {
"document.txt": "重要文档内容\n包含多行文本信息",
"config/settings.cfg": "[general]\ndebug = true\nlanguage = zh_CN",
"scripts/backup.sh": "#!/bin/bash\necho '开始备份...'",
"logs/app.log": "2024-01-01 10:00:00 INFO 应用启动\n2024-01-01 10:01:00 ERROR 连接失败"
}
for file_path, content in source_files.items():
full_path = os.path.join(source_dir, file_path)
os.makedirs(os.path.dirname(full_path), exist_ok=True)
with open(full_path, 'w', encoding='utf-8') as f:
f.write(content)
# 创建TAR文件
tar_file = os.path.join(temp_dir, "backup.tar")
with tarfile.open(tar_file, 'w') as tar:
tar.add(source_dir, arcname="backup_data")
print(f"创建TAR文件: {tar_file}")
# 解压所有文件
extract_dir = os.path.join(temp_dir, "extracted")
with tarfile.open(tar_file, 'r') as tar:
tar.extractall(extract_dir)
print(f"已解压到: {extract_dir}")
# 验证解压结果
print("\n解压后的目录结构:")
for root, dirs, files in os.walk(extract_dir):
level = root.replace(extract_dir, '').count(os.sep)
indent = ' ' * 2 * level
print(f"{indent}{os.path.basename(root)}/")
subindent = ' ' * 2 * (level + 1)
for file in files:
file_path = os.path.join(root, file)
size = os.path.getsize(file_path)
print(f"{subindent}{file} ({size} 字节)")
# 选择性解压特定文件
selective_dir = os.path.join(temp_dir, "selective")
os.makedirs(selective_dir, exist_ok=True)
with tarfile.open(tar_file, 'r') as tar:
# 只解压配置文件
config_members = [m for m in tar.getmembers() if 'config' in m.name]
for member in config_members:
if member.isreg():
tar.extract(member, path=selective_dir)
print(f"\n已提取配置文件: {member.name}")
---
02.压缩TAR文件
a.创建压缩TAR归档
使用gzip/bzip2/xz等压缩算法压缩TAR文件
b.代码示例
---
import tarfile
import os
import tempfile
import gzip
import bz2
# 比较不同压缩格式的效果
with tempfile.TemporaryDirectory() as temp_dir:
# 创建大量文件用于测试
test_dir = os.path.join(temp_dir, "test_data")
os.makedirs(test_dir)
# 生成大量文本数据
for i in range(100):
file_path = os.path.join(test_dir, f"data_{i:03d}.txt")
with open(file_path, 'w', encoding='utf-8') as f:
f.write(f"测试文件 {i}\n" + "重复数据内容" * 50 + "\n")
# 计算原始大小
original_size = 0
for root, dirs, files in os.walk(test_dir):
for file in files:
original_size += os.path.getsize(os.path.join(root, file))
print(f"原始数据总大小: {original_size:,} 字节")
# 创建不同格式的压缩TAR文件
formats = [
('tar', 'w'), # 无压缩
('tar.gz', 'w:gz'), # gzip压缩
('tar.bz2', 'w:bz2'), # bzip2压缩
('tar.xz', 'w:xz') # xz压缩
]
compression_results = []
for ext, mode in formats:
tar_path = os.path.join(temp_dir, f"archive.{ext}")
start_time = datetime.now()
with tarfile.open(tar_path, mode) as tar:
tar.add(test_dir, arcname="data")
compress_time = (datetime.now() - start_time).total_seconds()
compressed_size = os.path.getsize(tar_path)
compression_ratio = (1 - compressed_size / original_size) * 100
compression_results.append({
'format': ext,
'size': compressed_size,
'ratio': compression_ratio,
'time': compress_time
})
# 显示压缩结果比较
print("\n压缩格式比较:")
print(f"{'格式':<8} {'大小(字节)':<12} {'压缩率':<8} {'耗时(秒)':<10}")
print("-" * 45)
for result in compression_results:
print(f"{result['format']:<8} {result['size']:<12,} {result['ratio']:<7.1f}% {result['time']:<9.3f}")
# 测试解压性能
print("\n解压性能测试:")
for ext, mode in formats:
if ext != 'tar': # 跳过无压缩格式
tar_path = os.path.join(temp_dir, f"archive.{ext}")
extract_dir = os.path.join(temp_dir, f"extract_{ext.replace('.', '_')}")
start_time = datetime.now()
read_mode = mode.replace('w', 'r')
with tarfile.open(tar_path, read_mode) as tar:
tar.extractall(extract_dir)
extract_time = (datetime.now() - start_time).total_seconds()
print(f"{ext:<8} 解压耗时: {extract_time:.3f} 秒")
---
c.增量备份和差异备份
实现基于时间的智能备份策略
d.代码示例
---
import tarfile
import os
import tempfile
import time
import json
from datetime import datetime
# 实现增量备份功能
class IncrementalBackup:
def __init__(self, base_dir):
self.base_dir = base_dir
self.manifest_file = os.path.join(base_dir, "backup_manifest.json")
self.load_manifest()
def load_manifest(self):
"""加载备份清单"""
try:
with open(self.manifest_file, 'r') as f:
self.manifest = json.load(f)
except FileNotFoundError:
self.manifest = {}
def save_manifest(self):
"""保存备份清单"""
with open(self.manifest_file, 'w') as f:
json.dump(self.manifest, f, indent=2)
def scan_directory(self, directory):
"""扫描目录，返回文件信息字典"""
file_info = {}
for root, dirs, files in os.walk(directory):
for file in files:
file_path = os.path.join(root, file)
rel_path = os.path.relpath(file_path, directory)
stat = os.stat(file_path)
file_info[rel_path] = {
'mtime': stat.st_mtime,
'size': stat.st_size
}
return file_info
def create_incremental_backup(self, source_dir, backup_path, backup_type='incremental'):
"""创建增量备份"""
current_files = self.scan_directory(source_dir)
if backup_type == 'full':
# 完整备份
files_to_backup = current_files.keys()
else:
# 增量备份：只包含新增或修改的文件
files_to_backup = []
for file_path, info in current_files.items():
if (file_path not in self.manifest or
info['mtime'] > self.manifest[file_path]['mtime'] or
info['size'] != self.manifest[file_path]['size']):
files_to_backup.append(file_path)
if files_to_backup:
# 创建压缩的TAR备份
with tarfile.open(backup_path, 'w:gz') as tar:
for file_path in files_to_backup:
full_path = os.path.join(source_dir, file_path)
tar.add(full_path, arcname=file_path)
# 更新清单
for file_path in files_to_backup:
self.manifest[file_path] = current_files[file_path]
self.save_manifest()
return len(files_to_backup)
else:
return 0
# 演示增量备份
with tempfile.TemporaryDirectory() as temp_dir:
# 创建测试数据
source_dir = os.path.join(temp_dir, "source")
os.makedirs(source_dir)
# 初始文件
initial_files = {
"doc1.txt": "初始文档内容",
"data1.csv": "列1,列2\n值1,值2\n值3,值4"
}
for file_path, content in initial_files.items():
full_path = os.path.join(source_dir, file_path)
with open(full_path, 'w', encoding='utf-8') as f:
f.write(content)
# 创建备份管理器
backup_dir = os.path.join(temp_dir, "backups")
os.makedirs(backup_dir)
backup_manager = IncrementalBackup(backup_dir)
# 第一次完整备份
full_backup = os.path.join(backup_dir, "backup_full.tar.gz")
file_count = backup_manager.create_incremental_backup(
source_dir, full_backup, 'full'
)
print(f"完整备份: {file_count} 个文件, 大小: {os.path.getsize(full_backup)} 字节")
# 等待一秒确保时间戳不同
time.sleep(1)
# 修改现有文件并添加新文件
with open(os.path.join(source_dir, "doc1.txt"), 'a', encoding='utf-8') as f:
f.write("\n新增的内容行")
with open(os.path.join(source_dir, "new_file.txt"), 'w', encoding='utf-8') as f:
f.write("新创建的文件内容")
# 创建增量备份
incremental_backup = os.path.join(backup_dir, "backup_incremental.tar.gz")
file_count = backup_manager.create_incremental_backup(
source_dir, incremental_backup, 'incremental'
)
print(f"增量备份: {file_count} 个文件, 大小: {os.path.getsize(incremental_backup)} 字节")
# 再次修改并创建另一个增量备份
time.sleep(1)
with open(os.path.join(source_dir, "data1.csv"), 'w', encoding='utf-8') as f:
f.write("列1,列2\n修改值1,修改值2\n修改值3,修改值4")
incremental_backup2 = os.path.join(backup_dir, "backup_incremental2.tar.gz")
file_count = backup_manager.create_incremental_backup(
source_dir, incremental_backup2, 'incremental'
)
print(f"第二次增量备份: {file_count} 个文件, 大小: {os.path.getsize(incremental_backup2)} 字节")
---
7. 网络编程
7.1 socket套接字
01.TCP套接字编程
a.创建TCP服务器
监听指定端口接受客户端连接
b.代码示例
---
import socket
import threading
import time
class TCPServer:
def __init__(self, host='localhost', port=8888):
self.host = host
self.port = port
self.socket = None
self.running = False
def start(self):
"""启动TCP服务器"""
try:
# 创建TCP套接字
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 设置地址重用
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# 绑定地址和端口
self.socket.bind((self.host, self.port))
# 开始监听，最多5个连接排队
self.socket.listen(5)
self.running = True
print(f"TCP服务器启动在 {self.host}:{self.port}")
# 接受客户端连接
while self.running:
try:
client_socket, client_address = self.socket.accept()
print(f"客户端连接: {client_address}")
# 为每个客户端创建新线程
client_thread = threading.Thread(
target=self.handle_client,
args=(client_socket, client_address)
)
client_thread.daemon = True
client_thread.start()
except OSError:
if self.running:
print("接受连接时出错")
break
except Exception as e:
print(f"服务器启动失败: {e}")
finally:
self.stop()
def handle_client(self, client_socket, client_address):
"""处理客户端连接"""
try:
# 发送欢迎消息
welcome_msg = f"欢迎连接到服务器! 你的地址: {client_address}\n"
client_socket.send(welcome_msg.encode('utf-8'))
# 接收客户端数据
while True:
data = client_socket.recv(1024)
if not data:
break
message = data.decode('utf-8').strip()
print(f"来自 {client_address}: {message}")
# 处理特殊命令
if message.lower() == 'quit':
response = "再见!\n"
client_socket.send(response.encode('utf-8'))
break
elif message.lower() == 'time':
response = f"服务器时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n"
elif message.lower() == 'help':
response = "可用命令: time, quit, help\n"
else:
response = f"服务器收到: {message}\n"
client_socket.send(response.encode('utf-8'))
except Exception as e:
print(f"处理客户端 {client_address} 时出错: {e}")
finally:
client_socket.close()
print(f"客户端 {client_address} 断开连接")
def stop(self):
"""停止服务器"""
self.running = False
if self.socket:
self.socket.close()
print("TCP服务器已停止")
# 示例使用
if __name__ == "__main__":
server = TCPServer('localhost', 8888)
try:
server.start()
except KeyboardInterrupt:
print("\n正在停止服务器...")
server.stop()
---
c.创建TCP客户端
连接到服务器并发送接收数据
d.代码示例
---
import socket
import threading
import time
class TCPClient:
def __init__(self, host='localhost', port=8888):
self.host = host
self.port = port
self.socket = None
self.connected = False
def connect(self):
"""连接到服务器"""
try:
# 创建TCP套接字
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 设置超时时间为5秒
self.socket.settimeout(5)
# 连接到服务器
self.socket.connect((self.host, self.port))
self.connected = True
print(f"已连接到服务器 {self.host}:{self.port}")
# 接收欢迎消息
welcome_msg = self.socket.recv(1024).decode('utf-8')
print(f"服务器消息: {welcome_msg.strip()}")
return True
except Exception as e:
print(f"连接失败: {e}")
return False
def send_message(self, message):
"""发送消息到服务器"""
if not self.connected:
print("未连接到服务器")
return False
try:
self.socket.send(message.encode('utf-8'))
# 接收服务器响应
response = self.socket.recv(1024).decode('utf-8')
print(f"服务器响应: {response.strip()}")
return True
except Exception as e:
print(f"发送消息失败: {e}")
self.connected = False
return False
def interactive_mode(self):
"""交互模式"""
print("交互模式 (输入 'quit' 退出):")
print("可用命令: time, help, quit")
while self.connected:
try:
message = input("客户端> ")
if not message:
continue
if message.lower() == 'quit':
self.send_message(message)
break
else:
self.send_message(message)
except KeyboardInterrupt:
print("\n正在断开连接...")
break
def disconnect(self):
"""断开连接"""
if self.socket:
self.socket.close()
self.connected = False
print("已断开连接")
# 示例使用
if __name__ == "__main__":
client = TCPClient('localhost', 8888)
if client.connect():
try:
# 发送一些测试消息
test_messages = ["hello", "time", "help", "quit"]
for msg in test_messages:
if client.send_message(msg):
if msg.lower() == 'quit':
break
time.sleep(1)
# 或者使用交互模式
# client.interactive_mode()
finally:
client.disconnect()
---
02.UDP套接字编程
a.UDP数据报通信
无连接的数据传输方式
b.代码示例
---
import socket
import threading
import time
import json
class UDPServer:
def __init__(self, host='localhost', port=9999):
self.host = host
self.port = port
self.socket = None
self.running = False
self.client_list = {} # 存储客户端信息
def start(self):
"""启动UDP服务器"""
try:
# 创建UDP套接字
self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 绑定地址和端口
self.socket.bind((self.host, self.port))
self.running = True
print(f"UDP服务器启动在 {self.host}:{self.port}")
# 接收数据报
while self.running:
try:
# 接收数据和客户端地址
data, client_address = self.socket.recvfrom(1024)
message = data.decode('utf-8')
print(f"收到来自 {client_address}: {message}")
# 处理消息并回复
response = self.process_message(message, client_address)
if response:
self.socket.sendto(response.encode('utf-8'), client_address)
except socket.timeout:
continue
except Exception as e:
if self.running:
print(f"接收数据时出错: {e}")
break
except Exception as e:
print(f"服务器启动失败: {e}")
finally:
self.stop()
def process_message(self, message, client_address):
"""处理收到的消息"""
try:
# 尝试解析JSON消息
data = json.loads(message)
command = data.get('command', '')
if command == 'register':
# 客户端注册
client_id = data.get('client_id')
self.client_list[client_address] = {
'id': client_id,
'last_seen': time.time()
}
response = json.dumps({
'status': 'success',
'message': f'客户端 {client_id} 注册成功'
})
elif command == 'list':
# 获取客户端列表
clients = [
{'address': addr, **info}
for addr, info in self.client_list.items()
]
response = json.dumps({
'status': 'success',
'clients': clients
})
elif command == 'ping':
# 心跳检测
if client_address in self.client_list:
self.client_list[client_address]['last_seen'] = time.time()
response = json.dumps({
'status': 'success',
'message': 'pong',
'timestamp': time.time()
})
else:
response = json.dumps({
'status': 'error',
'message': '未知命令'
})
except json.JSONDecodeError:
# 处理普通文本消息
if message.lower() == 'time':
response = f"服务器时间: {time.strftime('%Y-%m-%d %H:%M:%S')}"
elif message.lower() == 'clients':
count = len(self.client_list)
response = f"当前连接客户端数: {count}"
else:
response = f"服务器收到: {message}"
return response
def stop(self):
"""停止服务器"""
self.running = False
if self.socket:
self.socket.close()
print("UDP服务器已停止")
# 示例使用
if __name__ == "__main__":
server = UDPServer('localhost', 9999)
try:
server.start()
except KeyboardInterrupt:
print("\n正在停止服务器...")
server.stop()
---
c.UDP客户端实现
发送数据报并接收响应
d.代码示例
---
import socket
import time
import json
import threading
class UDPClient:
def __init__(self, host='localhost', port=9999):
self.host = host
self.port = port
self.socket = None
self.server_address = (host, port)
self.client_id = f"client_{int(time.time())}"
def start(self):
"""启动客户端"""
try:
# 创建UDP套接字
self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 设置超时时间
self.socket.settimeout(3)
print(f"UDP客户端启动, 服务器: {self.host}:{self.port}")
print(f"客户端ID: {self.client_id}")
return True
except Exception as e:
print(f"客户端启动失败: {e}")
return False
def register(self):
"""向服务器注册"""
register_data = json.dumps({
'command': 'register',
'client_id': self.client_id
})
return self.send_command(register_data)
def send_command(self, command):
"""发送命令到服务器"""
try:
# 发送数据
self.socket.sendto(command.encode('utf-8'), self.server_address)
# 接收响应
data, server_addr = self.socket.recvfrom(1024)
response = data.decode('utf-8')
print(f"服务器响应: {response}")
return response
except socket.timeout:
print("请求超时")
return None
except Exception as e:
print(f"发送命令失败: {e}")
return None
def ping_server(self):
"""发送心跳"""
ping_data = json.dumps({
'command': 'ping',
'client_id': self.client_id
})
return self.send_command(ping_data)
def get_client_list(self):
"""获取客户端列表"""
list_data = json.dumps({
'command': 'list'
})
return self.send_command(list_data)
def send_text_message(self, message):
"""发送文本消息"""
try:
self.socket.sendto(message.encode('utf-8'), self.server_address)
data, server_addr = self.socket.recvfrom(1024)
response = data.decode('utf-8')
print(f"服务器响应: {response}")
return response
except Exception as e:
print(f"发送消息失败: {e}")
return None
def heartbeat_loop(self):
"""心跳循环"""
while True:
try:
print("发送心跳...")
self.ping_server()
time.sleep(10) # 每10秒发送一次心跳
except KeyboardInterrupt:
print("停止心跳")
break
def close(self):
"""关闭客户端"""
if self.socket:
self.socket.close()
print("客户端已关闭")
# 示例使用
if __name__ == "__main__":
client = UDPClient('localhost', 9999)
if client.start():
try:
# 注册客户端
client.register()
# 发送一些测试消息
test_commands = [
json.dumps({'command': 'list'}),
json.dumps({'command': 'ping'}),
'你好，服务器!',
'time',
'clients'
]
for cmd in test_commands:
print(f"\n发送: {cmd}")
client.send_text_message(cmd)
time.sleep(1)
# 启动心跳线程
heartbeat_thread = threading.Thread(target=client.heartbeat_loop)
heartbeat_thread.daemon = True
heartbeat_thread.start()
# 等待用户输入退出
input("按回车键退出...")
except KeyboardInterrupt:
print("\n正在退出...")
finally:
client.close()
---
7.2 ssl TLS/SSL包装
01.SSL上下文
a.创建安全连接
ssl模块提供传输层安全协议支持，用于网络通信的加密和认证。
b.代码示例
---
import ssl
import socket
import pprint
# 创建默认SSL上下文
context = ssl.create_default_context()
print(f"默认协议: {context.protocol}")
print(f"验证模式: {context.verify_mode}")
# 创建客户端SSL上下文
client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
client_context.check_hostname = False
client_context.verify_mode = ssl.CERT_NONE
# 创建服务端SSL上下文
server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
server_context.load_cert_chain(certfile="server.pem", keyfile="server.key")
# 获取支持的协议版本
print("支持的协议版本:")
pprint.pprint([attr for attr in dir(ssl) if 'PROTOCOL_' in attr])
---
02.SSL包装器
a.包装套接字
使用ssl.wrap_socket方法为现有套接字添加SSL/TLS加密层。
b.代码示例
---
import ssl
import socket
import urllib.request
# HTTPS请求示例
def make_https_request(url):
# 创建SSL上下文
context = ssl.create_default_context()
# 创建HTTPS连接
with urllib.request.urlopen(url, context=context) as response:
data = response.read()
print(f"响应大小: {len(data)} 字节")
print(f"状态码: {response.getcode()}")
return data
# 手动SSL套接字包装
def ssl_socket_example():
# 创建TCP套接字
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 包装为SSL套接字
ssl_sock = ssl.wrap_socket(
sock,
keyfile="client.key",
certfile="client.crt",
server_side=False,
cert_reqs=ssl.CERT_REQUIRED,
ca_certs="ca.crt",
ssl_version=ssl.PROTOCOL_TLS
)
try:
# 连接到HTTPS服务器
ssl_sock.connect(("httpbin.org", 443))
print(f"SSL连接已建立，协议: {ssl_sock.version()}")
print(f"加密套件: {ssl_sock.cipher()}")
# 发送HTTP请求
request = b"GET /get HTTP/1.1\r\nHost: httpbin.org\r\n\r\n"
ssl_sock.send(request)
# 接收响应
response = ssl_sock.recv(4096)
print(f"响应: {response[:100]}...")
except ssl.SSLError as e:
print(f"SSL错误: {e}")
finally:
ssl_sock.close()
print("HTTPS请求示例:")
try:
make_https_request("https://httpbin.org/get")
except Exception as e:
print(f"请求失败: {e}")
print("\nSSL套接字包装示例:")
ssl_socket_example()
---
7.3 urllib URL处理
01.URL解析
a.解析URL组件
urllib.parse模块提供URL解析功能，可以将URL分解为各个组件。
b.代码示例
---
# urllib.parse基础示例
from urllib.parse import urlparse, urlunparse, parse_qs, urlencode
# 解析URL
url = "https://www.example.com:8080/path/to/page?name=John&age=25#section1"
parsed = urlparse(url)
print(f"URL解析结果:")
print(f"协议: {parsed.scheme}")
print(f"域名: {parsed.netloc}")
print(f"路径: {parsed.path}")
print(f"参数: {parsed.params}")
print(f"查询: {parsed.query}")
print(f"片段: {parsed.fragment}")
# 查询参数解析
query_dict = parse_qs(parsed.query)
print(f"\n查询参数: {query_dict}")
# 构建URL
new_url = urlunparse((
'https',
'example.com:8080',
'/api/data',
'',
'format=json&limit=10',
'results'
))
print(f"\n构建的URL: {new_url}")
# 编码查询参数
params = {'name': '张三', 'city': '北京', 'page': 1}
encoded = urlencode(params)
print(f"\n编码参数: {encoded}")
---
c.URL编码
处理URL中的特殊字符，确保URL的有效性和安全性。
d.代码示例
---
# URL编码示例
from urllib.parse import quote, unquote, quote_plus, unquote_plus
# 处理包含特殊字符的路径
path = "文档/测试 文件.txt"
encoded_path = quote(path)
print(f"原始路径: {path}")
print(f"编码后: {encoded_path}")
print(f"解码后: {unquote(encoded_path)}")
# 处理表单数据
form_data = "name=John Doe&email=john@example.com"
encoded_form = quote_plus(form_data)
print(f"\n原始表单: {form_data}")
print(f"编码后: {encoded_form}")
print(f"解码后: {unquote_plus(encoded_form)}")
# 编码中文URL
chinese_url = "https://example.com/search?q=Python编程"
safe_chars = ":/=?&"
encoded_chinese = quote(chinese_url, safe=safe_chars)
print(f"\n中文URL编码: {encoded_chinese}")
# 处理文件名
filename = "我的 文档(1).pdf"
encoded_filename = quote(filename)
print(f"\n文件名编码: {encoded_filename}")
print(f"文件名解码: {unquote(encoded_filename)}")
---
02.网络请求
a.GET请求
urllib.request模块用于发送HTTP请求，获取网络资源。
b.代码示例
---
# urllib.request GET请求示例
import urllib.request
import json
from urllib.parse import urlencode
# 简单GET请求
def simple_get():
try:
with urllib.request.urlopen('https://httpbin.org/get') as response:
data = response.read()
text = data.decode('utf-8')
result = json.loads(text)
print(f"状态码: {response.getcode()}")
print(f"响应头: {dict(response.headers)}")
print(f"URL: {result.get('url')}")
except Exception as e:
print(f"请求失败: {e}")
# 带参数的GET请求
def get_with_params():
params = {
'name': 'Alice',
'age': 25,
'format': 'json'
}
query_string = urlencode(params)
url = f'https://httpbin.org/get?{query_string}'
req = urllib.request.Request(url)
req.add_header('User-Agent', 'Python Client')
try:
with urllib.request.urlopen(req) as response:
result = json.loads(response.read().decode('utf-8'))
print(f"查询参数: {result['args']}")
print(f"请求头: {result['headers']}")
except Exception as e:
print(f"请求失败: {e}")
print("简单GET请求:")
simple_get()
print("\n带参数的GET请求:")
get_with_params()
---
c.POST请求
发送POST请求提交数据到服务器，支持表单数据和JSON格式。
d.代码示例
---
# urllib.request POST请求示例
import urllib.request
import json
from urllib.parse import urlencode
# 表单POST请求
def form_post():
data = {
'username': 'testuser',
'password': 'testpass',
'remember': 'on'
}
encoded_data = urlencode(data).encode('utf-8')
req = urllib.request.Request(
'https://httpbin.org/post',
data=encoded_data,
headers={'Content-Type': 'application/x-www-form-urlencoded'}
)
try:
with urllib.request.urlopen(req) as response:
result = json.loads(response.read().decode('utf-8'))
print(f"表单数据: {result['form']}")
print(f"请求大小: {result.get('size', 0)} 字节")
except Exception as e:
print(f"POST失败: {e}")
# JSON POST请求
def json_post():
data = {
'title': 'Python学习',
'content': 'urllib模块使用教程',
'tags': ['python', 'web', 'http']
}
json_data = json.dumps(data).encode('utf-8')
req = urllib.request.Request(
'https://httpbin.org/post',
data=json_data,
headers={
'Content-Type': 'application/json',
'Content-Length': len(json_data)
}
)
try:
with urllib.request.urlopen(req) as response:
result = json.loads(response.read().decode('utf-8'))
print(f"JSON数据: {result['json']}")
print(f"Content-Type: {result['headers']['Content-Type']}")
except Exception as e:
print(f"JSON POST失败: {e}")
print("表单POST请求:")
form_post()
print("\nJSON POST请求:")
json_post()
---
7.4 http HTTP协议
01.HTTP客户端
a.HTTPConnection
http.client模块提供低级HTTP协议支持，适合精细控制HTTP请求。
b.代码示例
---
# http.client基础示例
import http.client
import json
import time
# HTTP连接示例
def basic_http_request():
# 创建HTTP连接
conn = http.client.HTTPConnection("httpbin.org", 80)
try:
# 发送GET请求
conn.request("GET", "/get")
response = conn.getresponse()
print(f"状态码: {response.status}")
print(f"原因: {response.reason}")
print(f"版本: {response.version}")
print(f"响应头: {dict(response.getheaders())}")
# 读取响应内容
data = response.read().decode('utf-8')
result = json.loads(data)
print(f"请求URL: {result.get('url')}")
finally:
conn.close()
# HTTPS连接示例
def https_request():
conn = http.client.HTTPSConnection("httpbin.org", 443)
try:
# 设置请求头
headers = {
'User-Agent': 'Python HTTP Client',
'Accept': 'application/json',
'Custom-Header': 'Test-Value'
}
conn.request("GET", "/headers", headers=headers)
response = conn.getresponse()
result = json.loads(response.read().decode('utf-8'))
print(f"请求头: {result['headers']}")
finally:
conn.close()
print("基本HTTP请求:")
basic_http_request()
print("\nHTTPS请求:")
https_request()
---
c.POST请求
发送HTTP POST请求，支持表单数据和JSON格式。
d.代码示例
---
# HTTP POST请求示例
import http.client
import json
import urllib.parse
# 表单POST请求
def form_post_request():
conn = http.client.HTTPSConnection("httpbin.org")
# 准备表单数据
params = urllib.parse.urlencode({
'username': 'testuser',
'password': 'secret123',
'remember': 'true'
})
# 设置请求头
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'Content-Length': len(params)
}
try:
conn.request("POST", "/post", params, headers)
response = conn.getresponse()
result = json.loads(response.read().decode('utf-8'))
print(f"表单数据: {result['form']}")
print(f"请求大小: {result.get('size', 0)} 字节")
finally:
conn.close()
# JSON POST请求
def json_post_request():
conn = http.client.HTTPSConnection("httpbin.org")
# 准备JSON数据
data = {
'name': 'Python客户端',
'version': '3.8',
'features': ['HTTP', 'HTTPS', 'JSON']
}
json_data = json.dumps(data)
headers = {
'Content-Type': 'application/json',
'Content-Length': len(json_data)
}
try:
conn.request("POST", "/post", json_data, headers)
response = conn.getresponse()
result = json.loads(response.read().decode('utf-8'))
print(f"JSON数据: {result['json']}")
print(f"Content-Type: {result['headers']['Content-Type']}")
finally:
conn.close()
print("表单POST请求:")
form_post_request()
print("\nJSON POST请求:")
json_post_request()
---
02.HTTP服务器
a.BaseHTTPRequestHandler
http.server模块提供HTTP服务器功能，可以构建Web服务器。
b.代码示例
---
# HTTP服务器示例
import http.server
import socketserver
import json
import threading
import time
# 自定义请求处理器
class MyHandler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
"""处理GET请求"""
if self.path == '/':
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
response = '<h1>Python HTTP服务器</h1><p>欢迎使用!</p>'
self.wfile.write(response.encode('utf-8'))
elif self.path == '/api/data':
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
data = {
'message': 'API数据',
'time': time.time(),
'status': 'success'
}
json_data = json.dumps(data, ensure_ascii=False)
self.wfile.write(json_data.encode('utf-8'))
else:
self.send_error(404, "页面未找到")
def do_POST(self):
"""处理POST请求"""
if self.path == '/api/upload':
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length)
try:
data = json.loads(post_data.decode('utf-8'))
response_data = {
'received': data,
'status': 'success',
'message': '数据接收成功'
}
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
response = json.dumps(response_data, ensure_ascii=False)
self.wfile.write(response.encode('utf-8'))
except json.JSONDecodeError:
self.send_error(400, "无效的JSON数据")
else:
self.send_error(404, "接口不存在")
def log_message(self, format, *args):
"""自定义日志格式"""
print(f"[{self.log_date_time_string()}] {format % args}")
# 启动服务器
def run_server():
PORT = 8000
with socketserver.TCPServer(("", PORT), MyHandler) as httpd:
print(f"服务器启动在端口 {PORT}")
print("访问 http://localhost:8000")
print("API接口: http://localhost:8000/api/data")
try:
httpd.serve_forever()
except KeyboardInterrupt:
print("\n服务器停止")
httpd.shutdown()
# 测试服务器函数
def test_server():
import urllib.request
import json
# 等待服务器启动
time.sleep(2)
try:
# 测试GET请求
with urllib.request.urlopen('http://localhost:8000/api/data') as response:
data = json.loads(response.read().decode('utf-8'))
print(f"GET响应: {data}")
# 测试POST请求
post_data = json.dumps({'test': 'data'}).encode('utf-8')
req = urllib.request.Request(
'http://localhost:8000/api/upload',
data=post_data,
headers={'Content-Type': 'application/json'}
)
with urllib.request.urlopen(req) as response:
result = json.loads(response.read().decode('utf-8'))
print(f"POST响应: {result}")
except Exception as e:
print(f"测试失败: {e}")
# 在后台运行服务器
server_thread = threading.Thread(target=run_server, daemon=True)
server_thread.start()
# 测试服务器
test_server()
# 保持服务器运行一段时间
time.sleep(1)
print("\n服务器运行中... (按Ctrl+C停止)")
time.sleep(2)
---
7.5 email邮件处理
01.邮件创建
a.MIME消息
email.mime模块提供创建各种类型邮件消息的功能。
b.代码示例
---
# email.mime基础示例
import email.mime.text
import email.mime.multipart
import email.mime.base
import email.mime.application
import mimetypes
import io
# 创建纯文本邮件
def create_text_email():
text_msg = email.mime.text.MIMEText(
'这是一封纯文本邮件。\n\n此致\n敬礼',
'plain',
'utf-8'
)
text_msg['From'] = 'sender@example.com'
text_msg['To'] = 'recipient@example.com'
text_msg['Subject'] = '纯文本邮件测试'
print("纯文本邮件:")
print(text_msg.as_string())
return text_msg
# 创建HTML邮件
def create_html_email():
html_content = '''
<html>
<body>
<h1>HTML邮件测试</h1>
<p>这是一封<b>HTML格式</b>的邮件。</p>
<ul>
<li>支持格式化文本</li>
<li>支持图片和链接</li>
<li>支持表格</li>
</ul>
<p style="color: blue;">感谢使用Python email模块!</p>
</body>
</html>
'''
html_msg = email.mime.text.MIMEText(html_content, 'html', 'utf-8')
html_msg['From'] = 'sender@example.com'
html_msg['To'] = 'recipient@example.com'
html_msg['Subject'] = 'HTML邮件测试'
print("\nHTML邮件:")
print(html_msg.as_string())
return html_msg
# 创建多媒体邮件
def create_multipart_email():
# 创建multipart容器
msg = email.mime.multipart.MIMEMultipart()
msg['From'] = 'sender@example.com'
msg['To'] = 'recipient@example.com'
msg['Subject'] = '多媒体邮件测试'
# 添加文本部分
text_part = email.mime.text.MIMEText('这是纯文本部分', 'plain', 'utf-8')
msg.attach(text_part)
# 添加HTML部分
html_part = email.mime.text.MIMEText(
'<html><body><h1>HTML部分</h1><p>这是HTML格式的邮件内容。</p></body></html>',
'html',
'utf-8'
)
msg.attach(html_part)
print("\n多媒体邮件:")
print(msg.as_string())
return msg
# 创建带附件的邮件
def create_email_with_attachment():
msg = email.mime.multipart.MIMEMultipart()
msg['From'] = 'sender@example.com'
msg['To'] = 'recipient@example.com'
msg['Subject'] = '带附件的邮件'
# 邮件正文
body = "这是邮件正文，包含附件。"
msg.attach(email.mime.text.MIMEText(body, 'plain', 'utf-8'))
# 创建文本附件
attachment_content = "这是一个文本附件的内容。\n包含多行文本。"
attachment = email.mime.text.MIMEText(attachment_content, 'plain', 'utf-8')
attachment.add_header(
'Content-Disposition',
'attachment',
filename='readme.txt'
)
msg.attach(attachment)
# 创建二进制附件示例
binary_data = b'\x00\x01\x02\x03\x04\x05' # 示例二进制数据
binary_attachment = email.mime.base.MIMEBase('application', 'octet-stream')
binary_attachment.set_payload(binary_data)
import email.encoders
email.encoders.encode_base64(binary_attachment)
binary_attachment.add_header(
'Content-Disposition',
'attachment',
filename='data.bin'
)
msg.attach(binary_attachment)
print("\n带附件的邮件:")
print(msg.as_string())
return msg
# 执行所有示例
create_text_email()
create_html_email()
create_multipart_email()
create_email_with_attachment()
---
c.邮件地址处理
处理邮件地址的编码、解码和格式化。
d.代码示例
---
# 邮件地址处理示例
import email.utils
email.header
# 处理邮件地址
def process_email_addresses():
# 基本邮件地址
addresses = [
'user@example.com',
'User Name <user@example.com>',
'用户 <user@example.com>',
'"User, Name" <user@example.com>',
'user1@example.com, user2@example.com'
]
print("邮件地址解析:")
for addr in addresses:
parsed = email.utils.parseaddr(addr)
print(f"原始: {addr}")
print(f"解析: 姓名='{parsed[0]}', 地址='{parsed[1]}'")
print()
# 格式化邮件地址
formatted = email.utils.formataddr(('张三', 'zhangsan@example.com'))
print(f"格式化地址: {formatted}")
# 解析多个地址
multiple = 'Alice <alice@example.com>, Bob <bob@example.com>, charlie@example.com'
parsed_list = email.utils.getaddresses([multiple])
print(f"\n多个地址解析: {parsed_list}")
# 邮件头部编码
def encode_headers():
# 需要编码的头部信息
headers = {
'From': '发送者 <sender@example.com>',
'To': '张三 <zhangsan@example.com>',
'Subject': '测试邮件：包含中文的标题',
'CC': '李四 <lisi@example.com>'
}
print("\n邮件头部编码:")
for name, value in headers.items():
# 检查是否需要编码
try:
value.encode('ascii')
encoded = value
except UnicodeEncodeError:
encoded = email.header.Header(value, 'utf-8').encode()
print(f"{name}: {encoded}")
# 生成邮件ID
def generate_message_id():
domain = 'example.com'
msg_id = email.utils.make_msgid(domain=domain)
print(f"\n生成的邮件ID: {msg_id}")
# 自定义邮件ID
custom_id = f"<custom-{hash('test')}@{domain}>"
print(f"自定义邮件ID: {custom_id}")
# 执行示例
process_email_addresses()
encode_headers()
generate_message_id()
---
02.邮件解析
a.Parser解析器
email.parser模块提供解析邮件消息的功能。
b.代码示例
---
# 邮件解析示例
import email.parser
import email.policy
import io
# 解析简单邮件
def parse_simple_email():
# 邮件原始内容
raw_email = """From: sender@example.com
To: recipient@example.com
Subject: 简单邮件测试
Date: Thu, 24 Nov 2025 10:00:00 +0800
这是一封简单的邮件内容。
包含多行文本。
此致
敬礼
"""
# 创建解析器
parser = email.parser.Parser(policy=email.policy.default)
msg = parser.parsestr(raw_email)
print("解析简单邮件:")
print(f"发件人: {msg['From']}")
print(f"收件人: {msg['To']}")
print(f"主题: {msg['Subject']}")
print(f"日期: {msg['Date']}")
print(f"邮件正文:\n{msg.get_body(preferencelist=('plain')).get_content()}")
print()
return msg
# 解析多媒体邮件
def parse_multipart_email():
raw_multipart = """From: sender@example.com
To: recipient@example.com
Subject: 多媒体邮件测试
MIME-Version: 1.0
Content-Type: multipart/mixed; boundary="boundary123"
--boundary123
Content-Type: text/plain; charset=utf-8
这是邮件正文部分。
--boundary123
Content-Type: text/html; charset=utf-8
<html><body><h1>HTML部分</h1></body></html>
--boundary123
Content-Type: application/octet-stream
Content-Disposition: attachment; filename="test.txt"
文件附件内容
--boundary123--
"""
parser = email.parser.Parser(policy=email.policy.default)
msg = parser.parsestr(raw_multipart)
print("解析多媒体邮件:")
print(f"邮件类型: {msg.get_content_type()}")
print(f"是否为multipart: {msg.is_multipart()}")
# 遍历邮件部分
for i, part in enumerate(msg.iter_parts()):
print(f"\n部分 {i+1}:")
print(f" 类型: {part.get_content_type()}")
print(f" 字符集: {part.get_content_charset()}")
if part.get_content_disposition() == 'attachment':
print(f" 附件: {part.get_filename()}")
else:
content = part.get_content()
if isinstance(content, str) and len(content) < 100:
print(f" 内容: {content.strip()}")
else:
print(f" 内容长度: {len(str(content))}")
return msg
# 解析带附件的邮件
def parse_email_with_attachments():
raw_email_with_attachments = """From: sender@example.com
To: recipient@example.com
Subject: 带附件的邮件
MIME-Version: 1.0
Content-Type: multipart/mixed; boundary="boundary456"
--boundary456
Content-Type: text/plain; charset=utf-8
邮件正文内容。
--boundary456
Content-Type: application/pdf
Content-Transfer-Encoding: base64
Content-Disposition: attachment; filename="document.pdf"
JVBERi0xLjUKJeLjz9MKNCAwIG9iago8PC9MZW5ndGggNSAwIFIvRmlsdGVyIC9GbGF0ZURl
Y29kZT4+CnN0cmVhbQp4nGWQMQ7CMAxF0+6BGkghDDkQVj5gJ3cWCNIFGkiEASbUWSvJ585M
--boundary456--
"""
parser = email.parser.Parser(policy=email.policy.default)
msg = parser.parsestr(raw_email_with_attachments)
print("\n解析带附件的邮件:")
# 提取邮件正文
body = msg.get_body(preferencelist=('plain'))
if body:
print(f"邮件正文: {body.get_content().strip()}")
# 提取附件
attachments = msg.iter_attachments()
print("\n附件列表:")
for i, attachment in enumerate(attachments, 1):
filename = attachment.get_filename()
content_type = attachment.get_content_type()
size = len(attachment.get_content())
print(f" 附件 {i}: {filename}")
print(f" 类型: {content_type}")
print(f" 大小: {size} 字节")
return msg
# 执行所有解析示例
parse_simple_email()
parse_multipart_email()
parse_email_with_attachments()
---
8. 系统与进程
8.1 sys系统参数
01.系统信息
a.平台信息
sys模块提供访问系统相关参数和函数的功能。
b.代码示例
---
# sys系统信息示例
import sys
import os
import platform
# Python版本信息
print("Python版本信息:")
print(f"版本号: {sys.version}")
print(f"版本信息: {sys.version_info}")
print(f"主版本: {sys.version_info.major}")
print(f"次版本: {sys.version_info.minor}")
print(f"微版本: {sys.version_info.micro}")
# 系统平台信息
print(f"\n系统平台: {sys.platform}")
print(f"操作系统: {os.name}")
print(f"系统信息: {platform.system()} {platform.release()}")
# 实现信息
print(f"\nPython实现: {sys.implementation}")
print(f"编译器: {platform.python_compiler()}")
# 字节序信息
print(f"\n字节序: {sys.byteorder}")
print(f"Unicode大小: {sys.maxunicode}")
# 模块搜索路径
print(f"\n模块搜索路径数量: {len(sys.path)}")
print(f"前3个路径: {sys.path[:3]}")
# 系统限制
print(f"\n最大递归深度: {sys.getrecursionlimit()}")
print(f"最大整数大小: {sys.maxsize}")
print(f"浮点精度: {sys.float_info}")
# 标准流信息
print(f"\n标准输入: {sys.stdin}")
print(f"标准输出: {sys.stdout}")
print(f"标准错误: {sys.stderr}")
---
c.命令行参数
处理命令行传入的参数和选项。
d.代码示例
---
# 命令行参数处理示例
import sys
def process_command_line():
print("命令行参数处理:")
print(f"脚本名称: {sys.argv[0]}")
print(f"参数总数: {len(sys.argv) - 1}")
if len(sys.argv) > 1:
print("参数列表:")
for i, arg in enumerate(sys.argv[1:], 1):
print(f" 参数 {i}: {arg}")
else:
print("没有提供命令行参数")
# 解析常见参数模式
args = sys.argv[1:]
options = {}
positional = []
for arg in args:
if arg.startswith('--'):
if '=' in arg:
key, value = arg[2:].split('=', 1)
options[key] = value
else:
options[arg[2:]] = True
elif arg.startswith('-'):
options[arg[1:]] = True
else:
positional.append(arg)
print(f"\n解析结果:")
print(f"选项: {options}")
print(f"位置参数: {positional}")
# 程序退出处理
def exit_handling():
print(f"\n程序退出处理:")
# 正常退出
def normal_exit():
print("程序正常退出")
# sys.exit(0) # 实际使用时取消注释
# 错误退出
def error_exit():
print("程序因错误退出")
# sys.exit(1) # 实际使用时取消注释
# 异常处理退出
def exception_exit():
try:
raise ValueError("测试异常")
except Exception as e:
print(f"捕获异常: {e}")
# sys.exit(2) # 实际使用时取消注释
# 演示退出处理（不实际退出）
normal_exit()
error_exit()
exception_exit()
# 执行示例
process_command_line()
exit_handling()
---
02.模块管理
a.模块导入
sys模块提供模块导入和路径管理的功能。
b.代码示例
---
# 模块管理示例
import sys
import importlib
# 当前已导入模块
def list_imported_modules():
print("已导入的模块:")
modules = list(sys.modules.keys())
print(f"总数量: {len(modules)}")
# 显示前10个核心模块
core_modules = [m for m in modules if not '.' in m][:10]
print("核心模块示例:")
for module in core_modules:
module_obj = sys.modules.get(module)
if module_obj:
file_path = getattr(module_obj, '__file__', '内置模块')
print(f" {module}: {file_path}")
# 模块路径管理
def manage_module_paths():
print(f"\n模块路径管理:")
print(f"当前路径数量: {len(sys.path)}")
# 显示主要路径
print("主要模块路径:")
for i, path in enumerate(sys.path[:5]):
path_type = "内置路径" if i == 0 else "脚本目录" if i == 1 else "系统路径"
print(f" {path_type}: {path}")
# 添加临时路径
temp_path = "/tmp/python_modules"
if temp_path not in sys.path:
sys.path.insert(0, temp_path)
print(f"\n已添加临时路径: {temp_path}")
print(f"新路径数量: {len(sys.path)}")
# 移除临时路径
if temp_path in sys.path:
sys.path.remove(temp_path)
print(f"已移除临时路径: {temp_path}")
# 动态导入模块
def dynamic_import():
print(f"\n动态模块导入:")
# 尝试导入标准库模块
modules_to_try = ['math', 'random', 'datetime']
for module_name in modules_to_try:
try:
if module_name in sys.modules:
module_obj = sys.modules[module_name]
print(f"{module_name}: 已导入")
else:
module_obj = importlib.import_module(module_name)
print(f"{module_name}: 导入成功")
# 获取模块属性
attrs = [attr for attr in dir(module_obj) if not attr.startswith('_')]
print(f" 属性示例: {attrs[:5]}")
except ImportError as e:
print(f"{module_name}: 导入失败 - {e}")
# 模块查找
def find_module():
print(f"\n模块查找:")
# 查找模块位置
module_name = 'os'
if module_name in sys.modules:
module = sys.modules[module_name]
file_path = getattr(module, '__file__', '内置模块')
print(f"{module_name} 位置: {file_path}")
# 执行所有示例
list_imported_modules()
manage_module_paths()
dynamic_import()
find_module()
---
c.性能监控
监控程序执行的性能和资源使用情况。
d.代码示例
---
# 性能监控示例
import sys
import time
import tracemalloc
# 内存使用监控
def memory_monitoring():
print("内存使用监控:")
# 开始内存跟踪
tracemalloc.start()
# 创建一些对象
data = []
for i in range(1000):
data.append([j for j in range(100)])
# 获取内存统计
current, peak = tracemalloc.get_traced_memory()
print(f"当前内存使用: {current / 1024 / 1024:.2f} MB")
print(f"峰值内存使用: {peak / 1024 / 1024:.2f} MB")
tracemalloc.stop()
# 执行时间监控
def execution_monitoring():
print(f"\n执行时间监控:")
start_time = time.time()
# 模拟一些计算
result = sum(i * i for i in range(100000))
end_time = time.time()
execution_time = end_time - start_time
print(f"计算结果: {result}")
print(f"执行时间: {execution_time:.6f} 秒")
print(f"每秒操作数: {100000 / execution_time:.0f}")
# 系统调用跟踪
def system_calls():
print(f"\n系统信息:")
# 调用统计
print(f"函数调用次数: {sys.gettotalrefcount()}")
# 对象引用计数
test_str = "Hello World"
ref_count = sys.getrefcount(test_str)
print(f"字符串引用计数: {ref_count}")
# 线程信息
try:
thread_count = len(sys._current_frames())
print(f"活动线程数: {thread_count}")
except AttributeError:
print("无法获取线程信息")
# 性能优化建议
def performance_tips():
print(f"\n性能优化建议:")
# 检查递归深度
recursion_limit = sys.getrecursionlimit()
print(f"当前递归限制: {recursion_limit}")
if recursion_limit < 1000:
print("建议增加递归限制以支持深度递归")
sys.setrecursionlimit(2000)
print(f"新的递归限制: {sys.getrecursionlimit()}")
# 检查模块加载性能
start_time = time.time()
import_module_time = time.time() - start_time
print(f"模块导入时间: {import_module_time:.6f} 秒")
# 执行性能监控
memory_monitoring()
execution_monitoring()
system_calls()
performance_tips()
---
8.2 subprocess子进程管理
01.进程创建
a.run函数
subprocess.run()提供了运行子进程的简化接口。
b.代码示例
---
# subprocess基础示例
import subprocess
import sys
import os
# 使用run函数执行命令
def basic_run_examples():
print("subprocess.run基础示例:")
# 简单命令执行
try:
result = subprocess.run(['python', '--version'],
capture_output=True, text=True)
print(f"Python版本: {result.stdout.strip()}")
print(f"返回码: {result.returncode}")
except FileNotFoundError:
print("Python命令未找到")
# 带参数的命令
try:
result = subprocess.run(['echo', 'Hello subprocess'],
capture_output=True, text=True)
print(f"输出: {result.stdout.strip()}")
except Exception as e:
print(f"执行失败: {e}")
# 使用shell=True执行命令
try:
result = subprocess.run('echo "Shell command"',
shell=True, capture_output=True, text=True)
print(f"Shell输出: {result.stdout.strip()}")
except Exception as e:
print(f"Shell执行失败: {e}")
# 执行命令并处理错误
def command_with_error_handling():
print(f"\n错误处理示例:")
# 捕获异常
try:
result = subprocess.run(['nonexistent_command'],
capture_output=True, text=True, timeout=5)
except FileNotFoundError as e:
print(f"命令不存在: {e}")
except subprocess.TimeoutExpired as e:
print(f"命令超时: {e}")
except Exception as e:
print(f"其他错误: {e}")
# 检查返回码
try:
result = subprocess.run(['python', '-c', 'import sys; sys.exit(1)'],
capture_output=True, text=True)
if result.returncode != 0:
print(f"命令执行失败，返回码: {result.returncode}")
print(f"错误输出: {result.stderr}")
except Exception as e:
print(f"执行错误: {e}")
# 输入输出重定向
def io_redirection():
print(f"\n输入输出重定向:")
# 写入文件
try:
with open('test_output.txt', 'w') as f:
result = subprocess.run(['echo', 'Test content'],
stdout=f, text=True)
print("已写入文件")
# 读取文件内容
with open('test_output.txt', 'r') as f:
content = f.read().strip()
print(f"文件内容: {content}")
except Exception as e:
print(f"文件操作失败: {e}")
# 管道输入
try:
result = subprocess.run(['python', '-c',
'import sys; print(sys.stdin.read().upper())'],
input='hello world', text=True,
capture_output=True)
print(f"管道输出: {result.stdout.strip()}")
except Exception as e:
print(f"管道操作失败: {e}")
# 清理文件
try:
os.remove('test_output.txt')
except:
pass
# 执行所有示例
basic_run_examples()
command_with_error_handling()
io_redirection()
---
c.Popen构造器
Popen提供更灵活的进程控制能力。
d.代码示例
---
# Popen高级示例
import subprocess
import time
import threading
# 异步进程执行
def async_process_example():
print("异步进程执行:")
# 启动长时间运行的进程
try:
process = subprocess.Popen(['python', '-c',
'import time; print("开始"); time.sleep(2); print("结束")'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
print(f"进程已启动，PID: {process.pid}")
# 检查进程状态
while process.poll() is None:
print("进程运行中...")
time.sleep(0.5)
# 获取输出
stdout, stderr = process.communicate()
print(f"进程输出: {stdout.strip()}")
print(f"返回码: {process.returncode}")
except Exception as e:
print(f"异步进程失败: {e}")
# 管道连接
def pipe_example():
print(f"\n管道连接示例:")
try:
# 创建两个进程并连接管道
p1 = subprocess.Popen(['echo', 'Hello World'],
stdout=subprocess.PIPE, text=True)
p2 = subprocess.Popen(['python', '-c',
'import sys; data = sys.stdin.read(); print(f"接收: {data.strip()}")'],
stdin=p1.stdout, stdout=subprocess.PIPE, text=True)
p1.stdout.close() # 允许p1接收到SIGPIPE
output = p2.communicate()[0]
print(f"管道输出: {output.strip()}")
except Exception as e:
print(f"管道操作失败: {e}")
# 进程环境变量
def environment_example():
print(f"\n环境变量示例:")
# 设置自定义环境变量
custom_env = os.environ.copy()
custom_env['MY_VAR'] = 'Custom Value'
try:
result = subprocess.run(['python', '-c',
'import os; print(f"MY_VAR: {os.getenv(\"MY_VAR\", \"未设置\")}")'],
env=custom_env, capture_output=True, text=True)
print(f"环境变量输出: {result.stdout.strip()}")
except Exception as e:
print(f"环境变量设置失败: {e}")
# 进程组管理
def process_group_example():
print(f"\n进程组管理:")
try:
# 创建进程组
process = subprocess.Popen(['python', '-c',
'import time; print("子进程运行"); time.sleep(1); print("子进程结束")'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
print(f"子进程PID: {process.pid}")
# 等待进程结束
process.wait(timeout=5)
print("子进程已结束")
except subprocess.TimeoutExpired:
print("进程超时，终止进程")
process.terminate()
process.wait()
except Exception as e:
print(f"进程组管理失败: {e}")
# 执行高级示例
async_process_example()
pipe_example()
environment_example()
process_group_example()
---
02.进程通信
a.管道通信
使用管道实现进程间数据交换。
b.代码示例
---
# 进程间通信示例
import subprocess
import json
threading
import queue
# 标准输入输出通信
def stdio_communication():
print("标准输入输出通信:")
try:
# 发送数据到子进程
process = subprocess.Popen(
['python', '-c', '''
import json
import sys
while True:
try:
line = sys.stdin.readline()
if not line:
break
data = json.loads(line.strip())
result = {"received": data, "processed": True}
print(json.dumps(result))
except Exception as e:
print(json.dumps({"error": str(e)}))
'''],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# 发送多个消息
messages = [
{"message": "Hello", "id": 1},
{"message": "World", "id": 2},
{"message": "Python", "id": 3}
]
for msg in messages:
# 发送消息
process.stdin.write(json.dumps(msg) + '\n')
process.stdin.flush()
# 读取响应
response = process.stdout.readline()
if response:
result = json.loads(response.strip())
print(f"发送: {msg}, 接收: {result}")
# 关闭输入流
process.stdin.close()
process.wait()
except Exception as e:
print(f"标准通信失败: {e}")
# 队列通信模拟
def queue_communication():
print(f"\n队列通信模拟:")
# 创建通信队列
message_queue = queue.Queue()
response_queue = queue.Queue()
def worker_process():
"""模拟工作进程"""
while True:
try:
message = message_queue.get(timeout=2)
if message == "STOP":
break
# 处理消息
response = f"处理: {message.upper()}"
response_queue.put(response)
message_queue.task_done()
except queue.Empty:
break
# 启动工作线程（模拟子进程）
worker_thread = threading.Thread(target=worker_process, daemon=True)
worker_thread.start()
# 发送消息
messages = ["hello", "world", "python", "subprocess"]
for msg in messages:
message_queue.put(msg)
response = response_queue.get()
print(f"消息: {msg}, 响应: {response}")
# 停止工作线程
message_queue.put("STOP")
worker_thread.join(timeout=3)
# 文件通信
def file_communication():
print(f"\n文件通信:")
import tempfile
import os
try:
# 创建临时文件
with tempfile.NamedTemporaryFile(mode='w+', delete=False) as f:
temp_file = f.name
# 写入数据到临时文件
data = {"command": "process", "data": [1, 2, 3, 4, 5]}
with open(temp_file, 'w') as f:
json.dump(data, f)
# 子进程读取文件并处理
process = subprocess.Popen([
'python', '-c', f'''
import json
import sys
with open("{temp_file}", 'r') as f:
data = json.load(f)
result = sum(data["data"])
with open("{temp_file}", 'w') as f:
json.dump({{"result": result, "success": True}}, f)
'''
], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
process.wait()
# 读取处理结果
with open(temp_file, 'r') as f:
result = json.load(f)
print(f"处理结果: {result}")
# 清理临时文件
os.unlink(temp_file)
except Exception as e:
print(f"文件通信失败: {e}")
# 执行通信示例
stdio_communication()
queue_communication()
file_communication()
---
8.3 signal信号处理
01.信号基础
a.信号定义
signal模块提供处理异步信号事件的功能。
b.代码示例
---
# signal基础示例
import signal
import time
import threading
import sys
# 信号处理函数
def signal_handler(signum, frame):
print(f"\n接收到信号 {signum}")
print("信号帧信息:", frame)
if signum == signal.SIGINT:
print("收到中断信号(Ctrl+C)，准备退出...")
sys.exit(0)
# 注册信号处理器
def setup_signal_handlers():
print("设置信号处理器:")
# 注册SIGINT信号处理器(Ctrl+C)
signal.signal(signal.SIGINT, signal_handler)
print("已注册SIGINT信号处理器")
# 注册SIGTERM信号处理器
signal.signal(signal.SIGTERM, signal_handler)
print("已注册SIGTERM信号处理器")
# 在Windows上可用的信号
if hasattr(signal, 'SIGBREAK'):
signal.signal(signal.SIGBREAK, signal_handler)
print("已注册SIGBREAK信号处理器")
# 演示信号处理
def signal_demo():
print("\n信号处理演示:")
print("程序正在运行，按Ctrl+C发送SIGINT信号")
print("等待10秒...")
for i in range(10):
print(f"运行中... {i+1}/10")
time.sleep(1)
if i == 5:
print("提示：现在可以按Ctrl+C测试信号处理")
print("程序正常结束")
# 信号处理实践
def practice_signal_handling():
global running
running = True
def stop_handler(signum, frame):
global running
print(f"\n收到停止信号 {signum}")
running = False
# 设置停止信号处理器
signal.signal(signal.SIGINT, stop_handler)
signal.signal(signal.SIGTERM, stop_handler)
print("\n长期运行程序示例:")
print("按Ctrl+C可随时停止程序")
count = 0
while running and count < 20:
print(f"工作循环 {count + 1}")
time.sleep(0.5)
count += 1
if not running:
print("程序被信号中断")
else:
print("程序自然结束")
# 执行所有示例
try:
setup_signal_handlers()
signal_demo()
practice_signal_handling()
except KeyboardInterrupt:
print("\n程序被用户中断")
except Exception as e:
print(f"发生错误: {e}")
---
c.信号类型
不同操作系统的信号类型和用法。
d.代码示例
---
# 信号类型示例
import signal
import platform
import os
# 显示可用信号
def show_available_signals():
print("可用信号列表:")
# 获取所有信号常量
signals = []
for name in dir(signal):
if name.startswith('SIG') and not name.startswith('SIG_'):
sig_value = getattr(signal, name)
if isinstance(sig_value, int):
signals.append((name, sig_value))
# 按信号值排序
signals.sort(key=lambda x: x[1])
print(f"平台: {platform.system()}")
print(f"总信号数: {len(signals)}")
print("\n常用信号:")
# 显示常用信号
common_signals = ['SIGINT', 'SIGTERM', 'SIGUSR1', 'SIGUSR2']
for sig_name in common_signals:
if hasattr(signal, sig_name):
sig_value = getattr(signal, sig_name)
print(f" {sig_name}: {sig_value}")
print("\n所有可用信号:")
for sig_name, sig_value in signals[:10]: # 只显示前10个
print(f" {sig_name}: {sig_value}")
# 信号发送示例
def signal_sending_demo():
print(f"\n信号发送演示:")
# 获取当前进程ID
pid = os.getpid()
print(f"当前进程PID: {pid}")
# 在Unix-like系统上可以发送更多信号
if platform.system() != 'Windows':
print("\nUnix-like系统信号功能:")
# 定义信号处理函数
def custom_handler(signum, frame):
print(f"收到自定义信号: {signum}")
# 注册自定义信号处理器
signal.signal(signal.SIGUSR1, custom_handler)
signal.signal(signal.SIGUSR2, custom_handler)
print("已注册USR1和USR2信号处理器")
print("可以使用以下命令发送信号:")
print(f"kill -USR1 {pid}")
print(f"kill -USR2 {pid}")
# 模拟等待信号
print("等待信号5秒...")
time.sleep(5)
else:
print("Windows系统下信号功能受限")
# 信号掩码示例
def signal_mask_demo():
print(f"\n信号掩码演示:")
# Unix-like系统支持信号掩码
if platform.system() != 'Windows':
try:
import signal
# 阻塞某些信号
blocked_signals = [signal.SIGINT, signal.SIGTERM]
old_mask = signal.pthread_sigmask(signal.SIG_BLOCK, blocked_signals)
print("已阻塞SIGINT和SIGTERM信号")
# 在这里执行关键代码
print("执行关键操作...")
# 恢复信号处理
signal.pthread_sigmask(signal.SIG_SETMASK, old_mask)
print("已恢复信号处理")
except AttributeError:
print("当前系统不支持信号掩码")
except Exception as e:
print(f"信号掩码操作失败: {e}")
else:
print("Windows系统不支持信号掩码")
# 执行信号类型演示
show_available_signals()
signal_sending_demo()
signal_mask_demo()
---
02.高级信号处理
a.定时信号
使用alarm和setitimer设置定时信号。
b.代码示例
---
# 定时信号示例
import signal
import time
threading
# 简单定时器
def simple_alarm():
print("简单定时器示例:")
def alarm_handler(signum, frame):
print("定时器触发!")
# 设置5秒后触发信号
signal.signal(signal.SIGALRM, alarm_handler)
signal.alarm(5)
print("定时器已设置，5秒后触发")
print("等待中...")
for i in range(6):
print(f"{i}秒")
time.sleep(1)
# 取消定时器
signal.alarm(0)
print("定时器已取消")
# 高级定时器
def advanced_timer():
print(f"\n高级定时器示例:")
count = 0
def timer_handler(signum, frame):
nonlocal count
count += 1
print(f"定时器触发 #{count}")
if count >= 3:
print("达到触发次数，取消定时器")
signal.setitimer(signal.ITIMER_REAL, 0)
# 设置定时器，每2秒触发一次
signal.signal(signal.SIGALRM, timer_handler)
signal.setitimer(signal.ITIMER_REAL, 2) # 2秒后开始
print("定时器已设置，每2秒触发一次，最多3次")
# 等待定时器触发
time.sleep(8)
print(f"总共触发了 {count} 次")
# 线程安全的信号处理
def thread_safe_signal():
print(f"\n线程安全信号处理:")
import threading
# 使用事件在主线程和信号处理器间通信
stop_event = threading.Event()
def signal_handler(signum, frame):
print(f"\n信号 {signum} 被触发")
stop_event.set()
# 注册信号处理器
signal.signal(signal.SIGINT, signal_handler)
def worker_thread():
print("工作线程启动")
while not stop_event.is_set():
print("工作中...")
time.sleep(1)
print("工作线程停止")
# 启动工作线程
worker = threading.Thread(target=worker_thread, daemon=True)
worker.start()
print("主线程运行中，按Ctrl+C停止")
try:
# 等待信号或超时
worker.join(timeout=10)
except KeyboardInterrupt:
print("主线程收到中断")
finally:
stop_event.set()
worker.join(timeout=2)
# 信号与异常处理
def signal_exception_handling():
print(f"\n信号与异常处理:")
def exception_handler(signum, frame):
print(f"异常处理器: 信号 {signum}")
try:
# 模拟可能抛出异常的操作
raise ValueError("信号处理器中的异常")
except Exception as e:
print(f"捕获异常: {e}")
# 设置异常处理器
signal.signal(signal.SIGINT, exception_handler)
print("测试异常处理，按Ctrl+C触发")
# 使用超时来避免无限等待
for i in range(3):
print(f"等待中... {i+1}")
try:
time.sleep(1)
except KeyboardInterrupt:
print("主循环被中断")
break
print("异常处理测试完成")
# 执行定时信号示例
try:
simple_alarm()
advanced_timer()
thread_safe_signal()
signal_exception_handling()
except Exception as e:
print(f"定时信号示例出错: {e}")
---
8.4 logging日志记录
01.基础日志配置
a.日志级别和格式
logging模块提供灵活的日志记录功能，支持多种日志级别和自定义格式。
b.代码示例
---
import logging
import sys
# 基本配置
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
handlers=[
logging.FileHandler('app.log', encoding='utf-8'),
logging.StreamHandler(sys.stdout)
]
)
# 创建日志记录器
logger = logging.getLogger(__name__)
# 不同级别的日志消息
logger.debug("这是调试信息")
logger.info("这是普通信息")
logger.warning("这是警告信息")
logger.error("这是错误信息")
logger.critical("这是严重错误信息")
# 测试日志级别
print("\n当前日志级别:")
print(f"DEBUG: {logger.isEnabledFor(logging.DEBUG)}")
print(f"INFO: {logger.isEnabledFor(logging.INFO)}")
print(f"WARNING: {logger.isEnabledFor(logging.WARNING)}")
print(f"ERROR: {logger.isEnabledFor(logging.ERROR)}")
print(f"CRITICAL: {logger.isEnabledFor(logging.CRITICAL)}")
---
02.高级日志配置
a.自定义处理器和格式化器
创建自定义处理器和格式化器来满足特定的日志需求。
b.代码示例
---
import logging
import logging.handlers
import os
from datetime import datetime
# 创建自定义格式化器
class ColoredFormatter(logging.Formatter):
"""带颜色的格式化器"""
COLORS = {
'DEBUG': '\033[36m', # 青色
'INFO': '\033[32m', # 绿色
'WARNING': '\033[33m', # 黄色
'ERROR': '\033[31m', # 红色
'CRITICAL': '\033[35m', # 紫色
'RESET': '\033[0m' # 重置
}
def format(self, record):
log_color = self.COLORS.get(record.levelname, self.COLORS['RESET'])
reset_color = self.COLORS['RESET']
# 设置颜色
record.levelname = f"{log_color}{record.levelname}{reset_color}"
return super().format(record)
# 创建详细格式化器
detailed_formatter = logging.Formatter(
fmt='%(asctime)s | %(levelname)8s | %(name)15s | %(filename)20s:%(lineno)3d | %(funcName)20s() | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# 创建彩色格式化器
colored_formatter = ColoredFormatter(
fmt='%(asctime)s | %(levelname)s | %(name)s | %(message)s',
datefmt='%H:%M:%S'
)
# 创建多个处理器
handlers = []
# 文件处理器 - 按大小轮转
file_handler = logging.handlers.RotatingFileHandler(
'detailed.log',
maxBytes=1024*1024, # 1MB
backupCount=5,
encoding='utf-8'
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(detailed_formatter)
handlers.append(file_handler)
# 文件处理器 - 按时间轮转
time_handler = logging.handlers.TimedRotatingFileHandler(
'app_%Y%m%d.log',
when='midnight',
backupCount=7,
encoding='utf-8'
)
time_handler.setLevel(logging.INFO)
time_handler.setFormatter(detailed_formatter)
handlers.append(time_handler)
# 控制台处理器 - 带颜色
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(colored_formatter)
handlers.append(console_handler)
# 创建专用日志记录器
app_logger = logging.getLogger('myapp')
app_logger.setLevel(logging.DEBUG)
# 清除现有处理器
app_logger.handlers.clear()
# 添加处理器
for handler in handlers:
app_logger.addHandler(handler)
# 测试日志
print("测试专用日志记录器:")
app_logger.debug("调试信息 - 详细检查")
app_logger.info("应用启动")
app_logger.warning("配置项缺失，使用默认值")
app_logger.error("数据库连接失败")
app_logger.critical("系统无法启动")
# 创建子日志记录器
db_logger = logging.getLogger('myapp.database')
db_logger.setLevel(logging.DEBUG)
print("\n测试子日志记录器:")
db_logger.info("连接数据库")
db_logger.debug("执行SQL查询")
db_logger.warning("查询较慢，耗时2.3秒")
# 日志上下文管理
class LoggingContext:
"""日志上下文管理器"""
def __init__(self, logger, level=logging.DEBUG):
self.logger = logger
self.level = level
self.original_level = None
def __enter__(self):
self.original_level = self.logger.level
self.logger.setLevel(self.level)
return self.logger
def __exit__(self, exc_type, exc_val, exc_tb):
self.logger.setLevel(self.original_level)
# 使用日志上下文
print("\n测试日志上下文:")
with LoggingContext(app_logger, logging.DEBUG):
app_logger.debug("上下文中的调试信息")
app_logger.info("上下文中的普通信息")
print("\n上下文外的日志:")
app_logger.debug("上下文外的调试信息（可能不显示）")
# 清理日志文件
import glob
for log_file in ['detailed.log'] + glob.glob('app_*.log') + glob.glob('app_*.log.*'):
try:
if os.path.exists(log_file):
os.remove(log_file)
except:
pass
---
03.结构化日志
a.字典日志和上下文信息
使用结构化日志记录方式，包含丰富的上下文信息。
b.代码示例
---
import logging
import json
import uuid
import threading
# 结构化日志格式化器
class StructuredFormatter(logging.Formatter):
"""JSON格式的结构化日志"""
def format(self, record):
log_entry = {
'timestamp': self.formatTime(record),
'level': record.levelname,
'logger': record.name,
'message': record.getMessage(),
'module': record.module,
'function': record.funcName,
'line': record.lineno,
'thread': threading.current_thread().ident,
'process': os.getpid()
}
# 添加自定义字段
if hasattr(record, 'request_id'):
log_entry['request_id'] = record.request_id
if hasattr(record, 'user_id'):
log_entry['user_id'] = record.user_id
if hasattr(record, 'extra_data'):
log_entry.update(record.extra_data)
return json.dumps(log_entry, ensure_ascii=False)
# 请求上下文日志记录器
class RequestContextLogger:
"""带请求上下文的日志记录器"""
def __init__(self, name):
self.logger = logging.getLogger(name)
self.request_id = str(uuid.uuid4())
self.user_id = None
def set_user(self, user_id):
self.user_id = user_id
def _log(self, level, message, extra=None):
log_extra = {
'request_id': self.request_id,
'user_id': self.user_id,
'extra_data': extra or {}
}
getattr(self.logger, level)(message, extra=log_extra)
def info(self, message, extra=None):
self._log('info', message, extra)
def debug(self, message, extra=None):
self._log('debug', message, extra)
def error(self, message, extra=None):
self._log('error', message, extra)
# 配置结构化日志
structured_handler = logging.FileHandler('structured.log', encoding='utf-8')
structured_handler.setFormatter(StructuredFormatter())
# 创建应用日志记录器
app_logger = logging.getLogger('structured_app')
app_logger.setLevel(logging.DEBUG)
app_logger.handlers.clear()
app_logger.addHandler(structured_handler)
# 创建带上下文的日志记录器
context_logger = RequestContextLogger('structured_app')
context_logger.set_user('user123')
# 测试结构化日志
print("测试结构化日志:")
context_logger.info("用户登录", extra={'ip': '192.168.1.100', 'user_agent': 'Mozilla/5.0'})
context_logger.info("处理订单", extra={'order_id': 'ORD-2025-001', 'amount': 199.99})
context_logger.debug("数据库查询", extra={'query': 'SELECT * FROM users', 'duration': 0.05})
context_logger.error("支付失败", extra={'error_code': 'PAYMENT_DECLINED', 'amount': 199.99})
# 性能日志记录器
class PerformanceLogger:
"""性能监控日志记录器"""
def __init__(self, logger_name='performance'):
self.logger = logging.getLogger(logger_name)
self.metrics = {}
def start_timer(self, operation):
import time
self.metrics[operation] = {
'start_time': time.time()
}
def end_timer(self, operation):
import time
if operation in self.metrics:
duration = time.time() - self.metrics[operation]['start_time']
self.metrics[operation]['duration'] = duration
self.logger.info(f"操作完成: {operation}", extra={
'operation': operation,
'duration': duration,
'type': 'performance'
})
return duration
return 0
# 测试性能日志
perf_logger = PerformanceLogger()
perf_handler = logging.FileHandler('performance.log', encoding='utf-8')
perf_handler.setFormatter(StructuredFormatter())
perf_logger.logger.addHandler(perf_handler)
perf_logger.logger.setLevel(logging.INFO)
print("\n测试性能日志:")
perf_logger.start_timer('data_processing')
import time
time.sleep(0.1) # 模拟处理时间
duration = perf_logger.end_timer('data_processing')
print(f"数据处理耗时: {duration:.3f}秒")
# 清理日志文件
for log_file in ['structured.log', 'performance.log']:
try:
if os.path.exists(log_file):
os.remove(log_file)
except:
pass
---
8.5 argparse命令行参数
01.命令行解析器
a.参数解析
argparse模块是Python标准库中推荐的命令行参数解析模块，提供了强大的参数解析和帮助信息生成功能。
b.代码示例
---
# argparse命令行参数示例
import argparse
import sys
# 创建解析器对象
parser = argparse.ArgumentParser(
prog='程序名',
description='这是一个示例程序',
epilog='程序结束说明'
)
# 添加位置参数（必需参数）
parser.add_argument('filename', help='要处理的文件名')
# 添加可选参数
parser.add_argument('-v', '--verbose', action='store_true',
help='显示详细输出信息')
parser.add_argument('-o', '--output', default='output.txt',
help='输出文件名（默认：output.txt）')
parser.add_argument('-n', '--number', type=int, default=10,
help='处理次数（默认：10）')
# 添加互斥参数组
group = parser.add_mutually_exclusive_group()
group.add_argument('--fast', action='store_true', help='快速模式')
group.add_argument('--slow', action='store_true', help='慢速模式')
# 添加选择参数
parser.add_argument('--mode', choices=['read', 'write', 'append'],
default='read', help='操作模式')
# 解析命令行参数
args = parser.parse_args()
# 使用解析后的参数
print(f"文件名: {args.filename}")
print(f"输出文件: {args.output}")
print(f"处理次数: {args.number}")
print(f"详细模式: {args.verbose}")
print(f"操作模式: {args.mode}")
if args.fast:
print("使用快速模式")
elif args.slow:
print("使用慢速模式")
else:
print("使用正常模式")
---
c.子命令支持
支持创建复杂的命令行工具，包含多个子命令。
d.代码示例
---
# argparse子命令示例
import argparse
import sys
# 创建主解析器
main_parser = argparse.ArgumentParser(description='文件管理工具')
subparsers = main_parser.add_subparsers(dest='command', help='可用命令')
# 添加copy子命令
copy_parser = subparsers.add_parser('copy', help='复制文件')
copy_parser.add_argument('source', help='源文件')
copy_parser.add_argument('destination', help='目标文件')
copy_parser.add_argument('-f', '--force', action='store_true',
help='强制覆盖已存在的文件')
# 添加delete子命令
delete_parser = subparsers.add_parser('delete', help='删除文件')
delete_parser.add_argument('files', nargs='+', help='要删除的文件列表')
delete_parser.add_argument('-r', '--recursive', action='store_true',
help='递归删除目录')
# 添加list子命令
list_parser = subparsers.add_parser('list', help='列出文件')
list_parser.add_argument('directory', nargs='?', default='.',
help='目录路径（默认：当前目录）')
list_parser.add_argument('-a', '--all', action='store_true',
help='显示隐藏文件')
list_parser.add_argument('-l', '--long', action='store_true',
help='显示详细信息')
# 解析参数
args = main_parser.parse_args()
# 执行相应命令
if args.command == 'copy':
print(f"复制: {args.source} -> {args.destination}")
if args.force:
print("强制模式：覆盖已存在文件")
elif args.command == 'delete':
print(f"删除文件: {args.files}")
if args.recursive:
print("递归删除模式")
elif args.command == 'list':
print(f"列出目录: {args.directory}")
if args.all:
print("显示所有文件（包括隐藏文件）")
if args.long:
print("显示详细信息格式")
else:
main_parser.print_help()
---
02.自定义验证
a.类型检查
自定义参数类型验证和转换函数。
b.代码示例
---
# argparse自定义验证示例
import argparse
# 自定义类型函数：验证日期格式
def date_type(date_string):
try:
from datetime import datetime
return datetime.strptime(date_string, '%Y-%m-%d')
except ValueError:
raise argparse.ArgumentTypeError(
f"'{date_string}' 不是有效的日期格式 (YYYY-MM-DD)")
# 自定义类型函数：验证正整数
def positive_int(value):
ivalue = int(value)
if ivalue <= 0:
raise argparse.ArgumentTypeError(f"{value} 必须是正整数")
return ivalue
# 自定义类型函数：验证文件扩展名
def file_extension(value):
valid_extensions = ['.txt', '.csv', '.json']
if not any(value.endswith(ext) for ext in valid_extensions):
raise argparse.ArgumentTypeError(
f"文件扩展名必须是以下之一: {valid_extensions}")
return value
# 创建解析器
parser = argparse.ArgumentParser(description='自定义参数验证示例')
# 使用自定义类型
parser.add_argument('--start-date', type=date_type,
help='开始日期 (YYYY-MM-DD)')
parser.add_argument('--count', type=positive_int,
help='正整数数量')
parser.add_argument('--input-file', type=file_extension,
help='输入文件 (.txt, .csv, .json)')
# 使用choices和自定义函数
def valid_port(port):
port = int(port)
if not 1 <= port <= 65535:
raise argparse.ArgumentTypeError(f'端口必须在1-65535范围内')
return port
parser.add_argument('--port', type=valid_port,
help='端口号 (1-65535)')
# 解析参数
try:
args = parser.parse_args()
print("参数验证通过:")
print(f"开始日期: {args.start_date}")
print(f"数量: {args.count}")
print(f"输入文件: {args.input_file}")
print(f"端口: {args.port}")
except argparse.ArgumentTypeError as e:
print(f"参数错误: {e}")
except SystemExit:
print("参数解析失败，请检查输入参数")
---
9. 开发工具
9.1 unittest单元测试
01.测试框架基础
a.基本概念
unittest是Python的标准单元测试框架，提供了测试用例组织、测试运行、测试结果报告等功能。
b.代码示例
---
# unittest基本测试示例
import unittest
import math
# 定义要测试的函数
def add_numbers(a, b):
"""两数相加的函数"""
return a + b
def is_prime(n):
"""判断是否为质数的函数"""
if n < 2:
return False
for i in range(2, int(math.sqrt(n)) + 1):
if n % i == 0:
return False
return True
# 创建测试类
class MathFunctionsTest(unittest.TestCase):
"""数学函数测试类"""
def test_add_numbers(self):
"""测试add_numbers函数"""
# 正常情况
self.assertEqual(add_numbers(2, 3), 5)
self.assertEqual(add_numbers(-1, 1), 0)
self.assertEqual(add_numbers(0, 0), 0)
# 边界情况
self.assertEqual(add_numbers(999999, 1), 1000000)
# 测试异常情况
with self.assertRaises(TypeError):
add_numbers("a", 1)
def test_is_prime(self):
"""测试is_prime函数"""
# 质数
self.assertTrue(is_prime(2))
self.assertTrue(is_prime(3))
self.assertTrue(is_prime(5))
self.assertTrue(is_prime(7))
self.assertTrue(is_prime(11))
self.assertTrue(is_prime(13))
# 非质数
self.assertFalse(is_prime(1))
self.assertFalse(is_prime(0))
self.assertFalse(is_prime(-1))
self.assertFalse(is_prime(4))
self.assertFalse(is_prime(9))
self.assertFalse(is_prime(100))
# 大质数
self.assertTrue(is_prime(997))
# 运行测试
if __name__ == '__main__':
# 创建测试套件
suite = unittest.TestLoader().loadTestsFromTestCase(MathFunctionsTest)
# 创建测试运行器
runner = unittest.TextTestRunner(verbosity=2)
# 运行测试
result = runner.run(suite)
# 输出测试结果摘要
print(f"\n运行了 {result.testsRun} 个测试")
print(f"失败: {len(result.failures)}")
print(f"错误: {len(result.errors)}")
---
c.测试装置
测试前后的准备和清理工作。
d.代码示例
---
# unittest测试装置示例
import unittest
import os
import tempfile
class TestWithFixtures(unittest.TestCase):
"""测试装置示例类"""
def setUp(self):
"""每个测试方法前执行"""
print("\n setUp: 准备测试环境")
self.test_data = [1, 2, 3, 4, 5]
self.test_dir = tempfile.mkdtemp()
print(f"创建临时目录: {self.test_dir}")
def tearDown(self):
"""每个测试方法后执行"""
print(" tearDown: 清理测试环境")
# 删除临时目录
import shutil
shutil.rmtree(self.test_dir, ignore_errors=True)
@classmethod
def setUpClass(cls):
"""类加载时执行一次"""
print("\n setUpClass: 类级别准备")
cls.shared_resource = "共享资源"
@classmethod
def tearDownClass(cls):
"""类卸载时执行一次"""
print(" tearDownClass: 类级别清理")
def test_list_operations(self):
"""测试列表操作"""
print("运行 test_list_operations")
self.assertEqual(len(self.test_data), 5)
self.test_data.append(6)
self.assertEqual(len(self.test_data), 6)
def test_file_operations(self):
"""测试文件操作"""
print("运行 test_file_operations")
test_file = os.path.join(self.test_dir, "test.txt")
# 写入测试文件
with open(test_file, 'w') as f:
f.write("测试内容")
# 验证文件存在
self.assertTrue(os.path.exists(test_file))
# 读取并验证内容
with open(test_file, 'r') as f:
content = f.read()
self.assertEqual(content, "测试内容")
def test_shared_resource(self):
"""测试共享资源"""
print("运行 test_shared_resource")
self.assertEqual(self.shared_resource, "共享资源")
# 运行测试
if __name__ == '__main__':
unittest.main(verbosity=2)
---
02.断言方法
a.常用断言
unittest提供了多种断言方法来验证测试结果。
b.代码示例
---
# unittest断言方法示例
import unittest
class AssertionMethodsTest(unittest.TestCase):
"""断言方法测试示例"""
def test_equality_assertions(self):
"""测试相等性断言"""
# assertEqual 和 assertNotEqual
self.assertEqual(2 + 2, 4)
self.assertEqual("hello", "hello")
self.assertEqual([1, 2], [1, 2])
self.assertNotEqual(2 + 2, 5)
self.assertNotEqual("hello", "world")
def test_boolean_assertions(self):
"""测试布尔断言"""
self.assertTrue(5 > 3)
self.assertTrue(bool([1, 2, 3]))
self.assertTrue("hello")
self.assertFalse(5 < 3)
self.assertFalse(bool([]))
self.assertFalse("")
def test_comparison_assertions(self):
"""测试比较断言"""
self.assertGreater(10, 5)
self.assertGreaterEqual(10, 10)
self.assertLess(5, 10)
self.assertLessEqual(5, 5)
def test_collection_assertions(self):
"""测试集合断言"""
# 成员关系
self.assertIn(3, [1, 2, 3, 4, 5])
self.assertIn("a", "hello")
self.assertNotIn(6, [1, 2, 3, 4, 5])
# 数量断言
self.assertCountEqual([1, 2, 3], [3, 2, 1]) # 元素相同，顺序可以不同
self.assertCountEqual("abc", "cba") # 字符串中的字符相同
def test_type_assertions(self):
"""测试类型断言"""
self.assertIsInstance(5, int)
self.assertIsInstance("hello", str)
self.assertIsInstance([1, 2], list)
self.assertNotInstance(5, str)
self.assertNotInstance("hello", int)
def test_exception_assertions(self):
"""测试异常断言"""
# 测试是否抛出特定异常
with self.assertRaises(ValueError):
int("abc")
with self.assertRaises(ZeroDivisionError):
result = 1 / 0
# 测试异常消息
with self.assertRaises(ValueError) as context:
int("xyz")
self.assertIn("invalid literal", str(context.exception))
def test_warning_assertions(self):
"""测试警告断言"""
import warnings
# 测试是否产生警告
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
warnings.warn("这是一个警告", UserWarning)
self.assertEqual(len(w), 1)
self.assertTrue(issubclass(w[0].category, UserWarning))
def test_almost_equal_assertions(self):
"""测试浮点数近似断言"""
self.assertAlmostEqual(1.0 + 2.0, 3.0)
self.assertAlmostEqual(0.1 + 0.2, 0.3, places=7) # 7位小数精度
self.assertAlmostEqual(3.1415926, 3.1415927, delta=0.000001)
# 运行测试
if __name__ == '__main__':
unittest.main(verbosity=2)
---
9.2 doctest文档测试
01.doctest基础
a.基本概念
doctest模块可以在文档字符串中嵌入测试用例，将文档和测试结合起来，使文档始终保持准确。
b.代码示例
---
# doctest基础示例
def factorial(n):
"""
计算阶乘的函数
Args:
n (int): 非负整数
Returns:
int: n的阶乘
Examples:
>>> factorial(0)
1
>>> factorial(1)
1
>>> factorial(5)
120
>>> factorial(3) * factorial(4)
144
测试边界情况:
>>> factorial(-1)
Traceback (most recent call last):
...
ValueError: n必须是非负整数
测试大数:
>>> factorial(10)
3628800
"""
if n < 0:
raise ValueError("n必须是非负整数")
if n == 0:
return 1
return n * factorial(n - 1)
def fibonacci(n):
"""
计算斐波那契数列
Args:
n (int): 位置索引
Returns:
int: 第n个斐波那契数
Examples:
>>> fibonacci(0)
0
>>> fibonacci(1)
1
>>> fibonacci(5)
5
>>> fibonacci(10)
55
>>> [fibonacci(i) for i in range(6)]
[0, 1, 1, 2, 3, 5]
"""
if n <= 0:
return 0
elif n == 1:
return 1
else:
a, b = 0, 1
for _ in range(2, n + 1):
a, b = b, a + b
return b
class Calculator:
"""
简单的计算器类
Examples:
>>> calc = Calculator()
>>> calc.add(2, 3)
5
>>> calc.subtract(10, 4)
6
>>> calc.multiply(3, 4)
12
>>> calc.divide(10, 2)
5.0
测试除零错误:
>>> calc.divide(10, 0)
Traceback (most recent call last):
...
ZeroDivisionError: 除数不能为零
"""
def add(self, a, b):
"""加法"""
return a + b
def subtract(self, a, b):
"""减法"""
return a - b
def multiply(self, a, b):
"""乘法"""
return a * b
def divide(self, a, b):
"""除法"""
if b == 0:
raise ZeroDivisionError("除数不能为零")
return a / b
# 运行doctest
if __name__ == '__main__':
import doctest
doctest.testmod(verbose=True)
---
c.测试运行
运行和验证doctest测试用例的方法。
d.代码示例
---
# doctest运行示例
import doctest
import sys
def square_and_cube(x):
"""
计算平方和立方
Examples:
>>> square_and_cube(2)
(4, 8)
>>> square_and_cube(-3)
(9, -27)
>>> square_and_cube(0)
(0, 0)
"""
return x**2, x**3
def process_list(lst):
"""
处理列表：去重、排序、计算统计信息
Examples:
>>> process_list([3, 1, 2, 3, 1])
[1, 2, 3]
>>> process_list([])
[]
>>> process_list([5, 2, 8, 2, 1, 8])
[1, 2, 5, 8]
处理包含负数的列表:
>>> process_list([-1, 3, -2, 1])
[-2, -1, 1, 3]
"""
return sorted(list(set(lst)))
# 不同的运行方式
def run_doctest_examples():
"""展示不同的doctest运行方式"""
print("=== 方式1: 使用testmod() ===")
# 运行当前模块的所有doctest
result1 = doctest.testmod()
print(f"testmod结果: 运行了 {result1.attempted} 个测试，失败 {result1.failed} 个")
print("\n=== 方式2: 使用testfile() ===")
# 从文件运行doctest
test_string = '''
>>> 2 + 2
4
>>> [x*2 for x in range(3)]
[0, 2, 4]
>>> len("hello")
5
'''
result2 = doctest.testmod() # 简化演示
print(f"testfile结果: 运行了 {result2.attempted} 个测试，失败 {result2.failed} 个")
print("\n=== 方式3: 运行特定对象的测试 ===")
# 只测试特定函数
result3 = doctest.run_docstring_examples(square_and_cube, globals())
print("运行特定函数的doctest完成")
print("\n=== 方式4: 详细模式运行 ===")
# 详细模式显示每个测试
result4 = doctest.testmod(verbose=True)
print("\n=== 方式5: 选项设置 ===")
# 设置doctest选项
doctest_options = doctest.ELLIPSIS | doctest.NORMALIZE_WHITESPACE
result5 = doctest.testmod(optionflags=doctest_options)
return result1, result2, result4, result5
if __name__ == '__main__':
run_doctest_examples()
---
02.高级特性
a.测试选项
doctest提供了多种选项来控制测试行为。
b.代码示例
---
# doctest高级特性示例
import doctest
import re
def advanced_function(data, threshold=5):
"""
高级函数示例，展示doctest高级特性
Args:
data: 输入数据（列表或字符串）
threshold: 阈值，默认为5
Returns:
处理后的数据
Examples:
基本使用:
>>> advanced_function([1, 2, 3, 4, 5])
[1, 2, 3, 4, 5]
>>> advanced_function([1, 2, 10, 3])
[1, 2, 10, 3]
处理字符串（使用ELLIPSIS匹配模式）:
>>> advanced_function("hello world")
'hello...'
处理空输入:
>>> advanced_function([])
[]
测试字典输入（使用NORMALIZE_WHITESPACE忽略空白）:
>>> advanced_function({'a': 1, 'b': 2})
{'a': 1, 'b': 2}
测试异常情况:
>>> advanced_function(None)
Traceback (most recent call last):
...
TypeError: 不支持的数据类型
比较数字（使用绝对比较）:
>>> abs(advanced_function([3.14159, 2.71828])[0] - 3.14159) < 0.0001
True
"""
if data is None:
raise TypeError("不支持的数据类型")
if isinstance(data, str):
return data[:threshold] + "..."
if isinstance(data, (list, tuple)):
return list(data)
if isinstance(data, dict):
return data
return data
# 设置测试选项的示例
def demonstrate_options():
"""
演示doctest选项的使用
>>> # 使用ELLIPSIS选项进行模糊匹配
>>> long_text = "This is a very long text that we want to match"
>>> long_text[:10] + "..."
'This is...'
>>> # 使用NORMALIZE_WHITESPACE忽略空白差异
>>> result = {
... 'name': 'John',
... 'age': 30
... }
>>> result # doctest: +NORMALIZE_WHITESPACE
{'name': 'John', 'age': 30}
>>> # 使用SKIP选项跳过测试
>>> print("这行会被跳过") # doctest: +SKIP
这行会被跳过
"""
pass
class DataProcessor:
"""
数据处理器类示例
Examples:
创建处理器:
>>> processor = DataProcessor()
>>> processor is not None
True
处理数值列表:
>>> processor.process_numbers([1, 2, 3, 4, 5])
15
处理字符串列表:
>>> processor.process_strings(['a', 'b', 'c'])
'abc'
处理混合类型（使用ELLIPSIS忽略不重要的差异）:
>>> result = processor.process_mixed([1, 'a', True, None])
>>> isinstance(result, dict)
True
>>> 'numbers' in result
True
>>> 'strings' in result
True
"""
def process_numbers(self, numbers):
"""处理数字列表"""
return sum(numbers)
def process_strings(self, strings):
"""处理字符串列表"""
return ''.join(strings)
def process_mixed(self, data):
"""处理混合类型数据"""
result = {
'numbers': [x for x in data if isinstance(x, (int, float))],
'strings': [x for x in data if isinstance(x, str)],
'booleans': [x for x in data if isinstance(x, bool)],
'others': [x for x in data if x is not None and not isinstance(x, (int, float, str, bool))]
}
return result
# 运行带选项的doctest
if __name__ == '__main__':
# 设置组合选项
flags = doctest.ELLIPSIS | doctest.NORMALIZE_WHITESPACE | doctest.REPORT_ONLY_FIRST_FAILURE
print("运行带选项的doctest:")
print(f"ELLIPSIS: 允许模糊匹配")
print(f"NORMALIZE_WHITESPACE: 忽略空白差异")
print(f"REPORT_ONLY_FIRST_FAILURE: 只报告第一次失败")
result = doctest.testmod(optionflags=flags, verbose=True)
print(f"\n测试完成: 运行了 {result.attempted} 个测试，失败 {result.failed} 个")
---
9.3 pdb调试器
01.pdb基础使用
a.基本概念
pdb是Python的交互式调试器，可以让程序在断点处停止，检查变量值，单步执行代码，是强大的调试工具。
b.代码示例
---
# pdb基础调试示例
import pdb
def calculate_average(numbers):
"""计算列表中数字的平均值"""
# 在这里设置断点，调试程序
pdb.set_trace() # 程序会在这里暂停
print(f"输入的数字列表: {numbers}")
print(f"列表长度: {len(numbers)}")
if not numbers:
return 0
total = 0
for i, num in enumerate(numbers):
print(f"处理第{i+1}个数: {num}")
total += num
# 可以在这里检查每一步的结果
average = total / len(numbers)
print(f"总和: {total}, 平均值: {average}")
return average
def find_primes(n):
"""找出小于n的所有质数"""
primes = []
for num in range(2, n):
is_prime = True
pdb.set_trace() # 调试质数判断逻辑
for i in range(2, int(num**0.5) + 1):
if num % i == 0:
is_prime = False
break
if is_prime:
primes.append(num)
print(f"找到质数: {num}")
return primes
def process_data(data):
"""处理数据的复杂函数"""
# 第一步：数据预处理
pdb.set_trace()
processed = []
for item in data:
if isinstance(item, str):
processed.append(item.strip().upper())
elif isinstance(item, (int, float)):
processed.append(item * 2)
else:
processed.append(str(item))
# 第二步：数据分析
unique_items = list(set(processed))
counts = {}
for item in unique_items:
counts[item] = processed.count(item)
# 第三步：生成结果
result = {
'original_length': len(data),
'processed_length': len(processed),
'unique_count': len(unique_items),
'item_counts': counts
}
return result
# 演示调试过程
if __name__ == '__main__':
print("=== 演示1: 计算平均值 ===")
numbers = [1, 2, 3, 4, 5]
avg = calculate_average(numbers)
print(f"平均值为: {avg}")
print("\n=== 演示2: 查找质数 ===")
primes = find_primes(20)
print(f"小于20的质数: {primes}")
print("\n=== 演示3: 处理数据 ===")
test_data = ["hello", 123, " world ", 45.6, True, "test"]
result = process_data(test_data)
print(f"处理结果: {result}")
---
c.调试命令
pdb提供了丰富的命令来控制程序执行和检查状态。
d.代码示例
---
# pdb命令示例
import pdb
class Debugger:
"""演示pdb调试命令的类"""
def __init__(self, name):
self.name = name
self.counter = 0
def process_list(self, data_list):
"""处理列表数据的方法"""
print(f"开始处理列表，初始数据: {data_list}")
pdb.set_trace() # 主要调试点
# 过滤数据
filtered_data = []
for i, item in enumerate(data_list):
self.counter += 1
if item % 2 == 0: # 只保留偶数
filtered_data.append(item)
print(f"保留偶数: {item}")
# 转换数据
processed_data = []
for item in filtered_data:
processed_data.append(item * self.counter)
return processed_data
def analyze_string(self, text):
"""分析字符串的方法"""
pdb.set_trace() # 字符串分析调试点
words = text.split()
word_count = len(words)
char_count = len(text)
avg_word_length = sum(len(word) for word in words) / word_count if words else 0
analysis = {
'original_text': text,
'word_count': word_count,
'char_count': char_count,
'avg_word_length': avg_word_length,
'has_numbers': any(c.isdigit() for c in text),
'has_uppercase': any(c.isupper() for c in text)
}
return analysis
def recursive_function(n, depth=0):
"""递归函数调试示例"""
print(f" " * depth + f"递归调用: n={n}, depth={depth}")
pdb.set_trace() # 递归调试点
if n <= 1:
return n
else:
result = recursive_function(n-1, depth+1) + recursive_function(n-2, depth+1)
print(f" " * depth + f"返回结果: {result}")
return result
def demonstrate_debug_commands():
"""
演示常用调试命令的使用场景
pdb常用命令:
- l(list): 显示当前代码
- n(next): 执行下一行
- s(step): 进入函数
- c(continue): 继续执行
- p(print): 打印变量
- pp(pretty-print): 格式化打印
- a(args): 显示函数参数
- w(where): 显示调用栈
- b(break): 设置断点
- cl(clear): 清除断点
- h(help): 显示帮助
"""
debugger = Debugger("TestDebugger")
print("=== 测试1: 列表处理 ===")
test_list = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
result1 = debugger.process_list(test_list)
print(f"列表处理结果: {result1}")
print("\n=== 测试2: 字符串分析 ===")
test_text = "Hello World 123 Python Programming"
result2 = debugger.analyze_string(test_text)
print(f"字符串分析结果: {result2}")
print("\n=== 测试3: 递归函数 ===")
fib_result = recursive_function(5)
print(f"斐波那契数列第5项: {fib_result}")
if __name__ == '__main__':
demonstrate_debug_commands()
---
02.高级调试技巧
a.断点管理
设置条件断点、临时断点和断点管理的高级技巧。
b.代码示例
---
# pdb高级调试技巧示例
import pdb
import random
class DataAnalyzer:
"""数据分析器，演示高级调试技巧"""
def __init__(self):
self.data_cache = {}
self.processing_count = 0
def analyze_complex_data(self, data, threshold=10):
"""复杂数据分析方法"""
print(f"开始分析数据: {data}")
print(f"阈值设置: {threshold}")
# 条件断点示例：只在特定条件下触发
if len(data) > threshold:
pdb.set_trace() # 只在数据量大于阈值时调试
result = {
'original_count': len(data),
'filtered_data': [],
'statistics': {}
}
# 数据过滤阶段
for i, item in enumerate(data):
self.processing_count += 1
# 临时调试：查看每个处理步骤
if i % 5 == 0: # 每5个项目调试一次
print(f"处理进度: {i}/{len(data)}")
pdb.set_trace()
if self._is_valid_item(item):
result['filtered_data'].append(item)
# 统计分析阶段
result['statistics'] = self._calculate_statistics(result['filtered_data'])
return result
def _is_valid_item(self, item):
"""验证数据项的有效性"""
# 在特定条件下触发调试
if isinstance(item, str) and len(item) > 10:
pdb.set_trace() # 调试长字符串处理
if isinstance(item, (int, float)):
return item >= 0
elif isinstance(item, str):
return len(item.strip()) > 0
else:
return False
def _calculate_statistics(self, data):
"""计算数据统计信息"""
if not data:
return {}
pdb.set_trace() # 统计计算调试点
stats = {}
numbers = [x for x in data if isinstance(x, (int, float))]
strings = [x for x in data if isinstance(x, str)]
if numbers:
stats['numeric'] = {
'count': len(numbers),
'sum': sum(numbers),
'average': sum(numbers) / len(numbers),
'min': min(numbers),
'max': max(numbers)
}
if strings:
stats['string'] = {
'count': len(strings),
'total_length': sum(len(s) for s in strings),
'avg_length': sum(len(s) for s in strings) / len(strings)
}
return stats
def debugging_techniques():
"""演示各种调试技巧"""
# 技巧1: 函数断点
def debug_function_entry():
pdb.set_trace()
print("进入函数")
# 技巧2: 异常处理调试
def debug_with_exception_handling():
try:
# 可能出错的代码
result = 10 / 0
except Exception as e:
print(f"捕获异常: {e}")
pdb.set_trace() # 在异常处理时调试
result = None
return result
# 技巧3: 循环中的条件调试
def debug_in_loop(data):
for i, item in enumerate(data):
if i == len(data) // 2: # 在循环中间调试
pdb.set_trace()
# 处理item...
# 技巧4: 调用栈分析
def deep_function_a():
return deep_function_b()
def deep_function_b():
return deep_function_c()
def deep_function_c():
pdb.set_trace() # 查看完整调用栈
return "来自深层调用的结果"
print("=== 调试技巧演示 ===")
print("1. 函数入口调试:")
debug_function_entry()
print("\n2. 异常处理调试:")
debug_with_exception_handling()
print("\n3. 循环调试:")
test_data = list(range(10))
debug_in_loop(test_data)
print("\n4. 调用栈分析:")
result = deep_function_a()
print(f"深层函数结果: {result}")
# 主程序演示
if __name__ == '__main__':
# 创建数据分析器
analyzer = DataAnalyzer()
# 生成测试数据
test_data = [
1, 2, 3, "hello", "world", 4.5, 6.7,
"python", "programming", -1, 0, "test",
"debugging", 42, "数据分析", "complex_string_example"
]
print("=== 复杂数据分析演示 ===")
analysis_result = analyzer.analyze_complex_data(test_data, threshold=5)
print(f"分析结果: {analysis_result}")
print("\n=== 调试技巧演示 ===")
debugging_techniques()
---
9.4 timeit性能测试
01.timeit基础
a.基本概念
timeit模块专门用于精确测量小段代码的执行时间，可以避免系统时间精度不足的问题。
b.代码示例
---
# timeit基础性能测试示例
import timeit
import random
# 定义要测试的函数
def sum_with_loop(numbers):
"""使用循环求和"""
total = 0
for num in numbers:
total += num
return total
def sum_with_builtin(numbers):
"""使用内置sum函数"""
return sum(numbers)
def sum_with_reduce(numbers):
"""使用reduce求和"""
from functools import reduce
return reduce(lambda x, y: x + y, numbers)
# 基础timeit测试
def basic_timeit_tests():
"""演示基本的timeit使用方法"""
# 生成测试数据
numbers = list(range(1000))
print("=== 基础timeit测试 ===")
# 方法1: 使用timeit.timeit()
print("\n1. 使用timeit.timeit():")
loop_time = timeit.timeit('sum_with_loop(numbers)',
setup='from __main__ import sum_with_loop, numbers',
number=1000)
print(f"循环求和时间: {loop_time:.6f}秒")
builtin_time = timeit.timeit('sum_with_builtin(numbers)',
setup='from __main__ import sum_with_builtin, numbers',
number=1000)
print(f"内置函数时间: {builtin_time:.6f}秒")
reduce_time = timeit.timeit('sum_with_reduce(numbers)',
setup='from __main__ import sum_with_reduce, numbers',
number=1000)
print(f"reduce函数时间: {reduce_time:.6f}秒")
# 方法2: 使用timeit.repeat()
print("\n2. 使用timeit.repeat()进行多次测试:")
loop_times = timeit.repeat('sum_with_loop(numbers)',
setup='from __main__ import sum_with_loop, numbers',
number=1000, repeat=5)
print(f"循环求和多次测试: {loop_times}")
print(f"平均时间: {sum(loop_times)/len(loop_times):.6f}秒")
print(f"最快时间: {min(loop_times):.6f}秒")
print(f"最慢时间: {max(loop_times):.6f}秒")
# 方法3: 使用timeit.Timer对象
print("\n3. 使用timeit.Timer对象:")
timer = timeit.Timer('sum_with_builtin(numbers)',
'from __main__ import sum_with_builtin, numbers')
timer_time = timer.timeit(number=1000)
print(f"Timer对象测试时间: {timer_time:.6f}秒")
def test_string_operations():
"""测试字符串操作性能"""
print("\n=== 字符串操作性能测试 ===")
# 字符串拼接方式比较
test_list = ['hello'] * 100
# 方式1: 使用+拼接
join_plus_time = timeit.timeit(
'result = ""\nfor item in test_list: result += item',
setup='from __main__ import test_list',
number=1000
)
print(f"使用+拼接时间: {join_plus_time:.6f}秒")
# 方式2: 使用join方法
join_method_time = timeit.timeit(
'result = "".join(test_list)',
setup='from __main__ import test_list',
number=1000
)
print(f"使用join时间: {join_method_time:.6f}秒")
# 方式3: 使用StringIO
join_io_time = timeit.timeit(
'''
from io import StringIO
buffer = StringIO()
for item in test_list:
buffer.write(item)
result = buffer.getvalue()
''',
setup='from __main__ import test_list',
number=1000
)
print(f"使用StringIO时间: {join_io_time:.6f}秒")
# 性能提升比例
improvement = (join_plus_time - join_method_time) / join_plus_time * 100
print(f"join方法比+操作快 {improvement:.1f}%")
if __name__ == '__main__':
basic_timeit_tests()
test_string_operations()
---
c.性能比较
比较不同算法或实现的性能差异。
d.代码示例
---
# timeit性能比较示例
import timeit
import random
itertools
# 排序算法比较
def bubble_sort(arr):
"""冒泡排序"""
n = len(arr)
for i in range(n):
for j in range(0, n-i-1):
if arr[j] > arr[j+1]:
arr[j], arr[j+1] = arr[j+1], arr[j]
return arr
def quick_sort(arr):
"""快速排序"""
if len(arr) <= 1:
return arr
pivot = arr[len(arr) // 2]
left = [x for x in arr if x < pivot]
middle = [x for x in arr if x == pivot]
right = [x for x in arr if x > pivot]
return quick_sort(left) + middle + quick_sort(right)
def builtin_sort(arr):
"""内置排序"""
return sorted(arr)
# 查找算法比较
def linear_search(arr, target):
"""线性查找"""
for i, item in enumerate(arr):
if item == target:
return i
return -1
def binary_search(arr, target):
"""二分查找（要求有序数组）"""
left, right = 0, len(arr) - 1
while left <= right:
mid = (left + right) // 2
if arr[mid] == target:
return mid
elif arr[mid] < target:
left = mid + 1
else:
right = mid - 1
return -1
def compare_sorting_algorithms():
"""比较排序算法性能"""
print("=== 排序算法性能比较 ===")
# 测试不同规模的数据
sizes = [100, 500, 1000]
algorithms = [
('冒泡排序', 'bubble_sort(data.copy())'),
('快速排序', 'quick_sort(data.copy())'),
('内置排序', 'builtin_sort(data.copy())')
]
for size in sizes:
print(f"\n数据规模: {size}")
data = list(range(size, 0, -1)) # 逆序数据
for name, stmt in algorithms:
try:
# 限制复杂算法的测试次数
num_runs = 1 if name == '冒泡排序' and size > 500 else 10
time_taken = timeit.timeit(
stmt,
setup='from __main__ import ' + ', '.join([alg[0].replace(' ', '_').replace('排', '_').replace('序', '_') for alg in algorithms]) + ', data',
number=num_runs
)
avg_time = time_taken / num_runs
print(f"{name:10}: {avg_time:.6f}秒")
except Exception as e:
print(f"{name:10}: 测试失败 - {e}")
def compare_search_algorithms():
"""比较查找算法性能"""
print("\n=== 查找算法性能比较 ===")
sizes = [1000, 5000, 10000]
for size in sizes:
print(f"\n数据规模: {size}")
# 创建有序数据
sorted_data = list(range(size))
target = size - 1 # 查找最后一个元素
# 线性查找
linear_time = timeit.timeit(
'linear_search(sorted_data, target)',
setup='from __main__ import linear_search, sorted_data, target',
number=1000
)
print(f"线性查找: {linear_time:.6f}秒")
# 二分查找
binary_time = timeit.timeit(
'binary_search(sorted_data, target)',
setup='from __main__ import binary_search, sorted_data, target',
number=1000
)
print(f"二分查找: {binary_time:.6f}秒")
# 性能提升
improvement = (linear_time - binary_time) / linear_time * 100
print(f"性能提升: {improvement:.1f}%")
def compare_list_operations():
"""比较列表操作性能"""
print("\n=== 列表操作性能比较 ===")
test_size = 10000
test_data = list(range(test_size))
operations = [
('列表追加', 'test_data.append(0)'),
('列表插入(头部)', 'test_data.insert(0, 0)'),
('列表删除(尾部)', 'test_data.pop()'),
('列表删除(头部)', 'test_data.pop(0)'),
('列表查找', 'test_data.index(test_size//2)'),
('列表反转', 'test_data.reverse()'),
]
for name, operation in operations:
# 每次操作都使用新的数据副本
time_taken = timeit.timeit(
operation,
setup=f'test_data = list(range({test_size}))',
number=100
)
print(f"{name:15}: {time_taken:.6f}秒")
if __name__ == '__main__':
compare_sorting_algorithms()
compare_search_algorithms()
compare_list_operations()
---
02.高级应用
a.内存性能
使用timeit测试内存相关操作的性能。
b.代码示例
---
# timeit高级应用示例
import timeit
import random
import gc
import sys
try:
import psutil # 需要安装: pip install psutil
except ImportError:
psutil = None
def get_memory_usage():
"""获取当前内存使用量"""
process = psutil.Process()
return process.memory_info().rss / 1024 / 1024 # MB
def test_memory_operations():
"""测试内存相关操作的性能"""
print("=== 内存操作性能测试 ===")
# 测试列表推导vs循环
list_comp_time = timeit.timeit(
'[x**2 for x in range(10000)]',
number=100
)
loop_time = timeit.timeit(
'''
result = []
for x in range(10000):
result.append(x**2)
''',
number=100
)
print(f"列表推导: {list_comp_time:.6f}秒")
print(f"循环构造: {loop_time:.6f}秒")
# 测试生成器表达式vs列表推导
gen_exp_time = timeit.timeit(
'sum(x**2 for x in range(10000))',
number=100
)
list_sum_time = timeit.timeit(
'sum([x**2 for x in range(10000)])',
number=100
)
print(f"生成器表达式: {gen_exp_time:.6f}秒")
print(f"列表推导求和: {list_sum_time:.6f}秒")
def test_gc_performance():
"""测试垃圾回收对性能的影响"""
print("\n=== 垃圾回收性能测试 ===")
def create_objects():
"""创建大量临时对象"""
objects = []
for i in range(1000):
objects.append({'id': i, 'data': 'x' * 100})
return objects
# 启用垃圾回收
gc.enable()
with_gc_time = timeit.timeit('create_objects()',
setup='from __main__ import create_objects',
number=100)
# 禁用垃圾回收
gc.disable()
without_gc_time = timeit.timeit('create_objects()',
setup='from __main__ import create_objects',
number=100)
# 重新启用垃圾回收
gc.enable()
print(f"启用垃圾回收: {with_gc_time:.6f}秒")
print(f"禁用垃圾回收: {without_gc_time:.6f}秒")
print(f"GC开销: {((with_gc_time - without_gc_time) / without_gc_time * 100):.1f}%")
def comprehensive_performance_test():
"""综合性能测试套件"""
print("\n=== 综合性能测试 ===")
test_functions = [
('数字计算', 'sum(range(10000))'),
('字符串操作', '" ".join(["hello"] * 1000)'),
('列表操作', 'list(range(1000))[::2]'),
('字典操作', 'dict((i, i*2) for i in range(1000))'),
('集合操作', 'set(range(1000)) | set(range(500, 1500))'),
('文件操作模拟', 'open("temp.txt", "w").write("test" * 1000); open("temp.txt", "r").read()'),
]
results = []
for name, test_code in test_functions:
# 预热
try:
timeit.timeit(test_code, number=1)
# 实际测试
exec_time = timeit.timeit(test_code, number=10)
avg_time = exec_time / 10
results.append((name, avg_time))
print(f"{name:12}: {avg_time:.6f}秒")
except Exception as e:
print(f"{name:12}: 测试失败 - {e}")
# 排序并显示结果
results.sort(key=lambda x: x[1])
print("\n性能排名（从快到慢）:")
for i, (name, time_taken) in enumerate(results, 1):
print(f"{i}. {name:12}: {time_taken:.6f}秒")
def benchmark_with_profiling():
"""结合内存监控的性能测试"""
print("\n=== 内存监控性能测试 ===")
def memory_intensive_task():
"""内存密集型任务"""
big_data = []
for i in range(1000):
big_data.append([0] * 1000) # 创建大列表
return len(big_data)
# 监控内存使用
initial_memory = get_memory_usage()
print(f"初始内存使用: {initial_memory:.2f} MB")
# 执行性能测试
exec_time = timeit.timeit('memory_intensive_task()',
setup='from __main__ import memory_intensive_task',
number=10)
peak_memory = get_memory_usage()
print(f"峰值内存使用: {peak_memory:.2f} MB")
print(f"内存增长: {peak_memory - initial_memory:.2f} MB")
print(f"执行时间: {exec_time:.6f}秒")
# 清理内存
gc.collect()
final_memory = get_memory_usage()
print(f"清理后内存: {final_memory:.2f} MB")
if __name__ == '__main__':
try:
test_memory_operations()
test_gc_performance()
comprehensive_performance_test()
benchmark_with_profiling()
except ImportError:
print("注意: 需要安装psutil模块: pip install psutil")
# 即使没有psutil，也可以运行基本测试
test_memory_operations()
comprehensive_performance_test()
---
9.5 profile性能分析
01.profile基础
a.基本概念
profile模块提供Python程序的详细性能分析，可以统计每个函数的调用次数、执行时间等详细信息。
b.代码示例
---
# profile基础性能分析示例
import profile
import cProfile
import pstats
import io
import random
# 定义一些测试函数
def fibonacci_recursive(n):
"""递归实现的斐波那契函数"""
if n <= 1:
return n
return fibonacci_recursive(n-1) + fibonacci_recursive(n-2)
def fibonacci_iterative(n):
"""迭代实现的斐波那契函数"""
if n <= 1:
return n
a, b = 0, 1
for _ in range(2, n + 1):
a, b = b, a + b
return b
def process_data(data):
"""处理数据的函数"""
result = []
for item in data:
if isinstance(item, int):
result.append(item * 2)
elif isinstance(item, str):
result.append(item.upper())
else:
result.append(str(item))
return sorted(result)
def complex_algorithm(n):
"""复杂算法示例"""
# 生成测试数据
data = [random.randint(1, 100) for _ in range(n)]
strings = [f"item_{i}" for i in range(n)]
# 处理数据
processed_numbers = [x * 2 for x in data if x % 2 == 0]
filtered_strings = [s for s in strings if len(s) > 5]
# 计算统计信息
numbers_sum = sum(processed_numbers)
numbers_avg = numbers_sum / len(processed_numbers) if processed_numbers else 0
# 排序操作
sorted_numbers = sorted(processed_numbers)
sorted_strings = sorted(filtered_strings, reverse=True)
return {
'numbers_count': len(processed_numbers),
'strings_count': len(filtered_strings),
'numbers_sum': numbers_sum,
'numbers_avg': numbers_avg,
'sorted_data': process_data(data + strings)
}
def demonstrate_profile():
"""演示profile的基本使用"""
print("=== profile基础性能分析 ===")
# 1. 使用cProfile进行基本分析
print("\n1. 使用cProfile分析复杂算法:")
cProfile.run('complex_algorithm(1000)', 'algorithm_profile.prof')
# 2. 读取并显示分析结果
print("\n2. 分析结果:")
stats = pstats.Stats('algorithm_profile.prof')
stats.sort_stats('cumulative') # 按累计时间排序
stats.print_stats(10) # 显示前10个最耗时的函数
# 3. 按函数调用次数排序
print("\n3. 按调用次数排序:")
stats.sort_stats('calls')
stats.print_stats(10)
# 4. 比较不同实现
print("\n4. 比较递归和迭代斐波那契:")
test_n = 30
# 递归版本分析
print("\n递归版本分析:")
cProfile.run(f'fibonacci_recursive({test_n})', 'recursive_fib.prof')
recursive_stats = pstats.Stats('recursive_fib.prof')
recursive_stats.sort_stats('cumulative')
recursive_stats.print_stats(5)
# 迭代版本分析
print("\n迭代版本分析:")
cProfile.run(f'fibonacci_iterative({test_n})', 'iterative_fib.prof')
iterative_stats = pstats.Stats('iterative_fib.prof')
iterative_stats.sort_stats('cumulative')
iterative_stats.print_stats(5)
def detailed_profile_analysis():
"""详细的性能分析"""
print("\n=== 详细性能分析 ===")
# 创建分析器
profiler = cProfile.Profile()
profiler.enable()
# 执行要分析的代码
print("执行被分析的代码...")
result = complex_algorithm(500)
fibonacci_result = fibonacci_iterative(20)
data_result = process_data([1, 2, 'hello', 3.14, True, 'world'])
profiler.disable()
# 创建统计对象
stats = pstats.Stats(profiler)
# 生成各种格式的报告
print("\n=== 按累计时间排序 ===")
stats.sort_stats('cumulative')
stats.print_stats(15)
print("\n=== 按自身时间排序 ===")
stats.sort_stats('time')
stats.print_stats(15)
print("\n=== 按函数名排序 ===")
stats.sort_stats('filename')
stats.print_stats(15)
if __name__ == '__main__':
demonstrate_profile()
detailed_profile_analysis()
---
c.结果分析
如何解读profile输出结果和优化建议。
d.代码示例
---
# profile结果分析和优化示例
import cProfile
import pstats
import io
import time
def inefficient_string_operation(text_list):
"""低效的字符串操作"""
result = ""
for text in text_list:
result += text.upper() + " "
return result.strip()
def efficient_string_operation(text_list):
"""高效的字符串操作"""
return " ".join(text.upper() for text in text_list)
def inefficient_list_processing(data_list):
"""低效的列表处理"""
result = []
for i in range(len(data_list)):
for j in range(len(data_list)):
if data_list[i] == data_list[j] and i != j:
if data_list[i] not in result:
result.append(data_list[i])
return result
def efficient_list_processing(data_list):
"""高效的列表处理"""
seen = set()
result = []
for item in data_list:
if data_list.count(item) > 1 and item not in seen:
result.append(item)
seen.add(item)
return result
def bubble_sort(arr):
"""冒泡排序（低效）"""
n = len(arr)
for i in range(n):
for j in range(0, n-i-1):
if arr[j] > arr[j+1]:
arr[j], arr[j+1] = arr[j+1], arr[j]
return arr
def quick_sort(arr):
"""快速排序（高效）"""
if len(arr) <= 1:
return arr
pivot = arr[len(arr) // 2]
left = [x for x in arr if x < pivot]
middle = [x for x in arr if x == pivot]
right = [x for x in arr if x > pivot]
return quick_sort(left) + middle + quick_sort(right)
def analyze_and_compare():
"""分析并比较不同实现的性能"""
print("=== 性能分析和比较 ===")
# 测试数据
test_strings = ["hello"] * 1000
test_numbers = list(range(100)) + [50] * 10 # 包含重复元素
test_list = list(range(100, 0, -1))
# 1. 字符串操作比较
print("\n1. 字符串操作性能比较:")
print("\n低效实现:")
profiler1 = cProfile.Profile()
profiler1.enable()
start_time = time.time()
result1 = inefficient_string_operation(test_strings)
end_time = time.time()
profiler1.disable()
print(f"执行时间: {end_time - start_time:.6f}秒")
stats1 = pstats.Stats(profiler1)
stats1.sort_stats('cumulative')
stats1.print_stats(3)
print("\n高效实现:")
profiler2 = cProfile.Profile()
profiler2.enable()
start_time = time.time()
result2 = efficient_string_operation(test_strings)
end_time = time.time()
profiler2.disable()
print(f"执行时间: {end_time - start_time:.6f}秒")
stats2 = pstats.Stats(profiler2)
stats2.sort_stats('cumulative')
stats2.print_stats(3)
# 2. 列表处理比较
print("\n2. 列表处理性能比较:")
print("\n低效实现:")
profiler3 = cProfile.Profile()
profiler3.enable()
result3 = inefficient_list_processing(test_numbers[:50]) # 减少数据量避免太慢
profiler3.disable()
stats3 = pstats.Stats(profiler3)
stats3.sort_stats('cumulative')
stats3.print_stats(5)
print("\n高效实现:")
profiler4 = cProfile.Profile()
profiler4.enable()
result4 = efficient_list_processing(test_numbers)
profiler4.disable()
stats4 = pstats.Stats(profiler4)
stats4.sort_stats('cumulative')
stats4.print_stats(5)
# 3. 排序算法比较
print("\n3. 排序算法性能比较:")
print("\n冒泡排序:")
profiler5 = cProfile.Profile()
profiler5.enable()
result5 = bubble_sort(test_list[:20]) # 小数据量
profiler5.disable()
stats5 = pstats.Stats(profiler5)
stats5.sort_stats('cumulative')
stats5.print_stats(5)
print("\n快速排序:")
profiler6 = cProfile.Profile()
profiler6.enable()
result6 = quick_sort(test_list)
profiler6.disable()
stats6 = pstats.Stats(profiler6)
stats6.sort_stats('cumulative')
stats6.print_stats(5)
def generate_optimization_report():
"""生成优化报告"""
print("\n=== 优化建议报告 ===")
# 创建一个综合性能测试
profiler = cProfile.Profile()
profiler.enable()
# 执行一系列操作
data = list(range(1000))
strings = [f"test_{i}" for i in range(1000)]
# 模拟一些常见操作
sorted_data = sorted(data, reverse=True)
unique_items = list(set(data))
string_result = " ".join(strings)
data_dict = {i: f"value_{i}" for i in data}
profiler.disable()
# 分析结果并提供建议
stats = pstats.Stats(profiler, stream=io.StringIO())
stats.sort_stats('cumulative')
stats.print_stats()
# 获取分析结果
result_stream = stats.stream
result_stream.seek(0)
profile_output = result_stream.getvalue()
print("性能分析完成!")
print("主要发现:")
print("1. 内置函数（如sorted, set）通常比自定义实现更快")
print("2. 列表推导比传统循环更高效")
print("3. 字符串拼接使用join比+操作更高效")
print("4. 适当的数据结构选择可以显著提升性能")
print(f"\n分析了 {len(profile_output.split(chr(10)))} 行性能数据")
print("建议重点关注执行时间最长的函数进行优化")
if __name__ == '__main__':
analyze_and_compare()
generate_optimization_report()
---
02.高级分析
a.内存分析
结合内存使用情况的深度性能分析。
b.代码示例
---
# profile高级分析示例
import cProfile
import pstats
import io
import time
import gc
import tracemalloc
class PerformanceAnalyzer:
"""性能分析器类"""
def __init__(self):
self.profiler = cProfile.Profile()
self.memory_snapshots = []
def start_analysis(self):
"""开始性能分析"""
print("开始性能分析...")
gc.collect() # 清理内存
tracemalloc.start() # 开始内存跟踪
self.profiler.enable()
self.start_time = time.time()
def stop_analysis(self):
"""停止性能分析"""
self.profiler.disable()
self.end_time = time.time()
current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
execution_time = self.end_time - self.start_time
memory_used = current / 1024 / 1024 # MB
memory_peak = peak / 1024 / 1024 # MB
return {
'execution_time': execution_time,
'memory_used': memory_used,
'memory_peak': memory_peak
}
def get_detailed_stats(self):
"""获取详细的性能统计"""
stats = pstats.Stats(self.profiler, stream=io.StringIO())
stats.sort_stats('cumulative')
stats.print_stats()
# 获取统计结果
result_stream = stats.stream
result_stream.seek(0)
return result_stream.getvalue()
def memory_intensive_operation():
"""内存密集型操作"""
big_lists = []
for i in range(10):
big_list = [0] * 100000 # 创建大列表
big_lists.append(big_list)
# 模拟一些处理
processed = []
for lst in big_lists:
processed.append(sum(lst))
return processed
def cpu_intensive_operation():
"""CPU密集型操作"""
result = []
for i in range(100000):
# 模拟复杂计算
num = i * i
if num % 2 == 0:
result.append(num // 2)
else:
result.append(num * 3 + 1)
return result[:1000] # 只返回一部分结果
def io_simulated_operation():
"""模拟I/O操作"""
results = []
for i in range(100):
# 模拟文件操作
data = "x" * 1000
processed = data.upper()
results.append(len(processed))
time.sleep(0.001) # 模拟I/O延迟
return results
def complex_data_processing():
"""复杂数据处理操作"""
import random
random.seed(42) # 确保可重复性
# 生成测试数据
data = []
for i in range(5000):
item = {
'id': i,
'value': random.randint(1, 100),
'category': random.choice(['A', 'B', 'C']),
'timestamp': time.time() + i
}
data.append(item)
# 数据处理步骤
filtered_data = [item for item in data if item['value'] > 50]
grouped_data = {}
for item in filtered_data:
category = item['category']
if category not in grouped_data:
grouped_data[category] = []
grouped_data[category].append(item)
# 计算统计信息
statistics = {}
for category, items in grouped_data.items():
values = [item['value'] for item in items]
statistics[category] = {
'count': len(items),
'avg_value': sum(values) / len(values),
'max_value': max(values),
'min_value': min(values)
}
return statistics
def comprehensive_performance_analysis():
"""综合性能分析"""
print("=== 综合性能分析 ===")
analyzer = PerformanceAnalyzer()
operations = [
('内存密集型操作', memory_intensive_operation),
('CPU密集型操作', cpu_intensive_operation),
('模拟I/O操作', io_simulated_operation),
('复杂数据处理', complex_data_processing)
]
results = []
for name, operation in operations:
print(f"\n分析: {name}")
# 开始分析
analyzer.start_analysis()
# 执行操作
try:
operation_result = operation()
except Exception as e:
print(f"操作失败: {e}")
continue
# 停止分析并获取结果
perf_data = analyzer.stop_analysis()
print(f"执行时间: {perf_data['execution_time']:.4f}秒")
print(f"内存使用: {perf_data['memory_used']:.2f}MB")
print(f"内存峰值: {perf_data['memory_peak']:.2f}MB")
results.append({
'name': name,
'performance': perf_data,
'detailed_stats': analyzer.get_detailed_stats()
})
# 生成性能对比报告
print("\n=== 性能对比报告 ===")
results.sort(key=lambda x: x['performance']['execution_time'])
print("执行时间排名:")
for i, result in enumerate(results, 1):
perf = result['performance']
print(f"{i}. {result['name']}: {perf['execution_time']:.4f}秒")
print("\n内存使用排名:")
results.sort(key=lambda x: x['performance']['memory_peak'])
for i, result in enumerate(results, 1):
perf = result['performance']
print(f"{i}. {result['name']}: {perf['memory_peak']:.2f}MB")
# 显示最耗时的函数详情
print("\n=== 最耗时的函数 ===")
slowest = max(results, key=lambda x: x['performance']['execution_time'])
print(f"最慢操作: {slowest['name']}")
print("详细分析:")
print(slowest['detailed_stats'][:2000] + "...") # 显示前2000字符
def performance_optimization_suggestions():
"""性能优化建议"""
print("\n=== 性能优化建议 ===")
suggestions = [
"1. 使用内置函数和标准库：它们通常经过优化，比自定义实现更快",
"2. 选择合适的数据结构：列表、字典、集合等各有优势",
"3. 避免不必要的循环和嵌套：考虑使用列表推导或生成器",
"4. 内存管理：及时释放不需要的大对象，使用生成器处理大数据",
"5. 缓存结果：对于重复计算，考虑使用缓存或记忆化",
"6. 并行处理：对于CPU密集型任务，考虑多进程处理",
"7. 算法优化：选择时间复杂度更优的算法",
"8. 减少I/O操作：批量处理，减少频繁的文件或网络访问"
]
for suggestion in suggestions:
print(suggestion)
if __name__ == '__main__':
comprehensive_performance_analysis()
performance_optimization_suggestions()
---