Replace REST autocomplete with GraphQL Search_suggestions query

- Replace /suggest REST endpoint with GraphQL Search_suggestions query
- Use search_location field instead of individual city/county/state/postal_code fields
- Fix coordinate order to [lon, lat] (GeoJSON standard) for radius searches
- Extract mpr_id from addr: prefix for single address lookups

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
zacharyhampton
2025-12-04 21:08:01 -08:00
parent 0b283e18bd
commit a8c9d0fd66

View File

@@ -36,7 +36,6 @@ from .processors import (
class RealtorScraper(Scraper): class RealtorScraper(Scraper):
SEARCH_GQL_URL = "https://api.frontdoor.realtor.com/graphql" SEARCH_GQL_URL = "https://api.frontdoor.realtor.com/graphql"
ADDRESS_AUTOCOMPLETE_URL = "https://parser-external.geo.moveaws.com/suggest"
NUM_PROPERTY_WORKERS = 20 NUM_PROPERTY_WORKERS = 20
DEFAULT_PAGE_SIZE = 200 DEFAULT_PAGE_SIZE = 200
@@ -44,33 +43,70 @@ class RealtorScraper(Scraper):
super().__init__(scraper_input) super().__init__(scraper_input)
def handle_location(self): def handle_location(self):
# Get client_id from listing_type query = """query Search_suggestions($searchInput: SearchSuggestionsInput!) {
if self.listing_type is None: search_suggestions(search_input: $searchInput) {
client_id = "for-sale" geo_results {
elif isinstance(self.listing_type, list): type
client_id = self.listing_type[0].value.lower().replace("_", "-") if self.listing_type else "for-sale" text
else: geo {
client_id = self.listing_type.value.lower().replace("_", "-") _id
area_type
city
state_code
postal_code
county
centroid { lat lon }
slug_id
geo_id
}
}
}
}"""
params = { variables = {
"input": self.location, "searchInput": {
"client_id": client_id, "search_term": self.location
"limit": "1", }
"area_types": "city,state,county,postal_code,address,street,neighborhood,school,school_district,university,park",
} }
response = self.session.get( payload = {
self.ADDRESS_AUTOCOMPLETE_URL, "query": query,
params=params, "variables": variables,
) }
response = self.session.post(self.SEARCH_GQL_URL, json=payload)
response_json = response.json() response_json = response.json()
result = response_json["autocomplete"] if (
response_json is None
if not result: or "data" not in response_json
or response_json["data"] is None
or "search_suggestions" not in response_json["data"]
or response_json["data"]["search_suggestions"] is None
or "geo_results" not in response_json["data"]["search_suggestions"]
or not response_json["data"]["search_suggestions"]["geo_results"]
):
return None return None
return result[0] geo_result = response_json["data"]["search_suggestions"]["geo_results"][0]
geo = geo_result.get("geo", {})
result = {
"text": geo_result.get("text"),
"area_type": geo.get("area_type"),
"city": geo.get("city"),
"state_code": geo.get("state_code"),
"postal_code": geo.get("postal_code"),
"county": geo.get("county"),
"centroid": geo.get("centroid"),
}
if geo.get("area_type") == "address":
geo_id = geo.get("_id", "")
if geo_id.startswith("addr:"):
result["mpr_id"] = geo_id.replace("addr:", "")
return result
def get_latest_listing_id(self, property_id: str) -> str | None: def get_latest_listing_id(self, property_id: str) -> str | None:
query = """query Property($property_id: ID!) { query = """query Property($property_id: ID!) {
@@ -372,19 +408,13 @@ class RealtorScraper(Scraper):
) )
elif search_type == "area": #: general search, came from a general location elif search_type == "area": #: general search, came from a general location
query = """query Home_search( query = """query Home_search(
$city: String, $search_location: SearchLocation,
$county: [String],
$state_code: String,
$postal_code: String,
$offset: Int, $offset: Int,
) { ) {
home_search( home_search(
query: { query: {
%s %s
city: $city search_location: $search_location
county: $county
postal_code: $postal_code
state_code: $state_code
%s %s
%s %s
%s %s
@@ -520,24 +550,16 @@ class RealtorScraper(Scraper):
if not location_info.get("centroid"): if not location_info.get("centroid"):
return [] return []
coordinates = list(location_info["centroid"].values()) centroid = location_info["centroid"]
coordinates = [centroid["lon"], centroid["lat"]] # GeoJSON order: [lon, lat]
search_variables |= { search_variables |= {
"coordinates": coordinates, "coordinates": coordinates,
"radius": "{}mi".format(self.radius), "radius": "{}mi".format(self.radius),
} }
elif location_type == "postal_code": else: #: general search (city, county, postal_code, etc.)
search_variables |= { search_variables |= {
"postal_code": location_info.get("postal_code"), "search_location": {"location": location_info.get("text")},
}
else: #: general search, location
search_variables |= {
"city": location_info.get("city"),
"county": location_info.get("county"),
"state_code": location_info.get("state_code"),
"postal_code": location_info.get("postal_code"),
} }
if self.foreclosure: if self.foreclosure: