# ----------------------------------------------------------------------------- # Immo-Ka — Agrégateur de maisons à vendre (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/ubee.py : Ubee (ubee.com) — plateforme immobilière québécoise # API publique anonyme : POST api.ubee.ca/api/anonymous/Search/SearchProperties # (pageIndex=N, 24 résultats/page, JSON riche : adresse, GPS, prix, pièces, # superficies m², année, galerie Cloudinary complète). # Deux volets À VENDRE (listingType=Seller) : résidentiel + commercial. # # ENRICHISSEMENT DÉTAIL (fiche complète) : # GET api.ubee.ca/api/anonymous/Listing/{id} # retourne TOUT : description longue (FR/EN), n° Centris, ~30 listes de # caractéristiques énumérées (chauffage, sous-sol, garage, piscine, vue, # proximité…), taxes/frais (expenses), évaluation municipale, garantie # légale, délai d'occupation, stationnements, courtier inscripteur # (nom/agence/téléphone/courriel), visite virtuelle, lien Centris original. # Mise en cache BD (detail_cache) — clé = prix + photo de couverture, seules # les fiches nouvelles/modifiées sont re-lues (plafond IMMOKA_DETAIL_LIMIT). # NB : l'endpoint détail refuse les User-Agent non navigateur (403) — le # User-Agent de BaseConnector passe. Images Cloudinary : transformations # nommées strictes, t_default_size (1260 px) est le maximum public. # ----------------------------------------------------------------------------- from __future__ import annotations import os import re from ..schema import PropertyListing from ._detailutil import apply_detail from .base import BaseConnector API = "https://api.ubee.ca/api/anonymous/Search/SearchProperties" DETAIL_API = "https://api.ubee.ca/api/anonymous/Listing" SITE = "https://ubee.com" M2_TO_SQFT = 10.7639 _TYPES = { "Unifamiliale": "Maison", "Condo": "Condo", "Terrain": "Terrain", "Plex": "Immeuble à revenus", "Commercial": "Commercial", "Fermette": "Fermette", "Chalet": "Chalet", } # --- vocabulaire énuméré de l'API détail -> libellés français ---------------- _ENUM_FR = { # sous-sol / fondation / structure "Aucun": "Aucun", "Autre": "Autre", "EntreeExterieure": "Entrée extérieure", "MoinsDeSixPieds": "Moins de 6 pieds", "SixPiedsOuPlus": "6 pieds ou plus", "NonAmenage": "Non aménagé", "PartiellementAmenage": "Partiellement aménagé", "TotalementAmenage": "Totalement aménagé", "VideSanitaire": "Vide sanitaire", "BetonCoule": "Béton coulé", "BlocDeBeton": "Bloc de béton", "DalleDeBetonAuSol": "Dalle de béton au sol", # chauffage / énergie "AirSouffle": "Air soufflé", "PlinthesAConvection": "Plinthes à convection", "PlinthesElectriques": "Plinthes électriques", "Radiant": "Radiant", "Electricite": "Électricité", "GazNaturel": "Gaz naturel", "Mazout": "Mazout", "Propane": "Propane", "Bois": "Bois", "Geothermie": "Géothermie", "Solaire": "Solaire", # foyer / poêle "FoyerAuBois": "Foyer au bois", "FoyerAuGaz": "Foyer au gaz", "FoyerAuMazout": "Foyer au mazout", "PoeleAuBois": "Poêle au bois", "PoeleAuxGranules": "Poêle aux granules", # garage / abri / allée / stationnement "Attache": "Attaché", "Detache": "Détaché", "Integre": "Intégré", "Chauffe": "Chauffé", "DoubleOuPlus": "Double ou plus", "SimpleLargeur": "Simple largeur", "DoubleLargeurOuPlus": "Double largeur ou plus", "Asphalte": "Asphalte", "NonPave": "Non pavé", "PaveUni": "Pavé uni", "AvecPriseExterieure": "Avec prise extérieure", # revêtements / toiture / fenêtres "Acier": "Acier", "Aluminium": "Aluminium", "Brique": "Brique", "Pierre": "Pierre", "Stucco": "Stucco", "Vinyle": "Vinyle", "PVC": "PVC", "BardeauxDAsphalte": "Bardeaux d'asphalte", "BardeauxDeCedre": "Bardeaux de cèdre", "BitumeEtGravier": "Bitume et gravier", "MembraneElastomere": "Membrane élastomère", "Tole": "Tôle", "Crank": "À manivelle", "PorteFenetre": "Porte-fenêtre", "Sliding": "Coulissante", "Melamine": "Mélamine", "Polyester": "Polyester", # commodités "AdoucisseurDEau": "Adoucisseur d'eau", "Ascenseur": "Ascenseur", "AspirateurCentral": "Aspirateur central", "Balcon": "Balcon", "BorneDeRecharge": "Borne de recharge", "ClimatiseurCentral": "Climatiseur central", "ClimatiseurMural": "Climatiseur mural", "EchangeurDAir": "Échangeur d'air", "EspaceDeRangement": "Espace de rangement", "Interphone": "Interphone", "Meuble": "Meublé", "SemiMeuble": "Semi-meublé", "Sauna": "Sauna", "PorteDeGarageElectrique": "Porte de garage électrique", "SystemeDAlarme": "Système d'alarme", "ThermopompeCentrale": "Thermopompe centrale", "ThermopompeMurale": "Thermopompe murale", "WaterHeater": "Chauffe-eau", # salle de bains "AttachedToMasterBedroom": "Attenante à la chambre principale", "BaignoireARemous": "Baignoire à remous", "SeparateShower": "Douche séparée", # piscine "Chauffee": "Chauffée", "Creusee": "Creusée", "HorsTerre": "Hors terre", "Interieure": "Intérieure", # eau / égouts "ChampDEpuration": "Champ d'épuration", "FosseSeptique": "Fosse septique", "Municipalite": "Municipalité", "SystemeBIONEST": "Système BIONEST", "EauDuLac": "Eau du lac", "PuitsArtesien": "Puits artésien", "PuitsDeSurface": "Puits de surface", "AccesALEau": "Accès à l'eau", "BordeParLEau": "Bordé par l'eau", "Navigable": "Navigable", # vue / terrain / zonage / particularités "Panoramique": "Panoramique", "SurLEau": "Sur l'eau", "SurLaMontagne": "Sur la montagne", "SurLaVille": "Sur la ville", "Other": "Autre", "Boise": "Boisé", "Cloture": "Clôturé", "Paysage": "Paysagé", "Agricultural": "Agricole", "Commercial": "Commercial", "Holiday": "Villégiature", "Multifamily": "Multifamilial", "Recreational": "Récréatif", "Residential": "Résidentiel", "CoinDeRue": "Coin de rue", "CornerUnit": "Unité de coin", "CulDeSac": "Cul-de-sac", "ForestMixed": "Forêt mixte", "MotorizedBoatPermitted": "Bateau à moteur permis", "MotorlessBoatOnly": "Bateau sans moteur seulement", "PrivateRoad": "Chemin privé", "SansVoisinArriere": "Sans voisin à l'arrière", # proximité "Autoroute": "Autoroute", "Cegep": "Cégep", "EcolePrimaire": "École primaire", "EcoleSecondaire": "École secondaire", "Garderie": "Garderie", "Golf": "Golf", "Hopital": "Hôpital", "MetropolitanExpressNetwork": "REM", "Parc": "Parc", "PisteCyclable": "Piste cyclable", "SentierDeMotoneige": "Sentier de motoneige", "SentierDeVTT": "Sentier de VTT", "SkiAlpin": "Ski alpin", "SkiDeFond": "Ski de fond", "TransportEnCommun": "Transport en commun", "Universite": "Université", # divers "AnimauxPermis": "Animaux permis", "RezDeChaussee": "Rez-de-chaussée", "SousSol": "Sous-sol", "Flat": "Plat", "Slope": "En pente", } # champs-listes de la fiche détail -> groupe de caractéristiques (features) _FEATURE_GROUPS = [ ("heatingType", "Chauffage"), ("heatingEnergyType", "Énergie pour le chauffage"), ("basementType", "Sous-sol"), ("garageType", "Garage"), ("carportType", "Abri d'auto"), ("drivewayType", "Allée d'accès"), ("poolType", "Piscine"), ("fireplaceType", "Foyer/Poêle"), ("convenienceType", "Commodités"), ("cabinetType", "Armoires"), ("bathroomFeatureType", "Salle de bains"), ("exteriorCoveringType", "Revêtement extérieur"), ("roofingType", "Toiture"), ("foundationType", "Fondation"), ("fenestrationType", "Fenestration"), ("windowMechanismType", "Type de fenêtres"), ("sewerSystemType", "Système d'égouts"), ("waterSupplyType", "Approvisionnement en eau"), ("waterAccessType", "Accès à l'eau"), ("viewType", "Vue"), ("proximityType", "À proximité"), ("landType", "Terrain"), ("landZoningType", "Zonage"), ("particularityType", "Particularités"), ("reducedMobilityType", "Mobilité réduite"), ("animalPermissionType", "Animaux"), ("energyEfficiencyType", "Efficacité énergétique"), ("rentedApplianceType", "Équipement loué"), ("washerDryerLocationType", "Laveuse/sécheuse"), ("commercialListingUsage", "Usage commercial"), ("farmAnimalType", "Animaux de ferme"), ("farmBuildingType", "Bâtiments agricoles"), ] _EXPENSE_FR = { "Taxmun": "Taxes municipales", "Taxsco": "Taxes scolaires", "Fcop": "Frais de copropriété", "Taxeau": "Taxe d'eau", "Elec": "Électricité", "Chauffage": "Chauffage", "CommonExpenses": "Frais communs", "Assurance": "Assurances", } _FREQ_FR = {"Annual": "an", "Monthly": "mois", "Weekly": "sem.", "Daily": "jour"} _WARRANTY_FR = { "WithWarranty": "Avec garantie légale", "WithExclusions": "Avec exclusions", "WithoutWarranty": "Sans garantie légale (aux risques et périls de l'acheteur)", "NoWarranty": "Sans garantie légale (aux risques et périls de l'acheteur)", } _PARKING_FR = {"Garage": "garage", "Driveway": "allée", "Carport": "abri d'auto", "Street": "rue", "Interior": "intérieur", "Exterior": "extérieur"} _HALF_UNITS = [("oneAndHalfCount", "1½"), ("twoAndHalfCount", "2½"), ("threeAndHalfCount", "3½"), ("fourAndHalfCount", "4½"), ("fiveAndHalfCount", "5½"), ("sixAndHalfCount", "6½"), ("sevenAndHalfCount", "7½+")] _CAMEL_RE = re.compile(r"(?<=[a-z0-9])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])") def _fr(code) -> str: """Libellé français d'un code énuméré de l'API (repli : découpe CamelCase).""" code = str(code or "").strip() if not code: return "" return _ENUM_FR.get(code) or _CAMEL_RE.sub(" ", code) def _money(amount) -> str: try: return f"{float(amount):,.0f} $".replace(",", " ") except (TypeError, ValueError): return "" class UbeeConnector(BaseConnector): source_id = "ubee" request_delay = 0.5 # fiches détail (re)lues au plus par synchronisation — le cache BD rend les # cycles suivants quasi gratuits. IMMOKA_DETAIL_LIMIT pour un rattrapage. detail_limit = 300 def _search(self, body: dict) -> list[dict]: out, page = [], 0 while True: r = self.post(f"{API}?pageIndex={page}", json=body).json() results = r.get("results") or [] out.extend(results) if len(out) >= (r.get("totalCount") or 0) or not results: break page += 1 if page > 200: # garde-fou break return out def _to_listing(self, it: dict) -> PropertyListing | None: lid = str(it.get("id") or "") slug = it.get("slugFr") or it.get("slugEn") or "" if not lid or not slug: return None city = (it.get("city") or "").strip() url = f"{SITE}/a-vendre/{it.get('citySlug') or ''}/{slug}" images = [im["publicUrls"]["default_size"] for im in it.get("images") or [] if (im.get("publicUrls") or {}).get("default_size")] living = it.get("livingSurfaceInMeters") land = it.get("landSurfaceInMeters") ptype = _TYPES.get(it.get("inscriptionType") or "", it.get("inscriptionType") or "") return PropertyListing( source=self.source_id, external_id=lid, url=url, title=f"{ptype} à vendre — {city}" if city else f"{ptype} à vendre", address=it.get("address") or "", city=city, property_type=ptype, price=it.get("askPrice"), price_label=(f"{it['askPrice']:,.0f} $".replace(",", " ") if it.get("askPrice") else ""), bedrooms=it.get("nbBedrooms"), bathrooms=it.get("nbBathrooms"), powder_rooms=it.get("nbHalfBaths"), area_sqft=round(living * M2_TO_SQFT) if living else None, lot_sqft=round(land * M2_TO_SQFT) if land else None, year_built=it.get("yearBuilt"), details={k: it.get(k) for k in ("propertyType", "buildingType", "toBuild", "taxable", "openHouseDetail", "isOnlineSince") if it.get(k)}, features=[f for f in ( f"Type de bâtiment : {it['buildingType']}" if it.get("buildingType") else "", f"Sous-type : {it['propertyType']}" if it.get("propertyType") else "", "Neuf / à construire" if it.get("toBuild") else "", "Prix taxable (+tx)" if it.get("taxable") else "", ) if f], images=images, lat=it.get("latitude"), lng=it.get("longitude"), broker_name="Ubee", agency="Ubee Québec", ) def fetch(self) -> list[PropertyListing]: out: dict[str, PropertyListing] = {} for flags in ({"isResidential": True}, {"isCommercial": True}): body = {"sortBy": "DateDescending", "listingType": "Seller", **flags} for it in self._search(body): # Québec seulement (l'API est QC par nature, on double-vérifie) if (it.get("province") or "QC") != "QC": continue lst = self._to_listing(it) if lst is not None: out.setdefault(lst.uid, lst) listings = list(out.values()) self._enrich_details(listings) return listings # -- fiche détail (description, caractéristiques, taxes, courtier…) -------- def _enrich_details(self, listings: list[PropertyListing]) -> None: budget = int(os.environ.get("IMMOKA_DETAIL_LIMIT", self.detail_limit)) if budget <= 0: return from .. import db con = db.connect() try: for lst in listings: # clé de changement : prix + photo de couverture (l'URL Cloudinary # porte la version) — les fiches modifiées sont re-lues cover = lst.images[0] if lst.images else "" key = f"v1:{lst.price or ''}:{cover}" cached = db.get_cached_detail(con, self.source_id, lst.external_id, key) if cached is None: stale = db.get_stale_detail(con, self.source_id, lst.external_id) if budget <= 0: if stale: self._apply(lst, stale) continue try: raw = self.get(f"{DETAIL_API}/{lst.external_id}").json() cached = self._detail_payload(raw) except Exception: cached = {} if not cached and stale: # échec transitoire : on garde l'ancien payload self._apply(lst, stale) budget -= 1 continue db.put_cached_detail(con, self.source_id, lst.external_id, key, cached) budget -= 1 self._apply(lst, cached) finally: con.close() def _apply(self, lst: PropertyListing, payload: dict) -> None: if not payload: return if payload.get("mls") and not lst.mls: lst.mls = payload["mls"] # agence réelle du courtier inscripteur (remplace « Ubee Québec ») if payload.get("agency"): lst.agency = payload["agency"] apply_detail(lst, payload) def _detail_payload(self, d: dict) -> dict: """Payload compact (JSON-sérialisable, mis en cache) depuis la fiche API.""" if not isinstance(d, dict) or not d.get("id"): return {} out: dict = {} details: dict = {} features: list[str] = [] # -- description longue (préférence au texte FR) ------------------------- texts = d.get("descriptionTexts") or [] desc = next((t.get("plainText") or t.get("text") or "" for t in texts if (t.get("language") or "").lower() == "fr"), "") desc = (desc or d.get("plainTextDescription") or d.get("description") or "").strip() if desc: out["description"] = desc # -- galerie (même famille Cloudinary ; on garde la plus fournie) -------- images = [im["publicUrls"]["default_size"] for im in d.get("images") or [] if (im.get("publicUrls") or {}).get("default_size")] if images: out["images"] = images # -- identifiants / champs simples --------------------------------------- if d.get("mlsNumber"): out["mls"] = str(d["mlsNumber"]) details["Numéro Centris"] = str(d["mlsNumber"]) if d.get("nbRooms"): details["Nombre de pièces"] = d["nbRooms"] if d.get("numberOfFloors"): details["Nombre d'étages"] = d["numberOfFloors"] if d.get("numberOfParkings"): details["Stationnement (total)"] = d["numberOfParkings"] parks = [f"{_PARKING_FR.get(p.get('parkingKind'), _fr(p.get('parkingKind')).lower())}" f" ({p['count']})" for p in d.get("parkings") or [] if p.get("count")] if parks: features.append("Stationnement : " + ", ".join(parks)) land = d.get("landSurfaceInMeters") if land: out["lot_sqft"] = round(land * M2_TO_SQFT) if d.get("yearBuilt"): out["year_built"] = d["yearBuilt"] # -- taxes / frais / évaluation municipale -------------------------------- for e in d.get("expenses") or []: label = _EXPENSE_FR.get(e.get("expenseType") or "") or \ _fr(e.get("expenseType")) money = _money(e.get("amount")) if not (label and money): continue per = _FREQ_FR.get(e.get("paymentFrequency") or "", "") details[label] = f"{money}/{per}" if per else money lot_a = d.get("lotMunicipalAssessment") or 0 bld_a = d.get("buildingMunicipalAssessment") or 0 if lot_a or bld_a: year = d.get("municipalAssessmentYear") details["Évaluation municipale"] = _money(lot_a + bld_a) + \ (f" ({year})" if year else "") if lot_a: details["Évaluation municipale (terrain)"] = _money(lot_a) if bld_a: details["Évaluation municipale (bâtiment)"] = _money(bld_a) # -- conditions de vente / occupation ------------------------------------- if d.get("legalWarrantyType") in _WARRANTY_FR: details["Garantie légale"] = _WARRANTY_FR[d["legalWarrantyType"]] if d.get("coOwnershipKind") in ("Divise", "Indivise"): details["Copropriété"] = d["coOwnershipKind"] if d.get("coOwnershipPercentage"): details["Quote-part"] = f"{d['coOwnershipPercentage']} %" if d.get("moveInDelay"): details["Délai d'occupation"] = str(d["moveInDelay"]).strip() if d.get("moveInDate"): details["Date d'occupation"] = str(d["moveInDate"])[:10] if d.get("topographyType"): details["Topographie"] = _fr(d["topographyType"]) if d.get("waterBodyName"): details["Plan d'eau"] = str(d["waterBodyName"]) if d.get("constructionCharacteristic") == "New": features.append("Construction neuve") if d.get("shortTermRentalPolicy"): details["Location court terme"] = ( "Non permise" if d["shortTermRentalPolicy"] == "NotAllowed" else "Permise") # -- plex / commercial ----------------------------------------------------- units = [f"{lab} × {d[k]}" for k, lab in _HALF_UNITS if d.get(k)] for k, lab in (("commercialCount", "commercial"), ("officeCount", "bureau"), ("industrialCount", "industriel")): if d.get(k): units.append(f"{lab} × {d[k]}") if units: details["Unités"] = ", ".join(units) if d.get("potentialCommercialRevenue"): details["Revenus potentiels"] = _money(d["potentialCommercialRevenue"]) # -- caractéristiques énumérées (~30 listes) ------------------------------ for field, group in _FEATURE_GROUPS: vals = [_fr(v) for v in d.get(field) or []] vals = [v for v in vals if v] if vals: features.append(f"{group} : {', '.join(vals)}") # -- liens : fiche Centris originale, visite virtuelle --------------------- links = d.get("additionalLinks") or {} if links.get("listingLink"): details["listing_origin_url"] = links["listingLink"] tour = links.get("virtualTourFrLink") or links.get("virtualTourEnLink") if tour: details["Visite virtuelle"] = tour # -- courtier inscripteur (sinon la fiche reste « Ubee » = proprio direct) -- pro = d.get("professional") or {} name = (pro.get("fullname") or "").strip() if name: out["broker_name"] = name phone = (pro.get("displayPhoneNumber") or "").strip() if phone: out["broker_phone"] = phone agence = (pro.get("agenceName") or pro.get("corporationName") or "").strip() if agence: out["agency"] = agence details["Courtier inscripteur"] = name + \ (f" ({agence})" if agence else "") + (f" — {phone}" if phone else "") if pro.get("displayEmail"): details["Courriel du courtier"] = pro["displayEmail"] cols = [c.get("fullname") for c in d.get("coListingProfessionals") or [] if c.get("fullname")] if cols: details["Courtier collaborateur"] = ", ".join(cols) if features: out["features"] = features if details: out["details"] = details return out