address validation script using geoapify AI

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<addr>[^']+)'\)\s*,\s*\(SELECT[\s\S]*?\)\s*,\s*'(?P<type>[^']+)'", 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("========================================================")