#!/usr/bin/env python3 """ Generate SQL migration file to ingest Jordan (Amman, Zarqa, Irbid) checkpoints CSV into places_jordan and refresh the unified_search_index. """ import csv import sys import os import math from collections import defaultdict def haversine(lat1, lon1, lat2, lon2): R = 6371000 phi1, phi2 = math.radians(lat1), math.radians(lat2) dphi = math.radians(lat2 - lat1) dlam = math.radians(lon2 - lon1) a = math.sin(dphi/2)**2 + math.cos(phi1)*math.cos(phi2)*math.sin(dlam/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() s = s.replace('أ', 'ا').replace('إ', 'ا').replace('آ', 'ا').replace('ة', 'ه').replace('ى', 'ي') return ' '.join(s.split()).lower() def clean_str(val): if not val: return 'NULL' s = str(val).strip().replace("'", "''") return f"'{s}'" def calculate_popularity(row): reviews = 0 try: reviews = int(float(row.get('reviews_count') or 0)) except Exception: reviews = 0 rating = 0.0 try: rating = float(row.get('rating') or 0) except Exception: rating = 0.0 score = int(reviews * 0.2 + rating * 5) cat = row.get('category_queried', '') if any(k in cat for k in ['مستشفى', 'جامعة', 'كلية', 'مركز صحي', 'طوارئ', 'دفاع مدني']): score = max(score, 90) elif any(k in cat for k in ['صيدلية', 'بنك', 'صراف آلي', 'سوبر ماركت', 'محطة']): score = max(score, 60) elif any(k in cat for k in ['مدرسة', 'روضة', 'عيادة', 'بلدية', 'دائرة']): score = max(score, 50) return min(100, max(10, score)) def main(): csv_path = 'data/checkpoint_35242_records.csv' out_sql_path = 'infrastructure/sql/18_ingest_amman_zarqa_checkpoint.sql' if not os.path.exists(csv_path): print(f"Error: {csv_path} not found.") sys.exit(1) print(f"Reading {csv_path}...") records = [] with open(csv_path, 'r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) for row in reader: name = row.get('name', '').strip() if not name: continue try: lat = float(row['latitude']) lng = float(row['longitude']) except Exception: continue if not (29.15 <= lat <= 33.45 and 34.85 <= lng <= 39.35): 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 'نقطة اهتمام في الأردن' pop_score = calculate_popularity(row) 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 }) # Exact dedup seen_exact = {} seen_urls = {} for r in records: k = (r['norm_name'], round(r['lat'], 5), round(r['lng'], 5)) u = r['maps_url'] if k in seen_exact or (u and u in seen_urls): continue seen_exact[k] = r if u: seen_urls[u] = r step1 = list(seen_exact.values()) # Spatial near-duplicate dedup (<50m) by_norm = defaultdict(list) for r in step1: by_norm[r['norm_name']].append(r) final_records = [] for norm_name, items in by_norm.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: if haversine(candidate['lat'], candidate['lng'], existing['lat'], existing['lng']) < 50: is_dup = True break if not is_dup: kept.append(candidate) final_records.extend(kept) print(f"Total valid unique records for SQL ingestion: {len(final_records)}") source_tag = 'checkpoint_35242_amman_zarqa' with open(out_sql_path, 'w', encoding='utf-8') as out: out.write("-- ============================================================================\n") out.write("-- MIGRATION 18: INGEST AMMAN & ZARQA CHECKPOINT DATASET (35,000+ PLACES)\n") out.write("-- Ingests into places_jordan and refreshes unified_search_index\n") out.write("-- ============================================================================\n\n") out.write("BEGIN;\n\n") out.write(f"-- Remove previous imports with same source to ensure idempotency\n") out.write(f"DELETE FROM places_jordan WHERE source = '{source_tag}';\n\n") out.write("ALTER TABLE places_jordan DISABLE TRIGGER trg_sync_place_location_jordan;\n\n") batch_size = 1000 for i in range(0, len(final_records), batch_size): batch = final_records[i:i + batch_size] out.write("INSERT INTO places_jordan (\n") out.write(" name, name_ar, latitude, longitude, category, city, neighbourhood,\n") out.write(" address, description, popularity_score, source, location\n") out.write(") VALUES\n") value_lines = [] for r in batch: c_name = clean_str(r['name']) c_cat = clean_str(r['category']) c_city = clean_str(r['city']) c_area = clean_str(r['neighbourhood']) c_addr = clean_str(r['address']) c_desc = clean_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}, '{source_tag}', " f"ST_SetSRID(ST_MakePoint({lng:.7f}, {lat:.7f}), 4326))") value_lines.append(line) out.write(",\n".join(value_lines)) out.write(";\n\n") out.write("ALTER TABLE places_jordan ENABLE TRIGGER trg_sync_place_location_jordan;\n\n") out.write(f"UPDATE places_jordan SET governorate_id = 137 WHERE source = '{source_tag}' AND city = 'عمان';\n") out.write(f"UPDATE places_jordan SET governorate_id = 5 WHERE source = '{source_tag}' AND city = 'الزرقاء';\n") out.write(f"UPDATE places_jordan SET governorate_id = 184 WHERE source = '{source_tag}' AND city = 'إربد';\n\n") out.write("COMMIT;\n\n") out.write("-- Refresh Unified Search Index to instantly index all new Jordanian places\n") out.write("REFRESH MATERIALIZED VIEW CONCURRENTLY unified_search_index;\n\n") out.write("ANALYZE places_jordan;\n") print(f"✅ Generated SQL file: {out_sql_path} ({os.path.getsize(out_sql_path) / (1024*1024):.2f} MB)") if __name__ == '__main__': main()