#!/usr/bin/env python3 """Julia's Daily Trend Radar Scan Scans design/branding sources, cross-references against taste board, produces structured candidate entries for review. Output: products/taste-board/candidates/YYYY-MM-DD.json """ import json import os import sys import re from datetime import datetime, date from pathlib import Path import urllib.request WORKSPACE = Path("/root/.openclaw/workspace") TASTE_DATA = WORKSPACE / "public/taste/prototype/data.json" CANDIDATES_DIR = WORKSPACE / "products/taste-board/candidates" PATTERNS_DIR = WORKSPACE / "strategic-intelligence/patterns" def fetch_http(url): """Simple HTTP fetch.""" req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'}) try: with urllib.request.urlopen(req, timeout=15) as resp: return resp.read().decode('utf-8', errors='ignore') except Exception as e: print(f" HTTP error for {url}: {e}", file=sys.stderr) return "" def scrape_brandnew(): """Scrape Brand New via HTTP.""" html = fetch_http("https://www.underconsideration.com/brandnew/") entries = [] # Match h2 > a patterns for m in re.finditer(r']*>\s*]*href="([^"]+)"[^>]*>([^<]+)', html, re.I): url, title = m.group(1), m.group(2).strip() if len(title) > 10 and 'DNS' not in title and 'Back on' not in title: if not url.startswith('http'): url = 'https://www.underconsideration.com/brandnew/' + url entries.append({"title": title, "url": url, "source": "Brand New"}) # Fallback: h2 text only if not entries: for m in re.finditer(r']*>([^<]+)', html, re.I): title = m.group(1).strip() if len(title) > 10 and title.startswith('New '): entries.append({"title": title, "url": "https://www.underconsideration.com/brandnew/", "source": "Brand New"}) return entries[:8] def scrape_creativeboom(): """Scrape Creative Boom via Playwright (JS-rendered).""" try: from playwright.sync_api import sync_playwright with sync_playwright() as p: browser = p.chromium.launch(headless=True, args=['--no-sandbox']) page = browser.new_page() page.goto('https://www.creativeboom.com/inspiration/', wait_until='domcontentloaded', timeout=30000) page.wait_for_timeout(3000) entries = page.evaluate("""() => { const items = []; document.querySelectorAll('a').forEach(el => { const title = el.textContent.trim().replace(/\s+/g, ' '); const href = el.href; if (title.length > 20 && title.length < 120 && href.includes('/inspiration/') && href !== 'https://www.creativeboom.com/inspiration/' && !items.some(i => i.url === href)) { items.push({ title: title.substring(0, 100), url: href, source: 'Creative Boom' }); } }); return items.slice(0, 8); }""") browser.close() return entries except Exception as e: print(f" Playwright error (Creative Boom): {e}", file=sys.stderr) return [] def scrape_itsnicethat(): """Scrape It's Nice That via Playwright.""" try: from playwright.sync_api import sync_playwright with sync_playwright() as p: browser = p.chromium.launch(headless=True, args=['--no-sandbox']) page = browser.new_page() page.goto('https://www.itsnicethat.com/', wait_until='domcontentloaded', timeout=30000) page.wait_for_timeout(3000) entries = page.evaluate("""() => { const items = []; document.querySelectorAll('a').forEach(el => { const title = el.textContent.trim().replace(/\s+/g, ' '); const href = el.href; if (title.length > 20 && title.length < 120 && href.includes('itsnicethat.com/articles') && !items.some(i => i.url === href)) { items.push({ title: title.substring(0, 100), url: href, source: "It's Nice That" }); } }); return items.slice(0, 8); }""") browser.close() return entries except Exception as e: print(f" Playwright error (It's Nice That): {e}", file=sys.stderr) return [] def load_existing_urls(): """Load URLs already in taste board to avoid duplicates.""" urls = set() if TASTE_DATA.exists(): data = json.loads(TASTE_DATA.read_text()) for e in data: if e.get('url'): urls.add(e['url'].rstrip('/')) # Also check previous candidates if CANDIDATES_DIR.exists(): for f in CANDIDATES_DIR.glob('*.json'): try: candidates = json.loads(f.read_text()) for c in candidates: if c.get('url'): urls.add(c['url'].rstrip('/')) except: pass return urls def load_trend_patterns(): """Load known trend pattern names.""" patterns = [] if PATTERNS_DIR.exists(): for f in PATTERNS_DIR.glob('*.md'): patterns.append(f.stem) return patterns def main(): print("📡 Julia Daily Scan starting...", file=sys.stderr) existing_urls = load_existing_urls() print(f" {len(existing_urls)} existing URLs to skip", file=sys.stderr) # Scan all sources all_entries = [] print(" Scanning Brand New...", file=sys.stderr) all_entries.extend(scrape_brandnew()) print(" Scanning Creative Boom...", file=sys.stderr) all_entries.extend(scrape_creativeboom()) print(" Scanning It's Nice That...", file=sys.stderr) all_entries.extend(scrape_itsnicethat()) print(f" Found {len(all_entries)} total entries", file=sys.stderr) # Filter duplicates against existing new_entries = [] for e in all_entries: url_clean = e['url'].rstrip('/') if url_clean not in existing_urls: new_entries.append(e) existing_urls.add(url_clean) print(f" {len(new_entries)} new entries after dedup", file=sys.stderr) # Output as raw candidates (Julia agent will enrich with tags, why_interesting, trend_link) CANDIDATES_DIR.mkdir(parents=True, exist_ok=True) today = date.today().isoformat() output_path = CANDIDATES_DIR / f"{today}.json" candidates = [] for e in new_entries: candidates.append({ "source": "trend-radar", "found_on": e["source"], "url": e["url"], "title": e["title"], "tags": [], # Julia agent enriches "why_interesting": "", # Julia agent enriches "trend_link": None, # Julia agent enriches "confidence": "signal", "scanned_at": datetime.utcnow().isoformat() + "Z" }) output_path.write_text(json.dumps(candidates, indent=2)) print(f" Wrote {len(candidates)} candidates to {output_path}", file=sys.stderr) # Output summary to stdout (for posting to Discord) print(json.dumps({ "date": today, "total_scanned": len(all_entries), "new_candidates": len(candidates), "candidates": candidates, "sources": { "brand_new": len([e for e in all_entries if e["source"] == "Brand New"]), "creative_boom": len([e for e in all_entries if e["source"] == "Creative Boom"]), "its_nice_that": len([e for e in all_entries if e["source"] == "It's Nice That"]), } }, indent=2)) if __name__ == "__main__": main()