1. import os
  2. import schedule
  3. import time
  4. import logging
  6. # Configure logging
  7. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  9. def check_directories(directories):
  10. """
  11. Checks if the specified directories exist.
  12. Returns a dictionary of errors, or an empty dictionary if all directories exist.
  13. """
  14. errors = {}
  15. for directory in directories:
  16. if not os.path.isdir(directory):
  17. errors[directory] = "Directory does not exist"
  18. elif not os.access(directory, os.R_OK):
  19. errors[directory] = "Directory is not readable"
  20. return errors
  22. def dry_run_schedule(directories, job_function, schedule_time):
  23. """
  24. Performs a dry-run of the schedule, checking directories for errors.
  25. """
  26. errors = check_directories(directories)
  27. if errors:
  28. logging.warning("Directory errors detected during dry-run:")
  29. for directory, error in errors.items():
  30. logging.warning(f" {directory}: {error}")
  31. print("Dry-run aborted due to directory errors.")
  32. return False # Indicate dry-run failure
  33. else:
  34. logging.info("Dry-run successful. All directories exist and are readable.")
  35. print("Dry-run successful. All directories exist and are readable.")
  36. return True # Indicate dry-run success
  39. def scheduled_job(job_function, schedule_time):
  40. """
  41. The actual scheduled job.
  42. """
  43. try:
  44. job_function()
  45. logging.info(f"Job executed successfully at {time.strftime('%Y-%m-%d %H:%M:%S')}")
  46. except Exception as e:
  47. logging.error(f"Job failed: {e}")
  51. def main():
  52. """
  53. Main function to set up and run the dry-run and scheduling.
  54. """
  56. directories_to_check = ["/path/to/directory1", "/path/to/directory2", "/path/to/directory3"] # Replace with your directory paths
  57. schedule_time = "0 * * * *" # Run every minute (example)
  59. # Define a sample job function - replace with your actual job
  60. def my_job():
  61. print("Running my scheduled job...")
  62. time.sleep(5) # Simulate some work
  63. print("Job completed.")
  66. if dry_run_schedule(directories_to_check, my_job, schedule_time):
  67. schedule.every().minute.do(scheduled_job, my_job, schedule_time) #uncomment if you want to run the scheduled job
  68. while True:
  69. schedule.run_pending()
  70. time.sleep(1)
  74. if __name__ == "__main__":
  75. main()

Add your comment