10｜用Python构建办公数据处理与文件自动化管线

一封凌晨的求助邮件

周五凌晨两点，运营团队的负责人发来一封紧急邮件：下周一的管理层会议需要一份跨部门数据汇总表，数据散落在 87 个 Excel 文件、若干 CSV 导出文件和一个 JSON 格式的 API 响应中。手工整理至少需要两个工作日，但留给他们的时间只有一个周末。

这正是 Python 大显身手的场景。本章将系统讲解如何用 Python 读写 Excel、CSV、JSON 三种主流办公数据格式，如何用正则表达式从非结构化文本中提取信息，以及如何编写文件批量处理脚本。最终，我们会将所有技能整合成一个可投入使用的文件自动分拣系统。

第一部分：Excel 文件的程序化操作

为什么需要用代码操作 Excel？

手动操作 Excel 在数据量小的时候没有问题，但当面对数十上百个文件、数万行数据时，手动操作既慢又容易出错。Python 的 openpyxl 库可以在不打开 Excel 软件的情况下读写 .xlsx 文件，实现全自动化。

环境准备

pip install openpyxl

读取已有工作簿

Excel 文件的层级结构为：工作簿（Workbook）> 工作表（Sheet）> 单元格（Cell）。用代码访问时遵循同样的层级：

import openpyxl

# 加载已有的 Excel 文件
workbook = openpyxl.load_workbook("quarterly_revenue.xlsx")

# 查看所有工作表名称
print(workbook.sheetnames)  # 例如 ['Q1', 'Q2', 'Q3', 'Q4']

# 选择特定工作表
q1_sheet = workbook["Q1"]

# 读取单个单元格的值——两种方式
# 方式一：Excel 坐标表示法
print(q1_sheet["C4"].value)

# 方式二：行列号（均从 1 开始）
target_cell = q1_sheet.cell(row=4, column=3)
print(target_cell.value)
print(target_cell.coordinate)  # 'C4'

遍历工作表数据

# 获取数据范围
total_rows = q1_sheet.max_row
total_cols = q1_sheet.max_column
print(f"数据范围：{total_rows} 行 x {total_cols} 列")

# 逐行读取全部数据
for r in range(1, total_rows + 1):
    row_data = []
    for c in range(1, total_cols + 1):
        row_data.append(q1_sheet.cell(row=r, column=c).value)
    print(row_data)

# 按区域读取（例如 B2 到 D6）
for row_tuple in q1_sheet["B2":"D6"]:
    for cell in row_tuple:
        print(cell.value, end="\t")
    print()

创建并写入工作簿

下面创建一个设备巡检记录表：

import openpyxl

wb = openpyxl.Workbook()
ws = wb.active
ws.title = "巡检记录"

# 写入表头
col_headers = ["设备编号", "巡检日期", "温度(C)", "压力(kPa)", "状态"]
for idx, header in enumerate(col_headers, start=1):
    ws.cell(row=1, column=idx, value=header)

# 写入若干行数据
inspection_data = [
    ["DEV-001", "2025-11-01", 72.3, 101.2, "正常"],
    ["DEV-002", "2025-11-01", 89.7, 98.5, "异常"],
    ["DEV-003", "2025-11-01", 68.1, 102.0, "正常"],
]

for row_num, record in enumerate(inspection_data, start=2):
    for col_num, val in enumerate(record, start=1):
        ws.cell(row=row_num, column=col_num, value=val)
    # 利用 Excel 内置函数标记超温设备
    ws.cell(row=row_num, column=6,
            value=f'=IF(C{row_num}>85,"超温","达标")')

wb.save("inspection_log.xlsx")
print("巡检记录已生成")

样式与格式设置

from openpyxl.styles import Font, Alignment, PatternFill

# 设置表头样式（白色字体 + 深色背景，确保文字可见）
header_font = Font(name="Arial", size=12, bold=True, color="FFFFFF")
header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
for col_idx in range(1, 7):
    header_cell = ws.cell(row=1, column=col_idx)
    header_cell.font = header_font
    header_cell.fill = header_fill

# 调整列宽
ws.column_dimensions["A"].width = 14
ws.column_dimensions["B"].width = 16

# 合并单元格用作标题行
ws.insert_rows(1)
ws.merge_cells("A1:F1")
ws["A1"] = "2025年11月设备巡检汇总"
ws["A1"].alignment = Alignment(horizontal="center")
ws["A1"].font = Font(size=16, bold=True)

