import ast
import re
import pandas as pd
import tempfile
import os
from intent_classification.fd_classification import find_intent
from intent_classification.retrieval_classification import handle_query_classification,find_matching_card,generate_card_response_with_context
from recommender.retrieval_ranking import generate_multi_queries,convert_to_direct_query_gradio,retrieve_and_rank_cards,generate_credit_card_recommendation_gemini,cross_encoder
from data import eligibility_lookup,card_features_lookup,all_card_names,features
from recommender.graph_retrieval_vectordb import generate_cypher,run_cypher_query,Neo4jConnectionError
#function to pass the retrieved cards and generated response to the UI
def recommend_cards_gradio(user_query, preferences, income, cibil, age,
min_joining_fee, max_joining_fee,
min_annual_fee, max_annual_fee,
use_eligibility=True,):
try:
# print(user_query)
if(user_query):
result = handle_query_classification(user_query)
if result["intent"] == "no_retrieval":
return (
f"
",
[["No retrieval required", "Answered using LLM"]],
None,
[],
{},
"Answered without retrieval"
)
elif result["intent"] == "specific":
matched_card = find_matching_card(user_query)
if matched_card:
gemini_answer = generate_card_response_with_context(user_query, matched_card)
card_name = matched_card["name"]
card_desc = matched_card["description"]
card_lookup = {card_name: card_desc}
# Constructing eligibility info if available
eligibility_info = eligibility_lookup.get(card_name, "No eligibility or fee information available.")
chat_history_entry = f"{card_name}:\n{card_desc}\n\nEligibility & Fees:\n{eligibility_info}"
return (
f"",
[["Specific card detected", card_name]],
None,
[],
card_lookup,
user_query
)
else:
return (
"Card mentioned not found in database.",
[["Card not found", "Try another card name."]],
None,
[],
{},
"Card not found"
)
direct_query,excluded_cards = convert_to_direct_query_gradio(user_query, preferences, all_card_names=all_card_names,feature_list=features)
queries = generate_multi_queries(direct_query)
if cibil < 700 and use_eligibility:
query_intent = True
else:
query_intent = find_intent(user_query)
print(query_intent)
cypher_query = generate_cypher(direct_query, query_intent)
print("Generated Cypher:\n", cypher_query)
try:
faiss_index, filtered_mapping = run_cypher_query(
user_query, cypher_query, use_eligibility,
income, cibil, age,
min_joining_fee, max_joining_fee,
min_annual_fee, max_annual_fee,excluded_cards
)
except Neo4jConnectionError as graph_err:
return (
"Graph database connection failed. Please try again later.",
[["Graph database error", str(graph_err)]],
None,
[],
{},
"Graph DB connection error"
)
cards = retrieve_and_rank_cards(faiss_index, filtered_mapping, direct_query, queries, top_k=10)
gemini_summary = generate_credit_card_recommendation_gemini(user_query, direct_query, cards)
if not cards:
return (
"No eligible cards found.",
[["No eligible cards found", "Please try a different query or check your input values."]],
None,
[],
{},
"No eligible card found"
)
match = re.search(r"f\.name IN (\[.*?\])", cypher_query)
query_features = set(ast.literal_eval(match.group(1))) if match else set()
card_rows = []
for score, card in sorted(
zip(cross_encoder.predict([[direct_query, card["description"]] for card in cards]), cards),
reverse=True,
key=lambda x: x[0]
):
card_name = card["name"]
card_desc = card["description"]
matched_features = query_features.intersection(card_features_lookup.get(card_name, set()))
feature_str = ", ".join(matched_features) if matched_features else "None"
card_rows.append([card_name, feature_str, card_desc])
card_names = [row[0] for row in card_rows]
card_lookup = {row[0]: row[2] for row in card_rows}
top_card_html = f"""
"""
df_cards = pd.DataFrame(card_rows, columns=["Card Name", "Matched Features", "Description"])
filename = "recommended_cards.csv"
temp_dir = tempfile.gettempdir()
file_path = os.path.join(temp_dir, filename)
df_cards.to_csv(file_path, index=False)
return top_card_html, card_rows, file_path, card_names, card_lookup, direct_query
except Exception as e:
print("Error:", e)
return (
"An unexpected error occurred. Please try again in a few minutes.",
[["Something went wrong", "Please try again."]],
None,
[],
{},
"Unexpected error occurred, please try again in a while"
)