""" Enhanced Polish Real Estate Extractor - Core Logic ================================================== Pure extraction logic separated from domain knowledge. This module contains ONLY extraction algorithms and pattern matching. All vocabulary and mappings are loaded from mappings.py. Version: 2.0 Architecture: Clean separation of logic and data Dependencies: mappings.py only (no Django required) """ import re import os import json import unicodedata from decimal import Decimal, InvalidOperation from datetime import datetime, date from typing import Dict, Any, Optional, List, Tuple from .mappings import MAPPINGS class EnhancedPolishExtractor: """ Production-grade Polish real estate data extractor. Design Principles: - Pure Python (no external dependencies beyond stdlib) - Thread-safe (immutable after init) - Testable (no side effects) - Fast (regex-based, ~100-200 extractions/sec) - Extensible (easy to add new extraction methods) Usage: extractor = EnhancedPolishExtractor() data = extractor.extract_all(description_text) """ def __init__(self, external_mapping_path: Optional[str] = None): """ Initialize extractor with mappings. Args: external_mapping_path: Optional path to JSON file with additional mappings (e.g., from API providers like OtoDom) """ self.mappings = MAPPINGS self.external_mapping = self._load_external_mapping(external_mapping_path) # ======================================================================== # EXTERNAL MAPPING LOADER # ======================================================================== def _load_external_mapping(self, path: Optional[str] = None) -> Dict[str, Any]: """ Load external mapping JSON (e.g., message2.txt from OtoDom API). This allows integration with API providers who give us additional categorization data that can improve extraction accuracy. """ if path and os.path.exists(path): try: with open(path, 'r', encoding='utf-8') as f: data = json.load(f) return data if isinstance(data, dict) else {} except Exception: return {} # Auto-detect in same directory as this file try: here = os.path.dirname(os.path.abspath(__file__)) for fname in ('message2.txt', 'message.txt'): fpath = os.path.join(here, fname) if os.path.exists(fpath): with open(fpath, 'r', encoding='utf-8') as f: data = json.load(f) return data if isinstance(data, dict) else {} except Exception: pass return {} # ======================================================================== # TEXT NORMALIZATION # ======================================================================== def normalize_text(self, text: str) -> str: """ Enhanced text normalization for Polish language. Handles: - Unicode normalization (different encodings) - Lowercasing - Whitespace cleanup """ if not text: return "" # Lowercase first text = text.lower() # Normalize unicode (handle UTF-8, ISO-8859-2, Windows-1250) text = unicodedata.normalize('NFKC', text) # Collapse multiple whitespace but preserve structure text = ' '.join(text.split()) return text def normalize_number(self, num_str: str) -> Optional[Decimal]: """ Parse Polish and international number formats. Examples: "2 500" → 2500 "2.500,50" → 2500.50 (Polish) "2,500.50" → 2500.50 (International) "2500,50" → 2500.50 Returns: Decimal or None if parsing fails """ if not num_str: return None # Remove whitespace num_str = re.sub(r'\s+', '', num_str) # Handle mixed formats if ',' in num_str and '.' in num_str: # Determine which is decimal separator comma_pos = num_str.rindex(',') dot_pos = num_str.rindex('.') if comma_pos > dot_pos: # Polish: 2.500,50 → 2500.50 num_str = num_str.replace('.', '').replace(',', '.') else: # International: 2,500.50 → 2500.50 num_str = num_str.replace(',', '') # Only comma (Polish decimal) elif ',' in num_str: num_str = num_str.replace(',', '.') # Only dot: check if decimal or thousand separator elif '.' in num_str: parts = num_str.split('.') # If last part has >2 digits, it's thousand separator if len(parts[-1]) > 2: num_str = num_str.replace('.', '') try: return Decimal(num_str) except (InvalidOperation, ValueError): return None # ======================================================================== # PRICE EXTRACTION # ======================================================================== def extract_price(self, text: str) -> Optional[Decimal]: """ Extract rental/sale price with priority-based pattern matching. Looks for: - Explicit labels (czynsz, cena) - Currency indicators (zł, PLN) - Period indicators (/mc, miesięcznie) - Ranges (takes minimum) """ text = self.normalize_text(text) # Priority patterns: (regex, priority_score) patterns = [ (r'(\d[\d\s\.,]*)\s*(?:zł|pln)\s*(?:/|za|na)\s*(?:mc|miesiąc|miesięcznie|m-c)', 95), (r'(?:czynsz|wynajem|koszt)[\s:]+(\d[\d\s\.,]*)\s*(?:zł|pln)', 90), (r'cena[\s:]+(\d[\d\s\.,]*)\s*(?:zł|pln)', 85), (r'od\s+(\d[\d\s\.,]*)\s*(?:do\s+\d[\d\s\.,]*)?\s*(?:zł|pln)', 75), (r'(?:^|\n|\|)\s*(\d[\d\s\.,]*)\s*(?:zł|pln)', 70), (r'(\d[\d\s\.,]*)\s*(?:zł|złotych|pln)', 60), ] best_match = None best_priority = 0 for pattern, priority in patterns: match = re.search(pattern, text, re.IGNORECASE) if match and priority > best_priority: price = self.normalize_number(match.group(1)) if price and self.mappings.is_valid_value('price', float(price)): best_match = price best_priority = priority return best_match # ======================================================================== # AREA EXTRACTION # ======================================================================== def extract_area(self, text: str) -> Dict[str, Any]: """ Extract area with automatic unit conversion. Returns dict with: - square_footage: Original value - area_unit: Unit of measurement - area_m2: Normalized to square meters """ text = self.normalize_text(text) result = {'square_footage': None, 'area_unit': None, 'area_m2': None} # Patterns with priority patterns = [ (r'powierzchnia[\s:]+(\d+[,\.]?\d*)\s*(?:m2|m²|mkw)', 95), (r'(\d+[,\.]?\d*)\s*(?:m2|m²|mkw|metr[oó]w\s+kwadratowych)', 90), (r'(\d+[,\.]?\d*)\s*ha', 85), (r'(\d+[,\.]?\d*)\s*(?:ar[yów]*|a\.)', 80), (r'pow\.?\s*(\d+[,\.]?\d*)', 70), (r'(\d+[,\.]?\d*)\s*(?:ft2|ft²|sq\s*ft|sqft)', 80), ] for pattern, priority in patterns: match = re.search(pattern, text, re.IGNORECASE) if match: area_str = match.group(1).replace(',', '.') try: area_value = Decimal(area_str) full_match = match.group(0).lower() # Determine unit and convert if 'ha' in full_match: unit = 'ha' conversion = self.mappings.area_conversions['ha'] elif any(x in full_match for x in ['ar', ' a']): unit = 'ar' conversion = self.mappings.area_conversions['ar'] elif any(x in full_match for x in ['ft2', 'ft²', 'sq ft', 'sqft']): unit = 'ft2' conversion = self.mappings.area_conversions['ft2'] else: unit = 'm2' conversion = self.mappings.area_conversions['m2'] result['square_footage'] = area_value result['area_unit'] = unit result['area_m2'] = area_value * Decimal(str(conversion)) if self.mappings.is_valid_value('area', float(result['area_m2'])): break else: result = {'square_footage': None, 'area_unit': None, 'area_m2': None} except (InvalidOperation, ValueError): continue return result def extract_land_area(self, text: str) -> Optional[Decimal]: """Extract land/plot area (działka).""" text = self.normalize_text(text) patterns = [ r'działka[\s:]+(\d+[,\.]?\d*)\s*(?:m2|m²|mkw)', r'działka[\s:]+(\d+[,\.]?\d*)\s*(?:ha|ar)', r'powierzchnia\s+działki[\s:]+(\d+[,\.]?\d*)', ] for pattern in patterns: match = re.search(pattern, text, re.IGNORECASE) if match: area_str = match.group(1).replace(',', '.') try: area = Decimal(area_str) full_match = match.group(0).lower() # Convert to m² if 'ha' in full_match: area *= Decimal(str(self.mappings.area_conversions['ha'])) elif 'ar' in full_match: area *= Decimal(str(self.mappings.area_conversions['ar'])) return area except (InvalidOperation, ValueError): continue return None # ======================================================================== # COUNT FIELDS # ======================================================================== def extract_rooms(self, text: str) -> Optional[int]: """Extract number of rooms with special case handling (kawalerka=studio).""" text = self.normalize_text(text) # Special case: studio apartment if re.search(r'\bkawalerka\b', text): return 1 patterns = [ r'(?:liczba\s+)?pokoi[\s:]+(\d+)', r'(\d+)\s*[\-–]?\s*pokojow[eya]', r'(\d+)\s*pok\.?(?:\s|$|,)', r'mieszkanie\s+(\d+)\s*[\-–]?\s*pokojowe', r'(\d+)\s+pokoje?(?:\s|,|$|\.|;)', ] for pattern in patterns: match = re.search(pattern, text, re.IGNORECASE) if match: try: rooms = int(match.group(1)) if self.mappings.is_valid_value('rooms', rooms): return rooms except (ValueError, IndexError): continue return None def extract_bathrooms(self, text: str) -> Optional[int]: """Extract number of bathrooms.""" text = self.normalize_text(text) patterns = [ r'(\d+)\s*łazien[ekikę]+', r'łazien[ekika]+[\s:]+(\d+)', r'(\d+)\s*(?:wc|toalet[ya])', ] for pattern in patterns: match = re.search(pattern, text, re.IGNORECASE) if match: try: bathrooms = int(match.group(1)) if self.mappings.is_valid_value('bathrooms', bathrooms): return bathrooms except (ValueError, IndexError): continue return None # ======================================================================== # FLOOR EXTRACTION # ======================================================================== def extract_floor(self, text: str) -> Dict[str, Optional[int]]: """ Extract floor number and total floors. Handles: - Combined format: "3/5", "piętro 3/5" - Ground floor: "parter" - Separate mentions """ text = self.normalize_text(text) result = {'floor': None, 'floors_num': None} # Combined patterns: X/Y or X z Y combined_patterns = [ r'(?:piętro|pietro|pię)[\s:]*(\d+)\s*[/z]\s*(\d+)', r'(\d+)\s*[/z]\s*(\d+)\s*(?:piętro|pietro|pię)', ] for pattern in combined_patterns: match = re.search(pattern, text, re.IGNORECASE) if match: try: floor = int(match.group(1)) total = int(match.group(2)) if (self.mappings.is_valid_value('floor', floor) and 0 <= floor <= total): result['floor'] = floor result['floors_num'] = total return result except (ValueError, IndexError): continue # Ground floor if re.search(r'\bparter\b', text): result['floor'] = 0 # Single floor mention if result['floor'] is None: single_patterns = [ r'(?:na\s+)?(?:piętro|pietro)[\s:]+(\d+)', r'(\d+)\s*(?:\.?\s*)?(?:piętro|pietro)', ] for pattern in single_patterns: match = re.search(pattern, text, re.IGNORECASE) if match: try: floor = int(match.group(1)) if self.mappings.is_valid_value('floor', floor): result['floor'] = floor break except (ValueError, IndexError): continue # Total floors if result['floors_num'] is None: total_patterns = [ r'budynek\s+(?:ma|posiada|składa\s+się\s+z)?\s*(\d+)\s*(?:piętr|kondygnacji|poziomów)', r'(\d+)\s*[-–]\s*piętrowy', ] for pattern in total_patterns: match = re.search(pattern, text, re.IGNORECASE) if match: try: total = int(match.group(1)) if 1 <= total <= 100: result['floors_num'] = total break except (ValueError, IndexError): continue return result # ======================================================================== # DATE & TIME # ======================================================================== def extract_build_year(self, text: str) -> Optional[str]: """Extract construction year with validation.""" text = self.normalize_text(text) patterns = [ r'rok\s+budowy[\s:]+(\d{4})', r'budow[aany]+[\s:]+(\d{4})', r'(?:z|rok)\s+(\d{4})', r'(\d{4})\s*r\.?(?:\s+budowy)?', r'wybudowany\s+w\s+(\d{4})', ] current_year = datetime.now().year for pattern in patterns: match = re.search(pattern, text, re.IGNORECASE) if match: year = match.group(1) year_int = int(year) if self.mappings.is_valid_value('build_year', year_int): return year return None def extract_available_from(self, text: str) -> Optional[str]: """ Extract availability date. Handles: - ISO format: 2024-01-15 - Polish format: 15 stycznia 2024 - Immediate: "od zaraz", "natychmiast" """ text = self.normalize_text(text) # ISO format iso_pattern = r'dostępn[eya]+\s+od\s+(\d{4}-\d{2}-\d{2})' match = re.search(iso_pattern, text, re.IGNORECASE) if match: return match.group(1) # Polish date format pl_pattern = r'dostępn[eya]+\s+od\s+(\d{1,2})\s+(\w+)\s+(\d{4})' match = re.search(pl_pattern, text, re.IGNORECASE) if match: day = int(match.group(1)) month_name = match.group(2).lower() year = int(match.group(3)) month = self.mappings.months_pl.get(month_name) if month: try: date_obj = date(year, month, day) return date_obj.strftime('%Y-%m-%d') except ValueError: pass # Immediate availability for keyword in self.mappings.availability_keywords: if keyword in text: return datetime.now().strftime('%Y-%m-%d') return None # ======================================================================== # BOOLEAN FIELDS # ======================================================================== def extract_boolean_field(self, text: str, field: str) -> Optional[bool]: """ Extract boolean field with context-aware negation detection. Args: text: Description text field: Feature key from mappings.boolean_keywords Returns: True if feature present, False if explicitly absent, None if not mentioned """ text = self.normalize_text(text) if field not in self.mappings.boolean_keywords: return None keywords = self.mappings.boolean_keywords[field] for keyword in keywords: # Find all occurrences with word boundaries pattern = rf'\b{re.escape(keyword)}\w*' for match in re.finditer(pattern, text, re.IGNORECASE): pos = match.start() # Context window: 70 chars before, 30 after context_start = max(0, pos - 70) context_end = min(len(text), pos + len(keyword) + 30) context = text[context_start:context_end] # Check for negation within 15 chars before keyword neg_window = context[max(0, pos-context_start-15):pos-context_start+5] has_negative = any(neg in neg_window for neg in self.mappings.negative_keywords) return not has_negative return None # ======================================================================== # CATEGORICAL FIELDS # ======================================================================== def extract_condition(self, text: str) -> Optional[str]: """Extract property condition (returns Polish canonical label).""" text = self.normalize_text(text) # Sort by length for specificity (longer phrases first) sorted_conditions = sorted( self.mappings.condition_map.items(), key=lambda x: len(x[0]), reverse=True ) for variant, canonical in sorted_conditions: if variant in text: return canonical return None def _has_heating_context(self, text: str, start: int, end: int) -> bool: """Check if match is in heating context (avoids false positives).""" window_start = max(0, start - 30) window_end = min(len(text), end + 30) ctx = text[window_start:window_end] return any(kw in ctx for kw in self.mappings.heating_context_keywords) def extract_heating(self, text: str) -> Optional[str]: """ Extract heating type with context validation. Uses external mapping if available, falls back to internal mappings. """ text = self.normalize_text(text) # Try external mapping first (higher accuracy) if self.external_mapping and 'heating_type' in self.external_mapping: ex_map = self.external_mapping['heating_type'] strong_hits: List[Tuple[int, str]] = [] weak_hits: List[Tuple[int, str]] = [] for category, phrases in ex_map.items(): canonical = MAPPINGS.heating_category_to_polish(category) if not canonical: continue for phrase in (phrases or []): p = phrase.strip().lower() if not p: continue pos = text.find(p) if pos == -1: continue end = pos + len(p) # Strong if phrase itself contains heating anchor is_strong = any(a in p for a in ['ogrzew', 'c.o', 'centralne', 'sieć']) if is_strong: strong_hits.append((pos, canonical)) elif self._has_heating_context(text, pos, end): weak_hits.append((pos, canonical)) if strong_hits: strong_hits.sort(key=lambda x: x[0]) return strong_hits[0][1] if weak_hits: weak_hits.sort(key=lambda x: x[0]) return weak_hits[0][1] # Fallback to internal mapping (with strict context requirement) sorted_heating = sorted( self.mappings.heating_map.items(), key=lambda x: len(x[0]), reverse=True ) for pattern, canonical in sorted_heating: m = re.search(rf'\b{re.escape(pattern)}\b', text, re.IGNORECASE) if m and self._has_heating_context(text, m.start(), m.end()): return canonical return None def extract_building_type(self, text: str) -> Optional[str]: """Extract building type.""" text = self.normalize_text(text) sorted_types = sorted( self.mappings.building_type_map.items(), key=lambda x: len(x[0]), reverse=True ) for pattern, canonical in sorted_types: if pattern in text: return canonical return None def extract_market_type(self, text: str) -> Optional[str]: """Extract market type (primary/secondary).""" text = self.normalize_text(text) for pattern, canonical in self.mappings.market_type_map.items(): if pattern in text: return canonical return None def extract_ownership_form(self, text: str) -> Optional[str]: """Extract ownership form.""" text = self.normalize_text(text) sorted_ownership = sorted( self.mappings.ownership_map.items(), key=lambda x: len(x[0]), reverse=True ) for pattern, canonical in sorted_ownership: if pattern in text: return canonical return None def extract_windows(self, text: str) -> Optional[str]: """Extract window type or direction.""" text = self.normalize_text(text) # Direction hints (kept as-is) directions = ['na południe', 'na północ', 'na wschód', 'na zachód'] for direction in directions: if direction in text: return direction # Window types sorted_windows = sorted( self.mappings.window_map.items(), key=lambda x: len(x[0]), reverse=True ) for pattern, canonical in sorted_windows: if re.search(rf'(?:okna|okien)?[\s:]*\b{re.escape(pattern)}\b', text, re.IGNORECASE): return canonical return None def extract_building_material(self, text: str) -> Optional[str]: """Extract building material.""" text = self.normalize_text(text) # Sort by length for specificity sorted_materials = sorted(self.mappings.building_materials, key=len, reverse=True) for material in sorted_materials: if material in text: return material return None def extract_energy_certificate(self, text: str) -> Optional[str]: """Extract energy certificate class (A-G).""" text = self.normalize_text(text) patterns = [ r'certyfikat\s+energetyczny[\s:]+([A-G]\+?)', r'świadectwo\s+energetyczne[\s:]+([A-G]\+?)', r'klasa\s+energetyczna[\s:]+([A-G]\+?)', r'energy\s+certificate[\s:]+([A-G]\+?)', r'energy\s+class[\s:]+([A-G]\+?)', ] for pattern in patterns: match = re.search(pattern, text, re.IGNORECASE) if match: return match.group(1).upper() return None # ======================================================================== # COMPOSITE/COMPLEX FIELDS # ======================================================================== def extract_balcony(self, text: str) -> Optional[str]: """Extract balcony information with details.""" text = self.normalize_text(text) patterns = [ r'balkon[\s:]+(\w+(?:\s+\w+){0,2})', r'(\d+)\s*balkon[yów]*', r'loggia[\s:]+(\w+)', ] for pattern in patterns: match = re.search(pattern, text, re.IGNORECASE) if match: return match.group(1).strip() # Simple presence if any(w in text for w in ['balkon', 'balkony', 'loggia', 'balcony']): return 'tak' return None def extract_parking_space(self, text: str) -> Optional[str]: """Extract parking information.""" text = self.normalize_text(text) patterns = [ r'parking[\s:]+(\w+(?:\s+\w+){0,2})', r'miejsce\s+parkingowe[\s:]+(\w+(?:\s+\w+){0,2})', r'(\d+)\s*miejsc?\s*parkingowych?', r'garaż\s+(\w+)', ] for pattern in patterns: match = re.search(pattern, text, re.IGNORECASE) if match: return match.group(1).strip() # English patterns → Polish labels if 'underground garage' in text: return 'garaż podziemny' if 'on-street parking' in text or 'on street parking' in text: return 'parking na ulicy' if 'parking space' in text: return 'miejsce parkingowe' return None def extract_media(self, text: str) -> Optional[str]: """Extract available media as comma-separated string.""" text = self.normalize_text(text) media_list = [] for keyword, label in self.mappings.media_keywords.items(): if keyword in text: # Check not negated pos = text.find(keyword) context = text[max(0, pos-30):pos+len(keyword)+10] if not any(neg in context for neg in self.mappings.negative_keywords): if label not in media_list: media_list.append(label) return ', '.join(media_list) if media_list else None def extract_security(self, text: str) -> Optional[str]: """Extract security features.""" text = self.normalize_text(text) security_list = [] for keyword in self.mappings.security_keywords: if keyword in text and keyword not in security_list: security_list.append(keyword) return ', '.join(security_list) if security_list else None # ======================================================================== # ADDRESS & LOCATION # ======================================================================== def extract_address_components(self, text: str) -> Dict[str, Optional[str]]: """Extract address components (city, street, zipcode, etc).""" text = self.normalize_text(text) result = { 'city': None, 'street': None, 'district': None, 'zipcode': None, 'province': None, 'neighborhood': None } # Zipcode: XX-XXX zipcode_match = re.search(r'\b(\d{2}-\d{3})\b', text) if zipcode_match: result['zipcode'] = zipcode_match.group(1) # City extraction - check for word boundaries for city in self.mappings.polish_cities: pattern = rf'\b{re.escape(city)}' if re.search(pattern, text, re.IGNORECASE): result['city'] = city break # Street extraction for street_type in self.mappings.street_types: pattern = rf'{street_type}\s+([A-ZĄĆĘŁŃÓŚŹŻa-ząćęłńóśźż\s\-]+?)(?:\s+\d+|\s*,|\s*\.|$)' match = re.search(pattern, text, re.IGNORECASE) if match: street_name = match.group(1).strip() street_name = re.sub(r'\s+\d+.*', '', street_name) result['street'] = street_name.title() break # District/neighborhood district_patterns = [ r'dzielnica[\s:]+([A-ZĄĆĘŁŃÓŚŹŻa-ząćęłńóśźż\s\-]+?)(?:\,|\.|$)', r'osiedle[\s:]+([A-ZĄĆĘŁŃÓŚŹŻa-ząćęłńóśźż\s\-]+?)(?:\,|\.|$)', ] for pattern in district_patterns: match = re.search(pattern, text, re.IGNORECASE) if match: district_name = match.group(1).strip() result['district'] = district_name.title() break return result def extract_coordinates(self, text: str) -> Dict[str, Optional[Decimal]]: """Extract GPS coordinates with validation.""" result = {'lat': None, 'lon': None} # Pattern: lat: 52.229676, lon: 21.012229 coord_pattern = r'(?:lat|latitude)[\s:]+(\d+\.\d+)[\s,]+(?:lon|lng|longitude)[\s:]+(\d+\.\d+)' match = re.search(coord_pattern, text, re.IGNORECASE) if match: try: lat = Decimal(match.group(1)) lon = Decimal(match.group(2)) # Validate Polish coordinates if self.mappings.is_valid_coordinate(float(lat), float(lon)): result['lat'] = lat result['lon'] = lon except (InvalidOperation, ValueError): pass return result # ======================================================================== # MAIN EXTRACTION ORCHESTRATOR # ======================================================================== def extract_all(self, description: str, **additional_fields) -> Dict[str, Any]: """ Main extraction method - orchestrates all extraction methods. Args: description: Property description text (required) **additional_fields: Pre-extracted structured fields to merge Returns: Dictionary with all extracted fields Example: extractor = EnhancedPolishExtractor() data = extractor.extract_all( "Mieszkanie 3-pokojowe, 65m2, 2500zł...", source_url="https://..." ) """ if not description: return {} result = {} # ===== NUMERIC FIELDS ===== result['rent'] = self.extract_price(description) # Area with conversion area_data = self.extract_area(description) result.update(area_data) result['land_area'] = self.extract_land_area(description) # ===== COUNT FIELDS ===== result['rooms'] = self.extract_rooms(description) result['bathrooms'] = self.extract_bathrooms(description) # Floor information floor_data = self.extract_floor(description) result.update(floor_data) # ===== TEXT/CATEGORICAL FIELDS ===== result['estate_condition'] = self.extract_condition(description) result['heating_type'] = self.extract_heating(description) result['building_type'] = self.extract_building_type(description) result['market_type'] = self.extract_market_type(description) result['ownership_form'] = self.extract_ownership_form(description) result['windows'] = self.extract_windows(description) result['building_material'] = self.extract_building_material(description) result['energy_certificate'] = self.extract_energy_certificate(description) # ===== DATE FIELDS ===== result['build_year'] = self.extract_build_year(description) result['available_from'] = self.extract_available_from(description) # ===== COMPOSITE FIELDS ===== result['balcony'] = self.extract_balcony(description) result['parking_space'] = self.extract_parking_space(description) result['media'] = self.extract_media(description) result['security'] = self.extract_security(description) # ===== LOCATION FIELDS ===== address_data = self.extract_address_components(description) result.update(address_data) coord_data = self.extract_coordinates(description) result.update(coord_data) # ===== BOOLEAN FIELDS ===== for feature_key in self.mappings.get_all_feature_keys(): value = self.extract_boolean_field(description, feature_key) if value is not None: result[feature_key] = value # ===== MERGE ADDITIONAL FIELDS ===== # Only add if not already extracted or if extracted value is None for key, value in additional_fields.items(): if key not in result or result[key] is None: result[key] = value return result # ======================================================================== # BATCH PROCESSING # ======================================================================== def extract_batch(self, descriptions: List[str]) -> List[Dict[str, Any]]: """ Process multiple descriptions in batch. Args: descriptions: List of description texts Returns: List of extraction results """ return [self.extract_all(desc) for desc in descriptions] # ======================================================================== # UTILITY METHODS # ======================================================================== def get_extraction_coverage(self, result: Dict[str, Any]) -> Dict[str, Any]: """ Calculate extraction coverage statistics. Args: result: Extraction result from extract_all() Returns: Dict with coverage statistics """ total_fields = len(result) filled_fields = sum(1 for v in result.values() if v is not None) return { 'total_fields': total_fields, 'filled_fields': filled_fields, 'empty_fields': total_fields - filled_fields, 'coverage_percent': round((filled_fields / total_fields * 100), 2) if total_fields > 0 else 0, 'filled_field_names': [k for k, v in result.items() if v is not None], 'empty_field_names': [k for k, v in result.items() if v is None] } # ============================================================================ # MODULE-LEVEL CONVENIENCE FUNCTIONS # ============================================================================ def extract_from_text(text: str, **kwargs) -> Dict[str, Any]: """ Convenience function for one-off extractions. Usage: from extractors.core.extractor import extract_from_text data = extract_from_text("Mieszkanie 3-pokojowe...") """ extractor = EnhancedPolishExtractor() return extractor.extract_all(text, **kwargs) def create_extractor(external_mapping_path: Optional[str] = None) -> EnhancedPolishExtractor: """ Factory function to create configured extractor. Usage: from extractors.core.extractor import create_extractor extractor = create_extractor() # Reuse for multiple extractions data1 = extractor.extract_all(text1) data2 = extractor.extract_all(text2) """ return EnhancedPolishExtractor(external_mapping_path=external_mapping_path) # ============================================================================ # TESTING & VALIDATION # ============================================================================ if __name__ == "__main__": """Quick validation test.""" print("=" * 70) print("ENHANCED POLISH EXTRACTOR - CORE MODULE") print("=" * 70) # Test extraction test_text = ''' Mieszkanie 3-pokojowe, 65 m2, 5 piętro/7. Czynsz 2500 zł/mc. Warszawa, ul. Marszałkowska 15. Rok budowy 2020. Winda, klimatyzacja, balkon, garaż. Dostępne od zaraz. Stan idealny. Ogrzewanie miejskie. ''' extractor = EnhancedPolishExtractor() result = extractor.extract_all(test_text) print("\n📝 TEST EXTRACTION:") print("-" * 70) for key, value in result.items(): if value is not None: print(f" {key:25} → {value}") print("\n📊 COVERAGE STATISTICS:") print("-" * 70) stats = extractor.get_extraction_coverage(result) print(f" Total fields: {stats['total_fields']}") print(f" Filled fields: {stats['filled_fields']}") print(f" Coverage: {stats['coverage_percent']}%") print("\n✓ Core extractor validated successfully") print("=" * 70)