app.py

# ============================================================
# REMATCH - HUGGING FACE SPACE APP
# Keep generation.py and the assets folder beside this file.
# ============================================================

import hashlib
import html
import itertools
import json
import re
from functools import lru_cache
from pathlib import Path
from urllib.parse import quote

import gradio as gr
import joblib
import numpy as np
import pandas as pd
import sklearn
import torch

from huggingface_hub import hf_hub_download
from sentence_transformers import SentenceTransformer
from torch import nn

from generation import generate_property_explanation

# Resolve every local artifact relative to app.py. This keeps the paths stable
# whether the Space starts from the repository root or another working folder.
APP_DIR = Path(__file__).resolve().parent
ASSET_DIR = APP_DIR / "assets"
ASSET_URL_PREFIX = "assets"
TEXT_CLASSIFIER_PATH = (
    APP_DIR / "rematch_text_profile_classifiers.joblib"
)

# These images are immutable application assets, so let Gradio serve them
# directly with an inline content disposition and the correct MIME type.
gr.set_static_paths(paths=[ASSET_DIR])

PROPERTY_IMAGES = {
    "single_family": "illustrative_single_family_home.png",
    "single_family_home": "illustrative_single_family_home.png",
    "singlefamily": "illustrative_single_family_home.png",
    "townhouse": "illustrative_townhouse.png",
    "condo": "illustrative_condo.png",
    "multi_family": "illustrative_multi_family.png",
    "multifamily": "illustrative_multi_family.png",
    "luxury_home": "illustrative_luxury_home.png",
}
DEFAULT_PROPERTY_IMAGE = "illustrative_default_property.png"

REQUIRED_ASSETS = {
    "rematch_logo.png",
    "illustrative_single_family_home.png",
    "illustrative_townhouse.png",
    "illustrative_condo.png",
    "illustrative_multi_family.png",
    "illustrative_luxury_home.png",
    "illustrative_default_property.png",
    "eyal_ofer.jpg",
    "gary_barnett.jpg",
    "adam_neumann.jpg",
}

IMAGE_SIGNATURES = {
    ".png": b"\x89PNG\r\n\x1a\n",
    ".jpg": b"\xff\xd8\xff",
    ".jpeg": b"\xff\xd8\xff",
}


def validate_assets():
    """Fail early with an exact list instead of rendering broken images."""
    missing = sorted(
        filename
        for filename in REQUIRED_ASSETS
        if not (ASSET_DIR / filename).is_file()
    )

    if missing:
        raise FileNotFoundError(
            "The assets folder is missing required files: "
            f"{missing}"
        )

    invalid = []

    for filename in sorted(REQUIRED_ASSETS):
        path = ASSET_DIR / filename
        expected_signature = IMAGE_SIGNATURES[path.suffix.lower()]

        with path.open("rb") as file:
            signature = file.read(len(expected_signature))

        if signature != expected_signature:
            invalid.append(filename)

    if invalid:
        raise RuntimeError(
            "These assets are not real image binaries (they may be "
            f"Git LFS pointer files): {invalid}"
        )


def asset_url(filename):
    """Return Gradio's documented repository-relative static-file URL."""
    return "/gradio_api/file=" + quote(
        f"{ASSET_URL_PREFIX}/{filename}",
        safe="/",
    )


validate_assets()
REMATCH_LOGO_URL = asset_url("rematch_logo.png")


def property_image_url(property_type):
    """Return a Gradio-served URL for a generic illustrative property image."""
    normalized_type = (
        str(property_type or "")
        .strip()
        .lower()
        .replace(" ", "_")
        .replace("-", "_")
    )
    image_name = PROPERTY_IMAGES.get(normalized_type, DEFAULT_PROPERTY_IMAGE)
    return asset_url(image_name)

# Part 3 intentionally stays CPU-only.
DEVICE = torch.device("cpu")
print("Runtime device:", DEVICE)


# ============================================================
# LOAD APPROVED MODEL ARTIFACTS
# ============================================================

MODEL_REPO_ID = "omershahar/REmatch-DCN-v2"

MODEL_CONFIG_PATH = hf_hub_download(
    repo_id=MODEL_REPO_ID,
    filename="model_config.json",
)

VOCABULARY_PATH = hf_hub_download(
    repo_id=MODEL_REPO_ID,
    filename="preprocessor_vocabulary.json",
)

CHECKPOINT_PATH = hf_hub_download(
    repo_id=MODEL_REPO_ID,
    filename="dcn_v2_checkpoint.pt",
)

with open(MODEL_CONFIG_PATH, "r", encoding="utf-8") as file:
    model_config = json.load(file)

with open(VOCABULARY_PATH, "r", encoding="utf-8") as file:
    vocabulary_payload = json.load(file)

I = model_config["investor_cols"]
P = model_config["property_cols"]
F = model_config["feature_cols"]

assert F == I + P
assert vocabulary_payload["feature_cols"] == F


# ============================================================
# LOAD DATASET B DIRECTLY FROM HUGGING FACE DATASET REPO
# ============================================================

DATASET_REPO_ID = "omershahar/REmatch-Investment-Matching-Dataset"
DATASET_FILENAME = "rematch_properties.csv"
DATASET_REVISION = "c486db5f4a37f32be96f1f986f8d3e06f022bc17"

DATASET_B_PATH = hf_hub_download(
    repo_id=DATASET_REPO_ID,
    repo_type="dataset",
    filename=DATASET_FILENAME,
    revision=DATASET_REVISION,
)

B = pd.read_csv(DATASET_B_PATH)

required_property_columns = set(
    P
    + [
        "id",
        "formattedAddress",
        "city_rentcast",
        "state",
        "propertyType",
        "price",
        "monthly_rent",
        "gross_rental_yield_percent",
        "value_forecast_12months",
        "price_volatility_percent",
        "property_description",
    ]
)

missing_property_columns = required_property_columns - set(B.columns)

assert not missing_property_columns, (
    "Dataset B is missing required columns: "
    + str(sorted(missing_property_columns))
)

for column in P:
    B[column] = (
        B[column]
        .fillna("__MISSING__")
        .astype(str)
        .str.strip()
    )

print("Loaded Dataset B from Hugging Face")
print("Property inventory rows:", len(B))

# ============================================================
# LOAD SAVED DCN-v2 PROPERTY EMBEDDINGS FROM THE SPACE
# ============================================================

EMBEDDINGS_FILENAME = (
    "rematch_dataset_b_dcn_v2_embeddings.parquet"
)
EMBEDDINGS_SHA256 = (
    "731aa9733493b8ecc26e0348dd340840b78f879619"
    "fa91b1ad1c4933dd935983"
)
EMBEDDINGS_PATH = APP_DIR / EMBEDDINGS_FILENAME


def _sha256(path):
    digest = hashlib.sha256()

    with path.open("rb") as file:
        for chunk in iter(lambda: file.read(1024 * 1024), b""):
            digest.update(chunk)

    return digest.hexdigest()


def _has_parquet_magic_bytes(path):
    if not path.exists() or path.stat().st_size < 8:
        return False

    with path.open("rb") as file:
        header = file.read(4)
        file.seek(-4, 2)
        footer = file.read(4)

    return header == b"PAR1" and footer == b"PAR1"


if not _has_parquet_magic_bytes(EMBEDDINGS_PATH):
    raise RuntimeError(
        f"{EMBEDDINGS_FILENAME} is not the real Parquet binary. "
        "Upload the 570,688-byte file, not its 131-byte Git LFS "
        "pointer."
    )

if _sha256(EMBEDDINGS_PATH) != EMBEDDINGS_SHA256:
    raise RuntimeError(
        "The local property embedding artifact failed "
        "its SHA-256 integrity check."
    )

EMBEDDINGS = pd.read_parquet(EMBEDDINGS_PATH)