wb.save("inspection_log_styled.xlsx")

第二部分：CSV 文件读写

CSV（逗号分隔值）是最轻量的表格数据格式——它是纯文本，不依赖任何软件，几乎所有编程语言和数据工具都能处理。Python 标准库自带 csv 模块，无需安装第三方包。

读取

import csv

# 方式一：按行读取为列表
with open("sensor_data.csv", "r", encoding="utf-8") as fh:
    reader = csv.reader(fh)
    for row in reader:
        print(row)  # 每行是一个字符串列表

# 方式二：按行读取为字典（以首行为键名）
with open("sensor_data.csv", "r", encoding="utf-8") as fh:
    dict_reader = csv.DictReader(fh)
    for entry in dict_reader:
        print(f"设备 {entry['device_id']}：温度 {entry['temperature']}℃")

DictReader 是推荐做法，因为 entry['temperature'] 比 row[3] 更具自文档性，且不受列顺序变化影响。

写入

import csv

# 方式一：写入列表
with open("export.csv", "w", newline="", encoding="utf-8") as fh:
    writer = csv.writer(fh)
    writer.writerow(["device_id", "temperature", "humidity"])
    writer.writerows([
        ["sensor_A", 23.5, 45],
        ["sensor_B", 26.1, 52],
        ["sensor_C", 21.8, 60],
    ])

# 方式二：写入字典
with open("export_dict.csv", "w", newline="", encoding="utf-8") as fh:
    columns = ["device_id", "temperature", "humidity"]
    writer = csv.DictWriter(fh, fieldnames=columns)
    writer.writeheader()
    writer.writerow({"device_id": "sensor_A", "temperature": 23.5, "humidity": 45})

Windows 环境注意事项： 写入 CSV 时必须指定 newline=""，否则行间会出现多余空行。如果生成的文件需要用 Excel 打开且包含中文字符，编码应设为 utf-8-sig（带 BOM 头），否则 Excel 会显示乱码。

第三部分：JSON 数据处理

JSON 是 Web 时代的通用数据交换格式。Python 的 json 标准库模块提供了四个核心函数：

函数	方向	操作目标
json.dumps()	Python -> JSON 字符串	内存中的字符串
json.loads()	JSON 字符串 -> Python	内存中的字符串
json.dump()	Python -> JSON 文件	磁盘文件
json.load()	JSON 文件 -> Python	磁盘文件

命名规律： 函数名带 s 后缀的操作字符串（string），不带的操作文件（file）。

序列化与反序列化

import json

# Python 对象 -> JSON 字符串
monitoring_snapshot = {
    "host": "prod-web-03",
    "cpu_usage": 72.5,
    "services": ["nginx", "gunicorn", "redis"],
    "healthy": True,
    "last_error": None
}

json_text = json.dumps(monitoring_snapshot, ensure_ascii=False, indent=2)
print(json_text)

# JSON 字符串 -> Python 对象
restored = json.loads(json_text)
print(restored["host"])       # prod-web-03
print(restored["services"])   # ['nginx', 'gunicorn', 'redis']

Python 与 JSON 的类型对应关系：None <-> null，True/False <-> true/false，dict <-> object，list <-> array。

文件读写

import json

# 写入 JSON 文件
cluster_info = {
    "nodes": [
        {"id": "node-1", "region": "us-east", "capacity": 128},
        {"id": "node-2", "region": "eu-west", "capacity": 64}
    ]
}
with open("cluster.json", "w", encoding="utf-8") as out:
    json.dump(cluster_info, out, ensure_ascii=False, indent=2)

# 读取 JSON 文件
with open("cluster.json", "r", encoding="utf-8") as inp:
    loaded = json.load(inp)
    for node in loaded["nodes"]:
        print(f"{node['id']} @ {node['region']}: {node['capacity']}GB")

第四部分：批量文件重命名

场景一：统一编号

产品部门提交了一批截图文件，文件名格式混乱。需要按顺序统一编号。

from pathlib import Path

target_dir = Path("screenshots")

for seq, filepath in enumerate(sorted(target_dir.iterdir()), start=1):
    if filepath.suffix.lower() in [".png", ".jpg", ".jpeg", ".gif"]:
        new_filename = f"product_v2_{seq:04d}{filepath.suffix}"
        filepath.rename(target_dir / new_filename)
        print(f"{filepath.name} -> {new_filename}")

