Python 工程实践（五）：I/O、序列化与数据格式

1
2
3
4
5
6
7
8

# 正确方式：始终使用上下文管理器（context manager）
with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()

# 不用 'with' 会发生什么：
f = open("data.txt", "r")
content = f.read()
f.close()  # 容易遗漏，尤其当上方代码抛出异常时

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# 一次性读取整个文件为字符串
with open("data.txt", encoding="utf-8") as f:
    content = f.read()

# 读取为行列表
with open("data.txt", encoding="utf-8") as f:
    lines = f.readlines()
# 每行末尾包含换行符 '\n'

# 逐行迭代（对大文件内存友好）
with open("data.txt", encoding="utf-8") as f:
    for line in f:
        process(line.rstrip("\n"))

# 读取指定字节数
with open("data.bin", "rb") as f:
    header = f.read(4)  # 前 4 字节
    rest = f.read()     # 剩余全部字节

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# 写入字符串
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("Hello, world\n")

# 写入多行
lines = ["first", "second", "third"]
with open("output.txt", "w", encoding="utf-8") as f:
    for line in lines:
        f.write(line + "\n")
# 或者：
with open("output.txt", "w", encoding="utf-8") as f:
    f.writelines(line + "\n" for line in lines)

# 追加到已有文件
with open("log.txt", "a", encoding="utf-8") as f:
    f.write(f"[{timestamp}] Event occurred\n")

# 写入二进制
with open("output.bin", "wb") as f:
    f.write(b"\x00\x01\x02\x03")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from pathlib import Path

# 创建路径
project = Path("/home/user/project")
config = project / "config" / "settings.toml"
# 结果：PosixPath('/home/user/project/config/settings.toml')

# 当前目录与用户主目录
cwd = Path.cwd()
home = Path.home()

# 路径组成部分
p = Path("/home/user/project/data/file.csv")
p.name       # 'file.csv'
p.stem       # 'file'
p.suffix     # '.csv'
p.parent     # PosixPath('/home/user/project/data')
p.parents[1] # PosixPath('/home/user/project')
p.parts      # ('/', 'home', 'user', 'project', 'data', 'file.csv')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

from pathlib import Path

p = Path("data")

# 检查存在性
p.exists()       # True/False
p.is_file()      # 若为文件返回 True
p.is_dir()       # 若为目录返回 True

# 创建目录（自动创建父目录，且不报错若已存在）
p.mkdir(parents=True, exist_ok=True)

# 列出目录内容
for child in p.iterdir():
    print(child)

# Glob 模式匹配
for csv_file in p.glob("*.csv"):
    print(csv_file)

# 递归 Glob
for py_file in p.rglob("*.py"):
    print(py_file)

# 便捷读写方法（无需显式 open）
text = p.joinpath("config.txt").read_text(encoding="utf-8")
p.joinpath("output.txt").write_text("hello\n", encoding="utf-8")
data = p.joinpath("image.png").read_bytes()
p.joinpath("copy.png").write_bytes(data)

# 文件元数据
stat = p.stat()
stat.st_size     # 文件大小（字节）
stat.st_mtime    # 修改时间（Unix 时间戳）

# 重命名与删除
p.rename("new_name")
p.unlink()          # 删除文件
p.rmdir()           # 删除空目录

1
2
3
4

# 这段代码在你的 Mac 上能运行，但在 Windows 服务器上会失败：
with open("data.txt") as f:
    content = f.read()
# UnicodeDecodeError: 'cp1252' codec can't decode byte 0xe9

1
2
3

# 务必这样做
with open("data.txt", encoding="utf-8") as f:
    content = f.read()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 跳过非法字节
with open("messy.txt", encoding="utf-8", errors="ignore") as f:
    content = f.read()

# 将非法字节替换为 '?'
with open("messy.txt", encoding="utf-8", errors="replace") as f:
    content = f.read()

# 自动检测编码（当你完全不知道源编码时）
import chardet

with open("mystery.txt", "rb") as f:
    raw = f.read()
    detected = chardet.detect(raw)
    # {'encoding': 'utf-8', 'confidence': 0.99, 'language': ''}

content = raw.decode(detected["encoding"])

1
2
3
4
5
6
7

# 读取：自动剥离 BOM（如存在）
with open("windows_file.csv", encoding="utf-8-sig") as f:
    content = f.read()

# 写入：添加 BOM（提升 Windows 兼容性）
with open("output.csv", "w", encoding="utf-8-sig") as f:
    f.write("data\n")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import json

# 解析 JSON 字符串
data = json.loads('{"name": "Alice", "age": 30}')
# data = {'name': 'Alice', 'age': 30}

# 序列化为 JSON 字符串
text = json.dumps(data)
# '{"name": "Alice", "age": 30}'

# 美化输出
text = json.dumps(data, indent=2, sort_keys=True)
# {
#   "age": 30,
#   "name": "Alice"
# }

# 从文件读取
with open("config.json", encoding="utf-8") as f:
    config = json.load(f)

# 写入文件
with open("output.json", "w", encoding="utf-8") as f:
    json.dump(data, f, indent=2, ensure_ascii=False)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

