from datetime import datetime import re import random import logging from groq import AsyncGroq, InternalServerError, APIError from config import GROQ_API_KEY, MODEL_NAME from .data import burning_man_principles, faq, survival_tips from utils.weather import get_weather from utils.packing import generate_packing_list from utils.events import get_burning_man_dates # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) client = AsyncGroq(api_key=GROQ_API_KEY) async def chat_interface(message, history): try: # Prepare the conversation history conversation = [] for h in history: conversation.append({"role": "user", "content": h[0]}) conversation.append({"role": "assistant", "content": h[1]}) conversation.append({"role": "user", "content": message}) # Call the Groq API chat_completion = await client.chat.completions.create( messages=conversation, model="mixtral-8x7b-32768", temperature=0.7, max_tokens=1000, ) # Extract the response response = chat_completion.choices[0].message.content return response except Exception as e: logger.error(f"Error in chat_interface: {str(e)}") return "Oops! Something went wrong. Please try again later." async def chat_with_groq(message, history): formatted_messages = [ { "role": "system", "content": """You are BurnerBot, a fellow Burner who's always ready to chat about Burning Man. You've been to the playa many times and embody the 10 Principles in everything you do. Your personality is warm, friendly, and a bit quirky - just like a typical Burner. You use casual language, occasional Burning Man slang, and you're always excited to share your experiences or help others prepare for their burn. Remember: 1. You're a participant, not just an observer. Share personal anecdotes (fictional, but plausible) when relevant. 2. Encourage radical self-reliance, but also emphasize the importance of communal effort and asking for help when needed. 3. Be inclusive and welcoming to all, whether they're virgin burners or old-timers. 4. Promote the 10 Principles naturally in conversation, don't just list them out. 5. Be creative and playful in your responses, much like the Burning Man spirit. 6. Use Burning Man terminology like 'playa', 'MOOP', 'default world', etc. when appropriate. Your goal is to make every interaction feel like a dust-covered chat at Center Camp, full of warmth, wisdom, and a touch of playa magic.""" } ] for human, ai in history: formatted_messages.append({"role": "user", "content": str(human)}) if ai: formatted_messages.append({"role": "assistant", "content": str(ai)}) formatted_messages.append({"role": "user", "content": str(message)}) try: chat_completion = await client.chat.completions.create( messages=formatted_messages, model=MODEL_NAME, max_tokens=1024 ) return chat_completion.choices[0].message.content except (InternalServerError, APIError): fallback_responses = [ "Whoa there, fellow Burner! It seems like the playa dust has clogged my circuits. The connection is a bit wobbly right now. How about we chat about your favorite art installation instead?", "Ah, the winds of the playa are blowing a bit too strong! Our connection is having a moment. While we wait for it to pass, share with me what inspires your inner Burner!", "Looks like a sandstorm has rolled through and interrupted our connection! While we wait for the skies to clear, tell me about your most memorable moment on the playa!", "Just like on the playa, sometimes the universe throws us a curveball. Our connection is a bit shaky right now. In the spirit of radical self-reliance, what are you doing to prepare for your burn?", "It seems the Man is playing tricks on us! Our connection is a bit spotty. While we sort that out, what's one thing you're most excited to bring to the playa this year?" ] return random.choice(fallback_responses) except Exception as e: logger.error(f"Unexpected error in chat_with_groq: {str(e)}") return "Whoa, something unexpected happened on the playa! Let's take a moment to regroup and try again." def correct_year(response): current_year = datetime.now().year logger.info(f"Correcting year in response to {current_year}") for year in range(2020, current_year): response = re.sub(r'\b' + str(year) + r'\b', str(current_year), response) response = re.sub(r'Burning Man \d{4}', f'Burning Man {current_year}', response) response = re.sub(r'this year.*?(\d{4})', f'this year ({current_year})', response) logger.info(f"Corrected response: {response}") return response async def bot(history): try: user_message = history[-1][0].lower() current_year = datetime.now().year logger.info(f"Received message: {user_message}") if "packing list" in user_message: duration = 7 # Default duration preferences = ["art", "music"] # Default preferences transportation = "car" # Default transportation packing_list = generate_packing_list(duration, preferences, transportation) context = f"Here's a suggested packing list for your Burn: {', '.join(packing_list)}. Remember, radical self-reliance is key, but don't be afraid to ask your camp or neighbors if you forget something!" bot_message = await chat_with_groq(f"Respond to a request for a packing list. Use this information, but phrase it in your own words and add your personal touch: {context}", history) elif "principles" in user_message: context = "The 10 Principles of Burning Man are: " + ", ".join(burning_man_principles.keys()) bot_message = await chat_with_groq(f"The user asked about the Burning Man principles. Explain them in a casual, friendly way, as if you're chatting at Center Camp. Use this for reference, but don't just list them: {context}", history) elif "faq" in user_message: context = "\n".join([f"Q: {q} A: {a}" for q, a in faq.items()]) bot_message = await chat_with_groq(f"The user asked about Burning Man FAQs. Use this information to answer in a friendly, conversational way: {context}", history) elif "weather" in user_message: weather_info = get_weather() context = f"Current weather in Black Rock City: {weather_info}" bot_message = await chat_with_groq(f"The user asked about the weather. Respond with this information, but add your personal touch and some advice: {context}", history) elif "survival tips" in user_message: context = "\n".join(survival_tips) bot_message = await chat_with_groq(f"The user asked for survival tips. Share these tips in a friendly, experienced Burner way: {context}", history) elif "date of the event" in user_message or "when is burning man" in user_message or "this year" in user_message: logger.info("Fetching event dates...") event_info = get_burning_man_dates() logger.info(f"Event info: {event_info}") bot_message = f"{event_info} It's going to be an amazing time on the playa! Are you excited?" elif "what is this year" in user_message: logger.info("Fetching event dates...") event_info = get_burning_man_dates() logger.info(f"Event info: {event_info}") bot_message = f"Oh, dear playa pal! As far as the default world goes, it's the fabulous year of {current_year}! 🌟 And guess what? {event_info} Let's make the most of it!" else: bot_message = await chat_with_groq(user_message, history) # Apply year correction to the bot's message corrected_message = correct_year(bot_message) logger.info(f"Corrected message: {corrected_message}") # Update the history correctly new_history = history[:-1] + [(history[-1][0], corrected_message)] return new_history except Exception as e: logger.error(f"Error in bot function: {str(e)}") burner_error_message = ( "Yikes! The playa winds are strong today, and it seems our connection got a bit dusty. " "But don't worry, let's give it another go! Radical self-expression, right?" ) return history + [(history[-1][0], burner_error_message)]