1. import json
  2. import logging
  4. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  6. def batch_user_data_operations(input_file, output_file, operations, dry_run=False):
  7. """
  8. Performs batch operations on user data from a JSON file.
  10. Args:
  11. input_file (str): Path to the input JSON file containing user data.
  12. output_file (str): Path to the output JSON file to save processed data.
  13. operations (list): A list of dictionaries, where each dictionary
  14. specifies an operation to perform on each user.
  15. Example:
  16. [
  17. {"operation": "update", "field": "email", "value": "new_email@example.com"},
  18. {"operation": "add_tag", "tag": "active"}
  19. ]
  20. dry_run (bool): If True, the operations are simulated and no changes
  21. are made to the output file.
  22. """
  24. try:
  25. with open(input_file, 'r') as f:
  26. user_data = json.load(f)
  27. except FileNotFoundError:
  28. logging.error(f"Input file not found: {input_file}")
  29. return
  30. except json.JSONDecodeError:
  31. logging.error(f"Invalid JSON format in: {input_file}")
  32. return
  34. processed_data = []
  35. for user in user_data:
  36. processed_user = user.copy() # Create a copy to avoid modifying original data
  37. for op in operations:
  38. operation_type = op['operation']
  40. if operation_type == 'update':
  41. field = op['field']
  42. value = op['value']
  43. processed_user[field] = value
  44. elif operation_type == 'add_tag':
  45. tag = op['tag']
  46. if 'tags' not in processed_user:
  47. processed_user['tags'] = []
  48. if tag not in processed_user['tags']:
  49. processed_user['tags'].append(tag)
  50. elif operation_type == 'delete_tag':
  51. tag = op['tag']
  52. if 'tags' in processed_user and tag in processed_user['tags']:
  53. processed_user['tags'].remove(tag)
  54. elif operation_type == 'append_field':
  55. field = op['field']
  56. value = op['value']
  57. processed_user[field] = value
  58. else:
  59. logging.warning(f"Unknown operation: {operation_type}")
  61. processed_data.append(processed_user)
  63. if not dry_run:
  64. try:
  65. with open(output_file, 'w') as f:
  66. json.dump(processed_data, f, indent=4)
  67. logging.info(f"Successfully processed data and saved to: {output_file}")
  68. except IOError:
  69. logging.error(f"Error writing to output file: {output_file}")
  70. else:
  71. logging.info("Dry-run mode: No changes made to the output file.")
  74. if __name__ == '__main__':
  75. # Example Usage
  76. input_file = 'user_data.json'
  77. output_file = 'processed_user_data.json'
  79. # Sample user data (create a user_data.json file with this content)
  80. sample_data = [
  81. {"id": 1, "name": "Alice", "email": "alice@example.com", "tags": ["VIP"]},
  82. {"id": 2, "name": "Bob", "email": "bob@example.com"},
  83. {"id": 3, "name": "Charlie", "email": "charlie@example.com", "tags": ["regular"]}
  84. ]
  85. with open(input_file, 'w') as f:
  86. json.dump(sample_data, f, indent=4)
  88. operations = [
  89. {"operation": "update", "field": "email", "value": "alice.new@example.com"},
  90. {"operation": "add_tag", "tag": "premium"},
  91. {"operation": "append_field", "field": "city", "value": "New York"}
  92. ]
  94. batch_user_data_operations(input_file, output_file, operations, dry_run=False)

Add your comment