๐ ApnaLawyer
Secure Legal Assistance Portal
""", unsafe_allow_html=True)
choice = st.sidebar.radio("Select Option", ["Login", "Signup", "Forgot Password"], label_visibility="collapsed")
if choice == "Signup":
st.sidebar.markdown("### Create New Account")
name = st.sidebar.text_input("Full Name", key="signup_name")
email = st.sidebar.text_input("Email Address", key="signup_email")
password = st.sidebar.text_input("Password", type="password", key="signup_password")
confirm_password = st.sidebar.text_input("Confirm Password", type="password", key="signup_confirm")
if st.sidebar.button("Create Account", key="signup_button"):
if not name:
st.sidebar.error("Please enter your full name!")
elif not email:
st.sidebar.error("Email address is required!")
elif not password or not confirm_password:
st.sidebar.error("Password fields cannot be empty!")
elif password != confirm_password:
st.sidebar.error("Passwords do not match!")
else:
try:
user = auth.create_user_with_email_and_password(email, password)
if store_user_data(user['localId'], name, email):
auth.send_email_verification(user['idToken'])
st.sidebar.success("โ
Account created! Please verify your email before logging in.")
except Exception as e:
error_str = str(e)
if "EMAIL_EXISTS" in error_str:
st.sidebar.warning("โ Email already exists. Please try a different email address.")
elif "WEAK_PASSWORD" in error_str:
st.sidebar.warning("โ Password should be at least 6 characters long.")
else:
st.sidebar.error(f"Error: {error_str}")
elif choice == "Login":
st.sidebar.markdown("### Welcome Back")
email = st.sidebar.text_input("Email Address", key="login_email")
password = st.sidebar.text_input("Password", type="password", key="login_password")
if st.sidebar.button("Login", key="login_button"):
if not email:
st.sidebar.error("โ Please enter your email address")
elif not password:
st.sidebar.error("โ Please enter your password")
else:
try:
user = auth.sign_in_with_email_and_password(email, password)
user_info = auth.get_account_info(user['idToken'])
email_verified = user_info['users'][0]['emailVerified']
if email_verified:
user_name = get_user_name(user['localId'])
st.session_state.logged_in = True
st.session_state.user_email = email
st.session_state.user_token = user['idToken']
st.session_state.user_name = user_name
st.sidebar.success(f"๐ Welcome back, {user_name}!")
st.rerun()
else:
st.sidebar.warning("๐ง Email not verified. Please check your inbox.")
if st.sidebar.button("๐ Resend Verification Email", key="resend_verification"):
auth.send_email_verification(user['idToken'])
st.sidebar.info("๐ฌ Verification email sent again!")
except Exception as e:
error_str = str(e)
if "EMAIL_NOT_FOUND" in error_str or "no user record" in error_str.lower():
st.sidebar.error("๐ญ Account not found")
st.sidebar.warning("Don't have an account? Please sign up.")
elif "INVALID_PASSWORD" in error_str or "INVALID_LOGIN_CREDENTIALS" in error_str:
st.sidebar.error("๐ Incorrect password")
else:
st.sidebar.error("โ Login error. Please try again.")
elif choice == "Forgot Password":
st.sidebar.markdown("### Reset Your Password")
email = st.sidebar.text_input("Enter your email address", key="reset_email")
if st.sidebar.button("Send Reset Link", key="reset_button"):
if not email:
st.sidebar.error("Please enter your email address!")
else:
try:
auth.send_password_reset_email(email)
st.sidebar.success("๐ฌ Password reset link sent to your email.")
except Exception as e:
error_str = str(e)
if "EMAIL_NOT_FOUND" in error_str:
st.sidebar.error("โ No account found with this email address")
st.sidebar.warning("โ Don't have an account? Please sign up.")
else:
st.sidebar.error(f"Error: {error_str}")
# In your login_signup_ui() function, replace the Google login section with:
st.sidebar.markdown("---")
st.sidebar.markdown("### Or continue with")
if st.sidebar.button("Continue with Google", key="google_login"):
try:
# Generate a unique state token
state_token = hashlib.sha256(str(time.time()).encode()).hexdigest()
# Get Firebase config values
client_id = "546645596018-nvtkegm7mi8e83upfv771tv6t58c7snn.apps.googleusercontent.com" # Use actual OAuth client ID
firebase_auth_domain = firebaseConfig["authDomain"]
redirect_uri = f"https://{firebase_auth_domain}/auth/handler"
# Build the Google OAuth URL
auth_url = (
f"https://accounts.google.com/o/oauth2/v2/auth?"
f"response_type=code&"
f"client_id={client_id}&"
f"redirect_uri={urllib.parse.quote(redirect_uri)}&"
f"scope=email%20profile%20openid&"
f"state={state_token}"
)
# Store state token in session
st.session_state.oauth_state = state_token
# Open OAuth flow in new tab
components.html(
f"""
""",
height=0
)
st.sidebar.info("Google login window should open. Please allow popups if it doesn't appear.")
except Exception as e:
st.sidebar.error(f"Failed to start Google login: {str(e)}")
def check_google_callback():
try:
if 'code' in st.query_params and 'state' in st.query_params:
# Verify state token matches
if st.query_params['state'] != st.session_state.get('oauth_state'):
st.error("Security verification failed. Please try logging in again.")
return
auth_code = st.query_params['code']
# Initialize the Google Auth Provider
provider = firebase.auth.GoogleAuthProvider()
# Sign in with the auth code
credential = provider.credential(
None, # No ID token needed for code flow
auth_code
)
# Sign in with credential
user = auth.sign_in_with_credential(credential)
# Store user in session
st.session_state.logged_in = True
st.session_state.user_email = user['email']
st.session_state.user_name = user.get('displayName', 'Google User')
st.session_state.user_token = user['idToken']
# Store user data if new
if not user_exists(user['localId']):
store_user_data(
user['localId'],
st.session_state.user_name,
user['email']
)
# Clear the OAuth code from URL
st.query_params.clear()
st.rerun()
except Exception as e:
st.error(f"Google login failed: {str(e)}")
# Initialize speech-to-text model (cached to avoid reloading)
# @st.cache_resource
# def load_speech_to_text_model():
# return pipeline("automatic-speech-recognition", model="openai/whisper-base")
# Function to translate text
def transcribe_audio(audio_bytes, auto_detect=False):
"""Transcribe audio with auto language detection (English/Hindi)"""
if not SPEECHMATICS_API_KEY:
st.error("API key not configured!")
return None
try:
API_BASE_URL = "https://asr.api.speechmatics.com/v2"
MAX_FILE_SIZE = 5 * 1024 * 1024 # 5MB free tier limit
TIMEOUT = 30 # seconds
# 1. Validate audio size
if len(audio_bytes) > MAX_FILE_SIZE:
st.error(f"Audio exceeds {MAX_FILE_SIZE/1024/1024}MB free tier limit")
return None
# 2. Create temp WAV file
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmpfile:
tmpfile.write(audio_bytes)
tmp_path = tmpfile.name
# 3. Configure language settings
language_config = {
"language": "auto" if auto_detect else st.session_state.get('language', 'en')
}
# For auto-detect, specify possible languages (improves accuracy)
if auto_detect:
language_config["language_options"] = ["en", "hi"] # English/Hindi only
job_config = {
"type": "transcription",
"transcription_config": {
**language_config,
"operating_point": "standard",
"enable_entities": False
}
}
headers = {"Authorization": f"Bearer {SPEECHMATICS_API_KEY}"}
# 4. Create job
with open(tmp_path, 'rb') as audio_file:
response = requests.post(
f"{API_BASE_URL}/jobs",
headers=headers,
files={
'config': (None, json.dumps(job_config)),
'data_file': ('audio.wav', audio_file)
},
timeout=TIMEOUT
)
# 5. Handle response
if response.status_code != 201:
error_msg = response.json().get('error', {}).get('message', response.text)
st.error(f"Job creation failed: {error_msg}")
return None
job_id = response.json()['id']
st.session_state.current_job_id = job_id
# 6. Poll for completion
start_time = time.time()
detected_language = None
while True:
if time.time() - start_time > TIMEOUT:
raise Exception("Timeout waiting for transcription")
status_response = requests.get(
f"{API_BASE_URL}/jobs/{job_id}",
headers=headers,
timeout=TIMEOUT
)
status_data = status_response.json()
# Capture detected language if auto mode
if auto_detect and not detected_language:
detected_language = status_data['job'].get('detected_language')
if detected_language:
st.info(f"๐ Detected language: {detected_language.upper()}")
if status_data['job']['status'] == 'done':
break
elif status_data['job']['status'] == 'failed':
raise Exception(f"Transcription failed: {status_data.get('error')}")
time.sleep(2)
# 7. Get transcript
transcript_response = requests.get(
f"{API_BASE_URL}/jobs/{job_id}/transcript",
headers=headers,
params={'format': 'txt'},
timeout=TIMEOUT
)
if transcript_response.status_code != 200:
st.error(f"Failed to fetch transcript: {transcript_response.text}")
return None
return transcript_response.text
except Exception as e:
st.error(f"Transcription error: {str(e)}")
return None
finally:
if 'tmp_path' in locals() and os.path.exists(tmp_path):
try:
os.unlink(tmp_path)
except:
pass
def check_rate_limit():
"""Simple rate limiting for audio transcription"""
if 'last_transcription_time' not in st.session_state:
st.session_state.last_transcription_time = time.time()
return True
current_time = time.time()
time_since_last = current_time - st.session_state.last_transcription_time
if time_since_last < 10: # 10 second cooldown between transcriptions
return False
st.session_state.last_transcription_time = current_time
return True
def create_new_chat():
"""Properly reset the chat state and start a new chat session."""
# Save current chat if it has messages
if len(st.session_state.get('messages', [])) > 1: # More than just welcome message
save_current_chat()
# Reset conversation state
st.session_state.messages = [{
"role": "assistant",
"content": f"๐โจ Welcome {st.session_state.user_name}! โจ๐\n\nI'm ApnaLawyer, your AI legal assistant. "
"I can help explain Indian laws in simple terms. What would you like to know?"
}]
st.session_state.current_chat_id = None
st.session_state.memory = ConversationBufferWindowMemory(k=2, memory_key="chat_history", return_messages=True)
# Clear any audio processing flags
if 'audio_processed' in st.session_state:
del st.session_state.audio_processed
st.rerun()
def save_current_chat():
"""Save the current chat to history before starting a new one."""
if len(st.session_state.get('messages', [])) > 1: # More than just welcome message
chat_title = generate_chat_title(st.session_state.messages)
if hasattr(st.session_state, 'current_chat_id') and st.session_state.current_chat_id:
update_chat_history(st.session_state.user_id, st.session_state.current_chat_id, st.session_state.messages)
else:
chat_id = save_chat_to_history(st.session_state.user_id, chat_title, st.session_state.messages)
st.session_state.current_chat_id = chat_id
st.session_state.chat_history = get_chat_history(st.session_state.user_id)
def load_chat(chat_id):
"""Load a specific chat from history."""
# Save current chat if it has messages
if len(st.session_state.get('messages', [])) > 1: # More than just welcome message
save_current_chat()
# Load the selected chat
chat_data = next((chat for chat in st.session_state.chat_history if chat[0] == chat_id), None)
if chat_data:
st.session_state.messages = chat_data[1]['messages']
st.session_state.current_chat_id = chat_id
# Reinitialize memory with loaded messages
st.session_state.memory = ConversationBufferWindowMemory(k=2, memory_key="chat_history", return_messages=True)
for msg in chat_data[1]['messages'][:-2]: # Skip last 2 messages to maintain window size
if msg['role'] == 'user':
st.session_state.memory.save_context({"question": msg['content']}, {"answer": ""})
st.rerun()
def delete_chat(chat_id):
"""Delete a specific chat from the user's chat history."""
try:
user_id = st.session_state.user_id
if delete_chat_history(user_id, chat_id):
# Show toast notification instead of message in chat history
st.toast("Chat deleted successfully!", icon="โ
")
# Refresh chat history after deletion
st.session_state.chat_history = get_chat_history(user_id)
st.rerun() # Force UI refresh
else:
st.toast("Failed to delete chat.", icon="โ")
except Exception as e:
st.toast(f"Error deleting chat: {str(e)}", icon="โ")
def chatbot_ui():
# Initialize language state
target_lang_code = "en" # Default language is English
if "language" not in st.session_state:
st.session_state.language = target_lang_code # Default language is English
# Initialize with personalized welcome message if first time
if "messages" not in st.session_state:
st.session_state.messages = [{
"role": "assistant",
"content": f"๐โจ Welcome {st.session_state.user_name}! โจ๐\n\nI'm ApnaLawyer, your AI legal assistant. "
"I can help explain Indian laws in simple terms. What would you like to know?"
}]
if "memory" not in st.session_state:
st.session_state.memory = ConversationBufferWindowMemory(k=5, memory_key="chat_history", return_messages=True)
# Get user ID
if "user_id" not in st.session_state:
try:
user_info = auth.get_account_info(st.session_state.user_token)
st.session_state.user_id = user_info['users'][0]['localId']
st.session_state.chat_history = get_chat_history(st.session_state.user_id)
except Exception as e:
st.error(f"Error getting user info: {str(e)}")
@st.cache_resource
def load_embeddings():
return HuggingFaceEmbeddings(model_name="law-ai/InLegalBERT")
embeddings = load_embeddings()
db = FAISS.load_local("ipc_embed_db", embeddings, allow_dangerous_deserialization=True)
db_retriever = db.as_retriever(search_type="similarity", search_kwargs={"k": 3})
# Define the prompt template
prompt_template = PromptTemplate(
input_variables=["context", "question", "chat_history"],
template="""