spb/lou-ka Public

1# -----------------------------------------------------------------------------2# Lou-Ka — Agrégateur de logements à louer (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/headway.py : connecteur La Corporation Headway (headwayltee.com)5#   Site vitrine Wix (rendu serveur) sans liste d'unités individuelles :6#   une annonce par complexe immobilier (Place Prévert à Vanier, Place7#   Versant Nord / Place l'Heureux / Domaine Versant Nord à Ste-Foy,8#   Complexe Renaissance à Charlesbourg, Thibault et Curé-Pelletier à Lévis).9#   Le Domaine Anjou (Montréal) est exclu (hors Québec/Lévis).10# -----------------------------------------------------------------------------11from __future__ import annotations1213import html as htmllib14import re15import unicodedata1617from ..schema import Listing, normalize_unit_type18from .base import BaseConnector1920BASE = "https://www.headwayltee.com"2122# (url de page, [(nom du complexe, secteur, ville), ...])23PAGES: list[tuple[str, list[tuple[str, str, str]]]] = [24    (f"{BASE}/logements-a-louer/appartement-quebec",25     [("Place Prévert", "Vanier", "Québec")]),26    (f"{BASE}/ste-foy",27     [("Place Versant Nord", "Sainte-Foy", "Québec"),28      ("Place l'Heureux", "Sainte-Foy", "Québec"),29      ("Domaine Versant Nord", "Sainte-Foy", "Québec")]),30    (f"{BASE}/logements-a-louer/appartement-charlesbourg",31     [("Complexe Renaissance", "Charlesbourg", "Québec")]),32    (f"{BASE}/logements-a-louer/levis",33     [("Thibault", "Lévis", "Lévis"),34      ("Curé-Pelletier", "Lévis", "Lévis")]),35]363738def _slug(s: str) -> str:39    s = unicodedata.normalize("NFD", s.lower())40    s = "".join(c for c in s if unicodedata.category(c) != "Mn")41    return re.sub(r"[^a-z0-9]+", "-", s).strip("-")424344def _wix_image(url: str) -> str:45    """URL wixstatic pleine résolution (sans les transformations /v1/fill/...)."""46    return url.split("/v1/")[0]474849class HeadwayConnector(BaseConnector):50    source_id = "headway"51    request_delay = 0.65253    def fetch(self) -> list[Listing]:54        listings: list[Listing] = []55        for url, complexes in PAGES:56            try:57                html = self.get(url).text58            except Exception:59                continue60            try:61                listings.extend(self._parse_page(url, html, complexes))62            except Exception:63                continue64        return listings6566    # -- une page (1 à 3 complexes) ----------------------------------------------67    def _parse_page(self, url: str, html: str,68                    complexes: list[tuple[str, str, str]]) -> list[Listing]:69        # segments de texte visibles, avec leur position dans le HTML70        # (les segments contenant « { » sont du CSS/JS inliné par Wix : ignorés)71        segments = [(m.start(), htmllib.unescape(m.group(1)).replace("\xa0", " ").strip())72                    for m in re.finditer(r">([^<>]{2,400})<", html)]73        segments = [(p, t) for p, t in segments74                    if t and "{" not in t and "}" not in t75                    and not t.startswith(("var ", "window.", "/*", "//"))]7677        # images : balises <img alt="Nom du complexe ..."> (wixstatic)78        img_tags = [(m.start(), m.group(0)) for m in re.finditer(r"<img [^>]+>", html)]7980        # occurrences exactes des noms de complexes (titres de sections)81        name_slugs = {_slug(n): n for n, _, _ in complexes}82        name_events: list[tuple[int, str]] = []          # (pos, nom)83        for p, t in segments:84            if _slug(t) in name_slugs:85                name_events.append((p, name_slugs[_slug(t)]))8687        # blocs "Adresse :" (libellé seul) -> associés au titre précédent le plus88        # proche ; les pages à complexe unique prennent le premier bloc trouvé.89        # nom -> (adresse, types, téléphone du bureau de location)90        info_blocks: dict[str, tuple[str, list[str], str]] = {}91        for i, (p, t) in enumerate(segments):92            if not re.match(r"^Adresse\s*:?\s*$", t):93                continue94            parts: list[str] = []95            unit_types: list[str] = []96            phone = ""97            after_phone_label = False98            for _, t2 in segments[i + 1:i + 40]:99                if re.search(r"Heures d'ouverture", t2, re.I):100                    break101                if re.match(r"^Num[ée]ro de t[ée]l[ée]phone", t2, re.I):102                    after_phone_label = True103                    continue104                if after_phone_label and not phone:105                    pm = re.search(r"\b(\d{3}-\d{3}-\d{4})\b", t2)106                    if pm:107                        phone = pm.group(1)108                    continue109                if re.fullmatch(r"\d\s*(?:½|1/2)(?:\s*pi[èe]ces?)?|\d\s*pi[èe]ces", t2):110                    nt = normalize_unit_type(t2) or t2111                    if nt not in unit_types:112                        unit_types.append(nt)113                elif len(parts) < 3 and not re.search(114                        r"Composition|Logements disponibles|sous-sol|Étage", t2, re.I):115                    parts.append(t2)116            address = ", ".join(parts).strip(" ,")117            if "saint-sacrement" in address.lower():      # siège social, pas un immeuble118                continue119            owner = None120            for np, n in name_events:121                if np < p:122                    owner = n123            if owner is None and len(complexes) == 1:124                owner = complexes[0][0]125            if owner and owner not in info_blocks:126                info_blocks[owner] = (address, unit_types, phone)127128        # position de la section détaillée de chaque complexe (dernière129        # occurrence exacte du nom, sinon premier segment qui le contient)130        section_pos: dict[str, int | None] = {}131        for name, _, _ in complexes:132            name_flat = _slug(name)133            name_pos = None134            for p, n in name_events:135                if n == name:136                    name_pos = p137            if name_pos is None:138                for p, t in segments:139                    if name_flat in _slug(t):140                        name_pos = p141                        break142            section_pos[name] = name_pos143144        results = []145        for name, sector, city in complexes:146            name_flat = _slug(name)147            name_pos = section_pos[name]148            # fin de la section = début de la section détaillée suivante149            starts = sorted(p for p in section_pos.values()150                            if p is not None and name_pos is not None151                            and p > name_pos)152            section_end = starts[0] if starts else len(html)153154            # images dont l'attribut alt commence par le nom du complexe155            images: list[str] = []156            for _, tag in img_tags:157                alt_m = re.search(r'alt="([^"]*)"', tag)158                if not alt_m:159                    continue160                alt = htmllib.unescape(alt_m.group(1))161                if not _slug(alt).startswith(name_flat):162                    continue163                src_m = re.search(r'src="(https://static\.wixstatic\.com/media/[^"]+)"', tag)164                if src_m:165                    u = _wix_image(src_m.group(1))166                    if u not in images:167                        images.append(u)168169            # description = premier long paragraphe de la section du complexe170            description = ""171            if name_pos is not None:172                for p, t in segments:173                    if name_pos < p < section_end and len(t) > 120 \174                            and not t.startswith("*") and "veuillez svp" not in t:175                        description = t176                        break177178            address, unit_types, phone = info_blocks.get(name, ("", [], ""))179180            # services disponibles (liste de la section du complexe)181            start = name_pos if name_pos is not None else 0182            amenities = self._collect_list(183                segments, start, section_end,184                r"^services disponibles",185                r"^(à moins de|pourquoi|n'h[ée]sitez|si vous|acc[èe]s facile"186                r"|gr[âa]ce [àa]|press to zoom|contactez|\d/\d)")187188            # commerces/services « à moins de 5 minutes à pied » -> description189            proximity = self._collect_list(190                segments, start, section_end,191                r"^([àa] moins de 5 minutes|acc[èe]s facile)",192                r"^(press to zoom|n'h[ée]sitez|gr[âa]ce [àa]|pourquoi"193                r"|contactez|la corporation|\d/\d)")194195            if unit_types:196                comp = "Composition de l'immeuble : " + ", ".join(unit_types) + "."197                description = (description + " " + comp).strip() if description else comp198            if proximity:199                prox = "À moins de 5 minutes à pied : " + ", ".join(proximity) + "."200                description = (description + " " + prox).strip() if description else prox201202            details: dict = {}203            if phone:204                details["contact"] = {"phone": phone}205206            results.append(Listing(207                source=self.source_id,208                external_id=_slug(name),209                url=url,210                title=name,211                address=address,212                sector=sector,213                city=city,214                unit_type=unit_types[0] if len(unit_types) == 1 else "",215                availability="Sur demande (contacter l'agent de location)",216                description=description[:800],217                amenities=amenities,218                details=details,219                images=images[:20],220            ))221        return results222223    # -- liste à puces après un libellé (bornée à la section du complexe) --------224    @staticmethod225    def _collect_list(segments: list[tuple[int, str]], start: int, end: int,226                      head_re: str, break_re: str,227                      max_len: int = 110) -> list[str]:228        head_rx = re.compile(head_re, re.I)229        break_rx = re.compile(break_re, re.I)230        items: list[str] = []231        for i, (p, t) in enumerate(segments):232            if not (start <= p < end) or not head_rx.match(t.lower()):233                continue234            for p2, t2 in segments[i + 1:i + 30]:235                if p2 >= end or break_rx.match(t2.lower()):236                    break237                # phrases d'introduction / libellés : ignorés, pas des items238                if not (3 <= len(t2) <= max_len) or t2.endswith(":") \239                        or re.search(r"appelez-nous|t[ée]l[ée]phoner", t2, re.I):240                    continue241                if t2 not in items:242                    items.append(t2)243            break244        return items245