自动化必知的 Python 核心语法

第2章：自动化必知的 Python 核心语法

这一章不是 Python 入门教程，而是专门为自动化场景筛选出的核心工具集。如果说第1章是搭脚手架，这章就是备工具箱——六个主题，每个都是后续章节反复用到的基础。有经验的读者可以快速浏览，把重点放在并发和类型提示这两节。

路径处理：pathlib vs os.path

在 Python 3.4 之前，路径操作依赖 os.path：字符串拼接、平台相关的分隔符、可读性差。Python 3.4 引入了 pathlib，用面向对象的方式处理路径，代码更简洁，跨平台更可靠。

从今天起，忘掉 os.path，用 pathlib.Path。

旧方式 os.path

import os
base = "/data/projects"
sub = os.path.join(base, "2024", "report.csv")
name = os.path.basename(sub)
ext = os.path.splitext(name)[1]
exists = os.path.exists(sub)

新方式 pathlib

from pathlib import Path
base = Path("/data/projects")
sub = base / "2024" / "report.csv"
name = sub.name       # "report.csv"
ext = sub.suffix      # ".csv"
exists = sub.exists()

Python 3.11+

from pathlib import Path

# --- 基本属性 ---
p = Path("/data/projects/2024/report.csv")

print(p.name)       # "report.csv" — 文件名（含扩展名）
print(p.stem)       # "report"    — 文件名（不含扩展名）
print(p.suffix)     # ".csv"      — 扩展名
print(p.parent)     # /data/projects/2024 — 父目录
print(p.parts)      # ('/', 'data', 'projects', '2024', 'report.csv')

# --- 路径操作 ---
# / 运算符拼接路径（自动处理分隔符，跨平台）
output_dir = Path("data") / "output" / "2024"
output_dir.mkdir(parents=True, exist_ok=True)  # 递归创建目录

# 修改文件名或扩展名
new_path = p.with_stem("summary")    # /data/.../summary.csv
new_ext  = p.with_suffix(".xlsx")   # /data/.../report.xlsx

# --- 遍历目录 ---
src = Path("src")

# 直接子项（非递归）
for item in src.iterdir():
    print(item, "dir" if item.is_dir() else "file")

# glob 匹配（单层）：找所有 .py 文件
for py_file in src.glob("*.py"):
    print(py_file)

# rglob 递归匹配：找所有子目录下的 .csv 文件
data_dir = Path("data")
csv_files = list(data_dir.rglob("*.csv"))
print(f"找到 {len(csv_files)} 个 CSV 文件")

# --- 读写快捷方式（小文件） ---
config = Path("config.txt")
config.write_text("debug=true\n", encoding="utf-8")  # 写入
text = config.read_text(encoding="utf-8")            # 读取
raw = config.read_bytes()                             # 以字节读取

# --- 实用方法 ---
home = Path.home()         # 当前用户主目录（跨平台）
cwd = Path.cwd()           # 当前工作目录
abs_path = p.resolve()     # 转为绝对路径（解析符号链接）
rel = p.relative_to("/data")  # 相对路径：projects/2024/report.csv

**Windows 路径注意：**pathlib 在 Windows 上自动使用 \ 分隔符，在 macOS/Linux 使用 /。用 / 运算符拼接路径，永远不需要手写分隔符——这就是 pathlib 跨平台的核心价值。

文件读写：编码与上下文管理器

with 语句：为什么是强制要求

打开文件后必须关闭，否则文件句柄泄漏可能导致数据未写入磁盘、文件被锁定。with 语句保证无论发生什么（包括异常），文件都会被正确关闭：

Python 3.11+

from pathlib import Path

# --- 文本文件读写 ---

# 写入（覆盖模式）
with open("report.txt", "w", encoding="utf-8") as f:
    f.write("第一行\n")
    f.write("第二行\n")

