diff --git a/project/backend/src/ai/address_validation/validator.py b/project/backend/src/ai/address_validation/validator.py new file mode 100644 index 0000000..586481e --- /dev/null +++ b/project/backend/src/ai/address_validation/validator.py @@ -0,0 +1,348 @@ +import requests +import json +from pathlib import Path +import sqlite3 +from typing import Optional, Tuple, Dict +import re + +# ====================================================================== +# --- CONFIGURATION & API ENDPOINTS --- +# ====================================================================== + +# 1. GEOCoding API for Address Validity & Place ID +GEOCODING_API_KEY = "17ef1e91c39e4aad80e6dcde4c6774f7" +GEOCODING_ENDPOINT = "https://api.geoapify.com/v1/geocode/search" # Example Geoapify endpoint + +# 2. Place Details API for Building Type Classification +PLACE_DETAILS_API_KEY = "5048e3524f4e426c9603e7438493aa03" +# Using the correct Geoapify Place Details API endpoint +PLACE_DETAILS_ENDPOINT = "https://api.geoapify.com/v2/place-details" + +# ====================================================================== +# --- 1. MOCK DATABASE RETRIEVAL --- +# ====================================================================== + +# ---------------------------------------------------------------------- +# Infrastructure SQL helpers (Unchanged) +# ---------------------------------------------------------------------- +def find_infrastructure_dir(start: Optional[Path] = None, max_up: int = 6) -> Optional[Path]: + """Search upward from `start` (or this file) for an 'infrastructure' folder. + Returns the Path or None if not found. + """ + if start is None: + start = Path(__file__).resolve() + p = start + for _ in range(max_up + 1): + cand = p / "infrastructure/init" + if cand.is_dir(): + return cand + if p.parent == p: + break + p = p.parent + return None + + +def read_sql_file(filename: str, infra_dir: Optional[Path] = None) -> str: + """Read and return the contents of a SQL file inside the infrastructure dir. + + Raises FileNotFoundError if the directory or file cannot be found. + """ + if infra_dir is None: + infra_dir = find_infrastructure_dir() + if infra_dir is None: + raise FileNotFoundError("infrastructure directory not found (searched upward from validator.py)") + file_path = infra_dir / filename + if not file_path.is_file(): + raise FileNotFoundError(f"{file_path} not found") + return file_path.read_text(encoding="utf-8") + + +def load_sql_into_sqlite(sql_text: str, conn: sqlite3.Connection) -> None: + """Execute a SQL script (may contain multiple statements) into the given sqlite3 connection.""" + conn.executescript(sql_text) + + +def load_infrastructure_sqls(infra_dir: Optional[Path] = None, into_memory_db: bool = True) -> Tuple[Optional[sqlite3.Connection], Dict[str, str]]: + """Read `apartments.sql` and `buildings.sql` from infrastructure. + + If `into_memory_db` is True, execute them in a sqlite3 in-memory DB and return + (conn, {"apartments": text, "buildings": text}). Otherwise return (None, texts). + """ + infra_dir = infra_dir or find_infrastructure_dir() + if infra_dir is None: + raise FileNotFoundError("infrastructure directory not found") + + # Support different possible locations and filenames: some projects keep SQL under + # infrastructure/init and filenames may be misspelled (e.g. 'apartmentes.sql'). + candidates_dirs = [infra_dir, infra_dir / "init"] + + def find_file(names): + for d in candidates_dirs: + if d is None: + continue + for n in names: + p = d / n + if p.is_file(): + return p + return None + + apt_names = ["apartments.sql", "apartmentes.sql", "apartamente.sql"] + bld_names = ["buildings.sql", "cladiri.sql", "buildings.sql"] + + apt_path = find_file(apt_names) + bld_path = find_file(bld_names) + + if apt_path is None: + raise FileNotFoundError(f"None of {apt_names} found under {candidates_dirs}") + if bld_path is None: + raise FileNotFoundError(f"None of {bld_names} found under {candidates_dirs}") + + apartments_sql = apt_path.read_text(encoding="utf-8") + buildings_sql = bld_path.read_text(encoding="utf-8") + + texts = {"apartments": apartments_sql, "buildings": buildings_sql} + + if not into_memory_db: + return None, texts + + conn = sqlite3.connect(":memory:") + # Run buildings first in case apartments reference buildings (FKs) + load_sql_into_sqlite(buildings_sql, conn) + load_sql_into_sqlite(apartments_sql, conn) + return conn, texts + + +def parse_apartments_sql(sql_text: str): + """Parse apartments.sql content and return a list of listings with + keys 'full_address' and 'expected_type' (tip_imobil). + + This looks for rows where building_id is a subselect with the building + address and extracts that address plus the tip_imobil field. + """ + listings = [] + # Regex: find occurrences like ( (SELECT id FROM buildings WHERE address='ADDR'), (SELECT id FROM users ...), 'tip_imobil', + pattern = re.compile(r"\(\s*\(SELECT\s+id\s+FROM\s+buildings\s+WHERE\s+address\s*=\s*'(?P[^']+)'\)\s*,\s*\(SELECT[\s\S]*?\)\s*,\s*'(?P[^']+)'", re.IGNORECASE) + for m in pattern.finditer(sql_text): + addr = m.group('addr').strip() + tip = m.group('type').strip() + listings.append({'full_address': addr, 'expected_type': tip}) + return listings + + +def find_rentai_root(start: Optional[Path] = None) -> Optional[Path]: + """Find the ancestor directory named 'rentAI'. Returns Path or None.""" + if start is None: + start = Path(__file__).resolve() + for p in [start] + list(start.parents): + if p.name == 'rentAI': + return p + return None + + +def mock_database_retrieval(): + """ + Simulates retrieving the address and expected type from your SQL data. + + In a real app, you would use 'sqlite3' or 'psycopg2' to run a query: + SELECT B.address, A.tip_imobil FROM buildings B JOIN apartments A ON ... + """ + # Using a sample of addresses from your apartments.sql file + return [ + { + "full_address": "Calea Victoriei 125, București, România", + "expected_type": "apartament" + }, + { + "full_address": "Strada Traian 212, București, România", + "expected_type": "apartament" + }, + { + "full_address": "Bd. Iuliu Maniu 58, București, România", + "expected_type": "garsoniera" + }, + # Example of a known commercial location that should fail the check + { + "full_address": "Bulevardul Vasile Milea 4, București, România", # Near AFI Cotroceni Mall + "expected_type": "apartament" + }, + ] + +# ====================================================================== +# --- 2. VALIDATION STEP 1: ADDRESS VALIDITY & PLACE ID RETRIEVAL --- +# ====================================================================== + +def get_place_id_and_validity(address, api_key): + """ + Uses the AI Geocoding API to check address validity and get a Place ID. + (This is the first step of validation) + """ + street_address = address.split(',')[0].strip() + params = { + # Textul de căutat (adresa stradală) + "text": street_address, + "apiKey": api_key, + + # SOLUȚIA FINALĂ: Parametri de adresă structurată pentru restricție + "city": "Bucuresti", + "country": "Romania" # sau countrycode: ro + } + + try: + response = requests.get(GEOCODING_ENDPOINT, params=params) + + # Secțiunea de depanare: Afișează eroarea exactă returnată de Geoapify + if not response.ok: + error_status = f"HTTP {response.status_code}" + error_body = response.text[:150] + "..." if response.text else "No response body." + print(f"!!! DEBUG: API FAILED. Status: {error_status}. Response: {error_body}") + response.raise_for_status() # Aceasta va genera RequestException + + data = response.json() + except requests.exceptions.RequestException as e: + # Aici se ajunge daca eroarea e 401, 429, sau 400 + return {"status": "ERROR: Geocoding API Failed", "place_id": None} + + # ... (restul funcției) ... + + if not data.get('features'): + return {"status": "FAILURE: Address Not Found", "place_id": None} + + best_match = data['features'][0] + properties = best_match.get('properties', {}) + + # Extract the AI-powered confidence rank (Geoapify example) + confidence_rank = properties.get('rank', {}).get('confidence', 0.0) + + # Many Geocoding APIs return a unique ID which is often required for Place Details + # (Using 'place_id' as a generic name) + place_id = properties.get('place_id') + + if confidence_rank >= 0.75: + return {"status": f"SUCCESS: Validated with {confidence_rank} Confidence", "place_id": place_id} + else: + return {"status": f"WARNING: Low Confidence Match ({confidence_rank:.2f})", "place_id": place_id} + +# ====================================================================== +# --- 3. VALIDATION STEP 2: BUILDING TYPE CHECK --- +# ====================================================================== + +def check_building_type(place_id, expected_type, api_key): + """ + Uses the Place Details API to check the building's official type/categories. + (This is the second step of validation) + + Geoapify categories used for checking: + RESIDENTIAL: 'building.residential', 'accommodation.apartment', 'accommodation.house' + COMMERCIAL: 'commercial', 'office' + """ + if not place_id: + return "N/A - Cannot check type without Place ID" + + # --- START OF MODIFICATION --- + + # Geoapify categories that confirm a residential building (apartment or house) + # Using 'building.residential' for the building type itself, and 'accommodation.apartment' + # and 'accommodation.house' for typical POIs that might represent the building. + RESIDENTIAL_CONFIRM_TYPES = [ + "building.residential", + "accommodation.apartment", + "accommodation.house", + "accommodation.residence" + ] + + # Geoapify top-level categories that strongly suggest a commercial mismatch + COMMERCIAL_FLAG_TYPES = [ + "commercial", # e.g., shopping center, retail + "office", # e.g., office building + "catering", # e.g., restaurants, cafes, bars + "tourism", # e.g., attractions, sights + "sport", # e.g., gyms, stadiums + "healthcare", # e.g., hospitals, clinics + "education" # e.g., schools, universities + ] + + # --- MOCK API CALL START --- + + # In a real script, you would query the Place Details API here using 'place_id': + # You would also request the 'categories' feature from the Place Details API. + # params = {"id": place_id, "apiKey": api_key, "features": "categories"} + # response = requests.get(PLACE_DETAILS_ENDPOINT, params=params) + # data = response.json() + # categories = data.get('features', [{}])[0].get('properties', {}).get('categories', []) + + # For demonstration, we use MOCK DATA based on the Place ID's expected outcome: + if "Vasile Milea 4" in place_id or "AFI" in place_id: + # Mock data for AFI Cotroceni (A Mall) + categories = ["commercial.shopping_mall", "commercial", "retail"] + elif "Fabrica de Glucoză 9" in place_id: + # Mock for new residential complex with multiple types + categories = ["building.residential", "accommodation.apartment", "commercial.supermarket"] + else: + # Mock data for a typical residential building + categories = ["building.residential", "accommodation.apartment", "residential"] + # --- MOCK API CALL END --- + + # --- Validation Logic --- + + # Check for confirmed residential type (if ANY category matches the list) + if any(c in categories for c in RESIDENTIAL_CONFIRM_TYPES): + return "SUCCESS: Confirmed Residential Building." + + # Check for commercial mismatch (if ANY top-level category matches the flag list) + # We check if the *start* of any category matches a commercial flag + if any(any(c.startswith(flag) for c in categories) for flag in COMMERCIAL_FLAG_TYPES): + # Find the most specific commercial category found for a better error message + flagged_category = next((c for c in categories if any(c.startswith(flag) for flag in COMMERCIAL_FLAG_TYPES)), "N/A") + return f"FAILURE: Address resolves to a COMMERCIAL type ({flagged_category})." + + return "WARNING: Type classification inconclusive. No explicit residential or commercial flags found." + +# ====================================================================== +# --- MAIN EXECUTION --- +# ====================================================================== + +if __name__ == "__main__": + + if GEOCODING_API_KEY == "17ef1e91c39e4aad80e6dcde4c6774f7" or PLACE_DETAILS_API_KEY == "5048e3524f4e426c9603e7438493aa03": + print("!!! WARNING: Please update the API keys in the CONFIGURATION section to run this script with real data. !!!") + print("Running with mock data based on the full addresses.") + + # Retrieve all listings by parsing the SQL file located at the hardcoded path + # as requested: /mnt/d/adobeHack/rentAI/project/infrastructure/init/apartments.sql + sql_path = find_rentai_root() / Path("project/infrastructure/init/apartments.sql") + try: + sql_text = sql_path.read_text(encoding="utf-8") + # Modified the addresses to include the city for better geocoding results + listings = [ + {'full_address': f"{l['full_address']}, București, România", 'expected_type': l['expected_type']} + for l in parse_apartments_sql(sql_text) + ] + except Exception as e: + print(f"Error reading/parsing apartments.sql: {e}") + # Fall back to mock data + listings = mock_database_retrieval() + + validation_results = [] + + for listing in listings: + address = listing['full_address'] + expected = listing['expected_type'] + + print("\n========================================================") + print(f"Listing: {address} (Expected: {expected})") + + # 1. VALIDATION STEP 1: Address Validity + validity_check = get_place_id_and_validity(address, GEOCODING_API_KEY) + print(f"Status 1 (Validity): {validity_check['status']}") + + place_id = validity_check['place_id'] + + # Only proceed to type check if the address was found + if validity_check['status'].startswith("SUCCESS"): + + # 2. VALIDATION STEP 2: Building Type + type_check = check_building_type(place_id, expected, PLACE_DETAILS_API_KEY) + print(f"Status 2 (Type Check): {type_check}") + + # Final Summary + print("========================================================") \ No newline at end of file