import os import shutil import json import argparse def do_merge(new_json_path, old_json_path): with open(new_json_path, "r") as f: new_json = json.load(f) with open(old_json_path, "r") as f: old_json = json.load(f) if new_json.get("id") != old_json.get("id"): raise RuntimeError(f"{new_json.get('id')} not equal to {old_json.get('id')} ") # 基础属性,从新的里拿,不为空就覆盖。 base_params = [ "description", "owner", ] # 获取基础属性 for param in base_params: if new_json.get(param): old_json[param] = new_json.get(param) # 更新result new_results = new_json.get("results") old_results = old_json.get("results") for key in new_results: if new_results[key]: old_results[key] = new_results[key] # 更新metadata if new_json.get("metadata"): old_json["metadata"] = new_json.get("metadata") return old_json def merge(new_data_path, old_data_path): if not os.path.exists(new_data_path): raise RuntimeError(f"data_path 【{new_data_path}】 not exist!") # foreach new files for new_json_file in os.listdir(new_data_path): if not new_json_file.endswith("json"): print(f"file {new_json_file} not a json file,jump") continue # check old file exist or not if not os.path.exists(os.path.join(old_data_path, new_json_file)): print(f"new file 【{new_json_file}】 copy...") shutil.copyfile(os.path.join(new_data_path, new_json_file), os.path.join(old_data_path, new_json_file)) else: print(f"merge file 【{new_json_file}】") new_json_content = do_merge(os.path.join(new_data_path, new_json_file), os.path.join(old_data_path, new_json_file)) with open(os.path.join(old_data_path, new_json_file), "w") as f: json.dump(new_json_content, f, indent=4, ensure_ascii=False) if __name__ == '__main__': args = argparse.ArgumentParser() args.add_argument("--new_data", type=str, required=True) args.add_argument('--old_data', type=str, default=None, required=False) args = args.parse_args() new_data: str = args.new_data old_data: str = args.old_data or os.path.dirname(os.path.abspath(__file__)) merge(new_data, old_data)