Python 文件 IO 与数据序列化：从 CSV 到 Parquet 的完整指南

from pathlib import Path
import shutil

# 创建路径对象
p = Path("/Users/alice/data/report.txt")
# 或跨平台写法
p = Path.home() / "data" / "report.txt"

# 路径分解
print(p.name)         # report.txt（文件名）
print(p.stem)         # report（无扩展名）
print(p.suffix)       # .txt（扩展名）
print(p.parent)       # /Users/alice/data（父目录）
print(p.parts)        # ('/', 'Users', 'alice', 'data', 'report.txt')
print(p.exists())     # True / False

# 遍历目录
for item in Path(".").iterdir():
    if item.is_file():
        print(f"文件: {item.name}, 大小: {item.stat().st_size}")

# 递归查找（glob）
py_files = list(Path("src").rglob("*.py"))   # 递归找所有 .py
log_files = list(Path("logs").glob("*.log")) # 只找当前目录

# 文件操作
p = Path("hello.txt")
p.write_text("Hello, World!", encoding="utf-8")
content = p.read_text(encoding="utf-8")

# 创建目录
Path("output/2024/08").mkdir(parents=True, exist_ok=True)

# 复制/移动/删除
shutil.copy("src.txt", "dst.txt")
shutil.move("old.txt", "new.txt")
p.unlink()   # 删除文件
# p.rmdir()  # 删除空目录

from pathlib import Path

# 写
Path("data.txt").write_text("Line 1\nLine 2\n", encoding="utf-8")

# 读
content = Path("data.txt").read_text(encoding="utf-8")

# 逐行读取
for line in Path("data.txt").read_text(encoding="utf-8").splitlines():
    print(line.strip())

# 大文件：不一次性读入内存
with open("large.txt", "r", encoding="utf-8") as f:
    for line in f:          # 逐行迭代（内存友好）
        process(line)

# 模式说明
# 'r'  读取（默认）
# 'w'  写入（覆盖）
# 'a'  追加
# 'x'  独占创建（文件存在则报错）
# 'b'  二进制模式
# '+'  读写

with open("log.txt", "a", encoding="utf-8") as f:
    f.write("新日志行\n")

# 同时读写
with open("data.txt", "r+", encoding="utf-8") as f:
    content = f.read()
    f.seek(0)
    f.write("新头部\n" + content)

# 写入二进制
data = b"\x00\x01\x02\x03"
Path("binary.dat").write_bytes(data)

# 读取二进制
raw = Path("binary.dat").read_bytes()

# 处理图片/文件
with open("image.png", "rb") as f:
    header = f.read(8)
    if header.startswith(b"\x89PNG"):
        print("这是一个 PNG 文件")

import mmap

# 用内存映射读取大文件（不加载到内存）
with open("huge_file.bin", "rb") as f:
    with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
        # mm 像 bytes 对象一样使用
        print(mm[:100])          # 前 100 字节
        print(mm.find(b"target"))  # 查找（O(n)）

import csv
from pathlib import Path

# 写 CSV
rows = [
    ["name", "age", "city"],
    ["Alice", 25, "Beijing"],
    ["Bob", 30, "Shanghai"],
]

with open("users.csv", "w", newline="", encoding="utf-8-sig") as f:
    writer = csv.writer(f)
    writer.writerows(rows)

# 读 CSV
with open("users.csv", "r", encoding="utf-8-sig") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"{row['name']}: {row['age']} 岁")

# 大文件分块读取
def read_csv_in_chunks(filepath, chunksize=10000):
    import pandas as pd
    for chunk in pd.read_csv(filepath, chunksize=chunksize):
        yield chunk

# 特定编码（Excel 导出的 CSV 常用 gbk）
# pd.read_csv("file.csv", encoding="gbk")

import json
from pathlib import Path

data = {
    "name": "Alice",
    "age": 25,
    "skills": ["Python", "Go", "Rust"],
    "address": {"city": "Beijing", "zip": "100000"},
}

# 序列化（Python 对象 → JSON 字符串）
json_str = json.dumps(data, ensure_ascii=False, indent=2)
print(json_str)

# 写入文件
Path("user.json").write_text(json_str, encoding="utf-8")

