第8章：Python字符串、正则与文件 I/O——对账文件日清

1. 项目背景

业务场景

食光集市每天早上 8 点，财务部门需要完成前一天的日对账——将微信支付、支付宝、银联三家渠道返回的对账文件（CSV/JSON 混合，UTF-8 编码但偶尔混入 GBK 字符，部分文件带 BOM 头）与内部订单系统进行比对，找出金额不一致、字段缺失、多出或缺少的订单记录。

过去三个月来，这套流程靠的是"王姐 + Excel"——王姐手动下载三个文件，用 Excel 的 VLOOKUP 跑比对，发现有问题的行标黄色，然后截图发到群。随着日均订单从 500 涨到 8000 单，Excel 已经跑不动了——打开文件要等 20 秒，VLOOKUP 跨表查询直接死机，更致命的是 3 月 7 日王姐请了病假，对账任务没人接手，导致一笔 1400 元的差异到月底审计才被发现。

技术团队决定把这个流程自动化。需求看似简单——读文件、比对字段、输出差异报告。但真正动手做才发现坑多到爆炸：

• 文件编码不统一，有 UTF-8、UTF-8-BOM、GBK 三种
• 各家的金额格式不一致：微信用 19.90，支付宝用 1990（以分为单位），银联用 19.90 CNY
• 三家订单号的命名规则不同，需要正则提取
• 某家银行给的对账文件每行末尾有 3 个 tab，还不换行分隔

字符串、正则与文件 I/O——这三项是后端开发最日常也最容易踩坑的技能。

痛点

字符串和文件处理不当，会导致：

2. 项目设计

场景：王姐请假当天，CTO 要求"今天必须上线自动对账"。小胖自告奋勇，用半天写了第一版脚本，结果跑 3 个文件报 4 个错。大师把两人叫到会议室。

小胖（苦恼）：“大师，我就想读个 CSV，用 open('duizhang.csv') 打开，Python 就报编码错误。我试了 UTF-8 不行，加了 encoding='gbk' 又报另一个错——我到底怎么知道文件用什么编码？”

小白：“这是一个很实际的问题。Python 没有内置的编码检测功能，但有个 chardet 库可以做。但我想问个更根本的问题——为什么我们允许每个支付渠道返回不同编码的文件？服务端的 API 契约不应该统一成 UTF-8 吗？”

大师：“小白说到了架构层面——从源头统一编码是最优解。但现实是：你这周就要上线，银联那边改接口排期到两个月后。所以我们需要容忍混乱，但保持自己的边界。”

"编码问题上，有三个防守策略：

最好的方案：API 出口统一 UTF-8，文件入口容错处理。"

技术映射：编码问题 = 你在国外收到一封手写信——看着是中文（GBK），但邮递员（Python）按英文（UTF-8）解读就成乱码。你得先认出这是中文，再用中文词典读。

小胖：“编码我搞定了。但读写大文件时，我写了 data = f.read() 然后服务器内存就爆了——2GB 的对账文件啊！”

小白：“这个我知道——用 for line in f: 逐行读，Python 的文件对象本身就是迭代器。但我遇到过一个情况：文件里有字段本身包含换行符（CSV 中被双引号包裹的多行字段），逐行读就断了。这个怎么处理？”

大师：“好问题。逐行读取 for line in f 适用于每行对应一条逻辑记录的场景。对于 CSV 中有转义换行的场景，用 csv.reader 而不是手写解析——csv 模块已经处理好了引号内的换行。”

"大文件处理的三板斧：

“但有一个重要的坑——Windows 和 Linux 的换行符不同。Python 3 的 open() 默认启用通用换行模式（universal newlines），读入时 \\r\\n、\\r、\\n 都统一转换为 \\n。这在你做多平台文件处理时是救命的特性，但要注意：用 open(path, newline='') 可以禁用这个行为——比如你需要保留原始换行符写入时。”

技术映射：逐行读取 = 吃流水面——来一碗吃一碗，不需要一张能放 100 碗面的桌子。csv.reader = 寿司传送带——不仅逐盘拿，还把每个寿司的配料（字段）分好了。

