|
|
|
|
|
import requests |
|
import time |
|
from geopy.geocoders import Nominatim |
|
from .logging_config import logger |
|
|
|
geocoder = Nominatim(user_agent="indian_property_verifier", timeout=10) |
|
|
|
def verify_address(data): |
|
try: |
|
address_results = { |
|
'address_exists': False, |
|
'pincode_valid': False, |
|
'city_state_match': False, |
|
'coordinates_match': False, |
|
'confidence': 0.0, |
|
'issues': [], |
|
'verification_score': 0.0 |
|
} |
|
|
|
|
|
zip_code = str(data.get('zip', '')).strip() |
|
city = str(data.get('city', '')).strip() |
|
state = str(data.get('state', '')).strip() |
|
address = str(data.get('address', '')).strip() |
|
country = str(data.get('country', 'India')).strip() |
|
latitude = data.get('latitude', None) |
|
longitude = data.get('longitude', None) |
|
|
|
|
|
basic_score = 0.0 |
|
if zip_code: |
|
basic_score += 0.25 |
|
if city: |
|
basic_score += 0.25 |
|
if state: |
|
basic_score += 0.25 |
|
if address: |
|
basic_score += 0.15 |
|
if latitude and longitude: |
|
basic_score += 0.10 |
|
|
|
|
|
if zip_code: |
|
try: |
|
response = requests.get(f"https://api.postalpincode.in/pincode/{zip_code}", timeout=5) |
|
if response.status_code == 200: |
|
pin_data = response.json() |
|
if pin_data[0]['Status'] == 'Success': |
|
address_results['pincode_valid'] = True |
|
post_offices = pin_data[0]['PostOffice'] |
|
cities = {po['Name'].lower() for po in post_offices} |
|
states = {po['State'].lower() for po in post_offices} |
|
if city.lower() in cities or state.lower() in states: |
|
address_results['city_state_match'] = True |
|
else: |
|
address_results['issues'].append("City/state may not match pincode") |
|
else: |
|
address_results['issues'].append(f"Invalid pincode: {zip_code}") |
|
else: |
|
address_results['issues'].append("Pincode API error") |
|
except Exception as e: |
|
logger.error(f"Pincode API error: {str(e)}") |
|
|
|
if zip_code and len(zip_code) == 6 and zip_code.isdigit(): |
|
address_results['pincode_valid'] = True |
|
address_results['issues'].append("Pincode validation failed (API error)") |
|
else: |
|
address_results['issues'].append("Pincode validation failed") |
|
|
|
|
|
full_address = ', '.join(filter(None, [address, city, state, country, zip_code])) |
|
geocoding_success = False |
|
|
|
for attempt in range(2): |
|
try: |
|
location = geocoder.geocode(full_address) |
|
if location: |
|
address_results['address_exists'] = True |
|
address_results['confidence'] = 0.9 |
|
geocoding_success = True |
|
|
|
if latitude and longitude: |
|
try: |
|
provided_coords = (float(latitude), float(longitude)) |
|
geocoded_coords = (location.latitude, location.longitude) |
|
from geopy.distance import distance |
|
dist = distance(provided_coords, geocoded_coords).km |
|
address_results['coordinates_match'] = dist < 2.0 |
|
if not address_results['coordinates_match']: |
|
address_results['issues'].append(f"Coordinates {dist:.2f}km off") |
|
except Exception as e: |
|
logger.warning(f"Invalid coordinates: {str(e)}") |
|
address_results['issues'].append("Invalid coordinates") |
|
break |
|
time.sleep(1) |
|
except Exception as e: |
|
logger.error(f"Geocoding error on attempt {attempt + 1}: {str(e)}") |
|
time.sleep(1) |
|
|
|
if not geocoding_success: |
|
|
|
if address and city and state: |
|
address_results['address_exists'] = True |
|
address_results['confidence'] = 0.8 |
|
address_results['issues'].append("Address geocoding failed (using fallback validation)") |
|
elif city and state: |
|
address_results['address_exists'] = True |
|
address_results['confidence'] = 0.7 |
|
address_results['issues'].append("Address geocoding failed (using fallback validation)") |
|
else: |
|
address_results['issues'].append("Address geocoding failed") |
|
|
|
|
|
try: |
|
verification_points = ( |
|
float(address_results['address_exists']) * 0.4 + |
|
float(address_results['pincode_valid']) * 0.3 + |
|
float(address_results['city_state_match']) * 0.2 + |
|
float(address_results['coordinates_match']) * 0.1 |
|
) |
|
|
|
|
|
if verification_points == 0.0 and basic_score > 0.0: |
|
verification_points = basic_score * 0.8 |
|
|
|
except Exception as e: |
|
logger.warning(f"Error calculating verification points: {str(e)}") |
|
verification_points = basic_score * 0.8 |
|
|
|
|
|
if verification_points == 0.0 and (zip_code or city or state or address): |
|
verification_points = 0.4 |
|
|
|
|
|
if zip_code and city and state: |
|
verification_points = min(1.0, verification_points + 0.1) |
|
elif city and state: |
|
verification_points = min(1.0, verification_points + 0.05) |
|
|
|
|
|
verification_score = min(100.0, verification_points * 100) |
|
address_results['verification_score'] = verification_score |
|
|
|
return address_results |
|
except Exception as e: |
|
logger.error(f"Error verifying address: {str(e)}") |
|
|
|
return { |
|
'address_exists': False, |
|
'pincode_valid': False, |
|
'city_state_match': False, |
|
'coordinates_match': False, |
|
'confidence': 0.0, |
|
'issues': [f"Address verification error: {str(e)}"], |
|
'verification_score': 30.0 |
|
} |
|
|