import requests import time import sqlite3 import threading import queue from collections import deque from typing import List # -------------------------------- # Thread-Safe Leaky Bucket Rate Limiter # -------------------------------- class LeakyBucketRateLimiter: def __init__(self, max_rate: float = 15.0): self.max_rate = max_rate self.timestamps: deque = deque() self.lock = threading.Lock() self.window: float = 1.0 # Sliding window size in seconds def acquire(self) -> None: with self.lock: now = time.monotonic() # Remove timestamps outside the current window while self.timestamps and self.timestamps[0] <= now - self.window: self.timestamps.popleft() # If limit reached, sleep until the oldest request exits the window if len(self.timestamps) >= self.max_rate: sleep_time = 1.0 - (now - self.timestamps[0]) if sleep_time > 0: time.sleep(sleep_time) now = time.monotonic() while self.timestamps and self.timestamps[0] <= now - self.window: self.timestamps.popleft() self.timestamps.append(time.monotonic()) # -------------------------------- # Database Initialization # -------------------------------- def init_db(db_path: str = "uk_crimes.db") -> sqlite3.Connection: conn = sqlite3.connect(db_path) conn.execute(""" CREATE TABLE IF NOT EXISTS crimes ( id INTEGER PRIMARY KEY, category TEXT, month TEXT, latitude REAL, longitude REAL, street_name TEXT, outcome_category TEXT ) """) conn.commit() return conn # -------------------------------- # UK Grid Polygon Generator # -------------------------------- def generate_uk_polygons(lat_min: float = 49.5, lat_max: float = 61.0, lng_min: float = -13.0, lng_max: float = 5.0, step: float = 0.5) -> List[str]: polygons = [] lats = [l / 10.0 for l in range(int(lat_min * 10), int(lat_max * 10), int(step * 10))] lns = [ln / 10.0 for ln in range(int(lng_min * 10), int(lng_max * 10), int(step * 10))] for i in range(len(lats) - 1): for j in range(len(lns) - 1): poly = f"{lats[i]},{lns[j]}:{lats[i]},{lns[j+1]}:{lats[i+1]},{lns[j+1]}:{lats[i+1]},{lns[j]}" polygons.append(poly) return polygons # -------------------------------- # Producer Thread: Fetch & Queue # -------------------------------- def producer_worker(polygons: List[str], date: str, limiter: LeakyBucketRateLimiter, data_queue: queue.Queue): total = len(polygons) for i, poly in enumerate(polygons): # Skip polygons exceeding the 4094 character limit [1] if len(poly) > 4090: continue # Enforce strict 15 requests/second limit [1] limiter.acquire() try: response = requests.get( "https://data.police.uk/api/crimes-street/all-crime", params={"date": date, "poly": poly}, timeout=30 ) response.raise_for_status() data = response.json() # API returns 503 if a custom area contains >10,000 crimes [1] if not isinstance(data, list): print(f"[PRODUCER] Warning: Received non-list response for poly {poly[:30]}...") continue if not data: continue rows = [] for crime in data: loc = crime.get("location", {}) outcome = crime.get("outcome_status") or {} rows.append(( crime.get("id"), crime.get("category"), crime.get("month"), float(loc.get("latitude", 0)), float(loc.get("longitude", 0)), loc.get("street", {}).get("name"), outcome.get("category") )) # Push batch to memory queue (non-blocking if size allows) data_queue.put(rows) except requests.exceptions.RequestException as e: print(f"[PRODUCER] Network error for poly {poly[:30]}... | {e}") except Exception as e: print(f"[PRODUCER] Unexpected error: {e}") if (i + 1) % 100 == 0 or i == total - 1: print(f"[PRODUCER] Progress: {i+1}/{total} fetched | Queue size: {data_queue.qsize()}") # Signal completion to consumer data_queue.put(None) # -------------------------------- # Consumer Thread: Queue to DB # -------------------------------- def consumer_worker(data_queue: queue.Queue, conn: sqlite3.Connection): batch_size = 500 batch = [] while True: try: item = data_queue.get(timeout=5.0) except queue.Empty: # If queue is empty for 5s, assume production is done if not batch: break else: # Flush remaining batch conn.executemany( "INSERT OR IGNORE INTO crimes (id, category, month, latitude, longitude, street_name, outcome_category) VALUES (?, ?, ?, ?, ?, ?, ?)", batch ) conn.commit() batch = [] continue if item is None: # Flush last batch before exiting if batch: conn.executemany( "INSERT OR IGNORE INTO crimes (id, category, month, latitude, longitude, street_name, outcome_category) VALUES (?, ?, ?, ?, ?, ?, ?)", batch ) conn.commit() data_queue.task_done() break batch.extend(item) if len(batch) >= batch_size: conn.executemany( "INSERT OR IGNORE INTO crimes (id, category, month, latitude, longitude, street_name, outcome_category) VALUES (?, ?, ?, ?, ?, ?, ?)", batch ) conn.commit() batch = [] data_queue.task_done() # -------------------------------- # Main Orchestrator # -------------------------------- def main(): limiter = LeakyBucketRateLimiter(max_rate=15.0) conn = init_db("uk_crimes.db") data_queue = queue.Queue(maxsize=2000) # Backpressure to prevent memory overflow date = "2024-01" print("Generating UK coverage polygons...") polygons = generate_uk_polygons() print(f"Total regions to process: {len(polygons)}\n") # Start threads prod_thread = threading.Thread(target=producer_worker, args=(polygons, date, limiter, data_queue), daemon=True) cons_thread = threading.Thread(target=consumer_worker, args=(data_queue, conn), daemon=True) cons_thread.start() prod_thread.start() # Wait for producer to finish prod_thread.join() print("\n[MAIN] Producer finished. Waiting for consumer to drain queue...") # Wait for consumer to finish cons_thread.join() total = conn.execute('SELECT COUNT(*) FROM crimes').fetchone()[0] print(f"[MAIN] Completed. Total unique records stored in uk_crimes.db: {total}") conn.close() if __name__ == "__main__": main()