chaim
4bc054f315
The PDF host (free-justice.openapi.gov.il) requires the same x-client-id header as the search API, so a normal browser click on the URL fails. This tool carries the auth header automatically and saves the PDF to a configurable local directory. Safety properties: - SSRF guard: only download from free-justice.openapi.gov.il and pub-justice.openapi.gov.il. - Path-traversal guard: filename is reduced to its basename; arbitrary paths are stripped. - Content-type guard: rejects 200-OK responses whose body is not a real PDF (gateway sometimes returns JSON-error 200s). - Atomic write via .tmp + rename so partial downloads never replace the cached copy. - Caches by destination path; re-downloads are no-ops unless overwrite=True. DECISIVE_APPRAISER_DOWNLOAD_DIR env var configures the output dir (default: ./downloads/decisive_appraisals). 6 new unit tests cover the happy path, caching, and all four guards. End-to-end live test confirmed a 1.4MB real PDF lands on disk with valid `%PDF-1.7` header. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
371 lines
14 KiB
Python
371 lines
14 KiB
Python
"""
|
|
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"),
|
|
}
|