searcher/scripts/sync-gallery-photos.py

#!/usr/bin/env python3
"""
sync-gallery-photos.py

Controls the real Google Chrome browser via Selenium to fetch Unsplash data.
This carries real browser cookies, localStorage, and TLS fingerprint,
so it bypasses most fetch-level anti-bot measures.

Workflow:
1. Launch real Chrome (headless or headed).
2. Visit https://unsplash.com to warm up cookies.
3. Navigate to the internal napi endpoint (/napi/photos) — same-origin,
   so it inherits all browser state.
4. Read the JSON from the page body (or intercept the network response).
5. Map & upsert into Supabase.
"""

from __future__ import annotations

import argparse
import json
import os
import random
import re
import sys
import time
import traceback
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

# ---------------------------------------------------------------------------
# Dependency check
# ---------------------------------------------------------------------------
try:
    from selenium import webdriver
    from selenium.webdriver.chrome.service import Service as ChromeService
    from selenium.webdriver.common.by import By
    from selenium.webdriver.chrome.options import Options as ChromeOptions
    from selenium.webdriver.support.ui import WebDriverWait
    from selenium.webdriver.support import expected_conditions as EC
    from selenium.common.exceptions import TimeoutException
    from webdriver_manager.chrome import ChromeDriverManager
except ImportError as exc:  # pragma: no cover
    print(
        "ERROR: selenium / webdriver-manager is not installed.\n"
        "Run: pip install selenium webdriver-manager",
        file=sys.stderr,
    )
    raise SystemExit(1) from exc

try:
    from supabase import create_client, Client
except ImportError as exc:  # pragma: no cover
    print(
        "ERROR: supabase-py is not installed.\n"
        "Run: pip install -r scripts/requirements.txt",
        file=sys.stderr,
    )
    raise SystemExit(1) from exc

try:
    from dotenv import load_dotenv
except ImportError:
    load_dotenv = None  # type: ignore[assignment]

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
DEFAULT_PAGE = 1
DEFAULT_PER_PAGE = 20
UNSPLASH_BASE_URL = "https://unsplash.com/napi"


def build_source(topic: str | None) -> str:
    """Build the source identifier used in Supabase."""
    if topic:
        return f"unsplash-{topic}"
    return "unsplash-photos"


def build_api_url(topic: str | None) -> str:
    """Build the Unsplash napi URL base path."""
    if topic:
        return f"{UNSPLASH_BASE_URL}/topics/{topic}/photos"
    return f"{UNSPLASH_BASE_URL}/photos"

# ---------------------------------------------------------------------------
# Environment helpers
# ---------------------------------------------------------------------------

def load_env_local() -> dict[str, str]:
    """Load variables from .env.local on top of the existing os.environ."""
    env_map = dict(os.environ)
    dotenv_path = Path(".env.local")

    if dotenv_path.exists() and load_dotenv:
        load_dotenv(dotenv_path, override=True)
        env_map = dict(os.environ)
    elif dotenv_path.exists():
        for line in dotenv_path.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            key, *value_parts = line.split("=")
            value = "=".join(value_parts).strip()
            env_map[key.strip()] = re.sub(r'^["\']|["\']$', "", value)

    return env_map


def require_env(env: dict[str, str], key: str) -> str:
    value = env.get(key)
    if not value:
        raise RuntimeError(f"{key} is not configured.")
    return value


# ---------------------------------------------------------------------------
# Data helpers
# ---------------------------------------------------------------------------

def normalize_date(value: str | None) -> str | None:
    if not value:
        return None
    try:
        dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
        return dt.isoformat()
    except (ValueError, TypeError):
        return None


def get_color_family(color: str | None) -> str | None:
    if not color:
        return None

    normalized = color.strip().removeprefix("#")
    if not re.fullmatch(r"[0-9a-fA-F]{6}", normalized):
        return None

    r = int(normalized[0:2], 16)
    g = int(normalized[2:4], 16)
    b = int(normalized[4:6], 16)
    max_value = max(r, g, b)
    min_value = min(r, g, b)
    lightness = (max_value + min_value) / 510

    if lightness > 0.78:
        return "light"
    if lightness < 0.2:
        return "black"

    delta = max_value - min_value
    if delta < 24:
        return "light" if lightness > 0.52 else "black"

    if max_value == r:
        hue = ((g - b) / delta) % 6
    elif max_value == g:
        hue = (b - r) / delta + 2
    else:
        hue = (r - g) / delta + 4

    hue = round(hue * 60)
    if hue < 0:
        hue += 360

    if hue < 18 or hue >= 345:
        return "red"
    if hue < 45:
        return "orange"
    if hue < 70:
        return "yellow"
    if hue < 170:
        return "green"
    if hue < 255:
        return "blue"
    if hue < 345:
        return "purple"

    return None


