Skip to content

Dynamic Filtering

DynamicFilteringRecommender

Source code in engines/contentFilterEngine/special_techniques/dynamic_filtering.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class DynamicFilteringRecommender:
    def __init__(self, base_recommender: Any):
        """
        Initialize the DynamicFilteringRecommender with a base recommender.

        Parameters:
        - base_recommender (Any): An instance of a base recommender (e.g., LSA, LDA).
        """
        self.base_recommender = base_recommender
        # Keeps track of items that have been added or removed
        self.added_items: List[int] = []
        self.removed_items: List[int] = []
        logger.info("DynamicFilteringRecommender initialized with base recommender.")

    def add_item(self, item_id: int, item_features: Dict[str, Any]):
        """
        Add a new item to the recommender system.

        Parameters:
        - item_id (int): The ID of the new item.
        - item_features (Dict[str, Any]): The features of the new item.
        """
        logger.info(f"Adding item {item_id} to the base recommender.")
        if hasattr(self.base_recommender, 'add_item'):
            self.base_recommender.add_item(item_id, item_features)
            self.added_items.append(item_id)
            logger.info(f"Item {item_id} added to the base recommender successfully.")
        else:
            logger.warning("Base recommender does not support adding items dynamically.")

    def remove_item(self, item_id: int):
        """
        Remove an existing item from the recommender system.

        Parameters:
        - item_id (int): The ID of the item to remove.
        """
        logger.info(f"Removing item {item_id} from the base recommender.")
        if hasattr(self.base_recommender, 'remove_item'):
            self.base_recommender.remove_item(item_id)
            self.removed_items.append(item_id)
            logger.info(f"Item {item_id} removed from the base recommender successfully.")
        else:
            logger.warning("Base recommender does not support removing items dynamically.")

    def update_item_features(self, item_id: int, new_features: Dict[str, Any]):
        """
        Update the features of an existing item.

        Parameters:
        - item_id (int): The ID of the item to update.
        - new_features (Dict[str, Any]): The updated features of the item.
        """
        try:
            if hasattr(self.base_recommender, 'update_item_features'):
                self.base_recommender.update_item_features(item_id, new_features)
                logger.info(f"Updated features for item {item_id} in the base recommender.")
            else:
                logger.warning("Base recommender does not support updating item features dynamically.")
        except Exception as e:
            logger.error(f"Error updating features for item {item_id}: {e}")

    def handle_data_change(self, event: Dict[str, Any]):
        """
        Handle dynamic data changes such as adding or removing items.

        Parameters:
        - event (Dict[str, Any]): A dictionary containing the type of event and relevant data.
          Example:
          {
              'action': 'add',
              'item_id': 123,
              'item_features': {'genre': 'Comedy', 'duration': 120}
          }
        """
        action = event.get('action')
        if action == 'add':
            self.add_item(event['item_id'], event['item_features'])
        elif action == 'remove':
            self.remove_item(event['item_id'])
        elif action == 'update':
            self.update_item_features(event['item_id'], event['item_features'])
        else:
            logger.warning(f"Unsupported event action: {action}")

    def recommend(self, user_id: int, query: str, top_n: int = 10) -> List[int]:
        """
        Generate top-N item recommendations for a user, considering dynamic changes.

        Parameters:
        - user_id (int): The ID of the user.
        - query (str): The query text for generating recommendations.
        - top_n (int): Number of top recommendations to return.

        Returns:
        - List[int]: List of recommended item IDs.
        """
        logger.info(f"Generating recommendations for user {user_id} with query '{query}' using DynamicFilteringRecommender.")
        return self.base_recommender.recommend(query, top_n=top_n)

__init__(base_recommender)

Initialize the DynamicFilteringRecommender with a base recommender.

Parameters: - base_recommender (Any): An instance of a base recommender (e.g., LSA, LDA).

Source code in engines/contentFilterEngine/special_techniques/dynamic_filtering.py 
10
11
12
13
14
15
16
17
18
19
20
21
def __init__(self, base_recommender: Any):
    """
    Initialize the DynamicFilteringRecommender with a base recommender.

    Parameters:
    - base_recommender (Any): An instance of a base recommender (e.g., LSA, LDA).
    """
    self.base_recommender = base_recommender
    # Keeps track of items that have been added or removed
    self.added_items: List[int] = []
    self.removed_items: List[int] = []
    logger.info("DynamicFilteringRecommender initialized with base recommender.")

