"""Compare an Airbnb listing's price with nearby listings for the same dates.""" from __future__ import annotations import argparse import csv import os import statistics import sys from dataclasses import asdict, dataclass from datetime import date, datetime, timedelta, timezone from pathlib import Path from tapline import SyncTaplineClient, TaplineError SEARCH_RADIUS_KM = 1.5 NUM_SEARCH_PAGES = 2 OUTPUT_PREFIX = "airbnb-listings" @dataclass(frozen=True) class Subject: room_id: str title: str bedrooms: int lat: float lng: float per_night: float currency: str @dataclass(frozen=True) class NearbyListing: room_id: str title: str bedrooms: int rating: float | None stay_total: float per_night: float url: str def find_next_stay(tapline: SyncTaplineClient, room_id: str, nights: int) -> date: today = date.today() calendar = tapline.airbnb.get_calendar( room_id=room_id, month=today.month, year=today.year, count=3 ) days = {day.date: day for month in calendar.months for day in month.days} for offset in range(92): check_in = today + timedelta(days=offset) stay = [days.get((check_in + timedelta(days=night)).isoformat()) for night in range(nights + 1)] if None in stay: continue first_night, checkout_day = stay[0], stay[-1] if ( first_night.available_for_checkin and (first_night.min_nights or 1) <= nights and all(day.available for day in stay[:-1]) and checkout_day.available_for_checkout ): return check_in raise SystemExit(f"Listing {room_id} has no bookable {nights}-night stay in the next 3 months.") def fetch_subject( tapline: SyncTaplineClient, room_id: str, check_in: date, check_out: date, adults: int ) -> Subject: details = tapline.airbnb.get_details(room_id=room_id) quote = tapline.airbnb.get_price( room_id=room_id, check_in=check_in.isoformat(), check_out=check_out.isoformat(), adults=adults, ) if not quote.available or quote.requested_stay_total is None: raise SystemExit(f"Listing {room_id} is not bookable for {check_in} to {check_out}.") return Subject( room_id=room_id, title=details.title or "", bedrooms=details.bedrooms or 0, lat=details.location.lat, lng=details.location.lng, per_night=quote.requested_stay_total.amount / (check_out - check_in).days, currency=quote.currency, ) def fetch_nearby_listings( tapline: SyncTaplineClient, subject: Subject, check_in: date, check_out: date, adults: int ) -> list[NearbyListing]: nights = (check_out - check_in).days nearby: dict[str, NearbyListing] = {} cursor = "" for _ in range(NUM_SEARCH_PAGES): page = tapline.airbnb.search( lat=subject.lat, lon=subject.lng, radius_km=SEARCH_RADIUS_KM, check_in=check_in.isoformat(), check_out=check_out.isoformat(), adults=adults, min_bedrooms=subject.bedrooms, cursor=cursor, ) for search_listing in page.listings: if ( search_listing.room_id in (None, subject.room_id) or search_listing.bedrooms != subject.bedrooms or search_listing.price.amount is None ): continue nearby[search_listing.room_id] = NearbyListing( room_id=search_listing.room_id, title=search_listing.name or search_listing.title or "", bedrooms=search_listing.bedrooms, rating=search_listing.rating, stay_total=search_listing.price.amount, per_night=round(search_listing.price.amount / nights, 2), url=search_listing.url or "", ) if not page.pagination.next_cursor: break cursor = page.pagination.next_cursor return sorted(nearby.values(), key=lambda listing: listing.per_night) def find_price_neighbours(subject: Subject, nearby: list[NearbyListing]) -> list[NearbyListing]: cheaper = [listing for listing in nearby if listing.per_night < subject.per_night] pricier = [listing for listing in nearby if listing.per_night >= subject.per_night] return cheaper[-5:] + pricier[:5] def write_csv(nearby: list[NearbyListing], output_dir: Path) -> Path: output_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") output_path = output_dir / f"{OUTPUT_PREFIX}-{timestamp}.csv" with output_path.open("w", newline="", encoding="utf-8") as output: writer = csv.DictWriter(output, fieldnames=list(NearbyListing.__dataclass_fields__)) writer.writeheader() for listing in nearby: writer.writerow(asdict(listing)) return output_path def print_report( subject: Subject, nearby: list[NearbyListing], check_in: date, check_out: date ) -> None: per_night = [listing.per_night for listing in nearby] lower, median, upper = statistics.quantiles(per_night, n=4) cheaper = sum(1 for price in per_night if price < subject.per_night) money = subject.currency + " {:,.0f}" print(f"Your listing: {subject.title} ({subject.bedrooms} bedrooms)") print(f"Dates: {check_in} to {check_out}") print(f"Your price: {money.format(subject.per_night)} per night, fees included") print(f"Listings: {len(nearby)} with {subject.bedrooms} bedrooms within {SEARCH_RADIUS_KM} km") print( f"Median price: {money.format(median)} per night " f"(middle half {money.format(lower)} to {money.format(upper)})" ) print(f"Cheaper than you: {cheaper} of {len(nearby)} ({cheaper / len(nearby):.0%})") print() print("Closest listings by price:") for listing in find_price_neighbours(subject, nearby): rating = f"{listing.rating:.2f}" if listing.rating else "n/a" print(f" {money.format(listing.per_night):>10} {rating:>4} {listing.title[:48]} {listing.url}") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("room_id", help="Your Airbnb listing ID, from its URL") parser.add_argument( "--check-in", type=date.fromisoformat, help="Check-in date, YYYY-MM-DD (default: your listing's next bookable stay)", ) parser.add_argument("--nights", type=int, default=3, help="Nights to price (default: 3)") parser.add_argument("--adults", type=int, default=2, help="Guests to price for (default: 2)") parser.add_argument( "--output-dir", type=Path, default=Path("."), help="Directory for the timestamped CSV (default: current directory)", ) return parser.parse_args() def main() -> int: args = parse_args() try: with SyncTaplineClient(api_key=os.environ["TAPLINE_API_KEY"]) as tapline: check_in = args.check_in or find_next_stay(tapline, args.room_id, args.nights) check_out = check_in + timedelta(days=args.nights) subject = fetch_subject(tapline, args.room_id, check_in, check_out, args.adults) nearby = fetch_nearby_listings(tapline, subject, check_in, check_out, args.adults) except TaplineError as error: print(f"Tapline request failed: {error}", file=sys.stderr) return 1 if len(nearby) < 4: print(f"Only {len(nearby)} listings found; widen SEARCH_RADIUS_KM.", file=sys.stderr) return 1 print_report(subject, nearby, check_in, check_out) print() print(f"Saved {len(nearby)} listings to {write_csv(nearby, args.output_dir)}") return 0 if __name__ == "__main__": sys.exit(main())