chore: super me

app.py

@@ -0,0 +1,76 @@
+import streamlit as st
+from node import LibrarySystem
+from helper import response_generator
+from helper import send_message
+
+library = LibrarySystem()
+
+# Streamed response emulator
+
+st.set_page_config(
+    page_title="🤗💬 Ana",
+    layout="wide"
+
+)
+
+st.warning('This is a prototype of a chatbot for Group 28 main algorithm. It is still under development even though the main logic has been completed and run inside terminal', icon="⚠️")
+
+
+# Initialize chat history
+if "messages" not in st.session_state:
+    st.session_state.messages = [{"role": "assistant", "content": "Let's start chatting! 👇. You can start with hello or hi"}]
+
+if 'app_state' not in st.session_state:
+    st.session_state['app_state'] = 'initial'
+
+if 'book' not in st.session_state:
+    st.session_state['book'] = None
+
+if 'user' not in st.session_state:
+    st.session_state['user'] = ''
+
+
+with st.sidebar:
+    st.title('🤗💬 Ana')
+    st.markdown('📖 Ana is a library management and book recommendation app designed to enhance accessibility and convenience.')
+    st.markdown('Convenient Access – Users can interact with Ana anytime, anywhere, to access library services from their comfort zones.')
+    st.markdown('Smart Book Borrowing – Users can reserve books and pick them up at their convenience, eliminating scheduling conflicts.')
+    st.markdown('Personalized Recommendations – The system suggests books based on user preferences, enhancing the reading experience.')
+    
+# Display chat messages from history on app rerun
+for message in st.session_state.messages:
+    with st.chat_message(message["role"]):
+        st.markdown(message["content"])
+
+# Accept user input
+if prompt := st.chat_input("What is up?"):
+    #initialize library system object
+    # Add user message to chat history
+    st.session_state.messages.append({"role": "user", "content": prompt})
+    # Display user message in chat message container
+    with st.chat_message("user"):
+        st.markdown(prompt)
+   
+    # Display assistant response in chat message container
+    with st.chat_message("assistant"):
+        # with st.spinner("Thinking..."):
+        if prompt in ["hi", "hello", "hey", "/start"]:
+            data = library.startMessage()
+        elif st.session_state['app_state'] == 'welcome' :    
+            data = library.login_or_register(prompt)
+        elif st.session_state['app_state'] == 'login_search':
+            data = library.loginSearch(prompt)
+        elif st.session_state['app_state'] == 'waiting_for_name' or st.session_state['app_state'] == 'waiting_for_id' :
+            data = library.handle_registration_input(prompt)
+        elif st.session_state['app_state'] == 'default':
+            data = library.main_menu(prompt)
+        elif st.session_state['app_state'] == 'search':
+            data = library.fetchbook(prompt)
+        elif st.session_state['app_state'] == 'borrow':
+            data = library.hasborrowed(st.session_state['book'], prompt)
+        elif st.session_state['app_state'] == 'borrowing':
+            library.borrow_book()
+        else:
+            send_message("Invalid input. Please try again.")
+        
+    # Add assistant response to chat history

helper.py

@@ -0,0 +1,25 @@
+import streamlit as st
+import random
+import time
+
+# def send_message(message):
+#     response = st.write_stream(response_generator(message))
+#     st.session_state.messages.append({"role": "assistant", "content": response})
+
+def send_message(message):
+        message_placeholder = st.empty()
+        full_response = ""
+        # Simulate stream of response with milliseconds delay
+        for chunk in message.split():
+            full_response += chunk + " "
+            time.sleep(0.05)
+            # Add a blinking cursor to simulate typing
+            message_placeholder.markdown(full_response + "▌")
+        message_placeholder.markdown(full_response, unsafe_allow_html=True)
+        st.session_state.messages.append({"role": "assistant", "content": full_response})
+
+
+def response_generator(response):
+    for word in response.split():
+        yield word + " "
+        time.sleep(0.05)

node.py

