—— 数据表格的瑞士军刀
import csv
# 自定义分隔符和引号
with open("data.tsv", "r") as f:
reader = csv.DictReader(f, delimiter='\t')
for row in reader:
print(row)
# 写入时处理特殊字符
data = [
{"desc": "包含,逗号", "price": "99.9"},
{"desc": '包含"引号', "price": "199.9"},
]
with open("output.csv", "w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=["desc", "price"],
quoting=csv.QUOTE_ALL)
writer.writeheader()
writer.writerows(data)
# sniff 自动检测格式
with open("unknown.csv", "r") as f:
sample = f.read(2048)
dialect = csv.Sniffer().sniff(sample)
f.seek(0)
reader = csv.DictReader(f, dialect=dialect)import csv
from pathlib import Path
def process_large_csv(filepath, filter_func=None, chunk_size=10000):
"""
流式处理大CSV文件
- 逐行读取，不一次性加载到内存
- 支持过滤函数
- 每 chunk_size 行处理一次
"""
total = 0
matched = 0
chunk = []
with open(filepath, "r", encoding="utf-8-sig") as f:
reader = csv.DictReader(f)
for row in reader:
total += 1
if filter_func and not filter_func(row):
continue
matched += 1
chunk.append(row)
if len(chunk) >= chunk_size:
# 处理一批数据
yield chunk
chunk = []
if chunk:
yield chunk
print(f"总计 {total} 行, 匹配 {matched} 行")
# 使用示例
def high_value(row):
return float(row.get("amount", 0)) > 1000
# for chunk in process_large_csv("big_data.csv", filter_func=high_value):
# process_chunk(chunk)import csv
from collections import defaultdict
def filter_and_aggregate(filepath):
"""过滤+聚合：按部门统计薪资"""
# 按部门聚合
dept_stats = defaultdict(lambda: {"count": 0, "total_salary": 0.0})
with open(filepath, "r", encoding="utf-8-sig") as f:
reader = csv.DictReader(f)
for row in reader:
# 过滤：只看在职员工
if row["status"] != "在职":
continue
# 过滤：薪资 > 5000
salary = float(row["salary"])
if salary <= 5000:
continue
# 聚合
dept = row["department"]
dept_stats[dept]["count"] += 1
dept_stats[dept]["total_salary"] += salary
# 计算平均薪资
result = {}
for dept, stats in dept_stats.items():
avg = stats["total_salary"] / stats["count"]
result[dept] = {
"人数": stats["count"],
"总薪资": round(stats["total_salary"], 2),
"平均薪资": round(avg, 2),
}
return resultimport csv
from collections import defaultdict
def pivot_table(filepath, row_field, col_field, value_field, agg="sum"):
"""
简易数据透视表
row_field: 行维度
col_field: 列维度
value_field: 值字段
agg: 聚合方式 sum/count/avg
"""
# 收集数据
data = defaultdict(list)
col_values = set()
with open(filepath, "r", encoding="utf-8-sig") as f:
reader = csv.DictReader(f)
for row in reader:
r_key = row[row_field]
c_key = row[col_field]
val = float(row[value_field])
data[(r_key, c_key)].append(val)
col_values.add(c_key)
# 聚合
col_values = sorted(col_values)
pivot = {}
for (r_key, c_key), values in data.items():
if r_key not in pivot:
pivot[r_key] = {}
if agg == "sum":
pivot[r_key][c_key] = round(sum(values), 2)
elif agg == "count":
pivot[r_key][c_key] = len(values)
elif agg == "avg":
pivot[r_key][c_key] = round(sum(values) / len(values), 2)
# 输出为 CSV
output = []
# 表头
header = [row_field] + col_values + ["合计"]
output.append(header)
# 数据行
for r_key in sorted(pivot.keys()):
row_data = [r_key]
row_total = 0
for c_key in col_values:
val = pivot[r_key].get(c_key, 0)
row_data.append(val)
row_total += val
row_data.append(round(row_total, 2))
output.append(row_data)
return output
# 使用示例：按部门×月份透视销售额
# result = pivot_table("sales.csv", "department", "month", "amount", agg="sum")#!/usr/bin/env python3
"""第09课 CSV处理验证"""
import csv
import tempfile
import os
from collections import defaultdict
def test_csv_readwrite():
"""CSV基本读写"""
with tempfile.NamedTemporaryFile(suffix=".csv", delete=False,
newline="", encoding="utf-8-sig", mode="w") as f:
path = f.name
w = csv.DictWriter(f, fieldnames=["name", "age"])
w.writeheader()
w.writerow({"name": "张三", "age": "28"})
with open(path, "r", encoding="utf-8-sig") as f:
rows = list(csv.DictReader(f))
assert rows[0]["name"] == "张三"
os.unlink(path)
print("✅ CSV基本读写测试通过")
def test_csv_filter():
"""CSV过滤测试"""
data = [
["name", "dept", "salary"],
["张三", "技术", "15000"],
["李四", "市场", "8000"],
["王五", "技术", "20000"],
]
filtered = [r for r in data[1:] if int(r[2]) > 10000]
assert len(filtered) == 2
assert all(int(r[2]) > 10000 for r in filtered)
print("✅ CSV过滤测试通过")
def test_csv_aggregate():
"""CSV聚合测试"""
rows = [
{"dept": "技术", "salary": "15000"},
{"dept": "技术", "salary": "20000"},
{"dept": "市场", "salary": "8000"},
]
stats = defaultdict(lambda: {"count": 0, "total": 0})
for r in rows:
d = r["dept"]
stats[d]["count"] += 1
stats[d]["total"] += int(r["salary"])
assert stats["技术"]["total"] == 35000
assert stats["技术"]["count"] == 2
print("✅ CSV聚合测试通过")
def test_pivot():
"""透视表测试"""
rows = [
{"dept": "技术", "month": "1月", "amount": "100"},
{"dept": "技术", "month": "2月", "amount": "200"},
{"dept": "市场", "month": "1月", "amount": "300"},
]
# 手动透视
pivot = defaultdict(dict)
for r in rows:
pivot[r["dept"]][r["month"]] = int(r["amount"])
assert pivot["技术"]["1月"] == 100
assert pivot["市场"]["1月"] == 300
print("✅ 透视表测试通过")
def test_large_csv():
"""大CSV流式处理测试"""
with tempfile.NamedTemporaryFile(suffix=".csv", delete=False,
newline="", encoding="utf-8", mode="w") as f:
path = f.name
w = csv.writer(f)
w.writerow(["id", "value"])
for i in range(1000):
w.writerow([i, i * 10])
# 流式读取
total = 0
count = 0
with open(path, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
if int(row["value"]) > 5000:
total += int(row["value"])
count += 1
assert count > 0
os.unlink(path)
print("✅ 大CSV流式处理测试通过")
if __name__ == "__main__":
test_csv_readwrite()
test_csv_filter()
test_csv_aggregate()
test_pivot()
test_large_csv()
print("\n🎉 第09课全部验证通过！")