"""
Restaurant-specific dine-in platform discovery.

City coverage is not restaurant coverage: the fact that a dining platform
operates in Mumbai does not prove that a particular Taftoon branch can take
payment there.  This module searches every supported platform's public,
indexed restaurant pages and returns the evidence URL.  A missing result is
deliberately reported as ``not_found`` (not ``unavailable``), because search
absence cannot prove that a restaurant is absent from a logged-in app.

The app uses only ``available`` rows in its savings calculation.  Offer text in
the search result is carried as a labelled lead and is never priced: table/time
slot offers change and may not stack with a bank offer.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import tempfile
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from typing import Dict, List, Optional
from urllib.parse import urlparse

try:
    import fcntl
except ImportError:  # pragma: no cover - production and supported dev are Unix
    fcntl = None


SERPAPI_URL = "https://serpapi.com/search"
HTTP_TIMEOUT_S = float(os.getenv("DINING_AVAILABILITY_TIMEOUT_S", "12"))
CACHE_TTL_S = int(os.getenv("DINING_AVAILABILITY_TTL_S", str(4 * 3600)))
CACHE_FILE = os.getenv("DINING_AVAILABILITY_CACHE", "/tmp/rp_dining_availability_cache.json")
CACHE_VERSION = "5"  # v5: canonical identity match + exact-page verification

PLATFORMS = {
    "dineout": {
        "name": "Swiggy Dineout",
        "domain": "swiggy.com",
        "path": "/restaurants/",
        "marker": "dineout",
        "query_site": "site:swiggy.com/restaurants",
        "query_label": "Swiggy Dineout bank offer discount",
    },
    "district": {
        "name": "District (Zomato)",
        "domain": "district.in",
        "path": "/dining/",
        "marker": "dining",
        "query_site": "site:district.in/dining",
        "query_label": "District dining bank offer discount",
    },
    "eazydiner": {
        "name": "EazyDiner / PayEazy",
        "domain": "eazydiner.com",
        "path": "/",
        "marker": "",
        "query_site": "site:eazydiner.com",
        "query_label": "EazyDiner PayEazy restaurant offer discount",
        # A review page alone does not prove that a diner can book/pay for an
        # offer. Search text must also expose a booking/payment/deal capability.
        "needs_offer_marker": True,
        "page_marker": re.compile(
            r"pay\s*eazy|payment\s+offer|get\s+(?:up\s*to|upto)\s+\d+(?:\.\d+)?\s*%\s*off",
            re.I,
        ),
    },
    "magicpin": {
        "name": "magicpin",
        "domain": "magicpin.in",
        "path": "/restaurant/",
        "marker": "/store/",
        "query_site": "site:magicpin.in",
        "query_label": "magicpin dine in magicpay voucher offer discount",
        "needs_offer_marker": True,
    },
    "nearbuy": {
        "name": "nearbuy",
        "domain": "nearbuy.com",
        "path": "/",
        "marker": "",
        "query_site": "site:nearbuy.com",
        "query_label": "nearbuy restaurant walk-in deal voucher offer discount",
        "needs_offer_marker": True,
    },
}

_PAYMENT_MARKER_RE = re.compile(
    r"\b(?:pay\s*eazy|magicpay|voucher|deal|offer|"
    r"discount|cashback|save|walk[ -]?in|\d+(?:\.\d+)?\s*%\s*off)\b",
    re.I,
)


def _http_client(**kwargs):
    import httpx
    return httpx.Client(**kwargs)

_NAME_STOP = {
    "the", "and", "restaurant", "restaurants", "bar", "kitchen", "cafe",
    "coffee", "house", "eatery", "dining", "lounge", "bistro", "grill",
}
_OFFER_RE = re.compile(
    r"(?:\b(?:flat\s+|up\s*to\s+|upto\s+)?\d+(?:\.\d+)?\s*%\s*(?:off|discount|cashback)"
    r"(?:\s+up\s+to\s+(?:₹|rs\.?\s*)\s*[\d,]+)?"
    r"|\b(?:flat\s+)?(?:₹|rs\.?\s*)\s*[\d,]+\s*(?:off|discount|cashback)\b"
    r"|\bcashback\s+\d+(?:\.\d+)?\s*%)",
    re.I,
)
_ADDRESS_STOP = {
    "road", "rd", "street", "st", "opposite", "opp", "near", "block",
    "floor", "unit", "shop", "maharashtra", "india", "building", "centre",
    "center", "west", "east", "north", "south", "bank", "sidbi", "of",
}


def _compact(value: str) -> str:
    return re.sub(r"[^a-z0-9]", "", (value or "").lower())


def _tokens(value: str) -> List[str]:
    return [
        token for token in re.findall(r"[a-z0-9]+", (value or "").lower())
        if token not in _NAME_STOP and len(token) > 1
    ]


def _restaurant_match(restaurant: str, haystack: str) -> bool:
    """Conservative restaurant identity match for a search result.

    The compact full name is preferred.  The token fallback handles provider
    titles that omit generic suffixes such as "Bar & Kitchen".
    """
    compact_name = _compact(restaurant)
    compact_hay = _compact(haystack)
    if len(compact_name) >= 4 and compact_name in compact_hay:
        return True
    tokens = _tokens(restaurant)
    if not tokens:
        return False
    hay_tokens = set(_tokens(haystack))
    return all(token in hay_tokens for token in tokens)


def _platform_url(channel_key: str, link: str) -> bool:
    cfg = PLATFORMS[channel_key]
    try:
        parsed = urlparse(link)
    except Exception:
        return False
    host = parsed.netloc.lower().split(":")[0]
    path = parsed.path.lower()
    if host != cfg["domain"] and not host.endswith("." + cfg["domain"]):
        return False
    if cfg["path"] not in path:
        return False
    if cfg.get("marker") and cfg["marker"] not in path:
        return False
    # Each platform also indexes category, editorial, menu and brand pages.
    # Accept only the canonical shape of one restaurant branch.
    if channel_key == "eazydiner":
        parts = [p for p in path.split("/") if p]
        if len(parts) != 2 or not re.search(r"-\d{5,}$", parts[-1]):
            return False
    if channel_key == "magicpin" and not re.search(
        r"/restaurant/[^/]+/store/[a-z0-9]+/?$", path, re.I
    ):
        return False
    if channel_key == "nearbuy":
        parts = [p for p in path.split("/") if p]
        if len(parts) != 2 or parts[1] in {
            "clp", "collection", "location", "all-offers", "c",
        }:
            return False
    return True


def _city_match(city: str, haystack: str) -> bool:
    city_tokens = _tokens(city)
    if not city_tokens:
        return False
    hay_tokens = set(_tokens(haystack))
    aliases = {
        "mumbai": {"mumbai", "bombay"},
        "bengaluru": {"bengaluru", "bangalore"},
        "delhi": {"delhi", "gurgaon", "gurugram", "noida", "ncr"},
        "kolkata": {"kolkata", "calcutta"},
        "chennai": {"chennai", "madras"},
        "kochi": {"kochi", "cochin"},
    }
    wanted = aliases.get(city_tokens[0], {city_tokens[0]})
    return bool(wanted & hay_tokens)


def _offer_texts(text: str) -> List[str]:
    found = []
    seen = set()
    for match in _OFFER_RE.finditer(text or ""):
        value = " ".join(match.group(0).replace("*", "").split())
        key = value.casefold()
        if key in seen:
            continue
        seen.add(key)
        found.append(value)
        if len(found) >= 3:
            break
    return found


def _address_hint(address: Optional[str], city: str) -> str:
    """Best locality-like component from a provider address.

    Full postal addresses make Google queries brittle. Prefer a comma-separated
    component with several meaningful words ("Bandra Kurla Complex"), excluding
    state, road and unit fragments. The hint is also used to reject the wrong
    branch when several restaurants share a name in one city.
    """
    city_tokens = set(_tokens(city))
    choices = []
    parts = (address or "").split(",")
    # Google commonly starts a long address with the venue/building itself;
    # that is not the locality that platform URLs use.
    if len(parts) >= 4:
        parts = parts[1:]
    for part in parts:
        raw = " ".join(part.split())
        tokens = [t for t in _tokens(raw) if t not in _ADDRESS_STOP and t not in city_tokens and not t.isdigit()]
        if not tokens or any(char.isdigit() for char in raw):
            continue
        choices.append((len(tokens), len(raw), raw))
    return max(choices, default=(0, 0, ""))[2]


def _branch_match(address: Optional[str], city: str, haystack: str) -> bool:
    hint = _address_hint(address, city)
    if not hint:
        return True
    wanted = [t for t in _tokens(hint) if t not in _ADDRESS_STOP]
    if wanted == ["bkc"]:
        wanted = ["bandra", "kurla"]
    hay_tokens = set(_tokens(haystack))
    overlap = sum(1 for token in wanted if token in hay_tokens)
    # One-word localities (Powai, BKC) must match exactly. Longer hints need at
    # least two tokens, preventing "Taftoon Powai" from proving the BKC branch.
    return overlap >= min(2, len(wanted))


def parse_platform_results(
    channel_key: str, restaurant: str, city: str, data: Dict,
    address: Optional[str] = None,
) -> Dict:
    """Pick the first direct listing that proves restaurant + city + platform."""
    listed_without_payment = None
    for row in data.get("organic_results") or []:
        link = str(row.get("link") or "").strip()
        title = " ".join(str(row.get("title") or "").split())
        snippet = " ".join(str(row.get("snippet") or row.get("description") or "").split())
        haystack = " ".join((title, snippet, link))
        if not _platform_url(channel_key, link):
            continue
        # Search snippets often contain nearby/related restaurant carousels.
        # The canonical page title + URL must identify the requested venue;
        # snippets are evidence for branch and offer capability only.
        if not _restaurant_match(restaurant, " ".join((title, link))):
            continue
        if not _city_match(city, haystack):
            continue
        if not _branch_match(address, city, haystack):
            continue
        advertised = _offer_texts(" ".join((title, snippet)))
        matched = {
            "channel_key": channel_key,
            "channel_name": PLATFORMS[channel_key]["name"],
            "status": "available",
            "listing_url": link,
            "evidence_title": title or None,
            "evidence_snippet": snippet or None,
            "advertised_offer": advertised[0] if advertised else None,
            "advertised_offers": advertised,
        }
        if PLATFORMS[channel_key].get("needs_offer_marker"):
            # magicpin contains delivery-only virtual outlets alongside dine-in
            # stores. Never turn one of those into a bill-payment route.
            if "delivery only" in haystack.lower():
                continue
            if not _PAYMENT_MARKER_RE.search(" ".join((title, snippet))):
                matched["status"] = "listed"
                listed_without_payment = listed_without_payment or matched
                continue
        return matched
    if listed_without_payment:
        return listed_without_payment
    return {
        "channel_key": channel_key,
        "channel_name": PLATFORMS[channel_key]["name"],
        "status": "not_found",
        "listing_url": None,
        "evidence_title": None,
        "evidence_snippet": None,
        "advertised_offer": None,
        "advertised_offers": [],
    }


def _exact_page_has_payment(channel_key: str, url: str) -> bool:
    """Verify payment capability that an exact listing's snippet omitted.

    Only a canonical URL accepted by ``_platform_url`` reaches this fetch.
    Any failure leaves the result ``listed`` and therefore unpriced.
    """
    marker = PLATFORMS[channel_key].get("page_marker")
    if not marker or not _platform_url(channel_key, url):
        return False
    try:
        with _http_client(
            timeout=min(HTTP_TIMEOUT_S, 8.0),
            follow_redirects=True,
            headers={"User-Agent": "RewardPilot/1.0 restaurant-offer-verifier"},
        ) as client:
            response = client.get(url)
            response.raise_for_status()
            if not _platform_url(channel_key, str(response.url)):
                return False
            return bool(marker.search(response.text[:1_000_000]))
    except Exception:
        return False


_CACHE_THREAD_LOCK = threading.Lock()
_CACHE_LOCK_FILE = f"{CACHE_FILE}.lock"


def _cache_load() -> Dict:
    try:
        with open(CACHE_FILE, encoding="utf-8") as fh:
            return json.load(fh)
    except Exception:  # noqa: BLE001
        return {}


def _cache_save(data: Dict) -> None:
    temp_path = None
    try:
        now = time.time()
        pruned = {k: v for k, v in data.items() if now - v.get("at", 0) < CACHE_TTL_S}
        cache_dir = os.path.dirname(CACHE_FILE) or "."
        fd, temp_path = tempfile.mkstemp(prefix=".rp-dining-cache-", dir=cache_dir)
        with os.fdopen(fd, "w", encoding="utf-8") as fh:
            json.dump(pruned, fh)
            fh.flush()
            os.fsync(fh.fileno())
        os.replace(temp_path, CACHE_FILE)
        temp_path = None
    except Exception:  # noqa: BLE001
        pass
    finally:
        if temp_path:
            try:
                os.unlink(temp_path)
            except OSError:
                pass


def _cache_get(key: str) -> Optional[Dict]:
    with _CACHE_THREAD_LOCK:
        lock_fh = None
        try:
            lock_fh = open(_CACHE_LOCK_FILE, "a+", encoding="utf-8")
            if fcntl is not None:
                fcntl.flock(lock_fh.fileno(), fcntl.LOCK_EX)
            return _cache_load().get(key)
        except OSError:
            return _cache_load().get(key)
        finally:
            if lock_fh is not None:
                if fcntl is not None:
                    fcntl.flock(lock_fh.fileno(), fcntl.LOCK_UN)
                lock_fh.close()


def _cache_put(key: str, payload: Dict) -> None:
    with _CACHE_THREAD_LOCK:
        lock_fh = None
        try:
            lock_fh = open(_CACHE_LOCK_FILE, "a+", encoding="utf-8")
            if fcntl is not None:
                fcntl.flock(lock_fh.fileno(), fcntl.LOCK_EX)
            cache = _cache_load()
            cache[key] = {"at": time.time(), "payload": payload}
            _cache_save(cache)
        except OSError:
            pass
        finally:
            if lock_fh is not None:
                if fcntl is not None:
                    fcntl.flock(lock_fh.fileno(), fcntl.LOCK_UN)
                lock_fh.close()


def _cache_key(restaurant: str, city: str, address: Optional[str] = None) -> str:
    raw = f"v{CACHE_VERSION}|{restaurant.strip().lower()}|{city.strip().lower()}|{(address or '').strip().lower()}"
    return hashlib.sha1(raw.encode("utf-8")).hexdigest()[:20]


def _search(channel_key: str, restaurant: str, city: str, address: Optional[str] = None) -> Dict:
    cfg = PLATFORMS[channel_key]
    hint = _address_hint(address, city)
    # A strict site: query proves identity but Google returns a skeletal snippet.
    # The natural-language form keeps the direct page near the top AND exposes
    # the listing's visible offer text; parse_platform_results still enforces
    # the exact domain/path, restaurant, city and branch before accepting it.
    query = f'"{restaurant}" "{city}" {cfg["query_label"]}'
    if hint:
        query += f' "{hint}"'
    params = {
        "engine": "google",
        "q": query,
        "hl": "en",
        "num": 10,
        "api_key": (os.getenv("SERPAPI_KEY") or "").strip(),
    }
    with _http_client(timeout=HTTP_TIMEOUT_S) as client:
        response = client.get(SERPAPI_URL, params=params)
        response.raise_for_status()
        return response.json()


def availability_for(restaurant: str, city: str, address: Optional[str] = None) -> Dict:
    """Return evidence-backed listing status for every supported dining app."""
    name = " ".join((restaurant or "").split())
    where = " ".join((city or "").split())
    checked_at = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
    empty = {
        "live": False,
        "source": "none",
        "restaurant": name,
        "city": where,
        "address": address or None,
        "checked_at": checked_at,
        "platforms": [
            {
                "channel_key": key,
                "channel_name": cfg["name"],
                "status": "unverified",
                "listing_url": None,
                "evidence_title": None,
                "evidence_snippet": None,
                "advertised_offer": None,
                "advertised_offers": [],
            }
            for key, cfg in PLATFORMS.items()
        ],
    }
    if len(name) < 2 or len(where) < 2:
        return dict(empty, source="none (restaurant and city required)")
    if not (os.getenv("SERPAPI_KEY") or "").strip():
        return dict(empty, source="none (no SERPAPI_KEY)")

    cache_key = _cache_key(name, where, address)
    cached = _cache_get(cache_key)
    if cached and time.time() - cached.get("at", 0) < CACHE_TTL_S:
        payload = dict(cached["payload"])
        payload["source"] = payload.get("source", "serpapi") + " (cached)"
        if address:
            payload["address"] = address
        return payload

    def check_platform(key: str):
        try:
            data = _search(key, name, where, address)
            if data.get("error"):
                raise RuntimeError(str(data["error"]))
            row = parse_platform_results(key, name, where, data, address)
            if row.get("status") == "listed" and row.get("listing_url"):
                if _exact_page_has_payment(key, row["listing_url"]):
                    row["status"] = "available"
            return row, True
        except Exception:  # noqa: BLE001 - each platform degrades independently
            return {
                "channel_key": key,
                "channel_name": PLATFORMS[key]["name"],
                "status": "unverified",
                "listing_url": None,
                "evidence_title": None,
                "evidence_snippet": None,
                "advertised_offer": None,
                "advertised_offers": [],
            }, False

    # Platforms are independent; parallel requests cap perceived latency at one
    # upstream timeout instead of five. The registry order remains stable.
    with ThreadPoolExecutor(max_workers=len(PLATFORMS)) as executor:
        checked = list(executor.map(check_platform, PLATFORMS.keys()))
    platforms = [row for row, _ok in checked]
    searches_ok = sum(1 for _row, ok in checked if ok)

    payload = {
        "live": searches_ok == len(PLATFORMS),
        "source": "serpapi" if searches_ok else "serpapi error",
        "restaurant": name,
        "city": where,
        "address": address or None,
        "checked_at": checked_at,
        "platforms": platforms,
    }
    # Cache a complete search only. A transient platform error must not become
    # the restaurant's answer for the next four hours.
    if searches_ok == len(PLATFORMS):
        _cache_put(cache_key, payload)
    return payload