# -----------------------------------------------------------------------------
# Lou-Ka — Agrégateur de logements à louer (province de Québec)
# Auteur : Simon-Pierre Boucher — contact@spboucher.ai
# connectors/apartments_com.py : Apartments.com (CoStar) — réseau US avec un
#   vrai parc québécois (~500 propriétés / 8 400 unités via /qc/, surtout de
#   grands gestionnaires). Le site est verrouillé par Akamai (403 en direct,
#   via Scrapfly ET via Oxylabs — sondé 2026-08-18/22) : le scraping est
#   délégué à l'ACTEUR APIFY MAISON gorgeous_thistle/ka-apartments-com
#   (source : actors/ka-apartments-com de ce repo) qui passe par le Web
#   Unlocker Bright Data — HTML server-rendered, pas de rendu JS.
#   UNE annonce = UNE PROPRIÉTÉ (immeuble) : prix « à partir de » (minimum
#   des plans d'unités), tous les plans dans details["plans"]. Les fiches
#   détail sont lues en FRANÇAIS (/fr/) : description, GPS, commodités,
#   galerie. Cache détail 30 j + CADENCE : l'inventaire CoStar bouge
#   lentement, l'acteur ne tourne qu'aux LOUKA_APTS_INTERVAL_H heures ;
#   entre deux runs on ré-émet les annonces actives depuis la BD (coût 0).
# -----------------------------------------------------------------------------
from __future__ import annotations

import json
import os
import re
import time

import requests as _requests

from ..schema import Listing
from .base import BaseConnector
from . import _detailutil as du

APIFY_API = "https://api.apify.com/v2"
ACTOR = os.environ.get("LOUKA_APTS_ACTOR", "gorgeous_thistle~ka-apartments-com")

SEARCH_URLS = ["https://www.apartments.com/qc/"]     # toute la province
MAX_PAGES = int(os.environ.get("LOUKA_APTS_MAX_PAGES", "25"))
DETAIL_LIMIT = int(os.environ.get("LOUKA_APTS_DETAIL_LIMIT", "250"))
TTL_DAYS = float(os.environ.get("LOUKA_APTS_TTL_DAYS", "30"))
INTERVAL_H = float(os.environ.get("LOUKA_APTS_INTERVAL_H", "12"))
CONCURRENCY = int(os.environ.get("LOUKA_APTS_CONCURRENCY", "4"))
RUN_TIMEOUT = int(os.environ.get("LOUKA_APTS_RUN_TIMEOUT", "3600"))  # s

DETAIL_KEY = "v1"
PRICE_MIN, PRICE_MAX = 300, 12000     # garde-fous de loyer mensuel

_MONEY_RX = re.compile(r"\$\s*([\d\s,  ]+)")
_BEDS_LABEL_RX = re.compile(r"(\d+)\s*(?:bed|lit|c\.?\s*à\.?\s*c)", re.I)
_SQFT_RX = re.compile(r"([\d\s,  ]+)\s*(?:pi²|sq\s*ft)", re.I)


def _money(text: str | None) -> float | None:
    m = _MONEY_RX.search(text or "")
    if not m:
        return None
    try:
        return float(re.sub(r"[^\d]", "", m.group(1)))
    except ValueError:
        return None


def _beds_count(label: str) -> float | None:
    low = (label or "").casefold()
    if "studio" in low or "bachelor" in low:
        return 0.0
    m = _BEDS_LABEL_RX.search(low)
    return float(m.group(1)) if m else None


def _unit_type(beds: float | None) -> str:
    if beds is None:
        return ""
    if beds <= 0:
        return "Studio"
    return "6½+" if beds >= 4 else f"{int(beds) + 2}½"


def _fmt_price(price: float) -> str:
    return f"{price:,.0f} $/mois".replace(",", " ")


def _split_address(addr: str) -> tuple[str, str]:
    """« 1411 St Du Fort, Montréal, QC H3H 2N7 » -> (rue, ville)."""
    parts = [p.strip() for p in (addr or "").split(",") if p.strip()]
    if len(parts) >= 3:
        return ", ".join(parts[:-2]), parts[-2]
    return addr or "", ""


