# -----------------------------------------------------------------------------
# Home-Ka — US real-estate aggregator (Groupe KA)
# Author: Simon-Pierre Boucher — contact@spboucher.ai
# connectors/json/jsonld_site.py : schema.org JSON-LD site connector.
#
# Many brokerage sites (Real Estate Webmasters and other server-rendered
# platforms) publish each listing as JSON-LD (`Product` + `RealEstateListing`
# / `SingleFamilyResidence` nodes) — clean, stable data with full address,
# geo, price and photos, no proxy needed. This is the SECONDARY crawler path
# (authority 3): use it to bootstrap coverage while converting the brokerage
# to a direct feed. Ported from immo-ka connectors/jsonld.py.
#
#   {"id": "michael_saunders", "connector_type": "jsonld", "config": {
#       "base_url": "https://www.michaelsaunders.com",
#       "url_include": "/listing/",
#       "external_id_regex": "/listing/([A-Z0-9-]+)/",   # from the URL
#       "max_pages": 400, "static": {"state": "FL"}}}
# -----------------------------------------------------------------------------
from __future__ import annotations

import hashlib
import json
import re
import time

from ..base import BaseConnector
from ...normalize import parse_price
from ...schema import Listing

_LD_RE = re.compile(r'<script[^>]+type=["\']application/ld\+json["\'][^>]*>(.*?)</script>',
                    re.S | re.I)
_LOC_RE = re.compile(r"<loc>\s*(?:<!\[CDATA\[)?\s*(.*?)\s*(?:\]\]>)?\s*</loc>",
                     re.S | re.I)

_LISTING_TYPES = {
    "realestatelisting", "singlefamilyresidence", "house", "residence",
    "apartment", "product", "accommodation", "condominium", "townhouse",
    # some SSR platforms (Repliers/Next.js) mark the listing as a plain Place;
    # safe here because pages are pre-filtered by url_include
    "place",
}

# URL slug fallback: ".../7039-s-straight-avenue-homosassa-fl-34446/"
_SLUG_ADDR_RE = re.compile(r"/([\w-]+)-([a-z]{2})-(\d{5})/?$", re.I)

# `name` fallback: several platforms (Real Geeks, @properties/Christie's)
# put the full address in Product.name — "123 Main St #4, Austin, TX 78701"
_NAME_ADDR_RE = re.compile(
    r"^(.+?),\s*([^,]+?),\s*([A-Za-z]{2})[,\s]+(\d{5})(?:-\d{4})?\s*$")


# Next.js "flight" streaming: some SSR sites (Douglas Elliman) embed the
# JSON-LD as self.__next_s.push([0,{"type":"application/ld+json",
# "children":"<escaped JSON>"}]) instead of a <script> tag.
_NEXT_LD_RE = re.compile(
    r'application/ld\+json\\?"[^}]*?\\?"children\\?":\\?"((?:[^"\\]|\\.)*)"',
    re.S)


def _blocks(html: str):
    yield from _LD_RE.findall(html)
    for esc in _NEXT_LD_RE.findall(html):
        try:
            yield json.loads(f'"{esc}"')   # unescape the JS string
        except ValueError:
            continue


def iter_ld(html: str):
    """Iterate JSON-LD objects of a page (each block, flattened from @graph).
    Tolerates non-standard keys without the '@' prefix (John L. Scott)."""
    for block in _blocks(html):
        block = block.strip()
        if not block:
            continue
        try:
            # strict=False: some platforms leave raw control chars in remarks
            data = json.loads(block, strict=False)
        except ValueError:
            continue
        if isinstance(data, dict):
            nodes = data.get("@graph") or data.get("graph") or [data]
        else:
            nodes = data
        for node in (nodes if isinstance(nodes, list) else [nodes]):
            if isinstance(node, dict):
                if "@type" not in node and "type" in node:
                    node = {**node, "@type": node["type"]}
                yield node


def _qv(value):
    """Unwrap schema.org QuantitativeValue ({'value': 2, ...} → 2)."""
    if isinstance(value, dict):
        return value.get("value")
    return value