EMBEDDING_COLS = sorted(
    [
        column
        for column in EMBEDDINGS.columns
        if column.startswith("dcn_v2_property_embedding_")
    ],
    key=lambda column: int(column.rsplit("_", 1)[1]),
)

assert len(EMBEDDING_COLS) == 60, (
    f"Expected 60 DCN-v2 embedding columns, found {len(EMBEDDING_COLS)}."
)

assert "id" in EMBEDDINGS.columns, (
    "The embedding Parquet file must contain an id column."
)

# IDs are the bridge between the vector index and Dataset B.
B["id"] = B["id"].astype(str)
EMBEDDINGS["id"] = EMBEDDINGS["id"].astype(str)

EMBEDDINGS = EMBEDDINGS[
    ["id"] + EMBEDDING_COLS
].drop_duplicates("id")

missing_embedding_ids = set(B["id"]) - set(EMBEDDINGS["id"])

assert not missing_embedding_ids, (
    "Some Dataset B properties have no saved embedding. "
    f"Missing count: {len(missing_embedding_ids)}"
)

print("Loaded DCN-v2 property embedding index from Parquet")
print("Embedding rows:", len(EMBEDDINGS))
print("Embedding dimensions:", len(EMBEDDING_COLS))

# ============================================================
# LOAD DATASET A FOR REPRESENTATIVE INVESTOR DESCRIPTIONS
# ============================================================

DATASET_A_PATH = hf_hub_download(
    repo_id=DATASET_REPO_ID,
    repo_type="dataset",
    filename="rematch_investor_profiles.csv",
    revision=DATASET_REVISION,
)

A = pd.read_csv(DATASET_A_PATH)

required_investor_columns = {
    "investor_id",
    "budget_level",
    "max_budget_usd",
    "financing_willingness",
    "liquidity_importance",
    "risk_profile",
    "primary_goal",
    "investor_description",
}

missing_investor_columns = required_investor_columns - set(A.columns)

assert not missing_investor_columns, (
    "Dataset A is missing required columns: "
    + str(sorted(missing_investor_columns))
)

for column in [
    "budget_level",
    "financing_willingness",
    "liquidity_importance",
    "risk_profile",
    "primary_goal",
]:
    A[column] = A[column].astype(str).str.strip()

A["max_budget_usd"] = pd.to_numeric(
    A["max_budget_usd"],
    errors="coerce",
)

print("Loaded Dataset A from Hugging Face")
print("Investor profile rows:", len(A))


# ============================================================
# LOAD THE TRAINED TEXT-TO-PROFILE CLASSIFIERS
# ============================================================

if not TEXT_CLASSIFIER_PATH.exists():
    raise FileNotFoundError(
        "Missing rematch_text_profile_classifiers.joblib. "
        "Upload it to the root of this Hugging Face Space."
    )

text_bundle = joblib.load(TEXT_CLASSIFIER_PATH)

required_text_artifacts = {
    "classifiers",
    "embedding_model",
    "embedding_dimension",
    "sklearn_version",
}
missing_text_artifacts = (
    required_text_artifacts - set(text_bundle)
)

if missing_text_artifacts:
    raise ValueError(
        "Text classifier artifact is missing: "
        f"{sorted(missing_text_artifacts)}"
    )

trained_sklearn_version = str(
    text_bundle["sklearn_version"]
)

if sklearn.__version__ != trained_sklearn_version:
    raise RuntimeError(
        "The text classifiers were trained with scikit-learn "
        f"{trained_sklearn_version}, but the Space loaded "
        f"{sklearn.__version__}. Pin the training version in "
        "requirements.txt."
    )

TEXT_CLASSIFIERS = text_bundle["classifiers"]
missing_profile_classifiers = set(I) - set(TEXT_CLASSIFIERS)

if missing_profile_classifiers:
    raise ValueError(
        "Text classifier artifact is missing profile fields: "
        f"{sorted(missing_profile_classifiers)}"
    )

# Text encoding is deliberately CPU-only. ZeroGPU is reserved for Qwen,
# while one MPNet sentence is fast enough on CPU and avoids GPU hand-offs.
TEXT_ENCODER = SentenceTransformer(
    text_bundle["embedding_model"],
    revision=text_bundle.get("embedding_model_commit"),
    device="cpu",
)

if (
    TEXT_ENCODER.get_sentence_embedding_dimension()
    != int(text_bundle["embedding_dimension"])
):
    raise ValueError(
        "Text encoder dimension does not match the classifier artifact."
    )

print(
    "Loaded text profile classifiers:",
    text_bundle["embedding_model"],
)

# ============================================================
# APPROVED DCN-v2 ARCHITECTURE
# ============================================================

class Prep:
    def __init__(self, vocabularies):
        self.v = vocabularies

    def transform(self, dataframe):
        encoded = np.zeros(
            (len(dataframe), len(F)),
            dtype=np.int64,
        )

        for index, column in enumerate(F):
            encoded[:, index] = (
                dataframe[column]
                .astype(str)
                .map(self.v[column])
                .fillna(0)
                .astype(int)
            )

        return encoded

    @property
    def sizes(self):
        return [len(self.v[column]) for column in F]


class Emb(nn.Module):
    def __init__(self, sizes, dim=12):
        super().__init__()
        self.t = nn.ModuleList(
            [nn.Embedding(size, dim) for size in sizes]
        )

    def forward(self, inputs):
        return torch.stack(
            [
                embedding(inputs[:, index])
                for index, embedding in enumerate(self.t)
            ],
            dim=1,
        )


class Cross(nn.Module):
    def __init__(self, input_size):
        super().__init__()
        self.w = nn.Parameter(torch.randn(input_size) * 0.01)
        self.b = nn.Parameter(torch.zeros(input_size))

    def forward(self, initial_input, current_input):
        return (
            initial_input
            * (current_input * self.w).sum(1, keepdim=True)
            + self.b
            + current_input
        )


class DCNv2(nn.Module):
    def __init__(self, sizes, dim=12):
        super().__init__()

        self.e = Emb(sizes, dim)

        flattened_size = len(sizes) * dim

        self.c = nn.ModuleList(
            [
                Cross(flattened_size),
                Cross(flattened_size),
            ]
        )

        self.d = nn.Sequential(
            nn.Linear(flattened_size, 96),
            nn.ReLU(),
            nn.Dropout(0.15),
            nn.Linear(96, 48),
            nn.ReLU(),
        )

        self.o = nn.Linear(flattened_size + 48, 1)

    def forward(self, inputs):
        initial_input = self.e(inputs).flatten(1)
        crossed_input = initial_input

        for cross_layer in self.c:
            crossed_input = cross_layer(
                initial_input,
                crossed_input,
            )

        return self.o(
            torch.cat(
                [
                    crossed_input,
                    self.d(initial_input),
                ],
                dim=1,
            )
        ).squeeze(1)


# ============================================================
# LOAD TRAINED MODEL
# ============================================================

prep = Prep(vocabulary_payload["vocabularies"])

embedding_dimension = int(model_config["embedding_dim"])

model = DCNv2(
    sizes=prep.sizes,
    dim=embedding_dimension,
).to(DEVICE)

checkpoint = torch.load(
    CHECKPOINT_PATH,
    map_location=DEVICE,
    weights_only=True,
)

assert checkpoint["model_name"] == "DCN-v2"
assert checkpoint["feature_cols"] == F
assert checkpoint["investor_cols"] == I
assert checkpoint["property_cols"] == P

model.load_state_dict(checkpoint["state_dict"])
model.eval()

print("Loaded trained model: DCN-v2")


# ============================================================
# PART 3 - RECOMMENDATION ENGINE
# ============================================================

RECOMMENDATION_FIELDS = [
    "id",
    "formattedAddress",
    "city_rentcast",
    "state",
    "propertyType",
    "price",
    "monthly_rent",
    "gross_rental_yield_percent",
    "value_forecast_12months",
    "price_volatility_percent",
    "property_description",
]


