Merge pull request #99 from MaineDSA/fix_pricing_parser

fix parsing of costs

Author: bmos

pyproject.toml

@@ -8,7 +8,7 @@ dependencies = [
     "beautifulsoup4~=4.14.0",
     "google-auth~=2.49.1",
     "gspread~=6.2.1",
-    "patchright~=1.57.2",
+    "patchright~=1.58.2",
     "python-dotenv~=1.2.0",
     "tqdm~=4.67.0",
 ]
@@ -165,3 +165,7 @@ exclude_also = [
     'if __name__ == "__main__":',
     "main\\(\\)",
 ]
+
+[tool.mypy]
+local_partial_types = true
+strict_bytes = true

src/automation.py

@@ -19,7 +19,6 @@
     MIN_SCROLL_UP,
     MIN_WAIT_TIME,
     PROBABILITY_SCROLL_UP,
-    ZillowParseError,
 )
 from src.scraper import PropertyListing, ZillowHomeFinder
 
@@ -68,13 +67,14 @@ async def get_browser_page(context: BrowserContext, *, require_new_page: bool =
 async def close_modal_if_present(page: Page) -> None:
     """Close modal dialog by clicking button with class containing 'CloseButton', if present."""
     try:
+        await page.wait_for_load_state()
         close_button = page.locator("button[class*='CloseButton']").first
         is_visible = await close_button.is_visible()
-        if not is_visible:
-            msg = "Popup modal blocked page loading, cannot scrape."
-            raise ZillowParseError(msg)
-        logger.debug("Popup modal detected, closing it")
-        await close_button.click()
+        if is_visible:
+            logger.debug("Popup modal detected, attempting to close it")
+            await close_button.click()
+            return
+        logger.warning("Invisible CloseButton modal found")
     except TimeoutError as e:
         logger.debug("No CloseButton modal found or could not close: %s", e)
 
@@ -193,6 +193,7 @@ async def scroll_and_load_listings(page: Page, max_entries: int = 100, max_no_ch
     logger.debug("Lazy loading complete. Total property cards loaded: %s", final_count)
 
     await scroll_to_top(page)
+    await simulate_human_behavior(page)
 
 
 async def check_and_click_next_page(page: Page) -> bool:
@@ -246,6 +247,7 @@ async def sort_by_newest(page: Page) -> None:
 
     await sort_button.click()
     await page.wait_for_load_state()
+    await simulate_human_behavior(page)
 
     newest_button = page.get_by_text("Newest")
     if not newest_button:

src/form_submission.py

@@ -7,7 +7,8 @@
 from patchright.async_api import TimeoutError as PlaywrightTimeoutError
 from tqdm import tqdm
 
-from src.constants import MAX_WAIT_TIME, MIN_WAIT_TIME, GoogleFormConstants
+from src.automation import simulate_human_behavior
+from src.constants import GoogleFormConstants
 from src.scraper import PropertyListing
 
 logger = logging.getLogger(__name__)
@@ -17,7 +18,7 @@
 async def _submit_single_listing(page: Page, url: str, listing: PropertyListing) -> None:
     """Submit a single listing to the Google Form."""
     await page.goto(url)
-    await page.wait_for_timeout(cryptogen.randint(MIN_WAIT_TIME, MAX_WAIT_TIME))
+    await simulate_human_behavior(page)
 
     await page.fill(GoogleFormConstants.ADDRESS_INPUT_XPATH, listing.address)
     await page.fill(GoogleFormConstants.PRICE_INPUT_XPATH, listing.price)
@@ -31,7 +32,7 @@ async def _submit_single_listing(page: Page, url: str, listing: PropertyListing)
         error_msg = f"Form submission confirmation not received for {listing.address}"
         raise PlaywrightTimeoutError(error_msg) from e
 
-    await page.wait_for_timeout(cryptogen.randint(MIN_WAIT_TIME, MAX_WAIT_TIME))
+    await simulate_human_behavior(page)
 
 
 async def submit_listings(page: Page, form_url: str, listings: list[PropertyListing]) -> None:

src/main.py

@@ -35,6 +35,7 @@ async def scrape_listings(context: BrowserContext, config: Config) -> list[Prope
             raise BaseException(error_msg)
 
         await close_modal_if_present(page)
+        await simulate_human_behavior(page)
 
         logger.info("Scraping all listings...")
         await sort_by_newest(page)

src/scraper.py

@@ -48,8 +48,8 @@ def _parse_address(self) -> str:
 
     def _parse_main_link(self) -> str:
         """Extract main property link from property card."""
-        link_element = self.card.find("a", class_="property-card-link", attrs={"data-test": "property-card-link"})
-        if not isinstance(link_element, Tag):
+        link_element = self.card.find("a", class_=re.compile("property.+-link"), attrs={"data-test": re.compile("property.+-link")})
+        if not isinstance(link_element, Tag) or not link_element.get("href"):
             return ""
 
         href = cast("str", link_element.get("href", "")).strip()
@@ -118,11 +118,11 @@ def _format_price_range(self, prices: list[str]) -> str | None:
 
     def _get_units_count(self) -> int:
         """Extract number of available units."""
-        badge_area = self.card.find("div", class_=re.compile(r"StyledPropertyCardBadgeArea"))
+        badge_area = self.card.find("div", attrs={"data-c11n-component": "PropertyCard.BadgeArea"})
         if not badge_area or isinstance(badge_area, NavigableString):
             return 1
 
-        badges = badge_area.find_all("span", class_=re.compile(r"StyledPropertyCardBadge"))
+        badges = badge_area.find_all("span", attrs={"data-c11n-component": "PropertyCard.Badge"})
         for badge in badges:
             badge_text = badge.get_text(strip=True).lower()
             unit_match = self._PATTERN_UNIT_COUNT.search(badge_text)
@@ -151,7 +151,23 @@ def _get_main_price_listings(self) -> list[PropertyListing]:
         if not main_price_element:
             return []
 
-        price_text = self._clean_price_text(main_price_element.get_text(strip=True))
+        # The price container can contains multiple spans like "Fees may apply".
+        # get_text() on the outer span concatenates them without spaces, corrupting
+        # the price text (e.g. "$1,608+ 2 bdsFees may apply"). Instead, grab only
+        # the first nested span which contains the actual price string.
+        inner_span = main_price_element.find("span")
+        if inner_span and not isinstance(inner_span, Tag):
+            msg = f"inner_span type is incorrect: {type(inner_span)}"
+            raise TypeError(msg)
+
+        price_span = inner_span.find("span") if inner_span else None
+        if price_span and not isinstance(price_span, Tag):
+            msg = f"price_span type is incorrect: {type(price_span)}"
+            raise TypeError(msg)
+
+        raw_text = (price_span or inner_span or main_price_element).get_text(strip=True)
+
+        price_text = self._clean_price_text(raw_text)
         if not price_text:
             return []
 
@@ -170,22 +186,32 @@ def _get_inventory_listings(self) -> list[PropertyListing]:
         if not inventory_section or isinstance(inventory_section, NavigableString):
             return []
 
-        # Extract price and bedroom data
-        price_elements = inventory_section.find_all("span", class_=re.compile(r"PriceText"))
-        bed_elements = inventory_section.find_all("span", class_=re.compile(r"BedText"))
+        price_bed_pairs: list[tuple[str, str, str]] = []  # (price, bed_info, link)
+        for anchor in inventory_section.find_all("a"):
+            box = anchor.find("div", attrs={"data-testid": "PropertyCardInventoryBox"})
+            if not box or isinstance(box, NavigableString):
+                continue
+
+            spans = box.find_all("span")
+            if not spans:
+                continue
+
+            price_text = self._clean_price_text(spans[0].get_text(strip=True))
+            if not price_text:
+                continue
+
+            bed_info = spans[1].get_text(strip=True) if len(spans) > 1 else ""
+
+            href = cast("str", anchor.get("href", "")).strip()
+            link = href if href.startswith("http") else f"https://www.zillow.com{href}"
 
-        price_bed_pairs = []
-        for i, price_elem in enumerate(price_elements):
-            price_text = self._clean_price_text(price_elem.get_text(strip=True))
-            if price_text:
-                bed_info = bed_elements[i].get_text(strip=True) if i < len(bed_elements) else ""
-                price_bed_pairs.append((price_text, bed_info))
+            price_bed_pairs.append((price_text, bed_info, link))
 
         units_count = self._get_units_count()
 
         # Handle multiple units with price range
         if units_count > 1 and len(price_bed_pairs) > 1:
-            prices = [price for price, _ in price_bed_pairs]
+            prices = [price for price, _, __ in price_bed_pairs]
             price_range = cast("str", self._format_price_range(prices))
 
             # Calculate median of the range
@@ -199,15 +225,14 @@ def _get_inventory_listings(self) -> list[PropertyListing]:
 
         # Create individual listings
         listings = []
-        for price, bed_info in price_bed_pairs:
+        for price, bed_info, link in price_bed_pairs:
             address = self.address + (f" ({bed_info})" if bed_info else "")
-            specific_link = self._create_specific_link(bed_info)
 
             # For individual listings, median is same as price
             numeric_price = self._extract_numeric_price(price)
             median_price = str(numeric_price) if numeric_price else price
 
-            listings.append(PropertyListing(address, price, median_price, specific_link))
+            listings.append(PropertyListing(address, price, median_price, link))
 
         return listings
 

tests/conftest.py

@@ -8,7 +8,7 @@
 def zillow_search_page_html() -> str:
     """Load the vendored Zillow search results page as html text."""
     html_example_folder = Path("tests/vendored")
-    return (html_example_folder / "zillow-search-boston-20251128-1.html").read_text(encoding="utf-8")
+    return (html_example_folder / "zillow-search-boston-20260402-1.html").read_text(encoding="utf-8")
 
 
 @pytest.fixture

tests/test_form_submission.py

@@ -53,7 +53,7 @@ def wait_for_selector_side_effect(*_, **__) -> AsyncMock:
     mock_page.wait_for_selector.side_effect = wait_for_selector_side_effect
     form_url = "https://example.com/form"
 
-    with caplog.at_level("INFO"), patch("src.form_submission.cryptogen.randint", return_value=250):
+    with caplog.at_level("INFO"), patch("src.automation.cryptogen.randint", return_value=250):
         await submit_listings(mock_page, form_url, sample_listings)
 
     assert mock_page.goto.call_count == 3
@@ -67,7 +67,7 @@ async def test_submit_listings_all_succeed(caplog: LogCaptureFixture, sample_lis
     mock_page = AsyncMock()
     form_url = "https://example.com/form"
 
-    with caplog.at_level("INFO"), patch("src.form_submission.cryptogen.randint", return_value=250):
+    with caplog.at_level("INFO"), patch("src.automation.cryptogen.randint", return_value=250):
         await submit_listings(mock_page, form_url, sample_listings)
 
     assert "3 successful, 0 failed" in caplog.text
@@ -82,7 +82,7 @@ async def test_submit_listings_all_fail(caplog: LogCaptureFixture, sample_listin
     mock_page.wait_for_selector.side_effect = PlaywrightTimeoutError("Timeout")
     form_url = "https://example.com/form"
 
-    with caplog.at_level("INFO"), patch("src.form_submission.cryptogen.randint", return_value=250):
+    with caplog.at_level("INFO"), patch("src.automation.cryptogen.randint", return_value=250):
         await submit_listings(mock_page, form_url, sample_listings)
 
     assert "0 successful, 3 failed" in caplog.text
@@ -107,7 +107,7 @@ async def test_submit_single_listing_flow_order() -> None:
     mock_page.wait_for_selector.side_effect = lambda *_, **__: call_order.append("wait_for_selector")
     mock_page.wait_for_timeout.side_effect = lambda _: call_order.append("wait_for_timeout")
 
-    with patch("src.form_submission.cryptogen.randint", return_value=250):
+    with patch("src.automation.cryptogen.randint", return_value=250):
         await _submit_single_listing(mock_page, form_url, listing)
 
     # Verify the sequence
@@ -123,30 +123,6 @@ async def test_submit_single_listing_flow_order() -> None:
     ]
 
 
-@pytest.mark.asyncio
-async def test_submit_listings_uses_random_waits() -> None:
-    """Test that random wait times are used between submissions."""
-    mock_page = AsyncMock()
-    listings = [
-        PropertyListing("Addr1", "$1000", "1000", "http://link1"),
-        PropertyListing("Addr2", "$2000", "2000", "http://link2"),
-    ]
-    form_url = "https://example.com/form"
-
-    wait_times = []
-    mock_page.wait_for_timeout.side_effect = wait_times.append
-
-    with patch("src.form_submission.cryptogen.randint") as mock_randint:
-        # Return different values for each call
-        mock_randint.side_effect = [100, 150, 200, 250]
-        await submit_listings(mock_page, form_url, listings)
-
-    # Should have wait_for_timeout called multiple times (2 per listing)
-    assert len(wait_times) == 4
-    # Verify the random values were used
-    assert wait_times == [100, 150, 200, 250]
-
-
 @pytest.mark.parametrize(
     "empty_list_arg",
     [
@@ -177,7 +153,7 @@ async def test_submit_single_listing_field_mapping(mock_page: AsyncMock) -> None
     listing = PropertyListing(address="742 Evergreen Terrace", price="$2,500/mo", median_price="2500", link="https://zillow.com/listing/999")
     form_url = "https://example.com/form"
 
-    with patch("src.form_submission.cryptogen.randint", return_value=250):
+    with patch("src.automation.cryptogen.randint", return_value=250):
         await _submit_single_listing(mock_page, form_url, listing)
 
     fill_calls = mock_page.fill.call_args_list