1. import json
  2. import logging
  3. import os
  4. from datetime import datetime
  6. class MetadataLogger:
  7. def __init__(self, log_file="metadata_migration.log", version_mapping=None):
  8. self.log_file = log_file
  9. self.logger = logging.getLogger(__name__)
  10. self.logger.setLevel(logging.INFO)
  12. # Create a file handler
  13. file_handler = logging.FileHandler(log_file)
  14. formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
  15. file_handler.setFormatter(formatter)
  16. self.logger.addHandler(file_handler)
  18. self.version_mapping = version_mapping if version_mapping else {} # Dict: OldVersion: NewVersion
  20. def log_operation(self, operation_type, metadata, old_version=None, new_version=None):
  21. """Logs a metadata operation.
  23. Args:
  24. operation_type (str): The type of operation (e.g., 'migrate', 'transform', 'validate').
  25. metadata (dict): The metadata being operated on.
  26. old_version (str, optional): The version of the metadata before the operation. Defaults to None.
  27. new_version (str, optional): The version of the metadata after the operation. Defaults to None.
  28. """
  29. log_entry = {
  30. "operation_type": operation_type,
  31. "timestamp": datetime.now().isoformat(),
  32. "metadata": metadata,
  33. "old_version": old_version,
  34. "new_version": new_version
  35. }
  36. self.logger.info(json.dumps(log_entry))
  38. def register_version_mapping(self, old_version, new_version):
  39. """Registers a mapping between old and new metadata versions.
  41. Args:
  42. old_version (str): The old metadata version.
  43. new_version (str): The new metadata version.
  44. """
  45. self.version_mapping[old_version] = new_version
  46. self.logger.info(f"Registered version mapping: {old_version} -> {new_version}")
  48. def get_new_version(self, old_version):
  49. """Retrieves the new version for a given old version.
  51. Args:
  52. old_version (str): The old metadata version.
  54. Returns:
  55. str: The new metadata version, or None if no mapping exists.
  56. """
  57. return self.version_mapping.get(old_version)
  59. def load_metadata(filepath):
  60. """Loads metadata from a JSON file.
  62. Args:
  63. filepath (str): The path to the JSON metadata file.
  65. Returns:
  66. dict: The loaded metadata. Returns an empty dict if file not found.
  67. """
  68. try:
  69. with open(filepath, 'r') as f:
  70. metadata = json.load(f)
  71. return metadata
  72. except FileNotFoundError:
  73. print(f"Warning: Metadata file not found at {filepath}. Returning empty dictionary.")
  74. return {}
  75. except json.JSONDecodeError:
  76. print(f"Error decoding JSON from {filepath}. Returning empty dictionary.")
  77. return {}
  80. if __name__ == '__main__':
  81. # Example Usage
  82. logger = MetadataLogger()
  83. logger.register_version_mapping("v1.0", "v2.0")
  85. # Simulate a metadata migration
  86. metadata = load_metadata("metadata_v1.json") # Replace with your actual file
  88. if metadata:
  89. logger.log_operation("migrate", metadata, old_version="v1.0", new_version="v2.0")
  91. # Simulate a transformation
  92. transformed_metadata = {"name": "NewName", "value": 123}
  93. logger.log_operation("transform", transformed_metadata, old_version="v2.0")
  95. # Simulate a validation
  96. validated_metadata = {"valid": True}
  97. logger.log_operation("validate", validated_metadata, old_version="v2.0")
  99. # Retrieving the new version
  100. new_version = logger.get_new_version("v1.0")
  101. print(f"The new version for v1.0 is: {new_version}")

Add your comment