filetree.py 2.5 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283
  1. from typing import (
  2. Iterable,
  3. Optional,
  4. Tuple
  5. )
  6. from dataladmetadatamodel import JSONObject
  7. from dataladmetadatamodel.metadata import (
  8. ExtractorConfiguration,
  9. Metadata
  10. )
  11. from dataladmetadatamodel.mappableobject import MappableObject
  12. from dataladmetadatamodel.metadatapath import MetadataPath
  13. from dataladmetadatamodel.mtreenode import MTreeNode
  14. from dataladmetadatamodel.mtreeproxy import MTreeProxy
  15. from dataladmetadatamodel.mapper.reference import Reference
  16. class FileTree(MTreeProxy):
  17. def __init__(self,
  18. mtree: Optional[MTreeNode] = None,
  19. reference: Optional[Reference] = None):
  20. assert isinstance(mtree, (type(None), MTreeNode))
  21. assert isinstance(reference, (type(None), Reference))
  22. super().__init__(Metadata, mtree, reference)
  23. def add_metadata(self,
  24. path: MetadataPath,
  25. metadata: Metadata):
  26. self.mtree.add_child_at(metadata, path)
  27. def get_metadata(self,
  28. path: MetadataPath
  29. ) -> Optional[Metadata]:
  30. metadata = self.mtree.get_object_at_path(path)
  31. if metadata is None:
  32. return None
  33. assert isinstance(metadata, Metadata)
  34. metadata.ensure_mapped()
  35. return metadata
  36. def unget_metadata(self,
  37. metadata: Metadata,
  38. destination: Optional[str] = None):
  39. assert isinstance(metadata, Metadata)
  40. metadata.write_out(destination)
  41. metadata.purge()
  42. def add_extractor_run(self,
  43. path,
  44. time_stamp: Optional[float],
  45. extractor_name: str,
  46. author_name: str,
  47. author_email: str,
  48. configuration: ExtractorConfiguration,
  49. metadata_content: JSONObject):
  50. try:
  51. metadata = self.get_metadata(path)
  52. except AttributeError:
  53. metadata = Metadata()
  54. self.add_metadata(path, metadata)
  55. metadata.add_extractor_run(
  56. time_stamp,
  57. extractor_name,
  58. author_name,
  59. author_email,
  60. configuration,
  61. metadata_content
  62. )
  63. def get_paths_recursive(self,
  64. show_intermediate: Optional[bool] = False
  65. ) -> Iterable[Tuple[MetadataPath, MappableObject]]:
  66. yield from self.mtree.get_paths_recursive(show_intermediate)