改进建议概览
- 依据 PEP 8：统一小写下划线命名、避免裸 except、避免可变默认值、移除全局状态、使用 logging 替代 print。
- 类型注解：为函数、返回值与数据结构添加类型注解与文档字符串，提升 IDE 支持与测试可读性。
- 拆分纯函数与 I/O：将文件读取（I/O）与合并逻辑（纯函数）分离，便于单元测试。
- 使用 with 上下文管理：保证文件句柄正确关闭。
- 简化循环与检索：用字典索引替代嵌套循环 O(n*m)，并用 csv 模块替代手工 split。
- 细化异常类型：分别捕获 FileNotFoundError、json.JSONDecodeError、UnicodeDecodeError，并保持返回类型一致。
- 返回类型一致性与测试友好性：始终返回同一类型（例如 List[UserRecord] 或 MergeResult），并提供详细统计信息以便测试断言。
- 数据清洗：规范 id 为字符串、去除空白；将 active 统一为布尔类型；跳过空行/无效行。
推荐实现（拆分 I/O 与纯函数，保留统计信息）
from future import annotations
import csv
import json
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, Iterable, List, Mapping, Optional
logger = logging.getLogger(name)
@dataclass(frozen=True)
class UserRecord:
"""统一的数据结构，提升类型安全与可维护性。"""
id: str
name: str
active: bool
@dataclass(frozen=True)
class MergeResult:
"""返回合并结果与统计信息，测试更友好。"""
records: List[UserRecord]
processed: int
matched: int
skipped_ids: List[str]
def parse_bool(value: Any, default: bool = False) -> bool:
"""将多种形式的真值/假值规范为 bool。"""
if isinstance(value, bool):
return value
if isinstance(value, str):
s = value.strip().lower()
if s in {"true", "1", "yes", "y", "t"}:
return True
if s in {"false", "0", "no", "n", "f", ""}:
return False
return default
if isinstance(value, (int, float)):
return bool(value)
return default
def merge_records(
ids: Iterable[str],
index: Mapping[str, Mapping[str, Any]]
) -> MergeResult:
"""
纯函数：将 CSV 中的 id 列与 JSON 索引合并为统一记录。
- 不进行 I/O、无副作用。
- 保留统计信息，便于测试。
"""
records: List[UserRecord] = []
skipped: List[str] = []
processed = 0
matched = 0
for raw_id in ids:
processed += 1
item_id = str(raw_id).strip()
if not item_id:
skipped.append(item_id)
continue
found = index.get(item_id)
if found is None:
skipped.append(item_id)
continue
records.append(
UserRecord(
id=item_id,
name=str(found.get("name", "")),
active=parse_bool(found.get("active", False)),
)
)
matched += 1
return MergeResult(records=records, processed=processed, matched=matched, skipped_ids=skipped)
def load_ids_from_csv(csv_path: Path) -> List[str]:
"""
I/O 函数：从 CSV 读取第一列 id。
使用 csv 模块以避免手工 split 的边界问题（逗号、引号、空格等）。
"""
ids: List[str] = []
with csv_path.open(newline="", encoding="utf-8") as f:
reader = csv.reader(f)
for row in reader:
if not row:
continue
ids.append(row[0].strip())
return ids
def load_json_index(json_path: Path) -> Dict[str, Dict[str, Any]]:
"""
I/O 函数：读取 JSON 列表并用 id 建立索引，便于 O(1) 检索。
"""
with json_path.open(encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, list):
raise ValueError("JSON root must be a list of objects")
index: Dict[str, Dict[str, Any]] = {}
for obj in data:
if not isinstance(obj, dict):
continue
key = str(obj.get("id", "")).strip()
if key:
index[key] = obj
return index
def read_and_merge(csv_path: str | Path, json_path: str | Path) -> MergeResult:
"""
编排函数：进行 I/O 并调用纯函数合并。
- 细化异常并使用 logging。
- 保持返回类型一致，异常时返回空结果。
"""
csv_path = Path(csv_path)
json_path = Path(json_path)
try:
ids = load_ids_from_csv(csv_path)
index = load_json_index(json_path)
except FileNotFoundError as exc:
logger.error("File not found: %s", exc.filename)
return MergeResult(records=[], processed=0, matched=0, skipped_ids=[])
except json.JSONDecodeError as exc:
logger.error("Invalid JSON: %s", exc)
return MergeResult(records=[], processed=0, matched=0, skipped_ids=[])
except UnicodeDecodeError as exc:
logger.error("Encoding error: %s", exc)
return MergeResult(records=[], processed=0, matched=0, skipped_ids=[])
except ValueError as exc:
logger.error("Data format error: %s", exc)
return MergeResult(records=[], processed=0, matched=0, skipped_ids=[])
return merge_records(ids, index)
def records_to_dicts(records: Iterable[UserRecord]) -> List[Dict[str, Any]]:
"""
如果需要与原有 dict 格式兼容（例如对外 API），可将 dataclass 转回字典。
"""
return [{"id": r.id, "name": r.name, "active": r.active} for r in records]
if name == "main":
logging.basicConfig(level=logging.INFO)
result = read_and_merge("data/users.csv", "data/users.json")
# 保持输出简单，同时兼容原来的打印
print(records_to_dicts(result.records))
print({"processed": result.processed, "matched": result.matched, "skipped_ids": result.skipped_ids})
关键优化点详解
- 移除全局状态与可变默认值
- 原代码的 COUNTER 与 cache={} 会引入隐式共享状态与难测的副作用。改用 MergeResult 携带统计信息并作为返回值。
- PEP 8 与命名
- 函数与变量名统一为 snake_case；常量与类名分别使用全大写与驼峰。
- 返回类型一致性
- 异常情况下不返回 {} 与 List 混合；统一返回 MergeResult（空结果）并记录日志。
- with 上下文管理
- 细化异常类型
- 分别捕获 FileNotFoundError、json.JSONDecodeError、UnicodeDecodeError、ValueError；便于定位问题与单元测试断言。
- 简化循环与检索逻辑
- 用 dict 索引替代在 JSON 列表中的线性搜索；时间复杂度从 O(n*m) 降为 O(n+m)。
- 使用 csv.reader 而不是手工 split，避免引号、转义、空格等边界问题。
- 数据清洗与类型统一
- 统一将 id 规范为 str 并 strip；空 id 与空行会被跳过。
- active 字段统一为 bool，避免字符串 'false' 导致下游逻辑混乱；保留 parse_bool 以兼容已有数据的多种形式。
- 测试友好性
- 纯函数 merge_records 可直接用内存数据（list/dict）进行单元测试。
- I/O 函数分离，便于用临时文件或 mock 测试。
- MergeResult 提供详细统计，测试时可断言 processed/matched/skipped。
如需尽量保持原返回结构
- 把 read_and_merge 改为返回 List[Dict[str, Any]]，并在异常时返回空列表而非 {}。
- 若必须保留字符串 'false'，可在记录构造处将 active 替换为 str(parse_bool(...)).lower()，但不推荐。