def predict(model, encoded_features, batch_size=8192):
    model.eval()
    outputs = []

    with torch.inference_mode():
        for start in range(0, len(encoded_features), batch_size):
            batch = torch.tensor(
                encoded_features[start:start + batch_size],
                dtype=torch.long,
                device=DEVICE,
            )

            prediction = torch.sigmoid(model(batch))
            outputs.append(prediction.cpu().numpy())

    return np.concatenate(outputs)


def percentile_rank(series, ascending=True):
    ranks = (
        pd.to_numeric(series, errors="coerce")
        .rank(pct=True)
        .fillna(0.5)
    )

    return ranks if ascending else 1 - ranks


def cosine_similarity(vector_a, vector_b):
    """Cosine similarity for two already numeric property vectors."""
    denominator = np.linalg.norm(vector_a) * np.linalg.norm(vector_b)

    if denominator == 0:
        return 0.0

    return float(np.dot(vector_a, vector_b) / denominator)


def select_diverse_top_3(ranked_candidates, diversity_weight=0.015):
    """
    Select three strong recommendations while using the saved DCN-v2
    property embeddings to avoid returning three almost identical assets.
    DCN-v2 fit score remains the main ranking signal.
    The embedding similarity penalty is intentionally small.
    """
    # Greedy selection preserves fit while applying a small similarity penalty.
    chosen = []
    chosen_vectors = []
    remaining = ranked_candidates.head(100).copy()

    while len(chosen) < 3 and not remaining.empty:
        selection_scores = []

        for index, row in remaining.iterrows():
            candidate_vector = row[EMBEDDING_COLS].to_numpy(
                dtype=np.float32
            )

            if chosen_vectors:
                max_similarity = max(
                    cosine_similarity(
                        candidate_vector,
                        selected_vector,
                    )
                    for selected_vector in chosen_vectors
                )
            else:
                max_similarity = 0.0

            selection_score = (
                float(row["final_score"])
                - diversity_weight * max_similarity
            )

            selection_scores.append(
                (index, selection_score, max_similarity)
            )

        best_index, best_selection_score, best_similarity = max(
            selection_scores,
            key=lambda item: item[1],
        )

        selected_row = remaining.loc[best_index].copy()
        selected_row["embedding_similarity_penalty"] = best_similarity
        selected_row["selection_score"] = best_selection_score

        chosen.append(selected_row)
        chosen_vectors.append(
            selected_row[EMBEDDING_COLS].to_numpy(
                dtype=np.float32
            )
        )

        remaining = remaining.drop(index=best_index)

    return pd.DataFrame(chosen)


def recommend_top_3(
    budget_level,
    max_budget_usd,
    financing_willingness,
    liquidity_importance,
    risk_profile,
    primary_goal,
    enforce_budget=True,
):
    answers = dict(
        zip(
            I,
            [
                budget_level,
                financing_willingness,
                liquidity_importance,
                risk_profile,
                primary_goal,
            ],
        )
    )

    for column, value in answers.items():
        if value not in prep.v[column]:
            raise ValueError(f"Unsupported {column}: {value}")

    max_budget_usd = float(max_budget_usd)

    if not np.isfinite(max_budget_usd) or max_budget_usd <= 0:
        raise ValueError(
            "max_budget_usd must be greater than zero."
        )

    # Dataset B remains the source of truth for every property.
    candidates = B.copy()

    candidates["price"] = pd.to_numeric(
        candidates["price"],
        errors="coerce",
    )

    # Strict eligibility rules run before any recommendation scoring.
    candidates = candidates[
        candidates["price"].notna()
        & (candidates["price"] > 0)
    ]
    if enforce_budget:
        candidates = candidates[
        candidates["price"] <= max_budget_usd
    ]  

    if risk_profile == "conservative":
        candidates = candidates[
            (candidates["volatility_band"] != "high_volatility")
            & (candidates["forecast_band"] != "negative_forecast")
        ]

    if candidates.empty:
        return pd.DataFrame(
            columns=RECOMMENDATION_FIELDS
            + [
                "predicted_match_fit",
                "final_score",
                "selection_score",
            ]
        )

    # Join each live Dataset B property to its saved DCN-v2 embedding.
    # Only the embedding vectors come from the Parquet file.
    candidates = candidates.merge(
        EMBEDDINGS,
        on="id",
        how="inner",
        validate="one_to_one",
    )

    if candidates.empty:
        raise RuntimeError(
            "No eligible Dataset B properties could be matched "
            "to the saved embedding index."
        )

    # DCN-v2 computes personalized match quality from the five answers.
    model_features = candidates[P].copy()

    for column, value in answers.items():
        model_features[column] = value

    candidates["predicted_match_fit"] = predict(
        model,
        prep.transform(model_features[F]),
    )

    # Deterministic property-quality tie-breaker.
    tie_breaker = (
        0.35 * percentile_rank(
            candidates["gross_rental_yield_percent"]
        )
        + 0.30 * percentile_rank(
            candidates["value_forecast_12months"]
        )
        + 0.20 * percentile_rank(
            candidates["price_volatility_percent"],
            ascending=False,
        )
    )
    if enforce_budget:
        tie_breaker += (
            0.15
            * (
                1 - candidates["price"] / max_budget_usd
            ).clip(0, 1)
    )
    else:
    # No budget headroom exists in fallback mode.
    # Rescale the remaining 85% of the tie-breaker to 100%.
        tie_breaker = tie_breaker / 0.85

    candidates["final_score"] = (
        0.85 * candidates["predicted_match_fit"]
        + 0.15 * tie_breaker
    )

    ranked_candidates = candidates.sort_values(
        [
            "final_score",
            "predicted_match_fit",
            "price",
            "id",
        ],
        ascending=[
            False,
            False,
            True,
            True,
        ],
        kind="mergesort",
    )

    # The saved Parquet vectors now actively influence the final three:
    # selected results remain high-fit, but are less repetitive.
    result = select_diverse_top_3(ranked_candidates)

    assert len(result) <= 3
    
    if enforce_budget:
        assert (result["price"] <= max_budget_usd).all(), (
            "Exact budget guard failed"
        )

    if risk_profile == "conservative":
        assert not (
            result["volatility_band"] == "high_volatility"
        ).any(), "Conservative volatility guard failed"

        assert not (
            result["forecast_band"] == "negative_forecast"
        ).any(), "Conservative forecast guard failed"

    return result[
        RECOMMENDATION_FIELDS
        + [
            "predicted_match_fit",
            "final_score",
            "selection_score",
        ]
    ]

# ============================================================
# PART 5A - REGRESSION TEST
# ============================================================

regression_results = recommend_top_3(
    budget_level="medium",
    max_budget_usd=450000,
    financing_willingness="no",
    liquidity_importance="medium",
    risk_profile="conservative",
    primary_goal="income",
).reset_index(drop=True)

expected_address_starts = [
    "435 Canberra Dr",
    "3508 Dance Ave",
    "400 Dublin Dr",
]

assert len(regression_results) == 3

for actual, expected in zip(
    regression_results["formattedAddress"].astype(str).tolist(),
    expected_address_starts,
):
    assert actual.startswith(expected), (
        f"Expected {expected}, got {actual}"
    )

assert (regression_results["price"] <= 450000).all()

regression_details = regression_results.merge(
    B[
        [
            "id",
            "volatility_band",
            "forecast_band",
        ]
    ],
    on="id",
    how="left",
)

assert not (
    regression_details["volatility_band"] == "high_volatility"
).any()

assert not (
    regression_details["forecast_band"] == "negative_forecast"
).any()

print("Part 5A regression test passed.")


# ============================================================
# FREE-TEXT PROFILE INFERENCE AND FAST EXPECTED DCN SCORING
# ============================================================

PROFILE_COMBINATIONS = [
    tuple(str(value) for value in combination)
    for combination in itertools.product(
        *[
            TEXT_CLASSIFIERS[field].classes_
            for field in I
        ]
    )
]