小胖：“正则呢？我要从一堆支付回执里提取订单号，有的叫 ORDER_ID=ABC-123，有的叫 orderId=ABC-123，有的是 "订单号":"ABC-123"。——这怎么写一个正则全部匹配？”

小白：“正则是个好工具，但也是双刃剑。我见过一个正则有 200 个字符，写的人自己一周后都看不懂。怎么判断一个正则是’够用’还是’过度’了？还有 re.match、re.search、re.fullmatch 的区别，很多人搞不清。”

大师：“正则的核心思想是描述模式而非列举情况。以订单号提取为例——订单号的本质是’以 ABC- 开头，后跟三位数字’。不管你前面的标签叫 ORDER_ID 还是 orderId 还是 订单号，订单号的固有模式是 ABC-\\d{3}。”

import re

ORDER_ID_PATTERN = re.compile(r"ABC-\\d{3}")

# 三种都能匹配的文本
texts = [
'ORDER_ID=ABC-123',
'"orderId":"ABC-456"',
'订单号:ABC-789',
]

for text in texts:
match = ORDER_ID_PATTERN.search(text) # search 查找任意位置
if match:
print(match.group()) # ABC-123, ABC-456, ABC-789

"关于三个函数的区别：

• re.match()——从字符串开头开始匹配，相当于在 pattern 前自动加了 ^。适合校验整体格式。
• re.search()——在字符串任意位置搜索第一个匹配。适合提取信息。
• re.fullmatch()——整个字符串必须完全匹配。适合精确校验。

选择直觉：如果你在说’这个字符串是不是一个合法的手机号’→ 用 fullmatch。如果你在说’这个字符串里有没有手机号’→ 用 search。如果你在说’这个字符串开头的部分是不是合法手机号’→ 用 match。"

“至于正则是否过度——一个简单的判断标准：如果一个正则里出现了 3 个以上的嵌套分组 ((…))，或者需要注释才能读懂的 (?<=…) 零宽断言，考虑拆成多步处理。先提取粗粒度的文本块，再用简单的字符串方法做细粒度解析。”

技术映射：正则 = 通缉令——你不描述具体长什么样（高矮胖瘦），而是描述模式（左脸有疤、戴眼镜）。search = 全城搜捕，fullmatch = 在门口堵着查验身份证。

3. 项目实战：自动对账工具

环境准备

mkdir foodmarket-ch08 && cd foodmarket-ch08
python -m venv .venv
.venv\\Scripts\\activate

分步实现

步骤1：编写编码自适应文件读取器（目标：处理 UTF-8/GBK/BOM 混合编码）

src/file_reader.py：

"""编码自适应文件读取器"""

import csv
import json
from pathlib import Path
from typing import Iterator, Any

ENCODINGS_TO_TRY = ["utf-8-sig", "utf-8", "gbk", "gb2312", "latin-1"]

def detect_encoding(file_path: Path, sample_size: int = 10240) –> str:
"""尝试多个编码打开文件，返回第一个成功的编码"""
raw_bytes = file_path.read_bytes()[:sample_size]

# 简单启发式：如果前 3 字节是 BOM，选 utf-8-sig
if raw_bytes[:3] == b"\\xef\\xbb\\xbf":
return "utf-8-sig"

# 按顺序尝试
for enc in ENCODINGS_TO_TRY:
try:
raw_bytes.decode(enc)
return enc
except (UnicodeDecodeError, LookupError):
continue

# 兜底：latin-1 永远不会失败（0x00-0xFF 都有对应字符）
return "latin-1"

def safe_read_lines(file_path: str | Path) –> Iterator[str]:
"""安全逐行读取——自动检测编码 + 跳过空行"""
path = Path(file_path)
if not path.exists():
raise FileNotFoundError(f"对账文件不存在: {file_path}")

encoding = detect_encoding(path)
print(f"[文件] {path.name} → 检测编码: {encoding}")

with open(path, encoding=encoding, errors="replace") as f:
for line in f:
stripped = line.rstrip("\\n\\r")
if stripped: # 跳过空行
yield stripped

