|
| 1 | +import argparse |
| 2 | +import asyncio |
| 3 | +import logging |
| 4 | +import os |
| 5 | +import subprocess |
| 6 | +import sys |
| 7 | +from datetime import datetime |
| 8 | +from pathlib import Path |
| 9 | + |
| 10 | + |
| 11 | +def setup_environment(debug: bool): |
| 12 | + if not debug: |
| 13 | + os.environ['BROWSER_USE_SETUP_LOGGING'] = 'false' |
| 14 | + os.environ['BROWSER_USE_LOGGING_LEVEL'] = 'critical' |
| 15 | + logging.getLogger().setLevel(logging.CRITICAL) |
| 16 | + else: |
| 17 | + os.environ['BROWSER_USE_SETUP_LOGGING'] = 'true' |
| 18 | + os.environ['BROWSER_USE_LOGGING_LEVEL'] = 'info' |
| 19 | + |
| 20 | + |
| 21 | +parser = argparse.ArgumentParser(description='Generate ads from landing pages using browser-use + 🍌') |
| 22 | +parser.add_argument('url', nargs='?', help='Landing page URL to analyze') |
| 23 | +parser.add_argument('--debug', action='store_true', default=False, help='Enable debug mode (show browser, verbose logs)') |
| 24 | +args = parser.parse_args() |
| 25 | +setup_environment(args.debug) |
| 26 | + |
| 27 | +import aiofiles |
| 28 | +from google import genai |
| 29 | +from PIL import Image |
| 30 | + |
| 31 | +from browser_use import Agent, BrowserSession |
| 32 | +from browser_use.llm.google import ChatGoogle |
| 33 | + |
| 34 | +GOOGLE_API_KEY = os.getenv('GOOGLE_API_KEY') |
| 35 | + |
| 36 | + |
| 37 | +class LandingPageAnalyzer: |
| 38 | + def __init__(self, debug: bool = False): |
| 39 | + self.debug = debug |
| 40 | + self.llm = ChatGoogle(model='gemini-2.0-flash-exp', api_key=GOOGLE_API_KEY) |
| 41 | + self.output_dir = Path('output') |
| 42 | + self.output_dir.mkdir(exist_ok=True) |
| 43 | + |
| 44 | + async def analyze_landing_page(self, url: str) -> dict: |
| 45 | + browser_session = BrowserSession( |
| 46 | + headless=not self.debug, # headless=False only when debug=True |
| 47 | + ) |
| 48 | + |
| 49 | + agent = Agent( |
| 50 | + task=f"""Go to {url} and quickly extract key brand information for Instagram ad creation. |
| 51 | +
|
| 52 | +Steps: |
| 53 | +1. Navigate to the website |
| 54 | +2. From the initial view, extract ONLY these essentials: |
| 55 | + - Brand/Product name |
| 56 | + - Main tagline or value proposition (one sentence) |
| 57 | + - Primary call-to-action text |
| 58 | + - Any visible pricing or special offer |
| 59 | +3. Scroll down half a page, twice (0.5 pages each) to check for any key info |
| 60 | +4. Done - keep it simple and focused on the brand |
| 61 | +
|
| 62 | +Return ONLY the key brand info, not page structure details.""", |
| 63 | + llm=self.llm, |
| 64 | + browser_session=browser_session, |
| 65 | + max_actions_per_step=2, |
| 66 | + step_timeout=30, |
| 67 | + use_thinking=False, |
| 68 | + vision_detail_level='high', |
| 69 | + ) |
| 70 | + |
| 71 | + screenshot_path = None |
| 72 | + timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') |
| 73 | + |
| 74 | + async def screenshot_callback(agent_instance): |
| 75 | + nonlocal screenshot_path |
| 76 | + import asyncio |
| 77 | + |
| 78 | + await asyncio.sleep(4) |
| 79 | + screenshot_path = self.output_dir / f'landing_page_{timestamp}.png' |
| 80 | + active_session = agent_instance.browser_session |
| 81 | + screenshot_data = await active_session.take_screenshot(path=str(screenshot_path), full_page=False) |
| 82 | + |
| 83 | + import asyncio |
| 84 | + |
| 85 | + screenshot_task = asyncio.create_task(screenshot_callback(agent)) |
| 86 | + |
| 87 | + history = await agent.run() |
| 88 | + |
| 89 | + try: |
| 90 | + await screenshot_task |
| 91 | + except Exception as e: |
| 92 | + print(f'Screenshot task failed: {e}') |
| 93 | + |
| 94 | + analysis = history.final_result() |
| 95 | + if not analysis: |
| 96 | + analysis = 'No analysis content extracted' |
| 97 | + |
| 98 | + return {'url': url, 'analysis': analysis, 'screenshot_path': screenshot_path, 'timestamp': timestamp} |
| 99 | + |
| 100 | + |
| 101 | +class AdGenerator: |
| 102 | + def __init__(self, api_key: str | None = GOOGLE_API_KEY): |
| 103 | + if not api_key: |
| 104 | + raise ValueError('GOOGLE_API_KEY is missing or empty – set the environment variable or pass api_key explicitly') |
| 105 | + |
| 106 | + self.client = genai.Client(api_key=api_key) |
| 107 | + self.output_dir = Path('output') |
| 108 | + self.output_dir.mkdir(exist_ok=True) |
| 109 | + |
| 110 | + def create_ad_prompt(self, browser_analysis: str) -> str: |
| 111 | + prompt = f"""Create an Instagram ad for this brand: |
| 112 | +
|
| 113 | +{browser_analysis} |
| 114 | +
|
| 115 | +Create a vibrant, eye-catching Instagram ad image with: |
| 116 | +- Try to use the colors and style of the logo or brand, else: |
| 117 | +- Bold, modern gradient background with bright colors |
| 118 | +- Large, playful sans-serif text with the product/service name from the analysis |
| 119 | +- Trendy design elements: geometric shapes, sparkles, emojis |
| 120 | +- Fun bubbles or badges for any pricing or special offers mentioned |
| 121 | +- Call-to-action button with text from the analysis |
| 122 | +- Emphasizes the key value proposition from the analysis |
| 123 | +- Uses visual elements that match the brand personality |
| 124 | +- Square format (1:1 ratio) |
| 125 | +- Use color psychology to drive action |
| 126 | +
|
| 127 | +Style: Modern Instagram advertisement, (1:1), scroll-stopping, professional but playful, conversion-focused""" |
| 128 | + return prompt |
| 129 | + |
| 130 | + async def generate_ad_image(self, prompt: str, screenshot_path: Path | None = None) -> bytes | None: |
| 131 | + """Generate ad image bytes using Gemini. Returns None on failure.""" |
| 132 | + |
| 133 | + try: |
| 134 | + from typing import Any |
| 135 | + |
| 136 | + contents: list[Any] = [prompt] |
| 137 | + |
| 138 | + if screenshot_path and screenshot_path.exists(): |
| 139 | + screenshot_prompt = ( |
| 140 | + '\n\nHere is the actual landing page screenshot to reference for design inspiration, ' |
| 141 | + 'colors, layout, and visual style:' |
| 142 | + ) |
| 143 | + |
| 144 | + img = Image.open(screenshot_path) |
| 145 | + w, h = img.size |
| 146 | + side = min(w, h) |
| 147 | + img = img.crop(((w - side) // 2, (h - side) // 2, (w + side) // 2, (h + side) // 2)) |
| 148 | + |
| 149 | + contents = [prompt + screenshot_prompt, img] |
| 150 | + |
| 151 | + response = await self.client.aio.models.generate_content( |
| 152 | + model='gemini-2.5-flash-image-preview', |
| 153 | + contents=contents, |
| 154 | + ) |
| 155 | + |
| 156 | + cand = getattr(response, 'candidates', None) |
| 157 | + if cand: |
| 158 | + for part in getattr(cand[0].content, 'parts', []): |
| 159 | + inline = getattr(part, 'inline_data', None) |
| 160 | + if inline: |
| 161 | + return inline.data |
| 162 | + |
| 163 | + except Exception as e: |
| 164 | + print(f'❌ Image generation failed: {e}') |
| 165 | + |
| 166 | + return None |
| 167 | + |
| 168 | + async def save_results(self, ad_image: bytes, prompt: str, analysis: str, url: str, timestamp: str) -> str: |
| 169 | + image_path = self.output_dir / f'ad_{timestamp}.png' |
| 170 | + async with aiofiles.open(image_path, 'wb') as f: |
| 171 | + await f.write(ad_image) |
| 172 | + |
| 173 | + analysis_path = self.output_dir / f'analysis_{timestamp}.txt' |
| 174 | + async with aiofiles.open(analysis_path, 'w', encoding='utf-8') as f: |
| 175 | + await f.write(f'URL: {url}\n\n') |
| 176 | + await f.write('BROWSER-USE ANALYSIS:\n') |
| 177 | + await f.write(analysis) |
| 178 | + await f.write('\n\nGENERATED PROMPT:\n') |
| 179 | + await f.write(prompt) |
| 180 | + |
| 181 | + return str(image_path) |
| 182 | + |
| 183 | + |
| 184 | +def open_image(image_path: str): |
| 185 | + """Open image with default system viewer""" |
| 186 | + try: |
| 187 | + if sys.platform.startswith('darwin'): |
| 188 | + # macOS |
| 189 | + subprocess.run(['open', image_path], check=True) |
| 190 | + elif sys.platform.startswith('win'): |
| 191 | + # Windows |
| 192 | + subprocess.run(['cmd', '/c', 'start', '', image_path], check=True) |
| 193 | + else: |
| 194 | + # Linux |
| 195 | + subprocess.run(['xdg-open', image_path], check=True) |
| 196 | + except Exception as e: |
| 197 | + print(f'❌ Could not open image: {e}') |
| 198 | + |
| 199 | + |
| 200 | +async def create_ad_from_landing_page(url: str, debug: bool = False): |
| 201 | + analyzer = LandingPageAnalyzer(debug=debug) |
| 202 | + generator = AdGenerator() |
| 203 | + |
| 204 | + try: |
| 205 | + print(f'🚀 Analyzing {url}...') |
| 206 | + page_data = await analyzer.analyze_landing_page(url) |
| 207 | + |
| 208 | + prompt = generator.create_ad_prompt(page_data['analysis']) |
| 209 | + ad_image = await generator.generate_ad_image(prompt, page_data.get('screenshot_path')) |
| 210 | + if ad_image is None: |
| 211 | + raise RuntimeError('Ad image generation failed') |
| 212 | + result_path = await generator.save_results(ad_image, prompt, page_data['analysis'], url, page_data['timestamp']) |
| 213 | + |
| 214 | + print(f'🎨 Generated ad: {result_path}') |
| 215 | + if page_data.get('screenshot_path'): |
| 216 | + print(f'📸 Page screenshot: {page_data["screenshot_path"]}') |
| 217 | + open_image(result_path) |
| 218 | + |
| 219 | + return result_path |
| 220 | + |
| 221 | + except Exception as e: |
| 222 | + print(f'❌ Error: {e}') |
| 223 | + raise |
| 224 | + |
| 225 | + |
| 226 | +if __name__ == '__main__': |
| 227 | + url = args.url |
| 228 | + if not url: |
| 229 | + url = input('🔗 Enter URL: ').strip() or 'https://www.apple.com/iphone-17-pro/' |
| 230 | + |
| 231 | + asyncio.run(create_ad_from_landing_page(url, debug=args.debug)) |
0 commit comments