class JSONLDSiteConnector(BaseConnector):
    family = "jsonld"
    request_delay = 0.8

    # -- URL discovery -----------------------------------------------------
    def _listing_urls(self) -> list[str]:
        cfg = self.config
        include = re.compile(cfg.get("url_include", "/listing/|/property/"))
        urls: list[str] = []
        seen: set[str] = set()

        def add(u: str) -> None:
            if u not in seen and include.search(u):
                seen.add(u)
                urls.append(u)

        sitemaps = [cfg["sitemap_url"]] if cfg.get("sitemap_url") else []
        if not sitemaps and cfg.get("base_url"):
            sitemaps = [cfg["base_url"].rstrip("/") + "/sitemap.xml"]
        # skip nested sitemaps by name (e.g. Union Street Media's huge
        # "off-market-sitemap-N.xml" full of sold listings)
        sm_exclude = re.compile(cfg["sitemap_exclude"]) \
            if cfg.get("sitemap_exclude") else None
        depth = 0
        max_pages = int(cfg.get("max_pages", 400))
        max_children = int(cfg.get("sitemap_max_children", 80))
        _PRIORITY = re.compile(r"active|listing|residential|propert|homes",
                               re.I)
        while sitemaps and depth < 3 and len(urls) < max_pages:
            # likely-listing shards first — big indexes (@properties: 69
            # children) would otherwise exhaust the budget on agent/office maps
            sitemaps.sort(key=lambda u: 0 if _PRIORITY.search(
                u.rsplit("/", 1)[-1]) else 1)
            nxt: list[str] = []
            for sm in sitemaps[:max_children]:
                if len(urls) >= max_pages:
                    break
                if sm_exclude and sm_exclude.search(sm):
                    continue
                try:
                    resp = self.get(sm)
                    if sm.split("?")[0].endswith(".gz"):
                        import gzip
                        body = gzip.decompress(resp.content).decode(
                            "utf-8", errors="replace")
                    else:
                        body = resp.text
                    # bot wall on the sitemap itself → Firecrawl fallback
                    if ("<loc" not in body and "<urlset" not in body
                            and self.config.get("backend") == "firecrawl"):
                        body = self.fc_scrape(sm)
                except Exception:
                    if self.config.get("backend") != "firecrawl":
                        continue
                    try:
                        body = self.fc_scrape(sm)
                    except Exception:
                        continue
                for loc in _LOC_RE.findall(body):
                    # nested sitemap: .xml/.gz files, or dynamic shard URLs
                    # (sitemapindex.aspx / sitemaplistings.ashx à la Weichert)
                    low = loc.split("?")[0].lower()
                    if (low.endswith((".xml", ".gz"))
                            or "sitemap" in loc.lower().rsplit("/", 1)[-1]):
                        nxt.append(loc)
                    else:
                        add(loc)
            sitemaps = nxt
            depth += 1
        for lp in cfg.get("list_urls", []):
            try:
                body = self.get(lp).text
            except Exception:
                continue
            base = cfg.get("base_url", "").rstrip("/")
            for href in re.findall(r'href=["\']([^"\']+)["\']', body):
                u = href if href.startswith("http") else base + href
                add(u.split("#")[0])
        return urls[:max_pages]

    # -- page fetch backends -------------------------------------------------
    def _page(self, url: str) -> str:
        """One listing page: Firecrawl backend (config backend=firecrawl —
        parallel-safe, bypasses bot walls, spares the source) or polite
        direct GET."""
        if self.config.get("backend") == "firecrawl":
            return self.fc_scrape(url,
                                  proxy=self.config.get("fc_proxy", "auto"),
                                  render=bool(self.config.get("fc_render")))
        return self.get(url).text

    # -- fetch --------------------------------------------------------------
    def fetch(self) -> list[Listing]:
        cfg = self.config
        static = cfg.get("static") or {}
        ext_re = re.compile(cfg["external_id_regex"]) \
            if cfg.get("external_id_regex") else None
        # weekly cache bucket: each detail page is re-fetched at most once a
        # week (price/status refresh) and never re-fetched inside a cycle
        week = int(time.time() // (7 * 86400))
        items: list[tuple[str, str]] = []
        seen_ext: set[str] = set()
        for url in self._listing_urls():
            ext = ""
            if ext_re:
                m = ext_re.search(url)
                if m:
                    ext = m.group(1)
            if not ext:
                ext = hashlib.sha1(url.encode()).hexdigest()[:16]
            if ext not in seen_ext:
                seen_ext.add(ext)
                items.append((url, ext))

        from ... import db
        con = db.connect()
        payloads: dict[str, dict] = {}
        misses: list[tuple[str, str]] = []
        for url, ext in items:
            cached = db.get_cached_detail(con, self.source_id, ext,
                                          f"w{week}:{url}")
            if cached is not None:
                payloads[ext] = cached
            else:
                misses.append((url, ext))

        def _grab(url: str) -> dict:
            try:
                return {"ld": list(iter_ld(self._page(url)))}
            except Exception:
                return {}

        if cfg.get("backend") == "firecrawl" and misses:
            # Firecrawl absorbs the load — fetch in parallel, cache serially
            from concurrent.futures import ThreadPoolExecutor
            workers = int(cfg.get("fc_workers", 8))
            with ThreadPoolExecutor(max_workers=workers) as pool:
                results = pool.map(lambda t: _grab(t[0]), misses)
                for (url, ext), payload in zip(misses, results):
                    db.put_cached_detail(con, self.source_id, ext,
                                         f"w{week}:{url}", payload)
                    payloads[ext] = payload
        else:
            for url, ext in misses:
                payload = _grab(url)
                db.put_cached_detail(con, self.source_id, ext,
                                     f"w{week}:{url}", payload)
                payloads[ext] = payload
        con.close()

        out: list[Listing] = []
        for url, ext in items:
            lst = self._from_nodes(payloads.get(ext, {}).get("ld") or [],
                                   url, ext, static)
            if lst is not None:
                out.append(lst)
        return out

    # -- mapping --------------------------------------------------------------
    def _from_nodes(self, nodes: list[dict], url: str, ext: str,
                    static: dict) -> Listing | None:
        """Merge every listing-typed JSON-LD node of the page (REW splits the
        data across Product + SingleFamilyResidence)."""
        def _t(n):
            t = n.get("@type")
            if isinstance(t, list):   # e.g. [Land, RealEstateListing, Product]
                for x in t:
                    if str(x).lower() in _LISTING_TYPES:
                        return str(x).lower()
                return str(t[0]).lower() if t else ""
            return str(t or "").lower()

        picked = [n for n in nodes if _t(n) in _LISTING_TYPES]
        if not picked:
            return None
        # standalone Offer nodes (John L. Scott, Greybeard publish the price
        # in a sibling node instead of an `offers` property)
        loose_offers = [n for n in nodes if _t(n) == "offer"]
        merged: dict = {}
        addr: dict = {}
        geo: dict = {}
        offer: dict = {}
        subtype = ""
        # only meaningful dwelling types qualify as a property subtype —
        # generic wrappers (Product, RealEstateListing, Place...) say nothing
        _MEANINGFUL = {"singlefamilyresidence", "house", "condominium",
                       "townhouse", "apartment"}
        for n in picked:
            t = _t(n)
            if t in _MEANINGFUL and not subtype:
                subtype = t
            for k, v in n.items():
                if v in (None, "", [], {}):
                    continue
                if k == "address" and isinstance(v, dict):
                    addr = {**v, **addr}
                elif k == "geo" and isinstance(v, dict):
                    geo = {**v, **geo}
                elif k == "offers":
                    o = v[0] if isinstance(v, list) and v else v
                    if isinstance(o, dict):
                        offer = {**o, **offer}
                else:
                    merged.setdefault(k, v)
        for o in loose_offers:
            offer = {**o, **offer}
        if isinstance(merged.get("address"), str):
            addr.setdefault("streetAddress", merged["address"])

        imgs = merged.get("image") or merged.get("photo") or []
        if isinstance(imgs, str):
            imgs = [imgs]
        imgs = [i.get("url") if isinstance(i, dict) else i for i in imgs]

        price = parse_price(_qv(offer.get("price")))
        availability = str(offer.get("availability") or "")
        status = "active"
        if "soldout" in availability.lower().replace("_", ""):
            status = "sold"

        street = str(addr.get("streetAddress") or "")
        city = str(addr.get("addressLocality") or "")
        state = str(addr.get("addressRegion") or "")
        zip_code = str(addr.get("postalCode") or "")
        if not street:   # name fallback: "123 Main St, Austin, TX 78701"
            m = _NAME_ADDR_RE.match(str(merged.get("name") or "").strip())
            if m:
                street, city = m.group(1), city or m.group(2)
                state, zip_code = state or m.group(3).upper(), zip_code or m.group(4)
        if not street:   # slug fallback: .../123-main-street-austin-tx-78701/
            m = _SLUG_ADDR_RE.search(url)
            if m:
                state = state or m.group(2).upper()
                zip_code = zip_code or m.group(3)

        beds = _qv(merged.get("numberOfBedrooms") or merged.get("numberOfRooms"))
        baths = _qv(merged.get("numberOfBathroomsTotal")
                    or merged.get("numberOfFullBathrooms"))
        floor = _qv(merged.get("floorSize"))

        kw = dict(
            source=self.source_id, external_id=ext, url=url,
            title=str(merged.get("name") or ""),
            street_address=street, city=city, state=state, zip_code=zip_code,
            list_price=price, status=status,
            mls_id=ext if not ext.startswith(("http", "sha")) and len(ext) <= 20 else "",
            property_type=subtype,
            property_subtype=subtype,
            bedrooms=beds, bathrooms=baths,
            living_area_sqft=floor,
            year_built=_qv(merged.get("yearBuilt")),
            description=str(merged.get("description") or ""),
            images=[i for i in imgs if isinstance(i, str)],
            lat=_qv(geo.get("latitude")), lng=_qv(geo.get("longitude")),
            details={"jsonld_types": [str(n.get("@type")) for n in picked],
                     "availability": availability},
        )
        for k, v in static.items():
            if not kw.get(k):
                kw[k] = v
        if not kw["street_address"] and not kw["title"]:
            return None
        return Listing(**kw)