""" Screenshot Studios — takes a screenshot of each studio's website and uploads it as cover_image via api-screenshots.php endpoint. Usage: python screenshot_studios.py Requirements: pip install playwright Pillow requests playwright install chromium """ import json import sys import time from pathlib import Path from io import BytesIO import requests from PIL import Image from playwright.sync_api import sync_playwright # ─── Config ─────────────────────────────────────────────────────────────── BASE_URL = "https://autodetailing.lv" TOKEN = "scr33nsh0t_2026_secret" LIST_URL = f"{BASE_URL}/api-screenshots.php?token={TOKEN}&action=list" UPLOAD_URL = f"{BASE_URL}/api-screenshots.php?token={TOKEN}&action=upload" # Note: .php extension is required — the rewrite rule allows direct access SCREENSHOT_DIR = Path(__file__).parent / "storage" / "screenshots" VIEWPORT_W = 1280 VIEWPORT_H = 800 # visible area height OUTPUT_W = 1280 OUTPUT_H = 800 QUALITY = 82 # WebP quality TIMEOUT_MS = 30000 # page load timeout SKIP_EXISTING = True # skip studios that already have a local screenshot file REUPLOAD_ONLY = True # skip screenshotting, just re-upload existing local files def normalize_url(url: str) -> str: """Ensure URL has a scheme.""" url = url.strip() if not url: return "" if not url.startswith(("http://", "https://")): url = "https://" + url return url def fetch_studios() -> list[dict]: """Fetch studio list from the API.""" print(f"Fetching studios from {LIST_URL[:60]}...") resp = requests.get(LIST_URL, timeout=30) resp.raise_for_status() data = resp.json() if not data.get("success"): print("API error:", data) sys.exit(1) return data["studios"] def take_screenshot(page, url: str, output_path: Path) -> bool: """Navigate to URL and save a cropped WebP screenshot.""" loaded = False # Strategy 1: networkidle (best quality — waits for all assets) try: page.goto(url, wait_until="networkidle", timeout=TIMEOUT_MS) page.wait_for_timeout(2000) loaded = True except Exception as e: err_msg = str(e) if "interrupted by another navigation" in err_msg: try: print(" → Following redirect...") page.wait_for_load_state("networkidle", timeout=TIMEOUT_MS) page.wait_for_timeout(2000) loaded = True except Exception: pass if "Timeout" in err_msg or "net::ERR" in err_msg: pass # fall through to strategy 2 elif not loaded: print(f" ✗ Failed to load: {e}") return False # Strategy 2: domcontentloaded fallback (for heavy sites) if not loaded: print(" → Retrying with domcontentloaded...") try: page.goto(url, wait_until="domcontentloaded", timeout=TIMEOUT_MS) page.wait_for_timeout(5000) # extra wait for visuals loaded = True except Exception as e2: err2 = str(e2) if "interrupted by another navigation" in err2: try: page.wait_for_load_state("domcontentloaded", timeout=TIMEOUT_MS) page.wait_for_timeout(5000) loaded = True except Exception: pass if not loaded: print(f" ✗ Failed to load: {e2}") return False final_url = page.url if final_url != url: print(f" → Redirected to: {final_url}") try: # Close common popups/cookie banners for selector in [ "button:has-text('Accept')", "button:has-text('Agree')", "button:has-text('OK')", "button:has-text('Close')", "button:has-text('Piekrītu')", "button:has-text('Принять')", "[class*='cookie'] button", "[id*='cookie'] button", "[class*='consent'] button", "[class*='gdpr'] button", ]: try: btn = page.locator(selector).first if btn.is_visible(timeout=500): btn.click(timeout=1000) page.wait_for_timeout(300) break except Exception: continue except Exception: pass # Take viewport screenshot raw = page.screenshot(type="png", full_page=False) img = Image.open(BytesIO(raw)) # Crop to desired dimensions img = img.crop((0, 0, min(img.width, OUTPUT_W), min(img.height, OUTPUT_H))) img = img.resize((OUTPUT_W, OUTPUT_H), Image.LANCZOS) # Save as WebP output_path.parent.mkdir(parents=True, exist_ok=True) img.save(str(output_path), "WEBP", quality=QUALITY) return True def upload_screenshot(studio_id: int, slug: str, file_path: Path) -> bool: """Upload screenshot to the server via API.""" try: with open(file_path, "rb") as f: resp = requests.post( UPLOAD_URL, data={"studio_id": studio_id, "slug": slug}, files={"screenshot": (file_path.name, f, "image/webp")}, timeout=30, ) data = resp.json() if data.get("success"): print(f" ✓ Uploaded → {data['cover_image']}") return True else: print(f" ✗ Upload error: {data}") return False except Exception as e: print(f" ✗ Upload failed: {e}") return False def main(): studios = fetch_studios() print(f"Found {len(studios)} studios total.\n") # ─── Re-upload mode: just push existing local files to new path ──────── if REUPLOAD_ONLY: with_website = [s for s in studios if s.get("website")] has_local = [ s for s in with_website if (SCREENSHOT_DIR / f"{s['slug']}.webp").exists() ] print(f" Re-upload mode: {len(has_local)} local screenshots found\n") success_count = 0 for i, studio in enumerate(has_local, 1): sid = int(studio["id"]) slug = studio["slug"] path = SCREENSHOT_DIR / f"{slug}.webp" print(f"[{i}/{len(has_local)}] {studio['name']}") if upload_screenshot(sid, slug, path): success_count += 1 print(f"\nRe-uploaded: {success_count}/{len(has_local)}") return # ─── Normal screenshot mode ─────────────────────────────────────────── with_website = [s for s in studios if s.get("website")] without_website = [s for s in studios if not s.get("website")] print(f" With website: {len(with_website)}") print(f" Without website: {len(without_website)}") if SKIP_EXISTING: before = len(with_website) with_website = [ s for s in with_website if not (SCREENSHOT_DIR / f"{s['slug']}.webp").exists() ] print(f" Already done (local file): {before - len(with_website)}") print(f" To process: {len(with_website)}") print() if not with_website: print("Nothing to do.") return SCREENSHOT_DIR.mkdir(parents=True, exist_ok=True) success_count = 0 fail_count = 0 with sync_playwright() as p: browser = p.chromium.launch(headless=True) context = browser.new_context( viewport={"width": VIEWPORT_W, "height": VIEWPORT_H}, locale="lv-LV", timezone_id="Europe/Riga", ) page = context.new_page() for i, studio in enumerate(with_website, 1): sid = studio["id"] slug = studio["slug"] name = studio["name"] url = normalize_url(studio["website"]) print(f"[{i}/{len(with_website)}] {name}") print(f" URL: {url}") if not url: print(" ✗ Empty URL, skipping") fail_count += 1 continue output_path = SCREENSHOT_DIR / f"{slug}.webp" if take_screenshot(page, url, output_path): size_kb = output_path.stat().st_size / 1024 print(f" ✓ Screenshot saved ({size_kb:.0f} KB)") if upload_screenshot(int(sid), slug, output_path): success_count += 1 else: fail_count += 1 else: fail_count += 1 print() browser.close() print("=" * 50) print(f"Done! Success: {success_count}, Failed: {fail_count}") print(f"Screenshots saved in: {SCREENSHOT_DIR}") if __name__ == "__main__": main()