for field_index, field in enumerate(I):
    unsupported_labels = {
        combination[field_index]
        for combination in PROFILE_COMBINATIONS
    } - set(prep.v[field])

    if unsupported_labels:
        raise ValueError(
            f"Unsupported classifier labels for {field}: "
            f"{sorted(unsupported_labels)}"
        )

B = B.reset_index(drop=True)
B["price"] = pd.to_numeric(B["price"], errors="coerce")
B["_property_row"] = np.arange(len(B), dtype=np.int32)

# Join property metadata and embeddings once. Requests then use cheap row masks
# instead of repeatedly copying and merging the complete inventory.
PROPERTY_INDEX = B.merge(
    EMBEDDINGS,
    on="id",
    how="inner",
    validate="one_to_one",
    sort=False,
)

if len(PROPERTY_INDEX) != len(B):
    raise RuntimeError(
        "The property embedding index is not aligned with Dataset B."
    )

PROPERTY_INDEX = PROPERTY_INDEX.sort_values(
    "_property_row",
    kind="mergesort",
).reset_index(drop=True)

PROPERTY_EMBEDDING_MATRIX = PROPERTY_INDEX[
    EMBEDDING_COLS
].to_numpy(dtype=np.float32)
embedding_norms = np.linalg.norm(
    PROPERTY_EMBEDDING_MATRIX,
    axis=1,
    keepdims=True,
)
PROPERTY_EMBEDDING_MATRIX = np.divide(
    PROPERTY_EMBEDDING_MATRIX,
    embedding_norms,
    out=np.zeros_like(PROPERTY_EMBEDDING_MATRIX),
    where=embedding_norms > 0,
)


def _encode_known_categories(dataframe, columns):
    """Encode validated categorical columns without constructing F-sized frames."""
    encoded_columns = []

    for column in columns:
        encoded = dataframe[column].astype(str).map(prep.v[column])

        if encoded.isna().any():
            unknown = sorted(
                dataframe.loc[encoded.isna(), column]
                .astype(str)
                .unique()
                .tolist()
            )
            raise ValueError(
                f"Unknown values in {column}: {unknown[:5]}"
            )

        encoded_columns.append(encoded.to_numpy(dtype=np.int64))

    return np.column_stack(encoded_columns)


def _precompute_profile_scores(profile_batch_size=16):
    """
    Precompute exact DCN scores for every profile/property pair.
    The resulting float32 matrix is only about 10 MB. A request can therefore
    integrate the classifier's complete probability distribution with one
    matrix multiplication instead of running DCN inference interactively.
    """
    investor_encoded = np.asarray(
        [
            [
                prep.v[field][value]
                for field, value in zip(I, combination)
            ]
            for combination in PROFILE_COMBINATIONS
        ],
        dtype=np.int64,
    )
    property_encoded = _encode_known_categories(B, P)
    score_matrix = np.empty(
        (len(PROFILE_COMBINATIONS), len(B)),
        dtype=np.float32,
    )

    for start in range(
        0,
        len(PROFILE_COMBINATIONS),
        profile_batch_size,
    ):
        stop = min(
            start + profile_batch_size,
            len(PROFILE_COMBINATIONS),
        )
        profile_block = investor_encoded[start:stop]
        block_size = len(profile_block)
        encoded_features = np.concatenate(
            [
                np.repeat(profile_block, len(B), axis=0),
                np.tile(property_encoded, (block_size, 1)),
            ],
            axis=1,
        )
        score_matrix[start:stop] = predict(
            model,
            encoded_features,
        ).reshape(block_size, len(B))

    return score_matrix


print(
    "Precomputing expected-score index for",
    len(PROFILE_COMBINATIONS),
    "investor profiles...",
)
PROFILE_SCORE_MATRIX = _precompute_profile_scores()
print(
    "Expected-score index ready:",
    PROFILE_SCORE_MATRIX.shape,
)


MONEY_PATTERN = re.compile(
    r"(?i)(?:\$\s*|usd\s*)"
    r"([0-9][0-9,]*(?:\.[0-9]+)?)\s*"
    r"(k|m|million|thousand)?\b"
)
BUDGET_PATTERN = re.compile(
    r"(?i)(?:budget|up\s+to|maximum|max|spend|afford|"
    r"purchase\s+price)[^0-9$]{0,30}\$?\s*"
    r"([0-9][0-9,]*(?:\.[0-9]+)?)\s*"
    r"(k|m|million|thousand)?\b"
)
TRAILING_MONEY_PATTERN = re.compile(
    r"(?i)\b([0-9][0-9,]*(?:\.[0-9]+)?)\s*"
    r"(k|m|million|thousand)?\s*(?:usd|dollars?)\b"
)


def _money_value(number, suffix):
    value = float(str(number).replace(",", ""))
    suffix = str(suffix or "").lower()

    if suffix in {"k", "thousand"}:
        value *= 1_000
    elif suffix in {"m", "million"}:
        value *= 1_000_000

    return value


def extract_budget(description):
    """Return a positive USD amount from explicit money/budget language."""
    for pattern in (
        MONEY_PATTERN,
        BUDGET_PATTERN,
        TRAILING_MONEY_PATTERN,
    ):
        match = pattern.search(str(description or ""))

        if match:
            value = _money_value(
                match.group(1),
                match.group(2),
            )

            if np.isfinite(value) and value > 0:
                return value

    return None


def _normalize_description(description):
    raw_text = str(description or "").strip()

    if not raw_text:
        raise ValueError(
            "Please describe yourself as an investor first."
        )

    if len(raw_text.splitlines()) > 3:
        raise ValueError(
            "Please keep the description to three lines or fewer."
        )

    return " ".join(raw_text.split())


@lru_cache(maxsize=256)
def _classify_description(normalized_description):
    """Cache immutable probability outputs for repeated examples/requests."""
    vector = TEXT_ENCODER.encode(
        [normalized_description],
        convert_to_numpy=True,
        normalize_embeddings=True,
        show_progress_bar=False,
    )
    outputs = []

    for field in I:
        # Exact purchase price is authoritative for budget level, so the
        # budget classifier is not evaluated during live inference.
        if field == "budget_level":
            continue

        classifier = TEXT_CLASSIFIERS[field]
        probabilities = classifier.predict_proba(vector)[0]
        outputs.append(
            (
                field,
                tuple(str(label) for label in classifier.classes_),
                tuple(float(value) for value in probabilities),
            )
        )

    return tuple(outputs)


def infer_profile(description, manual_budget=None):
    """Infer field distributions and produce DCN profile weights."""
    normalized_description = _normalize_description(description)
    classifier_outputs = _classify_description(
        normalized_description
    )
    distributions = {
        field: dict(zip(labels, probabilities))
        for field, labels, probabilities in classifier_outputs
    }
    profile = {
        field: max(distribution, key=distribution.get)
        for field, distribution in distributions.items()
    }

    budget = extract_budget(normalized_description)

    if budget is None and manual_budget not in (None, ""):
        budget = float(manual_budget)

    if budget is None:
        return None, profile, None

    if not np.isfinite(budget) or budget <= 0:
        raise ValueError(
            "The maximum property price must be greater than zero."
        )

    budget_level = derive_budget_level(budget)
    distributions["budget_level"] = {
        str(label): float(str(label) == budget_level)
        for label in TEXT_CLASSIFIERS[
            "budget_level"
        ].classes_
    }
    profile["budget_level"] = budget_level

    weights = np.ones(
        len(PROFILE_COMBINATIONS),
        dtype=np.float64,
    )

    for field_index, field in enumerate(I):
        distribution = distributions[field]
        weights *= np.fromiter(
            (
                distribution.get(
                    combination[field_index],
                    0.0,
                )
                for combination in PROFILE_COMBINATIONS
            ),
            dtype=np.float64,
            count=len(PROFILE_COMBINATIONS),
        )

    weight_sum = weights.sum()

    if not np.isfinite(weights).all() or weight_sum <= 0:
        raise RuntimeError(
            "The inferred profile probabilities are invalid."
        )

    return (
        float(budget),
        profile,
        (weights / weight_sum).astype(np.float32),
    )


