""" 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 logging import time from typing import Any, Dict, List, Optional 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" # 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, )