SafeVixAI/SafeVixAI-Dataset-Hub

1from __future__ import annotations2 3import argparse4import csv5import json6import urllib.parse7import urllib.request8from pathlib import Path9from typing import Iterable10 11 12DEFAULT_ENDPOINTS = (13    'https://overpass-api.de/api/interpreter',14    'https://overpass.kumi.systems/api/interpreter',15    'https://lz4.overpass-api.de/api/interpreter',16)17DEFAULT_HEADERS = {18    'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8',19    'User-Agent': 'SafeVixAI bootstrap scripts/1.0',20}21CSV_COLUMNS = [22    'osm_id',23    'osm_type',24    'name',25    'lat',26    'lon',27    'phone',28    'type',29    'city',30    'state',31    'address',32    'opening_hours',33    'website',34    'source',35]36 37 38def build_arg_parser(description: str, default_output: Path) -> argparse.ArgumentParser:39    parser = argparse.ArgumentParser(description=description)40    parser.add_argument(41        '--output',42        type=Path,43        default=default_output,44        help=f'CSV path to write. Defaults to {default_output}',45    )46    parser.add_argument(47        '--endpoint',48        help='Optional Overpass endpoint override. Defaults to a built-in fallback list.',49    )50    parser.add_argument(51        '--timeout',52        type=int,53        default=300,54        help='HTTP timeout in seconds. Defaults to 300.',55    )56    return parser57 58 59def build_india_query(selectors: Iterable[str], *, timeout: int) -> str:60    joined_selectors = '\n  '.join(selector.strip() for selector in selectors if selector.strip())61    return (62        f'[out:json][timeout:{timeout}];\n'63        'area["ISO3166-1"="IN"][admin_level=2]->.searchArea;\n'64        '(\n'65        f'  {joined_selectors}\n'66        ');\n'67        'out center tags;'68    )69 70 71def fetch_elements(query: str, *, endpoint: str | None, timeout: int, **kwargs) -> list[dict]:72    payload = urllib.parse.urlencode({'data': query}).encode('utf-8')73    endpoints = [endpoint] if endpoint else list(DEFAULT_ENDPOINTS)74    last_error: Exception | None = None75 76    for url in endpoints:77        request = urllib.request.Request(url, data=payload, headers=DEFAULT_HEADERS, method='POST')78        try:79            with urllib.request.urlopen(request, timeout=timeout) as response:80                decoded = response.read().decode('utf-8')81            data = json.loads(decoded)82            return list(data.get('elements', []))83        except Exception as exc:  # pragma: no cover - network failure path84            last_error = exc85 86    raise SystemExit(f'Unable to fetch data from Overpass. Last error: {last_error}')87 88 89def extract_point(element: dict) -> tuple[float | None, float | None]:90    if 'lat' in element and 'lon' in element:91        return float(element['lat']), float(element['lon'])92 93    center = element.get('center') or {}94    if 'lat' in center and 'lon' in center:95        return float(center['lat']), float(center['lon'])96 97    return None, None98 99 100def compose_address(tags: dict[str, str]) -> str:101    parts = [102        tags.get('addr:housenumber'),103        tags.get('addr:street'),104        tags.get('addr:suburb'),105        tags.get('addr:city') or tags.get('addr:town') or tags.get('addr:village'),106        tags.get('addr:state'),107    ]108    return ', '.join(part for part in parts if part)109 110 111def normalize_row(element: dict, *, default_type: str, fallback_name: str, **kwargs) -> dict | None:112    lat, lon = extract_point(element)113    if lat is None or lon is None:114        return None115 116    tags = element.get('tags', {})117    amenity_type = tags.get('amenity') or tags.get('healthcare') or tags.get('emergency') or default_type118    return {119        'osm_id': str(element.get('id', '')),120        'osm_type': str(element.get('type', '')),121        'name': tags.get('name') or fallback_name,122        'lat': f'{lat:.6f}',123        'lon': f'{lon:.6f}',124        'phone': tags.get('phone') or tags.get('contact:phone') or tags.get('emergency:phone') or '',125        'type': amenity_type,126        'city': tags.get('addr:city') or tags.get('addr:town') or tags.get('addr:village') or '',127        'state': tags.get('addr:state') or '',128        'address': compose_address(tags),129        'opening_hours': tags.get('opening_hours') or '',130        'website': tags.get('website') or tags.get('contact:website') or '',131        'source': 'overpass',132    }133 134 135def dedupe_rows(rows: Iterable[dict]) -> list[dict]:136    seen: set[tuple[str, str, str, str]] = set()137    deduped: list[dict] = []138    for row in rows:139        key = (140            row.get('name', '').strip().lower(),141            row.get('type', '').strip().lower(),142            row.get('lat', ''),143            row.get('lon', ''),144        )145        if key in seen:146            continue147        seen.add(key)148        deduped.append(row)149    deduped.sort(key=lambda item: (item['state'], item['city'], item['name']))150    return deduped151 152 153def write_rows(path: Path, rows: Iterable[dict]) -> int:154    path.parent.mkdir(parents=True, exist_ok=True)155    materialized = dedupe_rows(rows)156    with path.open('w', newline='', encoding='utf-8') as handle:157        writer = csv.DictWriter(handle, fieldnames=CSV_COLUMNS)158        writer.writeheader()159        writer.writerows(materialized)160    return len(materialized)161