def _select_diverse_expected_top_3(
    ranked_candidates,
    diversity_weight=0.015,
):
    """Vectorized greedy diversity selection from the 100 best fits."""
    pool = ranked_candidates.head(100).reset_index(drop=True)
    remaining = np.arange(len(pool), dtype=np.int32)
    selected_pool_rows = []
    selected_records = []

    while len(selected_records) < 3 and len(remaining):
        property_rows = pool.iloc[remaining][
            "_property_row"
        ].to_numpy(dtype=np.int64)

        if selected_pool_rows:
            selected_property_rows = pool.iloc[
                selected_pool_rows
            ]["_property_row"].to_numpy(dtype=np.int64)
            similarities = (
                PROPERTY_EMBEDDING_MATRIX[property_rows]
                @ PROPERTY_EMBEDDING_MATRIX[
                    selected_property_rows
                ].T
            )
            max_similarities = similarities.max(axis=1)
        else:
            max_similarities = np.zeros(
                len(remaining),
                dtype=np.float32,
            )

        selection_scores = (
            pool.iloc[remaining]["final_score"].to_numpy(
                dtype=np.float64
            )
            - diversity_weight * max_similarities
        )
        best_remaining_position = int(
            np.argmax(selection_scores)
        )
        best_pool_row = int(
            remaining[best_remaining_position]
        )
        selected_row = pool.iloc[best_pool_row].copy()
        selected_row["embedding_similarity_penalty"] = float(
            max_similarities[best_remaining_position]
        )
        selected_row["selection_score"] = float(
            selection_scores[best_remaining_position]
        )
        selected_records.append(selected_row)
        selected_pool_rows.append(best_pool_row)
        remaining = np.delete(
            remaining,
            best_remaining_position,
        )

    return pd.DataFrame(selected_records)


def recommend_from_distribution(
    profile_weights,
    max_budget_usd,
    profile,
    enforce_budget=True,
):
    """Rank properties using the classifier's full profile distribution."""
    budget = float(max_budget_usd)
    valid_price = (
        PROPERTY_INDEX["price"].notna()
        & (PROPERTY_INDEX["price"] > 0)
    )
    eligible = valid_price.copy()

    if enforce_budget:
        eligible &= PROPERTY_INDEX["price"] <= budget

    if profile["risk_profile"] == "conservative":
        eligible &= (
            PROPERTY_INDEX["volatility_band"]
            != "high_volatility"
        )
        eligible &= (
            PROPERTY_INDEX["forecast_band"]
            != "negative_forecast"
        )

    property_rows = np.flatnonzero(
        eligible.to_numpy(dtype=bool)
    )

    if not len(property_rows):
        return pd.DataFrame()

    candidates = PROPERTY_INDEX.iloc[property_rows].copy()
    candidates["predicted_match_fit"] = (
        profile_weights
        @ PROFILE_SCORE_MATRIX[:, property_rows]
    )
    tie_breaker = (
        0.35
        * percentile_rank(
            candidates["gross_rental_yield_percent"]
        )
        + 0.30
        * percentile_rank(
            candidates["value_forecast_12months"]
        )
        + 0.20
        * percentile_rank(
            candidates["price_volatility_percent"],
            ascending=False,
        )
    )

    if enforce_budget:
        tie_breaker += 0.15 * (
            1 - candidates["price"] / budget
        ).clip(0, 1)
    else:
        tie_breaker /= 0.85

    candidates["final_score"] = (
        0.85 * candidates["predicted_match_fit"]
        + 0.15 * tie_breaker
    )
    ranked = candidates.sort_values(
        [
            "final_score",
            "predicted_match_fit",
            "price",
            "id",
        ],
        ascending=[False, False, True, True],
        kind="mergesort",
    )
    result = _select_diverse_expected_top_3(ranked)

    if enforce_budget and not result.empty:
        assert (result["price"] <= budget).all(), (
            "Exact budget guard failed"
        )

    if profile["risk_profile"] == "conservative":
        assert not (
            result["volatility_band"] == "high_volatility"
        ).any(), "Conservative volatility guard failed"
        assert not (
            result["forecast_band"] == "negative_forecast"
        ).any(), "Conservative forecast guard failed"

    return result[
        RECOMMENDATION_FIELDS
        + [
            "predicted_match_fit",
            "final_score",
            "selection_score",
        ]
    ]


# ============================================================
# PART 4 - GROUNDED EXPLANATIONS
# ============================================================

def derive_budget_level(max_budget_usd):
    amount = float(max_budget_usd)

    if amount < 150000:
        return "low"

    if amount < 300000:
        return "lower_mid"

    if amount < 500000:
        return "medium"

    return "high"

def render_representative_investor_profile(
    max_budget_usd,
    financing_willingness,
    liquidity_importance,
    risk_profile,
    primary_goal,
):
    """Return one deterministic Dataset A description matching the user."""

    budget_level = derive_budget_level(max_budget_usd)
    budget = float(max_budget_usd)

    matches = A[
        (A["budget_level"] == budget_level)
        & (A["financing_willingness"] == financing_willingness)
        & (A["liquidity_importance"] == liquidity_importance)
        & (A["risk_profile"] == risk_profile)
        & (A["primary_goal"] == primary_goal)
        & A["max_budget_usd"].notna()
        & A["investor_description"].notna()
    ].copy()

    if matches.empty:
        return ""

    matches["budget_distance"] = (
        matches["max_budget_usd"] - budget
    ).abs()

    selected = matches.sort_values(
        ["budget_distance", "investor_id"],
        ascending=[True, True],
        kind="mergesort",
    ).iloc[0]

    description = html.escape(
        str(selected["investor_description"]).strip()
    )

    return f"""
    <div class="representative-profile-card">
        <div class="representative-profile-label">
            YOUR INVESTOR PROFILE
        </div>
        <h2>A profile that matches your description</h2>
        <p>“{description}”</p>
    </div>
    """


PART4_PROPERTY_FIELDS = [
    "id",
    "formattedAddress",
    "city_rentcast",
    "propertyType",
    "yield_band",
    "liquidity_band",
    "volatility_band",
    "forecast_band",
]

PROPERTY_RECORD_BY_ID = {
    str(row["id"]): {
        field: row[field]
        for field in PART4_PROPERTY_FIELDS
        if field in B.columns
    }
    for _, row in B.iterrows()
}


def python_value(value):
    if isinstance(value, np.generic):
        return value.item()

    if pd.isna(value):
        return None

    return value


def property_record_for_generation(property_id):
    record = PROPERTY_RECORD_BY_ID.get(str(property_id))

    if record is None:
        raise ValueError(
            f"No property found for ID {property_id}."
        )

    return {
        field: python_value(value)
        for field, value in record.items()
    }


def generate_top_3_explanations(
    investor_profile,
    property_records,
):
    generation_results = []

    for property_record in property_records:
        explanation = generate_property_explanation(
            investor_profile=investor_profile,
            property_record=property_record,
        )

        generation_results.append(explanation)

    return generation_results


