"""
HomeMatch AI — Hugging Face Space (Gradio)
==========================================
AI-powered real-estate recommendations with embeddings + grounded generation,
in a polished real-estate-style UI with illustrative property imagery.
- Artifacts read from HF Dataset repo: BarWachsman7/HomeMatch-AI-Dataset
(local file fallback supported). Files/models cached so they load once.
- Embedding model name read from homematch_embedding_model_info.json and loaded
with SentenceTransformer (HF model hub). Filters applied BEFORE cosine ranking.
- Generation: google/flan-t5-base (fallback google/flan-t5-small), deterministic,
grounded only in selected listing facts. Banned broker/agent/agency/realtor
wording is detected & cleaned.
- Images live in ./assets and are assigned by property type (illustrative only).
- top_k is capped at 4 in the UI to match 4 image sets per property type and
avoid repeated image sets in the same search.
No OpenAI, no paid APIs, no secret tokens.
"""
import os
import re
import io
import json
import base64
import functools
import numpy as np
import pandas as pd
import gradio as gr
# --------------------------------------------------------------------------- #
# Configuration
# --------------------------------------------------------------------------- #
DATASET_REPO = "BarWachsman7/HomeMatch-AI-Dataset"
F_METADATA = "homematch_embedding_metadata.csv"
F_EMB = "homematch_embeddings.npy"
F_MODEL = "homematch_embedding_model_info.json"
PRIMARY_GEN_MODEL = "google/flan-t5-base"
FALLBACK_GEN_MODEL = "google/flan-t5-small"
DEFAULT_EMBED_MODEL = "sentence-transformers/multi-qa-MiniLM-L6-cos-v1"
BANNED_TERMS = [
"broker", "brokers", "agent", "agents",
"agency", "agencies", "realtor", "realtors"
]
_BANNED_RE = re.compile(
r"\b(brokers?|agents?|agenc(?:y|ies)|realtors?)\b",
re.IGNORECASE
)
REC_COLUMNS = [
"rank", "similarity_score", "listing_id", "city", "neighborhood",
"property_type", "rooms", "size_sqm", "price_million_nis",
"target_audience", "has_elevator", "has_parking", "has_balcony",
"renovated", "near_schools", "distance_to_transport", "owner_name"
]
ASSET_ROOT = "assets"
PTYPE_TO_FOLDER = {
"Apartment": "apartment",
"Studio": "studio",
"Private House": "private_house",
"Penthouse": "penthouse",
"Duplex": "duplex",
"Garden Apartment": "garden_apartment",
}
IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".jfif"}
# --------------------------------------------------------------------------- #
# Small utilities
# --------------------------------------------------------------------------- #
def to_bool(value):
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
return value != 0
return str(value).strip().lower() in {"true", "1", "yes", "y", "t"}
def _as_list(v):
return list(v) if isinstance(v, (list, tuple, set)) else [v]
def contains_banned_terms(text):
return bool(_BANNED_RE.search(str(text)))
def clean_banned_terms(text):
return re.sub(r"\s{2,}", " ", _BANNED_RE.sub("owner", str(text))).strip()
def generation_quality_check(text, context_terms=None):
text = str(text)
if context_terms is None:
grounded = True
else:
terms = [
str(t).lower() for t in _as_list(context_terms)
if t is not None and str(t).strip()
]
grounded = (len(terms) == 0) or any(t in text.lower() for t in terms)
checks = {
"not_empty": len(text.strip()) > 0,
"no_banned_terms": not contains_banned_terms(text),
"length_ok": 15 <= len(text) <= 3000,
"grounded": grounded,
}
checks["passed"] = all(checks.values())
return checks
def _esc(s):
return str(s).replace("&", "&").replace("<", "<").replace(">", ">")
# --------------------------------------------------------------------------- #
# Cached loaders
# --------------------------------------------------------------------------- #
def _resolve_file(filename, repo_type="dataset"):
try:
from huggingface_hub import hf_hub_download
return hf_hub_download(repo_id=DATASET_REPO, filename=filename, repo_type=repo_type)
except Exception as e:
if os.path.exists(filename):
print(f"[load] Hub download failed for {filename} ({e}); using local copy.")
return filename
raise
@functools.lru_cache(maxsize=1)
def load_artifacts():
meta = pd.read_csv(_resolve_file(F_METADATA))
emb = np.load(_resolve_file(F_EMB))
selected_model = DEFAULT_EMBED_MODEL
try:
info = json.load(open(_resolve_file(F_MODEL), "r", encoding="utf-8"))
selected_model = info.get("selected_model", DEFAULT_EMBED_MODEL)
except Exception as e:
print(f"[load] model info JSON unavailable ({e}); using default embed model.")
if len(meta) != len(emb):
raise ValueError(f"Row mismatch: metadata={len(meta)} vs embeddings={len(emb)}")
return meta, emb, selected_model
@functools.lru_cache(maxsize=1)
def get_embed_model():
from sentence_transformers import SentenceTransformer
_, _, selected_model = load_artifacts()
print(f"[load] embedding model: {selected_model}")
return SentenceTransformer(selected_model)
@functools.lru_cache(maxsize=1)
def get_generator():
from transformers import pipeline
try:
return pipeline("text2text-generation", model=PRIMARY_GEN_MODEL, device=-1), PRIMARY_GEN_MODEL
except Exception as e:
print(f"[load] {PRIMARY_GEN_MODEL} failed ({e}); fallback {FALLBACK_GEN_MODEL}.")
return pipeline("text2text-generation", model=FALLBACK_GEN_MODEL, device=-1), FALLBACK_GEN_MODEL
def encode_query(text):
return get_embed_model().encode([text], normalize_embeddings=True, convert_to_numpy=True)[0]
# --------------------------------------------------------------------------- #
# Image helpers
# --------------------------------------------------------------------------- #
@functools.lru_cache(maxsize=512)
def _img_data_uri(path, max_w):
if not os.path.exists(path):
return None
data = None
try:
from PIL import Image
im = Image.open(path).convert("RGB")
if im.width > max_w:
new_h = max(1, int(im.height * max_w / im.width))
im = im.resize((max_w, new_h))
buf = io.BytesIO()
im.save(buf, format="JPEG", quality=82)
data = buf.getvalue()
except Exception:
try:
with open(path, "rb") as f:
data = f.read()
except Exception:
return None
return "data:image/jpeg;base64," + base64.b64encode(data).decode()
def _norm_asset_name(value):
value = str(value or "").strip().lower()
value = value.replace("-", "_").replace(" ", "_")
value = re.sub(r"[^a-z0-9_]+", "", value)
value = re.sub(r"_+", "_", value).strip("_")
return value
def _asset_root_candidates():
candidates = ["assets", "./assets", "Assets", "assets:"]
seen = []
for c in candidates:
if c not in seen:
seen.append(c)
return [c for c in seen if os.path.isdir(c)]
def _image_files_in_folder(folder_path):
files = []
if not folder_path or not os.path.isdir(folder_path):
return files
for root, _, filenames in os.walk(folder_path):
for filename in filenames:
if filename.startswith("."):
continue
ext = os.path.splitext(filename)[1].lower()
if ext in IMAGE_EXTS:
files.append(os.path.join(root, filename))
return sorted(files)
def _find_property_type_folder(property_type):
mapped = PTYPE_TO_FOLDER.get(str(property_type), str(property_type))
wanted_names = {
_norm_asset_name(mapped),
_norm_asset_name(property_type),
_norm_asset_name(str(property_type).replace("Apartment", "apartment")),
}
direct_candidates = []
for root in _asset_root_candidates():
direct_candidates.extend([
os.path.join(root, mapped),
os.path.join(root, str(property_type)),
os.path.join(root, _norm_asset_name(mapped)),
os.path.join(root, _norm_asset_name(property_type)),
os.path.join(root, str(property_type).lower().replace(" ", "_")),
os.path.join(root, str(property_type).lower().replace(" ", "-")),
])
for path in direct_candidates:
if os.path.isdir(path) and _image_files_in_folder(path):
return path
for root in _asset_root_candidates():
try:
for name in os.listdir(root):
path = os.path.join(root, name)
if not os.path.isdir(path):
continue
norm = _norm_asset_name(name)
if norm in wanted_names and _image_files_in_folder(path):
return path
except Exception:
pass
return None
def _set_sort_key(path):
name = os.path.basename(path).lower()
m = re.search(r"set_(\d+)", name)
if m:
return (0, int(m.group(1)), name)
return (1, 9999, name)
def _list_set_folders(type_folder):
if not type_folder or not os.path.isdir(type_folder):
return []
out = []
for name in os.listdir(type_folder):
path = os.path.join(type_folder, name)
if os.path.isdir(path) and name.lower().startswith("set_"):
if _image_files_in_folder(path):
out.append(path)
return sorted(out, key=_set_sort_key)
def _uris_from_paths(paths):
widths = [520, 160, 160]
out = []
for path, w in zip(paths[:3], widths):
uri = _img_data_uri(path, w)
if uri:
out.append(uri)
return out
def _pick_images_from_files(files, start_idx=0, limit=3):
if not files:
return []
start_idx = start_idx % len(files)
ordered = files[start_idx:] + files[:start_idx]
return ordered[:limit]
def listing_images(property_type, listing_id):
"""
Default fallback: deterministic images by property type and listing_id.
Used when we don't pre-assign a unique set at the search-result level.
"""
type_folder = _find_property_type_folder(property_type)
if type_folder:
set_folders = _list_set_folders(type_folder)
if set_folders:
try:
idx = int(float(listing_id)) % len(set_folders)
except Exception:
idx = 0
files = _image_files_in_folder(set_folders[idx])
uris = _uris_from_paths(files)
if uris:
return uris
files = _image_files_in_folder(type_folder)
try:
start = int(float(listing_id)) % max(len(files), 1)
except Exception:
start = 0
selected = _pick_images_from_files(files, start, limit=3)
uris = _uris_from_paths(selected)
if uris:
return uris
all_files = []
for root in _asset_root_candidates():
all_files.extend(_image_files_in_folder(root))
all_files = sorted(set(all_files))
try:
start = int(float(listing_id)) % max(len(all_files), 1)
except Exception:
start = 0
selected = _pick_images_from_files(all_files, start, limit=3)
return _uris_from_paths(selected)
@functools.lru_cache(maxsize=1)
def hero_data_uri():
for root in _asset_root_candidates():
hero_folder = os.path.join(root, "hero")
for ext in IMAGE_EXTS:
uri = _img_data_uri(os.path.join(hero_folder, f"hero_home{ext}"), 1400)
if uri:
return uri
hero_files = _image_files_in_folder(hero_folder)
if hero_files:
return _img_data_uri(hero_files[0], 1400)
for root in _asset_root_candidates():
files = _image_files_in_folder(root)
if files:
return _img_data_uri(files[0], 1400)
return None
# --------------------------------------------------------------------------- #
# Recommendation logic (filters BEFORE ranking)
# --------------------------------------------------------------------------- #
def recommend_properties(user_query, top_k=4, filters=None):
meta, emb, _ = load_artifacts()
mask = np.ones(len(meta), dtype=bool)
if filters:
if filters.get("city"):
mask &= meta["city"].isin(_as_list(filters["city"])).values
if filters.get("property_type"):
mask &= meta["property_type"].isin(_as_list(filters["property_type"])).values
if filters.get("max_price") is not None:
mask &= (meta["price_million_nis"] <= float(filters["max_price"])).values
if filters.get("min_rooms") is not None:
mask &= (meta["rooms"] >= float(filters["min_rooms"])).values
idxs = np.where(mask)[0]
if len(idxs) == 0:
return pd.DataFrame()
qv = encode_query(user_query)
sims = emb[idxs] @ qv
order = np.argsort(-sims)[:int(top_k)]
sel = idxs[order]
cols = [c for c in REC_COLUMNS[2:] if c in meta.columns]
out = meta.iloc[sel][cols].copy()
out.insert(0, "similarity_score", np.round(sims[order], 4))
out.insert(0, "rank", range(1, len(out) + 1))
return out.reset_index(drop=True)
# --------------------------------------------------------------------------- #
# Context + generation
# --------------------------------------------------------------------------- #
def _amenities_list(row):
return [
("Elevator", to_bool(row.get("has_elevator"))),
("Parking", to_bool(row.get("has_parking"))),
("Balcony", to_bool(row.get("has_balcony"))),
("Renovated", to_bool(row.get("renovated"))),
("Near schools", to_bool(row.get("near_schools"))),
]
def _amenities_text(row):
return ", ".join([
"elevator" if to_bool(row.get("has_elevator")) else "no elevator",
"parking" if to_bool(row.get("has_parking")) else "no parking",
"balcony" if to_bool(row.get("has_balcony")) else "no balcony",
"renovated" if to_bool(row.get("renovated")) else "not renovated",
"near schools" if to_bool(row.get("near_schools")) else "not near schools",
])
def build_listing_context(rec_df, max_items=3):
meta, _, _ = load_artifacts()
ref = meta.set_index("listing_id")
lines = []
for _, row in rec_df.head(max_items).iterrows():
lid = int(row["listing_id"])
rooms = row["rooms"]
rooms = int(rooms) if float(rooms).is_integer() else rooms
owner = row.get("owner_name")
if (owner is None or pd.isna(owner)) and lid in ref.index and "owner_name" in ref.columns:
owner = ref.loc[lid, "owner_name"]
line = (
f"Rank {int(row['rank'])} | listing_id {lid}: "
f"{row['property_type']} in {row['neighborhood']}, {row['city']}; "
f"{rooms} rooms; {int(row['size_sqm'])} sqm; "
f"{float(row['price_million_nis']):.2f} million NIS; "
f"target audience: {row['target_audience']}; "
f"amenities: {_amenities_text(row)}; "
f"{int(row['distance_to_transport'])} minutes to public transport; "
f"owner: {owner}."
)
if lid in ref.index and "owner_message_template" in ref.columns and pd.notna(ref.loc[lid, "owner_message_template"]):
line += f" Owner message template: {ref.loc[lid, 'owner_message_template']}"
lines.append(line)
return "\n".join(lines)
def generate_buyer_summary(user_query, rec_df, max_items=3):
pipe, _ = get_generator()
context = build_listing_context(rec_df, max_items=max_items)
prompt = (
"You are HomeMatch, a direct owner-to-buyer real-estate assistant.\n"
"Write a short professional buyer summary in English.\n"
"Explain why the top properties match the buyer request.\n"
"Use 2-4 short bullet points.\n"
"Use ONLY the facts provided.\n"
"Do NOT invent details.\n"
"Do NOT mention brokers, agents, agencies, or realtors.\n\n"
f"BUYER REQUEST:\n{user_query}\n\n"
f"FACTS:\n{context}\n\n"
"ANSWER:"
)
out = pipe(prompt, max_new_tokens=180, do_sample=False)[0]["generated_text"].strip()
return clean_banned_terms(out) if contains_banned_terms(out) else out
def generate_owner_inquiry(user_query, top_listing):
pipe, _ = get_generator()
owner = top_listing.get("owner_name")
context = build_listing_context(pd.DataFrame([top_listing]), max_items=1)
prompt = (
"You are helping a buyer contact a property owner directly on HomeMatch.\n"
"Write a short polite message in English from the buyer to the owner"
+ (f" named {owner}" if owner is not None and not pd.isna(owner) else "") + ".\n"
"Mention the property briefly and ask to schedule a viewing or receive more details.\n"
"Use ONLY the facts provided.\n"
"Do NOT invent details.\n"
"Do NOT mention brokers, agents, agencies, or realtors.\n\n"
f"BUYER REQUEST:\n{user_query}\n\n"
f"FACTS:\n{context}\n\n"
"MESSAGE:"
)
out = pipe(prompt, max_new_tokens=140, do_sample=False)[0]["generated_text"].strip()
return clean_banned_terms(out) if contains_banned_terms(out) else out
# --------------------------------------------------------------------------- #
# Card image assignment per search result set
# --------------------------------------------------------------------------- #
def assign_images_for_results(rec_df):
"""
Pre-assign images for the current result set.
Goal:
- avoid repeating the same set of images within the same search result list
- keep assignment stable and by property type
- if there are 4 set folders, the first 4 cards of that property type get different sets
"""
assigned = {}
for ptype, group in rec_df.groupby("property_type", sort=False):
type_folder = _find_property_type_folder(ptype)
if not type_folder:
continue
set_folders = _list_set_folders(type_folder)
if not set_folders:
continue
group_indexes = list(group.index)
try:
offset = int(float(group.iloc[0]["listing_id"])) % len(set_folders)
except Exception:
offset = 0
rotated = set_folders[offset:] + set_folders[:offset]
for j, idx in enumerate(group_indexes):
# unique set assignment while available
if j < len(rotated):
files = _image_files_in_folder(rotated[j])
uris = _uris_from_paths(files)
if uris:
assigned[idx] = uris
return assigned
# --------------------------------------------------------------------------- #
# Property cards
# --------------------------------------------------------------------------- #
def _card_html(row, imgs=None):
if imgs is None:
imgs = listing_images(row["property_type"], row["listing_id"])
if imgs:
main = f''
thumbs = "".join(f'
' for u in imgs[1:3])
thumbs_html = f'
AI-powered property recommendations with embeddings and grounded generation