event_schema.py (1610B)
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 import json 4 from collections import defaultdict 5 from typing import List 6 7 8 class EventSchema: 9 def __init__(self, type_list, role_list, type_role_dict): 10 self.type_list = type_list 11 self.role_list = role_list 12 self.type_role_dict = type_role_dict 13 14 @staticmethod 15 def read_from_file(filename): 16 lines = open(filename).readlines() 17 type_list = json.loads(lines[0]) 18 role_list = json.loads(lines[1]) 19 type_role_dict = json.loads(lines[2]) 20 return EventSchema(type_list, role_list, type_role_dict) 21 22 def write_to_file(self, filename): 23 with open(filename, 'w') as output: 24 output.write(json.dumps(self.type_list) + '\n') 25 output.write(json.dumps(self.role_list) + '\n') 26 output.write(json.dumps(self.type_role_dict) + '\n') 27 28 29 def merge_schema(schema_list: List[EventSchema]): 30 type_set = set() 31 role_set = set() 32 type_role_dict = defaultdict(list) 33 34 for schema in schema_list: 35 36 for type_name in schema.type_list: 37 type_set.add(type_name) 38 39 for role_name in schema.role_list: 40 role_set.add(role_name) 41 42 for type_name in schema.type_role_dict: 43 type_role_dict[type_name] += schema.type_role_dict[type_name] 44 45 for type_name in type_role_dict: 46 type_role_dict[type_name] = list(set(type_role_dict[type_name])) 47 48 return EventSchema(type_list=list(type_set), 49 role_list=list(role_set), 50 type_role_dict=type_role_dict, 51 )