Phase 3: Create govmap package structure (step 1/3)
Created modular package structure: - validators.py: Input validation functions (3 functions, ~100 lines) - utils.py: Helper utilities (3 functions, ~140 lines) - filters.py: Deal filtering logic (1 main function, ~140 lines) - statistics.py: Statistical calculations (2 functions, ~130 lines) All functions extracted from monolithic govmap.py as pure functions. Next: Extract market analysis and create client.py with API methods. Part of Phase 3 refactoring - no functionality changes yet. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Nitzan Pomerantz
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user