#!/usr/bin/env python3
"""
Google Places API (New) bulk scraper for UK company lists.

Features:
- SQLite checkpoint DB
- resumable import/run/export
- retries with exponential backoff
- rate limiting
- duplicate input handling
- production-safe field masks

Google Places API (New) does not expose public GMB verification/claim status.
"""
from __future__ import annotations

import argparse
import csv
import json
import os
import sqlite3
import sys
import time
from datetime import datetime, timezone
from typing import Any, Dict, Iterable, Optional

import requests
from dotenv import load_dotenv

load_dotenv()

DEFAULT_DB = os.getenv("DB_PATH", "places_bulk.sqlite3")
API_KEY = os.getenv("GOOGLE_PLACES_API_KEY", "").strip()
DEFAULT_INPUT_CSV = os.getenv("INPUT_CSV", "2026-05-29_-_Worker_and_Temporary_Worker.csv")
DEFAULT_OUTPUT_CSV = os.getenv("OUTPUT_CSV", "google_places_results.csv")
DEFAULT_QPS = float(os.getenv("QPS", "2"))
MAX_RETRIES = int(os.getenv("MAX_RETRIES", "5"))

SEARCH_URL = "https://places.googleapis.com/v1/places:searchText"

FIELD_MASK = ",".join([
    "places.id",
    "places.displayName",
    "places.formattedAddress",
    "places.addressComponents",
    "places.location",
    "places.nationalPhoneNumber",
    "places.internationalPhoneNumber",
    "places.websiteUri",
    "places.googleMapsUri",
    "places.businessStatus",
    "places.rating",
    "places.userRatingCount",
    "places.primaryType",
    "places.types",
])


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def clean(value: Any) -> str:
    if value is None:
        return ""
    return str(value).strip()


def connect(db_path: str = DEFAULT_DB) -> sqlite3.Connection:
    con = sqlite3.connect(db_path)
    con.row_factory = sqlite3.Row
    con.execute("PRAGMA journal_mode=WAL")
    con.execute("PRAGMA synchronous=NORMAL")
    return con


def init_db(con: sqlite3.Connection) -> None:
    con.executescript(
        """
        CREATE TABLE IF NOT EXISTS companies (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            organisation_name TEXT NOT NULL,
            town_city TEXT,
            county TEXT,
            type_rating TEXT,
            route TEXT,
            query TEXT NOT NULL,
            query_hash TEXT UNIQUE NOT NULL,
            status TEXT NOT NULL DEFAULT 'pending',
            attempts INTEGER NOT NULL DEFAULT 0,
            last_error TEXT,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        );

        CREATE TABLE IF NOT EXISTS places_results (
            company_id INTEGER PRIMARY KEY,
            input_organisation_name TEXT,
            input_town_city TEXT,
            input_county TEXT,
            found INTEGER NOT NULL DEFAULT 0,
            place_id TEXT,
            found_name TEXT,
            formatted_address TEXT,
            city TEXT,
            postal_code TEXT,
            country TEXT,
            latitude REAL,
            longitude REAL,
            national_phone TEXT,
            international_phone TEXT,
            website TEXT,
            google_maps_url TEXT,
            business_status TEXT,
            rating REAL,
            reviews_count INTEGER,
            primary_type TEXT,
            types TEXT,
            confidence_score INTEGER NOT NULL DEFAULT 0,
            raw_json TEXT,
            updated_at TEXT NOT NULL,
            FOREIGN KEY(company_id) REFERENCES companies(id) ON DELETE CASCADE
        );

        CREATE INDEX IF NOT EXISTS idx_companies_status ON companies(status);
        CREATE INDEX IF NOT EXISTS idx_companies_name_city ON companies(organisation_name, town_city);
        """
    )
    con.commit()


def make_query_hash(name: str, city: str, county: str) -> str:
    return f"{name.lower()}|{city.lower()}|{county.lower()}"


def build_query(name: str, city: str, county: str) -> str:
    parts = [name]
    if city:
        parts.append(city)
    if county and county.lower() != city.lower():
        parts.append(county)
    parts.append("United Kingdom")
    return ", ".join([p for p in parts if p])


def import_csv(csv_path: str, db_path: str) -> None:
    con = connect(db_path)
    init_db(con)
    inserted = skipped = 0
    now = utc_now()

    with open(csv_path, "r", encoding="utf-8-sig", newline="") as f:
        reader = csv.DictReader(f)
        required = {"Organisation Name", "Town/City", "County", "Type & Rating", "Route"}
        missing = required - set(reader.fieldnames or [])
        if missing:
            raise SystemExit(f"CSV missing columns: {', '.join(sorted(missing))}")

        batch = []
        for row in reader:
            name = clean(row.get("Organisation Name"))
            city = clean(row.get("Town/City"))
            county = clean(row.get("County"))
            if not name:
                skipped += 1
                continue
            query = build_query(name, city, county)
            query_hash = make_query_hash(name, city, county)
            batch.append((name, city, county, clean(row.get("Type & Rating")), clean(row.get("Route")), query, query_hash, now, now))
            if len(batch) >= 1000:
                inserted += insert_batch(con, batch)
                batch.clear()
        if batch:
            inserted += insert_batch(con, batch)

    print(f"Import complete. Inserted/kept unique rows: {inserted}. Skipped blank rows: {skipped}.")