def read_csv_records(file_path: str | Path) –> list[dict[str, str]]:
"""读取 CSV 对账文件为字典列表"""
path = Path(file_path)
encoding = detect_encoding(path)

rows = []
with open(path, encoding=encoding, errors="replace", newline="") as f:
reader = csv.DictReader(f)
for row in reader:
# 去除字段名和值的首尾空格
cleaned = {k.strip(): v.strip() for k, v in row.items() if k}
if any(cleaned.values()): # 跳过全空行
rows.append(cleaned)

return rows

def read_json_records(file_path: str | Path) –> list[dict[str, Any]]:
"""读取 JSON 对账文件——自动处理 JSONL（每行一个 JSON）"""
path = Path(file_path)
encoding = detect_encoding(path)

raw_text = path.read_text(encoding=encoding, errors="replace").strip()

# 尝试解析为 JSON 数组
try:
data = json.loads(raw_text)
if isinstance(data, list):
return data
except json.JSONDecodeError:
pass

# 尝试 JSONL 格式（每行一个 JSON 对象）
records = []
for line in raw_text.split("\\n"):
line = line.strip()
if line:
try:
records.append(json.loads(line))
except json.JSONDecodeError:
print(f"[警告] 跳过无法解析的行: {line[:80]}…")
return records

步骤2：实现订单号正则提取与金额标准化（目标：正则实战 + 字符串处理）

src/normalizer.py：

"""订单号提取与金额标准化"""

import re
from decimal import Decimal, InvalidOperation

# ── 订单号提取 ──

# 食光集市订单号格式：SG-年月日-序号，例如 SG-20250315-00123
ORDER_ID_PATTERN = re.compile(r"SG-\\d{8}-\\d{5}")

def extract_order_id(text: str) –> str | None:
"""从任意文本中提取食光集市订单号"""
match = ORDER_ID_PATTERN.search(text)
return match.group() if match else None

# ── 金额标准化 ──

AMOUNT_CLEAN_PATTERN = re.compile(r"[^\\d.-]") # 移除非金额字符

def normalize_amount(raw: str) –> Decimal | None:
"""将各种金额格式标准化为 Decimal

支持输入：'19.90', '1990'（分）, '19.90 CNY', '￥19.90', '19,90'
"""
if not raw or not raw.strip():
return None

cleaned = raw.strip()

# 中文金额符号
if "￥" in cleaned or "¥" in cleaned:
cleaned = cleaned.replace("￥", "").replace("¥", "").strip()

# 检测是否为"分"单位（纯整数且大于 100 的可能是分）
try:
int_val = int(cleaned.replace(",", ""))
# 如果整数 > 1000 且不是标准价格范围，假设是"分"
# 注意：这个启发式需要业务校准
except ValueError:
pass

# 移除货币后缀和多余字符
cleaned = AMOUNT_CLEAN_PATTERN.sub("", cleaned.replace(",", "."))
cleaned = cleaned.strip()

try:
return Decimal(cleaned).quantize(Decimal("0.01"))
except InvalidOperation:
return None

# ── 手机号与联系方式提取 ──

PHONE_PATTERN = re.compile(r"1[3-9]\\d{9}")

def extract_phone(text: str) –> str | None:
"""提取文本中的手机号"""
match = PHONE_PATTERN.search(text)
return match.group() if match else None

# ── 支付方式标准化 ──

PAYMENT_METHOD_MAP = {
"微信": "wechat",
"微信支付": "wechat",
"wechat": "wechat",
"支付宝": "alipay",
"alipay": "alipay",
"银联": "unionpay",
"unionpay": "unionpay",
"云闪付": "unionpay",
}

def normalize_payment_method(raw: str) –> str:
"""标准化支付方式"""
cleaned = raw.strip().lower()
return PAYMENT_METHOD_MAP.get(cleaned, cleaned)

步骤3：实现对账比对引擎（目标：字典比对 + 差异报告生成）

src/reconciler.py：

"""对账比对引擎——内部订单 ↔ 外部渠道对账"""

import json
from datetime import datetime
from decimal import Decimal
from typing import Any

from src.normalizer import extract_order_id, normalize_amount, normalize_payment_method

