""" NLP-Enhanced Polish Real Estate Extractor ========================================== Hybrid approach: Regex spotting + NLP contextual validation Architecture: - Layer 1 (REGEX): Fast pattern matching to spot potential candidates - Layer 2 (NLP): Contextual validation using spaCy to pick correct value - Layer 3 (VALIDATION): Domain knowledge validation from mappings This approach gives us: - Speed of regex (spots 100+ candidates/sec) - Accuracy of NLP (validates context, avoids false positives) - 10x better precision on ambiguous descriptions Version: 3.0 Dependencies: spaCy (pl_core_news_lg model required) """ from __future__ import annotations import re import logging from typing import Dict, Any, Optional, List, Tuple from decimal import Decimal, InvalidOperation # spaCy imports with graceful fallback try: import spacy from spacy.tokens import Doc, Token, Span SPACY_AVAILABLE = True except ImportError: SPACY_AVAILABLE = False logging.warning("spaCy not installed. NLP features disabled. Run: pip install spacy && python -m spacy download pl_core_news_lg") from .extractor import EnhancedPolishExtractor from .mappings import MAPPINGS from .semantic_matcher import SemanticMatcher, ContextAwareExtractor logger = logging.getLogger(__name__) class NLPEnhancedExtractor(EnhancedPolishExtractor): """ NLP-enhanced extractor with hybrid regex + spaCy approach. Key Features: 1. Regex spots potential values (fast, broad) 2. spaCy validates context (accurate, smart) 3. Fallback to pure regex if NLP unavailable Example: "Dom ma 4 sypialnie i 9 pokoi" - Regex spots: 4, 9 - NLP sees: "4" near "sypialnie" (bedrooms), "9" near "pokoi" (rooms) - Returns: rooms=9 (not 4, which is bedrooms) """ def __init__(self, external_mapping_path: Optional[str] = None, use_nlp: bool = True): """ Initialize NLP-enhanced extractor. Args: external_mapping_path: Path to external JSON mappings use_nlp: Enable NLP features (requires spaCy + pl_core_news_lg) """ super().__init__(external_mapping_path) self.use_nlp = use_nlp and SPACY_AVAILABLE self.nlp = None self.semantic_matcher = None self.context_extractor = None if self.use_nlp: try: # Load Polish language model (large - best accuracy) self.nlp = spacy.load("pl_core_news_lg") logger.info("✓ spaCy Polish model loaded successfully (pl_core_news_lg)") # Initialize semantic matching (v3.1) self.semantic_matcher = SemanticMatcher(self.nlp, use_semantic=True) self.context_extractor = ContextAwareExtractor(self.nlp) logger.info("✓ Semantic matching enabled (fuzzy + embeddings)") except OSError: logger.warning("Polish spaCy model not found. Run: python -m spacy download pl_core_news_lg") self.use_nlp = False self.nlp = None else: if not SPACY_AVAILABLE: logger.info("NLP features disabled: spaCy not installed") else: logger.info("NLP features disabled by configuration") # ======================================================================== # NLP UTILITIES - Context Analysis # ======================================================================== def _get_context_window(self, doc: Doc, token: Token, window: int = 5) -> List[Token]: """ Get surrounding context tokens for analysis. Args: doc: spaCy Doc object token: Target token window: Number of tokens before/after to include Returns: List of context tokens """ start = max(0, token.i - window) end = min(len(doc), token.i + window + 1) return list(doc[start:end]) def _find_nearest_keyword(self, token: Token, keywords: List[str], max_distance: int = 5) -> Optional[Tuple[str, int]]: """ Find nearest keyword to a token within max_distance. Args: token: Target token keywords: List of keywords to search for max_distance: Maximum token distance Returns: Tuple of (keyword, distance) or None """ doc = token.doc nearest_keyword = None min_distance = max_distance + 1 # Search backwards for i in range(max(0, token.i - max_distance), token.i): text_lower = doc[i].text.lower() lemma_lower = doc[i].lemma_.lower() for keyword in keywords: if keyword in text_lower or keyword in lemma_lower: distance = token.i - i if distance < min_distance: min_distance = distance nearest_keyword = keyword # Search forwards for i in range(token.i + 1, min(len(doc), token.i + max_distance + 1)): text_lower = doc[i].text.lower() lemma_lower = doc[i].lemma_.lower() for keyword in keywords: if keyword in text_lower or keyword in lemma_lower: distance = i - token.i if distance < min_distance: min_distance = distance nearest_keyword = keyword return (nearest_keyword, min_distance) if nearest_keyword else None def _has_dependency_relation(self, token: Token, dep_types: List[str]) -> bool: """ Check if token has specific dependency relation. Args: token: Token to check dep_types: List of dependency types (e.g., ['nummod', 'amod']) Returns: True if token has any of the specified dependencies """ return token.dep_ in dep_types def _is_negated_context(self, token: Token, window: int = 3) -> bool: """ Check if token appears in negated context. Args: token: Token to check window: Search window size Returns: True if negation detected in context """ negation_keywords = ['nie', 'bez', 'brak', 'nie ma', 'nieposiada'] context = self._get_context_window(token.doc, token, window) for ctx_token in context: if any(neg in ctx_token.text.lower() for neg in negation_keywords): return True return False # ======================================================================== # NLP-ENHANCED EXTRACTION - ROOMS # ======================================================================== def extract_rooms(self, text: str) -> Optional[int]: """ Extract number of rooms with NLP context validation. Regex spots numbers, NLP validates they refer to rooms (not bedrooms, bathrooms, etc.) Example: "Dom ma 4 sypialnie i 9 pokoi" - Regex: finds [4, 9] - NLP: 4 near "sypialnie" (reject), 9 near "pokoi" (accept) - Returns: 9 """ if not self.use_nlp or not self.nlp: # Fallback to parent regex-only method return super().extract_rooms(text) text = self.normalize_text(text) # Special case: kawalerka (studio) if re.search(r'\bkawalerka\b', text): return 1 # Step 1: REGEX - Spot all potential room numbers candidates = [] patterns = [ (r'(\d+)\s*pokoi', 'pokoi'), (r'(\d+)\s*[\-–]?\s*pokojow[eya]', 'pokojowe'), (r'(\d+)\s*pok\.?(?:\s|$|,)', 'pok'), (r'mieszkanie\s+(\d+)\s*[\-–]?\s*pokojowe', 'mieszkanie_pokojowe'), (r'(\d+)\s+pokoje?(?:\s|,|$|\.|;)', 'pokoje'), ] for pattern, context_type in patterns: for match in re.finditer(pattern, text, re.IGNORECASE): try: num = int(match.group(1)) if self.mappings.is_valid_value('rooms', num): candidates.append({ 'value': num, 'start': match.start(), 'end': match.end(), 'context_type': context_type, 'matched_text': match.group(0) }) except (ValueError, IndexError): continue if not candidates: return None # Step 2: NLP - Process text and validate context doc = self.nlp(text) # Room-related keywords (positive context) room_keywords = ['pokoi', 'pokoje', 'pokojowe', 'pokojowy', 'pokój', 'pok'] # Exclusion keywords (negative context - bedrooms, bathrooms, etc.) exclusion_keywords = ['sypialnia', 'sypialnie', 'łazienka', 'łazienki', 'garaż', 'garażowe', 'balkon', 'taras'] # Step 3: Score each candidate based on NLP context scored_candidates = [] for candidate in candidates: score = 50 # Base score # Find the token corresponding to this number target_token = None for token in doc: if token.idx >= candidate['start'] and token.idx < candidate['end']: if token.like_num or token.text.isdigit(): target_token = token break if not target_token: continue # Check for negation if self._is_negated_context(target_token, window=3): score -= 100 # Strong penalty for negation continue # Check proximity to room keywords (positive signal) nearest_room = self._find_nearest_keyword(target_token, room_keywords, max_distance=5) if nearest_room: keyword, distance = nearest_room score += (30 - distance * 5) # Closer = higher score # Check proximity to exclusion keywords (negative signal) nearest_exclusion = self._find_nearest_keyword(target_token, exclusion_keywords, max_distance=5) if nearest_exclusion: keyword, distance = nearest_exclusion score -= (40 - distance * 5) # Closer = bigger penalty # Check dependency relations if self._has_dependency_relation(target_token, ['nummod', 'amod']): # Check what it modifies head = target_token.head if any(kw in head.text.lower() for kw in room_keywords): score += 20 elif any(kw in head.text.lower() for kw in exclusion_keywords): score -= 30 # Context type bonus if candidate['context_type'] in ['pokoi', 'pokoje']: score += 10 scored_candidates.append({ **candidate, 'score': score }) # Step 4: Return highest-scored candidate if scored_candidates: best = max(scored_candidates, key=lambda x: x['score']) if best['score'] > 30: # Threshold for confidence logger.debug(f"NLP rooms extraction: {best['value']} (score: {best['score']}, context: {best['matched_text']})") return best['value'] # Fallback to regex-only if NLP didn't find confident result return super().extract_rooms(text) # ======================================================================== # NLP-ENHANCED EXTRACTION - AREA # ======================================================================== def extract_area(self, text: str) -> Dict[str, Any]: """ Extract area with NLP context validation. Handles ambiguous cases like: - "działka 500m2, dom 150m2" -> picks dom (building), not działka (plot) - "pokój 30 metrów" -> validates it's a room, not total area """ if not self.use_nlp or not self.nlp: return super().extract_area(text) text = self.normalize_text(text) result = {'square_footage': None, 'area_unit': None, 'area_m2': None} # Step 1: REGEX - Spot all area mentions candidates = [] patterns = [ (r'powierzchnia[\s:]+(\d+[,\.]?\d*)\s*(?:m2|m²|mkw)', 'powierzchnia', 100), (r'dom[\s\w]*(\d+[,\.]?\d*)\s*(?:m2|m²|mkw)', 'dom', 90), (r'mieszkanie[\s\w]*(\d+[,\.]?\d*)\s*(?:m2|m²|mkw)', 'mieszkanie', 90), (r'(\d+[,\.]?\d*)\s*(?:m2|m²|mkw|metr[oó]w\s+kwadratowych)', 'general', 70), (r'działka[\s:]+(\d+[,\.]?\d*)', 'dzialka', 50), # Lower priority - usually plot ] for pattern, area_type, base_score in patterns: for match in re.finditer(pattern, text, re.IGNORECASE): area_str = match.group(1).replace(',', '.') try: area_value = Decimal(area_str) full_match = match.group(0).lower() # Determine unit if 'ha' in full_match: unit = 'ha' elif any(x in full_match for x in ['ar', ' a']): unit = 'ar' else: unit = 'm2' candidates.append({ 'value': area_value, 'unit': unit, 'area_type': area_type, 'start': match.start(), 'end': match.end(), 'matched_text': match.group(0), 'base_score': base_score }) except (InvalidOperation, ValueError): continue if not candidates: return result # Step 2: NLP - Validate context doc = self.nlp(text) # Building area keywords (what we want) building_keywords = ['dom', 'mieszkanie', 'powierzchnia', 'użytkowa', 'budynek'] # Plot area keywords (what we want to avoid for building area) plot_keywords = ['działka', 'działki', 'teren', 'gruntu', 'plac'] scored_candidates = [] for candidate in candidates: score = candidate['base_score'] # Find token target_token = None for token in doc: if token.idx >= candidate['start'] and token.idx < candidate['end']: if token.like_num or '.' in token.text or ',' in token.text: target_token = token break if not target_token: continue # Check context nearest_building = self._find_nearest_keyword(target_token, building_keywords, max_distance=7) if nearest_building: keyword, distance = nearest_building score += (30 - distance * 3) nearest_plot = self._find_nearest_keyword(target_token, plot_keywords, max_distance=7) if nearest_plot: keyword, distance = nearest_plot score -= (25 - distance * 3) # Validate area range area_m2 = candidate['value'] * Decimal(str(self.mappings.area_conversions[candidate['unit']])) if not self.mappings.is_valid_value('area', float(area_m2)): score -= 50 scored_candidates.append({ **candidate, 'score': score, 'area_m2': area_m2 }) # Step 3: Return best candidate if scored_candidates: best = max(scored_candidates, key=lambda x: x['score']) if best['score'] > 40: result['square_footage'] = best['value'] result['area_unit'] = best['unit'] result['area_m2'] = best['area_m2'] logger.debug(f"NLP area extraction: {best['value']} {best['unit']} (score: {best['score']})") return result return super().extract_area(text) # ======================================================================== # NLP-ENHANCED EXTRACTION - PRICE # ======================================================================== def extract_price(self, text: str) -> Dict[str, Any]: """ Extract price with NLP context validation. Handles: - "Cena wynajmu 10 000 złotych miesięcznie" vs "Dodatkowe opłaty 2 500 złotych" - Distinguishes rent price from deposit, utilities, etc. """ if not self.use_nlp or not self.nlp: return super().extract_price(text) text = self.normalize_text(text) result = {'price': None, 'currency': None} # Step 1: REGEX - Spot all price mentions candidates = [] price_patterns = [ (r'cena[\s:]+(\d+[\s\d]*)\s*(?:zł|złotych|pln)', 'cena', 100), (r'wynajmu[\s:]+(\d+[\s\d]*)\s*(?:zł|złotych|pln)', 'wynajem', 95), (r'(\d+[\s\d]*)\s*(?:zł|złotych|pln)\s*miesięcznie', 'miesieczne', 90), (r'cena\s+sprzedaży[\s:]+(\d+[\s\d]*)', 'sprzedaz', 90), (r'(\d+[\s\d]*)\s*(?:zł|złotych|pln|€|euro|usd)', 'general', 60), ] for pattern, price_type, base_score in price_patterns: for match in re.finditer(pattern, text, re.IGNORECASE): price_str = match.group(1).replace(' ', '').replace(',', '.') try: price_value = Decimal(price_str) # Determine currency full_match = match.group(0).lower() if '€' in full_match or 'euro' in full_match: currency = 'EUR' elif '$' in full_match or 'usd' in full_match: currency = 'USD' else: currency = 'PLN' candidates.append({ 'value': price_value, 'currency': currency, 'price_type': price_type, 'start': match.start(), 'end': match.end(), 'matched_text': match.group(0), 'base_score': base_score }) except (InvalidOperation, ValueError): continue if not candidates: return result # Step 2: NLP - Validate context doc = self.nlp(text) # Main price keywords (positive) main_price_keywords = ['cena', 'wynajmu', 'sprzedaży', 'miesięcznie', 'kosztuje'] # Exclusion keywords (utilities, deposit, etc.) exclusion_keywords = ['opłaty', 'dodatkowe', 'kaucja', 'energia', 'prąd', 'woda', 'ogrzewanie', 'czynsz', 'administracyjne'] scored_candidates = [] for candidate in candidates: score = candidate['base_score'] # Find token target_token = None for token in doc: if token.idx >= candidate['start'] and token.idx < candidate['end']: if token.like_num or token.text.replace(' ', '').isdigit(): target_token = token break if not target_token: continue # Check for exclusion context nearest_exclusion = self._find_nearest_keyword(target_token, exclusion_keywords, max_distance=10) if nearest_exclusion: keyword, distance = nearest_exclusion score -= (60 - distance * 5) # Strong penalty for utility/deposit context # Check for main price context nearest_main = self._find_nearest_keyword(target_token, main_price_keywords, max_distance=8) if nearest_main: keyword, distance = nearest_main score += (25 - distance * 2) # Validate price range if not self.mappings.is_valid_value('price', float(candidate['value'])): score -= 30 scored_candidates.append({ **candidate, 'score': score }) # Step 3: Return best candidate if scored_candidates: best = max(scored_candidates, key=lambda x: x['score']) if best['score'] > 50: result['price'] = best['value'] result['currency'] = best['currency'] logger.debug(f"NLP price extraction: {best['value']} {best['currency']} (score: {best['score']})") return result return super().extract_price(text) # ======================================================================== # SEMANTIC-ENHANCED CATEGORICAL EXTRACTION (v3.1) # ======================================================================== def extract_condition(self, text: str) -> Optional[str]: """ Extract estate condition with semantic matching + context awareness. Enhancements: - Fuzzy matching: "dobry stam" → matches "dobry stan" - Synonym detection: "odnowiony" → matches "po remoncie" - Context validation: "dobry" only extracted if near "stan" keywords """ if not self.use_nlp or not self.semantic_matcher: return super().extract_condition(text) text = self.normalize_text(text) # Get condition vocabulary from mappings (variant -> canonical mapping) condition_map = self.mappings.condition_map if not condition_map: return super().extract_condition(text) # Context keywords for condition context_keywords = ['stan', 'techniczny', 'wykończenie', 'kondycja', 'budynku'] # Build canonical -> keywords mapping canonical_keywords = {} for keyword, canonical in condition_map.items(): if canonical not in canonical_keywords: canonical_keywords[canonical] = [] canonical_keywords[canonical].append(keyword) # Try semantic matching with context validation for canonical_value, keywords in canonical_keywords.items(): if self.semantic_matcher.match_with_context(text, keywords, context_keywords, window=50): if not self.semantic_matcher.is_negated(text, keywords[0]): logger.debug(f"Semantic extraction (condition): {canonical_value}") return canonical_value # Fallback to regex return super().extract_condition(text) def extract_heating(self, text: str) -> Optional[str]: """ Extract heating type with semantic matching. Enhancements: - Typo tolerance: "gazow" → matches "gazowe" - Synonym detection: "centralne" → matches "miejskie" """ if not self.use_nlp or not self.semantic_matcher: return super().extract_heating(text) text = self.normalize_text(text) # Get heating vocabulary (variant -> canonical mapping) heating_map = self.mappings.heating_map if not heating_map: return super().extract_heating(text) # Context keywords context_keywords = self.mappings.heating_context_keywords or ['ogrzewanie', 'c.o.', 'ciepło', 'grzewcz', 'podłogowe'] # Build canonical -> keywords mapping canonical_keywords = {} for keyword, canonical in heating_map.items(): if canonical not in canonical_keywords: canonical_keywords[canonical] = [] canonical_keywords[canonical].append(keyword) # Try semantic matching for canonical_value, keywords in canonical_keywords.items(): if self.semantic_matcher.match_with_context(text, keywords, context_keywords, window=40): if not self.semantic_matcher.is_negated(text, keywords[0]): logger.debug(f"Semantic extraction (heating): {canonical_value}") return canonical_value return super().extract_heating(text) def extract_building_type(self, text: str) -> Optional[str]: """ Extract building type with semantic matching. Enhancements: - Handles variations: "apartamentowic" → "apartamentowiec" - Context awareness: Only extracts near "budynek", "typ", "zabudowa" """ if not self.use_nlp or not self.semantic_matcher: return super().extract_building_type(text) text = self.normalize_text(text) building_map = self.mappings.building_type_map if not building_map: return super().extract_building_type(text) context_keywords = ['budynek', 'typ', 'zabudowa', 'budowla'] # Build canonical -> keywords mapping canonical_keywords = {} for keyword, canonical in building_map.items(): if canonical not in canonical_keywords: canonical_keywords[canonical] = [] canonical_keywords[canonical].append(keyword) for canonical_value, keywords in canonical_keywords.items(): if self.semantic_matcher.match_with_context(text, keywords, context_keywords, window=50): if not self.semantic_matcher.is_negated(text, keywords[0]): logger.debug(f"Semantic extraction (building_type): {canonical_value}") return canonical_value return super().extract_building_type(text) def extract_windows(self, text: str) -> Optional[str]: """ Extract window type with semantic matching. Enhancements: - Typo tolerance: "plastikow" → "plastikowe" - Context validation: Only near "okna", "stolarka" """ if not self.use_nlp or not self.semantic_matcher: return super().extract_windows(text) text = self.normalize_text(text) windows_map = self.mappings.window_map if not windows_map: return super().extract_windows(text) context_keywords = ['okna', 'stolarka', 'okienna'] # Build canonical -> keywords mapping canonical_keywords = {} for keyword, canonical in windows_map.items(): if canonical not in canonical_keywords: canonical_keywords[canonical] = [] canonical_keywords[canonical].append(keyword) for canonical_value, keywords in canonical_keywords.items(): if self.semantic_matcher.match_with_context(text, keywords, context_keywords, window=40): if not self.semantic_matcher.is_negated(text, keywords[0]): logger.debug(f"Semantic extraction (windows): {canonical_value}") return canonical_value return super().extract_windows(text) def extract_building_material(self, text: str) -> Optional[str]: """ Extract building material with semantic matching. Note: material_map not available in current mappings, fallback to parent method. """ # Material mapping not yet implemented in JSON config # Fallback to regex-based parent method return super().extract_building_material(text) # ======================================================================== # WRAPPER METHOD - Extract All # ======================================================================== def extract_all(self, text: str) -> Dict[str, Any]: """ Extract all fields using NLP-enhanced methods. This method automatically uses NLP validation when available, falls back to regex-only for unsupported fields. """ if not self.use_nlp: logger.info("Using regex-only extraction (NLP disabled)") return super().extract_all(text) logger.info("Using NLP-enhanced extraction (spaCy + Semantic v3.1)") # Call parent method - our overridden methods will be used automatically return super().extract_all(text)