def insert_batch(con: sqlite3.Connection, batch: list[tuple]) -> int:
    before = con.total_changes
    con.executemany(
        """
        INSERT OR IGNORE INTO companies
        (organisation_name, town_city, county, type_rating, route, query, query_hash, created_at, updated_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        batch,
    )
    con.commit()
    return con.total_changes - before


def extract_address_components(place: Dict[str, Any]) -> Dict[str, str]:
    out = {"city": "", "postal_code": "", "country": ""}
    for comp in place.get("addressComponents", []) or []:
        types = set(comp.get("types", []) or [])
        text = clean(comp.get("longText") or comp.get("shortText"))
        if not text:
            continue
        if "postal_code" in types:
            out["postal_code"] = text
        elif "locality" in types or "postal_town" in types:
            out["city"] = out["city"] or text
        elif "country" in types:
            out["country"] = text
    return out


def confidence(place: Optional[Dict[str, Any]]) -> int:
    if not place:
        return 0
    score = 0
    if place.get("businessStatus") == "OPERATIONAL":
        score += 25
    if place.get("googleMapsUri"):
        score += 15
    if place.get("websiteUri"):
        score += 15
    if place.get("nationalPhoneNumber") or place.get("internationalPhoneNumber"):
        score += 15
    if (place.get("userRatingCount") or 0) > 0:
        score += 15
    if place.get("rating"):
        score += 5
    if place.get("primaryType"):
        score += 10
    return min(score, 100)


def call_google(query: str) -> Optional[Dict[str, Any]]:
    if not API_KEY:
        raise RuntimeError("GOOGLE_PLACES_API_KEY is missing. Add it to .env")

    payload = {
        "textQuery": query,
        "regionCode": "GB",
        "languageCode": "en",
        "pageSize": 1,
    }
    headers = {
        "Content-Type": "application/json",
        "X-Goog-Api-Key": API_KEY,
        "X-Goog-FieldMask": FIELD_MASK,
    }

    last_error = None
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            response = requests.post(SEARCH_URL, headers=headers, json=payload, timeout=30)
            if response.status_code in (429, 500, 502, 503, 504):
                last_error = f"HTTP {response.status_code}: {response.text[:300]}"
                time.sleep(min(60, 2 ** attempt))
                continue
            response.raise_for_status()
            data = response.json()
            places = data.get("places", []) or []
            return places[0] if places else None
        except requests.RequestException as exc:
            last_error = str(exc)
            time.sleep(min(60, 2 ** attempt))
    raise RuntimeError(last_error or "Unknown Google API error")


def save_result(con: sqlite3.Connection, company: sqlite3.Row, place: Optional[Dict[str, Any]]) -> None:
    now = utc_now()
    if not place:
        con.execute(
            """
            INSERT OR REPLACE INTO places_results
            (company_id, input_organisation_name, input_town_city, input_county, found, confidence_score, updated_at)
            VALUES (?, ?, ?, ?, 0, 0, ?)
            """,
            (company["id"], company["organisation_name"], company["town_city"], company["county"], now),
        )
        con.execute("UPDATE companies SET status='not_found', updated_at=? WHERE id=?", (now, company["id"]))
        con.commit()
        return

    addr = extract_address_components(place)
    loc = place.get("location") or {}
    display_name = place.get("displayName") or {}
    types = place.get("types") or []

    con.execute(
        """
        INSERT OR REPLACE INTO places_results
        (company_id, input_organisation_name, input_town_city, input_county, found, place_id,
         found_name, formatted_address, city, postal_code, country, latitude, longitude,
         national_phone, international_phone, website, google_maps_url, business_status,
         rating, reviews_count, primary_type, types, confidence_score, raw_json, updated_at)
        VALUES (?, ?, ?, ?, 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            company["id"], company["organisation_name"], company["town_city"], company["county"],
            clean(place.get("id")), clean(display_name.get("text")), clean(place.get("formattedAddress")),
            addr["city"], addr["postal_code"], addr["country"], loc.get("latitude"), loc.get("longitude"),
            clean(place.get("nationalPhoneNumber")), clean(place.get("internationalPhoneNumber")),
            clean(place.get("websiteUri")), clean(place.get("googleMapsUri")), clean(place.get("businessStatus")),
            place.get("rating"), place.get("userRatingCount"), clean(place.get("primaryType")),
            ",".join(types), confidence(place), json.dumps(place, ensure_ascii=False), now,
        ),
    )
    con.execute("UPDATE companies SET status='done', updated_at=? WHERE id=?", (now, company["id"]))
    con.commit()


def next_companies(con: sqlite3.Connection, limit: int, only_errors: bool) -> Iterable[sqlite3.Row]:
    if only_errors:
        where = "status='error'"
    else:
        where = "status IN ('pending','error')"
    return con.execute(
        f"SELECT * FROM companies WHERE {where} ORDER BY id ASC LIMIT ?", (limit,)
    ).fetchall()


