直接遍历MongoDB聚合结果会漏掉深层字段，因for k,v in doc.items()仅访问第一层键，而嵌套的dict/list需递归展开；须动态判断类型、记录路径、控制深度并处理空值/缺失键/ObjectId等边界情况。
为什么直接遍历 MongoDB 聚合结果会漏掉深层字段
MongoDB 聚合（$group、$lookup、$facet）常产出多层嵌套结构，比如 { "data": [{ "user": { "profile": { "tags": ["a", "b"] } } } ] }。Python 的 for k, v in doc.items() 只能访问第一层键，v 是 dict 或 list 时，内部字段就“消失”在循环外了。这不是数据丢了，是没递进进去。
常见错误现象：
– 用 json.dumps(doc, indent=2) 看起来完整，但用 pandas.json_normalize() 却只提取出顶层字段
– 手动写三层 for 循环后发现第4层又出现新结构，被迫补丁式加循环
– 把 list 当 dict 处理，触发 TypeError: string indices must be integers
- 聚合结果中
list和dict混杂，必须动态判断类型再分支处理 - 同一字段名可能在不同层级重复出现（如多个
name），扁平化时需带路径前缀避免覆盖 - 空值（
None）、缺失键（KeyError）、ObjectId类型需提前转换或跳过，否则递归中断
用递归函数展开嵌套：核心逻辑与安全边界
关键不是“写个递归”，而是控制展开深度、路径记录方式和类型兜底。下面这个函数不依赖外部库，适配绝大多数聚合输出：
def flatten_doc(doc, prefix="", result=None, max_depth=8):
if result is None:
result = {}
if max_depth <= 0:
result[f"{prefix}__TRUNCATED"] = str(doc)
return result
if isinstance(doc, dict):
for k, v in doc.items():
new_key = f"{prefix}.{k}" if prefix else k
flatten_doc(v, new_key, result, max_depth - 1)
elif isinstance(doc, list):
for i, item in enumerate(doc):
new_key = f"{prefix}.{i}" if prefix else str(i)
flatten_doc(item, new_key, result, max_depth - 1)
else:
result[prefix] = doc if not isinstance(doc, ObjectId) else str(doc)
return result
使用场景：
– 直接传入 collection.aggregate([...]) 返回的 Cursor 中单个文档
– 配合 pandas.DataFrame([flatten_doc(d) for d in cursor]) 构建宽表
– 替代 json_normalize 处理含混合 list/dict 的非规则结构
立即学习“Python免费学习笔记（深入）”；
-
max_depth必须设限，防止无限嵌套（如意外的循环引用）导致栈溢出 -
ObjectId不转字符串会报TypeError: Object of type ObjectId is not JSON serializable，所以统一转str() - 路径分隔符用
.而非_，方便后续用df.filter(regex=r"^user\.profile\.")筛选子树
聚合结果中 $lookup 产生的数组怎么避免字段爆炸
$lookup 常返回 "orders": [{"id": 1, "items": [...]}, {"id": 2}]，直接 flatten_doc 会展开成 orders.0.id、orders.0.items.0.name、orders.1.id……列数随数组长度线性增长，pandas DataFrame 会严重稀疏。
更实用的做法是：对已知的“一对多”字段（如 orders、tags）提前切片或聚合，而不是无差别展开：
- 在递归前检查键名，对黑名单字段（
["orders", "history", "comments"]）改用len(v)、sum(1 for _ in v)或json.dumps(v[:3], separators=(',', ':'))存为字符串 - 若需保留明细，先用
[flatten_doc(item, f"order_{i}") for i, item in enumerate(doc.get("orders", []))]单独处理，再合并到主表 - 注意
$lookup结果为空数组[]时，flatten_doc仍会生成orders.0这类不存在的键——应在递归入口加if not doc: return
调试时怎么快速定位某字段在哪一层嵌套里
别靠肉眼扫 print(json.dumps(doc, indent=2))。写个轻量搜索函数：
def find_key(doc, target, path=""):
if isinstance(doc, dict):
for k, v in doc.items():
new_path = f"{path}.{k}" if path else k
if k == target:
print(f"Found: {new_path}")
find_key(v, target, new_path)
elif isinstance(doc, list):
for i, item in enumerate(doc):
find_key(item, target, f"{path}.{i}")
调用 find_key(result_doc, "email")，输出类似 Found: user.contact.primary.email，立刻知道该去哪层取值。比翻 200 行缩进的 JSON 快得多。
真正麻烦的不是嵌套深，是聚合阶段用了 $cond 或 $switch 导致某些文档有 user.profile，另一些只有 user.basic——这种结构不一致，必须在递归里做 try/except KeyError 或用 dict.get(k, default)，不能假设路径存在。
