Spaces:
Sleeping
Sleeping
| import time | |
| import streamlit as st | |
| import openai | |
| import os | |
| import json | |
| import re | |
| import requests | |
| from PIL import Image | |
| from io import BytesIO | |
| from urllib.parse import quote | |
| # ------------------ Configuration ------------------ | |
| st.set_page_config(page_title="Forrestdale Technical Drawing Assistant", layout="wide") | |
| st.markdown(""" | |
| <h1 style='font-size: 2.5rem;'>🏪 Forrestdale Technical Drawing Assistant</h1> | |
| <p style='font-size: 1rem;'>Ask about plans, drawings or components</p> | |
| """, unsafe_allow_html=True) | |
| # ------------------ Environment ------------------ | |
| OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") | |
| ASSISTANT_ID = os.getenv("ASSISTANT_ID") | |
| if not OPENAI_API_KEY or not ASSISTANT_ID: | |
| st.error("❌ Missing environment variables: OPENAI_API_KEY or ASSISTANT_ID.") | |
| st.stop() | |
| client = openai.OpenAI(api_key=OPENAI_API_KEY) | |
| # ------------------ Helper Functions ------------------ | |
| def extract_json_from_response(response_text): | |
| try: | |
| match = re.search(r"```json\s*([\s\S]+?)```", response_text.strip()) | |
| if not match: | |
| raise ValueError("No valid JSON block found.") | |
| json_str = match.group(1).strip() | |
| return json.loads(json_str) | |
| except Exception as e: | |
| st.error("⚠️ Could not parse assistant response as JSON.") | |
| st.stop() | |
| def display_card(item): | |
| st.markdown(""" | |
| <div style="border: 1px solid #333; border-radius: 12px; padding: 16px; margin: 10px; background-color: #111;"> | |
| <h3>📁 {} ({})</h3> | |
| <p><b>Discipline:</b> {}</p> | |
| <p><b>Summary:</b> {}</p> | |
| <div style="display: flex; flex-wrap: wrap; gap: 10px;"> | |
| """.format(item["drawing_number"], item.get("drawing_type", ""), item["discipline"], item["summary"]), unsafe_allow_html=True) | |
| for img in item.get("images", []): | |
| try: | |
| response = requests.get(img) | |
| if response.status_code == 200: | |
| image = Image.open(BytesIO(response.content)) | |
| st.image(image, caption=os.path.basename(img), width=200) | |
| except: | |
| st.warning(f"Image failed to load: {img}") | |
| st.markdown("</div></div>", unsafe_allow_html=True) | |
| # ------------------ Input ------------------ | |
| query = st.text_input("Ask about plans, drawings or components", placeholder="Show me all architectural plans") | |
| if query: | |
| with st.spinner("🔍 Querying assistant..."): | |
| try: | |
| if "thread_id" not in st.session_state: | |
| thread = client.beta.threads.create() | |
| st.session_state.thread_id = thread.id | |
| client.beta.threads.messages.create( | |
| thread_id=st.session_state.thread_id, | |
| role="user", | |
| content=query | |
| ) | |
| run = client.beta.threads.runs.create( | |
| thread_id=st.session_state.thread_id, | |
| assistant_id=ASSISTANT_ID | |
| ) | |
| while True: | |
| run_status = client.beta.threads.runs.retrieve( | |
| thread_id=st.session_state.thread_id, | |
| run_id=run.id | |
| ) | |
| if run_status.status in ["completed", "failed", "cancelled"]: | |
| break | |
| time.sleep(1) | |
| if run_status.status != "completed": | |
| st.error(f"❌ Assistant run failed: {run_status.status}") | |
| else: | |
| messages = client.beta.threads.messages.list(thread_id=st.session_state.thread_id) | |
| for message in reversed(messages.data): | |
| if message.role == "assistant": | |
| response_text = message.content[0].text.value | |
| data = extract_json_from_response(response_text) | |
| for item in data: | |
| display_card(item) | |
| break | |
| except Exception as e: | |
| st.error(f"❌ Error during assistant call: {e}") | |