adds celery logging rotation

app/worker/main.py

@@ -14,6 +14,7 @@ from app.shared.settings import get_settings
 from app.shared.log import log_error
 from app.shared.aa_utils import get_all_urls
 from app.shared.db import worker_crud
+from app.worker.worker_log import setup_celery_logger
 
 settings = get_settings()
 
@@ -22,6 +23,8 @@ Redis = get_redis()
 
 USER_GROUPS_FILENAME = settings.USER_GROUPS_FILENAME
 
+setup_celery_logger(celery)
+
 # TODO: these are temporary PATCHES for new aa's functionality
 # logger.add("app/worker/worker_log.log", level="DEBUG") 
 logger.remove = lambda x: print(f"logger.remove({x})")

app/worker/worker_log.py

@@ -0,0 +1,29 @@
+from loguru import logger
+from celery import Celery
+import sys
+
+from app.shared.task_messaging import get_celery
+
+celery = get_celery("worker")
+
+def setup_celery_logger(celery):
+	# Remove Celery's default handlers to prevent duplicate logs
+	celery_logger = celery.log.get_default_logger()
+	for handler in celery_logger.handlers[:]:
+		celery_logger.removeHandler(handler)
+
+	# Set up Loguru logging
+	logger.add("logs/celery_logs.log", retention="30 days", level="DEBUG")
+	logger.add("logs/celery_error_logs.log", retention="30 days", level="ERROR")
+
+	# Redirect Celery logs to Loguru
+	class InterceptHandler:
+		def write(self, message):
+			if message.strip():
+				logger.info(message.strip())
+		# Required to prevent issues with buffered output
+		def flush(self): pass  
+		def isatty(self): return False
+
+	sys.stdout = InterceptHandler()
+	sys.stderr = InterceptHandler()