# -----------------------------------------------------------------------------
# Rent-Ka — Rental listings aggregator (Canada, outside Québec)
# Author: Simon-Pierre Boucher — contact@spboucher.ai
# imgcheck.py : contrôle qualité des images d'annonces
#   - liens morts (404, timeout, non-image)          -> retirés
#   - images minuscules / pixellisées                 -> retirées
#   - placeholders de la source (même image générique
#     répétée sur une grande part des annonces)       -> retirés
#   - doublons dans une même galerie (hash contenu)   -> dédupliqués
#   Résultat par annonce : listings.images_ok (galerie nettoyée, ordre
#   d'origine conservé) + listings.img_audit (JSON de traçabilité).
#   Les vérifications par URL sont mises en cache dans image_checks.
# -----------------------------------------------------------------------------
from __future__ import annotations

import io
import json
import hashlib
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from urllib.parse import urlsplit

import requests

# seuils de qualité
MIN_WIDTH = 250
MIN_HEIGHT = 180
MIN_BYTES = 6_000          # en dessous : icône / pixel de tracking
MAX_DOWNLOAD = 6_000_000   # on ne lit jamais plus de 6 Mo par image
TIMEOUT = 12
RECHECK_FAILED_DAYS = 7    # une image en échec est retentée après 7 jours

# placeholder : même contenu présent sur >= PLACEHOLDER_MIN annonces distinctes,
# >= PLACEHOLDER_RATIO des annonces actives de la source ET réparti sur
# >= PLACEHOLDER_ADDRS adresses différentes. (Les photos d'immeuble ou de
# projet partagées entre les unités d'une MÊME adresse — piscine, gym,
# façade — sont légitimes : c'est la dispersion multi-adresses qui trahit
# le visuel générique « image non disponible ».)
PLACEHOLDER_MIN = 12
PLACEHOLDER_RATIO = 0.30
PLACEHOLDER_ADDRS = 8

_UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
       "(KHTML, like Gecko) Chrome/126 Safari/537.36 RentKaImgBot/1.0 "
       "(+https://www.rent-ka.com/bot; contact@spboucher.ai)")

# politesse : au plus 3 téléchargements simultanés PAR HÔTE (24 threads sur un
# même CDN = refus/limitation, et des images vivantes classées mortes à tort)
_HOST_SEMS: dict[str, threading.Semaphore] = {}
_HOST_LOCK = threading.Lock()


def _host_sem(url: str) -> threading.Semaphore:
    host = urlsplit(url).netloc
    with _HOST_LOCK:
        if host not in _HOST_SEMS:
            _HOST_SEMS[host] = threading.Semaphore(3)
        return _HOST_SEMS[host]


def _check_url(url: str) -> dict:
    """Télécharge et inspecte une image. Retourne {status,width,height,bytes,sha1}."""
    out = {"url": url, "status": "dead", "width": None, "height": None,
           "bytes": None, "sha1": None}
    if not url or not url.startswith(("http://", "https://")):
        return out
    try:
        with _host_sem(url):
            resp = requests.get(url, timeout=TIMEOUT, stream=True,
                                headers={"User-Agent": _UA,
                                         "Accept": "image/*,*/*;q=0.5"})
            if resp.status_code in (404, 410):
                return out            # lien mort, définitif
            if resp.status_code >= 400:
                out["status"] = "error"   # 403/429/5xx : indéterminé, à retenter
                return out
            buf = io.BytesIO()
            for chunk in resp.iter_content(65536):
                buf.write(chunk)
                if buf.tell() > MAX_DOWNLOAD:
                    break
            data = buf.getvalue()
    except requests.RequestException:
        out["status"] = "error"           # timeout/refus réseau : indéterminé
        return out
    out["bytes"] = len(data)
    out["sha1"] = hashlib.sha1(data).hexdigest()
    ctype = (resp.headers.get("content-type") or "").lower()
    if "svg" in ctype:      # SVG : pas de dimensions bitmap — au moins pas mort
        out["status"] = "ok" if len(data) >= 500 else "small"
        return out
    try:
        from PIL import Image
        img = Image.open(io.BytesIO(data))
        img.load()
        out["width"], out["height"] = img.size
    except Exception:
        out["status"] = "dead"   # corrompue / format non décodable
        return out
    if (len(data) < MIN_BYTES or out["width"] < MIN_WIDTH
            or out["height"] < MIN_HEIGHT):
        out["status"] = "small"
    else:
        out["status"] = "ok"
    return out


def _placeholder_hashes(con) -> set[str]:
    """Hashes « placeholder » : contenu identique répété massivement dans une
    même source (logo générique, « image non disponible »…)."""
    totals = {r["source"]: r["n"] for r in con.execute(
        "SELECT source, COUNT(*) n FROM listings WHERE active=1 GROUP BY source")}
    rows = con.execute(
        """SELECT l.source s, c.sha1 h, COUNT(DISTINCT l.uid) n,
                  COUNT(DISTINCT COALESCE(NULLIF(l.address,''),
                        NULLIF(l.sector,''), l.uid)) addrs
           FROM listings l, json_each(l.images) j
           JOIN image_checks c ON c.url = j.value
           WHERE l.active=1 AND c.sha1 IS NOT NULL AND c.status='ok'
           GROUP BY l.source, c.sha1 HAVING n >= ?""",
        (PLACEHOLDER_MIN,)).fetchall()
    out = set()
    for r in rows:
        if (r["n"] >= max(PLACEHOLDER_MIN, PLACEHOLDER_RATIO * totals.get(r["s"], 0))
                and r["addrs"] >= PLACEHOLDER_ADDRS):
            out.add(r["h"])
    return out


