1. import json
  2. import argparse
  3. import os
  5. def transform_data(input_file, output_file, schema_file, overrides):
  6. """
  7. Transforms data from an input file based on a schema and manual overrides.
  9. Args:
  10. input_file (str): Path to the input data file (JSON).
  11. output_file (str): Path to the output data file (JSON).
  12. schema_file (str): Path to the schema file (JSON).
  13. overrides (dict): Dictionary of field overrides. Key is field name, value is new value.
  14. """
  16. try:
  17. with open(input_file, 'r') as f:
  18. data = json.load(f)
  19. except FileNotFoundError:
  20. print(f"Error: Input file not found: {input_file}")
  21. return
  22. except json.JSONDecodeError:
  23. print(f"Error: Invalid JSON in input file: {input_file}")
  24. return
  26. try:
  27. with open(schema_file, 'r') as f:
  28. schema = json.load(f)
  29. except FileNotFoundError:
  30. print(f"Error: Schema file not found: {schema_file}")
  31. return
  32. except json.JSONDecodeError:
  33. print(f"Error: Invalid JSON in schema file: {schema_file}")
  34. return
  37. transformed_data = []
  38. for item in data:
  39. transformed_item = {}
  40. for field in schema['fields']:
  41. field_name = field['name']
  42. field_type = field['type']
  44. #Apply override if present
  45. if field_name in overrides:
  46. transformed_item[field_name] = overrides[field_name]
  47. continue
  49. if field_type == 'string':
  50. transformed_item[field_name] = str(item.get(field_name, "")) #Handle missing fields
  51. elif field_type == 'integer':
  52. try:
  53. transformed_item[field_name] = int(item.get(field_name, 0)) #Handle missing fields
  54. except (ValueError, TypeError):
  55. transformed_item[field_name] = 0 #Default value if conversion fails
  56. elif field_type == 'float':
  57. try:
  58. transformed_item[field_name] = float(item.get(field_name, 0.0)) #Handle missing fields
  59. except (ValueError, TypeError):
  60. transformed_item[field_name] = 0.0 #Default value if conversion fails
  61. elif field_type == 'boolean':
  62. transformed_item[field_name] = bool(item.get(field_name, False)) #Handle missing fields
  63. elif field_type == 'array':
  64. transformed_item[field_name] = item.get(field_name, []) #Handle missing fields
  65. else:
  66. transformed_item[field_name] = item.get(field_name, "") #Default to empty string for unknown types
  68. transformed_data.append(transformed_item)
  70. try:
  71. with open(output_file, 'w') as f:
  72. json.dump(transformed_data, f, indent=4)
  73. except IOError:
  74. print(f"Error: Could not write to output file: {output_file}")
  75. return
  76. print(f"Data transformed and saved to: {output_file}")
  79. if __name__ == "__main__":
  80. parser = argparse.ArgumentParser(description="Transform data from a JSON file.")
  81. parser.add_argument("input_file", help="Path to the input JSON file.")
  82. parser.add_argument("output_file", help="Path to the output JSON file.")
  83. parser.add_argument("schema_file", help="Path to the schema JSON file.")
  84. parser.add_argument("--overrides", help="Path to a JSON file containing field overrides.")
  86. args = parser.parse_args()
  88. overrides = {}
  89. if args.overrides:
  90. try:
  91. with open(args.overrides, 'r') as f:
  92. overrides = json.load(f)
  93. except FileNotFoundError:
  94. print(f"Error: Overrides file not found: {args.overrides}")
  95. exit(1)
  96. except json.JSONDecodeError:
  97. print(f"Error: Invalid JSON in overrides file: {args.overrides}")
  98. exit(1)

Add your comment