feat: Implement structured logging and replace prints

README.md

@@ -42,6 +42,7 @@ pandas >= 2.0.3
 numpy >= 1.24.4
 scikit-learn >= 1.3.2
 pickle >= 0.7.5
+rich >= 14.1.0
 optuna >= 3.6.1 (If you need automatic hyper-parameter tuning) 
 ```
 

main.py

@@ -7,6 +7,8 @@
 from config.model_param import model_specific_param
 from model import AVAILABLE_MODELS
 from util.databuilder import ColdStartDataBuilder
+import logging
+from util.logger import setup_logging
 
 class Config:
     """
@@ -37,16 +39,16 @@ def __init__(self, args: argparse.Namespace):
         warm_item_idx = data_info_dict['warm_item']
         cold_user_idx = data_info_dict['cold_user']
         cold_item_idx = data_info_dict['cold_item']
-        print(f"Dataset: {args.dataset}, User num: {user_num}, Item num: {item_num}.")
+        logger.info(f"Dataset: {args.dataset}, User num: {user_num}, Item num: {item_num}.")
 
         # Content obtaining
         user_content, item_content = None, None
         if args.cold_object == 'user':
             user_content = np.load(f'./data/{args.dataset}/{args.dataset}_{args.cold_object}_content.npy')
-            print(f'user content shape: {user_content.shape}')
+            logger.info(f'user content shape: {user_content.shape}')
         if args.cold_object == 'item':
             item_content = np.load(f'./data/{args.dataset}/{args.dataset}_{args.cold_object}_content.npy')
-            print(f'item content shape: {item_content.shape}')
+            logger.info(f'item content shape: {item_content.shape}')
 
         self.data = ColdStartDataBuilder(training_data, warm_valid_data, cold_valid_data, all_valid_data,
                                      warm_test_data, cold_test_data, all_test_data, user_num, item_num,
@@ -108,23 +110,24 @@ def parse_args() -> argparse.Namespace:
 
 if __name__ == '__main__':
     args = parse_args()
-    print(args)
+    log_path = setup_logging(dataset=args.dataset, model=args.model)
+    logger = logging.getLogger(__name__)
+    logger.info(args)
 
     config = Config(args)
-
     top_Ns = args.topN.split(',')
     results = {setting: {metric: [[] for _ in top_Ns] for metric in ['hit', 'precision', 'recall', 'ndcg']} for setting in ['all', 'cold', 'warm']}
 
     time_results = []
 
     for round_num in range(args.runs):
-        print(f"Start round {round_num} running!")
+        logger.info(f"Start round {round_num} running!")
 
         seed = args.seed if args.runs == 1 else round_num
         set_seed(seed, args.use_gpu)
 
         model = model_factory(config)
-        print(f"Registered model: {args.model}.")
+        logger.info(f"Registered model: {args.model}.")
 
         model.run()
 
@@ -139,9 +142,9 @@ def parse_args() -> argparse.Namespace:
         time_results.append((model.train_end_time - model.train_start_time) / args.epochs)
 
     for i, top_n in enumerate(top_Ns):
-        print("*" * 80)
+        logger.info("*" * 80)
         for setting_name, setting_key in [('Overall', 'all'), ('Cold-Start', 'cold'), ('Warm-Start', 'warm')]:
-            print(f"Top-{top_n} {setting_name} Test Performance:")
+            logger.info(f"Top-{top_n} {setting_name} Test Performance:")
 
             metrics = {
                 'Hit': (np.mean(results[setting_key]['hit'][i]), np.std(results[setting_key]['hit'][i])),
@@ -150,8 +153,8 @@ def parse_args() -> argparse.Namespace:
                 'NDCG': (np.mean(results[setting_key]['ndcg'][i]), np.std(results[setting_key]['ndcg'][i]))
             }
 
-            print(', '.join([f"{name}@{top_n}: {mean:.4f}±{std:.4f}" for name, (mean, std) in metrics.items()]))
+            logger.info(', '.join([f"{name}@{top_n}: {mean:.4f}±{std:.4f}" for name, (mean, std) in metrics.items()]))
 
-    print(f"Efficiency Performance:")
+    logger.info(f"Efficiency Performance:")
     mean_time, std_time = np.mean(time_results), np.std(time_results)
-    print(f"Time: {mean_time:.4f}±{std_time:.4f} seconds per epoch.")
+    logger.info(f"Time: {mean_time:.4f}±{std_time:.4f} seconds per epoch.")

model/BaseRecommender.py

@@ -7,7 +7,9 @@
 from util.evaluator import ranking_evaluation
 import time
 from util.utils import process_bar
+import logging
 
+logger = logging.getLogger(__name__)
 
 class BaseColdStartTrainer(ABC):
     """
@@ -41,14 +43,14 @@ def __init__(self, config):
             self.max_early_stop_patience = self.args.early_stop
 
     def print_basic_info(self):
-        print('*' * 80)
-        print('Model: ', self.model_name)
-        print('Dataset: ', self.dataset_name)
-        print('Embedding Dimension:', self.emb_size)
-        print('Maximum Epoch:', self.maxEpoch)
-        print('Learning Rate:', self.lr)
-        print('Batch Size:', self.batch_size)
-        print('*' * 80)
+        logger.info('*' * 80)
+        logger.info(f'Model: {self.model_name}')
+        logger.info(f'Dataset: {self.dataset_name}')
+        logger.info(f'Embedding Dimension: {self.emb_size}')
+        logger.info(f'Maximum Epoch: {self.maxEpoch}')
+        logger.info(f'Learning Rate: {self.lr}')
+        logger.info(f'Batch Size: {self.batch_size}')
+        logger.info('*' * 80)
 
     def timer(self, start=True):
         if start:
@@ -200,8 +202,8 @@ def full_evaluation(self, rec_list: Dict[str, List[Tuple[str, float]]], test_typ
             self.cold_test_results = test_performance
         elif test_type == 'all':
             self.overall_test_results = test_performance
-        print('*' * 80)
-        print(f'[{test_type} setting] The result of %s:\n%s' % (self.model_name, ''.join(self.result)))
+        logger.info('*' * 80)
+        logger.info(f"[{test_type} setting] The result of {self.model_name}:\n{''.join(self.result)}")
 
     def fast_evaluation(self, epoch: int, valid_type: str = 'all') -> List[str]:
         """
@@ -222,7 +224,7 @@ def fast_evaluation(self, epoch: int, valid_type: str = 'all') -> List[str]:
             valid_set = self.data.overall_valid_set
         else:
             raise ValueError('Invalid evaluation type!')
-        print(f'Evaluating the model under the {valid_type} setting...')
+        logger.info(f'Evaluating the model under the {valid_type} setting...')
         rec_list = self.valid(valid_type)
         measure, _ = ranking_evaluation(valid_set, rec_list, [self.max_N])
         if len(self.bestPerformance) > 0:
@@ -253,24 +255,24 @@ def fast_evaluation(self, epoch: int, valid_type: str = 'all') -> List[str]:
                 performance[k] = float(v)
             self.bestPerformance.append(performance)
             self.save()
-        print('-' * 120)
-        print('Performance ' + ' (Top-' + str(self.max_N) + ' Recommendation)')
+        logger.info('-' * 120)
+        logger.info('Performance ' + ' (Top-' + str(self.max_N) + ' Recommendation)')
         measure = [m.strip() for m in measure[1:]]
-        print('*Current Performance*')
-        print('Epoch:', str(epoch + 1) + ',', '  |  '.join(measure))
+        logger.info('*Current Performance*')
+        logger.info(f"Epoch:{epoch + 1}, {'  |  '.join(measure)}")
         bp = ''
         bp += 'Hit Ratio' + ':' + str(self.bestPerformance[1]['Hit Ratio']) + '  |  '
         bp += 'Precision' + ':' + str(self.bestPerformance[1]['Precision']) + '  |  '
         bp += 'Recall' + ':' + str(self.bestPerformance[1]['Recall']) + '  |  '
         bp += 'NDCG' + ':' + str(self.bestPerformance[1]['NDCG'])
-        print(f'*Best {valid_type} Performance* ')
-        print('Epoch:', str(self.bestPerformance[0]) + ',', bp)
+        logger.info(f'*Best {valid_type} Performance* ')
+        logger.info(f'Epoch:{str(self.bestPerformance[0])}, {bp}')
         if self.early_stop_flag:
             if self.early_stop_patience <= 0:
-                print(f"Stopping early at epoch {epoch + 1}.")
+                logger.info(f"Stopping early at epoch {epoch + 1}.")
             else:
-                print(f"Early stopping patience left: {self.early_stop_patience}.")
-        print('-' * 120)
+                logger.info(f"Early stopping patience left: {self.early_stop_patience}.")
+        logger.info('-' * 120)
         return measure
 
     def run(self) -> None:
@@ -281,11 +283,11 @@ def run(self) -> None:
         training, testing on different settings, and evaluation.
         """
         self.print_basic_info()
-        print('Training Model...')
+        logger.info('Training Model...')
         self.train()
         for test_type in ['all', 'cold', 'warm']:
-            print('*' * 80)
-            print(f'Testing under [{test_type}] setting...')
+            logger.info('*' * 80)
+            logger.info(f'Testing under [{test_type}] setting...')
             rec_list = self.test(test_type=test_type)
-            print(f'Evaluating under [{test_type}] setting...')
+            logger.info(f'Evaluating under [{test_type}] setting...')
             self.full_evaluation(rec_list, test_type=test_type)

model/GAR.py

@@ -2,7 +2,9 @@
 import torch.nn as nn
 from .BaseRecommender import BaseColdStartTrainer
 from util.utils import next_batch_pairwise, bpr_loss, l2_reg_loss, mse_loss
+import logging
 
+logger = logging.getLogger(__name__)
 
 # Following the source code process: https://github.com/zfnWong/GAR
 class GAR(BaseColdStartTrainer):
@@ -33,7 +35,7 @@ def train(self):
                 batch_loss.backward()
                 optimizer.step()
                 if n % 50 == 0:
-                    print('training:', epoch + 1, 'batch', n, 'batch_loss:', batch_loss.item())
+                    logger.info(f'training:{epoch + 1} batch: {n} batch_loss:{batch_loss.item()}')
 
             with torch.no_grad():
                 model.eval()

util/logger.py

@@ -0,0 +1,157 @@
+import copy
+import logging
+import logging.config
+import os
+import sys
+import time
+
+import torch
+from rich.text import Text
+from rich.traceback import install as install_rich_tracebacks
+
+# ==============================================================================
+# CONFIGURATION
+# ==============================================================================
+
+
+# Note the special '()' key. It tells dictConfig to import and instantiate
+# this class. The path must be importable from where the app runs.
+# Assuming this file is in 'utils/logger.py', the path is correct.
+class RemoveRichMarkupFilter(logging.Filter):
+    """A logging filter to remove rich markup from log messages."""
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        record.msg = Text.from_markup(record.getMessage()).plain
+        record.args = ()
+        return True
+
+
+LOGGING_CONFIG = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "filters": {
+        "remove_markup": {
+            "()": __name__ + ".RemoveRichMarkupFilter",
+        }
+    },
+    "formatters": {
+        "plain_text": {
+            "format": "%(message)s",
+        },
+    },
+    "handlers": {
+        "console": {
+            "class": "rich.logging.RichHandler",
+            "level": "INFO",
+            "rich_tracebacks": True,
+            "markup": True,
+            "show_path": False,
+            "show_level": False,
+            "show_time": False,
+        },
+        "file": {
+            "class": "logging.FileHandler",
+            "level": "INFO",
+            "filename": "result.log",
+            "mode": "a",
+            "encoding": "utf-8",
+            "formatter": "plain_text",
+            "filters": ["remove_markup"],
+        },
+    },
+    "root": {
+        "level": "DEBUG",
+        "handlers": ["console", "file"],
+    },
+}
+
+
+# ==============================================================================
+# SETUP LOGIC
+# ==============================================================================
+
+_logger_initialized = False
+
+
+def setup_logging(
+    dataset: str,
+    model: str,
+    no_log: bool = False,
+    log_dir: str = "./log",
+    console_log_level: int = logging.INFO,
+    file_log_level: int = logging.INFO,
+):
+    """
+    Configures the logging system using a dictionary configuration.
+    This should be called once at the start of the application.
+
+    Returns:
+        The path to the log subdirectory if successful, otherwise None.
+    """
+    global _logger_initialized
+    if _logger_initialized:
+        logging.warning("Logger has already been initialized.")
+        return None
+
+    if no_log:
+        logging.disable(logging.CRITICAL + 1)
+        return None
+
+    try:
+        config = copy.deepcopy(LOGGING_CONFIG)
+
+        timestamp = time.strftime("%Y-%m%d-%H%M", time.localtime())
+        log_subdir = os.path.join(log_dir, dataset, model, timestamp)
+        os.makedirs(log_subdir, exist_ok=True)
+        log_filename = os.path.join(log_subdir, "result.log")
+
+        config["handlers"]["file"]["filename"] = log_filename
+        config["handlers"]["console"]["level"] = logging.getLevelName(console_log_level)
+        config["handlers"]["file"]["level"] = logging.getLevelName(file_log_level)
+
+        logging.config.dictConfig(config)
+
+        # Actions that are not part of dictConfig (like global hooks)
+        install_rich_tracebacks(show_locals=True)
+        sys.excepthook = handle_exception
+
+        _logger_initialized = True
+        logging.info(f"Logger initialized. Log file at: {log_filename}")
+        return log_subdir
+
+    except Exception as e:
+        print(f"Failed to initialize logger: {e}", file=sys.stderr)
+        import traceback
+
+        traceback.print_exc(file=sys.stderr)
+        return None
+
+
+def handle_exception(exc_type, exc_value, exc_traceback):
+    """A global exception hook to log unhandled exceptions before exiting."""
+    if issubclass(exc_type, KeyboardInterrupt):
+        sys.__excepthook__(exc_type, exc_value, exc_traceback)
+        return
+
+    if _logger_initialized:
+        logging.critical("Unhandled exception occurred:", exc_info=(exc_type, exc_value, exc_traceback))
+    else:
+        print("Unhandled exception occurred (logger not initialized):", file=sys.stderr)
+        sys.__excepthook__(exc_type, exc_value, exc_traceback)
+
+
+# ==============================================================================
+# TENSORBOARD UTILITY FUNCTION
+# ==============================================================================
+
+
+def log_training_details_to_tensorboard(writer, step, data_dict):
+    """Logs metrics to a TensorBoard SummaryWriter instance."""
+    for key, value in data_dict.items():
+        if isinstance(value, torch.Tensor):
+            if value.dim() == 0:
+                writer.add_scalar(key, value.item(), step)
+            else:
+                writer.add_histogram(key, value, step)
+        elif isinstance(value, (float, int)):
+            writer.add_scalar(key, value, step)