Explore/ Craigslist apartment search
Public workflow
Craigslist apartment search
By Malik Ammar Faisal · used 0 times · updated 2026-10-18
How it runs
Distilled · v2 Includes a distilled script, so your copy runs it from its first run. If a page has changed, it falls back to the AI agent for that run.
Instructions
Open $city.craigslist.org Search apartments for rent under $max_price with at least $beds bedrooms. Return the first $count: title, price, bedrooms, neighborhood, posted_date, url.
You provide “city”, “max_price”, “beds”, “count” each time you run it.
Review the distilled script
This Python runs in your browser session when you use the workflow. Review it before running.
import csvimport hashlibimport jsonimport refrom datetime import datetime, timezonefrom html.parser import HTMLParserfrom urllib.parse import parse_qs, urlencode, urlparse from playwright.sync_api import expect class _TimeParser(HTMLParser): def __init__(self): super().__init__() self.values = [] def handle_starttag(self, tag, attrs): if tag == "time": value = dict(attrs).get("datetime") if value: self.values.append(value) def run(page, context, browser): with browser.scope( "open-apartment-search", goal="Open the requested city's Craigslist site and enter apartments / housing for rent.", returns={ "type": "object", "properties": { "inputs": {"type": "object"}, "route": {"type": "string"}, "city_url": {"type": "string"}, }, "required": ["inputs", "route", "city_url"], }, ) as opening: inputs = dict(browser.input["promptInputs"]) city = str(inputs["city"]).strip() normalized = re.sub(r"[^a-z0-9]", "", city.lower()) host = {"newyorkcity": "newyork"}.get(normalized, normalized) if not host: opening.deopt(reason="The requested city cannot be resolved to a hostname.", evidence=inputs) page.goto("https://" + host + ".craigslist.org") page.wait_for_load_state("domcontentloaded") link = page.get_by_role("link", name="apts / housing", exact=True) expect(link).to_be_visible(timeout=10000) city_url = page.url parsed = urlparse(city_url) city_text = page.locator("body").inner_text() route_match = re.search(r"/area/([^/?#]+)", parsed.path) actual_host = parsed.hostname or "" route = route_match.group(1) if route_match else actual_host.split(".")[0] if route != host or not city_text.strip(): opening.deopt( reason="The redirected Craigslist city does not match the requested city.", evidence={"city": city, "url": city_url, "route": route}, ) link.click() page.get_by_placeholder("max", exact=True).first.wait_for(timeout=20000) opening.value = {"inputs": inputs, "route": route, "city_url": city_url} with browser.scope( "verify-applied-filters", goal="Apply and verify the requested exclusive price maximum and minimum bedrooms without relying on a navigation event.", returns={ "type": "object", "properties": { "inputs": {"type": "object"}, "search_url": {"type": "string"}, "search_path": {"type": "string"}, "page_text": {"type": "string"}, "total_matches": {"type": ["integer", "null"]}, }, "required": ["inputs", "search_url", "search_path", "page_text", "total_matches"], }, ) as filters: inputs = opening.value["inputs"] maximum = int(inputs["max_price"]) - 1 minimum_beds = int(inputs["beds"]) requested_count = int(inputs["count"]) if maximum < 0 or minimum_beds < 0 or requested_count < 0: filters.deopt(reason="Invalid numeric search inputs.", evidence=inputs) max_field = page.get_by_placeholder("max", exact=True).first expect(max_field).to_be_editable(timeout=10000) max_field.fill(str(maximum)) bed_field = page.locator("input[name=min_bedrooms]") expect(bed_field).to_be_editable(timeout=10000) bed_field.fill(str(inputs["beds"])) apply_button = page.get_by_role("button", name="apply", exact=True) expect(apply_button).to_be_visible(timeout=10000) apply_button.click() expect(page.locator("input[name=max_price]")).to_have_value(str(maximum), timeout=10000) expect(bed_field).to_have_value(str(minimum_beds), timeout=10000) expect(page).to_have_url( re.compile(r"(?=.*[?&]max_price=" + str(maximum) + r"(?:&|#|$))" r"(?=.*[?&]min_bedrooms=" + str(minimum_beds) + r"(?:&|#|$)).*"), timeout=10000, ) search_url = page.url max_value = page.locator("input[name=max_price]").input_value() beds_value = page.locator("input[name=min_bedrooms]").input_value() parsed = urlparse(search_url) query = parse_qs(parsed.query) expected_path = "/search/area/" + opening.value["route"] alternative_path = "/search/apa" if ( max_value != str(maximum) or beds_value != str(minimum_beds) or query.get("max_price") != [str(maximum)] or query.get("min_bedrooms") != [str(minimum_beds)] or not ( (parsed.path == expected_path and query.get("cat") == ["apa"]) or (parsed.path == alternative_path and (parsed.hostname or "").split(".")[0] == opening.value["route"]) ) or query.get("sort", ["date"]) != ["date"] ): filters.deopt( reason="Search URL or filter readback does not match the requested search.", evidence={"url": search_url, "max_price": max_value, "min_bedrooms": beds_value}, ) candidates = page.locator(".cl-search-result") no_results = page.get_by_text( re.compile(r"(no results|nothing found|no matching results)", re.I) ).first if candidates.count() > 0: expect(candidates.first).to_be_visible(timeout=10000) else: expect(no_results).to_be_visible(timeout=10000) page_text = page.locator("body").inner_text() match = re.search(r"\b[\d,]+\s*[-–]\s*[\d,]+\s+of\s+([\d,]+)", page_text) total_matches = int(match.group(1).replace(",", "")) if match else None if candidates.count() == 0: if not no_results.is_visible(): filters.deopt(reason="Neither listings nor a verified empty result state is available.", evidence={"url": search_url}) total_matches = 0 search_path = parsed.path.removeprefix("/search/").strip("/") if search_path == "apa": search_path = opening.value["route"] filters.value = { "inputs": inputs, "search_url": search_url, "search_path": search_path, "page_text": page_text, "total_matches": total_matches, } with browser.scope( "extract-and-validate-listings", goal="Extract current listings in displayed order, decode exact timestamps, and verify the first timestamp against its detail page.", returns={ "type": "object", "properties": { "records": {"type": "array", "items": {"type": "object"}}, "tables": {"type": "object"}, "detail_url": {"type": ["string", "null"]}, "timestamp_verified": {"type": "boolean"}, "results_order": {"type": ["string", "null"]}, }, "required": ["records", "tables", "detail_url", "timestamp_verified", "results_order"], }, ) as extraction: limit = int(filters.value["inputs"]["count"]) records = page.locator(".cl-search-result").evaluate_all( """(es, limit)=>es.slice(0,limit).map(e=>({ title:e.querySelector(".posting-title").textContent, price:e.querySelector(".priceinfo")?.textContent, bedrooms:e.querySelector(".post-bedrooms")?.textContent, neighborhood:e.querySelector(".result-location")?.textContent, posted_date:e.querySelector(".result-posted-date")?.textContent, url:e.querySelector(".posting-title").href,pid:e.dataset.pid }))""", limit, ) if not records and limit and filters.value["total_matches"] != 0: extraction.deopt(reason="Listing extraction returned no usable records.", evidence={"requested": limit}) total = filters.value["total_matches"] if total is not None and len(records) < min(limit, total): extraction.deopt( reason="The current DOM does not contain all requested available listings; additional loading is needed.", evidence={"extracted": len(records), "requested": limit, "total": total}, ) detail_url = None verified = False results_order = None if records: api_url = "https://sapi.craigslist.org/web/v8/postings/search/full?" + urlencode({ "batch": "0-0-360-0-0", "cat": "apa", "max_price": int(filters.value["inputs"]["max_price"]) - 1, "min_bedrooms": int(filters.value["inputs"]["beds"]), "searchPath": urlparse(filters.value["search_url"]).path.removeprefix("/search/").strip("/") if urlparse(filters.value["search_url"]).path != "/search/apa" else opening.value["route"], "lang": "en", "cc": "us", }) response = page.evaluate("(url) => fetch(url).then(r=>r.json())", api_url) data = response.get("data") if isinstance(response, dict) else None if not isinstance(data, dict) or not isinstance(data.get("decode"), dict) or not isinstance(data.get("items"), list): extraction.deopt(reason="Search API decoder structure is unusable.", evidence={"response": response}) base = data["decode"] if not all(isinstance(base.get(k), (int, float)) for k in ("minPostingId", "minPostedDate")): extraction.deopt(reason="Search API timestamp bases are missing.", evidence={"decode": base}) results_order = data.get("resultsOrder") if results_order != "date": extraction.deopt(reason="The search API is not in newest-first order.", evidence={"order": results_order}) by_id = {} for item in data["items"]: if not isinstance(item, list) or len(item) < 4: extraction.deopt(reason="A search API item cannot be decoded.", evidence={"item": item}) by_id[str(base["minPostingId"] + item[0])] = item for record in records: if not all(record.get(k) is not None for k in ("title", "price", "bedrooms", "neighborhood", "url", "pid")): extraction.deopt(reason="A listing is missing requested fields.", evidence={"record": record}) if not record["title"].strip() or record["pid"] not in by_id: extraction.deopt(reason="A listing cannot be joined to current API data.", evidence={"record": record}) if len({r["pid"] for r in records}) != len(records): extraction.deopt(reason="Displayed listing IDs are not unique.", evidence={"ids": [r["pid"] for r in records]}) detail = context.new_page() detail.goto(records[0]["url"]) detail.wait_for_load_state("domcontentloaded") detail.locator("time").first.wait_for(timeout=15000) time_values = detail.locator("time").evaluate_all('(els)=>els.map(e=>({datetime:e.getAttribute("datetime"),text:e.textContent}))') observed_datetime = detail.locator("time").first.get_attribute("datetime") first_item = by_id[records[0]["pid"]] first_epoch = base["minPostedDate"] + first_item[1] if not observed_datetime or not time_values or time_values[0].get("datetime") != observed_datetime: extraction.deopt(reason="Detail page has no consistent exact posting timestamp.", evidence={"times": time_values, "datetime": observed_datetime}) detail_timestamp = datetime.fromisoformat(observed_datetime.replace("Z", "+00:00")) if detail_timestamp.tzinfo is None or detail_timestamp.timestamp() != first_epoch: extraction.deopt( reason="Decoded timestamp disagrees with the detail page.", evidence={"detail_timestamp": observed_datetime, "decoded_epoch": first_epoch}, ) verified = True detail_url = detail.url previous_epoch = None for record in records: item = by_id[record["pid"]] epoch = base["minPostedDate"] + item[1] price_text = record["price"].replace("$", "").replace(",", "").strip() beds_text = record["bedrooms"].replace("br", "").strip() if not price_text.isdigit() or not beds_text.isdigit(): extraction.deopt(reason="Price or bedroom value cannot be normalized.", evidence={"record": record}) record["price"] = int(price_text) record["bedrooms"] = int(beds_text) if record["price"] != item[3]: extraction.deopt(reason="Displayed price disagrees with the joined API listing.", evidence={"record": record, "api_price": item[3]}) record["posted_date"] = datetime.fromtimestamp(epoch, timezone.utc).isoformat() if ( record["price"] >= int(filters.value["inputs"]["max_price"]) or record["bedrooms"] < int(filters.value["inputs"]["beds"]) or (previous_epoch is not None and epoch > previous_epoch) ): extraction.deopt(reason="A listing violates the requested filters or newest-first ordering.", evidence={"record": record}) previous_epoch = epoch raw_tables = browser.store.tables() tables = raw_tables.get("tables", raw_tables) if isinstance(raw_tables, dict) else None if not isinstance(tables, dict): extraction.deopt(reason="Store table inventory has an unusable shape.", evidence={"tables": raw_tables}) extraction.value = { "records": records, "tables": tables, "detail_url": detail_url, "timestamp_verified": verified, "results_order": results_order, } with browser.scope( "save-results-and-report", goal="Save validated listings with terminal outcomes, verify the saved count, recreate artifacts, and report fresh results.", returns={"type": "object"}, ) as saving: records = extraction.value["records"] table = "craigslist_apartment_results" columns = { "listing_id": {"type": "text"}, "rank": {"type": "int"}, "title": {"type": "text"}, "price": {"type": "int"}, "bedrooms": {"type": "int"}, "neighborhood": {"type": "text"}, "posted_date": {"type": "timestamptz"}, "url": {"type": "text"}, } raw_inventory = browser.store.tables() inventory = raw_inventory.get("tables", raw_inventory) if isinstance(raw_inventory, dict) else None if not isinstance(inventory, dict): saving.deopt(reason="Store inventory cannot identify the destination.", evidence={"inventory": raw_inventory}) if table not in inventory: browser.store.define_table(table=table, columns=columns, primaryKey=["listing_id"]) else: definition = inventory[table] if not isinstance(definition, dict) or definition.get("primaryKey") != ["listing_id"] or not isinstance(definition.get("columns"), dict) or not set(columns).issubset(definition["columns"]): saving.deopt(reason="Destination schema does not provide the required columns and listing identity.", evidence={"definition": definition}) before_count = browser.store.query(sql="SELECT COUNT(*) AS count FROM craigslist_apartment_results") if not isinstance(before_count, dict) or before_count.get("truncated") is not False or not isinstance(before_count.get("rows"), list) or len(before_count["rows"]) != 1 or "count" not in before_count["rows"][0]: saving.deopt(reason="Destination initial count is unusable.", evidence={"query": before_count}) initial_table_count = int(before_count["rows"][0]["count"]) existing = browser.store.query(sql="SELECT * FROM craigslist_apartment_results") if not isinstance(existing, dict) or existing.get("truncated") is not False or not isinstance(existing.get("rows"), list): saving.deopt(reason="Destination query is truncated or unusable.", evidence={"query": existing}) existing_by_id = {} for row in existing["rows"]: if not isinstance(row, dict) or not set(columns).issubset(row) or not row["listing_id"]: saving.deopt(reason="An existing destination row lacks its identity or required columns.", evidence={"row": row}) identity = str(row["listing_id"]) if identity in existing_by_id: saving.deopt(reason="Destination identities are duplicated.", evidence={"listing_id": identity}) existing_by_id[identity] = row acknowledged = 0 for rank, record in enumerate(records, 1): browser.record.outcome( collection=table, recordKey=record["pid"], status="extracted", reason=("Extracted displayed rank " + str(rank) + ": " + record["title"])[:1000], ) row = {k: v for k, v in record.items() if k != "pid"} row.update(listing_id=record["pid"], rank=rank) prior = existing_by_id.get(record["pid"]) equivalent = prior is not None and all(prior[k] == row[k] for k in columns if k != "posted_date") if equivalent: equivalent = datetime.fromisoformat(prior["posted_date"].replace("Z", "+00:00")) == datetime.fromisoformat(row["posted_date"].replace("Z", "+00:00")) if equivalent: browser.record.outcome( collection=table, recordKey=record["pid"], status="skipped", reason=("Identical listing already present: " + record["title"])[:1000], ) acknowledged += 1 continue digest = hashlib.sha256( json.dumps(row, sort_keys=True, ensure_ascii=False).encode() ).hexdigest() commit = browser.store.upsert( table=table, row=row, idempotencyKey=table + ":" + record["pid"] + ":" + digest, ) if not isinstance(commit, dict) or commit.get("committed") is not True: browser.record.outcome( collection=table, recordKey=record["pid"], status="failed", reason=("Destination did not acknowledge listing " + record["pid"])[:1000], ) saving.deopt(reason="The store did not acknowledge the upsert.", evidence={"commit": commit, "listing_id": record["pid"]}) browser.record.outcome( collection=table, recordKey=record["pid"], status="saved", reason=("Saved extracted listing: " + record["title"])[:1000], ) acknowledged += 1 id_sql = ",".join("'" + str(record["pid"]).replace("'", "''") + "'" for record in records) verification_sql = "SELECT * FROM craigslist_apartment_results WHERE listing_id IN (" + id_sql + ") ORDER BY rank" if records else "SELECT * FROM craigslist_apartment_results WHERE 1=0 ORDER BY rank" confirmed = browser.store.query(sql=verification_sql) if not isinstance(confirmed, dict) or confirmed.get("truncated") is not False or not isinstance(confirmed.get("rows"), list): saving.deopt(reason="Current saved-listing verification is truncated or unusable.", evidence={"query": confirmed}) confirmed_rows = confirmed["rows"] if len(confirmed_rows) != len(records) or acknowledged != len(records): saving.deopt(reason="Current saved IDs do not account for every extracted listing.", evidence={"verified": len(confirmed_rows), "extracted": len(records), "acknowledged": acknowledged}) for expected_rank, (saved, original) in enumerate(zip(confirmed_rows, records), 1): if not isinstance(saved, dict) or not set(columns).issubset(saved): saving.deopt(reason="A verified destination row lacks required columns.", evidence={"row": saved}) if str(saved["listing_id"]) != str(original["pid"]) or saved["rank"] != expected_rank or any(saved[field] != original[field] for field in ("title", "price", "bedrooms", "neighborhood", "url")): saving.deopt(reason="Current saved listing fields or rank disagree with extraction.", evidence={"saved": saved, "original": original, "rank": expected_rank}) if datetime.fromisoformat(saved["posted_date"].replace("Z", "+00:00")) != datetime.fromisoformat(original["posted_date"].replace("Z", "+00:00")): saving.deopt(reason="The saved posting timestamp differs from extraction.", evidence={"saved": saved, "original": original}) saved_count = len(confirmed_rows) count_query = browser.store.query(sql="SELECT COUNT(*) AS count FROM craigslist_apartment_results") if not isinstance(count_query, dict) or count_query.get("truncated") is not False or not isinstance(count_query.get("rows"), list) or len(count_query["rows"]) != 1 or "count" not in count_query["rows"][0]: saving.deopt(reason="Shared-table count query is unusable.", evidence={"query": count_query}) shared_table_count = int(count_query["rows"][0]["count"]) detail_url = extraction.value["detail_url"] if detail_url is not None: matching_pages = [candidate for candidate in context.pages if candidate != page and candidate.url == detail_url] if len(matching_pages) != 1: saving.deopt(reason="The detail page cannot be uniquely reacquired for closure.", evidence={"url": detail_url, "count": len(matching_pages)}) matching_pages[0].close() page.bring_to_front() output_fields = ["title", "price", "bedrooms", "neighborhood", "posted_date", "url"] output = [{field: record[field] for field in output_fields} for record in records] json_path = "/workspace/craigslist_apartment_results.json" csv_path = "/workspace/craigslist_apartment_results.csv" with open(json_path, "w", encoding="utf-8") as handle: json.dump(output, handle, indent=2, ensure_ascii=False) with open(csv_path, "w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=["title", "price", "bedrooms", "neighborhood", "posted_date", "url"]) writer.writeheader() writer.writerows(output) expect(page.locator("input[name=max_price]")).to_have_value(str(int(filters.value["inputs"]["max_price"]) - 1), timeout=10000) expect(page.locator("input[name=min_bedrooms]")).to_have_value(str(int(filters.value["inputs"]["beds"])), timeout=10000) current_max = page.locator("input[name=max_price]").input_value() current_beds = page.locator("input[name=min_bedrooms]").input_value() current_page_text = page.locator("body").inner_text() current_url = page.url if current_url != filters.value["search_url"]: saving.deopt(reason="Search evidence no longer identifies the extracted search.", evidence={"url": current_url, "extracted_search": filters.value["search_url"]}) saving.value = { "city": filters.value["inputs"]["city"], "criteria": { "price_under_usd": int(filters.value["inputs"]["max_price"]), "minimum_bedrooms": int(filters.value["inputs"]["beds"]), }, "sort": extraction.value["results_order"], "total_matches": filters.value["total_matches"], "requested_count": int(filters.value["inputs"]["count"]), "returned_count": len(output), "results": output, "files": {"csv": csv_path, "json": json_path}, "evidence": { "search_url": current_url, "search_page_text": current_page_text, "max_price_filter": current_max, "min_bedrooms_filter": current_beds, "saved_records_verified": saved_count, "shared_table_initial_count": initial_table_count, "shared_table_total_count": shared_table_count, "verified_all_returned_match_filters": all( r["price"] < int(filters.value["inputs"]["max_price"]) and r["bedrooms"] >= int(filters.value["inputs"]["beds"]) for r in output ), "first_posting_timestamp_cross_checked": extraction.value["timestamp_verified"], "detail_url": detail_url, }, "saved_table": table, "saved_count": saved_count, } browser.done(result=saving.value)