def map_photo_to_row(photo: dict[str, Any], source: str) -> dict[str, Any]:
    return {
        "unsplash_id": photo["id"],
        "slug": photo.get("slug"),
        "description": photo.get("description"),
        "alt_description": photo.get("alt_description"),
        "width": photo.get("width"),
        "height": photo.get("height"),
        "color": photo.get("color"),
        "color_family": get_color_family(photo.get("color")),
        "blur_hash": photo.get("blur_hash"),
        "likes": photo.get("likes"),
        "photo_created_at": normalize_date(photo.get("created_at")),
        "photo_updated_at": normalize_date(photo.get("updated_at")),
        "urls": photo.get("urls") or {},
        "links": photo.get("links") or {},
        "user_info": photo.get("user") or {},
        "raw": photo,
        "source": source,
    }


# ---------------------------------------------------------------------------
# Chrome setup
# ---------------------------------------------------------------------------

def create_chrome_driver(headless: bool = True) -> webdriver.Chrome:
    """Launch a real Chrome browser with anti-detection options."""
    options = ChromeOptions()

    if headless:
        options.add_argument("--headless=new")  # new headless mode

    # Anti-bot / stability flags
    options.add_argument("--no-sandbox")
    options.add_argument("--disable-dev-shm-usage")
    options.add_argument("--disable-blink-features=AutomationControlled")
    options.add_argument("--disable-infobars")
    options.add_argument("--window-size=1920,1080")
    options.add_argument(
        "--user-agent=Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36"
    )

    # Exclude automation switches so navigator.webdriver === undefined
    options.add_experimental_option("excludeSwitches", ["enable-automation"])
    options.add_experimental_option("useAutomationExtension", False)

    service = ChromeService(ChromeDriverManager().install())
    driver = webdriver.Chrome(service=service, options=options)

    # Remove the webdriver property via CDP so sites can't detect Selenium
    driver.execute_cdp_cmd(
        "Page.addScriptToEvaluateOnNewDocument",
        {
            "source": """
                Object.defineProperty(navigator, 'webdriver', {
                    get: () => undefined
                });
            """
        },
    )

    return driver


# ---------------------------------------------------------------------------
# Browser fetch via real Chrome
# ---------------------------------------------------------------------------

