"""
maps_source.py — bulk business discovery via Google Places API (New), used as
the cheap/bounded replacement for the LLM grounded-search discovery agent.

POST https://places.googleapis.com/v1/places:searchText
Field mask kept to a handful of fields (id, displayName, websiteUri,
formattedAddress, internationalPhoneNumber) to stay in the cheapest Places
API (New) pricing tier -- a wider mask (reviews, ratings, opening hours, ...)
moves the same call into a more expensive SKU.

Needs a Google Cloud project with billing enabled and "Places API (New)"
turned on; there is a monthly free-usage credit but the key will not
authenticate at all without billing enabled. Never raises -- returns ([], 0)
on any failure and lets the caller log, matching contactout.py / signalhire.py.
"""

from __future__ import annotations

import logging
import re
import time
from urllib.parse import urlparse

import requests

from app.services.secrets import get_google_maps_key

log = logging.getLogger("outreach.maps_source")

# User-reported, 2026-10-07: searching "find UK BASED campers and trailers company" (max_leads=6) returned
# exactly 1 result, already a duplicate -- 0 new leads. Root cause: reps type an AI-prompt-style instruction
# into this box (same habit as the AI Discovery tab, which genuinely IS a prompt to Gemini) -- the
# placeholder models a plain phrase ("e.g. custom golf cart dealers in Texas"), but Google's Places Text
# Search is a literal keyword match against business listings, not an LLM -- an imperative like "find X"
# measurably hurts its matching versus just searching "X". Stripped here, ONLY on the Google Maps path
# (never the AI Discovery prompt sent to Gemini, which should stay exactly as typed).
_IMPERATIVE_PREFIX_RE = re.compile(r"^\s*find(\s+me)?(\s+(a|an|the|top|\d+))?\s+", re.IGNORECASE)


def _clean_query(query: str) -> str:
    cleaned = _IMPERATIVE_PREFIX_RE.sub("", query, count=1).strip()
    return cleaned or query.strip()

BASE_URL = "https://places.googleapis.com/v1/places:searchText"
_TIMEOUT = 25
_FIELD_MASK = (
    "places.id,places.displayName,places.websiteUri,"
    "places.formattedAddress,places.internationalPhoneNumber,nextPageToken"
)
# Google's own ceiling on Text Search pagination (20 results/page, 3 pages).
_MAX_PAGES = 3
# A fresh nextPageToken needs a short delay before Google accepts it.
_PAGE_TOKEN_DELAY_S = 2


def available() -> bool:
    return bool(get_google_maps_key())


def _search_page(query: str, page_size: int, page_token: str = "", region_code: str = "") -> dict:
    """One Text Search page. {} on any failure -- never raises."""
    key = get_google_maps_key()
    if not key or not query:
        return {}
    body: dict = {"textQuery": query, "pageSize": max(1, min(20, page_size))}
    if page_token:
        body["pageToken"] = page_token
    if region_code:
        # ISO 3166-1 alpha-2 -- biases (does not hard-filter) results toward
        # this country, same code format leads.country already uses.
        body["regionCode"] = region_code
    headers = {
        "Content-Type": "application/json",
        "X-Goog-Api-Key": key,
        "X-Goog-FieldMask": _FIELD_MASK,
    }
    try:
        r = requests.post(BASE_URL, json=body, headers=headers, timeout=_TIMEOUT)
    except requests.RequestException as exc:
        log.warning("maps search failed: %s", exc)
        return {}
    if r.status_code != 200:
        log.warning("maps search HTTP %s: %s", r.status_code, r.text[:200])
        return {}
    try:
        return r.json() or {}
    except ValueError:
        return {}


def find_businesses(
    query: str, max_results: int, region_code: str = "", require_website: bool = True
) -> tuple[list[dict], int]:
    """
    Text-search Google Places for `query`, paging up to _MAX_PAGES times, and
    return up to `max_results` businesses that have a website -- rows without
    one cannot be scanned/qualified downstream, so they are dropped here and
    counted instead of returned. Dedupes by place id.

    require_website=False keeps a row with no website as long as it has a
    listed phone (the caller saves it as a call-only lead -- see
    qualify.save_call_only); a row with neither is still dropped and counted.

    region_code (ISO 3166-1 alpha-2, optional) biases results toward that
    country -- e.g. "US" so "custom golf cart dealers" doesn't also pull in
    an unrelated UAE/Saudi result.

    Returns (rows, dropped_no_website_count). Never raises. Row shape:
    {"company_name", "company_website", "address", "phone", "place_id"}.
    """
    query = _clean_query(query or "")
    max_results = max(1, int(max_results or 1))
    if not query or not available():
        return [], 0

    rows: list[dict] = []
    seen_ids: set[str] = set()
    dropped = 0
    page_token = ""

    for _ in range(_MAX_PAGES):
        if page_token:
            time.sleep(_PAGE_TOKEN_DELAY_S)
        data = _search_page(query, max_results, page_token, region_code)
        places = data.get("places")
        if not isinstance(places, list):
            break

        for p in places:
            place_id = (p.get("id") or "").strip()
            if not place_id or place_id in seen_ids:
                continue
            seen_ids.add(place_id)

            website = (p.get("websiteUri") or "").strip()
            phone = (p.get("internationalPhoneNumber") or "").strip()
            if not website and (require_website or not phone):
                dropped += 1
                continue

            name = ((p.get("displayName") or {}).get("text") or "").strip()
            if not name:
                continue

            rows.append({
                "company_name": name,
                "company_website": website,
                "address": (p.get("formattedAddress") or "").strip(),
                "phone": phone,
                "place_id": place_id,
            })
            if len(rows) >= max_results:
                return rows, dropped

        page_token = (data.get("nextPageToken") or "").strip()
        if not page_token:
            break

    return rows, dropped


def _host(url: str) -> str:
    """Lower-cased hostname without a leading "www." ("" if unparseable)."""
    raw = (url or "").strip()
    if not raw:
        return ""
    if "://" not in raw:
        raw = "https://" + raw
    host = (urlparse(raw).netloc or "").lower().split(":")[0]
    return host[4:] if host.startswith("www.") else host


def _same_site(a: str, b: str) -> bool:
    """True when two URLs are on the same domain (or one is a subdomain of the other)."""
    ha, hb = _host(a), _host(b)
    if not ha or not hb:
        return False
    return ha == hb or ha.endswith("." + hb) or hb.endswith("." + ha)


def lookup_business(company_name: str, website: str, region_code: str = "") -> dict | None:
    """
    Look up ONE already-known company on Google Places and return its row --
    the same shape as find_businesses() rows -- but ONLY if a result's own
    website is on the same domain as `website`. That domain match is the safety
    check: a name search alone happily returns a different business with a
    similar name. None when there is no such result. Never raises.

    One Text Search request per call (billed like any other Text Search).
    """
    name = (company_name or "").strip()
    host = _host(website)
    if not name or not host or not available():
        return None
    data = _search_page(f"{name} {host}", 5, "", region_code)
    places = data.get("places")
    if not isinstance(places, list):
        return None
    for p in places:
        if not _same_site((p.get("websiteUri") or ""), website):
            continue
        return {
            "company_name": ((p.get("displayName") or {}).get("text") or "").strip(),
            "company_website": (p.get("websiteUri") or "").strip(),
            "address": (p.get("formattedAddress") or "").strip(),
            "phone": (p.get("internationalPhoneNumber") or "").strip(),
            "place_id": (p.get("id") or "").strip(),
        }
    return None