File size: 5,731 Bytes
47211ec | 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 136 137 138 139 140 141 142 143 144 | import os
import pickle
import streamlit as st
import requests
import time
from requests.exceptions import RequestException
# Initialize session state for pagination
if 'page' not in st.session_state:
st.session_state.page = 1
if 'similar_indices_list' not in st.session_state:
st.session_state.similar_indices_list = []
if 'selected_movie_index' not in st.session_state:
st.session_state.selected_movie_index = None
if 'loaded_recommendations' not in st.session_state:
st.session_state.loaded_recommendations = {'names': [], 'posters': []}
# --- Fetch Poster (cached for speed) ---
@st.cache_data(show_spinner=False)
def fetch_poster(movie_id):
url = f"https://api.themoviedb.org/3/movie/{movie_id}?api_key=8265bd1679663a7ea12ac168da84d2e8&language=en-US"
try:
response = requests.get(url, timeout=5)
response.raise_for_status()
data = response.json()
poster_path = data.get('poster_path')
if poster_path:
return "https://image.tmdb.org/t/p/w500/" + poster_path
else:
return "https://via.placeholder.com/300x450.png?text=No+Poster"
except RequestException as e:
print(f"Error fetching poster for movie ID {movie_id}: {e}")
return "https://via.placeholder.com/300x450.png?text=Error"
# --- Load Pickled Files ---
try:
movies = pickle.load(open('artifacts/movies.pkl', 'rb')).reset_index(drop=True)
similar_indices = pickle.load(open('artifacts/all_neighbors.pkl', 'rb'))
except FileNotFoundError:
st.error("Model files not found. Please make sure the 'artifacts' directory contains the required model files.")
st.stop()
# --- Get Similar Movie Indices ---
def get_similar_indices(movie):
if movie not in movies['title'].values:
return []
index = movies[movies['title'] == movie].index[0]
# Get up to 100 similar movies (excluding the selected movie itself)
return similar_indices[index][1:101]
# --- Load Batch of Recommendations ---
def load_recommendations_batch(indices, start_idx, batch_size=10):
"""Load a batch of movie recommendations"""
end_idx = min(start_idx + batch_size, len(indices))
names = []
posters = []
# Show a progress bar while loading
progress_bar = st.progress(0)
for i, movie_idx in enumerate(indices[start_idx:end_idx]):
movie_id = movies.iloc[movie_idx]['id']
names.append(movies.iloc[movie_idx]['title'])
posters.append(fetch_poster(movie_id))
# Update progress bar
progress = (i + 1) / (end_idx - start_idx)
progress_bar.progress(progress)
# Remove progress bar when done
progress_bar.empty()
return names, posters, end_idx
# --- Streamlit UI ---
st.markdown("<h1 style='text-align: center; color: red;'>CineML: Movie Recommender</h1>", unsafe_allow_html=True)
movie_list = movies['title'].values
selected_movie = st.selectbox("Type or Select a movie to get recommendations", movie_list)
# Reset pagination when a new movie is selected
if st.session_state.selected_movie_index is None or st.session_state.selected_movie_index != selected_movie:
st.session_state.page = 5
st.session_state.loaded_recommendations = {'names': [], 'posters': []}
if st.button('Show Recommendations'):
# Reset previous recommendations
st.session_state.loaded_recommendations = {'names': [], 'posters': []}
st.session_state.page = 1
# Get similar movie indices
st.session_state.similar_indices_list = get_similar_indices(selected_movie)
st.session_state.selected_movie_index = selected_movie
if len(st.session_state.similar_indices_list) > 0:
# Load first batch of recommendations
names, posters, _ = load_recommendations_batch(
st.session_state.similar_indices_list,
0,
batch_size=10
)
st.session_state.loaded_recommendations['names'] = names
st.session_state.loaded_recommendations['posters'] = posters
else:
st.error(f"No recommendations found for {selected_movie}")
# Display loaded recommendations
if len(st.session_state.loaded_recommendations['names']) > 0:
st.subheader(f"Recommendations for '{st.session_state.selected_movie_index}'")
# Display in a grid (5 movies per row)
total = len(st.session_state.loaded_recommendations['names'])
rows = (total + 4) // 5
for row in range(rows):
cols = st.columns(5)
for i in range(5):
idx = row * 5 + i
if idx < total:
with cols[i]:
st.text(st.session_state.loaded_recommendations['names'][idx])
st.image(st.session_state.loaded_recommendations['posters'][idx])
# Load more button
current_loaded = len(st.session_state.loaded_recommendations['names'])
similar_indices_length = len(st.session_state.similar_indices_list)
if current_loaded < similar_indices_length:
if st.button("Load More Recommendations"):
# Load next batch
names, posters, _ = load_recommendations_batch(
st.session_state.similar_indices_list,
current_loaded,
batch_size=10
)
# Append to existing recommendations
st.session_state.loaded_recommendations['names'].extend(names)
st.session_state.loaded_recommendations['posters'].extend(posters)
# Force a rerun to display the new recommendations
st.rerun()
|