# 追加模式（不覆盖，在末尾追加）
with open("report.txt", "a", encoding="utf-8") as f:
    f.write("追加的一行\n")

# 读取全部内容
with open("report.txt", "r", encoding="utf-8") as f:
    content = f.read()  # 返回字符串

# 逐行读取（大文件推荐，不会一次性加载到内存）
with open("large_log.txt", "r", encoding="utf-8") as f:
    for line in f:  # f 是可迭代对象，每次产出一行（含 \n）
        line = line.rstrip("\n")  # 去掉行尾换行符
        if "ERROR" in line:
            print(line)

# 读取所有行到列表
with open("report.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()  # ['第一行\n', '第二行\n', ...]

# --- 编码问题处理 ---

# 处理来源不明的文件，先尝试 UTF-8，失败再尝试 GBK
def read_unknown_encoding(filepath: Path) -> str:
    """尝试多种编码读取文件，返回文本内容。"""
    for encoding in ["utf-8", "gbk", "utf-8-sig", "latin-1"]:
        try:
            return filepath.read_text(encoding=encoding)
        except UnicodeDecodeError:
            continue
    # 最后兜底：忽略无法解码的字节
    return filepath.read_text(encoding="utf-8", errors="ignore")

# --- CSV 文件（推荐用标准库 csv 模块，不要手动 split） ---
import csv

# 写入 CSV
data = [
    ["姓名", "部门", "薪资"],
    ["张三", "技术部", 15000],
    ["李四", "市场部", 12000],
]
with open("employees.csv", "w", newline="", encoding="utf-8-sig") as f:
    # utf-8-sig 在 Windows Excel 中打开不会乱码
    writer = csv.writer(f)
    writer.writerows(data)

# 读取 CSV 为字典列表
with open("employees.csv", "r", encoding="utf-8-sig") as f:
    reader = csv.DictReader(f)
    employees = [row for row in reader]
    # [{'姓名': '张三', '部门': '技术部', '薪资': '15000'}, ...]

**编码陷阱：**Windows 中文系统导出的文件通常是 GBK 编码，直接用 UTF-8 读取会报 UnicodeDecodeError。写入给 Windows Excel 用的 CSV 文件时，用 utf-8-sig（带 BOM 的 UTF-8）才能避免中文乱码。

正则表达式：re 模块

正则表达式在自动化中极为常用：从日志里提取错误信息、验证数据格式、批量替换文本内容。

Python 3.11+

import re

# --- 编译 pattern（推荐：重复使用时性能更好）---
# re.compile 在循环外编译一次，比每次 re.search 重新编译快很多
EMAIL_RE = re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}")
DATE_RE  = re.compile(r"\d{4}-\d{2}-\d{2}")
URL_RE   = re.compile(r"https?://[^\s<>\"']+")
PHONE_RE = re.compile(r"1[3-9]\d{9}")  # 中国大陆手机号

# --- 核心方法 ---

text = "联系我: [email protected], 电话: 13812345678, 日期: 2024-01-15"

# search: 找第一个匹配（返回 Match 对象或 None）
match = EMAIL_RE.search(text)
if match:
    print(match.group())   # "[email protected]"

# findall: 返回字符串列表
emails = EMAIL_RE.findall(text)    # ['[email protected]']
dates  = DATE_RE.findall(text)     # ['2024-01-15']

# finditer: 返回 Match 对象迭代器（大文本推荐，节省内存）
for m in EMAIL_RE.finditer(text):
    print(f"邮箱: {m.group()} 位置: {m.start()}-{m.end()}")

# sub: 替换匹配内容（支持 lambda 函数）
anonymized = PHONE_RE.sub(lambda m: m.group()[:3] + "****" + m.group()[-4:], text)

# --- 命名分组捕获（提取结构化日志数据）---
LOG_RE = re.compile(
    r"(?P<date>\d{4}-\d{2}-\d{2})\s+(?P<time>\d{2}:\d{2}:\d{2})\s+"
    r"(?P<level>\w+)\s+\[(?P<module>\w+)\]\s+(?P<message>.+)"
)
m = LOG_RE.match("2024-01-15 14:32:01 ERROR [database] Connection timeout after 30s")
if m:
    print(m.group("level"))   # "ERROR"
    print(m.groupdict())      # 所有命名分组的字典

数据结构进阶：推导式、collections 与 dataclasses

Python 3.11+

from collections import defaultdict, Counter
from dataclasses import dataclass, field

# --- 列表/字典推导式 ---
# 比 for 循环更简洁，通常也更快（CPython 优化）
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

evens = [n for n in numbers if n % 2 == 0]        # [2, 4, 6, 8, 10]
squares = {n: n**2 for n in numbers}               # {1:1, 2:4, ...}
unique_ext = {f.suffix for f in Path(".").glob("*")}  # 扩展名集合（去重）

# 嵌套推导式：展平二维列表
matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
flat = [x for row in matrix for x in row]  # [1,2,3,4,5,6,7,8,9]

# --- defaultdict：避免 KeyError 的字典 ---
# 场景：按部门分组员工，普通 dict 需要先判断 key 是否存在
employees = [
    {"name": "张三", "dept": "技术部"},
    {"name": "李四", "dept": "市场部"},
    {"name": "王五", "dept": "技术部"},
]

# 普通字典（繁琐）
by_dept = {}
for e in employees:
    dept = e["dept"]
    if dept not in by_dept:
        by_dept[dept] = []
    by_dept[dept].append(e["name"])

# defaultdict（简洁）
by_dept = defaultdict(list)
for e in employees:
    by_dept[e["dept"]].append(e["name"])
# {"技术部": ["张三", "王五"], "市场部": ["李四"]}

# --- Counter：元素计数 ---
words = ["apple", "banana", "apple", "cherry", "banana", "apple"]
count = Counter(words)
print(count.most_common(2))  # [('apple', 3), ('banana', 2)]

# 统计文件扩展名分布
files = list(Path(".").rglob("*.*"))
ext_count = Counter(f.suffix for f in files)
print(ext_count.most_common(5))  # 最多的5种扩展名

# --- dataclasses：结构化数据容器 ---
@dataclass
class FileRecord:
    """表示一个需要处理的文件记录。"""
    path: Path
    size: int                     # 字节数
    processed: bool = False       # 默认值
    tags: list[str] = field(default_factory=list)  # 可变默认值必须用 field

    # dataclass 自动生成 __init__, __repr__, __eq__

    @property
    def size_kb(self) -> float:
        """文件大小（KB），保留2位小数。"""
        return round(self.size / 1024, 2)

    def mark_done(self) -> None:
        self.processed = True

# 使用
record = FileRecord(path=Path("report.csv"), size=204800)
print(record)              # FileRecord(path=..., size=204800, processed=False, tags=[])
print(record.size_kb)      # 200.0
record.tags.append("finance")
record.mark_done()
print(record.processed)    # True

并发基础：选哪种模型？

Python 并发有三种主要模型，选错了性能不升反降：

自动化脚本最常用的是 ThreadPoolExecutor——处理批量文件、API 调用等 I/O 场景，代码最简洁：

Python 3.11+

from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
import time

def process_file(filepath: Path) -> dict:
    """
    模拟处理单个文件（实际场景可能是：读取 CSV、调用 API、转换格式等）。
    返回处理结果字典。
    """
    start = time.time()
    # --- 模拟耗时操作（如网络请求、文件解析）---
    content = filepath.read_text(encoding="utf-8")
    line_count = len(content.splitlines())
    time.sleep(0.1)  # 模拟 I/O 等待
    elapsed = time.time() - start

    return {
        "file": filepath.name,
        "lines": line_count,
        "elapsed": round(elapsed, 3),
    }

def batch_process(data_dir: Path, max_workers: int = 8) -> list[dict]:
    """
    并发处理目录下的所有 .txt 文件。

    Args:
        data_dir: 数据目录
        max_workers: 最大并发线程数（I/O场景通常设为 CPU数*2~4）

    Returns:
        所有文件的处理结果列表
    """
    files = list(data_dir.glob("*.txt"))
    results = []
    errors = []

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任务，得到 Future 对象字典
        future_to_file = {
            executor.submit(process_file, f): f
            for f in files
        }

        # as_completed 按完成顺序迭代（先完成先处理）
        for future in as_completed(future_to_file):
            filepath = future_to_file[future]
            try:
                result = future.result()
                results.append(result)
                print(f"完成: {filepath.name} ({result['lines']} 行)")
            except Exception as exc:
                errors.append({"file": filepath.name, "error": str(exc)})
                print(f"失败: {filepath.name} — {exc}")

    print(f"\n汇总: 成功 {len(results)}, 失败 {len(errors)}")
    return results

# 使用示例
# results = batch_process(Path("data/input"), max_workers=8)

**max_workers 怎么设？**对于网络 I/O（API 调用、爬虫），通常设 10-50；对于本地文件 I/O，设 4-8 即可。不是越大越好——线程过多会导致上下文切换开销反而拖慢速度。

类型提示：为什么自动化脚本也需要它

很多人认为类型提示是"大型项目的事"。错。自动化脚本往往在无人监控的情况下运行，一个参数类型错误可能导致整个批量任务静默失败。类型提示让 IDE 在运行前就发现这类问题。

Python 3.11+

from pathlib import Path
from typing import Sequence

# --- Python 3.10+ 简化语法 ---
# 旧语法（3.9及以下）：from typing import Optional, Union, List, Dict
# 新语法（3.10+，本书采用）：直接用 | 和内置类型

def read_config(
    config_path: Path,
    fallback: str | None = None,  # 旧写法: Optional[str] = None
) -> dict[str, str]:              # 旧写法: Dict[str, str]
    """读取配置文件，返回键值字典。若文件不存在返回空字典。"""
    if not config_path.exists():
        if fallback:
            return {"_fallback": fallback}
        return {}
    # 实际解析逻辑...
    return {}

def batch_rename(
    files: Sequence[Path],          # Sequence 接受 list, tuple 等任何序列
    prefix: str = "",
    suffix: str = "",
    dry_run: bool = False,          # 布尔标志：True=仅打印不实际操作
) -> list[tuple[Path, Path]]:      # 返回 (原路径, 新路径) 的列表
    """
    批量重命名文件。

    Args:
        files: 要重命名的文件路径列表
        prefix: 添加到文件名开头的前缀
        suffix: 添加到文件名结尾（扩展名之前）的后缀
        dry_run: 若为 True，仅打印计划，不执行实际重命名

    Returns:
        (原路径, 新路径) 元组的列表
    """
    plan: list[tuple[Path, Path]] = []

    for old_path in files:
        new_name = f"{prefix}{old_path.stem}{suffix}{old_path.suffix}"
        new_path = old_path.parent / new_name
        plan.append((old_path, new_path))

        if dry_run:
            print(f"[DRY RUN] {old_path.name} -> {new_name}")
        else:
            old_path.rename(new_path)
            print(f"已重命名: {old_path.name} -> {new_name}")

    return plan

# 使用示例
files = list(Path("data").glob("*.csv"))
plan = batch_rename(files, prefix="2024_", dry_run=True)  # 先演习

**类型检查工具：**写完代码后，在终端运行 mypy your_script.py（需要 pip install mypy），它会在不运行代码的情况下发现类型错误。VS Code + Pylance 插件也会实时在编辑器内高亮类型问题。这是自动化脚本上线前的必要检查步骤。

上一章

下一章
第3章：AI 辅助编程方法论