def run(db_path: str, qps: float, limit: Optional[int], only_errors: bool) -> None:
    con = connect(db_path)
    init_db(con)
    delay = 1.0 / max(qps, 0.1)
    processed = 0
    batch_size = min(limit or 1000, 1000)

    while True:
        remaining_limit = None if limit is None else max(limit - processed, 0)
        if remaining_limit == 0:
            break
        rows = next_companies(con, min(batch_size, remaining_limit or batch_size), only_errors)
        if not rows:
            break

        for company in rows:
            start = time.time()
            try:
                con.execute(
                    "UPDATE companies SET status='processing', attempts=attempts+1, updated_at=? WHERE id=?",
                    (utc_now(), company["id"]),
                )
                con.commit()
                place = call_google(company["query"])
                save_result(con, company, place)
                processed += 1
                status = "FOUND" if place else "NOT_FOUND"
                print(f"[{processed}] {status}: {company['organisation_name']} | {company['town_city']}")
            except Exception as exc:
                msg = str(exc)[:1000]
                con.execute(
                    "UPDATE companies SET status='error', last_error=?, updated_at=? WHERE id=?",
                    (msg, utc_now(), company["id"]),
                )
                con.commit()
                processed += 1
                print(f"[{processed}] ERROR: {company['organisation_name']} | {msg}", file=sys.stderr)

            elapsed = time.time() - start
            if elapsed < delay:
                time.sleep(delay - elapsed)

    print(f"Run complete. Processed this run: {processed}")


def export_csv(db_path: str, out_path: str) -> None:
    con = connect(db_path)
    init_db(con)
    rows = con.execute(
        """
        SELECT c.id, c.organisation_name, c.town_city, c.county, c.type_rating, c.route,
               c.status, c.attempts, c.last_error,
               r.found, r.place_id, r.found_name, r.formatted_address, r.city, r.postal_code,
               r.country, r.latitude, r.longitude, r.national_phone, r.international_phone,
               r.website, r.google_maps_url, r.business_status, r.rating, r.reviews_count,
               r.primary_type, r.types, r.confidence_score
        FROM companies c
        LEFT JOIN places_results r ON r.company_id = c.id
        ORDER BY c.id ASC
        """
    ).fetchall()
    fields = [d[0] for d in con.execute(
        """
        SELECT c.id, c.organisation_name, c.town_city, c.county, c.type_rating, c.route,
               c.status, c.attempts, c.last_error,
               r.found, r.place_id, r.found_name, r.formatted_address, r.city, r.postal_code,
               r.country, r.latitude, r.longitude, r.national_phone, r.international_phone,
               r.website, r.google_maps_url, r.business_status, r.rating, r.reviews_count,
               r.primary_type, r.types, r.confidence_score
        FROM companies c LEFT JOIN places_results r ON r.company_id = c.id LIMIT 1
        """
    ).description]
    with open(out_path, "w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fields)
        writer.writeheader()
        for row in rows:
            writer.writerow(dict(row))
    print(f"Exported {len(rows)} rows to {out_path}")


def stats(db_path: str) -> None:
    con = connect(db_path)
    init_db(con)
    print("Status counts:")
    for row in con.execute("SELECT status, COUNT(*) count FROM companies GROUP BY status ORDER BY status"):
        print(f"  {row['status']}: {row['count']}")
    found = con.execute("SELECT COUNT(*) c FROM places_results WHERE found=1").fetchone()["c"]
    print(f"Found places: {found}")


def reset_errors(db_path: str) -> None:
    con = connect(db_path)
    init_db(con)
    con.execute("UPDATE companies SET status='pending', last_error=NULL, updated_at=? WHERE status='error'", (utc_now(),))
    con.commit()
    print("Errors reset to pending.")


def main() -> None:
    parser = argparse.ArgumentParser(description="Bulk Google Places API scraper for UK companies")
    parser.add_argument("--db", default=DEFAULT_DB, help="SQLite DB path")
    sub = parser.add_subparsers(dest="cmd", required=True)

    p_import = sub.add_parser("import")
    p_import.add_argument("--csv", default=DEFAULT_INPUT_CSV)

    p_run = sub.add_parser("run")
    p_run.add_argument("--qps", type=float, default=DEFAULT_QPS, help="Requests per second")
    p_run.add_argument("--limit", type=int, default=None, help="Max records to process this run")
    p_run.add_argument("--only-errors", action="store_true", help="Process only rows currently in error status")

    p_export = sub.add_parser("export")
    p_export.add_argument("--out", default=DEFAULT_OUTPUT_CSV)

    sub.add_parser("stats")
    sub.add_parser("reset-errors")

    args = parser.parse_args()
    if args.cmd == "import":
        import_csv(args.csv, args.db)
    elif args.cmd == "run":
        run(args.db, args.qps, args.limit, args.only_errors)
    elif args.cmd == "export":
        export_csv(args.db, args.out)
    elif args.cmd == "stats":
        stats(args.db)
    elif args.cmd == "reset-errors":
        reset_errors(args.db)


if __name__ == "__main__":
    main()