场景二：批量替换文件名中的关键词

from pathlib import Path

project_dir = Path("deliverables")

for item in project_dir.iterdir():
    if "draft" in item.name.lower():
        updated_name = item.name.replace("draft", "final").replace("Draft", "Final")
        item.rename(project_dir / updated_name)
        print(f"Renamed: {item.name} -> {updated_name}")

安全建议： 在执行重命名前，先用 print() 打印计划变更。确认无误后再取消注释实际的 rename 调用。批量重命名是不可逆操作，谨慎为上。

第五部分：目录结构自动整理

问题：如何把一个混乱的目录按文件类型自动分拣？

import shutil
from pathlib import Path

def sort_directory(root_path):
    """根据扩展名将文件分拣到对应子目录"""
    type_mapping = {
        "docs":       [".doc", ".docx", ".pdf", ".txt", ".xlsx", ".xls",
                       ".pptx", ".ppt", ".csv", ".md"],
        "media_img":  [".jpg", ".jpeg", ".png", ".gif", ".bmp", ".svg",
                       ".webp", ".ico"],
        "media_vid":  [".mp4", ".avi", ".mkv", ".mov", ".wmv", ".flv"],
        "media_aud":  [".mp3", ".wav", ".flac", ".aac", ".ogg"],
        "archives":   [".zip", ".rar", ".7z", ".tar", ".gz"],
        "source":     [".py", ".js", ".html", ".css", ".java", ".cpp",
                       ".go", ".rs", ".sql"],
    }

    root = Path(root_path)
    counter = 0

    for item in root.iterdir():
        if not item.is_file():
            continue

        extension = item.suffix.lower()
        destination = "misc"
        for folder_name, ext_list in type_mapping.items():
            if extension in ext_list:
                destination = folder_name
                break

        dest_dir = root / destination
        dest_dir.mkdir(exist_ok=True)

        shutil.move(str(item), str(dest_dir / item.name))
        print(f"  [{destination}] {item.name}")
        counter += 1

    print(f"\nComplete. {counter} files sorted.")

递归遍历子目录

os.walk() 能深度遍历整棵目录树，每层返回当前路径、子目录列表和文件列表：

import os

for current_dir, subdirs, files in os.walk("/var/log"):
    print(f"Directory: {current_dir}")
    for fname in files:
        full_path = os.path.join(current_dir, fname)
        size_kb = os.path.getsize(full_path) / 1024
        print(f"  {fname} ({size_kb:.1f} KB)")

第六部分：正则表达式文本处理

正则表达式是一种模式描述语言，用于在文本中查找、匹配和替换符合特定结构的内容。Python 标准库的 re 模块提供了完整的正则表达式支持。

第一个正则匹配

import re

log_line = "2025-11-15 14:32:07 [WARNING] Disk usage at 92% on /dev/sda1"

# 匹配百分比数字
pct_pattern = r"\d+%"
match = re.search(pct_pattern, log_line)
if match:
    print(f"Found: {match.group()}")  # 92%

为什么使用 r 前缀？ r"..." 是原始字符串，其中的反斜杠不会被 Python 解释器转义。正则表达式大量使用 \d、\w 等元字符，原始字符串可以避免与 Python 字符串转义产生冲突。

模式速查表

字符匹配：

模式	语义	例子
\d	一个数字	\d{4} 匹配四位年份
\D	一个非数字	\D+ 匹配连续非数字
\w	字母/数字/下划线	\w+ 匹配标识符
\s	空白字符	\s+ 匹配空白区域
.	任意字符（除换行）	a.b 匹配 a_b、a1b 等
[A-Fa-f0-9]	十六进制字符	匹配 hex 值
[^a-z]	非小写字母	匹配大写、数字、符号

量词：

模式	语义
*	零次或多次
+	一次或多次
?	零次或一次
{n}	恰好 n 次
{n,m}	n 到 m 次

锚点：

模式	语义
^	字符串开头
$	字符串结尾

核心方法

import re

server_log = "Error on node-03 at 10.0.1.15, fallback to 10.0.1.16"

# search：第一个匹配
ip_match = re.search(r"\d+\.\d+\.\d+\.\d+", server_log)
print(ip_match.group())  # 10.0.1.15

# findall：所有匹配
all_ips = re.findall(r"\d+\.\d+\.\d+\.\d+", server_log)
print(all_ips)  # ['10.0.1.15', '10.0.1.16']

