MongoDB导出特殊格式列转换及数据查询方案咨询

一、将非标准格式转换为事件字典列表 #

你拿到的格式是MongoDB扩展JSON的变种，用=替代了标准JSON的:，且未给键值对添加引号，需要先做字符串清洗和结构化解析：

1. 清洗原始字符串，提取单个事件块：

raw_str = "'[{idEvento.$oid=63ffaec3cdc01e6352729bad, dataHoraEvento.$date=1677690003377, codigoTipoEvento=1, mesAnoReferenciaContabilizacao=032023}, {idEvento.$oid=63ffb5c8cdc01e6352729bae, dataHoraEvento.$date=1677691800676, codigoTipoEvento=3, mesAnoReferenciaContabilizacao=032023}, {idEvento.$oid=6405cc8711c78c20369b4033, dataHoraEvento.$date=1678090851560, codigoTipoEvento=8, mesAnoReferenciaContabilizacao=032023}, {idEvento.$oid=6422b4c97e45dd75abb4f831, dataHoraEvento.$date=1679985307560, codigoTipoEvento=6, mesAnoReferenciaContabilizacao=032023, _class=br.com.bb.rcp.model.vantagens.HistoricoContabil}, {idEvento.$oid=6422b4c97e45dd75abb4f832, dataHoraEvento.$date=1679985309584, codigoTipoEvento=6, mesAnoReferenciaContabilizacao=032023, _class=br.com.bb.rcp.model.vantagens.HistoricoContabil}]'"

# 去除首尾无效字符，分割每个事件块
cleaned_str = raw_str.strip("'[]")
event_blocks = [block.strip() for block in cleaned_str.split('}, {')]

2. 编写解析函数，将事件块转为结构化字典，同时处理特殊字段：

import datetime

def parse_event(block):
    event_dict = {}
    # 分割键值对
    key_value_pairs = [pair.strip() for pair in block.split(',')]
    for pair in key_value_pairs:
        key_part, value = pair.split('=', 1)  # 仅分割第一个=，避免值含=的异常
        # 处理MongoDB的ObjectID字段
        if key_part == 'idEvento.$oid':
            event_dict['idEvento'] = value
        # 处理时间戳字段，转为datetime格式
        elif key_part == 'dataHoraEvento.$date':
            timestamp = int(value)
            event_dict['dataHoraEvento'] = datetime.datetime.fromtimestamp(timestamp/1000)
        # 数值类型字段转换
        elif key_part == 'codigoTipoEvento':
            event_dict[key_part] = int(value)
        # 其他字段直接映射
        else:
            event_dict[key_part] = value
    return event_dict

# 生成标准事件字典列表
event_list = [parse_event(block) for block in event_blocks]

二、事件内容的查询过滤方案 #

不建议用字符串查询——字符串匹配易受字段顺序、空格等因素影响，可靠性低。推荐两种更稳妥的方式：

方式1：先过滤再展开（适合大数据量场景） #

先在列表层面过滤符合条件的事件，再展开为行，减少后续数据处理量：

import pandas as pd

# 假设原始DataFrame为df，目标字段为'historico_contabil'
# 先将字段转为事件列表
df['event_list'] = df['historico_contabil'].apply(lambda x: parse_event_list(x))  # parse_event_list是整合上述清洗+解析逻辑的函数

# 过滤codigoTipoEvento=6的事件
df['filtered_events'] = df['event_list'].apply(lambda events: [e for e in events if e['codigoTipoEvento'] == 6])

# 展开为单行单事件，并拆分字典为列
expanded_df = df.explode('filtered_events').dropna(subset=['filtered_events'])
expanded_df = pd.concat([expanded_df, expanded_df['filtered_events'].apply(pd.Series)], axis=1)

方式2：先展开再过滤（逻辑更直观） #

先将事件列表展开为每行一个事件，再用pandas布尔索引做精准过滤：

# 展开事件列表为单行单事件
expanded_df = df.explode('event_list').dropna(subset=['event_list'])
# 将事件字典拆分为独立列
expanded_df = pd.concat([expanded_df, expanded_df['event_list'].apply(pd.Series)], axis=1)

# 多条件过滤示例：codigoTipoEvento=6且会计参考年月为032023
filtered_df = expanded_df[(expanded_df['codigoTipoEvento'] == 6) & (expanded_df['mesAnoReferenciaContabilizacao'] == '032023')]

这种方式逻辑简单，适合数据量不大的场景，过滤条件编写更直观。

内容的提问来源于stack exchange，提问作者FábioRB