1. import os
  2. import time
  3. import datetime
  4. import shutil
  5. import logging
  6. def archive_log_files(log_dir, archive_dir, retry_interval=60):
  7. """
  8. Archives log files from a directory to an archive directory with retry.
  9. Args:
  10. log_dir (str): Path to the directory containing log files.
  11. archive_dir (str): Path to the directory where archived logs will be stored.
  12. retry_interval (int): Retry interval in seconds.
  13. """
  14. if not os.path.exists(archive_dir):
  15. os.makedirs(archive_dir)
  16. for filename in os.listdir(log_dir):
  17. if filename.endswith(".log"):
  18. log_filepath = os.path.join(log_dir, filename)
  19. archive_filepath = os.path.join(archive_dir, filename)
  20. retries = 0
  21. while retries < 3: # Retry up to 3 times
  22. try:
  23. shutil.move(log_filepath, archive_filepath)
  24. logging.info(f"Archived {filename} to {archive_dir}")
  25. break # Exit retry loop if successful
  26. except Exception as e:
  27. retries += 1
  28. logging.warning(f"Failed to archive {filename} (attempt {retries}/{3}): {e}")
  29. time.sleep(retry_interval)
  30. if retries == 3:
  31. logging.error(f"Failed to archive {filename} after multiple retries.")
  32. if __name__ == '__main__':
  33. # Example usage:
  34. log_directory = "/path/to/your/logs" # Replace with your log directory
  35. archive_directory = "/path/to/your/archive" # Replace with your archive directory
  36. #Create some dummy log files for testing
  37. if not os.path.exists(log_directory):
  38. os.makedirs(log_directory)
  39. with open(os.path.join(log_directory, "test_log1.log"), "w") as f:
  40. f.write("This is test log 1")
  41. with open(os.path.join(log_directory, "test_log2.log"), "w") as f:
  42. f.write("This is test log 2")
  43. archive_log_files(log_directory, archive_directory)

Add your comment