def prepare_text_match(description, manual_budget=None):
    """Convert free text into a weighted profile and prepare three matches."""
    budget, profile, profile_weights = infer_profile(
        description,
        manual_budget,
    )

    if budget is None:
        return None, profile, [], {}, []

    results = recommend_from_distribution(
        profile_weights=profile_weights,
        max_budget_usd=budget,
        profile=profile,
        enforce_budget=True,
    ).reset_index(drop=True)
    budget_fallback = False

    if results.empty:
        results = recommend_from_distribution(
            profile_weights=profile_weights,
            max_budget_usd=budget,
            profile=profile,
            enforce_budget=False,
        ).reset_index(drop=True)
        budget_fallback = not results.empty

    investor_profile = {
        field: profile[field]
        for field in I
    }
    ranked_records = []

    for record in results.to_dict(orient="records"):
        clean_record = {
            key: python_value(value)
            for key, value in record.items()
        }
        clean_record["budget_fallback"] = budget_fallback
        clean_record["requested_budget_usd"] = budget
        ranked_records.append(clean_record)

    property_records = [
        property_record_for_generation(record["id"])
        for record in ranked_records
    ]

    return (
        budget,
        profile,
        ranked_records,
        investor_profile,
        property_records,
    )

def format_explanation_html(explanation):
    """
    Removes the duplicate opening heading and converts important
    explanation section titles into bold visual headings.
    """

    text = str(explanation or "").strip()

    lines = text.splitlines()

    # Remove blank lines at the beginning.
    while lines and not lines[0].strip():
        lines.pop(0)

    # Remove duplicate first line shown beneath the card heading.
    if lines:
        first_line = lines[0].strip().strip("*").lower()

        if first_line in {
            "why this property may fit:",
            "why this property may fit",
            "why this property matches you:",
            "why this property matches you",
        }:
            lines.pop(0)

    headings = {
        "key consideration:",
        "key property indicators:",
        "what to verify:",
    }

    formatted_lines = []

    for line in lines:
        clean_line = line.strip()

        if not clean_line:
            formatted_lines.append("<div class='explanation-space'></div>")
            continue

        escaped_line = html.escape(clean_line)
        normalized_line = clean_line.strip("*").lower()

        if normalized_line in headings:
            formatted_lines.append(
                f"<div class='explanation-heading'>{escaped_line}</div>"
            )
        else:
            formatted_lines.append(
                f"<div class='explanation-text'>{escaped_line}</div>"
            )

    return "".join(formatted_lines)
# ============================================================
# RESULT CARDS
# ============================================================

def render_part_4_results(
    ranked_records,
    generation_results,
):
    if not ranked_records:
        return """
        <div class="no-results">
            <h2>No eligible properties found</h2>
            <p>Try adjusting the budget or investor preferences.</p>
        </div>
        """

    if len(ranked_records) != len(generation_results):
        raise ValueError(
            "Recommendation and generation result counts do not match."
        )

    cards = []

    budget_fallback = bool(
        ranked_records[0].get("budget_fallback", False)
    )

    fallback_message = ""

    if budget_fallback:
        requested_budget = float(
            ranked_records[0]["requested_budget_usd"]
        )

        lowest_displayed_price = min(
            float(record["price"])
            for record in ranked_records
        )
        
        gap = lowest_displayed_price - requested_budget

        fallback_message = f"""
        <div class="budget-fallback">
            <strong>No eligible property was found within your
            ${requested_budget:,.0f} budget.</strong>
            <span>
               We’re showing the strongest matching alternatives
               outside your current budget. The lowest-priced option
               shown is ${lowest_displayed_price:,.0f},
               which is ${gap:,.0f} above your entered budget.
            </span>
        </div>
        """

    for rank, (property_record, explanation) in enumerate(
        zip(ranked_records, generation_results),
        start=1,
    ):
        address = html.escape(
            str(property_record.get("formattedAddress", "Address unavailable"))
        )

        property_type = html.escape(
            str(property_record.get("propertyType", "Property"))
        )

        image_url = property_image_url(
            property_record.get("propertyType", "")
        )

        try:
            price = f"${float(property_record.get('price')):,.0f}"
        except (TypeError, ValueError):
            price = "Price unavailable"

        try:
            price_value = float(property_record.get("price"))
        except (TypeError, ValueError):
            price_value = None

        try:
            monthly_rent = float(
                property_record.get("monthly_rent")
            )
        except (TypeError, ValueError):
            monthly_rent = None

        try:
            rental_yield = float(
                property_record.get("gross_rental_yield_percent")
            )
        except (TypeError, ValueError):
            rental_yield = None
            
        try:
            value_forecast = float(
                property_record.get("value_forecast_12months")
            )
        except (TypeError, ValueError):
            value_forecast = None


        annual_rent = (
            monthly_rent * 12
            if monthly_rent is not None
            else None
        )
        
        projected_value_change = (
            price_value * value_forecast / 100
            if price_value is not None
            and value_forecast is not None
            else None
        )
        
        estimated_gross_return = (
            rental_yield + value_forecast
            if rental_yield is not None
            and value_forecast is not None
            else None
        )
        
        rental_yield_text = (
            f"{rental_yield:.1f}%"
            if rental_yield is not None
            else "N/A"
            )

        annual_rent_text = (
            f"~${annual_rent:,.0f}/yr"
            if annual_rent is not None
            else "N/A"
        )

        forecast_text = (
            f"{value_forecast:+.1f}%"
            if value_forecast is not None
            else "N/A"
        )
        
        value_change_text = (
            f"~${projected_value_change:+,.0f}"
            if projected_value_change is not None
            else "N/A"
        )

        gross_return_text = (
            f"{estimated_gross_return:.1f}%"
            if estimated_gross_return is not None
            else "N/A"
        )

        explanation_text = format_explanation_html(explanation)

        cards.append(
            f"""
            <div class="property-card">
                <div class="property-image-wrap">
                    <img
                        class="property-image"
                        src="{image_url}"
                        alt="Illustrative {property_type} exterior"
                    />
                    <div class="property-rank">MATCH #{rank}</div>
                </div>
                <div class="property-content">
                    <h2 class="property-address">{address}</h2>
                    <div class="property-details">
                        <div>
                            <span class="detail-label">PROPERTY TYPE</span>
                            <span class="detail-value">{property_type}</span>
                        </div>
                        <div>
                            <span class="detail-label">PRICE</span>
                            <span class="property-price">{price}</span>
                        </div>
                    </div>
                    <div class="return-panel">
                        <div class="return-main">
                            <span class="return-label">
                               EST. 12-MONTH GROSS RETURN
                            </span>
                            <span class="return-value">
                               {gross_return_text}
                            </span>
                        </div>
                        
                        <div class="return-components">
                            <div class="return-component">
                                <span class="return-component-label">
                                   RENTAL INCOME
                                </span>
                                <strong>{rental_yield_text}</strong>
                                <small>{annual_rent_text}</small>
                        </div>
                        <div class="return-plus">+</div>
                        <div class="return-component">
                            <span class="return-component-label">
                               VALUE FORECAST
                            </span>
                            <strong>{forecast_text}</strong>
                            <small>{value_change_text}</small>
                        </div>
                    </div>
                    <div class="return-note">
                        Gross estimate before expenses, taxes,
                        financing and transaction costs.
                    </div>
                </div>
                    <div class="illustrative-note">Illustrative property image</div>
                    <div class="explanation-section">
                        <h3>Why this property matches you</h3>
                        <div>{explanation_text}</div>
                    </div>
                </div>
            </div>
            """
        )

    return f"""
    {fallback_message}
    <div class="results-header">
        <h1>Your top matches</h1>
        <p>Three properties selected for your investment profile.</p>
    </div>
    <div class="property-grid">
        {''.join(cards)}
    </div>
    """


# ============================================================
# QUICK STARTER USER EXAMPLES
# ============================================================

EXAMPLE_TEXT = {
    "Custom profile": "",
    "Eyal Ofer": (
        "I can invest up to $1,000,000 without financing. "
        "I seek long-term growth, accept aggressive risk, "
        "and do not need much liquidity."
    ),
    "Gary Barnett": (
        "My maximum purchase price is $450,000 and I will not "
        "use financing. I want a balanced, preservation-focused "
        "investment with medium liquidity."
    ),
    "Adam Neumann": (
        "My budget is $125,000 and I am willing to finance. "
        "I prefer conservative income investments and high liquidity."
    ),
}

