wallora/searcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
luoyangwei
2026-06-06 00:49:07 +08:00
commit 2b860d201e
134 changed files with 22773 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/sync-gallery-photos.py
Executable
+488
View File
@@ -0,0 +1,488 @@
#!/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)