如何在MongoDB变更流中监听嵌套数组指定字段变更并获取对应更新项

解决方法分三步实现： #

1. 优化更新操作写法（核心前提）

你当前的更新是直接重写整个matches数组，导致change stream只能识别到整个数组变更，无法定位到具体元素。请改用MongoDB数组过滤更新语法，仅更新目标比赛的win字段：

# 示例：更新match_id为3A4j0sp26的比赛win字段
client.mydb.match.update_one(
    {"tournament_id": "P1oi12mwj10b1b"},
    {"$set": {"matches.$[dateGroup].matches.$[match].win": "team2"}},
    array_filters = [
        {"dateGroup.date_order": 1},
        {"match.match_id": "3A4j0sp26"}
    ]
)

执行该更新后，change stream的updateDescription.updatedFields会返回具体的字段路径，如matches.0.matches.1.win，而非整个数组内容。

2. 配置change stream过滤器

开启updateLookup参数获取更新后的完整文档，过滤器仅筛选win字段变更的更新操作和文档替换操作：

import pymongo
from bson.json_util import dumps
import re

MONGO_URI = 'mongodb://localhost/mydb'
client = pymongo.MongoClient(MONGO_URI)

# 过滤器配置
filters = [{
    "$match": {
        "$or": [
            # 匹配嵌套win字段的更新操作
            {
                "operationType": "update",
                "$expr": {
                    "$gt": [
                        {"$size": {"$filter": {
                            "input": {"$objectToArray": "$updateDescription.updatedFields"},
                            "as": "field",
                            "cond": {"$regexMatch": {"input": "$$field.k", "regex": r"^matches\.\d+\.matches\.\d+\.win$"}}
                        }}},
                        0
                    ]
                }
            },
            # 兼容全文档替换的场景
            {"operationType": "replace"}
        ]
    }
}]

# 开启updateLookup获取完整文档
change_stream = client.mydb.match.watch(filters, full_document='updateLookup')

3. 解析变更事件，提取目标比赛条目

# 若存在replace操作，需缓存上一版本的文档用于diff
doc_cache = {}

for change in change_stream:
    full_doc = change['fullDocument']
    doc_id = str(full_doc['_id'])
    result_list = []

    if change['operationType'] == 'update':
        # 从更新字段路径提取索引，定位目标比赛
        for field_key in change['updateDescription']['updatedFields'].keys():
            match = re.match(r'matches\.(\d+)\.matches\.(\d+)\.win', field_key)
            if match:
                date_idx = int(match.group(1))
                match_idx = int(match.group(2))
                date_group = full_doc['matches'][date_idx]
                match_item = date_group['matches'][match_idx]
                # 拼装结果
                result = match_item.copy()
                result['tournament_id'] = full_doc['tournament_id']
                result['date_order'] = date_group['date_order']
                result_list.append(result)
    
    elif change['operationType'] == 'replace':
        # 替换操作需和缓存的旧文档做diff找win变更项
        old_doc = doc_cache.get(doc_id)
        if old_doc:
            # 遍历所有比赛对比win字段
            for old_date_group in old_doc['matches']:
                date_order = old_date_group['date_order']
                new_date_group = next(g for g in full_doc['matches'] if g['date_order'] == date_order)
                for old_match in old_date_group['matches']:
                    match_id = old_match['match_id']
                    new_match = next(m for m in new_date_group['matches'] if m['match_id'] == match_id)
                    if old_match.get('win') != new_match.get('win'):
                        result = new_match.copy()
                        result['tournament_id'] = full_doc['tournament_id']
                        result['date_order'] = date_order
                        result_list.append(result)
        # 更新缓存
        doc_cache[doc_id] = full_doc
    
    # 打印结果
    for res in result_list:
        print(dumps(res))

特殊情况兼容

如果你无法修改现有的更新操作逻辑（必须重写整个matches数组），则可以跳过过滤器配置，所有变更事件都走replace操作的diff逻辑即可。

内容的提问来源于stack exchange，提问作者Saurav Pathak