@@ -0,0 +1,297 @@
+from datetime import date
+from enum import Enum
+import pandas as pd
+from fuzzywuzzy import fuzz
+from tinydb import TinyDB, Query
+from enum import Enum
+from helper import send_message, response_generator
+import streamlit as st
+
+
+class Database:
+    def __init__(self, db_name):
+        self.db = TinyDB(db_name)
+        self.students = self.db.table("students")
+        self.User = Query()
+        self.trackBook = self.db.table("trackBook")
+
+    def createUser(self, data):
+      if not self.students.contains(self.User.id == data['id']):
+        self.students.insert(data)
+      else:
+        print("User with this ID already exists.")
+
+    def deleteUser(self, id, user_type):
+      self.students.remove(self.User.id == id)
+
+    def searchUser(self, id):
+      search = Query()
+      item = self.students.search(search.id == id )
+      return item[0] if item else None
+
+    def addBorrowedBook(self, data):
+      self.trackBook.insert(data)
+
+    def updateBorrowedBook(self, data):
+      self.trackBook.update(data)
+
+    def listBorrowedBooks(self):
+      print(self.trackBook.all())
+
+    def listBorrowedBooks(self, studentId):
+      search = Query()
+      items = self.trackBook.search(search.studentId == studentId )
+      return items
+
+    def searchBorrowedBookById(self, id):
+      search = Query()
+      item = self.trackBook.search(search.isbn == id )
+      return item[0] if item else None
+
+
+
+class RecommendationSystem:
+    def __init__(self):
+        self.books = pd.read_csv("books-2.csv")
+        self.books = self.books.drop(columns=['isbn10', 'subtitle', 'published_year', 'num_pages'])
+
+    def getBookByTitle(self, title):
+      return self.books[self.books['title'].str.contains(title, case=False)]
+
+    # def recommend_by_titles(self,book_titles, num_recommendations=5):
+    #   reader = Reader(rating_scale = (1, 10))
+    #   data = Dataset.load_from_df(books[["title","categories","average_rating"]], reader)
+    #   trainset = data.build_full_trainset()
+
+    #   model_svd = SVD()
+    #   model_svd.fit(trainset)
+    #   books['categories'] = books['categories'].astype(str)
+
+    #   similar_books = self.books[self.books['title'].apply(lambda x: any(fuzz.ratio(x, book_title) > 80 for book_title in book_titles))]
+
+    #   if similar_books.empty:
+    #     print("No books found with similar titles. Recommending books.")
+    #     return books.sort_values(by='average_rating', ascending=False).head(num_recommendations)
+
+    #   similar_categories = set()
+    #   for categories in similar_books['categories']:
+    #     similar_categories.update(categories.split(', '))
+
+    #   recommended_books = books[books['categories'].apply(lambda x: any(cat in similar_categories for cat in x.split(', ')))]
+
+    #   return recommended_books.head(num_recommendations)
+
+
+class LibrarySystem:
+    def __init__(self):
+        self.user = None
+        self.database = Database('db.json')
+        self.recommendation = RecommendationSystem()
+        
+
+    def display_menu(self, options):
+        for i, option in enumerate(options, 1):
+           return f"{i}. {option}"
+
+    def get_choice(self, options, incoming_msg=None):
+        while True:
+          try:
+            choice = int(incoming_msg)
+            if 1 <= choice <= len(options):
+              return choice
+            print("Invalid choice. Try again.")
+          except ValueError:
+            print("Invalid input. Enter a number.")
+
+    def startMessage(self):
+        message = '''
+            Welcome to our chatbot! I'm here to help you with your needs.
+            You can interact with me to learn about our services, get support, or provide feedback. 
+            Please choose one of the following options:\n\n
+
+            **1. Login**
+            
+            **2. Register**
+        '''
+        st.session_state.app_state = 'welcome'
+        send_message(message)
+        return message
+
+
+    def login_or_register(self, incoming_msg):
+        if self.get_choice(["Login", "Register"], incoming_msg) == 1:
+            st.session_state.app_state = 'login'
+            self.login(incoming_msg)
+        else:
+            self.register()
+
+    def login(self, incoming_msg=None):
+       if st.session_state.app_state == "login":
+        message = "Hi there! 😊 Please enter your ID to continue(eg. 10944444)"
+        response = st.write_stream(response_generator(message))
+        st.session_state.messages.append({"role": "assistant", "content": response})
+        st.session_state.app_state = 'login_search'
+
+ 
+       
+    def loginSearch(self, incoming_msg):
+        self.user = self.database.searchUser(incoming_msg)
+        print(self.user)
+        if self.user != None:
+            st.session_state.app_state = "default"
+            st.session_state.user = self.user["name"]
+            self.run_main_app()
+        else:
+            response = st.write_stream(response_generator("User not found"))
+            st.session_state.messages.append({"role": "assistant", "content": response})
+
+    def register(self):
+        send_message("Enter your name: ")
+        # response = st.write_stream(response_generator("enter your name: "))
+        # st.session_state.messages.append({"role": "assistant", "content": response})
+        # we need to set the state here to wait for the user to provide a username
+        st.session_state.app_state = "waiting_for_name"
+        
+
+    # new function to handle the user input for registration
+    def handle_registration_input(self, incoming_msg):
+        if st.session_state.app_state == "waiting_for_name":
+            self.user = incoming_msg
+            st.session_state.user = self.user
+            st.session_state.app_state = "waiting_for_id"
+            send_message("Enter your ID: ")
+        elif st.session_state.app_state == "waiting_for_id":
+            user_id = incoming_msg
+            self.user = st.session_state.user
+            self.database.createUser({"name": self.user, "id": user_id})
+            st.session_state.app_state = "default" # we set to default here when registration is done
+            self.user = self.database.searchUser(user_id)
+            self.run_main_app()
+
+
+    def run_main_app(self):
+      message = f"""Welcome {st.session_state.user.title()}! 😊\nPlease choose an option:
+                1️⃣ Search Book  
+                2️⃣ Borrow Book  
+                3️⃣ Return Book  
+                4️⃣ List Borrowed Books  
+                5️⃣ Recommend Book  
+                6️⃣ Exit  
+
+                Reply with the number of your choice.
+                """
+      send_message(message)
+
+    def main_menu(self, incoming_msg):
+          if incoming_msg in ["1", "2", "3", "4", "5", "6"]:
+            self.handle_student_choice(int(incoming_msg), incoming_msg)
+          else:
+            print(f"Invalid input. Enter a number. {incoming_msg}")
+            send_message("Please enter a valid number")
+   
+
+    def handle_student_choice(self, choice, incoming_msg):
+        if choice == 1:
+            self.search_book(incoming_msg)
+        else:
+           send_message("We are still working on this feature. Please try again later.")
+        # elif choice == 2:
+        #     self.borrow_book()
+        # elif choice == 3:
+        #     self.return_book()
+        # elif choice == 4:
+        #     self.list_borrowed_books()
+        # elif choice == 5:
+        #     self.recommendBook()
+
+    def search_book(self, incoming_msg):
+        message = f"📚 , please enter the title of the book:"
+        st.session_state.app_state = "search" # we set to default here when 
+        send_message(message)
+
+
+    def fetchbook(self, incoming_msg):
+        book = self.recommendation.getBookByTitle(incoming_msg)
+        st.session_state.book = book
+        if book.empty:
+            message = "❌ Book not found. Please check the title and try again."
+            send_message(message)
+            return
+        message = f'''
+        📖 Title: {book.iloc[0]["title"]}
+        📖 ISBN: {book.iloc[0]["isbn13"]}
+
+        📖 Borrow this book? (yes/no):
+        '''
+        send_message(message)
+        st.session_state.app_state = "borrow" # we set to default here when 
+
+    def hasborrowed(self, book, incoming_msg):
+        if incoming_msg.strip().lower() == "yes":
+            self.borrow_book_by_isbn(book.iloc[0]["isbn13"], book.iloc[0]["title"])
+
+    def borrow_book_by_isbn(self, isbn, title):
+        data = {
+            "title": title,
+            "isbn": str(isbn),
+            "studentId": st.session_state.user["id"],
+            "returned": False,
+            "borrowedDate": date.today().isoformat(),
+            "returnDate": ""
+        }
+        self.database.addBorrowedBook(data)
+        message = "✅ Book borrowed successfully!"
+        send_message(message)
+
+    def borrow_book(self):
+        send_message("Enter the title of the book to borrow: ")
+        if st.session_state.state == "borrowing":
+           isbn = self.incoming_msg
+           books = self.recommendation.getBookByTitle(isbn)
+           if books.empty:  # Check if the DataFrame is empty
+            send_message("Book not found.")
+            return
+           book_titles = books["title"].tolist()
+           send_message(book_titles)
+
+        choice = self.get_choice(book_titles)
+        book = books.iloc[choice - 1:choice]
+
+        self.borrow_book_by_isbn(book.iloc[0]["isbn13"], book.iloc[0]["title"])
+
+    def recommendBook(self):
+        borrowedBookByUser = self.database.listBorrowedBooks(self.user["id"])
+        print(borrowedBookByUser)
+        if not borrowedBookByUser:
+            print("No books borrowed, cannot recommend.")
+            return
+        book_titles = [book['title'] for book in borrowedBookByUser]
+        recommendations = self.recommendation.recommend_by_titles(book_titles)
+
+        print("Recommended Books:")
+        self.display_menu(recommendations['title'])
+        choice = self.get_choice(recommendations['title'])
+
+        shouldBorrow = input("Do you want to borrow this book? (yes/no): ")
+        if shouldBorrow == 'yes':
+          book = recommendations.iloc[choice - 1:choice]
+          self.borrow_book_by_isbn(book.iloc[0]["isbn13"], book.iloc[0]["title"])
+
+    def return_book(self):
+        isbn = input("Enter the ISBN of the book to return: ")
+        borrowed_book = self.database.searchBorrowedBookById(isbn)
+        if borrowed_book:
+            borrowed_book["returned"] = True
+            borrowed_book["actualReturnDate"] = date.today().isoformat()
+            self.database.updateBorrowedBook(borrowed_book)
+            print("Book returned successfully!")
+        else:
+            print("Book not found in borrowed list.")
+
+    def list_borrowed_books(self):
+        books = self.database.listBorrowedBooks(self.user["id"])
+        if books:
+            for book in books:
+                print(book)
+        else:
+            print("No books borrowed.")

requirements.txt

@@ -0,0 +1,13 @@
+streamlit
+python-decouple
+sqlalchemy
+psycopg2-binary
+numpy
+pandas
+python-multipart
+pyngrok
+# scikit-surprise
+fuzzywuzzy
+tinydb
+python-dotenv
+logging