Ujeshhh's picture
Update app.py
299af4d verified
raw
history blame
16.5 kB
import gradio as gr
import google.generativeai as genai
import os
from datetime import datetime
import plotly.express as px
import pandas as pd
from langdetect import detect
import random
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
import base64
from email.mime.text import MIMEText
import re
import json
# Configure Gemini API
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
# Gmail API configuration
GMAIL_ADDRESS = os.getenv("GMAIL_ADDRESS")
GMAIL_TOKEN_JSON = os.getenv("GMAIL_TOKEN_JSON")
# Initialize Gmail API client
def get_gmail_service():
try:
creds = Credentials.from_authorized_user_info(json.loads(GMAIL_TOKEN_JSON))
return build("gmail", "v1", credentials=creds), None
except Exception as e:
return None, f"Failed to initialize Gmail API: {str(e)}"
# Coping Strategies Library
coping_strategies = {
"happy": [
"Keep the positivity flowing! Try writing down three things you're grateful for today.",
"Share your joy! Call a friend or loved one to spread the good vibes."
],
"sad": [
"It's okay to feel this way. Try a gentle activity like listening to calming music or taking a short walk.",
"Write down your thoughts in a journal to process what's on your mind."
],
"anxious": [
"Take slow, deep breaths: inhale for 4 seconds, hold for 4, exhale for 4.",
"Try a grounding exercise: name 5 things you see, 4 you can touch, 3 you hear, 2 you smell, 1 you taste."
],
"stressed": [
"Pause for a moment and stretch your body to release tension.",
"Break tasks into smaller steps and tackle one at a time."
],
"other": [
"Reflect on what's on your mind with a quick mindfulness moment: focus on your breath for 60 seconds.",
"Engage in a favorite hobby to lift your spirits."
]
}
# Regional Resources
regional_resources = {
"USA": [
"National Suicide Prevention Lifeline: 1-800-273-8255",
"Crisis Text Line: Text HOME to 741741",
"[MentalHealth.gov](https://www.mentalhealth.gov/)"
],
"India": [
"Vandrevala Foundation: 1860-2662-345",
"AASRA Suicide Prevention: +91-9820466726",
"[iCall Helpline](https://icallhelpline.org/)"
],
"UK": [
"Samaritans: 116 123",
"Shout Crisis Text Line: Text SHOUT to 85258",
"[Mind UK](https://www.mind.org.uk/)"
],
"Global": [
"Befrienders Worldwide: [Find a helpline](https://befrienders.org/)",
"[WHO Mental Health Resources](https://www.who.int/health-topics/mental-health)"
]
}
# Mock Therapist Database
therapists = {
"Dr. Jane Smith": {
"specialty": "Anxiety and Depression",
"email": "[email protected]",
"times": ["09:00", "09:30", "10:00", "10:30", "11:00", "14:00", "14:30", "15:00", "16:00"]
},
"Dr. Amit Patel": {
"specialty": "Stress Management",
"email": "[email protected]",
"times": ["10:00", "10:30", "11:00", "11:30", "12:00", "15:00", "15:30", "16:00", "17:00"]
},
"Dr. Sarah Brown": {
"specialty": "Trauma and PTSD",
"email": "[email protected]",
"times": ["08:00", "08:30", "09:00", "13:00", "13:30", "14:00", "15:30", "16:00", "18:00"]
}
}
# Initialize Gemini model
model = genai.GenerativeModel("gemini-1.5-flash",
generation_config={"temperature": 0.7, "max_output_tokens": 200})
# Chatbot function
def chatbot_function(message, mood, conversation_mode, region, state):
if "chat_history" not in state:
state["chat_history"] = []
history = state["chat_history"]
if not message.strip():
return "Please enter a message.", state
try:
lang = detect(message)
except:
lang = "en"
if mood and mood != "Select mood":
history.append([f"I'm feeling {mood.lower()}.", None])
history.append([message, None])
tone_instruction = {
"Calm": "Respond in a soothing, gentle tone to promote relaxation.",
"Motivational": "Use an uplifting, encouraging tone to inspire confidence.",
"Neutral": "Maintain a balanced, empathetic tone."
}.get(conversation_mode, "Maintain a balanced, empathetic tone.")
prompt = f"""
You are a compassionate mental health support chatbot. Engage in a supportive conversation with the user based on their input: {message}.
- Provide empathetic, sensitive responses in the user's language (detected as {lang}).
- {tone_instruction}
- If signs of distress are detected, suggest coping strategies relevant to their mood or input.
- Recommend professional resources tailored to the user's region ({region}).
- Keep responses concise, warm, and encouraging.
"""
try:
response = model.generate_content(prompt)
response_text = response.text
except Exception as e:
response_text = "I'm here for you, but something went wrong. Please try again."
if mood and mood != "Select mood":
mood_key = mood.lower()
if mood_key in coping_strategies:
strategy = random.choice(coping_strategies[mood_key])
response_text += f"\n\n**Coping Strategy**: {strategy}"
region_key = region if region in regional_resources else "Global"
resources = "\n\n**Resources**:\n" + "\n".join(regional_resources[region_key])
response_text += resources
history[-1][1] = response_text
state["chat_history"] = history
chat_display = ""
for user_msg, bot_msg in history:
if user_msg:
chat_display += f"**You**: {user_msg}\n\n"
if bot_msg:
chat_display += f"**Bot**: {bot_msg}\n\n"
return chat_display, state
# Clear chat history
def clear_chat(state):
state["chat_history"] = []
return "", state
# Mood journal function
def log_mood(mood, state):
if mood and mood != "Select mood":
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
state["mood_journal"].append({"timestamp": timestamp, "mood": mood.lower()})
return "Mood logged!", state
return "Please select a mood.", state
# Mood trend visualization
def show_mood_trends(state):
if not state["mood_journal"]:
return "No moods logged yet.", state
df = pd.DataFrame(state["mood_journal"])
fig = px.line(df, x="timestamp", y="mood", title="Mood Trends", markers=True)
return fig, state
# Feedback function
def submit_feedback(feedback_text, rating, state):
if feedback_text or rating:
state["feedback"].append({"text": feedback_text, "rating": rating, "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
return "Feedback submitted!", state
return "Please provide feedback or a rating.", state
# Emergency resources
def show_emergency_resources():
return """
**Crisis Support:**
- National Suicide Prevention Lifeline: 1-800-273-8255
- Crisis Text Line: Text HOME to 741741
- [MentalHealth.gov](https://www.mentalhealth.gov/)
"""
# Get available times for selected therapist
def get_available_times(therapist):
if therapist and therapist in therapists:
return therapists[therapist]["times"], f"Available times for {therapist} loaded."
return ["Select time"], "Please select a therapist."
# Create MIME message for Gmail API
def create_message(to, subject, message_text):
message = MIMEText(message_text)
message["to"] = to
message["from"] = GMAIL_ADDRESS
message["subject"] = subject
raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
return {"raw": raw}
# Send emails using Gmail API
def send_emails(therapist, time_slot, date, user_email, therapist_email, state):
if "failed_emails" not in state:
state["failed_emails"] = []
therapist_body = f"""
Dear {therapist},
You have a new appointment:
- Date: {date}
- Time: {time_slot}
- Client Email: {user_email}
Please confirm with the client.
Best,
Mental Health Chatbot
"""
therapist_message = create_message(therapist_email, "New Appointment", therapist_body)
user_body = f"""
Dear User,
Your appointment is booked:
- Therapist: {therapist}
- Date: {date}
- Time: {time_slot}
Expect a confirmation from {therapist}.
Best,
Mental Health Chatbot
"""
user_message = create_message(user_email, "Appointment Confirmation", user_body)
service, error = get_gmail_service()
if not service:
state["failed_emails"].append({
"therapist_email": {"to": therapist_email, "subject": "New Appointment", "body": therapist_body},
"user_email": {"to": user_email, "subject": "Appointment Confirmation", "body": user_body},
"error": error,
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
})
return False, error
try:
service.users().messages().send(userId="me", body=therapist_message).execute()
service.users().messages().send(userId="me", body=user_message).execute()
return True, ""
except HttpError as e:
error_msg = f"Gmail API error: {str(e)}"
state["failed_emails"].append({
"therapist_email": {"to": therapist_email, "subject": "New Appointment", "body": therapist_body},
"user_email": {"to": user_email, "subject": "Appointment Confirmation", "body": user_body},
"error": error_msg,
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
})
return False, error_msg
except Exception as e:
error_msg = f"Unexpected error: {str(e)}"
state["failed_emails"].append({
"therapist_email": {"to": therapist_email, "subject": "New Appointment", "body": therapist_body},
"user_email": {"to": user_email, "subject": "Appointment Confirmation", "body": user_body},
"error": error_msg,
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
})
return False, error_msg
# Schedule appointment
def schedule_appointment(therapist, time_slot, date, user_email, state):
if not therapist or therapist not in therapists:
return "Please select a therapist.", state
if not time_slot or time_slot == "Select time":
return "Please select a time slot.", state
if not date:
return "Please select a date.", state
if not user_email or not re.match(r"[^@]+@[^@]+\.[^@]+", user_email):
return "Please enter a valid email.", state
try:
appointment_date = datetime.strptime(date, "%Y-%m-%d")
if appointment_date < datetime.now():
return "Please select a future date.", state
except ValueError:
return "Invalid date format (use YYYY-MM-DD).", state
try:
appointment_time = datetime.strptime(time_slot, "%H:%M").strftime("%H:%M")
if appointment_time not in therapists[therapist]["times"]:
return f"Time {appointment_time} is not available for {therapist}.", state
except ValueError:
return "Invalid time format (use HH:MM).", state
appointment = {
"therapist": therapist,
"time": appointment_time,
"date": date,
"user_email": user_email,
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
state["appointments"].append(appointment)
therapist_email = therapists[therapist]["email"]
success, error_msg = send_emails(therapist, appointment_time, date, user_email, therapist_email, state)
if success:
return f"Appointment booked with {therapist} on {date} at {appointment_time}. Emails sent!", state
else:
return (f"Appointment booked with {therapist} on {date} at {appointment_time}. "
f"Email sending failed: {error_msg}. Please contact {therapist_email}."), state
# Gradio Interface
with gr.Blocks(title="Mental Health Support Chatbot") as demo:
state = gr.State({"chat_history": [], "mood_journal": [], "feedback": [], "appointments": [], "failed_emails": []})
gr.Markdown("## Mental Health Support Chatbot")
gr.Markdown("A safe space to share your feelings and access support.")
with gr.Row():
with gr.Column(scale=1):
mood = gr.Dropdown(
choices=["Select mood", "Happy", "Sad", "Anxious", "Stressed", "Other"],
label="Mood",
value="Select mood"
)
conversation_mode = gr.Radio(
choices=["Neutral", "Calm", "Motivational"],
label="Conversation Style",
value="Neutral"
)
region = gr.Dropdown(
choices=["USA", "India", "UK", "Global"],
label="Region",
value="Global"
)
with gr.Column(scale=2):
chatbot = gr.Textbox(
label="Conversation",
value="",
interactive=False,
lines=10,
show_copy_button=True
)
user_input = gr.Textbox(
placeholder="Type your message here...",
label="Your Message",
lines=2
)
with gr.Row():
submit_btn = gr.Button("Send Message")
clear_btn = gr.Button("Clear Chat")
emergency_btn = gr.Button("Show Crisis Support")
emergency_output = gr.Markdown("")
with gr.Accordion("Mood Journal"):
log_mood_btn = gr.Button("Log Mood")
mood_log_output = gr.Textbox(label="Mood Log Status", interactive=False)
mood_trend_btn = gr.Button("View Mood Trends")
mood_trend_output = gr.Plot()
with gr.Accordion("Provide Feedback"):
feedback_text = gr.Textbox(label="Your Feedback", placeholder="How can we improve?")
feedback_rating = gr.Slider(minimum=1, maximum=5, step=1, label="Rating")
feedback_btn = gr.Button("Submit Feedback")
feedback_output = gr.Textbox(label="Feedback Status", interactive=False)
with gr.Accordion("Schedule Appointment"):
therapist = gr.Dropdown(
choices=list(therapists.keys()),
label="Select Therapist"
)
date = gr.Textbox(
label="Appointment Date",
placeholder="Enter date (YYYY-MM-DD)"
)
time_slot = gr.Dropdown(
choices=["Select time"],
label="Appointment Time",
interactive=True,
allow_custom_value=True
)
user_email = gr.Textbox(
label="Your Email",
placeholder="e.g., [email protected]"
)
schedule_btn = gr.Button("Book Appointment")
schedule_output = gr.Textbox(label="Booking Status", interactive=False)
submit_btn.click(
fn=chatbot_function,
inputs=[user_input, mood, conversation_mode, region, state],
outputs=[chatbot, state]
)
user_input.submit(
fn=chatbot_function,
inputs=[user_input, mood, conversation_mode, region, state],
outputs=[chatbot, state]
)
clear_btn.click(
fn=clear_chat,
inputs=[state],
outputs=[chatbot, state]
)
emergency_btn.click(
fn=show_emergency_resources,
inputs=None,
outputs=emergency_output
)
log_mood_btn.click(
fn=log_mood,
inputs=[mood, state],
outputs=[mood_log_output, state]
)
mood_trend_btn.click(
fn=show_mood_trends,
inputs=[state],
outputs=[mood_trend_output, state]
)
feedback_btn.click(
fn=submit_feedback,
inputs=[feedback_text, feedback_rating, state],
outputs=[feedback_output, state]
)
therapist.change(
fn=get_available_times,
inputs=therapist,
outputs=[time_slot, schedule_output]
)
schedule_btn.click(
fn=schedule_appointment,
inputs=[therapist, time_slot, date, user_email, state],
outputs=[schedule_output, state]
)
gr.Markdown("""
**Resources:**
- [Suicide Prevention Lifeline](https://suicidepreventionlifeline.org/)
- [MentalHealth.gov](https://www.mentalhealth.gov/)
- [Crisis Text Line](https://www.crisistextline.org/)
""")
if __name__ == "__main__":
demo.launch()