# Upload these three authorised image files to the assets folder. The app uses
# an initials-based placeholder until the corresponding image has been added.
EXAMPLE_DETAILS = {
    "Eyal Ofer": {
        "photo": "eyal_ofer.jpg",
        "initials": "EO",
        "bio": (
            "Eyal Ofer founded Ofer Global, a private portfolio of international businesses. "
            "Its areas of activity include maritime shipping, real estate and hotels, technology, banking and energy. "
            "This example profile represents a high-budget, growth-oriented investment approach."
        ),
    },
    "Gary Barnett": {
        "photo": "gary_barnett.jpg",
        "initials": "GB",
        "bio": (
            "Gary Barnett is the founder and chairman of Extell Development. "
            "Extell develops luxury residential, commercial and hospitality properties in New York City and beyond. "
            "This example profile represents a balanced, preservation-oriented investment approach."
        ),
    },
    "Adam Neumann": {
        "photo": "adam_neumann.jpg",
        "initials": "AN",
        "bio": (
            "Adam Neumann co-founded WeWork and later founded Flow, a residential real-estate company. "
            "Flow focuses on a technology-enabled residential experience for owners, operators and residents. "
            "This example profile represents a lower-budget, income-oriented investment approach."
        ),
    },
}


def example_image_url(example):
    """Use an authorised local image when present, otherwise show initials."""
    photo_path = ASSET_DIR / EXAMPLE_DETAILS[example]["photo"]

    if photo_path.exists():
        return asset_url(EXAMPLE_DETAILS[example]["photo"])

    initials = EXAMPLE_DETAILS[example]["initials"]
    return (
        "data:image/svg+xml;utf8,"
        + quote(
            f"<svg xmlns='http://www.w3.org/2000/svg' width='160' height='160'>"
            f"<rect width='100%' height='100%' fill='%23003b95'/>"
            f"<text x='50%' y='55%' text-anchor='middle' fill='white' "
            f"font-family='Arial' font-size='52' font-weight='700'>{initials}</text>"
            f"</svg>"
        )
    )


def render_selected_example(example_name):
    """Show an investor bio only when the visitor selects an example."""
    if example_name == "Custom profile":
        return ""

    details = EXAMPLE_DETAILS[example_name]

    return f"""
    <div class="selected-example-card">
        <img
            class="selected-example-photo"
            src="{example_image_url(example_name)}"
            alt="{html.escape(example_name)}"
        />
        <div class="selected-example-copy">
            <div class="selected-example-label">INVESTOR EXAMPLE</div>
            <h2>{html.escape(example_name)}</h2>
            <p>{html.escape(details['bio'])}</p>
        </div>
    </div>
    """


INITIAL_RESULTS = """
<div class="no-results">
    <h2>Ready when you are</h2>
    <p>
        Describe yourself as an investor, then select
        <strong>Match Me Up!</strong>
    </p>
</div>
"""


def run_match(description, manual_budget=None):
    """Complete profile inference, ranking, and explanation generation."""
    try:
        (
            budget,
            profile,
            ranked_records,
            investor_profile,
            property_records,
        ) = prepare_text_match(
            description,
            manual_budget,
        )

        if budget is None:
            budget_prompt = """
            <div class="budget-question">
                <strong>One more thing:</strong>
                What is the maximum property price you can afford in USD?
            </div>
            """
            waiting_results = """
            <div class="no-results">
                <h2>Budget needed</h2>
                <p>
                    Add a dollar amount so REmatch can enforce
                    your price limit.
                </p>
            </div>
            """
            return (
                budget_prompt,
                "",
                waiting_results,
                gr.update(visible=True, value=None),
                gr.update(visible=True),
            )

        representative_profile = (
            render_representative_investor_profile(
                budget,
                profile["financing_willingness"],
                profile["liquidity_importance"],
                profile["risk_profile"],
                profile["primary_goal"],
            )
        )
        explanations = generate_top_3_explanations(
            investor_profile,
            property_records,
        )
        results = render_part_4_results(
            ranked_records,
            explanations,
        )

        return (
            "",
            representative_profile,
            results,
            gr.update(visible=False),
            gr.update(visible=False),
        )

    except Exception as error:
        safe_message = html.escape(str(error))
        error_message = f"""
        <div class="budget-fallback">
            <strong>We could not complete the match.</strong>
            <span>{safe_message}</span>
        </div>
        """
        return (
            error_message,
            "",
            "",
            gr.update(),
            gr.update(),
        )


def apply_example(example_name):
    description = EXAMPLE_TEXT[example_name]
    selected_example = (
        render_selected_example(example_name)
        if example_name != "Custom profile"
        else ""
    )

    if not description:
        return (
            "",
            "",
            "",
            "",
            INITIAL_RESULTS,
            gr.update(visible=False),
            gr.update(visible=False),
        )

    return (
        description,
        selected_example,
        *run_match(description),
    )

# ============================================================
# GRADIO UI
# ============================================================

with gr.Blocks(
    title="rematch | Property matching",
    css="""
    :root { --blue:#003b95; --blue-dark:#002b6d; --yellow:#febb02; --ink:#1a1a1a; --muted:#6b6b6b; }
    .gradio-container { background:#f5f5f5 !important; font-family:Arial,Helvetica,sans-serif !important; }
    .hero { background:linear-gradient(112deg,var(--blue-dark),var(--blue)); border-radius:0 0 20px 20px; color:white; margin:-8px -8px 0; padding:42px max(24px,calc((100vw - 1120px)/2)) 72px; }
    .brand-row { align-items:center; display:flex; gap:14px; margin:0 0 14px; }
    .brand-logo { height:92px; object-fit:contain; width:92px; }
    .brand { color:#ffffff !important; font-size:56px; font-weight:800; letter-spacing:-2px; line-height:1; margin:0; }
    .hero-subtitle { color:#ffffff !important; font-size:19px; line-height:1.4; margin:0; opacity:.96; }
    .search-shell { max-width:1120px; margin:36px auto 0; position:relative; z-index:2; background:var(--yellow); border-radius:12px; padding:5px; box-shadow:0 6px 22px rgba(0,0,0,.18); }
    .search-card { background:white; border-radius:8px; padding:18px; }
    .search-title { color:var(--ink); font-size:21px; font-weight:700; margin:0 0 4px; }
    .search-subtitle { color:var(--muted); margin:0 0 18px; }
    .input-row { align-items:flex-end; gap:10px !important; }
    .field-question { align-items:flex-start; color:#262626; display:flex; font-size:14px; font-weight:700; gap:6px; line-height:1.25; margin:0 0 3px; min-height:28px; }
    .help-icon { align-items:center; align-self:center; background:#003b95; border-radius:50%; color:white; cursor:help; display:inline-flex; flex:0 0 18px; font-size:12px; font-weight:800; height:18px; justify-content:center; width:18px; }
    .search-field { min-width:0 !important; }
    .match-button { margin-top:44px !important; }
    .quick-starter { background:#eef5ff; border:1px solid #c8ddff; border-radius:8px; margin:24px auto 0; max-width:1120px; padding:8px 16px; }
    .selected-example { max-width:1120px; margin:16px auto 0; }
    .selected-example-card { align-items:center; background:white; border:1px solid #c8ddff; border-radius:10px; display:flex; gap:18px; padding:18px; }
    .selected-example-photo { border-radius:50%; flex:0 0 110px; height:110px; object-fit:cover; width:110px; }
    .selected-example-copy h2 { color:#262626; font-size:22px; margin:3px 0 8px; }
    .selected-example-copy p { color:#555; line-height:1.55; margin:0; }
    .selected-example-label { color:#003b95; font-size:11px; font-weight:800; letter-spacing:.1em; }
    .representative-profile { max-width:1120px; margin:18px auto 0; }
    .representative-profile-card { background:white; border-left:5px solid var(--blue); border-radius:8px; box-shadow:0 2px 8px rgba(0,0,0,.08); padding:20px 22px; }
    .representative-profile-label { color:var(--blue); font-size:11px; font-weight:800; letter-spacing:.1em; }
    .representative-profile-card h2 { color:#262626; font-size:21px; margin:5px 0 10px; }
    .representative-profile-card p { color:#404040; font-size:16px; font-style:italic; line-height:1.6; margin:0; }
    .match-button { background:#0071c2 !important; border:1px solid #0071c2 !important; border-radius:6px !important; color:white !important; font-size:17px !important; font-weight:700 !important; min-height:50px !important; }
    .match-button:hover { background:#005fa3 !important; }
    .page-content { max-width:1120px; margin:22px auto 44px; }
    .results-header { margin:28px 0 16px; text-align:left; } .results-header h1 { color:var(--ink); font-size:27px; margin-bottom:6px; } .results-header p { color:var(--muted); }
    .property-grid { display:grid; grid-template-columns:repeat(3,minmax(0,1fr)); gap:16px; align-items:stretch; margin-top:20px; }
    .property-card { background:white; border:1px solid #d0d0d0; border-radius:8px; box-shadow:0 2px 8px rgba(0,0,0,.09); overflow:hidden; }
    .property-image-wrap { height:190px; position:relative; overflow:hidden; background:#e8eef5; } .property-image { width:100%; height:100%; display:block; object-fit:cover; }
    .property-content { display:flex; flex-direction:column; padding:18px; min-height:390px; }
    .property-rank { background:var(--blue); border-radius:4px; color:white; font-size:11px; font-weight:800; letter-spacing:.1em; padding:6px 8px; position:absolute; top:12px; left:12px; }
    .property-address { color:#262626; font-size:20px; line-height:1.25; margin:0 0 18px; }
    .property-details { display:grid; grid-template-columns:1fr 1fr; gap:12px; background:#f5f5f5; border-radius:6px; padding:12px; margin-bottom:8px; }
    .detail-label { display:block; color:var(--muted); font-size:10px; font-weight:800; letter-spacing:.08em; margin-bottom:4px; } .detail-value { color:#262626; font-weight:700; }
    .property-price { color:#008009; font-size:19px; font-weight:800; } .illustrative-note { color:#777; font-size:11px; margin:2px 0 14px; }
    .return-panel { background:#f7fbff; border:1px solid #d7e7f7; border-radius:8px; margin:8px 0 12px; padding:14px; }
    .return-main { text-align:center; margin-bottom:12px; }
    .return-label { color:#6b6b6b; display:block; font-size:10px; font-weight:800; letter-spacing:.08em; }
    .return-value { color:#008009; display:block; font-size:27px; font-weight:800; margin-top:3px; }
    .return-components { align-items:center; display:grid; grid-template-columns:1fr auto 1fr; gap:8px; text-align:center; }
    .return-component-label { color:#6b6b6b; display:block; font-size:9px; font-weight:800; letter-spacing:.07em; }
    .return-component strong { color:#262626; display:block; font-size:16px; margin-top:3px; }
    .return-component small { color:#777; display:block; font-size:11px; margin-top:2px; }
    .return-plus { color:#003b95; font-size:20px; font-weight:800; }
    .return-note { border-top:1px solid #e1eaf3; color:#777; font-size:9px; margin-top:10px; padding-top:8px; text-align:center; }
    .budget-fallback { background:#fff7ed; border:1px solid #f5c38b; border-left:5px solid #f59e0b; border-radius:8px; color:#704214; margin:24px 0 8px; padding:16px 18px; }
    .budget-fallback strong { display:block; font-size:16px; margin-bottom:5px; }
    .budget-fallback span { display:block; font-size:14px; line-height:1.5; }
    .explanation-section { border-top:1px solid #e2e2e2; padding-top:16px; margin-top:auto; } .explanation-section h3 { color:#262626; font-size:15px; margin:0 0 8px; }
    .no-results { background:#fff7ed; border:1px solid #fed7aa; border-radius:8px; color:#854d0e; padding:24px; text-align:center; }
    .explanation-heading { color:#262626; font-size:14px; font-weight:800; margin-top:14px; margin-bottom:5px; } .explanation-text { color:#454545; font-size:14px; line-height:1.65; margin-bottom:4px; } .explanation-space { height:8px; }
    .text-search textarea { min-height:110px !important; font-size:17px !important; line-height:1.45 !important; }
    .search-card .match-button { margin-top:12px !important; }
    .budget-question { max-width:1120px; margin:18px auto 0; background:white; border-left:5px solid var(--yellow); border-radius:8px; padding:14px 18px; box-shadow:0 2px 10px rgba(0,0,0,.08); }
    .budget-followup { max-width:520px; margin:12px auto 0; }
    @media (max-width:900px) { .property-grid { grid-template-columns:1fr; } .hero { padding:32px 22px 48px; } .brand-logo { height:70px; width:70px; } .brand { font-size:42px; } .selected-example-card { align-items:flex-start; flex-direction:column; } }
    """,
) as demo:
    gr.HTML(f"""
    <div class="hero">
      <div class="brand-row">
        <img class="brand-logo" src="{REMATCH_LOGO_URL}" alt="REmatch logo" />
        <h1 class="brand">Rematch</h1>
      </div>
      <p class="hero-subtitle">Find investment properties that fit the way you invest.</p>
    </div>
    """)

    with gr.Group(elem_classes="search-shell"):
        with gr.Column(elem_classes="search-card"):
            gr.HTML("""
            <h2 class="search-title">
                Tell us about yourself as an investor
            </h2>
            <p class="search-subtitle">
                Write naturally in up to three lines. Include your
                maximum property budget if you know it.
            </p>
            """)
            investor_text = gr.Textbox(
                label="Your investor description",
                lines=3,
                max_lines=3,
                placeholder=(
                    "Example: I can invest up to $350,000. I want "
                    "steady rental income, balanced risk, medium "
                    "liquidity, and I can use financing."
                ),
                elem_classes="text-search",
            )
            match_button = gr.Button(
                "Match Me Up!",
                variant="primary",
                elem_classes="match-button",
            )
            with gr.Row(elem_classes="budget-followup"):
                budget_input = gr.Number(
                    label="Maximum property price (USD)",
                    minimum=1,
                    precision=0,
                    visible=False,
                )
                budget_button = gr.Button(
                    "Continue",
                    variant="primary",
                    visible=False,
                )

    status_output = gr.HTML(value="")

    with gr.Group(elem_classes="quick-starter"):
        quick_starter = gr.Dropdown(
            choices=list(EXAMPLE_TEXT.keys()),
            value="Custom profile",
            label="Try an investor example",
            info=(
                "Choosing a named example writes its description "
                "and immediately loads recommendations."
            ),
        )

    with gr.Column(elem_classes="selected-example"):
        selected_example_output = gr.HTML(value="")

    with gr.Column(elem_classes="representative-profile"):
        representative_profile_output = gr.HTML(value="")

    with gr.Column(elem_classes="page-content"):
        full_output = gr.HTML(value=INITIAL_RESULTS)

    match_outputs = [
        status_output,
        representative_profile_output,
        full_output,
        budget_input,
        budget_button,
    ]

    match_button.click(
        fn=run_match,
        inputs=[investor_text],
        outputs=match_outputs,
    )
    investor_text.submit(
        fn=run_match,
        inputs=[investor_text],
        outputs=match_outputs,
    )
    budget_button.click(
        fn=run_match,
        inputs=[investor_text, budget_input],
        outputs=match_outputs,
    )
    quick_starter.change(
        fn=apply_example,
        inputs=[quick_starter],
        outputs=[
            investor_text,
            selected_example_output,
            *match_outputs,
        ],
    )


demo.queue().launch(
    share=False,
    debug=False,
    show_error=True,
)