#!/usr/bin/env python3 """Full sync: Notion Inspiration Library → Taste Prototype data.json + screenshots.""" import argparse import json import os import ssl import sys import time import urllib.request import urllib.error from pathlib import Path # Allow importing smart capture sys.path.insert(0, str(Path(__file__).parent)) from pathlib import Path NOTION_DB_ID = "2ff330c2-8646-81f0-bbd9-ec474393d7a5" SCREENSHOTS_DIR = Path(__file__).parent.parent / "public" / "taste" / "screenshots" DATA_JSON_PATH = Path(__file__).parent.parent / "public" / "taste" / "prototype" / "data.json" API_KEY_PATH = Path("/home/clawd/secrets/notion/api_key") # SSL context that skips verification (some Notion CDN URLs need it) _nossl = ssl.create_default_context() _nossl.check_hostname = False _nossl.verify_mode = ssl.CERT_NONE def get_api_key() -> str: return API_KEY_PATH.read_text().strip() def notion_request(url: str, api_key: str, body: dict | None = None) -> dict: """Make a Notion API request.""" data = json.dumps(body).encode() if body else None req = urllib.request.Request( url, data=data, headers={ "Authorization": f"Bearer {api_key}", "Notion-Version": "2022-06-28", "Content-Type": "application/json", }, method="POST" if body is not None else "GET", ) with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read()) def query_all_pages(api_key: str, limit: int | None = None) -> list[dict]: """Paginate through entire Notion DB.""" pages = [] start_cursor = None url = f"https://api.notion.com/v1/databases/{NOTION_DB_ID}/query" while True: body: dict = {"page_size": 100} if start_cursor: body["start_cursor"] = start_cursor result = notion_request(url, api_key, body) pages.extend(result.get("results", [])) print(f" Fetched {len(pages)} pages so far...") if limit and len(pages) >= limit: pages = pages[:limit] break if not result.get("has_more"): break start_cursor = result["next_cursor"] time.sleep(0.3) return pages def short_id(page_id: str) -> str: """First 20 chars of the page ID (with hyphens).""" return page_id[:20] def extract_entry(page: dict) -> dict: """Extract a data entry from a Notion page object.""" props = page.get("properties", {}) # Title title_prop = props.get("Name", props.get("Title", {})) title_arr = title_prop.get("title", []) title = "".join(t.get("plain_text", "") for t in title_arr).strip() # URL (Link property) link_prop = props.get("Link", props.get("URL", {})) url = link_prop.get("url", "") or "" # Tags tags_prop = props.get("Tags", {}) tags = [t["name"] for t in tags_prop.get("multi_select", [])] # Cover image cover = page.get("cover") cover_url = None if cover: if cover["type"] == "external": cover_url = cover["external"]["url"] elif cover["type"] == "file": cover_url = cover["file"]["url"] sid = short_id(page["id"]) created = page.get("created_time", "") return { "id": sid, "title": title, "url": url, "desc": "", "note": "", "tags": tags, "type": "WebPage", "created": created, "image": f"screenshots/{sid}.jpg", "source": "notion", "_cover_url": cover_url, # internal, stripped before saving } def download_cover(cover_url: str, dest: Path) -> bool: """Download a cover image. Returns True on success.""" req = urllib.request.Request(cover_url, headers={ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36", "Accept": "image/*,*/*", }) # Notion S3 signed URLs don't need extra headers, just follow redirects try: with urllib.request.urlopen(req, timeout=20, context=_nossl) as resp: data = resp.read() if len(data) < 1000: return False dest.write_bytes(data) return True except Exception as e: print(f" Cover download failed: {e}") return False def main(): parser = argparse.ArgumentParser(description="Sync Notion Inspiration Library → Taste Prototype") parser.add_argument("--dry-run", action="store_true", help="Show what would be synced without downloading") parser.add_argument("--limit", type=int, default=None, help="Limit number of entries to process") parser.add_argument("--no-capture", action="store_true", help="Skip smart capture (covers only)") args = parser.parse_args() SCREENSHOTS_DIR.mkdir(parents=True, exist_ok=True) api_key = get_api_key() print(f"Querying Notion DB {NOTION_DB_ID}...") pages = query_all_pages(api_key, limit=args.limit) print(f"Got {len(pages)} pages.\n") # Lazy-load smart capture only if needed smart_capture = None if not args.no_capture and not args.dry_run: try: from importlib.util import spec_from_file_location, module_from_spec spec = spec_from_file_location("taste_smart_capture", str(Path(__file__).parent / "taste-smart-capture.py")) mod = module_from_spec(spec) spec.loader.exec_module(mod) smart_capture = mod.capture_work_image print("Smart capture loaded.\n") except Exception as e: print(f"Warning: Could not load smart capture: {e}\n") entries = [] stats = {"cover": 0, "capture": 0, "skip": 0, "existing": 0} capture_queue = [] # (entry, url) for batch smart capture for i, page in enumerate(pages): entry = extract_entry(page) entries.append(entry) img_path = SCREENSHOTS_DIR / f"{entry['id']}.jpg" label = "skip" if img_path.exists(): stats["existing"] += 1 label = "existing" elif entry["_cover_url"]: if args.dry_run: label = "cover(dry)" else: if download_cover(entry["_cover_url"], img_path): label = "cover" stats["cover"] += 1 else: # Fallback to capture if cover download fails if entry["url"] and smart_capture: capture_queue.append((entry, entry["url"])) label = "capture(queued)" else: label = "skip" stats["skip"] += 1 elif entry["url"]: if args.dry_run: label = "capture(dry)" elif smart_capture: capture_queue.append((entry, entry["url"])) label = "capture(queued)" else: label = "skip" stats["skip"] += 1 else: stats["skip"] += 1 print(f"Syncing {i+1}/{len(pages)}: {entry['title'][:50]}... [{label}]") # Process capture queue (with rate limiting) if capture_queue: print(f"\nRunning smart capture for {len(capture_queue)} entries...") for j, (entry, url) in enumerate(capture_queue): img_path = SCREENSHOTS_DIR / f"{entry['id']}.jpg" print(f" Capturing {j+1}/{len(capture_queue)}: {entry['title'][:50]}...") try: result = smart_capture(url, str(SCREENSHOTS_DIR), f"{entry['id']}.jpg") if result and img_path.exists(): stats["capture"] += 1 else: stats["skip"] += 1 except Exception as e: print(f" Capture failed: {e}") stats["skip"] += 1 if j < len(capture_queue) - 1: time.sleep(0.5) # Strip internal fields and build final data for entry in entries: del entry["_cover_url"] # Only include image path if file actually exists img_path = SCREENSHOTS_DIR / f"{entry['id']}.jpg" if not img_path.exists(): entry["image"] = "" # Sort by created time descending entries.sort(key=lambda e: e.get("created", ""), reverse=True) # Write data.json if not args.dry_run: DATA_JSON_PATH.parent.mkdir(parents=True, exist_ok=True) DATA_JSON_PATH.write_text(json.dumps(entries, indent=2, ensure_ascii=False)) print(f"\nWrote {len(entries)} entries to {DATA_JSON_PATH}") print(f"\nStats: {stats}") print(f" Existing: {stats['existing']}, Covers: {stats['cover']}, " f"Captured: {stats['capture']}, Skipped: {stats['skip']}") if __name__ == "__main__": main()