Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| import re | |
| import random | |
| import datetime | |
| import sqlite3 | |
| # Initialize DB | |
| def init_db(): | |
| conn = sqlite3.connect('orders.db') | |
| c = conn.cursor() | |
| c.execute('''CREATE TABLE IF NOT EXISTS orders ( | |
| id INTEGER PRIMARY KEY AUTOINCREMENT, order_name TEXT, order_type TEXT, weight_kg REAL, | |
| pickup TEXT, delivery TEXT, date TEXT, payment TEXT, dimensions TEXT, instructions TEXT, | |
| created TEXT, completed BOOLEAN DEFAULT 0)''') | |
| conn.commit() | |
| return conn | |
| # Extract details with LLM-like pattern recognition | |
| def extract_details(text): | |
| details = {} | |
| # Order type pattern - looking for what's being shipped | |
| patterns = { | |
| 'order_type': [r'ship (?:a |an |some |)(.+?)(?:weigh|from|to|of|that|\.|,)', | |
| r'order (?:for |of |some |)(.+?)(?:weigh|from|to|\.|,)', | |
| r'sending (?:a |an |some |)(.+?)(?:weigh|from|to|\.|,)'], | |
| 'weight_kg': [r'(\d+(?:\.\d+)?)(?:\s*)(kg|kilo|kilograms?|g|grams?|lbs?|pounds?)'], | |
| 'pickup': [r'from (.+?)(?:to|and|\.|,|$)', | |
| r'pickup (?:from |at |in |)(.+?)(?:to|and|\.|,|$)'], | |
| 'delivery': [r'to (.+?)(?:by|on|for|\.|,|$)', | |
| r'deliver (?:to |at |in |)(.+?)(?:by|on|\.|,|$)'], | |
| 'date': [r'(?:on|by) (.+?)(?:via|\.|,|$)', | |
| r'(?:ship|deliver|send) (?:on|by) (.+?)(?:via|\.|,|$)'], | |
| 'payment': [r'(?:pay|payment) (?:via|using|with|by) (.+?)(?:\.|\,|$)', | |
| r'(?:using|via|with) (.+?) (?:payment|card|method)'], | |
| 'dimensions': [r'dimensions?:? (.+?)(?:cm|m|\.|,|$)', | |
| r'size:? (.+?)(?:cm|m|\.|,|$)', | |
| r'(\d+)(?:\s*)[xX×](\d+)(?:\s*)[xX×](\d+)'], | |
| 'instructions': [r'(?:special |)instructions?:? (.+?)(?:\.|\,|$)', | |
| r'note:? (.+?)(?:\.|\,|$)', | |
| r'(?:please|kindly) (.+?)(?:\.|\,|$)'] | |
| } | |
| # Process date values | |
| date_values = { | |
| 'today': datetime.datetime.now().strftime("%Y-%m-%d"), | |
| 'tomorrow': (datetime.datetime.now() + datetime.timedelta(days=1)).strftime("%Y-%m-%d"), | |
| 'next week': (datetime.datetime.now() + datetime.timedelta(days=7)).strftime("%Y-%m-%d") | |
| } | |
| # Try each pattern for each field | |
| for field, field_patterns in patterns.items(): | |
| for pattern in field_patterns: | |
| match = re.search(pattern, text, re.IGNORECASE) | |
| if match: | |
| if field == 'weight_kg' and match.group(2): | |
| # Convert weight to kg | |
| value = float(match.group(1)) | |
| unit = match.group(2).lower() | |
| if unit.startswith(('g', 'gram')): | |
| value /= 1000 | |
| elif unit.startswith(('lb', 'pound')): | |
| value *= 0.45359237 | |
| details[field] = value | |
| elif field == 'date' and match.group(1).lower() in date_values: | |
| details[field] = date_values[match.group(1).lower()] | |
| elif field == 'dimensions' and len(match.groups()) > 1: | |
| # Format dimensions like 10x20x30 | |
| details[field] = f"{match.group(1)}x{match.group(2)}x{match.group(3)}" | |
| else: | |
| details[field] = match.group(1).strip() | |
| break | |
| return details | |
| # Save order to DB | |
| def save_order(conn, details): | |
| try: | |
| c = conn.cursor() | |
| name = details.get('order_name', f"Order-{datetime.datetime.now().strftime('%Y%m%d')}-{random.randint(1000,9999)}") | |
| c.execute('''INSERT INTO orders (order_name, order_type, weight_kg, pickup, delivery, date, | |
| payment, dimensions, instructions, created) | |
| VALUES (?,?,?,?,?,?,?,?,?,?)''', | |
| (name, details.get('order_type'), details.get('weight_kg'), | |
| details.get('pickup'), details.get('delivery'), details.get('date'), | |
| details.get('payment'), details.get('dimensions'), details.get('instructions'), | |
| datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))) | |
| conn.commit() | |
| return c.lastrowid | |
| except Exception as e: | |
| st.error(f"Error: {e}") | |
| return None | |
| # Get orders from DB | |
| def get_orders(conn): | |
| c = conn.cursor() | |
| c.execute('SELECT * FROM orders ORDER BY created DESC LIMIT 5') | |
| return c.fetchall() | |
| # Main app | |
| def main(): | |
| st.set_page_config(page_title="Transport Assistant", page_icon="🚚") | |
| conn = init_db() | |
| # Initialize session state | |
| if 'messages' not in st.session_state: | |
| st.session_state.messages = [{"role": "assistant", "content": "👋 Hello! I'm your Transport Assistant. What would you like to ship?"}] | |
| if 'order' not in st.session_state: | |
| st.session_state.order = {} | |
| if 'state' not in st.session_state: | |
| st.session_state.state = "collecting" | |
| if 'current_field' not in st.session_state: | |
| st.session_state.current_field = None | |
| # Required fields including dimensions | |
| required_fields = ['order_type', 'weight_kg', 'pickup', 'delivery', 'date', 'payment', 'dimensions'] | |
| # App UI | |
| st.title("🚚 Transport Order Assistant") | |
| # Display chat history | |
| for msg in st.session_state.messages: | |
| with st.chat_message(msg["role"]): | |
| st.markdown(msg["content"]) | |
| # User input handling | |
| user_input = st.chat_input("Type your shipping details...") | |
| if user_input: | |
| # Add user message to chat | |
| st.session_state.messages.append({"role": "user", "content": user_input}) | |
| # Process input based on current state | |
| if st.session_state.state == "collecting": | |
| # If asking for specific field | |
| if st.session_state.current_field: | |
| field = st.session_state.current_field | |
| # Direct field extraction | |
| if field == 'weight_kg': | |
| weight_match = re.search(r'(\d+(?:\.\d+)?)(?:\s*)(kg|kilo|kilograms?|g|grams?|lbs?|pounds?)?', user_input, re.IGNORECASE) | |
| if weight_match: | |
| value = float(weight_match.group(1)) | |
| unit = weight_match.group(2).lower() if weight_match.group(2) else 'kg' | |
| if unit.startswith(('g', 'gram')): | |
| value /= 1000 | |
| elif unit.startswith(('lb', 'pound')): | |
| value *= 0.45359237 | |
| st.session_state.order[field] = value | |
| else: | |
| try: | |
| st.session_state.order[field] = float(user_input) | |
| except: | |
| st.session_state.order[field] = user_input | |
| elif field == 'date': | |
| date_values = { | |
| 'today': datetime.datetime.now().strftime("%Y-%m-%d"), | |
| 'tomorrow': (datetime.datetime.now() + datetime.timedelta(days=1)).strftime("%Y-%m-%d"), | |
| 'next week': (datetime.datetime.now() + datetime.timedelta(days=7)).strftime("%Y-%m-%d") | |
| } | |
| st.session_state.order[field] = date_values.get(user_input.lower(), user_input) | |
| else: | |
| st.session_state.order[field] = user_input | |
| st.session_state.current_field = None | |
| else: | |
| # Extract all possible details from user input | |
| extracted = extract_details(user_input) | |
| # Update order with extracted details | |
| for key, value in extracted.items(): | |
| if value: | |
| st.session_state.order[key] = value | |
| # Check for missing fields | |
| missing = [f for f in required_fields if f not in st.session_state.order or not st.session_state.order[f]] | |
| if not missing: | |
| # All required fields collected - show summary | |
| response = "Please confirm your order:\n\n" | |
| response += f"📦 **Item:** {st.session_state.order['order_type']}\n" | |
| response += f"⚖️ **Weight:** {st.session_state.order['weight_kg']} kg\n" | |
| response += f"📍 **From:** {st.session_state.order['pickup']}\n" | |
| response += f"🏁 **To:** {st.session_state.order['delivery']}\n" | |
| response += f"📅 **Date:** {st.session_state.order['date']}\n" | |
| response += f"💳 **Payment:** {st.session_state.order['payment']}\n" | |
| response += f"📏 **Dimensions:** {st.session_state.order['dimensions']}\n" | |
| if 'instructions' in st.session_state.order: | |
| response += f"📝 **Instructions:** {st.session_state.order['instructions']}\n" | |
| response += "\nType 'confirm' to place your order or tell me what to change." | |
| st.session_state.messages.append({"role": "assistant", "content": response}) | |
| st.session_state.state = "confirming" | |
| else: | |
| # Ask for the first missing field | |
| field = missing[0] | |
| field_questions = { | |
| 'order_type': "What are you shipping?", | |
| 'weight_kg': "What is the weight of your package?", | |
| 'pickup': "Where should we pick up the package from?", | |
| 'delivery': "Where should we deliver the package to?", | |
| 'date': "When do you want it shipped?", | |
| 'payment': "How will you pay for the shipping?", | |
| 'dimensions': "What are the dimensions of your package? (LxWxH)" | |
| } | |
| # Show current progress | |
| collected = {k: v for k, v in st.session_state.order.items() if v} | |
| if collected: | |
| response = "Here's what I have so far:\n\n" | |
| for k, v in collected.items(): | |
| field_name = k.replace('_', ' ').title() | |
| if k == 'weight_kg': | |
| response += f"• **{field_name}:** {v} kg\n" | |
| else: | |
| response += f"• **{field_name}:** {v}\n" | |
| response += f"\n{field_questions[field]}" | |
| else: | |
| response = field_questions[field] | |
| st.session_state.messages.append({"role": "assistant", "content": response}) | |
| st.session_state.current_field = field | |
| elif st.session_state.state == "confirming": | |
| # Process confirmation or changes | |
| if re.search(r'\b(yes|confirm|correct|ok|right|good)\b', user_input.lower()): | |
| # Save the order | |
| order_id = save_order(conn, st.session_state.order) | |
| if order_id: | |
| response = f"✅ Order #{order_id} confirmed! Your {st.session_state.order['order_type']} will be picked up from {st.session_state.order['pickup']} on {st.session_state.order['date']}." | |
| st.session_state.messages.append({"role": "assistant", "content": response}) | |
| # Reset for new order | |
| st.session_state.order = {} | |
| st.session_state.state = "collecting" | |
| st.session_state.current_field = None | |
| # Prompt for new order | |
| st.session_state.messages.append({ | |
| "role": "assistant", | |
| "content": "Would you like to place another order?" | |
| }) | |
| else: | |
| response = "Sorry, there was an error saving your order. Please try again." | |
| st.session_state.messages.append({"role": "assistant", "content": response}) | |
| else: | |
| # Check for field to change | |
| fields = { | |
| 'order_type': ['item', 'product', 'shipping'], | |
| 'weight_kg': ['weight', 'kg', 'heavy'], | |
| 'pickup': ['pickup', 'from', 'origin'], | |
| 'delivery': ['delivery', 'to', 'destination'], | |
| 'date': ['date', 'when', 'day', 'time'], | |
| 'payment': ['payment', 'pay', 'card', 'cash'], | |
| 'dimensions': ['dimensions', 'size', 'measurements'], | |
| 'instructions': ['instructions', 'notes', 'special'] | |
| } | |
| field_to_edit = None | |
| for field, keywords in fields.items(): | |
| if any(k in user_input.lower() for k in keywords): | |
| field_to_edit = field | |
| break | |
| if field_to_edit: | |
| response = f"What would you like to change the {field_to_edit.replace('_', ' ')} to?" | |
| st.session_state.messages.append({"role": "assistant", "content": response}) | |
| st.session_state.state = "editing" | |
| st.session_state.current_field = field_to_edit | |
| else: | |
| # Extract from entire message | |
| extracted = extract_details(user_input) | |
| updated = False | |
| for key, value in extracted.items(): | |
| if value: | |
| st.session_state.order[key] = value | |
| updated = True | |
| if updated: | |
| response = "I've updated your order. Please confirm:\n\n" | |
| response += f"📦 **Item:** {st.session_state.order.get('order_type', 'N/A')}\n" | |
| response += f"⚖️ **Weight:** {st.session_state.order.get('weight_kg', 'N/A')} kg\n" | |
| response += f"📍 **From:** {st.session_state.order.get('pickup', 'N/A')}\n" | |
| response += f"🏁 **To:** {st.session_state.order.get('delivery', 'N/A')}\n" | |
| response += f"📅 **Date:** {st.session_state.order.get('date', 'N/A')}\n" | |
| response += f"💳 **Payment:** {st.session_state.order.get('payment', 'N/A')}\n" | |
| response += f"📏 **Dimensions:** {st.session_state.order.get('dimensions', 'N/A')}\n" | |
| if 'instructions' in st.session_state.order: | |
| response += f"📝 **Instructions:** {st.session_state.order['instructions']}\n" | |
| response += "\nIs this correct now?" | |
| else: | |
| response = "I didn't understand what you want to change. Please specify which field to update." | |
| st.session_state.messages.append({"role": "assistant", "content": response}) | |
| elif st.session_state.state == "editing": | |
| # Update the specified field | |
| field = st.session_state.current_field | |
| if field == 'weight_kg': | |
| weight_match = re.search(r'(\d+(?:\.\d+)?)(?:\s*)(kg|kilo|kilograms?|g|grams?|lbs?|pounds?)?', user_input, re.IGNORECASE) | |
| if weight_match: | |
| value = float(weight_match.group(1)) | |
| unit = weight_match.group(2).lower() if weight_match.group(2) else 'kg' | |
| if unit.startswith(('g', 'gram')): | |
| value /= 1000 | |
| elif unit.startswith(('lb', 'pound')): | |
| value *= 0.45359237 | |
| st.session_state.order[field] = value | |
| else: | |
| try: | |
| st.session_state.order[field] = float(user_input) | |
| except: | |
| st.session_state.order[field] = user_input | |
| elif field == 'date': | |
| date_values = { | |
| 'today': datetime.datetime.now().strftime("%Y-%m-%d"), | |
| 'tomorrow': (datetime.datetime.now() + datetime.timedelta(days=1)).strftime("%Y-%m-%d"), | |
| 'next week': (datetime.datetime.now() + datetime.timedelta(days=7)).strftime("%Y-%m-%d") | |
| } | |
| st.session_state.order[field] = date_values.get(user_input.lower(), user_input) | |
| else: | |
| st.session_state.order[field] = user_input | |
| # Return to confirmation | |
| response = "I've updated your order. Please confirm:\n\n" | |
| response += f"📦 **Item:** {st.session_state.order.get('order_type', 'N/A')}\n" | |
| response += f"⚖️ **Weight:** {st.session_state.order.get('weight_kg', 'N/A')} kg\n" | |
| response += f"📍 **From:** {st.session_state.order.get('pickup', 'N/A')}\n" | |
| response += f"🏁 **To:** {st.session_state.order.get('delivery', 'N/A')}\n" | |
| response += f"📅 **Date:** {st.session_state.order.get('date', 'N/A')}\n" | |
| response += f"💳 **Payment:** {st.session_state.order.get('payment', 'N/A')}\n" | |
| response += f"📏 **Dimensions:** {st.session_state.order.get('dimensions', 'N/A')}\n" | |
| if 'instructions' in st.session_state.order: | |
| response += f"📝 **Instructions:** {st.session_state.order['instructions']}\n" | |
| else: | |
| response += "\nWould you like to add any special instructions? (optional)" | |
| response += "\nIs this correct now?" | |
| st.session_state.messages.append({"role": "assistant", "content": response}) | |
| st.session_state.state = "confirming" | |
| st.session_state.current_field = None | |
| st.rerun() | |
| # Show recent orders in sidebar | |
| with st.sidebar: | |
| st.header("Recent Orders") | |
| orders = get_orders(conn) | |
| if orders: | |
| for order in orders: | |
| with st.expander(f"Order #{order[0]} - {order[1]}"): | |
| st.write(f"**Type:** {order[2]}") | |
| st.write(f"**Weight:** {order[3]} kg") | |
| st.write(f"**From:** {order[4]}") | |
| st.write(f"**To:** {order[5]}") | |
| st.write(f"**Date:** {order[6]}") | |
| st.write(f"**Payment:** {order[7]}") | |
| st.write(f"**Dimensions:** {order[8]}") | |
| if order[9]: | |
| st.write(f"**Instructions:** {order[9]}") | |
| else: | |
| st.write("No orders yet") | |
| if __name__ == "__main__": | |
| main() |