From dd6d78b48b4a5ce54d60c9e36adb26714a83e733 Mon Sep 17 00:00:00 2001 From: Nitzan Pomerantz <9297302+nitzpo@users.noreply.github.com> Date: Sun, 13 Jul 2025 01:03:15 +0300 Subject: [PATCH] Refactor govmap --- nadlan_mcp/__init__.py | 2 +- nadlan_mcp/govmap.py | 333 +++++++++++++++++++++++++++ nadlan_mcp/main.py | 335 +--------------------------- nadlan_mcp/simple_fastmcp_server.py | 2 +- simple_mcp_server.py | 2 +- tests/test_govmap_client.py | 2 +- 6 files changed, 338 insertions(+), 338 deletions(-) create mode 100644 nadlan_mcp/govmap.py diff --git a/nadlan_mcp/__init__.py b/nadlan_mcp/__init__.py index bb83ab2..cd1ebdc 100644 --- a/nadlan_mcp/__init__.py +++ b/nadlan_mcp/__init__.py @@ -5,7 +5,7 @@ A Python-based Mission Control Program to interact with the Israeli government's public real estate data API (Govmap). """ -from .main import GovmapClient +from .govmap import GovmapClient __version__ = "1.0.0" __all__ = ["GovmapClient"] \ No newline at end of file diff --git a/nadlan_mcp/govmap.py b/nadlan_mcp/govmap.py new file mode 100644 index 0000000..dd95b15 --- /dev/null +++ b/nadlan_mcp/govmap.py @@ -0,0 +1,333 @@ +import requests +import logging + +import json +from datetime import datetime, timedelta +from typing import Any, Dict, List, Tuple, Optional + +logger = logging.getLogger(__name__) + + +class GovmapClient: + """ + A client for interacting with the Israeli government's Govmap API. + + This class provides methods to search for properties, find block/parcel information, + and retrieve real estate deal data. + """ + + def __init__(self, base_url: str = "https://www.govmap.gov.il/api/"): + """ + Initialize the GovmapClient. + + Args: + base_url: The base URL for the Govmap API + """ + self.base_url = base_url.rstrip('/') + self.session = requests.Session() + self.session.headers.update({ + 'Content-Type': 'application/json', + 'User-Agent': 'NadlanMCP/1.0.0' + }) + + def autocomplete_address(self, search_text: str) -> Dict[str, Any]: + """ + Find the most likely match for a given address using autocomplete. + + Args: + search_text: The address to search for (e.g., "סוקולוב 38 חולון") + + Returns: + Dict containing the JSON response from the API with coordinates + + Raises: + requests.RequestException: If the API request fails + ValueError: If the response is invalid + """ + url = f"{self.base_url}/search-service/autocomplete" + + payload = { + "searchText": search_text, + "language": "he", + "isAccurate": False, + "maxResults": 10 + } + + try: + logger.info(f"Searching for address: {search_text}") + response = self.session.post(url, json=payload) + response.raise_for_status() + + data = response.json() + if not data or 'results' not in data: + raise ValueError("Invalid response format from autocomplete API") + + return data + + except requests.RequestException as e: + logger.error(f"Error calling autocomplete API: {e}") + raise + except json.JSONDecodeError as e: + logger.error(f"Error parsing JSON response: {e}") + raise ValueError("Invalid JSON response from API") + + def get_gush_helka(self, point: Tuple[float, float]) -> Dict[str, Any]: + """ + Get Gush (Block) and Helka (Parcel) information for a coordinate point. + + Args: + point: A tuple of (longitude, latitude) + + Returns: + Dict containing the JSON response with block and parcel data + + Raises: + requests.RequestException: If the API request fails + ValueError: If the response is invalid + """ + url = f"{self.base_url}/layers-catalog/entitiesByPoint" + + payload = { + "point": list(point), + "layers": [{"layerId": "16"}], + "tolerance": 0 + } + + try: + logger.info(f"Getting Gush/Helka for point: {point}") + response = self.session.post(url, json=payload) + response.raise_for_status() + + data = response.json() + return data + + except requests.RequestException as e: + logger.error(f"Error calling entitiesByPoint API: {e}") + raise + except json.JSONDecodeError as e: + logger.error(f"Error parsing JSON response: {e}") + raise ValueError("Invalid JSON response from API") + + def get_deals_by_radius(self, point: Tuple[float, float], radius: int = 50) -> List[Dict[str, Any]]: + """ + Find real estate deals within a specified radius of a point. + + Args: + point: A tuple of (longitude, latitude) + radius: The search radius in meters (default: 50) + + Returns: + List of deals found within the radius + + Raises: + requests.RequestException: If the API request fails + """ + url = f"{self.base_url}/real-estate/deals/{point[0]},{point[1]}/{radius}" + + try: + logger.info(f"Getting deals by radius for point: {point}, radius: {radius}m") + response = self.session.get(url) + response.raise_for_status() + + data = response.json() + return data if isinstance(data, list) else [] + + except requests.RequestException as e: + logger.error(f"Error calling deals by radius API: {e}") + raise + except json.JSONDecodeError as e: + logger.error(f"Error parsing JSON response: {e}") + return [] + + def get_street_deals(self, polygon_id: str, limit: int = 10, + start_date: Optional[str] = None, end_date: Optional[str] = None) -> List[Dict[str, Any]]: + """ + Retrieve detailed information about deals on a specific street. + + Args: + polygon_id: The ID of the lot's polygon + limit: Maximum number of deals to return (default: 10) + start_date: Start date for search in 'YYYY-MM' format + end_date: End date for search in 'YYYY-MM' format + + Returns: + List of detailed deal information for the street + + Raises: + requests.RequestException: If the API request fails + """ + url = f"{self.base_url}/real-estate/street-deals/{polygon_id}" + + params: Dict[str, Any] = {"limit": limit} + if start_date: + params["startDate"] = start_date + if end_date: + params["endDate"] = end_date + + try: + logger.info(f"Getting street deals for polygon: {polygon_id}") + response = self.session.get(url, params=params) + response.raise_for_status() + + data = response.json() + # API returns {data: [...], totalCount: ..., limit: ..., offset: ...} + if isinstance(data, dict) and 'data' in data: + return data['data'] if isinstance(data['data'], list) else [] + return data if isinstance(data, list) else [] + + except requests.RequestException as e: + logger.error(f"Error calling street deals API: {e}") + raise + except json.JSONDecodeError as e: + logger.error(f"Error parsing JSON response: {e}") + return [] + + def get_neighborhood_deals(self, polygon_id: str, limit: int = 10, + start_date: Optional[str] = None, end_date: Optional[str] = None) -> List[Dict[str, Any]]: + """ + Retrieve deals within the same neighborhood as the given polygon_id. + + Args: + polygon_id: The ID of the lot's polygon + limit: Maximum number of deals to return (default: 10) + start_date: Start date for search in 'YYYY-MM' format + end_date: End date for search in 'YYYY-MM' format + + Returns: + List of deals in the neighborhood + + Raises: + requests.RequestException: If the API request fails + """ + url = f"{self.base_url}/real-estate/neighborhood-deals/{polygon_id}" + + params: Dict[str, Any] = {"limit": limit} + if start_date: + params["startDate"] = start_date + if end_date: + params["endDate"] = end_date + + try: + logger.info(f"Getting neighborhood deals for polygon: {polygon_id}") + response = self.session.get(url, params=params) + response.raise_for_status() + + data = response.json() + # API returns {data: [...], totalCount: ..., limit: ..., offset: ...} + if isinstance(data, dict) and 'data' in data: + return data['data'] if isinstance(data['data'], list) else [] + return data if isinstance(data, list) else [] + + except requests.RequestException as e: + logger.error(f"Error calling neighborhood deals API: {e}") + raise + except json.JSONDecodeError as e: + logger.error(f"Error parsing JSON response: {e}") + return [] + + def find_recent_deals_for_address(self, address: str, years_back: int = 2) -> List[Dict[str, Any]]: + """ + Find all relevant real estate deals for a given address from the last few years. + + This is the main use case function that ties everything together. + + Args: + address: The address to search for + years_back: How many years back to search (default: 2) + + Returns: + List of deals found for the address area + + Raises: + ValueError: If address cannot be found or processed + requests.RequestException: If API requests fail + """ + try: + # Step 1: Get coordinates for the address + logger.info(f"Starting search for address: {address}") + autocomplete_result = self.autocomplete_address(address) + + if not autocomplete_result.get('results'): + raise ValueError(f"No results found for address: {address}") + + # Get the best match (first result) + best_match = autocomplete_result['results'][0] + if 'shape' not in best_match: + raise ValueError("No coordinates found in autocomplete result") + + # Parse coordinates from WKT POINT string + # Format: "POINT(longitude latitude)" + shape_str = best_match['shape'] + if not shape_str.startswith('POINT('): + raise ValueError("Invalid coordinate format in autocomplete result") + + # Extract coordinates from "POINT(x y)" + coords_str = shape_str[6:-1] # Remove "POINT(" and ")" + coords = coords_str.split() + if len(coords) != 2: + raise ValueError("Invalid coordinate format in autocomplete result") + + point = (float(coords[0]), float(coords[1])) + logger.info(f"Found coordinates: {point}") + + # Step 2: Get deals by radius to find polygon IDs + nearby_deals = self.get_deals_by_radius(point, radius=30) # Slightly larger radius + + # Extract unique polygon IDs + polygon_ids = set() + for deal in nearby_deals: + if 'polygon_id' in deal: + polygon_ids.add(str(deal['polygon_id'])) + + logger.info(f"Found {len(polygon_ids)} unique polygon IDs") + + # Step 3: Calculate date range + end_date = datetime.now() + start_date = end_date - timedelta(days=years_back * 365) + start_date_str = start_date.strftime('%Y-%m') + end_date_str = end_date.strftime('%Y-%m') + + # Step 4: Get street and neighborhood deals for each polygon + all_deals = [] + seen_deals = set() # For deduplication + + for polygon_id in polygon_ids: + try: + # Get street deals + street_deals = self.get_street_deals( + polygon_id, limit=50, + start_date=start_date_str, end_date=end_date_str + ) + + # Get neighborhood deals + neighborhood_deals = self.get_neighborhood_deals( + polygon_id, limit=50, + start_date=start_date_str, end_date=end_date_str + ) + + # Combine deals + combined_deals = street_deals + neighborhood_deals + + # Add to results with deduplication + for deal in combined_deals: + # Create a unique identifier for the deal + deal_id = f"{deal.get('dealId', '')}{deal.get('address', '')}{deal.get('dealDate', '')}" + + if deal_id not in seen_deals: + seen_deals.add(deal_id) + deal['source_polygon_id'] = polygon_id # Add source for reference + all_deals.append(deal) + + except Exception as e: + logger.warning(f"Error processing polygon {polygon_id}: {e}") + continue + + # Step 5: Sort by date (newest first) + all_deals.sort(key=lambda x: x.get('dealDate', ''), reverse=True) + + logger.info(f"Found {len(all_deals)} total deals for address: {address}") + return all_deals + + except Exception as e: + logger.error(f"Error in find_recent_deals_for_address: {e}") + raise \ No newline at end of file diff --git a/nadlan_mcp/main.py b/nadlan_mcp/main.py index 8fffc8f..18a461e 100644 --- a/nadlan_mcp/main.py +++ b/nadlan_mcp/main.py @@ -5,340 +5,7 @@ This module provides the GovmapClient class for interacting with the Israeli government's public real estate data API (Govmap). """ -import requests -from typing import Dict, List, Tuple, Optional, Any -from datetime import datetime, timedelta -import json -import logging - -# Configure logging -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) - - -class GovmapClient: - """ - A client for interacting with the Israeli government's Govmap API. - - This class provides methods to search for properties, find block/parcel information, - and retrieve real estate deal data. - """ - - def __init__(self, base_url: str = "https://www.govmap.gov.il/api/"): - """ - Initialize the GovmapClient. - - Args: - base_url: The base URL for the Govmap API - """ - self.base_url = base_url.rstrip('/') - self.session = requests.Session() - self.session.headers.update({ - 'Content-Type': 'application/json', - 'User-Agent': 'NadlanMCP/1.0.0' - }) - - def autocomplete_address(self, search_text: str) -> Dict[str, Any]: - """ - Find the most likely match for a given address using autocomplete. - - Args: - search_text: The address to search for (e.g., "סוקולוב 38 חולון") - - Returns: - Dict containing the JSON response from the API with coordinates - - Raises: - requests.RequestException: If the API request fails - ValueError: If the response is invalid - """ - url = f"{self.base_url}/search-service/autocomplete" - - payload = { - "searchText": search_text, - "language": "he", - "isAccurate": False, - "maxResults": 10 - } - - try: - logger.info(f"Searching for address: {search_text}") - response = self.session.post(url, json=payload) - response.raise_for_status() - - data = response.json() - if not data or 'results' not in data: - raise ValueError("Invalid response format from autocomplete API") - - return data - - except requests.RequestException as e: - logger.error(f"Error calling autocomplete API: {e}") - raise - except json.JSONDecodeError as e: - logger.error(f"Error parsing JSON response: {e}") - raise ValueError("Invalid JSON response from API") - - def get_gush_helka(self, point: Tuple[float, float]) -> Dict[str, Any]: - """ - Get Gush (Block) and Helka (Parcel) information for a coordinate point. - - Args: - point: A tuple of (longitude, latitude) - - Returns: - Dict containing the JSON response with block and parcel data - - Raises: - requests.RequestException: If the API request fails - ValueError: If the response is invalid - """ - url = f"{self.base_url}/layers-catalog/entitiesByPoint" - - payload = { - "point": list(point), - "layers": [{"layerId": "16"}], - "tolerance": 0 - } - - try: - logger.info(f"Getting Gush/Helka for point: {point}") - response = self.session.post(url, json=payload) - response.raise_for_status() - - data = response.json() - return data - - except requests.RequestException as e: - logger.error(f"Error calling entitiesByPoint API: {e}") - raise - except json.JSONDecodeError as e: - logger.error(f"Error parsing JSON response: {e}") - raise ValueError("Invalid JSON response from API") - - def get_deals_by_radius(self, point: Tuple[float, float], radius: int = 50) -> List[Dict[str, Any]]: - """ - Find real estate deals within a specified radius of a point. - - Args: - point: A tuple of (longitude, latitude) - radius: The search radius in meters (default: 50) - - Returns: - List of deals found within the radius - - Raises: - requests.RequestException: If the API request fails - """ - url = f"{self.base_url}/real-estate/deals/{point[0]},{point[1]}/{radius}" - - try: - logger.info(f"Getting deals by radius for point: {point}, radius: {radius}m") - response = self.session.get(url) - response.raise_for_status() - - data = response.json() - return data if isinstance(data, list) else [] - - except requests.RequestException as e: - logger.error(f"Error calling deals by radius API: {e}") - raise - except json.JSONDecodeError as e: - logger.error(f"Error parsing JSON response: {e}") - return [] - - def get_street_deals(self, polygon_id: str, limit: int = 10, - start_date: Optional[str] = None, end_date: Optional[str] = None) -> List[Dict[str, Any]]: - """ - Retrieve detailed information about deals on a specific street. - - Args: - polygon_id: The ID of the lot's polygon - limit: Maximum number of deals to return (default: 10) - start_date: Start date for search in 'YYYY-MM' format - end_date: End date for search in 'YYYY-MM' format - - Returns: - List of detailed deal information for the street - - Raises: - requests.RequestException: If the API request fails - """ - url = f"{self.base_url}/real-estate/street-deals/{polygon_id}" - - params: Dict[str, Any] = {"limit": limit} - if start_date: - params["startDate"] = start_date - if end_date: - params["endDate"] = end_date - - try: - logger.info(f"Getting street deals for polygon: {polygon_id}") - response = self.session.get(url, params=params) - response.raise_for_status() - - data = response.json() - # API returns {data: [...], totalCount: ..., limit: ..., offset: ...} - if isinstance(data, dict) and 'data' in data: - return data['data'] if isinstance(data['data'], list) else [] - return data if isinstance(data, list) else [] - - except requests.RequestException as e: - logger.error(f"Error calling street deals API: {e}") - raise - except json.JSONDecodeError as e: - logger.error(f"Error parsing JSON response: {e}") - return [] - - def get_neighborhood_deals(self, polygon_id: str, limit: int = 10, - start_date: Optional[str] = None, end_date: Optional[str] = None) -> List[Dict[str, Any]]: - """ - Retrieve deals within the same neighborhood as the given polygon_id. - - Args: - polygon_id: The ID of the lot's polygon - limit: Maximum number of deals to return (default: 10) - start_date: Start date for search in 'YYYY-MM' format - end_date: End date for search in 'YYYY-MM' format - - Returns: - List of deals in the neighborhood - - Raises: - requests.RequestException: If the API request fails - """ - url = f"{self.base_url}/real-estate/neighborhood-deals/{polygon_id}" - - params: Dict[str, Any] = {"limit": limit} - if start_date: - params["startDate"] = start_date - if end_date: - params["endDate"] = end_date - - try: - logger.info(f"Getting neighborhood deals for polygon: {polygon_id}") - response = self.session.get(url, params=params) - response.raise_for_status() - - data = response.json() - # API returns {data: [...], totalCount: ..., limit: ..., offset: ...} - if isinstance(data, dict) and 'data' in data: - return data['data'] if isinstance(data['data'], list) else [] - return data if isinstance(data, list) else [] - - except requests.RequestException as e: - logger.error(f"Error calling neighborhood deals API: {e}") - raise - except json.JSONDecodeError as e: - logger.error(f"Error parsing JSON response: {e}") - return [] - - def find_recent_deals_for_address(self, address: str, years_back: int = 2) -> List[Dict[str, Any]]: - """ - Find all relevant real estate deals for a given address from the last few years. - - This is the main use case function that ties everything together. - - Args: - address: The address to search for - years_back: How many years back to search (default: 2) - - Returns: - List of deals found for the address area - - Raises: - ValueError: If address cannot be found or processed - requests.RequestException: If API requests fail - """ - try: - # Step 1: Get coordinates for the address - logger.info(f"Starting search for address: {address}") - autocomplete_result = self.autocomplete_address(address) - - if not autocomplete_result.get('results'): - raise ValueError(f"No results found for address: {address}") - - # Get the best match (first result) - best_match = autocomplete_result['results'][0] - if 'shape' not in best_match: - raise ValueError("No coordinates found in autocomplete result") - - # Parse coordinates from WKT POINT string - # Format: "POINT(longitude latitude)" - shape_str = best_match['shape'] - if not shape_str.startswith('POINT('): - raise ValueError("Invalid coordinate format in autocomplete result") - - # Extract coordinates from "POINT(x y)" - coords_str = shape_str[6:-1] # Remove "POINT(" and ")" - coords = coords_str.split() - if len(coords) != 2: - raise ValueError("Invalid coordinate format in autocomplete result") - - point = (float(coords[0]), float(coords[1])) - logger.info(f"Found coordinates: {point}") - - # Step 2: Get deals by radius to find polygon IDs - nearby_deals = self.get_deals_by_radius(point, radius=30) # Slightly larger radius - - # Extract unique polygon IDs - polygon_ids = set() - for deal in nearby_deals: - if 'polygon_id' in deal: - polygon_ids.add(str(deal['polygon_id'])) - - logger.info(f"Found {len(polygon_ids)} unique polygon IDs") - - # Step 3: Calculate date range - end_date = datetime.now() - start_date = end_date - timedelta(days=years_back * 365) - start_date_str = start_date.strftime('%Y-%m') - end_date_str = end_date.strftime('%Y-%m') - - # Step 4: Get street and neighborhood deals for each polygon - all_deals = [] - seen_deals = set() # For deduplication - - for polygon_id in polygon_ids: - try: - # Get street deals - street_deals = self.get_street_deals( - polygon_id, limit=50, - start_date=start_date_str, end_date=end_date_str - ) - - # Get neighborhood deals - neighborhood_deals = self.get_neighborhood_deals( - polygon_id, limit=50, - start_date=start_date_str, end_date=end_date_str - ) - - # Combine deals - combined_deals = street_deals + neighborhood_deals - - # Add to results with deduplication - for deal in combined_deals: - # Create a unique identifier for the deal - deal_id = f"{deal.get('dealId', '')}{deal.get('address', '')}{deal.get('dealDate', '')}" - - if deal_id not in seen_deals: - seen_deals.add(deal_id) - deal['source_polygon_id'] = polygon_id # Add source for reference - all_deals.append(deal) - - except Exception as e: - logger.warning(f"Error processing polygon {polygon_id}: {e}") - continue - - # Step 5: Sort by date (newest first) - all_deals.sort(key=lambda x: x.get('dealDate', ''), reverse=True) - - logger.info(f"Found {len(all_deals)} total deals for address: {address}") - return all_deals - - except Exception as e: - logger.error(f"Error in find_recent_deals_for_address: {e}") - raise +from nadlan_mcp.govmap import GovmapClient # Example usage functions diff --git a/nadlan_mcp/simple_fastmcp_server.py b/nadlan_mcp/simple_fastmcp_server.py index a9759fe..60a7e72 100644 --- a/nadlan_mcp/simple_fastmcp_server.py +++ b/nadlan_mcp/simple_fastmcp_server.py @@ -10,7 +10,7 @@ import json import logging from typing import List, Dict from mcp.server.fastmcp import FastMCP -from .main import GovmapClient +from .govmap import GovmapClient # Configure logging logging.basicConfig(level=logging.INFO) diff --git a/simple_mcp_server.py b/simple_mcp_server.py index 7db552e..1ec9f3a 100644 --- a/simple_mcp_server.py +++ b/simple_mcp_server.py @@ -14,7 +14,7 @@ from typing import Any, Dict, List # Simple MCP server implementation class SimpleMCPServer: def __init__(self): - from nadlan_mcp.main import GovmapClient + from nadlan_mcp.govmap import GovmapClient self.client = GovmapClient() self.tools = { "find_recent_deals_for_address": { diff --git a/tests/test_govmap_client.py b/tests/test_govmap_client.py index e029fcb..6a2d86c 100644 --- a/tests/test_govmap_client.py +++ b/tests/test_govmap_client.py @@ -5,7 +5,7 @@ Tests for the GovmapClient class. import pytest import requests from unittest.mock import Mock, patch -from nadlan_mcp.main import GovmapClient +from nadlan_mcp.govmap import GovmapClient class TestGovmapClient: