narty-2027/gen_images.py

#!/usr/bin/env python3
"""Pobiera zdjęcia obiektów noclegowych do katalogu pages/img/.

Dla każdej oferty (`type: oferta*`) z polem `www` we frontmatterze skrypt
otwiera stronę obiektu i wyciąga z niej 3–5 zdjęć:

    1. og:image / twitter:image — zdjęcie, które obiekt sam wybrał na wizytówkę,
    2. galeria ze strony — <img src|data-src|srcset>, <a href="...jpg">
       oraz background-image w stylach inline.

Kandydaci są odsiewani dwoma filtrami, w tej kolejności: po nazwie pliku
(logotypy, ikony, mapy, piksele śledzące) i po samym pliku — waga > 30 KB
oraz proporcje w przedziale krajobrazowym (patrz MIN_RATIO/MAX_RATIO,
MIN_WIDTH). Co przejdzie, ląduje jako JPEG przeskalowany do MAX_WIDTH px
szerokości, jeśli w systemie jest Pillow; bez Pillow zapisywany jest oryginał
z własnym rozszerzeniem.

Zapis: pages/img/<id-oferty>/01.jpg, 02.jpg, ... — `<id-oferty>` to pole `id`
z frontmattera. Mapowanie oferta → zdjęcia jest wyłącznie konwencją
katalogową: pliki .md NIE są modyfikowane, tak samo jak w gen_pages.py
i gen_viz.py. Katalog czyta potem gen_pages.py (galeria na karcie, miniatury
w tabeli porównawczej, tło kafli regionów).

Błąd jednej oferty (strona nie oddaje zdjęć, blokada scrapingu, timeout) nie
przerywa reszty — trafia na listę braków wypisywaną na końcu. Znany przypadek:
booking.com odrzuca automatyczne pobrania i taka oferta zostaje bez zdjęć.

Uruchomienie:
    python3 gen_images.py            # pomija oferty, które mają już zdjęcia
    python3 gen_images.py --force    # pobiera wszystko od nowa
    python3 gen_images.py --only kristall-schattbergxpress
"""

from __future__ import annotations

import argparse
import io
import re
import ssl
import sys
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from html.parser import HTMLParser
from pathlib import Path
from urllib.parse import unquote, urljoin, urlsplit

import yaml

BUNDLE_ROOT = Path(__file__).resolve().parent
IMG_DIR = BUNDLE_ROOT / "pages" / "img"

WANTED = 5           # ile zdjęć chcemy na obiekt
MIN_WANTED = 3       # poniżej tego oferta idzie na listę „niepełnych”
MAX_CANDIDATES = 40  # ile adresów maksymalnie próbujemy pobrać
MIN_BYTES = 30 * 1024
MIN_WIDTH = 600
MIN_RATIO = 0.85     # wysokość ≥ szerokości to zwykle baner/logo w pionie
MAX_RATIO = 2.60     # a bardzo szerokie paski to nagłówki, nie zdjęcia
MAX_TRANSPARENT = 0.03  # udział pikseli z alfą — powyżej to logo/plakietka
MAX_FLAT = 0.35         # udział jednego koloru — powyżej to grafika, nie zdjęcie
MAX_WIDTH = 1200
TIMEOUT = 20

USER_AGENT = (
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/126.0 Safari/537.36"
)

IMAGE_EXTS = (".jpg", ".jpeg", ".png", ".webp")

_RETINA_RE = re.compile(r"@[234]x(?=\.[a-z]+$)")
MAX_HASH_DISTANCE = 10  # dystans Hamminga, poniżej którego to to samo zdjęcie

# Nazwy plików, które prawie nigdy nie są zdjęciem obiektu.
NOISE_RE = re.compile(
    r"(logo|icon|favicon|sprite|pixel|tracking|placeholder|avatar|banner|"
    r"button|arrow|flag|map|karte|mappa|plan|piktogram|wetter|weather|"
    r"footer|header|badge|award|siegel|social|whatsapp|facebook|instagram|"
    r"partner|gutschein|zertifikat|prospekt|katalog)",
    re.I,
)

try:
    from PIL import Image
except ModuleNotFoundError:  # pragma: no cover
    Image = None

PILLOW = Image is not None


# --- oferty z bundle'a -------------------------------------------------


@dataclass
class Offer:
    concept_id: str
    offer_id: str
    name: str
    url: str
    operator: str = ""


def load_offers() -> list[Offer]:
    offers: list[Offer] = []
    for path in sorted(BUNDLE_ROOT.rglob("*.md")):
        if ".git" in path.parts or "pages" in path.parts:
            continue
        text = path.read_text(encoding="utf-8")
        if not text.startswith("---\n"):
            continue
        end = text.find("\n---", 3)
        if end == -1:
            continue
        frontmatter = yaml.safe_load(text[4:end]) or {}
        if not str(frontmatter.get("type", "")).startswith("oferta"):
            continue
        offers.append(
            Offer(
                concept_id=path.relative_to(BUNDLE_ROOT).with_suffix("").as_posix(),
                offer_id=str(frontmatter.get("id") or "").strip(),
                name=str(frontmatter.get("nazwa") or frontmatter.get("title") or ""),
                url=str(frontmatter.get("www") or "").strip(),
                operator=str(frontmatter.get("operator") or ""),
            )
        )
    return offers


# --- pobieranie --------------------------------------------------------

_LAX_SSL = ssl.create_default_context()
_LAX_SSL.check_hostname = False
_LAX_SSL.verify_mode = ssl.CERT_NONE


def fetch(url: str, referer: str = "") -> tuple[bytes, str]:
    """Zwraca (bajty, content-type). Rzuca OSError/HTTPError przy porażce."""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "text/html,application/xhtml+xml,image/avif,image/webp,*/*;q=0.8",
        "Accept-Language": "pl,en;q=0.8,de;q=0.6",
    }
    if referer:
        headers["Referer"] = referer
    request = urllib.request.Request(url, headers=headers)
    try:
        with urllib.request.urlopen(request, timeout=TIMEOUT) as response:
            return response.read(), response.headers.get("Content-Type", "")
    except urllib.error.URLError as exc:
        # Część małych obiektów ma łańcuch certyfikatów niepełny — dla zdjęć
        # z publicznej strony to akceptowalne ryzyko, więc jedna próba obok.
        if isinstance(getattr(exc, "reason", None), ssl.SSLError):
            with urllib.request.urlopen(
                request, timeout=TIMEOUT, context=_LAX_SSL
            ) as response:
                return response.read(), response.headers.get("Content-Type", "")
        raise


# --- wyciąganie adresów zdjęć ze strony --------------------------------


class ImageCollector(HTMLParser):
    """Zbiera kandydatów na zdjęcia: og:image, <img>, <a href=*.jpg>, tła."""

    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.meta: list[str] = []
        self.gallery: list[str] = []

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        data = {k.lower(): (v or "") for k, v in attrs}
        if tag == "meta":
            key = (data.get("property") or data.get("name") or "").lower()
            if key in {"og:image", "og:image:url", "og:image:secure_url", "twitter:image"}:
                self._add(self.meta, data.get("content", ""))
        elif tag == "img":
            for attribute in ("data-src", "data-lazy-src", "data-original", "src"):
                if data.get(attribute):
                    self._add(self.gallery, data[attribute])
                    break
            for attribute in ("srcset", "data-srcset"):
                for part in data.get(attribute, "").split(","):
                    url = part.strip().split(" ")[0]
                    self._add(self.gallery, url)
        elif tag == "a":
            href = data.get("href", "")
            if href.split("?")[0].lower().endswith(IMAGE_EXTS):
                self._add(self.gallery, href)
        if "style" in data:
            for match in re.finditer(r"url\(['\"]?([^)'\"]+)", data["style"]):
                self._add(self.gallery, match.group(1))

    def _add(self, bucket: list[str], url: str) -> None:
        url = (url or "").strip()
        if url and not url.startswith("data:") and url not in bucket:
            bucket.append(url)


def candidates(page_url: str, markup: str, tokens: list[str] = ()) -> list[str]:
    collector = ImageCollector()
    try:
        collector.feed(markup)
    except Exception:  # parser HTML-a nie może wywrócić całego przebiegu
        pass

    ordered: list[str] = []
    seen: set[str] = set()
    for url in collector.meta + collector.gallery:
        absolute = urljoin(page_url, url)
        if not absolute.lower().startswith(("http://", "https://")):
            continue
        path = urlsplit(absolute).path
        if path.lower().endswith(".svg") or NOISE_RE.search(path):
            continue
        # To samo zdjęcie wraca w srcset w kilku gęstościach (@2x, @3x)
        # i z parametrami skalowania w query — liczy się raz.
        key = _RETINA_RE.sub("", unquote(path).lower())
        if key in seen:
            continue
        seen.add(key)
        ordered.append(absolute)

    # Na stronach portalowych (marka z wieloma obiektami, katalog noclegów)
    # w tym samym HTML-u siedzą zdjęcia cudzych obiektów. Jeśli którykolwiek
    # adres niesie w ścieżce nazwę własną obiektu, bierzemy tylko takie —
    # a gdy żaden jej nie niesie, zostaje pełna lista (strona własna obiektu).
    if tokens:
        relevant = [
            url
            for url in ordered
            if any(token in unquote(urlsplit(url).path).lower() for token in tokens)
        ]
        if relevant:
            ordered = relevant
    return ordered[:MAX_CANDIDATES]


# Człony nazw, które nie identyfikują obiektu — są w co drugiej ofercie.
GENERIC_TOKENS = {
    "apartment", "apartments", "appartement", "appartementhaus", "apart",
    "hotel", "house", "residence", "chalet", "lodge", "ferienwohnung",
    "wohnung", "central", "sport", "sportchalet",
}


def object_tokens(offer: Offer) -> list[str]:
    """Nazwy własne obiektu, po których poznamy jego zdjęcia na portalu.

    Odpadają człony rodzajowe („apartments”), nazwa operatora („AlpenParks”)
    i wszystko, co siedzi już w domenie — jeśli cała witryna należy do
    obiektu, filtrowanie po nazwie nie ma czego zawężać.
    """
    host = urlsplit(offer.url).netloc.lower().replace("-", "").replace(".", "")
    brand = re.sub(r"[^a-z0-9]", "", (offer.operator or "").lower())
    tokens = []
    for part in re.split(r"[^a-z0-9]+", offer.offer_id.lower()):
        if len(part) < 5 or part in GENERIC_TOKENS:
            continue
        if part in host or (brand and part in brand):
            continue
        tokens.append(part)
    return tokens


def object_page(page_url: str, markup: str, tokens: list[str]) -> str:
    """Podstrona obiektu wylinkowana ze strony głównej marki/portalu."""
    if not tokens or urlsplit(page_url).path.strip("/"):
        return ""  # weszliśmy już w konkretną podstronę, nie w korzeń
    for match in re.finditer(r'href=["\']([^"\'#]+)["\']', markup, re.I):
        target = urljoin(page_url, match.group(1))
        if urlsplit(target).netloc != urlsplit(page_url).netloc:
            continue
        path = unquote(urlsplit(target).path).lower()
        if path.strip("/") and any(token in path for token in tokens):
            return target
    return ""


def decode(raw: bytes, content_type: str) -> str:
    match = re.search(r"charset=([\w-]+)", content_type, re.I)
    if not match:
        match = re.search(rb'charset=["\']?([\w-]+)', raw[:4096], re.I)
        encoding = match.group(1).decode("ascii", "ignore") if match else "utf-8"
    else:
        encoding = match.group(1)
    try:
        return raw.decode(encoding, "replace")
    except LookupError:
        return raw.decode("utf-8", "replace")


# --- zapis -------------------------------------------------------------


def acceptable(raw: bytes, seen: list[int] = ()) -> tuple[bool, str, int]:
    """(czy bierzemy, powód odrzucenia, odcisk percepcyjny)."""
    if raw[:1024].lstrip().lower().startswith((b"<!doctype", b"<html", b"<?xml")):
        # Serwer oddał stronę pod adresem .jpg — tak wygląda blokada bota.
        return False, "HTML zamiast obrazka (blokada scrapingu)", 0
    if len(raw) < MIN_BYTES:
        return False, f"{len(raw) / 1024:.0f} KB < {MIN_BYTES // 1024} KB", 0
    if not PILLOW:
        return True, "", 0
    try:
        with Image.open(io.BytesIO(raw)) as image:
            width, height = image.size
            transparent = _transparency(image)
            flat = _flatness(image)
            fingerprint = _ahash(image)
    except Exception as exc:
        return False, f"nieczytelny obraz ({type(exc).__name__})", 0
    if width < MIN_WIDTH:
        return False, f"szerokość {width}px < {MIN_WIDTH}px", 0
    ratio = width / height if height else 0
    if not MIN_RATIO <= ratio <= MAX_RATIO:
        return False, f"proporcje {ratio:.2f} poza [{MIN_RATIO}, {MAX_RATIO}]", 0
    # Logotypy i plakietki przechodzą filtr wagi i proporcji (bywają duże),
    # ale zdradza je kanał alfa i płaskie tło — zdjęcie ma jedno i drugie inne.
    if transparent > MAX_TRANSPARENT:
        return False, f"grafika z przezroczystością ({transparent:.0%})", 0
    if flat > MAX_FLAT:
        return False, f"płaskie tło ({flat:.0%} jednego koloru) — grafika", 0
    # To samo zdjęcie bywa na stronie pod kilkoma adresami i w kilku
    # rozdzielczościach — po odcisku poznajemy je niezależnie od pliku.
    for other in seen:
        if bin(fingerprint ^ other).count("1") <= MAX_HASH_DISTANCE:
            return False, "duplikat wcześniejszego zdjęcia", fingerprint
    return True, "", fingerprint


def _ahash(image: "Image.Image") -> int:
    """Odcisk percepcyjny (average hash) — odporny na zmianę rozdzielczości."""
    pixels = image.convert("L").resize((8, 8)).tobytes()
    average = sum(pixels) / len(pixels)
    return sum(1 << n for n, value in enumerate(pixels) if value > average)


def _transparency(image: "Image.Image") -> float:
    if image.mode not in ("RGBA", "LA", "P") or (
        image.mode == "P" and "transparency" not in image.info
    ):
        return 0.0
    alpha = image.convert("RGBA").getchannel("A").resize((48, 48))
    histogram = alpha.getcolors(256) or []
    total = sum(count for count, _ in histogram) or 1
    return sum(count for count, value in histogram if value < 250) / total


def _flatness(image: "Image.Image") -> float:
    small = image.convert("RGB").resize((48, 48))
    histogram = small.getcolors(48 * 48)
    if not histogram:  # więcej kolorów niż pikseli próbki — na pewno zdjęcie
        return 0.0
    total = sum(count for count, _ in histogram) or 1
    return max(count for count, _ in histogram) / total


def save(raw: bytes, directory: Path, index: int, source_url: str) -> Path:
    directory.mkdir(parents=True, exist_ok=True)
    if PILLOW:
        with Image.open(io.BytesIO(raw)) as image:
            image = image.convert("RGB")
            if image.width > MAX_WIDTH:
                height = round(image.height * MAX_WIDTH / image.width)
                image = image.resize((MAX_WIDTH, height), Image.LANCZOS)
            out = directory / f"{index:02d}.jpg"
            image.save(out, "JPEG", quality=82, optimize=True, progressive=True)
        return out
    suffix = Path(urlsplit(source_url).path).suffix.lower()
    out = directory / f"{index:02d}{suffix if suffix in IMAGE_EXTS else '.jpg'}"
    out.write_bytes(raw)
    return out


# --- przebieg na ofertę ------------------------------------------------


@dataclass
class Result:
    offer: Offer
    saved: list[Path] = field(default_factory=list)
    reason: str = ""
    source: str = ""  # podstrona obiektu, jeśli weszliśmy głębiej niż `www`


def collect(offer: Offer, force: bool) -> Result:
    result = Result(offer=offer)
    directory = IMG_DIR / offer.offer_id

    if not offer.offer_id:
        result.reason = "brak pola `id` we frontmatterze"
        return result
    if not offer.url:
        result.reason = "brak pola `www` (oferta bez konkretnego obiektu)"
        return result

    existing = sorted(directory.glob("*")) if directory.is_dir() else []
    if existing and not force:
        result.saved = existing
        result.reason = "pominięte — zdjęcia już są (--force nadpisze)"
        return result

    try:
        raw, content_type = fetch(offer.url)
    except Exception as exc:
        result.reason = f"strona nie odpowiada: {_describe(exc)}"
        return result

    markup = decode(raw, content_type)
    source_url = offer.url
    tokens = object_tokens(offer)
    deeper = object_page(offer.url, markup, tokens)
    if deeper:
        try:
            raw, content_type = fetch(deeper)
            markup = decode(raw, content_type)
            source_url = deeper
            result.source = deeper
        except Exception:
            pass  # zostajemy przy stronie głównej

    urls = candidates(source_url, markup, tokens)
    if not urls:
        result.reason = "strona nie zawiera kandydatów na zdjęcia (HTML bez galerii)"
        return result

    if force and directory.is_dir():
        for old in directory.glob("*"):
            old.unlink()

    rejected: list[str] = []
    hashes: list[int] = []
    for url in urls:
        if len(result.saved) >= WANTED:
            break
        try:
            image_raw, image_type = fetch(url, referer=offer.url)
        except Exception as exc:
            rejected.append(f"{_short(url)}: {_describe(exc)}")
            continue
        if "image" not in image_type and not urlsplit(url).path.lower().endswith(
            IMAGE_EXTS
        ):
            rejected.append(f"{_short(url)}: nie obrazek ({image_type})")
            continue
        ok, why, fingerprint = acceptable(image_raw, hashes)
        if not ok:
            rejected.append(f"{_short(url)}: {why}")
            continue
        hashes.append(fingerprint)
        try:
            result.saved.append(
                save(image_raw, directory, len(result.saved) + 1, url)
            )
        except Exception as exc:
            rejected.append(f"{_short(url)}: zapis — {_describe(exc)}")

    if not result.saved:
        head = "; ".join(rejected[:3]) or "brak przyczyny"
        result.reason = f"żaden kandydat nie przeszedł filtrów ({head})"
    elif len(result.saved) < MIN_WANTED:
        result.reason = f"tylko {len(result.saved)} zdj. (cel: {MIN_WANTED}–{WANTED})"
    return result


def _describe(exc: Exception) -> str:
    if isinstance(exc, urllib.error.HTTPError):
        return f"HTTP {exc.code}"
    reason = getattr(exc, "reason", None)
    return f"{type(exc).__name__}: {reason or exc}"


def _short(url: str) -> str:
    name = urlsplit(url).path.rsplit("/", 1)[-1] or urlsplit(url).netloc
    return name[:42]


# --- main --------------------------------------------------------------


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--force", action="store_true", help="pobierz od nowa")
    parser.add_argument("--only", metavar="ID", help="tylko ta oferta (pole `id`)")
    args = parser.parse_args()

    offers = load_offers()
    if args.only:
        offers = [o for o in offers if o.offer_id == args.only]
        if not offers:
            print(f"Nie ma oferty o id={args.only}", file=sys.stderr)
            return 1

    print(f"Pillow: {'jest — skalowanie do ' + str(MAX_WIDTH) + 'px' if PILLOW else 'BRAK — zapis oryginałów'}")
    print(f"Ofert do przetworzenia: {len(offers)}\n")

    results = []
    for offer in offers:
        print(f"→ {offer.offer_id or offer.concept_id} ({offer.url or 'brak www'})")
        result = collect(offer, args.force)
        results.append(result)
        if result.saved:
            names = ", ".join(p.name for p in result.saved)
            print(f"   {len(result.saved)} zdj.: {names}"
                  + (f"  [{result.reason}]" if result.reason else ""))
        else:
            print(f"   ✗ {result.reason}")

    ok = [r for r in results if len(r.saved) >= MIN_WANTED]
    partial = [r for r in results if 0 < len(r.saved) < MIN_WANTED]
    missing = [r for r in results if not r.saved]

    print(f"\nZe zdjęciami ({len(ok)}):")
    for result in ok:
        print(f"  + {result.offer.offer_id} — {len(result.saved)}")
    if partial:
        print(f"\nNiepełne ({len(partial)}):")
        for result in partial:
            print(f"  ~ {result.offer.offer_id} — {result.reason}")
    if missing:
        print(f"\nBez zdjęć ({len(missing)}):")
        for result in missing:
            print(f"  ! {result.offer.offer_id or result.offer.concept_id} — {result.reason}")

    total = sum(p.stat().st_size for p in IMG_DIR.rglob("*") if p.is_file())
    print(f"\nRazem w {IMG_DIR.relative_to(BUNDLE_ROOT)}: {total / 1024 / 1024:.1f} MiB")
    print("Teraz: python3 gen_pages.py (wpina zdjęcia w karty i tabelę)")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())