import streamlit as st
from openai import OpenAI
from PIL import Image
import time
import os
import uuid
import firebase_admin
from firebase_admin import credentials, firestore
import re
# 🔐 Firebase setup
if not firebase_admin._apps:
cred = credentials.Certificate("firebase-service-account.json")
firebase_admin.initialize_app(cred)
db = firestore.client()
# 🔐 OpenAI setup
openai_key = os.getenv("openai_key")
assistant_id = os.getenv("ASSISTANT_ID")
client = OpenAI(api_key=openai_key)
# 🌐 Streamlit Config
st.set_page_config(page_title="Carfind.co.za AI Assistant", layout="wide")
# 🎯 Session + User ID
if "user_id" not in st.session_state:
st.session_state["user_id"] = str(uuid.uuid4())
user_id = st.session_state["user_id"]
# 🖼️ Branding + Styling
st.markdown("""
""", unsafe_allow_html=True)
st.markdown("""
Powered by Carfind
""", unsafe_allow_html=True)
# 🔁 Firebase Chat Functions
def get_or_create_thread_id():
doc_ref = db.collection("users").document(user_id)
doc = doc_ref.get()
if doc.exists:
return doc.to_dict()["thread_id"]
else:
thread = client.beta.threads.create()
doc_ref.set({
"thread_id": thread.id,
"created_at": firestore.SERVER_TIMESTAMP
})
return thread.id
def save_message(role, content):
db.collection("users").document(user_id).collection("messages").add({
"role": role,
"content": content,
"timestamp": firestore.SERVER_TIMESTAMP
})
def display_chat_history():
messages = db.collection("users").document(user_id).collection("messages") \
.order_by("timestamp").stream()
assistant_icon_html = ""
for msg in list(messages)[::-1]:
data = msg.to_dict()
if data["role"] == "user":
st.markdown(
f""
f"👤 You: {data['content']}
", unsafe_allow_html=True
)
else:
st.markdown(
f""
f"{assistant_icon_html} Carfind Assistant: {data['content']}
",
unsafe_allow_html=True
)
# 🛍 Tabs: AI Chat | Car Identifier
tab1, tab2 = st.tabs(["AI Chat", "What car is that?"])
# 🤖 AI Chat Tab
with tab1:
input_col, clear_col = st.columns([9, 1])
with input_col:
user_input = st.chat_input("Type your message here...")
with clear_col:
if st.button("🗑️", key="clear-chat", help="Clear Chat"):
try:
user_doc_ref = db.collection("users").document(user_id)
for msg in user_doc_ref.collection("messages").stream():
msg.reference.delete()
user_doc_ref.delete()
st.session_state.clear()
st.rerun()
except Exception as e:
st.error(f"Failed to clear chat: {e}")
thread_id = get_or_create_thread_id()
display_chat_history()
if user_input:
client.beta.threads.messages.create(thread_id=thread_id, role="user", content=user_input)
save_message("user", user_input)
with st.spinner("Thinking and typing... 💭"):
run = client.beta.threads.runs.create(thread_id=thread_id, assistant_id=assistant_id)
while True:
run_status = client.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run.id)
if run_status.status == "completed":
break
time.sleep(1)
messages_response = client.beta.threads.messages.list(thread_id=thread_id)
latest_response = sorted(messages_response.data, key=lambda x: x.created_at)[-1]
assistant_message = latest_response.content[0].text.value
save_message("assistant", assistant_message)
time.sleep(0.5)
st.rerun()
# 🚗 "What car is that?" Tab
with tab2:
uploaded_image = st.file_uploader("Upload an image of a car and let Ai identify it for you", type=["jpg", "jpeg", "png"])
if uploaded_image:
col1, col2 = st.columns([1.2, 1.8])
with col1:
image = Image.open(uploaded_image)
st.image(image, caption="Uploaded Image", use_container_width=True)
with col2:
try:
image_thread = client.beta.threads.create()
file_response = client.files.create(file=uploaded_image, purpose="assistants")
client.beta.threads.messages.create(
thread_id=image_thread.id,
role="user",
content=[
{
"type": "image_file",
"image_file": {"file_id": file_response.id}
},
{
"type": "text",
"text": "Please identify this car from the image and include a full vehicle specification table, a short overview of the car, and who it's recommended for."
}
]
)
run = client.beta.threads.runs.create(
thread_id=image_thread.id,
assistant_id=assistant_id
)
with st.spinner("🔍 Analyzing image and identifying the car..."):
while True:
run_status = client.beta.threads.runs.retrieve(
thread_id=image_thread.id,
run_id=run.id
)
if run_status.status == "completed":
break
time.sleep(1)
messages = client.beta.threads.messages.list(thread_id=image_thread.id)
assistant_message = messages.data[0].content[0].text.value
cleaned_message = re.sub(r'\*\*(.*?)\*\*', r'\1', assistant_message)
cleaned_message = cleaned_message.replace("###", "").replace("\n", "
")
# Replace key labels with icons
cleaned_message = cleaned_message.replace("Transmission", "
Transmission")
cleaned_message = cleaned_message.replace("Engine Size", "
Engine Size")
st.success("✅ Identification Complete")
st.markdown(f"""
{cleaned_message}
""", unsafe_allow_html=True)
except Exception as e:
st.error(f"❌ Error during image analysis: {str(e)}")