""" Client for the Ministry of Justice "Decisive Appraiser" search API. The upstream endpoint sits behind an F5 WAF that fingerprints TLS clients and rejects standard `requests`. We use `curl_cffi` with Chrome impersonation to traverse it. A static `x-client-id` header issued to the gov.il SPA is required by the gateway; without it every request returns a generic 500 "General Error". """ from datetime import datetime import hashlib import logging import os from pathlib import Path import re import time from typing import Any, Dict, List, Optional, Union from urllib.parse import urlparse from curl_cffi import requests as cf_requests from nadlan_mcp.config import GovmapConfig, get_config from .models import ( AppraisalDecision, DecisiveAppraiserSearchResponse, ) logger = logging.getLogger(__name__) # Endpoint is public and stable; embedding it avoids forcing every operator # to set yet another env var. Override-able via config if the path moves. DEFAULT_API_URL = ( "https://pub-justice.openapi.gov.il/pub/moj/portal/rest/searchpredefinedapi/v1" "/SearchPredefinedApi/DecisiveAppraiser/SearchDecisions" ) DEFAULT_REFERER = "https://www.gov.il/he/departments/dynamiccollectors/decisive_appraisal_decisions" # Public client id assigned to the gov.il SPA and visible in any browser # DevTools session. Not a secret, but required by the gateway. DEFAULT_CLIENT_ID = "149a5bad-edde-49a6-9fb9-188bd17d4788" # PDFs are hosted on a sibling host (`free-justice.openapi.gov.il`) and # also require the same x-client-id header. SSRF guard: refuse to download # anything outside this host. ALLOWED_PDF_HOSTS = ("free-justice.openapi.gov.il", "pub-justice.openapi.gov.il") # Filters the upstream API understands. Anything else is silently dropped. _VALID_FILTER_KEYS = { "SearchText", "AppraisalHeader", "DecisiveAppraiser", "Block", "Plot", "Committee", "AppraisalType", "AppraiserType", "AppraisalVersion", "PublicityDate_from", "PublicityDate_to", "DecisionDate_from", "DecisionDate_to", } def _format_date(value: Optional[Any]) -> Optional[str]: """Coerce a date / datetime / string into the dd-MM-yyyy format the API expects.""" if value is None: return None if isinstance(value, datetime): return value.strftime("%d-%m-%Y") if hasattr(value, "strftime"): return value.strftime("%d-%m-%Y") s = str(value).strip() if not s: return None return s class DecisiveAppraiserClient: """ Search and download published decisive-appraiser decisions. Uses curl_cffi with Chrome 120 impersonation to satisfy the F5 WAF in front of pub-justice.openapi.gov.il. Honours the same retry / rate-limit knobs as `GovmapClient` so operators don't manage two configs. """ def __init__( self, config: Optional[GovmapConfig] = None, api_url: str = DEFAULT_API_URL, client_id: str = DEFAULT_CLIENT_ID, impersonate: str = "chrome120", ): self.config = config or get_config() self.api_url = api_url self.client_id = client_id self._impersonate = impersonate self._session = cf_requests.Session(impersonate=impersonate) self._session.headers.update( { "Accept": "application/json, text/plain, */*", "Accept-Language": "he-IL,he;q=0.9,en-US;q=0.8,en;q=0.7", "Content-Type": "application/json;charset=UTF-8", "Origin": "https://www.gov.il", "Referer": DEFAULT_REFERER, "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "cross-site", "x-client-id": self.client_id, } ) self._last_request_time = 0.0 def _rate_limit(self) -> None: min_interval = 1.0 / self.config.requests_per_second elapsed = time.time() - self._last_request_time if elapsed < min_interval: time.sleep(min_interval - elapsed) self._last_request_time = time.time() def _post(self, body: Dict[str, Any]) -> Dict[str, Any]: """POST `body` to the search endpoint with retries on transient failures.""" last_exc: Optional[Exception] = None for attempt in range(self.config.max_retries + 1): self._rate_limit() try: response = self._session.post( self.api_url, json=body, timeout=(self.config.connect_timeout, self.config.read_timeout), ) # 500 from this gateway typically means "you got past TLS but # the gateway rejected your shape/headers"; treat as fatal so # retries don't mask a misconfiguration. if response.status_code == 500: raise ValueError( f"DecisiveAppraiser API returned 500 — request shape or " f"x-client-id is likely wrong. Body keys sent: {list(body.keys())}" ) if response.status_code >= 500: raise cf_requests.RequestsError( f"Upstream {response.status_code}: {response.text[:200]}" ) if response.status_code != 200: raise ValueError( f"DecisiveAppraiser API returned {response.status_code}: " f"{response.text[:200]}" ) return response.json() except (cf_requests.RequestsError, OSError) as e: last_exc = e if attempt < self.config.max_retries: wait = min( self.config.retry_max_wait, self.config.retry_min_wait * (2**attempt), ) logger.warning( f"DecisiveAppraiser request failed (attempt {attempt + 1}/" f"{self.config.max_retries + 1}): {e}. Retrying in {wait}s" ) time.sleep(wait) else: raise # Unreachable, but satisfies the type checker. raise last_exc if last_exc else RuntimeError("retry loop exited unexpectedly") def search_decisions( self, block: Optional[str] = None, plot: Optional[str] = None, decisive_appraiser: Optional[str] = None, committee: Optional[str] = None, decision_date_from: Optional[Any] = None, decision_date_to: Optional[Any] = None, publicity_date_from: Optional[Any] = None, publicity_date_to: Optional[Any] = None, search_text: Optional[str] = None, appraisal_header: Optional[str] = None, skip: int = 0, ) -> DecisiveAppraiserSearchResponse: """ Search published decisive-appraiser decisions. All filters are optional — the API will return the latest 10 results when called with just `skip=0`. Page size is fixed at 10 server-side; use `skip` to paginate. Args: block: Land block (גוש) — exact string match. plot: Land plot (חלקה) — exact string match within `block`. decisive_appraiser: Appraiser name (substring/exact, Hebrew). committee: Local committee name (e.g. "תל אביב-יפו"). decision_date_from / decision_date_to: Filter by decision date. Accepts datetime, date, or "dd-MM-yyyy" string. publicity_date_from / publicity_date_to: Filter by publication date. search_text: Free-text search across the document body. appraisal_header: Search within the appraisal header text. skip: Pagination offset (multiples of 10). """ body: Dict[str, Any] = {"skip": int(skip)} if block: body["Block"] = str(block).strip() if plot: body["Plot"] = str(plot).strip() if decisive_appraiser: body["DecisiveAppraiser"] = decisive_appraiser.strip() if committee: body["Committee"] = committee.strip() if search_text: body["SearchText"] = search_text.strip() if appraisal_header: body["AppraisalHeader"] = appraisal_header.strip() df = _format_date(decision_date_from) if df: body["DecisionDate_from"] = df dt = _format_date(decision_date_to) if dt: body["DecisionDate_to"] = dt pf = _format_date(publicity_date_from) if pf: body["PublicityDate_from"] = pf pt = _format_date(publicity_date_to) if pt: body["PublicityDate_to"] = pt raw = self._post(body) # Unwrap {"Results": [{"Data": {...}}, ...]} into [decision, ...] decisions: List[AppraisalDecision] = [] for item in raw.get("Results", []) or []: data = item.get("Data") if isinstance(item, dict) else None if isinstance(data, dict): decisions.append(AppraisalDecision.model_validate(data)) return DecisiveAppraiserSearchResponse( results=decisions, total_results=int(raw.get("TotalResults") or 0), status=raw.get("Status"), message=raw.get("message"), ) def search_decisions_paged( self, max_results: int = 50, **filters: Any, ) -> DecisiveAppraiserSearchResponse: """ Page through `search_decisions` until `max_results` decisions are collected (or the upstream runs out). `filters` are forwarded to `search_decisions`; do not pass `skip`. """ if "skip" in filters: raise ValueError("search_decisions_paged manages skip itself") page_size = 10 # Server-side fixed. collected: List[AppraisalDecision] = [] total = 0 skip = 0 while len(collected) < max_results: page = self.search_decisions(skip=skip, **filters) total = page.total_results if not page.results: break collected.extend(page.results) skip += page_size if skip >= total: break return DecisiveAppraiserSearchResponse( results=collected[:max_results], total_results=total, ) def download_pdf( self, url: str, output_dir: Optional[Union[str, Path]] = None, filename: Optional[str] = None, overwrite: bool = False, ) -> Dict[str, Any]: """ Download a decisive-appraiser PDF to the local filesystem. The PDF host requires the same `x-client-id` header as the search API; a normal browser click on the URL will fail. This method carries the auth header automatically. Args: url: Direct PDF URL, as returned in `decision.documents[].file_url`. Must be on `free-justice.openapi.gov.il` or `pub-justice.openapi.gov.il` (SSRF guard). output_dir: Directory to save into. Defaults to the configured `decisive_appraiser_download_dir`. filename: Optional filename. If omitted, derived from a hash of the URL plus `.pdf`. Path components in the filename are stripped — only the basename is kept. overwrite: If False (default), aborts when the destination file already exists; returns its existing size unchanged. Returns: Dict with `path` (str), `size_bytes` (int), `from_cache` (bool), `content_type` (str). Raises ValueError on host/url issues and curl_cffi.RequestsError on network failures. """ parsed = urlparse(url) if parsed.scheme not in ("http", "https"): raise ValueError(f"Refusing non-http(s) URL: {url}") if parsed.hostname not in ALLOWED_PDF_HOSTS: raise ValueError( f"Refusing to download from untrusted host '{parsed.hostname}'. " f"Allowed: {ALLOWED_PDF_HOSTS}" ) # Resolve destination if output_dir is None: output_dir = getattr( self.config, "decisive_appraiser_download_dir", "./downloads/decisive_appraisals" ) out_dir = Path(output_dir).expanduser().resolve() out_dir.mkdir(parents=True, exist_ok=True) if filename: # Strip any path components — only basename allowed. safe_name = Path(filename).name # Sanitise a bit: keep alnum, dot, hyphen, underscore, Hebrew. safe_name = re.sub(r'[<>:"|?*\x00-\x1f]', "_", safe_name) if not safe_name: safe_name = None # type: ignore[assignment] if not filename: digest = hashlib.sha256(url.encode("utf-8")).hexdigest()[:16] safe_name = f"decisive_appraisal_{digest}.pdf" dest = out_dir / safe_name if dest.exists() and not overwrite: return { "path": str(dest), "size_bytes": dest.stat().st_size, "from_cache": True, "content_type": "application/pdf", } self._rate_limit() response = self._session.get( url, timeout=(self.config.connect_timeout, self.config.read_timeout * 4), headers={"Accept": "application/pdf,*/*"}, ) if response.status_code != 200: raise ValueError(f"PDF download returned {response.status_code}: {response.text[:200]}") content = response.content if not content.startswith(b"%PDF"): # Gateway sometimes returns JSON error with 200 — guard against it. raise ValueError( f"Response is not a PDF (first bytes: {content[:8]!r}, size={len(content)})" ) # Atomic write: write to .tmp then rename. tmp = dest.with_suffix(dest.suffix + ".tmp") tmp.write_bytes(content) os.replace(tmp, dest) return { "path": str(dest), "size_bytes": len(content), "from_cache": False, "content_type": response.headers.get("content-type", "application/pdf"), }