| import json
|
| import re
|
| import os
|
| import joblib
|
| import pandas as pd
|
| from Sastrawi.Stemmer.StemmerFactory import StemmerFactory
|
|
|
|
|
|
|
|
|
| import os
|
|
|
| GEMINI_ENABLED = False
|
| gemini_client = None
|
|
|
| try:
|
| from google import genai
|
|
|
| api_key = os.getenv("GEMINI_API_KEY")
|
|
|
| if api_key:
|
| gemini_client = genai.Client(api_key=api_key)
|
| GEMINI_ENABLED = True
|
| print("Gemini chatbot enabled ✅")
|
| else:
|
| print("No GEMINI_API_KEY found — fallback to rules only.")
|
|
|
| except Exception as e:
|
| print("Gemini init failed:", e)
|
| GEMINI_ENABLED = False
|
| gemini_client = None
|
|
|
|
|
|
|
|
|
| try:
|
| from inference_core import (
|
| predict_price,
|
| calibrate_prediction,
|
| recommend_locations,
|
| get_recommended_area_examples,
|
| get_target_area_overview,
|
| )
|
| except Exception as e:
|
| print("chatbot import error:", e)
|
|
|
|
|
|
|
|
|
|
|
| bandung_listings = pd.read_csv("bandung_listing_features.csv")
|
|
|
| factory = StemmerFactory()
|
| stemmer = factory.create_stemmer()
|
|
|
|
|
|
|
|
|
|
|
| def fmt_price(val_juta):
|
| if val_juta is None or pd.isna(val_juta):
|
| return "—"
|
| if val_juta >= 1000:
|
| return f"Rp {val_juta/1000:.2f} Miliar"
|
| return f"Rp {val_juta:,.0f} Juta"
|
|
|
|
|
| def extract_location(text):
|
| lowered = text.lower()
|
| for loc in bandung_listings["location"].dropna().unique():
|
| if loc.lower() in lowered:
|
| return loc
|
| return None
|
|
|
|
|
| def extract_number(text):
|
| match = re.search(r"(\d[\d.,]*)", text)
|
| if not match:
|
| return None
|
| return float(match.group(1).replace(",", "."))
|
|
|
|
|
| def extract_bedrooms(text):
|
| match = re.search(r"(\d+)\s*(kamar|bed)", text.lower())
|
| if match:
|
| return int(match.group(1))
|
| return None
|
|
|
|
|
| def gemini_chat_response(user_msg, history):
|
| if not GEMINI_ENABLED or gemini_client is None:
|
| return None
|
|
|
| try:
|
| prompt = f"""
|
| You are a smart Bandung housing advisor.
|
|
|
| User: {user_msg}
|
|
|
| Rules:
|
| - Respond in natural, complete sentences
|
| - Be helpful and specific (location, price range, or insight)
|
| - Keep it concise but NOT robotic
|
| - Avoid one-word or vague answers
|
| - Max 2–3 sentences
|
| - Under 60 words
|
| """
|
|
|
| if history:
|
| for turn in history[-4:]:
|
| prompt += f"{turn['role']}: {turn['content']}\n"
|
|
|
| prompt += f"user: {user_msg}"
|
|
|
| response = gemini_client.models.generate_content(
|
| model="gemini-2.5-flash-lite",
|
| contents=prompt,
|
| config={
|
| "max_output_tokens": 60,
|
| "temperature": 0.3,
|
| "top_p": 0.95
|
| }
|
| )
|
|
|
| if hasattr(response, "text") and response.text:
|
| return response.text.strip()
|
|
|
| except Exception as e:
|
| print("Gemini failed, fallback to rules:", e)
|
|
|
| return None
|
|
|
|
|
|
|
|
|
|
|
| EMPTY_STATE = json.dumps(
|
| {
|
| "location": None,
|
| "bedroom_total": None,
|
| "building_area_m2": None,
|
| "budget": None,
|
| }
|
| )
|
|
|
|
|
|
|
|
|
|
|
| def rule_chat(user_msg, history, state_str):
|
| state = json.loads(state_str)
|
|
|
| text = user_msg.strip()
|
|
|
| location = extract_location(text)
|
| bedrooms = extract_bedrooms(text)
|
| number = extract_number(text)
|
|
|
| if location:
|
| state["location"] = location
|
|
|
| if bedrooms:
|
| state["bedroom_total"] = bedrooms
|
|
|
| if number and "budget" in text.lower():
|
| state["budget"] = number
|
|
|
| lowered = text.lower()
|
|
|
|
|
|
|
|
|
| if any(w in lowered for w in ["harga", "price", "estimasi", "berapa"]):
|
| if not state.get("location"):
|
| bot_msg = "Area mana yang ingin dicek? Contoh: Dago, Buah Batu."
|
| elif not state.get("bedroom_total"):
|
| bot_msg = "Berapa kamar tidur?"
|
| else:
|
| try:
|
| features = {
|
| "location": state["location"],
|
| "building_type": "Rumah",
|
| "bedroom_total": state["bedroom_total"],
|
| "bathroom_total": 2,
|
| "carport_total": 1,
|
| "building_area_m2": 120,
|
| }
|
|
|
| pred = predict_price(features)
|
| result = calibrate_prediction(pred)
|
|
|
| bot_msg = (
|
| f"📍 **{state['location']}**\n"
|
| f"💰 Estimasi harga: **{fmt_price(result['Predicted Price (IDR Mn)'])}**\n"
|
| f"📉 Range: {fmt_price(result['Lower Range'])} — {fmt_price(result['Upper Range'])}"
|
| )
|
|
|
| except Exception as e:
|
| bot_msg = f"Prediction error: {e}"
|
|
|
|
|
|
|
|
|
| elif any(w in lowered for w in ["alternatif", "recommend", "mirip"]):
|
| if not state.get("location"):
|
| bot_msg = "Area targetnya apa dulu?"
|
| else:
|
| try:
|
| recs = recommend_locations(
|
| {"location": state["location"]},
|
| top_n=5
|
| )
|
|
|
| lines = [f"📍 Area mirip dengan **{state['location']}**:"]
|
| for _, row in recs.iterrows():
|
| lines.append(
|
| f"- {row['Alternative Area']} "
|
| f"({row['Similarity Score']:.3f})"
|
| )
|
|
|
| bot_msg = "\n".join(lines)
|
|
|
| except Exception as e:
|
| bot_msg = f"Recommendation error: {e}"
|
|
|
|
|
|
|
|
|
| else:
|
| bot_msg = (
|
| "Hi, aku bisa bantu:\n"
|
| "- estimasi harga properti\n"
|
| "- area alternatif\n"
|
| "- market context Bandung\n\n"
|
| "Contoh: *harga rumah 3 kamar di Dago*"
|
| )
|
|
|
| history = history + [
|
| {"role": "user", "content": user_msg},
|
| {"role": "assistant", "content": bot_msg},
|
| ]
|
|
|
| return history, json.dumps(state)
|
|
|
|
|
|
|
|
|
|
|
| def chat(user_msg, history, state_str):
|
| if history is None:
|
| history = []
|
|
|
| if not state_str:
|
| state_str = EMPTY_STATE
|
|
|
| text = user_msg.strip()
|
|
|
| if not text:
|
| return history, state_str
|
|
|
|
|
| gemini_reply = gemini_chat_response(user_msg, history)
|
| print("GEMINI CHAT REPLY:", gemini_reply)
|
|
|
| if gemini_reply and len(gemini_reply.split()) >= 5:
|
| history = history + [
|
| {"role": "user", "content": user_msg},
|
| {"role": "assistant", "content": gemini_reply},
|
| ]
|
| return history, state_str
|
|
|
|
|
| return rule_chat(user_msg, history, state_str) |