def run(limit_listings: int = 1500, workers: int = 24,
        source: str | None = None) -> dict:
    """Audite les images des annonces publiées sans audit (ou invalidées).

    Incrémental : chaque appel traite au plus `limit_listings` annonces ;
    la boucle `watch` en fait avancer un lot à chaque cycle.
    """
    from . import db
    con = db.connect()
    now = time.time()
    sql = ("SELECT uid, source, images FROM listings"
           " WHERE active=1 AND published=1 AND img_audit IS NULL"
           " AND images IS NOT NULL AND images NOT IN ('[]','null')")
    args: list = []
    if source:
        sql += " AND source=?"
        args.append(source)
    sql += " ORDER BY last_seen DESC LIMIT ?"
    args.append(limit_listings)
    listings = con.execute(sql, args).fetchall()
    if not listings:
        con.close()
        return {"listings": 0}

    # 1) URLs à vérifier (absentes du cache, ou en échec depuis > 7 jours)
    all_urls: list[str] = []
    per_listing: list[tuple[str, str, list[str]]] = []
    for r in listings:
        try:
            imgs = [u for u in json.loads(r["images"]) if isinstance(u, str)]
        except ValueError:
            imgs = []
        per_listing.append((r["uid"], r["source"], imgs))
        all_urls.extend(imgs)
    uniq = list(dict.fromkeys(all_urls))
    cached: dict[str, dict] = {}
    for i in range(0, len(uniq), 500):
        chunk = uniq[i:i + 500]
        q = ",".join("?" * len(chunk))
        for row in con.execute(
                f"SELECT * FROM image_checks WHERE url IN ({q})", chunk):
            d = dict(row)
            stale = (d["status"] != "ok"
                     and now - (d["checked_at"] or 0) > RECHECK_FAILED_DAYS * 86400)
            if not stale:
                cached[d["url"]] = d
    todo = [u for u in uniq if u not in cached]

    # 2) vérification parallèle PUIS écriture en rafale : jamais de transaction
    #    d'écriture ouverte pendant les téléchargements (verrou SQLite sinon
    #    tenu de longues secondes → « database is locked » pour les syncs)
    checked = 0
    if todo:
        with ThreadPoolExecutor(max_workers=workers) as pool:
            for i in range(0, len(todo), 400):
                chunk_res = list(pool.map(_check_url, todo[i:i + 400]))
                for res in chunk_res:
                    cached[res["url"]] = res
                con.executemany(
                    "INSERT OR REPLACE INTO image_checks"
                    " (url, status, width, height, bytes, sha1, checked_at)"
                    " VALUES (?,?,?,?,?,?,?)",
                    [(r["url"], r["status"], r["width"], r["height"],
                      r["bytes"], r["sha1"], now) for r in chunk_res])
                con.commit()
                checked += len(chunk_res)

    placeholders = _placeholder_hashes(con)

    # 3) galerie nettoyée par annonce
    stats = {"listings": len(per_listing), "urls_verifiees": checked,
             "dead": 0, "small": 0, "dup": 0, "placeholder": 0,
             "galeries_videes": 0}
    for uid, _src, imgs in per_listing:
        kept, seen_hashes = [], set()
        audit = {"n": len(imgs), "dead": 0, "small": 0, "dup": 0,
                 "placeholder": 0, "ts": int(now)}
        for u in imgs:
            c = cached.get(u)
            if c is None:          # non vérifiée (réseau) : bénéfice du doute
                kept.append(u)
                continue
            if c["status"] == "dead":
                audit["dead"] += 1
                continue
            if c["status"] == "error":    # indéterminé : bénéfice du doute
                audit["uncertain"] = audit.get("uncertain", 0) + 1
                kept.append(u)
                continue
            if c["status"] == "small":
                audit["small"] += 1
                continue
            if c["sha1"] and c["sha1"] in placeholders:
                audit["placeholder"] += 1
                continue
            if c["sha1"] and c["sha1"] in seen_hashes:
                audit["dup"] += 1
                continue
            if c["sha1"]:
                seen_hashes.add(c["sha1"])
            kept.append(u)
        audit["kept"] = len(kept)
        for k in ("dead", "small", "dup", "placeholder"):
            stats[k] += audit[k]
        if not kept and imgs:
            stats["galeries_videes"] += 1
        con.execute(
            "UPDATE listings SET images_ok=?, img_audit=? WHERE uid=?",
            (json.dumps(kept, ensure_ascii=False),
             json.dumps(audit, ensure_ascii=False), uid))
    con.commit()
    con.close()
    print(f"[rent-ka] imgcheck: {stats}")
    return stats