import json
from datetime import datetime
from pathlib import Path


def json_serializer(obj):
    """处理 json.dumps 无法直接序列化的类型"""
    if isinstance(obj, datetime):
        return obj.isoformat()
    if isinstance(obj, Path):
        return str(obj)
    if isinstance(obj, set):
        return sorted(obj)
    if isinstance(obj, bytes):
        return obj.decode("utf-8", errors="replace")
    raise TypeError(f"Type {type(obj)} is not JSON serializable")


data = {
    "timestamp": datetime.now(),
    "path": Path("/home/user/data"),
    "tags": {"python", "coding"},
}

text = json.dumps(data, default=json_serializer, indent=2)

1
2
3
4
5

# 美化打印 JSON 文件
$ python -m json.tool data.json

# 从管道输入
$ curl -s https://api.example.com/data | python -m json.tool

1

(.venv) $ pip install pyyaml

1
2
3
4
5
6
7
8
9

import yaml

# 读取 YAML
with open("config.yaml", encoding="utf-8") as f:
    config = yaml.safe_load(f)

# 写入 YAML
with open("output.yaml", "w", encoding="utf-8") as f:
    yaml.dump(config, f, default_flow_style=False, allow_unicode=True)

1
2
3

# 下面这段 YAML 在使用 yaml.load() 时会执行系统命令：
!!python/object/apply:os.system
  args: ["rm -rf /"]

1
2
3
4
5
6
7
8
9

# YAML 存在令人意外的类型强制转换：
norway: NO       # 解析为布尔值 False！
version: 3.10    # 解析为浮点数 3.1！
port: 8080       # 解析为整数（通常符合预期）
zip: 01onal      # 解析为字符串

# 对任何可能被误判为布尔值或数字的值，务必加引号：
norway: "NO"
version: "3.10"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# Python 3.11+
import tomllib

with open("config.toml", "rb") as f:
    config = tomllib.load(f)

# 或从字符串解析
config = tomllib.loads("""
[server]
host = "0.0.0.0"
port = 8080
debug = false

[database]
url = "postgresql://localhost/mydb"
pool_size = 5
""")

1

(.venv) $ pip install tomli

1
2
3
4

import tomli

with open("config.toml", "rb") as f:
    config = tomli.load(f)

1

(.venv) $ pip install tomli-w

1
2
3
4
5
6
7
8
9

import tomli_w

config = {
    "server": {"host": "0.0.0.0", "port": 8080},
    "database": {"url": "postgresql://localhost/mydb"},
}

with open("config.toml", "wb") as f:
    tomli_w.dump(config, f)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import csv

# 作为列表读取
with open("data.csv", encoding="utf-8") as f:
    reader = csv.reader(f)
    header = next(reader)
    for row in reader:
        print(row)  # ['Alice', '30', 'alice@example.com']

# 作为字典读取（通常更推荐）
with open("data.csv", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(row["name"], row["age"])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

import csv

# 使用 DictWriter 写入
rows = [
    {"name": "Alice", "age": 30, "email": "alice@example.com"},
    {"name": "Bob", "age": 25, "email": "bob@example.com"},
]

with open("output.csv", "w", encoding="utf-8", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=["name", "age", "email"])
    writer.writeheader()
    writer.writerows(rows)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 制表符分隔（TSV）
with open("data.tsv", encoding="utf-8") as f:
    reader = csv.reader(f, delimiter="\t")

# 分号分隔（欧洲地区常见）
with open("data.csv", encoding="utf-8") as f:
    reader = csv.reader(f, delimiter=";")

# 处理由 Excel 导出的带 BOM 的 CSV
with open("excel_export.csv", encoding="utf-8-sig") as f:
    reader = csv.DictReader(f)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import pickle

data = {"key": [1, 2, 3], "nested": {"a": "b"}}

# 序列化
with open("data.pkl", "wb") as f:
    pickle.dump(data, f)

# 反序列化
with open("data.pkl", "rb") as f:
    loaded = pickle.load(f)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

import struct

# 将数据打包为字节
packed = struct.pack(">IHB", 1024, 256, 42)
# > = 大端序，I = uint32，H = uint16，B = uint8
# 结果：b'\x00\x00\x04\x00\x01\x00\x2a'

# 将字节解包为数值
values = struct.unpack(">IHB", packed)
# (1024, 256, 42)

1

(.venv) $ pip install msgpack

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

import msgpack

data = {"name": "Alice", "scores": [95, 87, 91]}

# 序列化
packed = msgpack.packb(data)
# b'\x82\xa4name\xa5Alice\xa6scores\x93_W['

# 反序列化
unpacked = msgpack.unpackb(packed)

1

(.venv) $ pip install pyarrow pandas

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import pandas as pd

# 读取 CSV 并写入 Parquet
df = pd.read_csv("large_data.csv")
df.to_parquet("large_data.parquet", engine="pyarrow")

# 读取 Parquet
df = pd.read_parquet("large_data.parquet")

# 仅读取特定列（Parquet 可跳过未使用列）
df = pd.read_parquet("large_data.parquet", columns=["name", "age"])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from pathlib import Path

def process_large_log(path: Path) -> dict[str, int]:
    """统计日志级别——无需将整个文件加载到内存。"""
    counts: dict[str, int] = {}
    with open(path, encoding="utf-8") as f:
        for line in f:  # 每次产出一行
            level = line.split("|", 2)[1].strip() if "|" in line else "UNKNOWN"
            counts[level] = counts.get(level, 0) + 1
    return counts

1
2
3
4
5
6
7
8

def sha256_file(path: Path, chunk_size: int = 65536) -> str:
    """哈希大文件，不全量加载到内存。"""
    import hashlib
    h = hashlib.sha256()
    with open(path, "rb") as f:
        while chunk := f.read(chunk_size):
            h.update(chunk)
    return h.hexdigest()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

from typing import Iterator
import gzip
import json

def read_jsonl_gz(path: Path) -> Iterator[dict]:
    """从 gzip 压缩的 JSON Lines 文件流式读取记录。"""
    with gzip.open(path, "rt", encoding="utf-8") as f:
        for line in f:
            if line.strip():
                yield json.loads(line)

def filter_recent(records: Iterator[dict], days: int = 7) -> Iterator[dict]:
    """只保留最近 N 天的记录。"""
    from datetime import datetime, timedelta
    cutoff = datetime.now() - timedelta(days=days)
    for record in records:
        if datetime.fromisoformat(record["timestamp"]) > cutoff:
            yield record

def extract_errors(records: Iterator[dict]) -> Iterator[dict]:
    """只保留 ERROR 级别记录。"""
    for record in records:
        if record.get("level") == "ERROR":
            yield record

# 组合：在迭代之前什么都不执行
pipeline = extract_errors(filter_recent(read_jsonl_gz(Path("app.jsonl.gz"))))

# 逐条处理——无论文件多大，内存恒定
for error in pipeline:
    print(f"{error['timestamp']}: {error['message']}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import mmap
from pathlib import Path

def search_in_large_file(path: Path, pattern: bytes) -> list[int]:
    """用 mmap 在大文件中查找所有匹配位置。"""
    offsets = []
    with open(path, "rb") as f:
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
            pos = 0
            while True:
                pos = mm.find(pattern, pos)
                if pos == -1:
                    break
                offsets.append(pos)
                pos += 1
    return offsets

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// user.proto
syntax = "proto3";

message User {
  int32 id = 1;
  string name = 2;
  string email = 3;
  repeated string tags = 4;

  enum Role {
    UNKNOWN = 0;
    ADMIN = 1;
    USER = 2;
  }
  Role role = 5;
}

1
2

$ pip install grpcio-tools
$ python -m grpc_tools.protoc -I. --python_out=. --pyi_out=. user.proto

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from user_pb2 import User

# 创建
user = User(id=1, name="Alice", email="alice@example.com", role=User.ADMIN)
user.tags.append("staff")

# 序列化（字节）
data: bytes = user.SerializeToString()
print(len(data))  # ~40 字节（相比 JSON 等效的 ~120 字节）

# 反序列化
restored = User()
restored.ParseFromString(data)
assert restored.name == "Alice"

1

(.venv) $ pip install duckdb

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import duckdb

# 用 SQL 查询 CSV 文件
result = duckdb.sql("""
    SELECT department, COUNT(*) as headcount, AVG(salary) as avg_salary
    FROM 'employees.csv'
    GROUP BY department
    ORDER BY avg_salary DESC
""").fetchdf()  # 返回 pandas DataFrame

# 查询 Parquet 文件（支持通配符）
result = duckdb.sql("""
    SELECT date_trunc('month', created_at) as month, COUNT(*) as orders
    FROM 'orders/*.parquet'
    GROUP BY 1
    ORDER BY 1
""")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

import duckdb

# CSV → Parquet（带压缩）
duckdb.sql("""
    COPY (SELECT * FROM 'large_dataset.csv')
    TO 'output.parquet' (FORMAT PARQUET, COMPRESSION ZSTD)
""")

# Parquet → CSV（筛选后）
duckdb.sql("""
    COPY (SELECT * FROM 'data.parquet' WHERE year = 2024)
    TO 'filtered.csv' (FORMAT CSV, HEADER)
""")

1

(.venv) $ pip install python-dotenv

1
2
3
4
5

# .env
DATABASE_URL=postgresql://localhost/mydb
API_KEY=sk-abc123
DEBUG=true
SECRET_KEY=super-secret-key-do-not-commit

1
2
3
4
5
6
7
8

from dotenv import load_dotenv
import os

load_dotenv()  # 将 .env 加载至 os.environ

database_url = os.environ["DATABASE_URL"]
api_key = os.environ["API_KEY"]
debug = os.environ.get("DEBUG", "false").lower() == "true"

1
2
3
4
5

# .env.example
DATABASE_URL=postgresql://localhost/mydb
API_KEY=your-api-key-here
DEBUG=false
SECRET_KEY=generate-a-random-key