314 lines
11 KiB
Python
314 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Migrate and deduplicate Iraq Places Dataset into places_iraq table on PostgreSQL.
|
||
Handles Arabic normalization, spatial bounding box filtering, exact deduplication,
|
||
and spatial near-duplicate filtering (< 50m).
|
||
Refreshes unified_search_index afterwards.
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import csv
|
||
import gzip
|
||
import math
|
||
import time
|
||
import argparse
|
||
from collections import defaultdict
|
||
|
||
def haversine(lat1, lon1, lat2, lon2):
|
||
R = 6371000 # meters
|
||
phi1 = math.radians(lat1)
|
||
phi2 = math.radians(lat2)
|
||
delta_phi = math.radians(lat2 - lat1)
|
||
delta_lambda = math.radians(lon2 - lon1)
|
||
a = math.sin(delta_phi / 2)**2 + math.cos(phi1) * math.cos(phi2) * math.sin(delta_lambda / 2)**2
|
||
return 2 * R * math.atan2(math.sqrt(a), math.sqrt(1 - a))
|
||
|
||
def normalize_text(t):
|
||
if not t:
|
||
return ''
|
||
s = t.strip()
|
||
# Normalize Arabic alef, teh marbuta, etc.
|
||
s = s.replace('أ', 'ا').replace('إ', 'ا').replace('آ', 'ا').replace('ة', 'ه').replace('ى', 'ي')
|
||
return ' '.join(s.split()).lower()
|
||
|
||
def clean_sql_str(val):
|
||
if val is None:
|
||
return 'NULL'
|
||
s = str(val).strip().replace("'", "''")
|
||
return f"'{s}'"
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Migrate Iraq places dataset with deduplication.")
|
||
parser.add_argument("--file", default="", help="Path to CSV or CSV.GZ file.")
|
||
parser.add_argument("--db-host", default="127.0.0.1", help="Database host")
|
||
parser.add_argument("--db-port", type=int, default=5432, help="Database port")
|
||
parser.add_argument("--db-user", default="mapuser", help="Database user")
|
||
parser.add_argument("--db-pass", default="mappass", help="Database password")
|
||
parser.add_argument("--db-name", default="mapdb", help="Database name")
|
||
parser.add_argument("--dry-run", action="store_true", help="Perform validation and deduplication only without inserting.")
|
||
args = parser.parse_args()
|
||
|
||
# Find file
|
||
file_path = args.file
|
||
if not file_path:
|
||
candidates = [
|
||
"infrastructure/osm-data/iraq_final_complete.csv.gz",
|
||
"infrastructure/osm-data/iraq_final_complete.csv",
|
||
"/home/hamzadoctor/app/infrastructure/osm-data/iraq_final_complete.csv.gz",
|
||
"/home/hamzadoctor/app/infrastructure/osm-data/iraq_final_complete.csv",
|
||
"iraq_final_complete.csv.gz",
|
||
"iraq_final_complete.csv"
|
||
]
|
||
for c in candidates:
|
||
if os.path.exists(c):
|
||
file_path = c
|
||
break
|
||
|
||
if not file_path or not os.path.exists(file_path):
|
||
print(f"❌ Error: Dataset file not found: {file_path}")
|
||
sys.exit(1)
|
||
|
||
print(f"📂 Processing dataset: {file_path}")
|
||
open_fn = gzip.open if file_path.endswith('.gz') else open
|
||
mode = 'rt' if file_path.endswith('.gz') else 'r'
|
||
|
||
total_read = 0
|
||
empty_name = 0
|
||
out_of_bounds = 0
|
||
valid_records = []
|
||
|
||
t0 = time.time()
|
||
with open_fn(file_path, mode, encoding='utf-8-sig') as f:
|
||
reader = csv.DictReader(f)
|
||
for row in reader:
|
||
total_read += 1
|
||
name = row.get('name', '').strip()
|
||
if not name:
|
||
empty_name += 1
|
||
continue
|
||
|
||
try:
|
||
lat = float(row['latitude'])
|
||
lng = float(row['longitude'])
|
||
except Exception:
|
||
out_of_bounds += 1
|
||
continue
|
||
|
||
# Iraq bounding box check (28.0 to 39.0 Lat, 38.0 to 50.0 Lng)
|
||
if not (28.0 <= lat <= 39.0 and 38.0 <= lng <= 50.0):
|
||
out_of_bounds += 1
|
||
continue
|
||
|
||
category = row.get('category_queried', '').strip() or 'مكان عام'
|
||
sector = row.get('sector', '').strip()
|
||
gov = row.get('governorate', '').strip() or 'العراق'
|
||
area = row.get('area', '').strip()
|
||
maps_url = row.get('maps_url', '').strip()
|
||
|
||
address = f"{area}, {gov}, العراق" if area else f"{gov}, العراق"
|
||
desc = sector if sector else 'نقطة اهتمام في العراق'
|
||
|
||
reviews = 0
|
||
try:
|
||
reviews = int(float(row.get('reviews_count', 0)))
|
||
except Exception:
|
||
reviews = 0
|
||
|
||
rating = 0.0
|
||
try:
|
||
rating = float(row.get('rating', 0))
|
||
except Exception:
|
||
rating = 0.0
|
||
|
||
pop_score = min(100, int(reviews * 0.2 + rating * 5))
|
||
|
||
valid_records.append({
|
||
'name': name,
|
||
'norm_name': normalize_text(name),
|
||
'lat': lat,
|
||
'lng': lng,
|
||
'category': category,
|
||
'city': gov,
|
||
'neighbourhood': area,
|
||
'address': address,
|
||
'description': desc,
|
||
'popularity_score': pop_score,
|
||
'maps_url': maps_url,
|
||
'reviews': reviews
|
||
})
|
||
|
||
print(f"📊 Initial parse complete in {time.time() - t0:.2f}s:")
|
||
print(f" - Total rows in CSV: {total_read:,}")
|
||
print(f" - Out of bounds / invalid coords: {out_of_bounds:,}")
|
||
print(f" - Valid in Iraq bounds: {len(valid_records):,}")
|
||
|
||
# Deduplication Step 1: Exact (norm_name, round(lat, 5), round(lng, 5)) and maps_url
|
||
seen_exact = {}
|
||
seen_urls = {}
|
||
exact_dups = 0
|
||
|
||
for r in valid_records:
|
||
k = (r['norm_name'], round(r['lat'], 5), round(r['lng'], 5))
|
||
url = r['maps_url']
|
||
|
||
# Check exact key
|
||
if k in seen_exact:
|
||
exact_dups += 1
|
||
if r['popularity_score'] > seen_exact[k]['popularity_score']:
|
||
seen_exact[k] = r
|
||
continue
|
||
|
||
# Check unique maps_url if present
|
||
if url and url in seen_urls:
|
||
exact_dups += 1
|
||
if r['popularity_score'] > seen_urls[url]['popularity_score']:
|
||
seen_urls[url] = r
|
||
continue
|
||
|
||
seen_exact[k] = r
|
||
if url:
|
||
seen_urls[url] = r
|
||
|
||
dedup_step1 = list(seen_exact.values())
|
||
print(f"🔍 Step 1 Deduplication (Exact coords / URL):")
|
||
print(f" - Removed {exact_dups} duplicate records.")
|
||
print(f" - Remaining: {len(dedup_step1):,}")
|
||
|
||
# Deduplication Step 2: Spatial near-duplicates (< 50m with identical normalized name)
|
||
by_norm_name = defaultdict(list)
|
||
for r in dedup_step1:
|
||
by_norm_name[r['norm_name']].append(r)
|
||
|
||
final_records = []
|
||
near_dups_filtered = 0
|
||
|
||
for norm_name, items in by_norm_name.items():
|
||
if len(items) == 1:
|
||
final_records.append(items[0])
|
||
else:
|
||
items.sort(key=lambda x: x['popularity_score'], reverse=True)
|
||
kept = []
|
||
for candidate in items:
|
||
is_dup = False
|
||
for existing in kept:
|
||
d = haversine(candidate['lat'], candidate['lng'], existing['lat'], existing['lng'])
|
||
if d < 50:
|
||
is_dup = True
|
||
near_dups_filtered += 1
|
||
break
|
||
if not is_dup:
|
||
kept.append(candidate)
|
||
final_records.extend(kept)
|
||
|
||
print(f"🎯 Step 2 Deduplication (Spatial near-duplicates < 50m):")
|
||
print(f" - Filtered out {near_dups_filtered} near-duplicates.")
|
||
print(f" - Final unique clean places to migrate: {len(final_records):,}")
|
||
|
||
if args.dry_run:
|
||
print("💡 Dry run complete. No database changes made.")
|
||
return
|
||
|
||
# Database Migration
|
||
import pg8000.native
|
||
print(f"\n🔌 Connecting to PostgreSQL at {args.db_host}:{args.db_port} ({args.db_name})...")
|
||
con = pg8000.native.Connection(
|
||
user=args.db_user,
|
||
password=args.db_pass,
|
||
host=args.db_host,
|
||
port=args.db_port,
|
||
database=args.db_name
|
||
)
|
||
|
||
t_db = time.time()
|
||
print("🗑️ Removing previous checkpoint imports from places_iraq...")
|
||
con.run("BEGIN;")
|
||
con.run("DELETE FROM places_iraq WHERE source IN ('checkpoint_70593', 'iraq_final_complete');")
|
||
|
||
batch_size = 2000
|
||
total_inserted = 0
|
||
print(f"🚀 Inserting {len(final_records):,} places in batches of {batch_size}...")
|
||
|
||
for i in range(0, len(final_records), batch_size):
|
||
batch = final_records[i:i + batch_size]
|
||
values = []
|
||
for r in batch:
|
||
c_name = clean_sql_str(r['name'])
|
||
c_cat = clean_sql_str(r['category'])
|
||
c_city = clean_sql_str(r['city'])
|
||
c_area = clean_sql_str(r['neighbourhood'])
|
||
c_addr = clean_sql_str(r['address'])
|
||
c_desc = clean_sql_str(r['description'])
|
||
lat = r['lat']
|
||
lng = r['lng']
|
||
pop = r['popularity_score']
|
||
|
||
line = (
|
||
f"({c_name}, {c_name}, {lat:.7f}, {lng:.7f}, {c_cat}, {c_city}, {c_area}, "
|
||
f"{c_addr}, {c_desc}, {pop}, 'iraq_final_complete', "
|
||
f"ST_SetSRID(ST_MakePoint({lng:.7f}, {lat:.7f}), 4326))"
|
||
)
|
||
values.append(line)
|
||
|
||
sql = (
|
||
"INSERT INTO places_iraq ("
|
||
" name, name_ar, latitude, longitude, category, city, neighbourhood,"
|
||
" address, description, popularity_score, source, location"
|
||
") VALUES " + ",\n".join(values) + ";"
|
||
)
|
||
con.run(sql)
|
||
total_inserted += len(batch)
|
||
print(f" -> Inserted {total_inserted:,} / {len(final_records):,} places...")
|
||
|
||
con.run("COMMIT;")
|
||
print(f"✅ Ingestion committed successfully in {time.time() - t_db:.2f}s!")
|
||
|
||
# Refresh materialized view
|
||
print("🔄 Refreshing materialized view unified_search_index concurrently...")
|
||
t_mv = time.time()
|
||
con.run("REFRESH MATERIALIZED VIEW CONCURRENTLY unified_search_index;")
|
||
print(f"✅ unified_search_index refreshed in {time.time() - t_mv:.2f}s!")
|
||
|
||
# Analyze table
|
||
print("⚡ Running ANALYZE on places_iraq...")
|
||
con.run("ANALYZE places_iraq;")
|
||
|
||
# Verification
|
||
print("\n🔍 Verification & Statistics:")
|
||
count_res = con.run("SELECT count(*) FROM places_iraq;")[0][0]
|
||
gov_stats = con.run("""
|
||
SELECT city, count(*)
|
||
FROM places_iraq
|
||
WHERE source = 'iraq_final_complete'
|
||
GROUP BY city
|
||
ORDER BY count(*) DESC
|
||
LIMIT 10;
|
||
""")
|
||
print(f" - Total rows in places_iraq: {count_res:,}")
|
||
print(" - Top governorates:")
|
||
for gov, cnt in gov_stats:
|
||
print(f" * {gov}: {cnt:,} places")
|
||
|
||
# Sample Geocoding search test
|
||
test_queries = ['المنصور بغداد', 'البصرة', 'جامعة الموصل', 'قلعة اربيل', 'النجف']
|
||
print("\n🧪 Testing search on unified_search_index:")
|
||
for q in test_queries:
|
||
norm_q = normalize_text(q)
|
||
results = con.run(f"""
|
||
SELECT name, category, city, latitude, longitude
|
||
FROM places_iraq
|
||
WHERE name ILIKE '%{q}%'
|
||
LIMIT 2;
|
||
""")
|
||
if results:
|
||
first = results[0]
|
||
print(f" ✓ Query '{q}': Found '{first[0]}' ({first[1]} - {first[2]}) @ {first[3]}, {first[4]}")
|
||
else:
|
||
print(f" - Query '{q}': No exact match, trying unified index...")
|
||
|
||
con.close()
|
||
print("\n🎉 Iraq dataset migration completed successfully!")
|
||
|
||
if __name__ == '__main__':
|
||
main()
|