import argparse import logging import os import tempfile import time from pathlib import Path import shutdown from constants import DATA_DIR, REPO_DIR SOURCES = ("rightmove", "onthemarket", "zoopla") SOURCE_CHOICES = (*SOURCES, "all") TEST_MAX_PROPERTIES_PER_SOURCE = 100 TEST_OUTCODES = ( "E1", "N1", "NW1", "SE1", "SW1", "W1", "WC1", "BR1", "CR0", "TW1", ) log = logging.getLogger("finder") def configure_standalone_runtime() -> None: """Keep browser/cache/temp files on the project volume for local runs.""" runtime_dir = REPO_DIR / ".tmp" / "finder" cache_dir = runtime_dir / "cache" temp_dir = runtime_dir / "tmp" cache_dir.mkdir(parents=True, exist_ok=True) temp_dir.mkdir(parents=True, exist_ok=True) os.environ["XDG_CACHE_HOME"] = str(cache_dir) os.environ["TMPDIR"] = str(temp_dir) os.environ["TEMP"] = str(temp_dir) os.environ["TMP"] = str(temp_dir) tempfile.tempdir = str(temp_dir) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Run a manual Greater London-ish property scrape." ) parser.add_argument( "--source", default="all", metavar="SOURCES", help=( "Comma-separated portal(s) to scrape: any of " f"{', '.join(SOURCES)}, or 'all' (default). " "E.g. --source rightmove,onthemarket." ), ) parser.add_argument( "--output-dir", type=Path, default=DATA_DIR, help=f"Directory for parquet output. Defaults to {DATA_DIR}.", ) parser.add_argument( "--outcodes", type=str, default=None, help=( "Comma-separated outcodes to scrape (e.g. 'SW9' or 'SW9,E14,BR1') " "instead of the full Greater London set. Must fall within the " "London-ish areas; takes precedence over --test/--limit-outcodes." ), ) parser.add_argument( "--limit-outcodes", type=int, default=None, help="Limit outcodes for a quick manual smoke test.", ) parser.add_argument( "--max-properties-per-source", type=int, default=None, help="Stop each source after this many transformed listings.", ) parser.add_argument( "--test", action="store_true", help=( "Run a small standalone smoke test: use likely London outcodes and " f"fetch at most {TEST_MAX_PROPERTIES_PER_SOURCE} listings per source." ), ) return parser.parse_args() def configure_logging() -> None: logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("httpcore").setLevel(logging.WARNING) def selected_sources(source: str) -> list[str]: """Resolve --source: one or more comma-separated portals, or 'all'. Accepts e.g. ``rightmove,onthemarket`` (whitespace and case tolerant). Unknown values are rejected; the result is deduplicated and returned in the canonical ``SOURCES`` order so downstream merge/dedup stays deterministic.""" requested = [part.strip().lower() for part in source.split(",")] requested = [part for part in requested if part] if not requested: raise SystemExit("--source was empty") unknown = sorted(set(requested) - set(SOURCE_CHOICES)) if unknown: raise SystemExit( f"Unknown --source value(s): {', '.join(unknown)}. " f"Choose from {', '.join(SOURCE_CHOICES)} (comma-separated)." ) if "all" in requested: return list(SOURCES) return [portal for portal in SOURCES if portal in requested] def main() -> int: args = parse_args() configure_standalone_runtime() configure_logging() # Ctrl+C (and SIGTERM, e.g. `docker stop`) asks the scrapers to wind down # gracefully — each source stops at its next outcode boundary and the run # still persists detail caches and writes the listings collected so far. shutdown.install_signal_handlers() if args.limit_outcodes is not None and args.limit_outcodes < 1: raise SystemExit("--limit-outcodes must be greater than zero") if ( args.max_properties_per_source is not None and args.max_properties_per_source < 1 ): raise SystemExit("--max-properties-per-source must be greater than zero") output_dir = args.output_dir.expanduser().resolve() if args.test and args.output_dir == DATA_DIR: output_dir = (DATA_DIR / "test").expanduser().resolve() output_dir.mkdir(parents=True, exist_ok=True) from scraper import ( build_postcode_coords, build_postcode_index, filter_londonish_outcodes, load_outcodes, run_scrape, ) if args.outcodes is not None: requested = [code.strip().upper() for code in args.outcodes.split(",") if code.strip()] if not requested: raise SystemExit("--outcodes was empty") outcodes = filter_londonish_outcodes(requested) dropped = sorted(set(requested) - set(outcodes)) if dropped: log.warning("Ignoring outcodes outside the Greater London-ish areas: %s", ", ".join(dropped)) if not outcodes: raise SystemExit( "None of the requested outcodes are within the Greater London-ish areas " f"({', '.join(requested)})." ) else: outcodes = load_outcodes() if args.test and args.limit_outcodes is None: preferred = [outcode for outcode in TEST_OUTCODES if outcode in set(outcodes)] if preferred: outcodes = preferred if args.limit_outcodes is not None: outcodes = outcodes[: args.limit_outcodes] if not outcodes: raise SystemExit("No Greater London-ish outcodes loaded; nothing to scrape.") sources = selected_sources(args.source) max_properties_per_source = args.max_properties_per_source if args.test and max_properties_per_source is None: max_properties_per_source = TEST_MAX_PROPERTIES_PER_SOURCE log.info( "Starting sale scrape: source=%s outcodes=%d output_dir=%s test=%s", args.source, len(outcodes), output_dir, args.test, ) started = time.monotonic() pc_index = build_postcode_index() pc_coords = build_postcode_coords() if "zoopla" in sources else None result = run_scrape( outcodes, pc_index, pc_coords=pc_coords, sources=sources, output_dir=output_dir, max_properties_per_source=max_properties_per_source, ) elapsed = time.monotonic() - started if shutdown.stop_requested(): log.warning( "Scrape interrupted after %.1fs; partial results written to %s", elapsed, result.get("path"), ) log.info("Result: %s", result) return 130 # 128 + SIGINT, the conventional Ctrl+C exit code. log.info("Scrape finished in %.1fs", elapsed) log.info("Result: %s", result) if args.test and result.get("errors"): raise SystemExit("Test scrape failed; see errors in the result above.") return 0 if __name__ == "__main__": raise SystemExit(main())