# 反序列化
loaded = json.loads(json_str)
print(loaded["name"])

# 直接从文件读写
with open("user.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

with open("user.json", "r", encoding="utf-8") as f:
    data = json.load(f)

import json
from datetime import datetime
from decimal import Decimal

class CustomEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.isoformat()
        if isinstance(obj, Decimal):
            return float(obj)
        if isinstance(obj, set):
            return list(obj)
        return super().default(obj)

data = {
    "created_at": datetime.now(),
    "price": Decimal("99.99"),
    "tags": {"python", "json"},
}

json_str = json.dumps(data, cls=CustomEncoder, ensure_ascii=False, indent=2)
print(json_str)

# pip install pyyaml
import yaml

config = {
    "app": {"name": "MyApp", "debug": False},
    "database": {"host": "localhost", "port": 5432},
}

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

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

print(loaded["database"]["host"])

import pickle
from pathlib import Path

# 序列化任意 Python 对象
data = {"numbers": [1, 2, 3], "func": lambda x: x**2}

pickled = pickle.dumps(data)
print(f"Pickle 大小: {len(pickled)} bytes")

# 写入文件
Path("data.pkl").write_bytes(pickled)

# 反序列化
loaded = pickle.loads(Path("data.pkl").read_bytes())
print(loaded["numbers"])

# 注意：Pickle 不安全！不要反序列化不受信任的数据

# pip install msgpack
import msgpack

data = {"name": "Alice", "age": 25, "scores": [95, 88, 92]}

# 序列化（比 JSON 更快、更小）
packed = msgpack.packb(data)
print(f"MessagePack 大小: {len(packed)} bytes")

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

# pip install pyarrow
import pandas as pd

df = pd.DataFrame({
    "name": ["Alice", "Bob", "Charlie"],
    "age": [25, 30, 35],
    "salary": [5000.0, 6000.0, 7000.0],
})

# 写入 Parquet（列式存储，压缩率高）
df.to_parquet("data.parquet", engine="pyarrow", compression="snappy")

# 读取（只读需要的列，非常快）
df = pd.read_parquet("data.parquet", columns=["name", "age"])

from pathlib import Path

def process_large_file(filepath, process_func, chunk_size=8192):
    """分块处理大文件"""
    with open(filepath, "rb") as f:
        while chunk := f.read(chunk_size):
            process_func(chunk)

# 逐行处理超大日志
def process_logs(filepath):
    with open(filepath, "r", encoding="utf-8") as f:
        for line in f:
            yield line.strip()

# 使用
for line in process_logs("huge.log"):
    if "ERROR" in line:
        print(line)

# 自动检测编码
# pip install chardet
import chardet

raw = Path("unknown.txt").read_bytes()
result = chardet.detect(raw)
print(f"编码: {result['encoding']}, 置信度: {result['confidence']}")

# 按检测到的编码解码
text = raw.decode(result["encoding"])

# BOM 处理
# utf-8-sig 自动处理 UTF-8 BOM
Path("with_bom.txt").write_text("hello", encoding="utf-8-sig")

from io import StringIO, BytesIO

# 模拟文本文件操作（不落地磁盘）
output = StringIO()
output.write("第一行\n")
output.write("第二行\n")

# 获取完整内容
output.seek(0)
content = output.read()
print(content)

# BytesIO 处理二进制数据
bio = BytesIO()
bio.write(b"\x00\x01\x02\x03")
bio.seek(0)
print(bio.read())     # b'\x00\x01\x02\x03'

import tempfile
from pathlib import Path

# 方案 1：自动清理的临时文件
with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=True) as f:
    f.write("name,age\nAlice,25\n")
    f.flush()
    # 处理完成后自动删除

# 方案 2：临时目录（多个文件）
with tempfile.TemporaryDirectory() as tmpdir:
    raw_path = Path(tmpdir) / "raw_data.json"
    raw_path.write_text('{"key": "value"}', encoding="utf-8")
    processed_path = Path(tmpdir) / "processed.parquet"
    # 处理逻辑...
    # 退出 with 后整个目录自动清理

# 方案 3：跨平台安全路径（避免 /tmp 竞态条件）
secure_tmp = Path(tempfile.mkstemp(suffix=".dat")[1])