diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 42b4d57..dd78f84 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -7,6 +7,27 @@ on: branches: [main] jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - uses: actions/setup-python@v6 + with: + python-version: "3.12" + - run: pip install ruff + - run: ruff check boligwatch.py test_boligwatch.py + - run: ruff format --check boligwatch.py test_boligwatch.py + + typecheck: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - uses: actions/setup-python@v6 + with: + python-version: "3.12" + - run: pip install mypy + - run: mypy boligwatch.py + test: runs-on: ubuntu-latest strategy: diff --git a/CHANGELOG.md b/CHANGELOG.md index bc7e8cf..82e8f3d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,8 +6,17 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/). ## [Unreleased] +### Added + +- Cloudflare bypass via `curl_cffi` — when installed, API requests use Chrome TLS fingerprint impersonation to avoid Cloudflare bot challenges (HTTP 403). Install with `pip install curl_cffi`. Falls back to stdlib `urllib` when not installed. +- HTTP 403 errors are now retried with exponential backoff (previously only 429 and 5xx were retried). +- CI: ruff linting + formatting, mypy type checking, README badges. +- `pyproject.toml` with ruff and mypy configuration. + ### Fixed +- `Z` suffix in ISO 8601 dates is now handled on Python 3.10 (where `fromisoformat()` doesn't support it natively). + - Re-listing detection now parses ISO 8601 dates instead of comparing strings, fixing incorrect results when timezone formats differ (`Z` vs `+00:00`). - Added missing `--parking` and `--elevator` CLI flags (filters were already supported in config and MCP but had no argparse arguments). diff --git a/README.md b/README.md index c66e600..48854d2 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,17 @@ # BoligWatch +[![CI](https://github.com/nille/boligwatch/actions/workflows/ci.yml/badge.svg)](https://github.com/nille/boligwatch/actions/workflows/ci.yml) +[![Python 3.10+](https://img.shields.io/badge/python-3.10%2B-blue)](https://www.python.org/downloads/) +[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff) +[![License: MIT](https://img.shields.io/badge/license-MIT-green)](LICENSE) + CLI tool and MCP server that monitors [boligportal.dk](https://www.boligportal.dk) for new rental listings in Denmark. Polls the same public search API the website uses — no account or API key required. ## Requirements - Python 3.10+ -- No external dependencies for CLI mode (stdlib only) +- CLI mode works with stdlib only (zero required dependencies) +- `pip install curl_cffi` — recommended, bypasses Cloudflare bot protection (see [Cloudflare bypass](#cloudflare-bypass)) - `pip install mcp` for MCP server mode ## Installation @@ -19,9 +25,12 @@ cd boligwatch python3 -m venv .venv source .venv/bin/activate -# CLI mode works out of the box — no dependencies needed +# CLI mode works out of the box with stdlib only python boligwatch.py --help +# Recommended: install curl_cffi to bypass Cloudflare bot protection +pip install curl_cffi + # For MCP server mode, install the MCP SDK pip install mcp @@ -514,6 +523,20 @@ contact the landlord with a message in Danish, and notify me on Slack. Because the extension bridges into your existing browser session, Claude authenticates as you — no separate login flow, no stored credentials. +## Cloudflare bypass + +Boligportal.dk uses Cloudflare bot protection, which can block requests from standard HTTP clients like Python's `urllib` with an HTTP 403 and a JavaScript challenge. When this happens, the API becomes unreachable. + +Installing `curl_cffi` enables Chrome TLS fingerprint impersonation, which bypasses the challenge transparently: + +```bash +pip install curl_cffi +``` + +When `curl_cffi` is installed, BoligWatch automatically uses it for all API requests. When it's not installed, BoligWatch falls back to stdlib `urllib` — which works fine when Cloudflare isn't actively challenging requests. + +Both backends retry on HTTP 403, 429, and 5xx errors with exponential backoff. + ## Listing output format Each listing returned by the API includes: diff --git a/boligwatch.py b/boligwatch.py index 062a513..fe218d2 100755 --- a/boligwatch.py +++ b/boligwatch.py @@ -33,6 +33,13 @@ API_URL = "https://www.boligportal.dk/api/search/list" LISTING_BASE = "https://www.boligportal.dk" + +try: + from curl_cffi import requests as _cffi_requests + + _HAS_CURL_CFFI = True +except ImportError: + _HAS_CURL_CFFI = False DEFAULT_SEEN_FILE = Path(__file__).parent / ".boligwatch_seen.json" DEFAULT_CONFIG_FILE = Path(__file__).parent / "boligwatch_config.json" DEFAULT_LOG_FILE = Path(__file__).parent / "boligwatch.log" @@ -44,11 +51,10 @@ # -- Config ---------------------------------------------------------------- + @dataclass class SearchConfig: - categories: list[str] = field( - default_factory=lambda: ["rental_apartment", "rental_house", "rental_townhouse"] - ) + categories: list[str] = field(default_factory=lambda: ["rental_apartment", "rental_house", "rental_townhouse"]) city_level_1: list[str] | None = field(default_factory=lambda: ["københavn"]) city_level_2: list[str] | None = None min_lat: float | None = None @@ -189,6 +195,7 @@ def from_dict(cls, data: dict[str, Any]) -> SearchConfig: # -- API Client ------------------------------------------------------------ + @dataclass class Listing: id: int @@ -276,7 +283,10 @@ def format_short(self) -> str: date_str = "" if self.advertised_date: try: - dt = datetime.fromisoformat(self.advertised_date) + ad = self.advertised_date + if ad.endswith("Z"): + ad = ad[:-1] + "+00:00" + dt = datetime.fromisoformat(ad) date_str = dt.strftime(" [%Y-%m-%d %H:%M]") except (ValueError, TypeError): pass @@ -293,8 +303,57 @@ def format_short(self) -> str: BACKOFF_MAX = 300.0 -def _api_request(url: str, body_bytes: bytes) -> dict[str, Any]: - """POST to the boligportal API with exponential backoff + jitter.""" +def _backoff_delay(attempt: int) -> float: + delay = min(BACKOFF_BASE ** (attempt + 1), BACKOFF_MAX) + return delay + random.uniform(0, delay * 0.5) + + +def _api_request_cffi(url: str, body_bytes: bytes) -> dict[str, Any]: + """POST using curl_cffi with Chrome TLS fingerprint (bypasses Cloudflare).""" + for attempt in range(MAX_RETRIES): + try: + resp = _cffi_requests.post( + url, + headers={ + "Content-Type": "text/plain;charset=UTF-8", + "Origin": "https://www.boligportal.dk", + "Referer": "https://www.boligportal.dk/", + }, + data=body_bytes, + impersonate="chrome", + timeout=30, + ) + if resp.status_code == 200: + return dict(resp.json()) + if resp.status_code in (403, 429) or resp.status_code >= 500: + wait = _backoff_delay(attempt) + print( + f" HTTP {resp.status_code} -- backing off {wait:.0f}s (attempt {attempt + 1}/{MAX_RETRIES})", + file=sys.stderr, + ) + time.sleep(wait) + continue + raise urllib.error.HTTPError( + url, + resp.status_code, + resp.reason or "Error", + {}, # type: ignore[arg-type] + None, + ) + except (OSError, TimeoutError): + if attempt == MAX_RETRIES - 1: + raise + wait = _backoff_delay(attempt) + print( + f" Network error -- retrying in {wait:.0f}s (attempt {attempt + 1}/{MAX_RETRIES})", + file=sys.stderr, + ) + time.sleep(wait) + raise RuntimeError(f"Failed after {MAX_RETRIES} retries") + + +def _api_request_urllib(url: str, body_bytes: bytes) -> dict[str, Any]: + """POST using stdlib urllib (no external dependencies).""" for attempt in range(MAX_RETRIES): req = urllib.request.Request( url, @@ -307,12 +366,10 @@ def _api_request(url: str, body_bytes: bytes) -> dict[str, Any]: ) try: with urllib.request.urlopen(req, timeout=30) as resp: - return json.loads(resp.read().decode("utf-8")) + return dict(json.loads(resp.read().decode("utf-8"))) except urllib.error.HTTPError as e: - if e.code == 429 or e.code >= 500: - delay = min(BACKOFF_BASE ** (attempt + 1), BACKOFF_MAX) - jitter = random.uniform(0, delay * 0.5) - wait = delay + jitter + if e.code in (403, 429) or e.code >= 500: + wait = _backoff_delay(attempt) print( f" HTTP {e.code} -- backing off {wait:.0f}s (attempt {attempt + 1}/{MAX_RETRIES})", file=sys.stderr, @@ -323,9 +380,7 @@ def _api_request(url: str, body_bytes: bytes) -> dict[str, Any]: except (urllib.error.URLError, TimeoutError): if attempt == MAX_RETRIES - 1: raise - delay = min(BACKOFF_BASE ** (attempt + 1), BACKOFF_MAX) - jitter = random.uniform(0, delay * 0.5) - wait = delay + jitter + wait = _backoff_delay(attempt) print( f" Network error -- retrying in {wait:.0f}s (attempt {attempt + 1}/{MAX_RETRIES})", file=sys.stderr, @@ -334,6 +389,17 @@ def _api_request(url: str, body_bytes: bytes) -> dict[str, Any]: raise RuntimeError(f"Failed after {MAX_RETRIES} retries") +def _api_request(url: str, body_bytes: bytes) -> dict[str, Any]: + """POST to the boligportal API with exponential backoff + jitter. + + Uses curl_cffi when installed (bypasses Cloudflare bot challenges via + Chrome TLS fingerprint impersonation). Falls back to stdlib urllib. + """ + if _HAS_CURL_CFFI: + return _api_request_cffi(url, body_bytes) + return _api_request_urllib(url, body_bytes) + + def fetch_listings(config: SearchConfig) -> list[Listing]: body = config.to_api_body() body_bytes = json.dumps(body).encode("utf-8") @@ -356,6 +422,7 @@ def fetch_listings(config: SearchConfig) -> list[Listing]: # -- Seen-Listings Tracker ------------------------------------------------- + class SeenTracker: """Track which listings have been seen, detecting re-listings. @@ -376,9 +443,7 @@ def _load(self) -> None: self._seen = json.load(f) def _save(self) -> None: - fd, tmp = tempfile.mkstemp( - dir=self._path.parent, suffix=".tmp", prefix=".boligwatch_" - ) + fd, tmp = tempfile.mkstemp(dir=self._path.parent, suffix=".tmp", prefix=".boligwatch_") try: with open(fd, "w", encoding="utf-8") as f: json.dump(self._seen, f, indent=2, ensure_ascii=False) @@ -395,6 +460,8 @@ def _get_ad_date(entry: Any) -> str | None: @staticmethod def _parse_date(value: str) -> datetime: + if value.endswith("Z"): + value = value[:-1] + "+00:00" return datetime.fromisoformat(value) def is_new(self, listing_id: int, advertised_date: str | None = None) -> bool: @@ -449,6 +516,7 @@ def path(self) -> Path: # -- Logging --------------------------------------------------------------- + def setup_logging(log_file: Path | None, verbose: bool) -> None: level = logging.DEBUG if verbose else logging.INFO fmt = "%(asctime)s %(levelname)s %(message)s" @@ -460,6 +528,7 @@ def setup_logging(log_file: Path | None, verbose: bool) -> None: # -- CLI ------------------------------------------------------------------- + def print_header(config: SearchConfig, total: int, new_count: int) -> None: now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") if config.min_lat is not None: @@ -470,7 +539,9 @@ def print_header(config: SearchConfig, total: int, new_count: int) -> None: location = "all" rooms = "" if config.rooms_min is not None and config.rooms_max is not None: - rooms = f"{config.rooms_min}-{config.rooms_max}" if config.rooms_min != config.rooms_max else str(config.rooms_min) + rooms = ( + f"{config.rooms_min}-{config.rooms_max}" if config.rooms_min != config.rooms_max else str(config.rooms_min) + ) elif config.rooms_min is not None: rooms = f"{config.rooms_min}+" elif config.rooms_max is not None: @@ -479,12 +550,12 @@ def print_header(config: SearchConfig, total: int, new_count: int) -> None: size = f", min {config.min_size_m2}m\u00b2" if config.min_size_m2 else "" period = f", min {config.min_rental_period}mo" if config.min_rental_period else "" - print(f"\n{'='*60}") + print(f"\n{'=' * 60}") print(f"BoligWatch {now}") - print(f"{'='*60}") + print(f"{'=' * 60}") print(f"Search: {location} | {rooms} rooms{rent}{size}{period}") print(f"Found: {total} total, {new_count} new") - print(f"{'='*60}") + print(f"{'=' * 60}") def run_once( @@ -495,15 +566,15 @@ def run_once( peek: bool = False, ) -> list[Listing]: listings = fetch_listings(config) - new_listings = [l for l in listings if tracker.is_new(l.id, l.advertised_date)] + new_listings = [it for it in listings if tracker.is_new(it.id, it.advertised_date)] if json_output: if new_listings: - output = [l.to_json_dict() for l in new_listings] + output = [it.to_json_dict() for it in new_listings] print(json.dumps(output, ensure_ascii=False)) if not peek: - ad_dates = {l.id: l.advertised_date for l in new_listings} - tracker.mark_all_seen([l.id for l in new_listings], advertised_dates=ad_dates) + ad_dates = {it.id: it.advertised_date for it in new_listings} + tracker.mark_all_seen([it.id for it in new_listings], advertised_dates=ad_dates) else: print("[]") return new_listings @@ -516,8 +587,8 @@ def run_once( print() for listing in new_listings: print(f"\n{listing.format_short()}") - ad_dates = {l.id: l.advertised_date for l in new_listings} - tracker.mark_all_seen([l.id for l in new_listings], advertised_dates=ad_dates) + ad_dates = {it.id: it.advertised_date for it in new_listings} + tracker.mark_all_seen([it.id for it in new_listings], advertised_dates=ad_dates) elif not quiet: print("\nNo new listings since last check.") @@ -572,11 +643,26 @@ def load_config(path: Path | None) -> SearchConfig: # -- MCP Server ------------------------------------------------------------ _RESTRICTIVE_FILTERS = { - "rooms_min", "rooms_max", "max_rent", "min_size_m2", "min_rental_period", - "max_available_from", "pet_friendly", "balcony", "furnished", "parking", - "elevator", "shareable", "student_only", "senior_friendly", - "social_housing", "newbuild", "electric_charging_station", - "dishwasher", "washing_machine", "dryer", + "rooms_min", + "rooms_max", + "max_rent", + "min_size_m2", + "min_rental_period", + "max_available_from", + "pet_friendly", + "balcony", + "furnished", + "parking", + "elevator", + "shareable", + "student_only", + "senior_friendly", + "social_housing", + "newbuild", + "electric_charging_station", + "dishwasher", + "washing_machine", + "dryer", } @@ -675,10 +761,7 @@ def _build_search_config( if not explicit: return base - overrides: dict[str, Any] = { - k: v for k, v in base.to_dict().items() - if k not in _RESTRICTIVE_FILTERS - } + overrides: dict[str, Any] = {k: v for k, v in base.to_dict().items() if k not in _RESTRICTIVE_FILTERS} overrides.update(explicit) if "min_lat" in explicit: @@ -693,10 +776,7 @@ def run_mcp_server(config: SearchConfig, tracker: SeenTracker) -> None: from mcp.server.fastmcp import FastMCP except ImportError: print( - "MCP mode requires the 'mcp' package. Install it with:\n" - " pip install mcp\n" - "or:\n" - " uv pip install mcp", + "MCP mode requires the 'mcp' package. Install it with:\n pip install mcp\nor:\n uv pip install mcp", file=sys.stderr, ) sys.exit(1) @@ -778,21 +858,36 @@ def search_listings( max_pages: Maximum pages to fetch (18 listings per page, default 5). """ search = _build_search_config( - config, cities=cities, min_lat=min_lat, min_lng=min_lng, - max_lat=max_lat, max_lng=max_lng, rooms_min=rooms_min, - rooms_max=rooms_max, max_rent=max_rent, min_size_m2=min_size_m2, + config, + cities=cities, + min_lat=min_lat, + min_lng=min_lng, + max_lat=max_lat, + max_lng=max_lng, + rooms_min=rooms_min, + rooms_max=rooms_max, + max_rent=max_rent, + min_size_m2=min_size_m2, min_rental_period=min_rental_period, max_available_from=max_available_from, - pet_friendly=pet_friendly, balcony=balcony, furnished=furnished, - parking=parking, elevator=elevator, shareable=shareable, - student_only=student_only, senior_friendly=senior_friendly, - social_housing=social_housing, newbuild=newbuild, + pet_friendly=pet_friendly, + balcony=balcony, + furnished=furnished, + parking=parking, + elevator=elevator, + shareable=shareable, + student_only=student_only, + senior_friendly=senior_friendly, + social_housing=social_housing, + newbuild=newbuild, electric_charging_station=electric_charging_station, - dishwasher=dishwasher, washing_machine=washing_machine, - dryer=dryer, max_pages=max_pages, + dishwasher=dishwasher, + washing_machine=washing_machine, + dryer=dryer, + max_pages=max_pages, ) listings = fetch_listings(search) - return json.dumps([l.to_json_dict() for l in listings], ensure_ascii=False) + return json.dumps([it.to_json_dict() for it in listings], ensure_ascii=False) @mcp.tool() def get_new_listings( @@ -862,25 +957,40 @@ def get_new_listings( mark_as_seen: If True, mark returned listings as seen (default False). """ search = _build_search_config( - config, cities=cities, min_lat=min_lat, min_lng=min_lng, - max_lat=max_lat, max_lng=max_lng, rooms_min=rooms_min, - rooms_max=rooms_max, max_rent=max_rent, min_size_m2=min_size_m2, + config, + cities=cities, + min_lat=min_lat, + min_lng=min_lng, + max_lat=max_lat, + max_lng=max_lng, + rooms_min=rooms_min, + rooms_max=rooms_max, + max_rent=max_rent, + min_size_m2=min_size_m2, min_rental_period=min_rental_period, max_available_from=max_available_from, - pet_friendly=pet_friendly, balcony=balcony, furnished=furnished, - parking=parking, elevator=elevator, shareable=shareable, - student_only=student_only, senior_friendly=senior_friendly, - social_housing=social_housing, newbuild=newbuild, + pet_friendly=pet_friendly, + balcony=balcony, + furnished=furnished, + parking=parking, + elevator=elevator, + shareable=shareable, + student_only=student_only, + senior_friendly=senior_friendly, + social_housing=social_housing, + newbuild=newbuild, electric_charging_station=electric_charging_station, - dishwasher=dishwasher, washing_machine=washing_machine, - dryer=dryer, max_pages=max_pages, + dishwasher=dishwasher, + washing_machine=washing_machine, + dryer=dryer, + max_pages=max_pages, ) listings = fetch_listings(search) - new = [l for l in listings if tracker.is_new(l.id, l.advertised_date)] + new = [it for it in listings if tracker.is_new(it.id, it.advertised_date)] if mark_as_seen and new: - ad_dates = {l.id: l.advertised_date for l in new} - tracker.mark_all_seen([l.id for l in new], advertised_dates=ad_dates) - return json.dumps([l.to_json_dict() for l in new], ensure_ascii=False) + ad_dates = {it.id: it.advertised_date for it in new} + tracker.mark_all_seen([it.id for it in new], advertised_dates=ad_dates) + return json.dumps([it.to_json_dict() for it in new], ensure_ascii=False) @mcp.tool() def mark_seen(ids: list[int]) -> dict[str, Any]: @@ -911,6 +1021,7 @@ def get_seen_stats() -> dict[str, Any]: # -- Entry point ----------------------------------------------------------- + def main() -> None: parser = argparse.ArgumentParser( description="Monitor boligportal.dk for new rental listings.", @@ -929,42 +1040,116 @@ def main() -> None: """, ) parser.add_argument("--watch", "-w", action="store_true", help="continuously poll for new listings") - parser.add_argument("--interval", "-i", type=int, default=300, help="poll interval in seconds (default: 300)") - parser.add_argument("--json", action="store_true", dest="json_output", help="output new listings as JSON (for piping to agents)") - parser.add_argument("--peek", action="store_true", help="like --json but do NOT mark listings as seen (for retry-safe workflows)") - parser.add_argument("--mark-seen", nargs="+", type=int, metavar="ID", help="mark specific listing IDs as seen") + parser.add_argument( + "--interval", + "-i", + type=int, + default=300, + help="poll interval in seconds (default: 300)", + ) + parser.add_argument( + "--json", + action="store_true", + dest="json_output", + help="output new listings as JSON (for piping to agents)", + ) + parser.add_argument( + "--peek", + action="store_true", + help="like --json but do NOT mark listings as seen (for retry-safe workflows)", + ) + parser.add_argument( + "--mark-seen", + nargs="+", + type=int, + metavar="ID", + help="mark specific listing IDs as seen", + ) parser.add_argument("--config", "-c", type=Path, help="path to config JSON file") parser.add_argument("--init-config", action="store_true", help="generate a config template file") - parser.add_argument("--seen-file", type=Path, default=DEFAULT_SEEN_FILE, help="path to seen-listings tracker") + parser.add_argument( + "--seen-file", + type=Path, + default=DEFAULT_SEEN_FILE, + help="path to seen-listings tracker", + ) parser.add_argument("--log-file", type=Path, default=None, help="write log to file (default: none)") parser.add_argument("--verbose", "-v", action="store_true", help="verbose logging") - parser.add_argument("--reset", action="store_true", help="clear seen-listings history before running") + parser.add_argument( + "--reset", + action="store_true", + help="clear seen-listings history before running", + ) parser.add_argument("--mcp", action="store_true", help="start as MCP server (stdio transport)") # Inline filter overrides (take precedence over config file) parser.add_argument("--city", action="append", dest="cities", help="city to search (can repeat)") - parser.add_argument("--bbox", type=str, metavar="S,W,N,E", - help="bounding box as min_lat,min_lng,max_lat,max_lng (replaces --city)") + parser.add_argument( + "--bbox", + type=str, + metavar="S,W,N,E", + help="bounding box as min_lat,min_lng,max_lat,max_lng (replaces --city)", + ) parser.add_argument("--rooms-min", type=int, help="minimum rooms") parser.add_argument("--rooms-max", type=int, help="maximum rooms") parser.add_argument("--max-rent", type=int, help="maximum monthly rent in DKK") parser.add_argument("--min-size", type=int, help="minimum size in m2") parser.add_argument("--min-rental-period", type=int, help="minimum lease in months (12 = 1 year)") parser.add_argument("--max-pages", type=int, help="max pages to fetch (18 results each)") - parser.add_argument("--max-available-from", type=str, metavar="YYYY-MM-DD", help="latest move-in date") + parser.add_argument( + "--max-available-from", + type=str, + metavar="YYYY-MM-DD", + help="latest move-in date", + ) parser.add_argument("--pet-friendly", action="store_true", default=None, help="only pet-friendly") parser.add_argument("--balcony", action="store_true", default=None, help="must have balcony/terrace") parser.add_argument("--furnished", action="store_true", default=None, help="must be furnished") parser.add_argument("--parking", action="store_true", default=None, help="must have parking") parser.add_argument("--elevator", action="store_true", default=None, help="must have elevator") - parser.add_argument("--shareable", action="store_true", default=None, help="must be shareable (delevenlig)") - parser.add_argument("--student-only", action="store_true", default=None, help="student-only listings") - parser.add_argument("--senior-friendly", action="store_true", default=None, help="senior-friendly listings") - parser.add_argument("--social-housing", action="store_true", default=None, help="social housing only (almen bolig)") - parser.add_argument("--newbuild", action="store_true", default=None, help="new-build/project rentals only (projektudlejning)") - parser.add_argument("--ev-charging", action="store_true", default=None, help="must have EV charging station (ladestander)") + parser.add_argument( + "--shareable", + action="store_true", + default=None, + help="must be shareable (delevenlig)", + ) + parser.add_argument( + "--student-only", + action="store_true", + default=None, + help="student-only listings", + ) + parser.add_argument( + "--senior-friendly", + action="store_true", + default=None, + help="senior-friendly listings", + ) + parser.add_argument( + "--social-housing", + action="store_true", + default=None, + help="social housing only (almen bolig)", + ) + parser.add_argument( + "--newbuild", + action="store_true", + default=None, + help="new-build/project rentals only (projektudlejning)", + ) + parser.add_argument( + "--ev-charging", + action="store_true", + default=None, + help="must have EV charging station (ladestander)", + ) parser.add_argument("--dishwasher", action="store_true", default=None, help="must have dishwasher") - parser.add_argument("--washing-machine", action="store_true", default=None, help="must have washing machine") + parser.add_argument( + "--washing-machine", + action="store_true", + default=None, + help="must have washing machine", + ) parser.add_argument("--dryer", action="store_true", default=None, help="must have dryer") args = parser.parse_args() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..6a833ef --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,17 @@ +[tool.ruff] +target-version = "py310" +line-length = 120 + +[tool.ruff.lint] +select = ["E", "F", "W", "I"] + +[tool.mypy] +python_version = "3.10" +warn_return_any = true +warn_unused_configs = true +disallow_untyped_defs = true +check_untyped_defs = true + +[[tool.mypy.overrides]] +module = ["curl_cffi.*", "mcp.*"] +ignore_missing_imports = true diff --git a/test_boligwatch.py b/test_boligwatch.py index 29f4182..1c323f9 100644 --- a/test_boligwatch.py +++ b/test_boligwatch.py @@ -1,40 +1,52 @@ -"""Tests for SearchConfig, _build_search_config, to_api_body, and SeenTracker.""" +"""Tests for SearchConfig, _build_search_config, to_api_body, SeenTracker, and API request.""" from __future__ import annotations import json import logging from pathlib import Path +from unittest.mock import MagicMock, patch import pytest -from boligwatch import SearchConfig, SeenTracker, _build_search_config, _RESTRICTIVE_FILTERS, MAX_PAGES_CEILING - +from boligwatch import ( + _RESTRICTIVE_FILTERS, + MAX_PAGES_CEILING, + SearchConfig, + SeenTracker, + _api_request, + _api_request_urllib, + _backoff_delay, + _build_search_config, +) # -- Fixtures ---------------------------------------------------------------- + @pytest.fixture def base_config() -> SearchConfig: """Simulates a typical config file with restrictive filters set.""" - return SearchConfig.from_dict({ - "categories": ["rental_apartment", "rental_house"], - "city_level_1": None, - "min_lat": 55.63, - "min_lng": 12.48, - "max_lat": 55.73, - "max_lng": 12.80, - "rooms_min": 3, - "max_rent": 17000, - "min_rental_period": 12, - "order": "DEFAULT", - "max_pages": 10, - }) + return SearchConfig.from_dict( + { + "categories": ["rental_apartment", "rental_house"], + "city_level_1": None, + "min_lat": 55.63, + "min_lng": 12.48, + "max_lat": 55.73, + "max_lng": 12.80, + "rooms_min": 3, + "max_rent": 17000, + "min_rental_period": 12, + "order": "DEFAULT", + "max_pages": 10, + } + ) # -- _build_search_config: no filters = saved search ------------------------ -class TestBuildSearchConfigSavedSearch: +class TestBuildSearchConfigSavedSearch: def test_no_filters_returns_full_config(self, base_config: SearchConfig) -> None: result = _build_search_config(base_config) assert result.categories == ["rental_apartment", "rental_house"] @@ -49,9 +61,13 @@ def test_no_filters_returns_same_object(self, base_config: SearchConfig) -> None assert result is base_config def test_no_filters_preserves_boolean_filters(self) -> None: - config = SearchConfig.from_dict({ - "pet_friendly": True, "balcony": True, "max_rent": 15000, - }) + config = SearchConfig.from_dict( + { + "pet_friendly": True, + "balcony": True, + "max_rent": 15000, + } + ) result = _build_search_config(config) assert result.pet_friendly is True assert result.balcony is True @@ -60,8 +76,8 @@ def test_no_filters_preserves_boolean_filters(self) -> None: # -- _build_search_config: any filter = clean slate -------------------------- -class TestBuildSearchConfigAdHoc: +class TestBuildSearchConfigAdHoc: def test_any_filter_strips_restrictive(self, base_config: SearchConfig) -> None: result = _build_search_config(base_config, min_size_m2=200) assert result.min_size_m2 == 200 @@ -78,17 +94,30 @@ def test_structural_settings_inherited(self, base_config: SearchConfig) -> None: assert result.max_pages == 10 def test_all_restrictive_filters_stripped(self) -> None: - rich_config = SearchConfig.from_dict({ - "rooms_min": 2, "rooms_max": 5, "max_rent": 20000, - "min_size_m2": 60, "min_rental_period": 12, - "max_available_from": "2026-08-01", - "pet_friendly": True, "balcony": True, "furnished": True, - "parking": True, "elevator": True, "shareable": True, - "student_only": True, "senior_friendly": True, - "social_housing": True, "newbuild": True, - "electric_charging_station": True, "dishwasher": True, - "washing_machine": True, "dryer": True, - }) + rich_config = SearchConfig.from_dict( + { + "rooms_min": 2, + "rooms_max": 5, + "max_rent": 20000, + "min_size_m2": 60, + "min_rental_period": 12, + "max_available_from": "2026-08-01", + "pet_friendly": True, + "balcony": True, + "furnished": True, + "parking": True, + "elevator": True, + "shareable": True, + "student_only": True, + "senior_friendly": True, + "social_housing": True, + "newbuild": True, + "electric_charging_station": True, + "dishwasher": True, + "washing_machine": True, + "dryer": True, + } + ) result = _build_search_config(rich_config, rooms_min=4) for field_name in _RESTRICTIVE_FILTERS: if field_name == "rooms_min": @@ -99,7 +128,9 @@ def test_all_restrictive_filters_stripped(self) -> None: def test_explicit_filter_overrides_applied(self, base_config: SearchConfig) -> None: result = _build_search_config( base_config, - rooms_min=2, max_rent=15000, min_size_m2=80, + rooms_min=2, + max_rent=15000, + min_size_m2=80, ) assert result.rooms_min == 2 assert result.max_rent == 15000 @@ -117,7 +148,10 @@ def test_bbox_override_clears_city(self) -> None: city_config = SearchConfig.from_dict({"city_level_1": ["københavn"]}) result = _build_search_config( city_config, - min_lat=55.6, min_lng=12.4, max_lat=55.7, max_lng=12.6, + min_lat=55.6, + min_lng=12.4, + max_lat=55.7, + max_lng=12.6, ) assert result.min_lat == 55.6 assert result.city_level_1 is None @@ -133,7 +167,9 @@ def test_max_pages_override(self, base_config: SearchConfig) -> None: def test_boolean_filters_pass_through(self, base_config: SearchConfig) -> None: result = _build_search_config( base_config, - pet_friendly=True, dishwasher=True, electric_charging_station=True, + pet_friendly=True, + dishwasher=True, + electric_charging_station=True, ) assert result.pet_friendly is True assert result.dishwasher is True @@ -152,8 +188,8 @@ def test_max_pages_alone_triggers_ad_hoc(self, base_config: SearchConfig) -> Non # -- to_api_body mapping ---------------------------------------------------- -class TestToApiBody: +class TestToApiBody: def test_minimal_config_has_categories_and_order(self) -> None: config = SearchConfig.from_dict({}) body = config.to_api_body() @@ -198,21 +234,34 @@ def test_max_available_from(self) -> None: assert body["max_available_from"] == "2026-08-01" def test_bbox_coordinates(self) -> None: - config = SearchConfig.from_dict({ - "min_lat": 55.6, "min_lng": 12.4, "max_lat": 55.7, "max_lng": 12.6, - }) + config = SearchConfig.from_dict( + { + "min_lat": 55.6, + "min_lng": 12.4, + "max_lat": 55.7, + "max_lng": 12.6, + } + ) body = config.to_api_body() assert body["min_lat"] == 55.6 assert body["max_lng"] == 12.6 def test_boolean_filters_included_when_set(self) -> None: all_bools = { - "pet_friendly": True, "balcony": True, "furnished": True, - "parking": True, "elevator": True, "shareable": True, - "student_only": True, "senior_friendly": True, - "social_housing": True, "newbuild": True, - "electric_charging_station": True, "dishwasher": True, - "washing_machine": True, "dryer": True, + "pet_friendly": True, + "balcony": True, + "furnished": True, + "parking": True, + "elevator": True, + "shareable": True, + "student_only": True, + "senior_friendly": True, + "social_housing": True, + "newbuild": True, + "electric_charging_station": True, + "dishwasher": True, + "washing_machine": True, + "dryer": True, } config = SearchConfig.from_dict(all_bools) body = config.to_api_body() @@ -222,8 +271,16 @@ def test_boolean_filters_included_when_set(self) -> None: def test_boolean_filters_excluded_when_none(self) -> None: config = SearchConfig.from_dict({}) body = config.to_api_body() - for key in ["pet_friendly", "balcony", "social_housing", "newbuild", - "electric_charging_station", "dishwasher", "washing_machine", "dryer"]: + for key in [ + "pet_friendly", + "balcony", + "social_housing", + "newbuild", + "electric_charging_station", + "dishwasher", + "washing_machine", + "dryer", + ]: assert key not in body, f"{key} should not be in API body when None" def test_none_filters_excluded_from_body(self) -> None: @@ -238,8 +295,8 @@ def test_none_filters_excluded_from_body(self) -> None: # -- SeenTracker: re-listing detection -------------------------------------- -class TestSeenTrackerRelisting: +class TestSeenTrackerRelisting: @pytest.fixture def tracker(self, tmp_path: Path) -> SeenTracker: return SeenTracker(tmp_path / "seen.json") @@ -296,8 +353,8 @@ def test_reset_clears_all(self, tracker: SeenTracker) -> None: # -- Date comparison across timezone formats --------------------------------- -class TestDateComparisonTimezones: +class TestDateComparisonTimezones: @pytest.fixture def tracker(self, tmp_path: Path) -> SeenTracker: return SeenTracker(tmp_path / "seen.json") @@ -325,8 +382,8 @@ def test_malformed_date_does_not_crash(self, tracker: SeenTracker) -> None: # -- Atomic writes ----------------------------------------------------------- -class TestAtomicWrites: +class TestAtomicWrites: def test_save_creates_file_atomically(self, tmp_path: Path) -> None: path = tmp_path / "seen.json" tracker = SeenTracker(path) @@ -352,8 +409,8 @@ def test_file_survives_reload(self, tmp_path: Path) -> None: # -- Unknown config keys warning --------------------------------------------- -class TestUnknownConfigKeys: +class TestUnknownConfigKeys: def test_unknown_key_logged(self, caplog: pytest.LogCaptureFixture) -> None: with caplog.at_level(logging.WARNING, logger="boligwatch"): SearchConfig.from_dict({"max_rnet": 15000, "rooms_min": 2}) @@ -372,8 +429,8 @@ def test_unknown_keys_still_ignored(self) -> None: # -- max_pages ceiling ------------------------------------------------------- -class TestMaxPagesCeiling: +class TestMaxPagesCeiling: def test_excessive_max_pages_clamped(self) -> None: config = SearchConfig.from_dict({"max_pages": 1000}) assert config.max_pages == MAX_PAGES_CEILING @@ -394,8 +451,8 @@ def test_build_search_config_clamps_max_pages(self) -> None: # -- Missing CLI flags ------------------------------------------------------- -class TestParkingElevatorFlags: +class TestParkingElevatorFlags: def test_parking_in_api_body(self) -> None: config = SearchConfig.from_dict({"parking": True}) body = config.to_api_body() @@ -411,3 +468,62 @@ def test_parking_elevator_through_build(self) -> None: result = _build_search_config(base, parking=True, elevator=True) assert result.parking is True assert result.elevator is True + + +# -- Backoff helper ----------------------------------------------------------- + + +class TestBackoffDelay: + def test_delay_increases_with_attempt(self) -> None: + d0 = _backoff_delay(0) + d2 = _backoff_delay(2) + assert d2 > d0 + + def test_delay_has_jitter(self) -> None: + delays = {_backoff_delay(1) for _ in range(20)} + assert len(delays) > 1 + + +# -- API request dispatch ------------------------------------------------------ + + +class TestApiRequestDispatch: + def test_delegates_to_cffi_when_available(self) -> None: + with ( + patch("boligwatch._HAS_CURL_CFFI", True), + patch("boligwatch._api_request_cffi", return_value={"results": []}) as mock, + ): + result = _api_request("https://example.com", b"{}") + mock.assert_called_once_with("https://example.com", b"{}") + assert result == {"results": []} + + def test_delegates_to_urllib_when_cffi_unavailable(self) -> None: + with ( + patch("boligwatch._HAS_CURL_CFFI", False), + patch("boligwatch._api_request_urllib", return_value={"results": []}) as mock, + ): + result = _api_request("https://example.com", b"{}") + mock.assert_called_once_with("https://example.com", b"{}") + assert result == {"results": []} + + +# -- urllib backend retries 403 ------------------------------------------------ + + +class TestUrllibRetries403: + def test_403_is_retried(self) -> None: + import urllib.error + + effects = [ + urllib.error.HTTPError("url", 403, "Forbidden", {}, None), + urllib.error.HTTPError("url", 403, "Forbidden", {}, None), + MagicMock( + read=MagicMock(return_value=b'{"results": []}'), + __enter__=MagicMock(), + __exit__=MagicMock(return_value=False), + ), + ] + effects[2].__enter__.return_value = effects[2] + with patch("urllib.request.urlopen", side_effect=effects), patch("time.sleep"): + result = _api_request_urllib("https://example.com", b"{}") + assert result == {"results": []}