class ApartmentsComConnector(BaseConnector):
    source_id = "apartments_com"
    request_delay = 1.0

    # -- orchestration de l'acteur Apify --------------------------------------
    def _run_actor(self, payload: dict, token: str) -> list[dict]:
        """Lance l'acteur, attend la fin, retourne les items du dataset."""
        r = _requests.post(
            f"{APIFY_API}/acts/{ACTOR}/runs?waitForFinish=120",
            json=payload, timeout=180,
            headers={"Authorization": f"Bearer {token}"})
        r.raise_for_status()
        run = r.json()["data"]
        deadline = time.time() + RUN_TIMEOUT
        while run["status"] in ("READY", "RUNNING") and time.time() < deadline:
            time.sleep(10)
            run = _requests.get(
                f"{APIFY_API}/actor-runs/{run['id']}", timeout=60,
                headers={"Authorization": f"Bearer {token}"}).json()["data"]
        if run["status"] != "SUCCEEDED":
            raise RuntimeError(f"acteur {ACTOR} : run {run['id']} "
                               f"terminé en {run['status']}")
        items: list[dict] = []
        offset = 0
        while True:
            batch = _requests.get(
                f"{APIFY_API}/datasets/{run['defaultDatasetId']}/items"
                f"?limit=1000&offset={offset}", timeout=120,
                headers={"Authorization": f"Bearer {token}"}).json()
            items.extend(batch)
            if len(batch) < 1000:
                return items
            offset += 1000

    # -- cadence / ré-émission -------------------------------------------------
    def _last_ok_sync(self, cache: du.TtlDetailCache) -> float:
        row = cache.con.execute(
            "SELECT MAX(ts) AS t FROM sync_log WHERE source=? AND ok=1"
            " AND found > 0", (self.source_id,)).fetchone()
        return row["t"] or 0.0

    def _active_rows(self, cache: du.TtlDetailCache) -> list:
        return cache.con.execute(
            "SELECT external_id, url, title, address, sector, city, unit_type,"
            " bedrooms, bathrooms, price, price_label, availability,"
            " availability_date, area_sqft, pets, furnished, description,"
            " amenities, details, images, lat, lng"
            " FROM listings WHERE source=? AND active=1", (self.source_id,)
        ).fetchall()

    def _listing_from_row(self, r) -> Listing:
        def js(s, default):
            try:
                return json.loads(s) if s else default
            except ValueError:
                return default
        return Listing(
            source=self.source_id, external_id=r["external_id"],
            url=r["url"], title=r["title"] or "", address=r["address"] or "",
            sector=r["sector"] or "", city=r["city"] or "",
            unit_type=r["unit_type"] or "", bedrooms=r["bedrooms"],
            bathrooms=r["bathrooms"], price=r["price"],
            price_label=r["price_label"] or "",
            availability=r["availability"] or "",
            availability_date=r["availability_date"],
            area_sqft=r["area_sqft"], pets=r["pets"],
            furnished=(None if r["furnished"] is None else bool(r["furnished"])),
            description=r["description"] or "",
            amenities=js(r["amenities"], []), details=js(r["details"], {}),
            images=js(r["images"], []), lat=r["lat"], lng=r["lng"],
        )

    # -- payload détail -> champs Listing ---------------------------------------
    @staticmethod
    def _std_detail(d: dict) -> dict:
        std = {k: d[k] for k in ("description", "images", "lat", "lng",
                                 "amenities") if d.get(k)}
        if d.get("street"):
            std["address"] = d["street"]
        if d.get("city"):
            std["city"] = d["city"]
        plans, seen = [], set()
        for pl in d.get("plans") or []:      # grille dupliquée desktop/mobile
            sig = (pl.get("key"), pl.get("name"), pl.get("rent"))
            if sig not in seen:
                seen.add(sig)
                plans.append(pl)
        rents = [p for p in (_money(pl.get("rent")) for pl in plans)
                 if p is not None and PRICE_MIN <= p <= PRICE_MAX]
        if rents:
            std["price"] = min(rents)
            std["price_label"] = (f"à partir de {_fmt_price(min(rents))}"
                                  if len(plans) > 1 else _fmt_price(min(rents)))
        # immeuble multi-plans : minimum, cohérent avec le prix « à partir
        # de » — et ça évite que finalize() devine n'importe quoi dans la
        # description marketing (« appartements de 1 à 4 chambres »)
        beds = {pl.get("beds") for pl in plans if pl.get("beds") is not None}
        baths = {pl.get("baths") for pl in plans if pl.get("baths") is not None}
        if beds:
            std["bedrooms"] = min(beds)
            if len(beds) == 1:
                std["unit_type"] = _unit_type(std["bedrooms"])
        if baths:
            std["bathrooms"] = min(baths)
        if len(plans) == 1:
            m = _SQFT_RX.search(" ".join(plans[0].get("details") or []))
            if m:
                try:
                    std["area_sqft"] = float(re.sub(r"[^\d]", "", m.group(1)))
                except ValueError:
                    pass
        avails = [pl.get("availability") or "" for pl in plans]
        now = next((a for a in avails
                    if "maintenant" in a.casefold() or "now" in a.casefold()),
                   "")
        if now or any(avails):
            std["availability"] = now or next(a for a in avails if a)
        if plans:
            std["details"] = {"plans": [
                {k: pl[k] for k in ("name", "rent", "beds", "baths",
                                    "details", "availability") if k in pl}
                for pl in plans]}
        return std

    # -- pipeline principal ------------------------------------------------------
    def fetch(self) -> list[Listing]:
        token = os.environ.get("APIFY_TOKEN")
        bd_token = os.environ.get("BRIGHTDATA_API_KEY")
        if not token:
            raise RuntimeError("APIFY_TOKEN manquant (voir .env)")
        if not bd_token:
            raise RuntimeError("BRIGHTDATA_API_KEY manquant (voir .env)")

        cache = du.TtlDetailCache(self, budget=0, ttl_days=TTL_DAYS,
                                  key=DETAIL_KEY, fetch_html=lambda _u: "")
        try:
            # inventaire lent + acteur payant : entre deux vrais runs on
            # ré-émet les actives telles quelles (aucune requête réseau)
            age_h = (time.time() - self._last_ok_sync(cache)) / 3600
            actives = self._active_rows(cache)
            if actives and age_h < INTERVAL_H:
                return [self._listing_from_row(r) for r in actives]

            fresh = {r["external_id"] for r in cache.con.execute(
                "SELECT external_id FROM detail_cache"
                " WHERE source=? AND key=? AND fetched_at > ?",
                (self.source_id, DETAIL_KEY,
                 time.time() - TTL_DAYS * 86400)).fetchall()}
            # rattrapage : actives jamais enrichies (sans GPS ou description)
            extra = [f'{r["external_id"]}|{r["url"]}' for r in actives
                     if r["external_id"] not in fresh
                     and (r["lat"] is None or not (r["description"] or "").strip())]

            items = self._run_actor({
                "searchUrls": SEARCH_URLS,
                "maxPages": MAX_PAGES,
                "getDetails": True,
                "maxDetails": DETAIL_LIMIT,
                "skipDetailIds": sorted(fresh),
                "extraDetailIds": extra[:DETAIL_LIMIT],
                "brightdataToken": bd_token,
                "brightdataZone": os.environ.get("BRIGHTDATA_ZONE",
                                                 "web_unlocker1"),
                "concurrency": CONCURRENCY,
            }, token)

            for it in items:
                if it.get("kind") == "detail" and it.get("id"):
                    payload = {k: v for k, v in it.items()
                               if k not in ("kind",) and v is not None}
                    cache.put(str(it["id"]), payload)

            out: dict[str, Listing] = {}
            for it in items:
                if it.get("kind") != "listing":
                    continue
                pid = str(it["id"])
                street, city = _split_address(it.get("address") or "")
                rents = [(r.get("beds") or "", _money(r.get("price")))
                         for r in it.get("rents") or []]
                prices = [p for _b, p in rents
                          if p is not None and PRICE_MIN <= p <= PRICE_MAX]
                from_price = any("+" in (r.get("price") or "")
                                 for r in it.get("rents") or [])
                beds = {b for b in (_beds_count(lbl) for lbl, _p in rents)
                        if b is not None}
                lst = Listing(
                    source=self.source_id,
                    external_id=pid,
                    url=(it.get("url") or "").replace(
                        "www.apartments.com/", "www.apartments.com/fr/", 1),
                    title=(it.get("title") or street)[:200],
                    address=street,
                    city=city,
                    price=min(prices) if prices else None,
                    price_label=((f"à partir de {_fmt_price(min(prices))}"
                                  if from_price or len(prices) > 1
                                  else _fmt_price(min(prices)))
                                 if prices else ""),
                    unit_type=_unit_type(beds.pop() if len(beds) == 1
                                         else None),
                    amenities=it.get("amenities") or [],
                    images=[it["image"]] if it.get("image") else [],
                )
                detail, _f = cache.peek(pid)
                if detail:
                    du.apply_detail(lst, self._std_detail(detail))
                    if detail.get("name"):
                        lst.title = detail["name"][:200]
                if lst.price is None or not (PRICE_MIN <= lst.price
                                             <= PRICE_MAX):
                    continue          # « Call for Rent » : pas affichable
                out[pid] = lst
            return list(out.values())
        finally:
            cache.close()