1. import json
  2. import time
  3. import re
  5. class ArtifactCleaner:
  6. def __init__(self, cleanup_interval=60):
  7. """
  8. Initializes the ArtifactCleaner.
  10. Args:
  11. cleanup_interval (int): Time in seconds to run cleanup. Defaults to 60 seconds.
  12. """
  13. self.cleanup_interval = cleanup_interval
  14. self.artifacts = [] # Store artifacts (e.g., response bodies)
  15. self.last_cleanup = 0
  17. def store_artifact(self, artifact):
  18. """
  19. Stores an artifact.
  21. Args:
  22. artifact (any): The artifact to store.
  23. """
  24. self.artifacts.append(artifact)
  26. def cleanup(self):
  27. """
  28. Cleans up artifacts older than the specified interval.
  29. """
  30. now = time.time()
  31. artifacts_to_remove = []
  32. for artifact in self.artifacts:
  33. if (now - artifact['timestamp']) > self.cleanup_interval:
  34. artifacts_to_remove.append(artifact)
  35. for artifact in artifacts_to_remove:
  36. self.artifacts.remove(artifact)
  37. print(f"Cleaned up {len(artifacts_to_remove)} artifacts.")
  39. def run(self):
  40. """
  41. Runs the cleanup process periodically.
  42. """
  43. while True:
  44. self.cleanup()
  45. self.last_cleanup = time.time()
  46. time.sleep(self.cleanup_interval)
  49. def clean_response(response_text):
  50. """
  51. Cleans a response text by removing sensitive information.
  53. Args:
  54. response_text (str): The raw response text.
  56. Returns:
  57. str: The cleaned response text.
  58. """
  59. # Remove potential API keys or sensitive data using regex
  60. response_text = re.sub(r'\b(API_KEY|password|secret)\b', '[REDACTED]', response_text, flags=re.IGNORECASE)
  61. # Remove JSON formatting (optional, for brevity)
  62. response_text = re.sub(r'\{.*\}', '', response_text)
  63. response_text = re.sub(r'\[.*\]', '', response_text)
  64. return response_text
  67. if __name__ == '__main__':
  68. cleaner = ArtifactCleaner(cleanup_interval=10) # Cleanup every 10 seconds
  70. # Simulate API calls and artifact storage
  71. for i in range(15):
  72. # Simulate an API response
  73. response_data = {"timestamp": time.time(), "data": f"This is some data {i}"}
  74. response_text = json.dumps(response_data)
  75. cleaned_response = clean_response(response_text)
  76. cleaner.store_artifact(response_text)
  77. time.sleep(2) # Simulate API call frequency
  79. # Start the cleanup process in the background
  80. cleanup_thread = threading.Thread(target=cleaner.run)
  81. cleanup_thread.daemon = True # Allow main thread to exit even if cleanup is running
  82. cleanup_thread.start()
  84. time.sleep(15) #Let the cleanup thread run for a bit
  85. print("Done.")

Add your comment