def normalize_channel_record(record: dict[str, Any]) –> dict[str, Any]:
"""将外部渠道记录标准化为内部对账格式

每个渠道的字段名不同，需要映射到统一 schema
"""
# 识别渠道
raw_text = json.dumps(record, ensure_ascii=False)

# 提取订单号
order_id = None
for key in ("order_id", "orderId", "out_trade_no", "transaction_id", "订单号"):
if key in record:
order_id = str(record[key])
break
if not order_id:
order_id = extract_order_id(raw_text)

# 提取金额
amount = None
for key in ("amount", "total_fee", "totalFee", "交易金额"):
if key in record:
amount = normalize_amount(str(record[key]))
break

return {
"order_id": order_id,
"amount": amount,
"payment_method": normalize_payment_method(record.get("method", record.get("channel", ""))),
"status": record.get("status", record.get("trade_state", "")),
"source_raw": record,
}

def normalize_internal_record(record: dict[str, Any]) –> dict[str, Any]:
"""标准化内部订单记录"""
return {
"order_id": str(record.get("order_id", "")),
"amount": normalize_amount(str(record.get("amount", "0"))),
"payment_method": normalize_payment_method(record.get("payment_method", "")),
"status": record.get("status", ""),
}

@dataclass
class DiffRecord:
"""差异记录"""
order_id: str
diff_type: str # "amount_mismatch" | "missing_in_channel" | "missing_internal" | "status_mismatch"
internal_value: Any
channel_value: Any

def reconcile(
internal_records: list[dict],
channel_records: list[dict],
channel_name: str,
) –> dict[str, Any]:
"""执行单渠道对账

Returns:
dict: {
"channel": 渠道名,
"matched": 匹配数量,
"differences": 差异列表,
"summary": 汇总信息,
}
"""
# 标准化并建索引
internal_map: dict[str, dict] = {}
for rec in internal_records:
norm = normalize_internal_record(rec)
if norm["order_id"]:
internal_map[norm["order_id"]] = norm

channel_map: dict[str, dict] = {}
for rec in channel_records:
norm = normalize_channel_record(rec)
if norm["order_id"]:
channel_map[norm["order_id"]] = norm

internal_ids = set(internal_map)
channel_ids = set(channel_map)
common_ids = internal_ids & channel_ids

differences = []
total_amount_diff = Decimal("0")

# 1. 共同订单——比对金额和状态
for oid in sorted(common_ids):
internal = internal_map[oid]
channel = channel_map[oid]

if internal["amount"] is not None and channel["amount"] is not None:
if internal["amount"] != channel["amount"]:
diff_amount = internal["amount"] – channel["amount"]
total_amount_diff += diff_amount
differences.append({
"order_id": oid,
"type": "amount_mismatch",
"internal_amount": str(internal["amount"]),
"channel_amount": str(channel["amount"]),
"diff": str(diff_amount),
})

if internal["status"] and channel["status"]:
if internal["status"].lower() != channel["status"].lower():
differences.append({
"order_id": oid,
"type": "status_mismatch",
"internal_status": internal["status"],
"channel_status": channel["status"],
})

# 2. 内部有、渠道无
for oid in sorted(internal_ids – channel_ids):
differences.append({
"order_id": oid,
"type": "missing_in_channel",
"internal_amount": str(internal_map[oid].get("amount", "")),
})

# 3. 渠道有、内部无
for oid in sorted(channel_ids – internal_ids):
differences.append({
"order_id": oid,
"type": "missing_internal",
"channel_amount": str(channel_map[oid].get("amount", "")),
})

return {
"channel": channel_name,
"generated_at": datetime.now().isoformat(),
"internal_total": len(internal_ids),
"channel_total": len(channel_ids),
"matched": len(common_ids) – sum(
1 for d in differences if d["type"] == "amount_mismatch"
),
"amount_mismatches": sum(1 for d in differences if d["type"] == "amount_mismatch"),
"missing_in_channel": sum(1 for d in differences if d["type"] == "missing_in_channel"),
"missing_internal": sum(1 for d in differences if d["type"] == "missing_internal"),
"total_amount_diff": str(total_amount_diff),
"differences": differences,
}

