watch.py 2.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596
  1. """General utility functions."""
  2. import contextlib
  3. import os
  4. import shutil
  5. import time
  6. from watchdog.events import FileSystemEvent, FileSystemEventHandler
  7. from watchdog.observers import Observer
  8. from reflex.constants import Dirs
  9. from reflex.utils.prerequisites import get_web_dir
  10. class AssetFolderWatch:
  11. """Asset folder watch class."""
  12. def __init__(self, root):
  13. """Initialize the Watch Class.
  14. Args:
  15. root: root path of the public.
  16. """
  17. self.path = str(root / Dirs.APP_ASSETS)
  18. self.event_handler = AssetFolderHandler(root)
  19. def start(self):
  20. """Start watching asset folder."""
  21. self.observer = Observer()
  22. self.observer.schedule(self.event_handler, self.path, recursive=True)
  23. self.observer.start()
  24. class AssetFolderHandler(FileSystemEventHandler):
  25. """Asset folder event handler."""
  26. def __init__(self, root):
  27. """Initialize the AssetFolderHandler Class.
  28. Args:
  29. root: root path of the public.
  30. """
  31. super().__init__()
  32. self.root = root
  33. def on_modified(self, event: FileSystemEvent):
  34. """Event handler when a file or folder was modified.
  35. This is called every time after a file is created, modified and deleted.
  36. Args:
  37. event: Event information.
  38. """
  39. dest_path = self.get_dest_path(event.src_path)
  40. # wait 1 sec for fully saved
  41. time.sleep(1)
  42. if os.path.isfile(event.src_path):
  43. with contextlib.suppress(PermissionError):
  44. shutil.copyfile(event.src_path, dest_path)
  45. if os.path.isdir(event.src_path):
  46. if os.path.exists(dest_path):
  47. shutil.rmtree(dest_path)
  48. with contextlib.suppress(PermissionError):
  49. shutil.copytree(event.src_path, dest_path)
  50. def on_deleted(self, event: FileSystemEvent):
  51. """Event hander when a file or folder was deleted.
  52. Args:
  53. event: Event infomation.
  54. """
  55. dest_path = self.get_dest_path(event.src_path)
  56. if os.path.isfile(dest_path):
  57. # when event is about a file, pass
  58. # this will be deleted at on_modified function
  59. return
  60. if os.path.exists(dest_path):
  61. shutil.rmtree(dest_path)
  62. def get_dest_path(self, src_path: str) -> str:
  63. """Get public file path.
  64. Args:
  65. src_path: The asset file path.
  66. Returns:
  67. The public file path.
  68. """
  69. return src_path.replace(
  70. str(self.root / Dirs.APP_ASSETS),
  71. str(self.root / get_web_dir() / Dirs.PUBLIC),
  72. )