def fetch_photos_via_browser(
    driver: webdriver.Chrome, page_num: int, per_page: int, topic: str | None
) -> list[dict[str, Any]]:
    """
    1. Visit Unsplash homepage to warm up cookies.
    2. Navigate to the internal napi endpoint.
    3. Read the JSON from <pre> or page body.
    """
    # 1. Warm-up
    driver.get("https://unsplash.com")
    WebDriverWait(driver, 15).until(
        EC.presence_of_element_located((By.TAG_NAME, "body"))
    )

    # 2. Navigate to the internal API
    base_url = build_api_url(topic)
    api_url = f"{base_url}?page={page_num}&per_page={per_page}&order_by=latest"
    driver.get(api_url)

    # 3. Wait until the body contains JSON (starts with '[')
    try:
        WebDriverWait(driver, 15).until(
            lambda d: d.find_element(By.TAG_NAME, "body").text.strip().startswith("[")
        )
    except TimeoutException:
        body = driver.find_element(By.TAG_NAME, "body").text.strip()
        raise RuntimeError(
            f"Timeout waiting for JSON. First 200 chars: {body[:200]}"
        )

    body_text = driver.find_element(By.TAG_NAME, "body").text.strip()
    return json.loads(body_text)


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main() -> None:  # pragma: no cover
    parser = argparse.ArgumentParser(
        description="Sync Unsplash photos into Supabase via real Chrome."
    )
    parser.add_argument(
        "--max-pages",
        type=int,
        required=True,
        help="Maximum page number to fetch (script runs pages 1..max_pages inclusive).",
    )
    parser.add_argument(
        "--topic",
        type=str,
        default=None,
        help=(
            "Unsplash topic slug (e.g. 'wallpapers', '3d-renders', 'nature'). "
            "If omitted, uses the generic /napi/photos endpoint."
        ),
    )
    parser.add_argument(
        "--per-page",
        type=int,
        default=DEFAULT_PER_PAGE,
        help=f"Photos per page (default: {DEFAULT_PER_PAGE}).",
    )
    parser.add_argument(
        "--headless",
        action=argparse.BooleanOptionalAction,
        default=True,
        help="Run Chrome in headless mode (default: true).",
    )
    parser.add_argument(
        "--delay",
        type=int,
        default=3,
        help="Base delay in seconds between page requests (default: 3).",
    )
    args = parser.parse_args()

    max_pages: int = args.max_pages
    topic: str | None = args.topic
    per_page: int = args.per_page
    headless: bool = args.headless
    base_delay: int = args.delay

    source = build_source(topic)

    env = load_env_local()
    sb_url = require_env(env, "NEXT_PUBLIC_SUPABASE_URL")
    sb_key = require_env(env, "SUPABASE_SERVICE_ROLE_KEY")

    from supabase._sync.client import ClientOptions

    supabase: Client = create_client(
        sb_url,
        sb_key,
        options=ClientOptions(
            persist_session=False,
            auto_refresh_token=False,
        ),
    )

    # -----------------------------------------------------------------------
    # Read per_page from sync state if it exists
    # -----------------------------------------------------------------------
    state_resp = (
        supabase.table("wg_gallery_sync_state")
        .select("source,per_page")
        .eq("source", source)
        .limit(1)
        .execute()
    )
    state = state_resp.data[0] if state_resp.data else None
    if state and state.get("per_page") and args.per_page == DEFAULT_PER_PAGE:
        per_page = state["per_page"]

    # -----------------------------------------------------------------------
    # Launch Chrome once and reuse it for all pages
    # -----------------------------------------------------------------------
    driver = create_chrome_driver(headless=headless)

    total_seen = 0
    total_inserted = 0
    failed_pages: list[int] = []

    try:
        for page_num in range(1, max_pages + 1):
            print(f"[page {page_num}/{max_pages}] Fetching ...", flush=True)

            try:
                photos = fetch_photos_via_browser(driver, page_num, per_page, topic)
            except Exception as exc:
                print(f"[page {page_num}/{max_pages}] ERROR: {exc}", flush=True)
                _record_error(supabase, source, page_num, per_page, str(exc))
                failed_pages.append(page_num)
                continue

            # Validate shape
            if not isinstance(photos, list):
                msg = f"Expected a JSON array, got {type(photos).__name__}."
                print(f"[page {page_num}/{max_pages}] ERROR: {msg}", flush=True)
                _record_error(supabase, source, page_num, per_page, msg)
                failed_pages.append(page_num)
                continue

            # Upsert into Supabase
            rows = [map_photo_to_row(photo, source) for photo in photos]
            inserted_count = 0

            if rows:
                upsert_resp = (
                    supabase.table("wg_gallery_photos")
                    .upsert(rows, on_conflict="unsplash_id", ignore_duplicates=True)
                    .execute()
                )
                inserted_count = len(upsert_resp.data) if upsert_resp.data else 0

            total_seen += len(photos)
            total_inserted += inserted_count

            print(
                f"[page {page_num}/{max_pages}] seen={len(photos)} inserted={inserted_count}",
                flush=True,
            )

            # Random delay between pages to avoid rate-limiting
            if page_num < max_pages:
                sleep_sec = base_delay + random.uniform(0, 2)
                print(f"[page {page_num}/{max_pages}] Sleeping {sleep_sec:.1f}s ...", flush=True)
                time.sleep(sleep_sec)

        # -------------------------------------------------------------------
        # Final sync state update
        # -------------------------------------------------------------------
        now = datetime.now(timezone.utc).isoformat()
        next_page = max_pages + 1 if not failed_pages else failed_pages[0]
        (
            supabase.table("wg_gallery_sync_state")
            .upsert(
                {
                    "source": source,
                    "next_page": next_page,
                    "per_page": per_page,
                    "last_synced_at": now,
                    "last_inserted_count": total_inserted,
                    "last_seen_count": total_seen,
                    "last_error": None if not failed_pages else f"Failed pages: {failed_pages}",
                },
                on_conflict="source",
            )
            .execute()
        )

        result = {
            "ok": True,
            "source": source,
            "maxPages": max_pages,
            "failedPages": failed_pages,
            "totalSeen": total_seen,
            "totalInserted": total_inserted,
            "nextPage": next_page,
        }
        print(json.dumps(result, indent=2))

    finally:
        driver.quit()


def _record_error(
    supabase: Client, source: str, page_num: int, per_page: int, message: str
) -> None:
    """Persist error details so the next run can retry from the same page."""
    now = datetime.now(timezone.utc).isoformat()
    (
        supabase.table("wg_gallery_sync_state")
        .upsert(
            {
                "source": source,
                "next_page": page_num,
                "per_page": per_page,
                "last_synced_at": now,
                "last_inserted_count": 0,
                "last_seen_count": 0,
                "last_error": message,
            },
            on_conflict="source",
        )
        .execute()
    )


if __name__ == "__main__":
    try:
        main()
    except Exception:
        traceback.print_exc()
        sys.exit(1)