1. import time
  2. import os
  3. import logging
  4. import argparse
  5. # Configure logging
  6. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  7. def monitor_directory(directories, interval, edge_cases=False):
  8. """
  9. Monitors the state of specified directories for changes.
  10. Args:
  11. directories (list): A list of directory paths to monitor.
  12. interval (int): The time interval (in seconds) between checks.
  13. edge_cases (bool): If True, includes extra checks for empty directories
  14. or non-existent directories.
  15. """
  16. if edge_cases:
  17. all_dirs_exist = True
  18. for dir_path in directories:
  19. if not os.path.exists(dir_path):
  20. logging.warning(f"Directory {dir_path} does not exist.")
  21. all_dirs_exist = False
  22. if not all_dirs_exist:
  23. logging.info("Exiting due to missing directories.")
  24. return
  25. while True:
  26. for dir_path in directories:
  27. try:
  28. current_state = os.listdir(dir_path)
  29. previous_state = None # Initialize for the first iteration
  30. if previous_state is None:
  31. logging.info(f"Initial state of {dir_path}: {current_state}")
  32. else:
  33. new_items = [item for item in current_state if item not in previous_state]
  34. removed_items = [item for item in previous_state if item not in current_state]
  35. if new_items:
  36. logging.info(f"New items in {dir_path}: {new_items}")
  37. if removed_items:
  38. logging.info(f"Removed items from {dir_path}: {removed_items}")
  39. # Handle empty directory edge case
  40. if not current_state and edge_cases:
  41. logging.warning(f"{dir_path} is empty.")
  42. previous_state = current_state # Update for the next iteration
  43. except OSError as e:
  44. logging.error(f"Error accessing {dir_path}: {e}")
  45. time.sleep(interval)
  46. if __name__ == "__main__":
  47. parser = argparse.ArgumentParser(description="Monitor directories for changes.")
  48. parser.add_argument("directories", nargs="+", help="List of directories to monitor.")
  49. parser.add_argument("-i", "--interval", type=int, default=5, help="Interval between checks (seconds).")
  50. parser.add_argument("-e", "--edge_cases", action="store_true", help="Include extra checks (empty/missing dirs).")
  51. args = parser.parse_args()
  52. monitor_directory(args.directories, args.interval, args.edge_cases)

Add your comment