# sub：替换
sanitized = re.sub(r"\d+\.\d+\.\d+\.\d+", "[REDACTED]", server_log)
print(sanitized)  # Error on node-03 at [REDACTED], fallback to [REDACTED]

# 分组提取
ts_pattern = r"(\d{4})-(\d{2})-(\d{2})\s(\d{2}:\d{2}:\d{2})"
ts_text = "Event at 2025-11-15 14:32:07"
ts_match = re.search(ts_pattern, ts_text)
if ts_match:
    print(ts_match.group(0))  # 2025-11-15 14:32:07
    print(ts_match.group(1))  # 2025
    print(ts_match.group(2))  # 11
    print(ts_match.group(3))  # 15
    print(ts_match.group(4))  # 14:32:07

预编译正则

当同一个模式需要反复使用时，预编译为正则对象可以显著提升性能：

import re

ipv4_regex = re.compile(r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b")

log_samples = [
    "Connection from 192.168.0.101 established",
    "Timeout connecting to 10.20.30.40",
    "No IP in this line",
    "Resolved 8.8.8.8 and 8.8.4.4",
]

for line in log_samples:
    found = ipv4_regex.findall(line)
    if found:
        print(f"IPs: {found}")
    else:
        print("No IP found")

第七部分：综合实战——文件自动分拣系统

下面将前六部分的全部技能融合到一个完整的自动化项目中。这个系统能自动扫描指定目录，按文件类型分类，处理重名冲突，标记超大文件，并生成 CSV 报告和 JSON 统计摘要。

"""
文件自动分拣系统
================
功能清单：
  1. 按扩展名将文件移入对应类别子目录
  2. 超过 100MB 的文件放入类别下的 oversized 子目录
  3. 目标路径已存在同名文件时自动追加序号
  4. 生成 CSV 格式的操作日志
  5. 生成 JSON 格式的分类统计
"""
import os
import shutil
import csv
import json
from pathlib import Path
from datetime import datetime


def run_file_sorter(scan_path):
    """扫描并分拣目录中的所有文件"""
    root = Path(scan_path)
    if not root.exists():
        print(f"Path not found: {scan_path}")
        return

    # 分类规则
    classification = {
        "documents":  [".doc", ".docx", ".pdf", ".txt", ".xlsx", ".xls",
                       ".pptx", ".ppt", ".csv", ".md", ".rtf"],
        "images":     [".jpg", ".jpeg", ".png", ".gif", ".bmp", ".svg",
                       ".webp", ".ico", ".psd", ".tiff"],
        "videos":     [".mp4", ".avi", ".mkv", ".mov", ".wmv", ".flv",
                       ".webm", ".m4v"],
        "audio":      [".mp3", ".wav", ".flac", ".aac", ".ogg", ".m4a",
                       ".wma"],
        "archives":   [".zip", ".rar", ".7z", ".tar", ".gz", ".dmg",
                       ".iso"],
        "installers": [".exe", ".msi", ".pkg", ".deb", ".rpm", ".apk"],
        "source":     [".py", ".js", ".html", ".css", ".java", ".cpp",
                       ".go", ".rs", ".sql", ".json", ".xml", ".yaml"],
    }

    operation_log = []
    category_counts = {}

    print(f"Scanning: {root}")
    print("=" * 55)

    for item in root.iterdir():
        # 跳过目录和隐藏文件
        if not item.is_file() or item.name.startswith("."):
            continue

        ext = item.suffix.lower()
        file_bytes = item.stat().st_size
        file_mb = file_bytes / (1024 * 1024)

        # 确定类别
        category = "uncategorized"
        for cat_name, ext_list in classification.items():
            if ext in ext_list:
                category = cat_name
                break

        # 大文件标记
        if file_mb > 100:
            category = f"{category}/oversized"

        # 创建目标目录
        dest_dir = root / category
        dest_dir.mkdir(parents=True, exist_ok=True)

        # 处理重名冲突
        dest_file = dest_dir / item.name
        if dest_file.exists():
            base_name = item.stem
            suffix = item.suffix
            seq = 1
            while dest_file.exists():
                dest_file = dest_dir / f"{base_name}_{seq}{suffix}"
                seq += 1

        # 移动文件
        shutil.move(str(item), str(dest_file))

        # 记录日志
        operation_log.append({
            "filename": item.name,
            "category": category,
            "size_mb": f"{file_mb:.2f}",
            "extension": ext,
        })
        category_counts[category] = category_counts.get(category, 0) + 1
        print(f"  [{category}] {item.name} ({file_mb:.1f} MB)")

    # 写入 CSV 报告
    if operation_log:
        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
        csv_path = root / f"sort_report_{ts}.csv"
        with open(csv_path, "w", newline="", encoding="utf-8-sig") as fh:
            writer = csv.DictWriter(fh,
                fieldnames=["filename", "category", "size_mb", "extension"])
            writer.writeheader()
            writer.writerows(operation_log)

        # 写入 JSON 统计
        stats = {
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "total_files": len(operation_log),
            "breakdown": category_counts,
        }
        json_path = root / "sort_stats.json"
        with open(json_path, "w", encoding="utf-8") as fh:
            json.dump(stats, fh, ensure_ascii=False, indent=2)

        print("=" * 55)
        print(f"Done. {len(operation_log)} files processed.")
        for cat, cnt in sorted(category_counts.items()):
            print(f"  {cat}: {cnt}")
        print(f"Report: {csv_path}")
    else:
        print("No files to process.")


# 使用示例（取消注释后执行）
# run_file_sorter("/path/to/your/messy/directory")

预期运行输出

Scanning: /home/user/incoming
=======================================================
  [documents] quarterly_report.pdf (2.8 MB)
  [documents] meeting_notes.docx (0.1 MB)
  [images] architecture_diagram.png (0.4 MB)
  [videos/oversized] training_session.mp4 (312.5 MB)
  [archives] backup_2025.zip (18.7 MB)
  [source] pipeline.py (0.0 MB)
  [installers] ide_setup.dmg (105.2 MB)
  [uncategorized] debug.log (0.3 MB)
=======================================================
Done. 8 files processed.
  archives: 1
  documents: 2
  images: 1
  installers: 1
  source: 1
  uncategorized: 1
  videos/oversized: 1
Report: /home/user/incoming/sort_report_20251115_143052.csv

本章回顾

本章覆盖了自动化办公的五大核心能力：

能力	工具	要点
电子表格读写	openpyxl	load_workbook 读、Workbook 写，支持公式与样式
CSV 处理	csv 标准模块	DictReader/DictWriter 是最佳实践
JSON 处理	json 标准模块	带 s 操作字符串，不带 s 操作文件
文件操作	pathlib + shutil	Path 对象提供面向对象的文件系统接口
文本模式匹配	re 正则模块	findall 全量匹配、sub 替换、compile 预编译

自动化的核心价值不在于省下几分钟时间，而在于消除人为错误和建立可复用的流程。一个写好的脚本可以被团队里的每个人调用，可以被定时任务自动触发，可以在数据量增长 10 倍时依然可靠运行。当你开始用”这件事能否自动化”的视角审视日常工作时，你会发现到处都有优化空间。

延伸：自动化脚本的工程化建议

当你的自动化脚本从”自己用”变成”团队用”时，有几条工程实践值得注意：

添加命令行参数支持。 硬编码路径和参数会让脚本失去通用性。Python 标准库的 argparse 模块可以轻松实现命令行参数解析：

import argparse

parser = argparse.ArgumentParser(description="File sorting utility")
parser.add_argument("directory", help="Path to the directory to organize")
parser.add_argument("--dry-run", action="store_true",
                    help="Preview changes without actually moving files")
parsed = parser.parse_args()

if parsed.dry_run:
    print(f"[DRY RUN] Would organize: {parsed.directory}")
else:
    run_file_sorter(parsed.directory)

日志替代 print。 print() 适合调试，但生产脚本应使用 logging 模块。它支持日志级别（DEBUG/INFO/WARNING/ERROR）、输出到文件、格式定制等功能：

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("automation.log"),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)
logger.info("Starting file organization")
logger.warning("Large file detected: %s (%.1f MB)", filename, size_mb)

异常处理要具体。 不要使用空的 except 捕获所有异常。明确捕获你预期的异常类型（如 FileNotFoundError、PermissionError），对意料之外的异常让它自然抛出，以便快速定位问题。

幂等性设计。 一个好的自动化脚本应该能安全地重复执行。比如文件分拣脚本应该跳过已经在目标目录中的文件，而不是报错或产生重复文件。

将这些工程化思维融入你的自动化实践，你写出的不再是”一次性脚本”，而是可以长期运行、被团队信赖的工具。下一章，我们将学习如何从互联网上自动采集结构化数据。