def format_report(report: dict) –> str:
"""格式化差异报告为可读文本"""
lines = [
"=" * 60,
f" 食光集市 · 日对账报告",
f" 渠道: {report['channel']}",
f" 生成时间: {report['generated_at']}",
"=" * 60,
"",
f" 内部订单总数: {report['internal_total']}",
f" 渠道订单总数: {report['channel_total']}",
f" 完全匹配: {report['matched']}",
f" 金额不一致: {report['amount_mismatches']}",
f" 渠道缺失: {report['missing_in_channel']}",
f" 内部缺失: {report['missing_internal']}",
f" 金额总差异: {report['total_amount_diff']} 元",
"",
]

if report["differences"]:
lines.append("-" * 60)
lines.append(" 差异明细:")
lines.append("-" * 60)
for diff in report["differences"]:
oid = diff["order_id"]
dtype = diff["type"]
if dtype == "amount_mismatch":
lines.append(
f" [{dtype}] {oid}: 内部 {diff['internal_amount']} vs "
f"渠道 {diff['channel_amount']} (差 {diff['diff']})"
)
elif dtype == "missing_in_channel":
lines.append(f" [{dtype}] {oid}: 内部 {diff['internal_amount']} 但渠道无记录")
elif dtype == "missing_internal":
lines.append(f" [{dtype}] {oid}: 渠道 {diff['channel_amount']} 但内部无记录")
elif dtype == "status_mismatch":
lines.append(
f" [{dtype}] {oid}: 内部 {diff['internal_status']} vs "
f"渠道 {diff['channel_status']}"
)
else:
lines.append(" ✓ 全部匹配，无差异！")

return "\\n".join(lines)

步骤4：编写端到端测试（目标：用模拟数据验证全流程）

tests/test_reconcile.py：

import json
import pytest
from pathlib import Path
from src.file_reader import safe_read_lines, read_csv_records
from src.normalizer import extract_order_id, normalize_amount, normalize_payment_method
from src.reconciler import reconcile, normalize_channel_record

FIXTURES = Path(__file__).parent / "fixtures"

class TestOrderIdExtraction:
def test_standard_format(self):
assert extract_order_id("订单 SG-20250315-00123 已支付") == "SG-20250315-00123"

def test_json_format(self):
text = '"orderId": "SG-20250315-00456"'
assert extract_order_id(text) == "SG-20250315-00456"

def test_no_match(self):
assert extract_order_id("订单号 ABC-123 已支付") is None

class TestAmountNormalization:
def test_decimal_format(self):
assert normalize_amount("19.90") == Decimal("19.90")

def test_with_currency(self):
assert normalize_amount("19.90 CNY") == Decimal("19.90")

def test_chinese_yuan(self):
assert normalize_amount("￥19.90") == Decimal("19.90")

def test_empty(self):
assert normalize_amount("") is None
assert normalize_amount(None) is None

def test_invalid(self):
assert normalize_amount("not-a-number") is None

class TestPaymentMethod:
def test_wechat_variants(self):
assert normalize_payment_method("微信") == "wechat"
assert normalize_payment_method("微信支付") == "wechat"
assert normalize_payment_method("wechat") == "wechat"

class TestReconcileEngine:
def test_perfect_match(self):
internal = [
{"order_id": "SG-20250315-00001", "amount": "50.00", "payment_method": "wechat", "status": "paid"},
]
channel = [
{"orderId": "SG-20250315-00001", "total_fee": "50.00", "method": "wechat", "status": "paid"},
]
report = reconcile(internal, channel, "wechat")
assert report["amount_mismatches"] == 0
assert report["matched"] == 1

def test_amount_mismatch(self):
internal = [
{"order_id": "SG-20250315-00001", "amount": "50.00", "payment_method": "wechat", "status": "paid"},
]
channel = [
{"orderId": "SG-20250315-00001", "total_fee": "49.99", "method": "wechat", "status": "paid"},
]
report = reconcile(internal, channel, "wechat")
assert report["amount_mismatches"] == 1

def test_missing_in_channel(self):
internal = [
{"order_id": "SG-20250315-00001", "amount": "50.00", "payment_method": "wechat", "status": "paid"},
]
report = reconcile(internal, [], "wechat")
assert report["missing_in_channel"] == 1

