import sys import json import joblib import argparse import re import os import mysql.connector from dotenv import load_dotenv import locale import google.generativeai as genai from transformers import AutoTokenizer, AutoModel import numpy as np try: locale.setlocale(locale.LC_ALL, 'id_ID.UTF-8') except locale.Error: locale.setlocale(locale.LC_ALL, '') load_dotenv() try: GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") if not GEMINI_API_KEY: genai_configured = False else: genai.configure(api_key=GEMINI_API_KEY) genai_configured = True except Exception as e: genai_configured = False try: tokenizer = AutoTokenizer.from_pretrained("indobenchmark/indobert-base-p1") bert_model = AutoModel.from_pretrained("indobenchmark/indobert-base-p1") bert_model.eval() except Exception: tokenizer = None bert_model = None def get_property_detail(ref_id: str, intent: str) -> str: detail_map = { "tanya_nama_properti": ("c.cluster_apart_name", "Nama properti ini adalah"), "tanya_jumlah_lantai": ("c.property_floor", "Properti ini memiliki"), "tanya_harga": ("c.property_price", "Harga properti ini adalah"), "tanya_luas_bangunan": ("c.square_building", "Luas bangunannya adalah"), "tanya_luas_tanah": ("c.square_land", "Luas tanahnya adalah"), "tanya_lokasi_gmaps": (["c.latitude", "c.longitude"], "Tentu, ini lokasi propertinya di Google Maps:"), "tanya_dokumen": ("doc.name", "Status dokumen/sertifikat properti ini adalah"), "tanya_tipe_aset": ("at.name", "Tipe aset ini adalah"), "tanya_kondisi_bangunan": ("conb.name", "Kondisi bangunannya saat ini"), "tanya_kondisi_jalan": ("cr.name", "Kelas jalan di sekitar properti adalah"), "tanya_lalu_lintas": ("tv.name", "Volume lalu lintas di area tersebut tergolong"), "tanya_banjir": ("pf.name", "Untuk risiko banjir, lokasinya tercatat"), "tanya_penghuni": ("ob.name", "Tingkat hunian properti saat ini sekitar"), } if intent not in detail_map: return "Maaf, saya tidak mengerti detail apa yang Anda tanyakan." column_info, response_prefix = detail_map[intent] try: conn = mysql.connector.connect(host=os.getenv("SERVER_NAME"), user=os.getenv("USER_NAME"), password=os.getenv("PASSWORD"), database=os.getenv("DATABASE")) cursor = conn.cursor() columns_str = ", ".join(column_info) if isinstance(column_info, list) else column_info query = f"SELECT {columns_str} FROM contribution c LEFT JOIN document_m doc ON c.property_document = doc.code LEFT JOIN asset_type_m at ON c.asset_type = at.code LEFT JOIN condition_building_m conb ON c.condition_code = conb.code LEFT JOIN class_road_m cr ON c.streetclass_code = cr.code LEFT JOIN traffic_volume_m tv ON c.traffic_code = tv.code LEFT JOIN possible_flooding_m pf ON c.flood_code = pf.code LEFT JOIN occupancy_building_m ob ON c.occupancy_code = ob.code WHERE c.ref_id = %s" cursor.execute(query, (ref_id,)) result = cursor.fetchone() conn.close() if not result: return "Maaf, detail tersebut tidak tersedia untuk properti ini." if intent == 'tanya_lokasi_gmaps': lat, lon = result[0], result[1] return f"{response_prefix}\nhttp://maps.google.com/maps?q={lat},{lon}" if lat and lon else "Maaf, data koordinat tidak tersedia." else: value = result[0] if value is None: return "Maaf, detail tersebut tidak tersedia." if intent == 'tanya_harga': return f"{response_prefix} {locale.currency(float(value), grouping=True)}." elif intent in ['tanya_luas_bangunan', 'tanya_luas_tanah']: return f"{response_prefix} {value} m2." elif intent == 'tanya_jumlah_lantai': return f"{response_prefix} {value} lantai." else: return f"{response_prefix} {value}." except Exception: return "Terjadi kesalahan saat mengambil data detail." def retrieve_similar(city=None, price_max=None, prop_type=None, top_k=3): try: conn = mysql.connector.connect(host=os.getenv("SERVER_NAME"), user=os.getenv("USER_NAME"), password=os.getenv("PASSWORD"), database=os.getenv("DATABASE")) cursor = conn.cursor(dictionary=True) query = "SELECT c.ref_id, c.cluster_apart_name, c.address, c.city, c.property_price FROM contribution c WHERE 1=1 AND c.status='STT01' " params = [] if city: query += " AND (c.city LIKE %s OR c.province LIKE %s)" params.extend([f"%{city}%", f"%{city}%"]) if price_max: query += " AND c.property_price <= %s" params.append(price_max) if prop_type: query += " AND c.asset_type = (SELECT code FROM asset_type_m WHERE name LIKE %s LIMIT 1)" params.append(f"%{prop_type}%") query += " ORDER BY c.property_price DESC LIMIT %s" params.append(top_k) cursor.execute(query, tuple(params)) results = cursor.fetchall() conn.close() if not results: return [], "Maaf, tidak ada properti yang ditemukan dengan kriteria tersebut." response_parts = [] for i, prop in enumerate(results): harga_formatted = locale.currency(float(prop['property_price']), grouping=True) if prop['property_price'] else 'N/A' response_parts.append(f"\n{i+1}. {prop['cluster_apart_name']}\n Lokasi: {prop['address']}, {prop['city']}\n Harga: {harga_formatted}") response_parts.append("\nAnda bisa menanyakan detail seperti 'dimana lokasinya?' untuk properti nomor 1.") return results, "\n".join(response_parts) except Exception: return [], "Terjadi kesalahan saat mencari properti." def hitung_kalkulator_kpr(text: str) -> str: harga_match = re.search(r'(\d+(?:\.\d+)?)\s*(miliar|m|juta|jt)', text) if not harga_match: return "Mohon sebutkan harga rumah (contoh: 'hitung kpr rumah 1 miliar')." harga_rumah = float(harga_match.group(1)) * (10**9 if harga_match.group(2) in ['miliar', 'm'] else 10**6) dp_match = re.search(r'(dp|uang muka)\s*(\d+)', text) tenor_match = re.search(r'(\d+)\s*(tahun|thn)', text) dp_persen = float(dp_match.group(2)) if dp_match else 10 tenor_tahun = int(tenor_match.group(1)) if tenor_match else 15 suku_bunga = 6.5 pokok_pinjaman = harga_rumah * (1 - dp_persen / 100) suku_bunga_bulanan = (suku_bunga / 100) / 12 jumlah_bulan = tenor_tahun * 12 pembilang = pokok_pinjaman * suku_bunga_bulanan * (1 + suku_bunga_bulanan)**jumlah_bulan penyebut = (1 + suku_bunga_bulanan)**jumlah_bulan - 1 angsuran = pembilang / penyebut if penyebut != 0 else pokok_pinjaman / jumlah_bulan return f"Berikut hasil perhitungan KPR:\nHarga Rumah: {locale.currency(harga_rumah, grouping=True)}\nDP ({dp_persen}%): {locale.currency(harga_rumah * (dp_persen/100), grouping=True)}\nJangka Waktu: {tenor_tahun} tahun\n?? Estimasi Angsuran per Bulan: **{locale.currency(angsuran, grouping=True)}**" def hitung_simulasi_kemampuan(text: str) -> str: gaji_match = re.search(r'(\d+(?:\.\d+)?)\s*(juta|jt)', text) if not gaji_match: return "Mohon sebutkan gaji bulanan Anda (contoh: 'gaji 15 juta')." gaji_bulanan = float(gaji_match.group(1)) * 1_000_000 maks_cicilan = gaji_bulanan * 0.35 suku_bunga, tenor_tahun = 7.0, 20 suku_bunga_bulanan = (suku_bunga / 100) / 12 jumlah_bulan = tenor_tahun * 12 pembilang = maks_cicilan * ((1 + suku_bunga_bulanan)**jumlah_bulan - 1) penyebut = suku_bunga_bulanan * (1 + suku_bunga_bulanan)**jumlah_bulan maks_pokok_pinjaman = pembilang / penyebut estimasi_harga_rumah = maks_pokok_pinjaman / 0.90 _, rekomendasi_text = retrieve_similar(price_max=estimasi_harga_rumah, top_k=2) return f"Berdasarkan gaji {locale.currency(gaji_bulanan, grouping=True)}/bulan, berikut simulasi kemampuan KPR Anda:\n\nCicilan Maksimal per Bulan: **{locale.currency(maks_cicilan, grouping=True)}**\nEstimasi Harga Rumah: **{locale.currency(estimasi_harga_rumah, grouping=True)}**\n\nBerikut adalah beberapa rekomendasi properti yang mungkin cocok:\n{rekomendasi_text}" def extract_entities(text: str, available_locations: list): entities = {} text_lower = text.lower() for loc in available_locations: if re.search(r'\b' + re.escape(loc.lower()) + r'\b', text_lower): entities['lokasi'] = loc break price_match = re.search(r'(di bawah|maksimal|maks|sekitar|harga|<|>) (\d+(?:\.\d+)?)\s*(miliar|m|juta|jt)', text_lower) if price_match: limit_type, value, unit = price_match.groups() multiplier = 10**9 if unit in ['miliar', 'm'] else 10**6 price_value = float(value) * multiplier if limit_type in ['di bawah', 'maksimal', 'maks', '<']: entities['harga_maksimal'] = price_value else: entities['harga_sekitar'] = price_value property_types = {'rumah': ['rumah', 'hunian'], 'apartemen': ['apartemen'], 'ruko': ['ruko'], 'tanah': ['tanah', 'kavling']} for prop_type, keywords in property_types.items(): if any(keyword in text_lower for keyword in keywords): entities['jenis_properti'] = prop_type break return entities def get_bert_embedding(texts): if not bert_model or not tokenizer: return np.zeros((len(texts), 768)) embeddings = [] for text in texts: inputs = tokenizer(text, return_tensors="pt", truncation=True, padding=True, max_length=128) with torch.no_grad(): outputs = bert_model(**inputs) cls_embedding = outputs.last_hidden_state[:, 0, :].squeeze().numpy() embeddings.append(cls_embedding) return np.array(embeddings) def enhance_answer_with_gemini(user_question: str, factual_data: str) -> str: if not genai_configured: return factual_data try: model = genai.GenerativeModel('gemini-pro') prompt = f"Anda adalah PropertiBot, asisten properti yang ramah. Ubah data faktual berikut menjadi jawaban yang natural dalam Bahasa Indonesia, tanpa mengubah fakta (angka, nama, dll).\n\nPertanyaan Pengguna: '{user_question}'\nData Faktual dari Bot: '{factual_data}'\n\nJawaban Anda:" response = model.generate_content(prompt) return response.text except Exception: return factual_data def get_gemini_fallback_response(user_question: str) -> str: if not genai_configured: return "Maaf, saya adalah PropertiBot dan hanya bisa membantu terkait properti." try: model = genai.GenerativeModel('gemini-pro') prompt = f"Anda adalah PropertiBot. Pengguna bertanya sesuatu di luar topik properti. Jawab pertanyaannya dengan singkat, lalu arahkan kembali ke topik properti.\n\nPertanyaan: '{user_question}'\n\nJawaban Anda:" response = model.generate_content(prompt) if "properti" not in response.text[-50:]: return response.text + "\n\nAda lagi yang bisa saya bantu seputar properti?" return response.text except Exception: return "Maaf, terjadi kendala. Ada yang bisa saya bantu terkait properti?" def analyze_message(msg, user_context, intent_model, vectorizer, scaler, label_encoder, available_locations): X_tfidf = vectorizer.transform([msg.lower()]).toarray() X_bert = get_bert_embedding([msg.lower()]) X_combined = np.concatenate([X_tfidf, X_bert], axis=1) X_scaled = scaler.transform(X_combined) intent_encoded = intent_model.predict(X_scaled)[0] intent = label_encoder.inverse_transform([intent_encoded])[0] entities = extract_entities(msg, available_locations) factual_answer = None user_context.setdefault('entities', {}).update(entities) if intent == "sapa": factual_answer = "Halo! Ada yang bisa saya bantu terkait pencarian properti atau simulasi KPR?" elif intent == "cari_properti": current_entities = user_context.get('entities', {}) prop_type = current_entities.get('jenis_properti') location = current_entities.get('lokasi') price_max = current_entities.get('harga_maksimal') if not location: factual_answer = f"Tentu, Anda ingin mencari {prop_type or 'properti'} di kota atau provinsi mana?" else: search_desc = f"{prop_type or 'properti'} di {location.title()}" if price_max: search_desc += f" dengan harga di bawah {locale.currency(price_max, grouping=True)}" properties, response_text = retrieve_similar(city=location, price_max=price_max, prop_type=prop_type) if properties: user_context["last_properties"] = properties factual_answer = f"Baik, saya carikan {search_desc}.\n\n{response_text}" else: factual_answer = response_text elif intent.startswith("tanya_") and user_context.get("last_properties"): ref_id = user_context["last_properties"][0]["ref_id"] factual_answer = get_property_detail(ref_id, intent) elif intent == "kalkulator_kpr": factual_answer = hitung_kalkulator_kpr(msg) elif intent == "simulasi_kemampuan_kpr": factual_answer = hitung_simulasi_kemampuan(msg, available_locations) elif intent == "keluar": factual_answer = "Terima kasih, sampai jumpa lagi!" user_context = {} if factual_answer: final_answer = enhance_answer_with_gemini(msg, factual_answer) elif intent == "fallback": final_answer = get_gemini_fallback_response(msg) else: final_answer = "Maaf, saya kurang mengerti. Bisa coba tanyakan hal lain seputar properti?" return {"intent": intent, "answer": final_answer, "context": user_context, "entities": entities} def main(): parser = argparse.ArgumentParser(description="Chatbot Properti CLI") parser.add_argument("--message", required=True) parser.add_argument("--context", default='{}') args = parser.parse_args() try: intent_model = joblib.load('intent_model.pkl') vectorizer = joblib.load('vectorizer.pkl') scaler = joblib.load('scaler.pkl') label_encoder = joblib.load('label_encoder.pkl') conn = mysql.connector.connect(host=os.getenv("SERVER_NAME"), user=os.getenv("USER_NAME"), password=os.getenv("PASSWORD"), database=os.getenv("DATABASE")) cursor = conn.cursor() cursor.execute("SELECT DISTINCT city FROM contribution WHERE city IS NOT NULL AND city != '' UNION SELECT DISTINCT province FROM contribution WHERE province IS NOT NULL AND province != ''") available_locations = [row[0].strip() for row in cursor.fetchall()] conn.close() except Exception as e: print(json.dumps({"error": f"Gagal memuat model atau data: {e}"})) sys.exit(1) try: user_context = json.loads(args.context) except json.JSONDecodeError: user_context = {} result = analyze_message(args.message, user_context, intent_model, vectorizer, scaler, label_encoder, available_locations) print(json.dumps(result, default=str)) if __name__ == "__main__": main()