diff --git a/nadlan_mcp/govmap/filters.py b/nadlan_mcp/govmap/filters.py new file mode 100644 index 0000000..d94cf54 --- /dev/null +++ b/nadlan_mcp/govmap/filters.py @@ -0,0 +1,135 @@ +""" +Deal filtering functions. + +This module provides composable functions for filtering real estate deal data. +""" + +from typing import Any, Dict, List, Optional + +from .utils import extract_floor_number + + +def filter_deals_by_criteria( + deals: List[Dict[str, Any]], + property_type: Optional[str] = None, + min_rooms: Optional[float] = None, + max_rooms: Optional[float] = None, + min_price: Optional[float] = None, + max_price: Optional[float] = None, + min_area: Optional[float] = None, + max_area: Optional[float] = None, + min_floor: Optional[int] = None, + max_floor: Optional[int] = None, +) -> List[Dict[str, Any]]: + """ + Filter deals by various criteria. + + Args: + deals: List of deal dictionaries to filter + property_type: Property type to filter by (Hebrew description) + min_rooms: Minimum number of rooms + max_rooms: Maximum number of rooms + min_price: Minimum deal amount + max_price: Maximum deal amount + min_area: Minimum asset area (square meters) + max_area: Maximum asset area (square meters) + min_floor: Minimum floor number + max_floor: Maximum floor number + + Returns: + Filtered list of deals + + Raises: + ValueError: If filter criteria are invalid + """ + if not isinstance(deals, list): + raise ValueError("deals must be a list") + + # Validate numeric ranges + if min_rooms is not None and max_rooms is not None and min_rooms > max_rooms: + raise ValueError("min_rooms cannot be greater than max_rooms") + if min_price is not None and max_price is not None and min_price > max_price: + raise ValueError("min_price cannot be greater than max_price") + if min_area is not None and max_area is not None and min_area > max_area: + raise ValueError("min_area cannot be greater than max_area") + if min_floor is not None and max_floor is not None and min_floor > max_floor: + raise ValueError("min_floor cannot be greater than max_floor") + + filtered_deals = [] + + for deal in deals: + # Property type filter + if property_type is not None: + deal_type = deal.get( + "propertyTypeDescription", deal.get("assetTypeHeb", "") + ) + # Skip deals with missing property type data when filter is active + if not deal_type: + continue + + # Normalize both strings for flexible matching + property_type_normalized = property_type.lower().strip() + deal_type_normalized = deal_type.lower().strip() + + # Check if the filter term appears in the deal type + # This allows "דירה" to match "דירת גג", "דירה בבניין", etc. + if property_type_normalized not in deal_type_normalized: + continue + + # Room count filter + if min_rooms is not None or max_rooms is not None: + rooms = deal.get("assetRoomNum") + if rooms is None: + continue # Skip deals with missing room data when filter is active + try: + rooms = float(rooms) + if min_rooms is not None and rooms < min_rooms: + continue + if max_rooms is not None and rooms > max_rooms: + continue + except (TypeError, ValueError): + continue # Skip deals with invalid room data when filter is active + + # Price filter + if min_price is not None or max_price is not None: + price = deal.get("dealAmount") + if price is None: + continue # Skip deals with missing price data when filter is active + try: + price = float(price) + if min_price is not None and price < min_price: + continue + if max_price is not None and price > max_price: + continue + except (TypeError, ValueError): + continue # Skip deals with invalid price data when filter is active + + # Area filter + if min_area is not None or max_area is not None: + area = deal.get("assetArea") + if area is None: + continue # Skip deals with missing area data when filter is active + try: + area = float(area) + if min_area is not None and area < min_area: + continue + if max_area is not None and area > max_area: + continue + except (TypeError, ValueError): + continue # Skip deals with invalid area data when filter is active + + # Floor filter + if min_floor is not None or max_floor is not None: + floor_str = deal.get("floorNo", "") + if floor_str and isinstance(floor_str, str): + # Try to extract floor number (handles Hebrew floor descriptions) + floor_num = extract_floor_number(floor_str) + if floor_num is not None: + if min_floor is not None and floor_num < min_floor: + continue + if max_floor is not None and floor_num > max_floor: + continue + + filtered_deals.append(deal) + + return filtered_deals diff --git a/nadlan_mcp/govmap/statistics.py b/nadlan_mcp/govmap/statistics.py new file mode 100644 index 0000000..4c29338 --- /dev/null +++ b/nadlan_mcp/govmap/statistics.py @@ -0,0 +1,124 @@ +""" +Statistical calculation functions for deal data. + +This module provides pure mathematical functions for analyzing real estate deal data. +""" + +from collections import Counter +from typing import Any, Dict, List + + +def calculate_deal_statistics(deals: List[Dict[str, Any]]) -> Dict[str, Any]: + """ + Calculate statistical aggregations on deal data. + + Args: + deals: List of deal dictionaries + + Returns: + Dictionary with statistical metrics + + Raises: + ValueError: If deals is not a valid list + """ + if not isinstance(deals, list): + raise ValueError("deals must be a list") + + if not deals: + return { + "count": 0, + "price_stats": {}, + "area_stats": {}, + "price_per_sqm_stats": {}, + "room_distribution": {}, + } + + # Extract numeric values + prices = [] + areas = [] + price_per_sqm_values = [] + rooms = [] + + for deal in deals: + price = deal.get("dealAmount") + if isinstance(price, (int, float)) and price > 0: + prices.append(price) + + area = deal.get("assetArea") + if isinstance(area, (int, float)) and area > 0: + areas.append(area) + + pps = deal.get("price_per_sqm") + if pps is None and price and area and area > 0: + pps = price / area + if isinstance(pps, (int, float)) and pps > 0: + price_per_sqm_values.append(pps) + + room_count = deal.get("assetRoomNum") + if isinstance(room_count, (int, float)): + rooms.append(room_count) + + # Calculate statistics + stats: Dict[str, Any] = {"count": len(deals)} + + # Price statistics + if prices: + sorted_prices = sorted(prices) + stats["price_stats"] = { + "mean": round(sum(prices) / len(prices), 2), + "median": sorted_prices[len(sorted_prices) // 2], + "min": min(prices), + "max": max(prices), + "p25": sorted_prices[len(sorted_prices) // 4], + "p75": sorted_prices[(3 * len(sorted_prices)) // 4], + "std_dev": round(calculate_std_dev(prices), 2) if len(prices) > 1 else 0, + "total": sum(prices), + } + + # Area statistics + if areas: + sorted_areas = sorted(areas) + stats["area_stats"] = { + "mean": round(sum(areas) / len(areas), 2), + "median": sorted_areas[len(sorted_areas) // 2], + "min": min(areas), + "max": max(areas), + "p25": sorted_areas[len(sorted_areas) // 4], + "p75": sorted_areas[(3 * len(sorted_areas)) // 4], + } + + # Price per sqm statistics + if price_per_sqm_values: + sorted_pps = sorted(price_per_sqm_values) + stats["price_per_sqm_stats"] = { + "mean": round(sum(price_per_sqm_values) / len(price_per_sqm_values), 2), + "median": round(sorted_pps[len(sorted_pps) // 2], 2), + "min": round(min(price_per_sqm_values), 2), + "max": round(max(price_per_sqm_values), 2), + "p25": round(sorted_pps[len(sorted_pps) // 4], 2), + "p75": round(sorted_pps[(3 * len(sorted_pps)) // 4], 2), + } + + # Room distribution + if rooms: + room_counts = Counter(rooms) + stats["room_distribution"] = dict(sorted(room_counts.items())) + + return stats + + +def calculate_std_dev(values: List[float]) -> float: + """ + Calculate standard deviation of a list of values. + + Args: + values: List of numeric values + + Returns: + Standard deviation + """ + if len(values) < 2: + return 0.0 + mean = sum(values) / len(values) + variance = sum((x - mean) ** 2 for x in values) / (len(values) - 1) + return variance**0.5 diff --git a/nadlan_mcp/govmap/utils.py b/nadlan_mcp/govmap/utils.py new file mode 100644 index 0000000..05e590d --- /dev/null +++ b/nadlan_mcp/govmap/utils.py @@ -0,0 +1,141 @@ +""" +Utility functions for Govmap client. + +This module provides shared helper functions with no external dependencies +(except standard library). +""" + +from typing import Tuple + + +def calculate_distance(point1: Tuple[float, float], point2: Tuple[float, float]) -> float: + """ + Calculate Euclidean distance between two points in ITM coordinates. + + ITM (Israeli Transverse Mercator) uses meters as units, so Euclidean + distance provides accurate results for distances within Israel. + + Args: + point1: (longitude, latitude) in ITM + point2: (longitude, latitude) in ITM + + Returns: + Distance in meters + """ + dx = point2[0] - point1[0] + dy = point2[1] - point1[1] + return (dx * dx + dy * dy) ** 0.5 + + +def is_same_building(search_address: str, deal_address: str) -> bool: + """ + Check if a deal is from the same building as the search address. + + Args: + search_address: The normalized search address (lowercase, stripped) + deal_address: The normalized deal address (lowercase, stripped) + + Returns: + True if likely the same building, False otherwise + """ + if not search_address or not deal_address: + return False + + # Exact match + if search_address == deal_address: + return True + + # Extract key components for comparison + def extract_address_parts(addr: str) -> tuple: + """Extract street name and number from address""" + # Remove common prefixes/suffixes and normalize + addr_clean = ( + addr.replace("רח'", "") + .replace("רחוב", "") + .replace("שד'", "") + .replace("שדרות", "") + ) + addr_clean = addr_clean.replace(" ", " ").strip() + + # Try to extract number and street name + parts = addr_clean.split() + if len(parts) >= 2: + # Look for number (could be at start or end) + for i, part in enumerate(parts): + if part.isdigit() or any(c.isdigit() for c in part): + number = part + street_parts = parts[:i] + parts[i + 1 :] + street_name = " ".join(street_parts).strip() + return (street_name, number) + + return (addr_clean, "") + + search_street, search_number = extract_address_parts(search_address) + deal_street, deal_number = extract_address_parts(deal_address) + + # Same street and same number = same building + if ( + search_street + and deal_street + and search_number + and deal_number + and search_street == deal_street + and search_number == deal_number + ): + return True + + # Check if one address is contained in the other (for different formats of same address) + if len(search_address) > 5 and len(deal_address) > 5: + if search_address in deal_address or deal_address in search_address: + return True + + return False + + +def extract_floor_number(floor_str: str) -> int | None: + """ + Extract numeric floor number from Hebrew floor description. + + Args: + floor_str: Floor description string (e.g., "שלישית", "קומה 3", "3") + + Returns: + Floor number or None if cannot be extracted + """ + if not floor_str: + return None + + # Hebrew ordinal floor names to numbers + hebrew_floors = { + "קרקע": 0, + "מרתף": -1, + "ראשונה": 1, + "שניה": 2, + "שלישית": 3, + "רביעית": 4, + "חמישית": 5, + "שישית": 6, + "שביעית": 7, + "שמינית": 8, + "תשיעית": 9, + "עשירית": 10, + } + + floor_lower = floor_str.lower().strip() + + # Check for direct match with Hebrew names + for heb, num in hebrew_floors.items(): + if heb in floor_lower: + return num + + # Try to extract number from string + import re + + numbers = re.findall(r"\d+", floor_str) + if numbers: + try: + return int(numbers[0]) + except ValueError: + pass + + return None diff --git a/nadlan_mcp/govmap/validators.py b/nadlan_mcp/govmap/validators.py new file mode 100644 index 0000000..6497373 --- /dev/null +++ b/nadlan_mcp/govmap/validators.py @@ -0,0 +1,107 @@ +""" +Input validation functions for Govmap API client. + +This module provides pure validation functions with no dependencies on other modules. +All functions are stateless and raise ValueError on validation failure. +""" + +import logging +from typing import Optional, Tuple + +logger = logging.getLogger(__name__) + + +def validate_address(address: str) -> str: + """ + Validate and sanitize address input. + + Args: + address: Address string to validate + + Returns: + Sanitized address string + + Raises: + ValueError: If address is invalid + """ + if not address or not isinstance(address, str): + raise ValueError("Address must be a non-empty string") + address = address.strip() + if not address: + raise ValueError("Address cannot be empty or whitespace only") + if len(address) > 500: + raise ValueError("Address is too long (max 500 characters)") + return address + + +def validate_coordinates(point: Tuple[float, float]) -> Tuple[float, float]: + """ + Validate coordinate input. + + Args: + point: Tuple of (longitude, latitude) in ITM projection + + Returns: + Validated coordinate tuple + + Raises: + ValueError: If coordinates are invalid + """ + if not isinstance(point, (tuple, list)) or len(point) != 2: + raise ValueError("Point must be a tuple of (longitude, latitude)") + try: + lon, lat = float(point[0]), float(point[1]) + except (TypeError, ValueError): + raise ValueError("Coordinates must be numeric values") + + # Basic validation for Israeli coordinates (ITM projection) + if not (0 < lon < 400000): # Rough bounds for Israeli ITM longitude + logger.warning(f"Longitude {lon} may be outside Israeli bounds") + if not (0 < lat < 1400000): # Rough bounds for Israeli ITM latitude + logger.warning(f"Latitude {lat} may be outside Israeli bounds") + + return (lon, lat) + + +def validate_positive_int( + value: int, name: str, max_value: Optional[int] = None +) -> int: + """ + Validate positive integer input. + + Args: + value: Value to validate + name: Name of the parameter (for error messages) + max_value: Optional maximum allowed value + + Returns: + Validated integer + + Raises: + ValueError: If value is invalid + """ + if not isinstance(value, int): + raise ValueError(f"{name} must be an integer") + if value <= 0: + raise ValueError(f"{name} must be positive") + if max_value and value > max_value: + raise ValueError(f"{name} must be <= {max_value}") + return value + + +def validate_deal_type(deal_type: int) -> int: + """ + Validate deal type parameter. + + Args: + deal_type: Deal type (1=first hand/new, 2=second hand/used) + + Returns: + Validated deal type + + Raises: + ValueError: If deal type is invalid + """ + if deal_type not in (1, 2): + raise ValueError("deal_type must be 1 (first hand) or 2 (second hand)") + return deal_type