spb/wp20_uqo

1"""Prepare the listing text for the text-only extractor: concatenate description, features and broker details,2and MASK every price, tax and municipal-assessment amount so that the text model cannot read value information3(the WP14 leakage lesson: DuProprio descriptions embed the asking price).4"""5from __future__ import annotations67import json8import re910MASKED_KEYS = re.compile(r"(évaluation municipale|evaluation municipale|municipal assessment|taxes?|prix|price|revenus|"11                         r"asking|hypoth|mortgage|frais de condo|condo fees|img_audited|transaction)", re.I)12MONEY = re.compile(r"(\$\s?\d[\d\s.,]*\d|\d[\d\s]{2,}(?:[.,]\d+)?\s?(?:\$|dollars|k\$|CAD))", re.I)13ASSESS_SENT = re.compile(r"(municipal assessment|évaluation municipale|evaluation municipale|asking price|prix demandé|"14                         r"taxes municipales|taxes scolaires|school taxes|municipal taxes)[^.\n]{0,80}", re.I)151617def mask_money(s: str) -> str:18    s = ASSESS_SENT.sub("[masked]", s)19    return MONEY.sub("[amount]", s)202122def build_text(row) -> tuple[str, dict]:23    """Return (masked text, stats). `row` must have description, features (JSON list), details (JSON dict)."""24    parts = []25    desc = row.get("description") or ""26    parts.append("DESCRIPTION:\n" + mask_money(desc))27    try:28        feats = json.loads(row.get("features") or "[]")29    except Exception:30        feats = []31    if isinstance(feats, list) and feats:32        feats = [mask_money(str(f)) for f in feats[:80] if not MASKED_KEYS.search(str(f).split(":")[0])]33        parts.append("FEATURES: " + "; ".join(feats))34    try:35        det = json.loads(row.get("details") or "{}")36    except Exception:37        det = {}38    kept, dropped = [], 039    if isinstance(det, dict):40        for k, v in det.items():41            if MASKED_KEYS.search(str(k)):42                dropped += 143                continue44            if k == "pieces" and isinstance(v, list):45                rooms = "; ".join(f"{p.get('nom','')} ({p.get('niveau','')}, {p.get('dimensions','')}, {p.get('revetement','')})"46                                  for p in v[:25] if isinstance(p, dict))47                kept.append(f"ROOMS: {rooms}")48            elif isinstance(v, (str, int, float)):49                if k == "remarques_proprio" and str(v).strip()[:200] == desc.strip()[:200]:50                    continue  # duplicate of the description51                kept.append(f"{k}: {mask_money(str(v))}")52    if kept:53        parts.append("BROKER FIELDS:\n" + "\n".join(kept))54    text = "\n\n".join(parts)55    stats = {"desc_chars": len(desc), "n_features": len(feats) if isinstance(feats, list) else 0,56             "n_detail_kept": len(kept), "n_detail_masked_keys": dropped,57             "n_amounts_masked": len(MONEY.findall(desc)) + len(ASSESS_SENT.findall(desc))}58    return text[:12000], stats59