add_item(item_id, item_features)

Add a new item to the recommender system.

Parameters: - item_id (int): The ID of the new item. - item_features (Dict[str, Any]): The features of the new item.

Source code in engines/contentFilterEngine/special_techniques/dynamic_filtering.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def add_item(self, item_id: int, item_features: Dict[str, Any]):
    """
    Add a new item to the recommender system.

    Parameters:
    - item_id (int): The ID of the new item.
    - item_features (Dict[str, Any]): The features of the new item.
    """
    logger.info(f"Adding item {item_id} to the base recommender.")
    if hasattr(self.base_recommender, 'add_item'):
        self.base_recommender.add_item(item_id, item_features)
        self.added_items.append(item_id)
        logger.info(f"Item {item_id} added to the base recommender successfully.")
    else:
        logger.warning("Base recommender does not support adding items dynamically.")

handle_data_change(event)

Handle dynamic data changes such as adding or removing items.

  • event (Dict[str, Any]): A dictionary containing the type of event and relevant data. Example: { 'action': 'add', 'item_id': 123, 'item_features': {'genre': 'Comedy', 'duration': 120} }
Source code in engines/contentFilterEngine/special_techniques/dynamic_filtering.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def handle_data_change(self, event: Dict[str, Any]):
    """
    Handle dynamic data changes such as adding or removing items.

    Parameters:
    - event (Dict[str, Any]): A dictionary containing the type of event and relevant data.
      Example:
      {
          'action': 'add',
          'item_id': 123,
          'item_features': {'genre': 'Comedy', 'duration': 120}
      }
    """
    action = event.get('action')
    if action == 'add':
        self.add_item(event['item_id'], event['item_features'])
    elif action == 'remove':
        self.remove_item(event['item_id'])
    elif action == 'update':
        self.update_item_features(event['item_id'], event['item_features'])
    else:
        logger.warning(f"Unsupported event action: {action}")

recommend(user_id, query, top_n=10)

Generate top-N item recommendations for a user, considering dynamic changes.

Parameters: - user_id (int): The ID of the user. - query (str): The query text for generating recommendations. - top_n (int): Number of top recommendations to return.

Returns: - List[int]: List of recommended item IDs.

Source code in engines/contentFilterEngine/special_techniques/dynamic_filtering.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def recommend(self, user_id: int, query: str, top_n: int = 10) -> List[int]:
    """
    Generate top-N item recommendations for a user, considering dynamic changes.

    Parameters:
    - user_id (int): The ID of the user.
    - query (str): The query text for generating recommendations.
    - top_n (int): Number of top recommendations to return.

    Returns:
    - List[int]: List of recommended item IDs.
    """
    logger.info(f"Generating recommendations for user {user_id} with query '{query}' using DynamicFilteringRecommender.")
    return self.base_recommender.recommend(query, top_n=top_n)

remove_item(item_id)

Remove an existing item from the recommender system.

Parameters: - item_id (int): The ID of the item to remove.

Source code in engines/contentFilterEngine/special_techniques/dynamic_filtering.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def remove_item(self, item_id: int):
    """
    Remove an existing item from the recommender system.

    Parameters:
    - item_id (int): The ID of the item to remove.
    """
    logger.info(f"Removing item {item_id} from the base recommender.")
    if hasattr(self.base_recommender, 'remove_item'):
        self.base_recommender.remove_item(item_id)
        self.removed_items.append(item_id)
        logger.info(f"Item {item_id} removed from the base recommender successfully.")
    else:
        logger.warning("Base recommender does not support removing items dynamically.")

update_item_features(item_id, new_features)

Update the features of an existing item.

Parameters: - item_id (int): The ID of the item to update. - new_features (Dict[str, Any]): The updated features of the item.

Source code in engines/contentFilterEngine/special_techniques/dynamic_filtering.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def update_item_features(self, item_id: int, new_features: Dict[str, Any]):
    """
    Update the features of an existing item.

    Parameters:
    - item_id (int): The ID of the item to update.
    - new_features (Dict[str, Any]): The updated features of the item.
    """
    try:
        if hasattr(self.base_recommender, 'update_item_features'):
            self.base_recommender.update_item_features(item_id, new_features)
            logger.info(f"Updated features for item {item_id} in the base recommender.")
        else:
            logger.warning("Base recommender does not support updating item features dynamically.")
    except Exception as e:
        logger.error(f"Error updating features for item {item_id}: {e}")