app.py

from typing import Dict, Text
import numpy as np
import tensorflow as tf
import pandas as pd
import pickle
import numpy as np
import tensorflow as tf
import tensorflow_recommenders as tfrs
import streamlit as st
from html_information import html


import pandas as pd
import json

def read_json(file_name):
    with open(file_name) as json_file:
        data = json.load(json_file)
    return data

uid_name_map = read_json('uid_name_map.json')
uid_url_map = read_json('uid_url_map.json')

print(uid_name_map)
print(uid_url_map)
st.set_page_config(page_title="My App", page_icon=":guardsman:", layout="wide", initial_sidebar_state="auto")

@st.cache_resource
def load_model(path):
    loaded = tf.saved_model.load(path)
    return loaded


def inference(model, user_id):
    scores, titles = model([user_id])
    recs = titles[0, :15]
    extracted_rec = []
    for rec in recs:
        extracted_rec.append(int(rec.numpy().decode('utf-8')))
    return extracted_rec
    
    
def read_pickle_file(file_path):
    with open(file_path, 'rb') as f:
        data = pickle.load(f)
    return data

def streamlit_carousel(header_name: str, rec_item_url: list,
                       rec_item_name: list) -> None:
    st.header(header_name)
    mid_section = ""
    for index, value in enumerate(rec_item_url):
        mid_section += """<div class="item"><div id="image-container"><img src='""" + str(value) + """' /></div><p>""" + str(rec_item_name[index]) + """</p></div>"""

    mid_html = html+mid_section + """</div></div></body>"""
    st.markdown(mid_html, unsafe_allow_html=True)

def recall_at_k(ground_truth, recommended, k):
    """
    Calculate Recall@k.

    Parameters:
    - ground_truth (list): List of ground truth product IDs.
    - recommended (list): List of recommended product IDs.
    - k (int): Number of recommendations to consider.

    Returns:
    - recall (float): Recall@k value.
    """
    # Take only the top-k recommended items
    recommended_at_k = set(recommended[:k])
    
    # Count the number of relevant items in the ground truth
    relevant_items = set(ground_truth)
    
    # Calculate the intersection (number of relevant items in top-k)
    intersection = recommended_at_k.intersection(relevant_items)
    
    # Calculate Recall@k
    recall = len(intersection) / len(relevant_items) if len(relevant_items) > 0 else 0.0
    
    return recall




model_weights_name = 'gofynd_old_model.model'
k = 15
print("######## Running ########")
print(f"model_weights_name: {model_weights_name}")
print('########')
print()
loaded = load_model(model_weights_name)
print("######### Model Loaded #########")


# uid_name_map = read_pickle_file('new_uid_name_map.pkl')
# uid_url_map = read_pickle_file('new_uid_url_map.pkl')
# uid_url_map = 
user_product_dict = read_pickle_file('user_product_dict.pkl')
last_session_user_product_dict = read_pickle_file('final_sessions_fynd_pickle_filename.pkl')
user_with_multiple_sessions = read_pickle_file('users_with_multiple_sessions_filename.pkl')
initial_sessions_user_product_dict = read_pickle_file('initial_sessions_fynd_pickle_filename.pkl')
# avg_recall = read_pickle_file('Personalised_two_tower_fynd_recall.pkl')
# positive_recall = read_pickle_file("Personalised_two_twoer_fynd_positive_recall.pkl")
# total_count = read_pickle_file("Personalised_two_twoer_fynd_total_count.pkl")
# average_positive_recall = read_pickle_file("Personalised_two_twoer_fynd_average_positive_recall.pkl")
user_id_list = user_with_multiple_sessions

user_id_list.append("000000000000000000000004")
last_session_user_product_dict["000000000000000000000004"] = []
initial_sessions_user_product_dict["000000000000000000000004"] = []

# # st.set_page_config(page_title="My App", page_icon=":guardsman:", layout="wide", initial_sidebar_state="auto")
# st.header("Personalised Product Recommendations (Fynd)")
# st.subheader("Training Metrics")
# st.write(f"Average Recall@{k} on Test Set: {avg_recall}")
# st.write(f"Total Users Count: {total_count}")
# st.write(f"Users with Positive Recall@{k} on Test Set: {positive_recall}")
# st.write(f"% Users with Positive Recall@{k} on Test Set: {average_positive_recall}")
# col1, col2 = st.tabs(["Training & Test Loss", "Top 10 Test Accuracy"])
# with col1:
#     st.image('Personalised_two_tower_fynd_loss_graph.png')
# with col2:
#     st.image('Personalised_two_tower_fynd_top_10_accuracy_graph.png')
    
st.header("Personalised Product Recommendations")
st.write("Model trained with Clickstream data of GoFynd.com")
st.subheader("Choose a User")
index = st.selectbox("User List", range(len(user_id_list)), format_func=lambda x: user_id_list[x])
user_id = user_id_list[index]
print(f"User ID: {user_id}")


user_final_session = last_session_user_product_dict[user_id]
final_session_product_list = []
for all_session in user_final_session:
    for session in all_session:
        final_session_product_list.append(session['product_id'])

rec_list = inference(loaded, str(user_id))
print(f"Final Session Product List: {final_session_product_list}")
print(f"Recommendation List: {rec_list}")
recall_value = recall_at_k(final_session_product_list, rec_list, k)
print(f"Recall@{k}: {recall_value}")
st.write(f"Recommendation Score: {recall_value}")

initial_sessions = initial_sessions_user_product_dict[user_id]

tab1, tab2, tab3 = st.tabs(["Recommendations", "Test session data", "Train session data"])
with tab1:
    # print(product_id, str(product_id))
    print(rec_list)
    rec_list_name = [uid_name_map[str(product_id)] for product_id in rec_list]
    rec_list_url = [uid_url_map[str(product_id)] for product_id in rec_list]
    streamlit_carousel("Top 15 Personalised Product Recommendation", rec_list_url, rec_list_name)
with tab2:
    product_name_list = [uid_name_map[str(product_id)] for product_id in final_session_product_list]
    product_url_list = [uid_url_map[str(product_id)] for product_id in final_session_product_list]
    streamlit_carousel("User's Test Last Session Viewed Products", product_url_list, product_name_list)
with tab3:
    i=1
    for session in initial_sessions:
        temp_product_list = []
        for row in session:
            temp_product_list.append(row['product_id'])
        product_name_list = [uid_name_map[str(product_id)] for product_id in temp_product_list]
        product_url_list = [uid_url_map[str(product_id)] for product_id in temp_product_list]
        streamlit_carousel("Session "+str(i), product_url_list, product_name_list)
        i+=1