scrape.py

"""Airbnb scraper in Python: listings, prices and availability to CSV.

Runs the Airbnb Listings Scraper (an Apify Actor) for one or more places and
saves one row per listing to a CSV file.

    export APIFY_TOKEN="<your-apify-token>"
    python scrape.py "Lisbon, Portugal" --check-in 2026-11-20 --check-out 2026-11-23
    python scrape.py "Paris, France" "Rome, Italy" --currency EUR --max-results 30
"""

import argparse
import csv
import json
import os
import sys
from decimal import Decimal

from apify_client import ApifyClient

ACTOR = "myth84/airbnb-listings-scraper"

# Columns written to the CSV. Every row of the dataset has more (see the README).
COLUMNS = [
    "platformListingId",
    "title",
    "propertyType",
    "roomType",
    "city",
    "checkIn",
    "checkOut",
    "nights",
    "currency",
    "pricePerNight",
    "priceTotal",
    "priceBeforeDiscount",
    "taxesAndFees",
    "rating",
    "reviewCount",
    "bedrooms",
    "beds",
    "bathrooms",
    "badges",
    "latitude",
    "longitude",
    "searchLocation",
    "url",
]


def make_parser(description):
    parser = argparse.ArgumentParser(description=description)
    parser.add_argument("locations", nargs="+", help='places as typed on Airbnb, e.g. "Lisbon, Portugal"')
    parser.add_argument("--check-in", help="check-in date, YYYY-MM-DD (with --check-out)")
    parser.add_argument("--check-out", help="check-out date, YYYY-MM-DD")
    parser.add_argument("--check-in-days", type=int, default=30,
                        help="without fixed dates: check-in in N days, recomputed at every run (default 30)")
    parser.add_argument("--nights", type=int, default=3, help="without fixed dates: nights of the stay (default 3)")
    parser.add_argument("--adults", type=int, default=2)
    parser.add_argument("--currency", default="USD", help="ISO code: USD, EUR, GBP...")
    parser.add_argument("--room-type", action="append",
                        choices=["entire_home", "private_room", "shared_room", "hotel_room"],
                        help="repeat to allow several types (default: all)")
    parser.add_argument("--price-max", type=int, help="maximum nightly price, in the chosen currency")
    parser.add_argument("--max-results", type=int, default=50,
                        help="listings per place (default 50); you pay per listing saved")
    parser.add_argument("--max-charge", type=Decimal, help="stop the run once it has cost this many USD")
    return parser


def build_input(args):
    if bool(args.check_in) != bool(args.check_out):
        sys.exit("--check-in and --check-out go together.")
    run_input = {
        "locations": args.locations,
        "adults": args.adults,
        "currency": args.currency,
        "maxResults": args.max_results,
    }
    if args.check_in:
        run_input["checkIn"] = args.check_in
        run_input["checkOut"] = args.check_out
    else:
        run_input["checkInDaysFromNow"] = args.check_in_days
        run_input["nights"] = args.nights
    if args.room_type:
        run_input["roomTypes"] = args.room_type
    if args.price_max:
        run_input["priceMax"] = args.price_max
    return run_input


def run_actor(run_input, max_charge=None):
    """Runs the Actor, waits for it to finish and returns the rows of its dataset."""
    if not os.environ.get("APIFY_TOKEN"):
        sys.exit("Set APIFY_TOKEN first: Apify Console > Settings > API & Integrations.")
    client = ApifyClient(os.environ["APIFY_TOKEN"])
    run = client.actor(ACTOR).call(run_input=run_input, max_total_charge_usd=max_charge)
    if run is None:
        sys.exit("The run could not be started.")
    status = getattr(run.status, "value", run.status)  # SUCCEEDED, FAILED, TIMED-OUT...
    if status != "SUCCEEDED":
        sys.exit(f"The run ended as {status}: open its log in Apify Console.")
    return list(client.dataset(run.default_dataset_id).iterate_items())


def cell(value):
    """Lists and objects as JSON text, so that every CSV row has the same columns."""
    return json.dumps(value, ensure_ascii=False) if isinstance(value, (list, dict)) else value


def write_csv(rows, path, columns=COLUMNS, append=False):
    new_file = not (append and os.path.exists(path))
    with open(path, "a" if append else "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        if new_file:
            writer.writerow(columns)
        for row in rows:
            writer.writerow([cell(row.get(column)) for column in columns])


def main():
    parser = make_parser("Scrape Airbnb listings, prices and availability to CSV.")
    parser.add_argument("--output", default="airbnb_listings.csv", help="CSV file (default airbnb_listings.csv)")
    args = parser.parse_args()

    rows = run_actor(build_input(args), args.max_charge)
    write_csv(rows, args.output)
    print(f"{len(rows)} listings saved to {args.output}")

    priced = sorted((row for row in rows if row.get("pricePerNight") is not None),
                    key=lambda row: row["pricePerNight"])
    for row in priced[:5]:
        print(f'{row["pricePerNight"]:>9} {row["currency"]}/night  '
              f'{row.get("rating") or "-":>4}  {row["title"]}')


if __name__ == "__main__":
    main()