Python3中高效匹配JSON规则与数据以获取articleId的技术问询

需求说明 #

在Python3环境下完成以下功能：

• 基于rules.json中的规则集合，为data.json中的每个数据条目匹配对应的articleId
• 规则与数据的对应关系：rules中的propertyId对应data中的property_id，rules中的value对应data中的property_value
• 规则内的条件通过logicalOperator（目前为AND）关联，需满足所有条件才匹配该规则的articleId

规则示例（rules.json） #

{"rules": [
    {"articleId": "art1",
     "properties_rule": [
         {"condition": "EQ", "logicalOperator": "AND", "propertyId": 487, "value": "aaaa"},
         {"condition": "EQ", "logicalOperator": "", "propertyId": 487, "value": "zzzz"}
     ]},
    {"articleId": "art2",
     "properties_rule": [
         {"condition": "GTE", "logicalOperator": "AND", "propertyId": 487, "value": "bbbb"},
         {"condition": "LTE", "logicalOperator": "", "propertyId": 487, "value": "eeee"}
     ]},
    {"articleId": "art3",
     "properties_rule": [
         {"condition": "GTE", "logicalOperator": "", "propertyId": 487, "value": "ffff"}
     ]}
]}

数据示例（data.json） #

{"data": {
    "1": {"properties_values": [{"value": {"property_id": 487, "property_value": "aaaa", "response_id": 1}}]},
    "2": {"properties_values": [{"value": {"property_id": 487, "property_value": "bbbb", "response_id": 2}}]},
    "3": {"properties_values": [{"value": {"property_id": 487, "property_value": "eeee", "response_id": 3}}]}
}}

现有实现的问题 #

原代码采用四层嵌套循环（规则→规则条件→数据条目→属性值），时间复杂度为O(R*C*D*P)（R=规则数，C=单规则条件数，D=数据条目数，P=单条目的属性数）。当数据量或规则量较大时，重复遍历会导致效率急剧下降，且存在字段名错误（如原代码中误写properties_value、survey_response）。

优化方案 #

核心优化点 #

1. 预处理数据，构建属性值索引：将数据按property_id分组，存储该属性的所有取值集合，后续查询时直接获取，无需重复遍历整个数据集
2. 预定义条件判断函数：把EQ/GTE/LTE转换为可直接调用的函数，避免每次判断的分支开销
3. 规则匹配逻辑简化：利用预处理后的索引，快速验证规则条件是否满足

优化后代码 #

import json

def preprocess_data(raw_data):
    """预处理数据，构建{property_id: 取值集合}的索引"""
    property_index = {}
    for entry in raw_data.values():
        for prop_item in entry["properties_values"]:
            value_info = prop_item["value"]
            prop_id = value_info["property_id"]
            prop_value = value_info["property_value"]
            # 为每个property_id维护一个取值集合
            if prop_id not in property_index:
                property_index[prop_id] = set()
            property_index[prop_id].add(prop_value)
    return property_index

def get_condition_func(condition):
    """返回对应条件的判断函数"""
    condition_map = {
        "EQ": lambda x, y: x == y,
        "GTE": lambda x, y: x >= y,
        "LTE": lambda x, y: x <= y
    }
    return condition_map.get(condition, lambda x, y: False)

def match_rules(rules, property_index):
    """遍历规则，找到第一个满足所有条件的articleId"""
    for rule in rules:
        article_id = rule["articleId"]
        conditions = rule["properties_rule"]
        all_matched = True
        
        for cond in conditions:
            prop_id = cond["propertyId"]
            target_value = cond["value"]
            condition_func = get_condition_func(cond["condition"])
            
            # 先检查该属性是否有数据，没有则直接不匹配
            if prop_id not in property_index:
                all_matched = False
                break
            
            # 检查该属性的所有取值中是否有满足条件的
            has_match = any(condition_func(val, target_value) for val in property_index[prop_id])
            if not has_match:
                all_matched = False
                break
        
        if all_matched:
            return article_id
    # 无匹配规则时返回默认值
    return "Default Article ID"

if __name__ == "__main__":
    # 加载数据
    with open("rules.json", "r") as f:
        rules = json.load(f)["rules"]
    
    with open("data.json", "r") as f:
        raw_data = json.load(f)["data"]
    
    # 预处理数据并匹配规则
    property_index = preprocess_data(raw_data)
    result = match_rules(rules, property_index)
    print(f"Article ID: {result}")

优化效果说明 #

• 时间复杂度降低为O(D*P + R*C*V)（V为单属性的平均取值数），预处理仅执行一次，后续规则匹配无需重复遍历原始数据
• 利用集合查询和预定义函数，减少了分支判断和重复计算
• 代码结构更清晰，易于维护和扩展（如新增条件类型）

内容的提问来源于stack exchange，提问作者Sudipta Dhara