|
| 1 | +import json |
| 2 | +from ruamel.yaml import YAML |
| 3 | +import csv |
| 4 | +import os |
| 5 | +from io import StringIO |
| 6 | + |
| 7 | +def create_or_update_json_entry(rocrate, keys_path, new_value): |
| 8 | + """ |
| 9 | + Create or update a nested JSON entry in a ro-crate structure. |
| 10 | +
|
| 11 | + Args: |
| 12 | + rocrate (dict): The main ro-crate dictionary. |
| 13 | + keys_path (str): Dot-separated path to the key that needs updating. |
| 14 | + new_value (any): New value to be inserted or updated. |
| 15 | + """ |
| 16 | + # Split the keys path into individual components |
| 17 | + keys = keys_path.split('.') |
| 18 | + prefix = "" |
| 19 | + structure = rocrate |
| 20 | + |
| 21 | + # Traverse through the nested structure using keys except the last one |
| 22 | + for key in keys[:-1]: |
| 23 | + key = prefix + key |
| 24 | + |
| 25 | + # Handle potential './' prefix logic |
| 26 | + if key == "": |
| 27 | + prefix = "." |
| 28 | + continue |
| 29 | + else: |
| 30 | + prefix = "" |
| 31 | + |
| 32 | + if isinstance(structure, list): |
| 33 | + # Find the item with matching '@id' key |
| 34 | + for item in structure: |
| 35 | + if item.get("@id") == key: |
| 36 | + structure = item |
| 37 | + break |
| 38 | + else: |
| 39 | + print(f"Key '{key}' not found.") |
| 40 | + return |
| 41 | + elif key in structure: |
| 42 | + structure = structure[key] |
| 43 | + else: |
| 44 | + print(f"Key '{key}' not found.") |
| 45 | + return |
| 46 | + |
| 47 | + # The final key where the new value should be placed |
| 48 | + last_key = keys[-1] |
| 49 | + |
| 50 | + # Update the value at the final key |
| 51 | + if last_key in structure: |
| 52 | + if isinstance(structure[last_key], list): |
| 53 | + # Prepend only if the new value is not already in the list |
| 54 | + if new_value not in structure[last_key]: |
| 55 | + structure[last_key].insert(0, new_value) |
| 56 | + else: |
| 57 | + # Convert existing non-list value to a list if needed |
| 58 | + structure[last_key] = [new_value, structure[last_key]] |
| 59 | + else: |
| 60 | + # If the key doesn't exist, create a new list with the new value |
| 61 | + structure[last_key] = [new_value] |
| 62 | + |
| 63 | + |
| 64 | +def navigate_and_assign(source, path, value): |
| 65 | + """Navigate through a nested dictionary and assign a value to the specified path.""" |
| 66 | + keys = path.split('.') |
| 67 | + for i, key in enumerate(keys[:-1]): |
| 68 | + if key.isdigit(): # If the key is a digit, it's an index for a list |
| 69 | + key = int(key) |
| 70 | + while len(source) <= key: # Extend the list if necessary |
| 71 | + source.append({}) |
| 72 | + source = source[key] |
| 73 | + else: |
| 74 | + if i < len(keys) - 2 and keys[i + 1].isdigit(): # Next key is a digit, so ensure this key leads to a list |
| 75 | + source = source.setdefault(key, []) |
| 76 | + else: # Otherwise, it leads to a dictionary |
| 77 | + source = source.setdefault(key, {}) |
| 78 | + # Assign the value to the final key |
| 79 | + if keys[-1].isdigit(): # If the final key is a digit, it's an index for a list |
| 80 | + key = int(keys[-1]) |
| 81 | + while len(source) <= key: # Extend the list if necessary |
| 82 | + source.append(None) |
| 83 | + source[key] = value |
| 84 | + else: |
| 85 | + source[keys[-1]] = value |
| 86 | + |
| 87 | + |
| 88 | +def read_yaml_with_header(file_path): |
| 89 | + """ |
| 90 | + Read YAML content inside YAML header delimiters '---' |
| 91 | + """ |
| 92 | + |
| 93 | + with open(file_path,'r') as file: |
| 94 | + data = file.read() |
| 95 | + |
| 96 | + yaml = YAML() |
| 97 | + yaml_content = yaml.load(data.strip('---\n')) |
| 98 | + |
| 99 | + return yaml_content |
| 100 | + |
| 101 | +def update_csv_content(file_path, field, value): |
| 102 | + # Read the CSV file and update the field value |
| 103 | + updated_rows = [] |
| 104 | + field_exists = False |
| 105 | + with open(file_path, mode='r', newline='') as file: |
| 106 | + reader = csv.reader(file) |
| 107 | + for row in reader: |
| 108 | + if row and row[0] == field: |
| 109 | + row[1] = value |
| 110 | + field_exists = True |
| 111 | + updated_rows.append(row) |
| 112 | + |
| 113 | + # If the field does not exist, add a new line |
| 114 | + if not field_exists: |
| 115 | + updated_rows.append([field, value]) |
| 116 | + |
| 117 | + # Convert the updated rows back into a CSV-formatted string |
| 118 | + updated_csv_content = StringIO() |
| 119 | + writer = csv.writer(updated_csv_content) |
| 120 | + writer.writerows(updated_rows) |
| 121 | + updated_csv_string = updated_csv_content.getvalue() |
| 122 | + |
| 123 | + return updated_csv_string |
0 commit comments