def test_missing_internal(self):
channel = [
{"orderId": "SG-20250315-00099", "total_fee": "30.00", "method": "alipay", "status": "paid"},
]
report = reconcile([], channel, "alipay")
assert report["missing_internal"] == 1

class TestEncodingEdgeCases:
def test_utf8_bom_handling(self):
# 模拟 UTF-8 BOM 内容
content = "\\ufefforder_id,amount\\nSG-001,50.00\\n"
path = FIXTURES / "test_bom.csv"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8-sig")
records = read_csv_records(str(path))
assert records[0]["order_id"] == "SG-001"
path.unlink(missing_ok=True)

创建测试 fixtures 目录：

mkdir tests\\fixtures

运行：

python -m pytest tests/ -v

完整代码清单

foodmarket-ch08/
├── src/
│ ├── __init__.py
│ ├── file_reader.py # 编码自适应读取
│ ├── normalizer.py # 正则提取 + 金额标准化
│ └── reconciler.py # 对账比对引擎
├── tests/
│ ├── __init__.py
│ ├── fixtures/ # 测试数据
│ └── test_reconcile.py
└── requirements.txt

测试验证

python -m pytest tests/ -v
python src/file_reader.py # 编码检测演示
python src/normalizer.py # 正则提取演示

4. 项目总结

优点 & 缺点

适用场景

不适用场景

注意事项

• open() 的 encoding 参数一定要显式指定：不要依赖系统默认编码（Windows 上是 GBK，Linux 上是 UTF-8）。
• pathlib 优于 os.path：Path("data") / "2025" / "03.csv" 比 os.path.join("data", "2025", "03.csv") 直观且跨平台。
• 正则编译后复用：re.compile(pattern) 一次编译，多次使用。在循环中直接 re.search(pattern, text) 每次都会重新编译。
• errors='replace' 会丢信息：� 替换了原本的字符后无法还原。如果数据特别重要，记录下哪些文件的哪些行有编码问题，人工复核。

常见踩坑经验

故障案例1：Windows 路径中的反斜杠

现象：代码写 open("data\\output.csv")，Linux 上报 FileNotFoundError。 根因：\\o 在 Python 字符串中被解释为转义字符。Windows 碰巧能工作（因为 \\ 是路径分隔符）。 修复：用正斜杠 "data/output.csv"（Python 在所有平台都接受），或用 pathlib.Path("data") / "output.csv"。

故障案例2：csv.reader 在含引号行上行为异常

现象：CSV 里有一行 "张三","你好,"世界""，解析后 世界 被合并到一个字段。 根因：CSV 的引号转义规则是 "" 表示一个字面引号。"你好,"世界"" 实际上包含一个逗号。如果原始数据不是标准 CSV，需要自定义解析器。 修复：确认上游数据源的单据规范；必要时用 csv.reader(f, quotechar=None) 禁用引号处理。

故障案例3：正则 .* 的贪婪匹配导致跨越多行

现象：re.search(r"金额:(.*)", text) 在一个多行文本中把"金额: 50"后面直到文件末尾的全内容都匹配进去了。 根因：.* 默认不跨行，但如果 text 本身就是整个文件的内容（没有换行切割），.* 会贪婪匹配到结尾。 修复：用 .*?（非贪婪）或精确指定终止字符 r"金额:([\\d.]+)"。

思考题

答案见基础篇综合实战章附录。

延伸阅读与资源

Python 3实战精进：从脚本到高并发订单引擎 MongoDB 实战进阶与内核修炼 python入门：Rquests从菜鸟脚本到企业级SDK的网络实战圣经 Milvus向量数据库实战修炼：从 0 到 1精通向量检索与生产落地 后端工程师的 AI 转型第一课：Ollama 与私有化大模型实战 10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用 后端工程师转型AI第一课-Ollama 与私有化大模型实战 大型语言模型(LLM) vLLM 高性能推理落地实战 Agent开发之LlamaIndex 实战修炼与源码进阶 大语言模型Transformers 实战修炼与源码剖析