Spaces:
Sleeping
Sleeping
File size: 4,445 Bytes
806d445 | 1 2 3 4 5 6 7 8 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 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | """Collaborative filtering model for BookSage-AI."""
from typing import Any
import numpy as np
import pandas as pd
from scipy.sparse import csr_matrix
from sklearn.neighbors import NearestNeighbors
from app.core.config import Config
from app.core.logger import logger
class CollaborativeFilteringModel:
"""Collaborative filtering recommendation model using KNN."""
def __init__(self):
"""Initialize the collaborative filtering model."""
self.model: NearestNeighbors | None = None
self.book_pivot: pd.DataFrame | None = None
self.is_trained: bool = False
def train(self, final_rating: pd.DataFrame) -> bool:
"""
Train the collaborative filtering model.
Args:
final_rating: DataFrame with user ratings
Returns:
True if training successful, False otherwise
"""
try:
logger.info("Training collaborative filtering model...")
# Create user-item matrix
self.book_pivot = final_rating.pivot_table(
index="title",
columns="user_id",
values="rating"
).fillna(0)
book_sparse = csr_matrix(self.book_pivot.values)
# Build KNN model
self.model = NearestNeighbors(metric="cosine", algorithm="brute")
self.model.fit(book_sparse)
self.is_trained = True
logger.info("Collaborative filtering model trained successfully")
return True
except Exception as e:
logger.error(f"Error training collaborative filtering model: {e}")
self.is_trained = False
return False
def get_recommendations(
self,
book_title: str,
books_content: pd.DataFrame,
books: pd.DataFrame,
top_n: int = Config.DEFAULT_TOP_N
) -> list[dict[str, Any]]:
"""
Generate collaborative filtering recommendations.
Args:
book_title: Title of the book to get recommendations for
books_content: DataFrame with book content
books: DataFrame with all books
top_n: Number of recommendations to return
Returns:
List of recommendation dictionaries
"""
if not self.is_trained or self.model is None or self.book_pivot is None:
logger.warning("Model not trained yet")
return []
try:
if book_title not in self.book_pivot.index:
logger.warning(
f"Book '{book_title}' not found in collaborative filtering data"
)
return []
book_idx = np.where(self.book_pivot.index == book_title)[0][0]
distances, indices = self.model.kneighbors(
self.book_pivot.iloc[book_idx, :].values.reshape(1, -1),
n_neighbors=top_n + 1
)
recommendations = []
for i in range(1, len(indices.flatten())):
title = self.book_pivot.index[indices.flatten()[i]]
book_info = books_content[books_content["title"] == title]
if book_info.empty:
book_info = books[books["title"] == title]
if book_info.empty:
continue # pragma: no cover
book_info = book_info.iloc[0]
img_url = self._validate_image_url(book_info["img_url"])
recommendations.append({
"title": title,
"author": book_info["author"],
"year": book_info["year"],
"publisher": book_info["publisher"],
"image_url": img_url,
"score": float(1 - distances.flatten()[i]),
"type": "collaborative"
})
return recommendations[:top_n]
except Exception as e:
logger.error(f"Error in collaborative recommendations: {e}")
return []
def _validate_image_url(self, img_url: Any) -> str:
"""
Validate and return proper image URL.
Args:
img_url: Image URL to validate
Returns:
Valid image URL or default placeholder
"""
if not isinstance(img_url, str) or not img_url.startswith("http"):
return Config.DEFAULT_IMAGE_URL
return img_url
|