Update main.py

main.py

@@ -16,15 +16,16 @@ logging.basicConfig(level=logging.INFO,
 
 logger = logging.getLogger(__name__)
 
-
 # MongoDB connection setup
 db_name = 'property-listing'
 collection_name = 'activities'
+user_recommendation_collection_name = 'user_recommendation_collection'
 connection_string = os.getenv('CONNECTION_STRING')
 
 client = MongoClient(connection_string)
 db = client[db_name]
 collection = db[collection_name]
+user_recommendation_collection = db[user_recommendation_collection_name]
 
 # Load pre-trained SVD model and user-item matrix columns
 svd = joblib.load('svd_model.joblib')
@@ -67,28 +68,30 @@ async def check_for_new_session():
             logger.error(f"Error in check_for_new_session: {e}")
             await asyncio.sleep(5)  # Wait before retrying
 
-def generate_recommendations_for_session(session_id):
+def get_session_data(session_id):
     try:
-        # Retrieve all documents for the given session
         session_data = list(collection.find({'sessionId': session_id}))
         if not session_data:
             logger.warning(f"No data found for session {session_id}")
             return None
 
-        # Convert session data to a DataFrame
         raw_df = pd.DataFrame(session_data)
-        
-        # Debug: Print column names
         logger.debug(f"Columns in raw_df: {raw_df.columns.tolist()}")
-        
-        # Check if required columns exist
+
         required_columns = ['id', 'action']
         missing_columns = [col for col in required_columns if col not in raw_df.columns]
         if missing_columns:
             logger.error(f"Missing required columns: {missing_columns}")
             return None
 
-        # Determine aggregation based on presence of 'duration' column
+        return raw_df
+
+    except Exception as e:
+        logger.error(f"Error in get_session_data: {str(e)}")
+        return None
+
+def create_pivot_table(raw_df):
+    try:
         if 'duration' in raw_df.columns:
             aggregated_data = raw_df.groupby(['id', 'action']).agg(
                 presence=('action', 'size'),
@@ -99,7 +102,6 @@ def generate_recommendations_for_session(session_id):
                 presence=('action', 'size')
             ).reset_index()
 
-        # Create pivot table
         pivot_columns = ['presence', 'total_duration'] if 'duration' in raw_df.columns else ['presence']
         pivot_df = aggregated_data.pivot_table(
             index=['id'],
@@ -108,45 +110,100 @@ def generate_recommendations_for_session(session_id):
             fill_value=0
         )
 
-        # Flatten column names
         pivot_df.columns = ['_'.join(col).strip() for col in pivot_df.columns.values]
 
-        # Ensure all expected columns exist in the pivot table
         for col in ALL_COLUMNS:
             if f'presence_{col}' not in pivot_df.columns and col != 'time_spent':
                 pivot_df[f'presence_{col}'] = 0
             elif col == 'time_spent' and 'duration' in raw_df.columns and 'total_duration_time_spent' not in pivot_df.columns:
                 pivot_df['total_duration_time_spent'] = 0
 
-        # Calculate interaction score for each row
+        return pivot_df
+
+    except Exception as e:
+        logger.error(f"Error in create_pivot_table: {str(e)}")
+        return None
+
+def create_user_vector(pivot_df):
+    try:
         pivot_df['interaction_score'] = pivot_df.apply(calculate_interaction_score, axis=1)
 
-        # Create a user vector based on the interaction scores
         user_vector = pd.Series(index=user_item_matrix_columns, dtype=float).fillna(0)
         for property_id, score in pivot_df['interaction_score'].items():
             if property_id in user_vector.index:
                 user_vector[property_id] = score
 
-        # Transform the user vector using the SVD model
+        return user_vector
+
+    except Exception as e:
+        logger.error(f"Error in create_user_vector: {str(e)}")
+        return None
+
+def generate_recommendations(user_vector):
+    try:
         user_vector_array = user_vector.values.reshape(1, -1)
         user_latent = svd.transform(user_vector_array)
 
-        # Calculate similarity scores between the user vector and item factors
         similarity_scores = cosine_similarity(user_latent, item_factors)
-
-        # Get the indices of the top 10 most similar items
         top_indices = similarity_scores.argsort()[0][-10:][::-1]
-
-        # Get the corresponding property IDs for the top indices
         recommendations = user_item_matrix_columns[top_indices].tolist()
 
         return recommendations
 
     except Exception as e:
-        logger.error(f"Error in generate_recommendations_for_session: {str(e)}")
-        logger.debug(f"Raw dataframe info: {raw_df.info()}")
+        logger.error(f"Error in generate_recommendations: {str(e)}")
         return None
 
+def generate_recommendations_for_session(session_id):
+    try:
+        raw_df = get_session_data(session_id)
+        if raw_df is None:
+            return None
+
+        pivot_df = create_pivot_table(raw_df)
+        if pivot_df is None:
+            return None
+
+        user_vector = create_user_vector(pivot_df)
+        if user_vector is None:
+            return None
+
+        recommendations = generate_recommendations(user_vector)
+
+        # Check if recommendations already exist for the session
+        existing_recommendations = user_recommendation_collection.find_one({"sessionId": session_id})
+        
+        if existing_recommendations:
+            # Compare the existing recommendations with the new recommendations
+            if existing_recommendations["recommendations"] != recommendations:
+                # Update the recommendations if they are different
+                recommendation_data = {
+                    "sessionId": session_id,
+                    "recommendations": recommendations,
+                    "timestamp": datetime.now()
+                }
+                user_recommendation_collection.update_one(
+                    {"sessionId": session_id},
+                    {"$set": recommendation_data}
+                )
+                logger.info(f"Updated recommendations for session {session_id}: {recommendations}")
+            else:
+                logger.info(f"Recommendations for session {session_id} remain unchanged")
+        else:
+            # Save the recommendations if they don't exist for the session
+            recommendation_data = {
+                "sessionId": session_id,
+                "recommendations": recommendations,
+                "timestamp": datetime.now()
+            }
+            user_recommendation_collection.insert_one(recommendation_data)
+            logger.info(f"Saved recommendations for session {session_id}: {recommendations}")
+
+        return recommendations
+
+    except Exception as e:
+        logger.error(f"Error in generate_recommendations_for_session: {str(e)}")
+        return None
 
 def calculate_interaction_score(row):
     try:
@@ -200,4 +257,3 @@ async def get_recommendations():
     else:
         logger.info("No recommendations available")
         return []
-