diff --git a/.github/workflows/ci-cd.yml b/.github/workflows/ci-cd.yml index ebbe7bc..65a0ef4 100755 --- a/.github/workflows/ci-cd.yml +++ b/.github/workflows/ci-cd.yml @@ -32,13 +32,15 @@ jobs: poetry install --with dev - name: Ruff - run: poetry run ruff check app/ integrations/ utils.py tests/test_alo_source_parser.py + run: poetry run ruff check app/ integrations/ utils.py tests/test_alo_source_parser.py tests/test_imoti_source_parser.py - name: Ruff format check - run: poetry run ruff format --check app/ integrations/ utils.py tests/test_alo_source_parser.py + run: poetry run ruff format --check app/ integrations/ utils.py tests/test_alo_source_parser.py tests/test_imoti_source_parser.py - - name: ALO source parser regression test - run: poetry run python -m unittest discover -s tests -p "test_alo_source_parser.py" + - name: Source parser regression tests + run: | + poetry run python -m unittest discover -s tests -p "test_alo_source_parser.py" + poetry run python -m unittest discover -s tests -p "test_imoti_source_parser.py" typing: runs-on: ubuntu-latest diff --git a/app/scraper/sources/__init__.py b/app/scraper/sources/__init__.py index 27e2d5c..f8de65c 100644 --- a/app/scraper/sources/__init__.py +++ b/app/scraper/sources/__init__.py @@ -1,3 +1,4 @@ from app.scraper.sources.alo import AloSourceParser +from app.scraper.sources.imoti import ImotiSourceParser -__all__ = ["AloSourceParser"] +__all__ = ["AloSourceParser", "ImotiSourceParser"] diff --git a/app/scraper/sources/imoti.py b/app/scraper/sources/imoti.py new file mode 100644 index 0000000..5fe394a --- /dev/null +++ b/app/scraper/sources/imoti.py @@ -0,0 +1,348 @@ +from __future__ import annotations + +import re +from typing import Callable, Protocol + +from bs4 import BeautifulSoup, Tag + +from app.scraper.models import ScrapedListing + +extract_names: Callable[[str], list[str]] | None +looks_like_person_name: Callable[[str], bool] | None + +try: + from utils import ( + extract_names as _extract_names, + looks_like_person_name as _looks_like_person_name, + ) + + extract_names = _extract_names + looks_like_person_name = _looks_like_person_name +except ImportError: # pragma: no cover - fallback path for isolated runtimes + extract_names = None + looks_like_person_name = None + +EMAIL_RE = re.compile(r"[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}", flags=re.I) +PROPERTY_NAME_TOKENS: tuple[str, ...] = ( + "апартамент", + "апартаменти", + "едностаен", + "двустаен", + "тристаен", + "четиристаен", + "многостаен", + "квартира", + "квартири", + "аренда", + "наем", + "оренда", + "жилье", + "житло", + "apartment", + "flat", + "rent", +) + + +class ImotiParserContext(Protocol): + @staticmethod + def _attr_str(tag: Tag | None, name: str, default: str = "") -> str: ... + + def _normalize_link(self, base_url: str, href: str | None) -> str: ... + + def _clean_text(self, value: str) -> str: ... + + def _is_listing_candidate(self, title: str) -> bool: ... + + def _pick_card_container(self, anchor: Tag) -> Tag: ... + + def _extract_ad_id(self, link: str) -> str: ... + + def _extract_price(self, text: str) -> str: ... + + def _extract_location(self, text: str) -> str: ... + + def _extract_size(self, text: str) -> str: ... + + def _extract_image(self, card: Tag | None, base_url: str) -> str: ... + + def _extract_seller_name(self, card: Tag | None) -> str: ... + + def _detect_ad_type(self, seller_name: str) -> str: ... + + def _passes_filters(self, listing: ScrapedListing) -> bool: ... + + def _extract_phone_from_text(self, text: str) -> str: ... + + +class ImotiSourceParser: + def __init__( + self, + context: ImotiParserContext, + source_site: str, + selectors: dict[str, str], + ) -> None: + self._context = context + self._source_site = source_site + self._selectors = selectors + + def parse_listing_page(self, html: str, base_url: str) -> list[ScrapedListing]: + soup = BeautifulSoup(html, "html.parser") + exact_cards = soup.select("article.product-classic") + if exact_cards: + exact_results: list[ScrapedListing] = [] + seen_ids: set[str] = set() + for article in exact_cards: + listing = self._parse_exact_card(article, base_url) + if listing is None or listing.ad_id in seen_ids: + continue + seen_ids.add(listing.ad_id) + if self._context._passes_filters(listing): + exact_results.append(listing) + return exact_results + + soup = BeautifulSoup(html, "lxml") + links = soup.select(self._selectors.get("link", "a[href*='/наеми/']")) + results: list[ScrapedListing] = [] + seen_links: set[str] = set() + + for link_el in links: + link = self._context._normalize_link( + base_url, + self._context._attr_str(link_el, "href"), + ) + if not link or link in seen_links: + continue + seen_links.add(link) + + title = self._context._clean_text(link_el.get_text(" ", strip=True)) + if not self._context._is_listing_candidate(title): + continue + + card = self._context._pick_card_container(link_el) + card_text = card.get_text("\n", strip=True) if card else title + seller_name = self._context._extract_seller_name(card) + listing = ScrapedListing( + ad_id=self._context._extract_ad_id(link), + title=title, + price=self._context._extract_price(card_text), + location=self._context._extract_location(card_text), + size=self._context._extract_size(card_text), + link=link, + image_url=self._context._extract_image(card, base_url), + seller_name=seller_name, + ad_type=self._context._detect_ad_type(seller_name), + ) + if self._context._passes_filters(listing): + results.append(listing) + + return results + + def _parse_exact_card(self, article: Tag, base_url: str) -> ScrapedListing | None: + context = self._context + title_anchor = article.select_one("h4.product-classic-title a") + if title_anchor is None: + return None + + link = context._normalize_link(base_url, context._attr_str(title_anchor, "href")) + if not link: + return None + + ad_id = context._extract_ad_id(link) + if not ad_id: + return None + + title = context._clean_text(title_anchor.get_text(strip=True)) + if not context._is_listing_candidate(title): + return None + + price_el = article.select_one(".product-classic-price") + if price_el: + price_lines = [ + line.strip() for line in price_el.get_text("\n").splitlines() if line.strip() + ] + price = price_lines[0] if price_lines else "" + else: + price = "" + + location_el = article.select_one(".btext") + location = context._clean_text(location_el.get_text(strip=True) if location_el else "") + + size = "" + for li in article.select(".product-classic-list li"): + text = context._clean_text(li.get_text(strip=True)) + if "кв.м." in text or "м²" in text: + size = text.replace("м2", "м²") + break + + seller_name = self._extract_seller_name(article) + phone = self._extract_phone(article) + + return ScrapedListing( + ad_id=ad_id, + title=title, + price=price, + location=location, + size=size, + link=link, + image_url=context._extract_image(article, base_url), + source_site=self._source_site, + phone=phone, + seller_name=seller_name, + ad_type=context._detect_ad_type(seller_name), + ) + + def enrich_detail(self, soup: BeautifulSoup, listing: ScrapedListing) -> None: + context = self._context + + if not self._looks_like_real_seller_name(listing.seller_name): + listing.seller_name = "" + + for block in soup.select("div.block-person-link"): + icon = block.select_one("span.icon") + icon_classes = context._attr_str(icon, "class") + block_text = context._clean_text(block.get_text(" ", strip=True)) + + if "mdi-account" in icon_classes and block_text: + listing.seller_name = block_text + if not listing.contact_name or listing.contact_name == "-": + listing.contact_name = block_text + continue + + if "mdi-phone" in icon_classes and not listing.phone: + tel = block.select_one("a[href^='tel:']") + phone_source = context._attr_str(tel, "href") if tel is not None else block_text + phone = context._extract_phone_from_text(phone_source) + if phone: + listing.phone = phone + continue + + if "mdi-email" in icon_classes and ( + not listing.contact_email or listing.contact_email == "-" + ): + email_anchor = block.select_one("a[href]") + email_text = context._clean_text( + email_anchor.get_text(" ", strip=True) + if email_anchor is not None + else block_text + ) + email_match = EMAIL_RE.search(email_text) + if email_match: + listing.contact_email = email_match.group(0) + + if not listing.seller_name: + for selector in ( + "h1 a", + ".product-title a", + ".property-title", + "[class*='owner']", + "[class*='seller']", + "[class*='agency']", + ): + element = soup.select_one(selector) + if element is None: + continue + candidate = context._clean_text(element.get_text(" ", strip=True)) + if self._looks_like_real_seller_name(candidate): + listing.seller_name = candidate + break + + if not listing.contact_name or listing.contact_name == "-": + listing.contact_name = self._extract_contact_name(soup) + + if listing.seller_name: + listing.ad_type = context._detect_ad_type(listing.seller_name) + + def _extract_phone(self, article: Tag) -> str: + context = self._context + tel_link = article.select_one('a[href^="tel:"]') + if tel_link: + phone = context._extract_phone_from_text(context._attr_str(tel_link, "href")) + if phone: + return phone + + for selector in ( + "[class*='phone']", + "[class*='tel']", + ".contact-phone", + ".phone-number", + "span.phone", + ): + element = article.select_one(selector) + if element is not None: + phone = context._extract_phone_from_text(element.get_text(" ", strip=True)) + if phone: + return phone + + return context._extract_phone_from_text(article.get_text(" ", strip=True)) + + def _extract_seller_name(self, article: Tag) -> str: + context = self._context + for selector in ( + ".product-classic-agency", + ".agency-name", + ".seller-name", + "[class*='agency']", + "[class*='seller']", + ".block-info h3", + ): + element = article.select_one(selector) + if element is None: + continue + name = context._clean_text(element.get_text(strip=True)) + if self._looks_like_real_seller_name(name): + return name + + text_content = article.get_text(" ", strip=True) + if extract_names is not None: + names = extract_names(text_content) + if names: + return names[0] + + if looks_like_person_name is not None: + chunks = [ + context._clean_text(chunk) for chunk in text_content.split(" ") if chunk.strip() + ] + for chunk in chunks: + if looks_like_person_name(chunk): + return chunk + return "" + + def _extract_contact_name(self, soup: BeautifulSoup) -> str: + context = self._context + for block in soup.select("div.block-person-link"): + text = context._clean_text(block.get_text(" ", strip=True)) + if not text: + continue + if looks_like_person_name is not None and looks_like_person_name(text): + return text + + for selector in ( + ".contact-name", + ".agent-name", + ".block-agent-contact .name", + ".contact-person", + "[class*='contact-name']", + ): + element = soup.select_one(selector) + if element is None: + continue + candidate = context._clean_text(element.get_text(" ", strip=True)) + if not candidate: + continue + if looks_like_person_name is None or looks_like_person_name(candidate): + return candidate + + return "-" + + def _looks_like_real_seller_name(self, value: str) -> bool: + normalized = self._context._clean_text(value).lower() + if not normalized: + return False + if "*" in normalized: + return False + if any(word in normalized for word in PROPERTY_NAME_TOKENS): + return False + if any(token in normalized for token in ("кв.м", "месец", "eur", "лв", "€", "$")): + return False + return True diff --git a/app/services/async_scraper.py b/app/services/async_scraper.py index c822b2f..18d61e5 100755 --- a/app/services/async_scraper.py +++ b/app/services/async_scraper.py @@ -14,28 +14,21 @@ from app.core.logging import get_logger from app.scraper.models import ScrapedListing from app.scraper.sources.alo import AloSourceParser +from app.scraper.sources.imoti import ImotiSourceParser -extract_names: Callable[[str], list[str]] | None extract_phone_numbers: Callable[[str], list[str]] | None -looks_like_person_name: Callable[[str], bool] | None normalize_phone_number: Callable[[str], str] | None try: from utils import ( - extract_names as _extract_names, extract_phone_numbers as _extract_phone_numbers, - looks_like_person_name as _looks_like_person_name, normalize_phone_number as _normalize_phone_number, ) - extract_names = _extract_names extract_phone_numbers = _extract_phone_numbers - looks_like_person_name = _looks_like_person_name normalize_phone_number = _normalize_phone_number except ImportError: # pragma: no cover - fallback path for isolated runtimes - extract_names = None extract_phone_numbers = None - looks_like_person_name = None normalize_phone_number = None logger = get_logger("async_scraper") @@ -107,6 +100,11 @@ def __init__(self, site_config: SiteConfig, settings: Settings) -> None: ) self._proxy_index = 0 self._alo_parser = AloSourceParser(self, site_config.name) + self._imoti_parser = ImotiSourceParser( + self, + site_config.name, + site_config.selectors, + ) def _get_client_kwargs(self) -> dict[str, Any]: headers = { @@ -290,7 +288,7 @@ def _build_page_url(self, page: int) -> str: def _parse_listing_page(self, html: str, base_url: str) -> list[ScrapedListing]: parsers = { - "imoti.bg": self._parse_imoti_listing_page, + "imoti.bg": self._imoti_parser.parse_listing_page, "alo.bg": self._alo_parser.parse_listing_page, "dom.ria.com": self._parse_generic_anchor_page, "olx.ua": self._parse_generic_anchor_page, @@ -299,107 +297,6 @@ def _parse_listing_page(self, html: str, base_url: str) -> list[ScrapedListing]: parser = parsers.get(self.site_config.name, self._parse_generic_cards) return parser(html=html, base_url=base_url) - def _parse_imoti_listing_page(self, html: str, base_url: str) -> list[ScrapedListing]: - soup = BeautifulSoup(html, "html.parser") - exact_cards = soup.select("article.product-classic") - if exact_cards: - exact_results: list[ScrapedListing] = [] - seen_ids: set[str] = set() - for article in exact_cards: - listing = self._parse_imoti_card_exact(article, base_url) - if listing is None or listing.ad_id in seen_ids: - continue - seen_ids.add(listing.ad_id) - if self._passes_filters(listing): - exact_results.append(listing) - return exact_results - - soup = BeautifulSoup(html, "lxml") - links = soup.select(self.site_config.selectors.get("link", "a[href*='/наеми/']")) - results: list[ScrapedListing] = [] - seen_links: set[str] = set() - - for link_el in links: - link = self._normalize_link(base_url, self._attr_str(link_el, "href")) - if not link or link in seen_links: - continue - seen_links.add(link) - - title = self._clean_text(link_el.get_text(" ", strip=True)) - if not self._is_listing_candidate(title): - continue - - card = self._pick_card_container(link_el) - card_text = card.get_text("\n", strip=True) if card else title - listing = ScrapedListing( - ad_id=self._extract_ad_id(link), - title=title, - price=self._extract_price(card_text), - location=self._extract_location(card_text), - size=self._extract_size(card_text), - link=link, - image_url=self._extract_image(card, base_url), - seller_name=self._extract_seller_name(card), - ad_type=self._detect_ad_type(self._extract_seller_name(card)), - ) - if self._passes_filters(listing): - results.append(listing) - - return results - - def _parse_imoti_card_exact(self, article: Tag, base_url: str) -> ScrapedListing | None: - title_anchor = article.select_one("h4.product-classic-title a") - if title_anchor is None: - return None - - link = self._normalize_link(base_url, self._attr_str(title_anchor, "href")) - if not link: - return None - - ad_id = self._extract_ad_id(link) - if not ad_id: - return None - - title = self._clean_text(title_anchor.get_text(strip=True)) - if not self._is_listing_candidate(title): - return None - - price_el = article.select_one(".product-classic-price") - if price_el: - price_lines = [ - line.strip() for line in price_el.get_text("\n").splitlines() if line.strip() - ] - price = price_lines[0] if price_lines else "" - else: - price = "" - - location_el = article.select_one(".btext") - location = self._clean_text(location_el.get_text(strip=True) if location_el else "") - - size = "" - for li in article.select(".product-classic-list li"): - text = self._clean_text(li.get_text(strip=True)) - if "кв.м." in text or "м²" in text: - size = text.replace("м2", "м²") - break - - seller_name = self._extract_seller_name_from_imoti_card(article) - phone = self._extract_phone_from_imoti_card(article) - - return ScrapedListing( - ad_id=ad_id, - title=title, - price=price, - location=location, - size=size, - link=link, - image_url=self._extract_image(article, base_url), - source_site=self.site_config.name, - phone=phone, - seller_name=seller_name, - ad_type=self._detect_ad_type(seller_name), - ) - def _parse_generic_cards(self, html: str, base_url: str) -> list[ScrapedListing]: soup = BeautifulSoup(html, "lxml") cards = soup.select(self.site_config.selectors.get("card", "article, li, div")) @@ -517,7 +414,7 @@ async def _enrich_one( soup = BeautifulSoup(response.text, "lxml") if self.site_config.name == "imoti.bg": - self._enrich_imoti_detail(soup, listing) + self._imoti_parser.enrich_detail(soup, listing) elif self.site_config.name == "alo.bg": self._alo_parser.enrich_detail(soup, listing) else: @@ -550,101 +447,6 @@ def _enrich_generic_detail(self, soup: BeautifulSoup, listing: ScrapedListing) - if not listing.contact_name or listing.contact_name == "-": listing.contact_name = listing.seller_name or self._guess_contact_name(text) - def _enrich_imoti_detail(self, soup: BeautifulSoup, listing: ScrapedListing) -> None: - if not self._looks_like_real_seller_name(listing.seller_name): - listing.seller_name = "" - - for block in soup.select("div.block-person-link"): - icon = block.select_one("span.icon") - icon_classes = self._attr_str(icon, "class") - block_text = self._clean_text(block.get_text(" ", strip=True)) - - if "mdi-account" in icon_classes and block_text: - listing.seller_name = block_text - if not listing.contact_name or listing.contact_name == "-": - listing.contact_name = block_text - continue - - if "mdi-phone" in icon_classes and not listing.phone: - tel = block.select_one("a[href^='tel:']") - phone_source = self._attr_str(tel, "href") if tel is not None else block_text - phone = self._extract_phone_from_text(phone_source) - if phone: - listing.phone = phone - continue - - if "mdi-email" in icon_classes and ( - not listing.contact_email or listing.contact_email == "-" - ): - email_anchor = block.select_one("a[href]") - email_text = self._clean_text( - email_anchor.get_text(" ", strip=True) - if email_anchor is not None - else block_text - ) - email_match = EMAIL_RE.search(email_text) - if email_match: - listing.contact_email = email_match.group(0) - - if not listing.seller_name: - for sel in ( - "h1 a", - ".product-title a", - ".property-title", - "[class*='owner']", - "[class*='seller']", - "[class*='agency']", - ): - el = soup.select_one(sel) - if el is not None: - candidate = self._clean_text(el.get_text(" ", strip=True)) - if self._looks_like_real_seller_name(candidate): - listing.seller_name = candidate - break - - if not listing.contact_name or listing.contact_name == "-": - listing.contact_name = self._extract_contact_name_from_detail_soup(soup) - if listing.seller_name: - listing.ad_type = self._detect_ad_type(listing.seller_name) - - def _looks_like_real_seller_name(self, value: str) -> bool: - normalized = self._clean_text(value).lower() - if not normalized: - return False - if "*" in normalized: - return False - if any(word in normalized for word in APARTMENT_KEYWORDS): - return False - if any(token in normalized for token in ("кв.м", "месец", "eur", "лв", "€", "$")): - return False - return True - - def _extract_contact_name_from_detail_soup(self, soup: BeautifulSoup) -> str: - for block in soup.select("div.block-person-link"): - text = self._clean_text(block.get_text(" ", strip=True)) - if not text: - continue - if looks_like_person_name is not None and looks_like_person_name(text): - return text - - for sel in ( - ".contact-name", - ".agent-name", - ".block-agent-contact .name", - ".contact-person", - "[class*='contact-name']", - ): - el = soup.select_one(sel) - if el is None: - continue - candidate = self._clean_text(el.get_text(" ", strip=True)) - if not candidate: - continue - if looks_like_person_name is None or looks_like_person_name(candidate): - return candidate - - return "-" - def _extract_contact_email_from_detail_soup(self, soup: BeautifulSoup) -> str: for root in soup.select("div.block-person-link, div.block-info, div.block-agent"): mail = root.select_one("a[href^='mailto:']") @@ -846,59 +648,6 @@ def _guess_contact_name(self, text: str) -> str: return sentence[:120] return "-" - def _extract_phone_from_imoti_card(self, article: Tag) -> str: - tel_link = article.select_one('a[href^="tel:"]') - if tel_link: - phone = self._extract_phone_from_text(self._attr_str(tel_link, "href")) - if phone: - return phone - - for selector in ( - "[class*='phone']", - "[class*='tel']", - ".contact-phone", - ".phone-number", - "span.phone", - ): - el = article.select_one(selector) - if el is not None: - phone = self._extract_phone_from_text(el.get_text(" ", strip=True)) - if phone: - return phone - - full_text = article.get_text(" ", strip=True) - return self._extract_phone_from_text(full_text) - - def _extract_seller_name_from_imoti_card(self, article: Tag) -> str: - for selector in ( - ".product-classic-agency", - ".agency-name", - ".seller-name", - "[class*='agency']", - "[class*='seller']", - ".block-info h3", - ): - el = article.select_one(selector) - if el is not None: - name = self._clean_text(el.get_text(strip=True)) - if self._looks_like_real_seller_name(name): - return name - - text_content = article.get_text(" ", strip=True) - if extract_names is not None: - names = extract_names(text_content) - if names: - return names[0] - - if looks_like_person_name is not None: - chunks = [ - self._clean_text(chunk) for chunk in text_content.split(" ") if chunk.strip() - ] - for chunk in chunks: - if looks_like_person_name(chunk): - return chunk - return "" - def _title_from_url(self, link: str) -> str: path = urlparse(link).path.rstrip("/").split("/")[-1] path = re.sub(r"[-_]+", " ", path) diff --git a/tests/test_imoti_source_parser.py b/tests/test_imoti_source_parser.py new file mode 100644 index 0000000..cb2eaa2 --- /dev/null +++ b/tests/test_imoti_source_parser.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +import unittest + +from bs4 import BeautifulSoup + +from app.core.config import Settings, SiteConfig +from app.services.async_scraper import BaseScraper + + +class ImotiSourceParserTests(unittest.TestCase): + def setUp(self) -> None: + self.site = SiteConfig( + name="imoti.bg", + base_url="https://imoti.bg/наеми/page:{page}", + max_pages=1, + selectors={ + "link": "h4 a[href*='/наеми/'], a[href*='/наеми/']", + "seller": "[class*='agency'], [class*='seller']", + }, + allowed_domains=["imoti.bg"], + ) + self.settings = Settings( + sites=[self.site], + city_filter=None, + scrape_detail_pages=False, + _env_file=None, + ) + self.scraper = BaseScraper(self.site, self.settings) + + def test_exact_card_parsing_is_delegated_to_imoti_parser(self) -> None: + html = """ + +
+

+ + Двустаен апартамент под наем + +

+
+ 900 EUR + Допълнителен текст +
+
София, Лозенец
+ +
Example Agency
+ +359 888 123 456 + +
+ +
+

+ + Двустаен апартамент под наем + +

+
900 EUR
+
София, Лозенец
+
+ + """ + + listings = self.scraper._parse_listing_page( + html, + base_url="https://imoti.bg/наеми/", + ) + + self.assertEqual(len(listings), 1) + listing = listings[0] + self.assertEqual(listing.ad_id, "123456") + self.assertEqual(listing.title, "Двустаен апартамент под наем") + self.assertEqual(listing.price, "900 EUR") + self.assertEqual(listing.location, "София, Лозенец") + self.assertEqual(listing.size, "72 кв.м.") + self.assertEqual(listing.seller_name, "Example Agency") + self.assertEqual(listing.ad_type, "agency") + self.assertTrue(listing.phone) + self.assertEqual(listing.source_site, "imoti.bg") + self.assertEqual( + listing.link, + "https://imoti.bg/наеми/двустаен-апартамент-123456.htm", + ) + + def test_detail_enrichment_remains_source_specific(self) -> None: + listing = self.scraper._parse_listing_page( + """ +
+

+ + Апартамент под наем + +

+
800 EUR
+
София
+
+ """, + base_url="https://imoti.bg/наеми/", + )[0] + + soup = BeautifulSoup( + """ + + + + + + """, + "lxml", + ) + + self.scraper._imoti_parser.enrich_detail(soup, listing) + + self.assertEqual(listing.seller_name, "Example Agency") + self.assertEqual(listing.contact_name, "Example Agency") + self.assertTrue(listing.phone) + self.assertEqual(listing.contact_email, "agent@example.com") + self.assertEqual(listing.ad_type, "agency") + + +if __name__ == "__main__": + unittest.main()