#!/usr/bin/env python3 """ NYTimes Search Plugin Self-contained plugin for searching articles on nytimes.com. Supports two modes: - Direct navigation to nytimes URL - Search form interaction with human-like behavior """ from dataclasses import dataclass from typing import Dict, Any import asyncio import random from lib.primitive.click_element import ClickElement from lib.primitive.type_text import TypeText from lib.primitive.paste_text import PasteText from lib.primitive.press_enter import PressEnter from lib.primitive.press_tab import PressTab from lib.primitive.scroll_page import ScrollPage # Import plugin base classes from lib.plugin_loader import PluginMetadata, PluginParameter class NYTimesSearchPlugin: """ Search for articles on nytimes.com. Supports two modes: - Direct navigation to nytimes URL - Search form interaction with human-like behavior """ metadata = PluginMetadata( name="nytimes-search", description="NYTimes Search", methods=["POST"], bg_color=["#6C9EB5"], parameters=[ PluginParameter( name="query", type="text", label="Search Query or URL", required=True, placeholder="Enter search term or nytimes.com URL", help_text="Search term to look for, or full nytimes.com URL to navigate to" ), PluginParameter( name="direct_access_probability", type="number", label="Direct Access Probability", required=False, default=0.5, help_text="Probability (0.0-1.0) of directly navigating to URL vs using search form" ), PluginParameter( name="paste_probability", type="number", label="Paste Probability", required=False, default=0.8, help_text="Probability (0.0-1.0) of pasting vs typing the search query" ) ] ) def validate_params(self, params: Dict[str, Any]) -> str | None: """Custom validation for probability parameters.""" direct_prob = params.get('direct_access_probability', 0.5) paste_prob = params.get('paste_probability', 0.8) if not (0.0 <= direct_prob <= 1.0): return "direct_access_probability must be between 0.0 and 1.0" if not (0.0 <= paste_prob <= 1.0): return "paste_probability must be between 0.0 and 1.0" return None async def execute(self, browser_navigator, page_id: str, params: Dict[str, Any]) -> Dict[str, Any]: """ Execute NYTimes search operation. Args: browser_navigator: BrowserNavigator instance page_id: Page to use for operation params: Validated parameters (query, direct_access_probability, paste_probability) Returns: Dict with success status, method used, and page URL """ query = params['query'] direct_access_probability = params.get('direct_access_probability', 0.5) paste_probability = params.get('paste_probability', 0.8) print(f"🔍 NYTimes search: query={query}, direct_prob={direct_access_probability}, paste_prob={paste_probability}") # Get managed page managed_page = await browser_navigator._get_page_to_use(page_id) page = managed_page.page try: # Set page state to busy await managed_page.set_busy(f"searching nytimes for {query}") # Initialize primitives clicker = ClickElement() typer = TypeText() paster = PasteText() presser = PressEnter() scroller = ScrollPage() tabber = PressTab() # Check if query is a nytimes URL is_nytimes_url = "nytimes.com" in query.lower() # PATH 1: Direct navigation if probability met and query is nytimes URL if is_nytimes_url and random.random() < direct_access_probability: print(f"🎯 Direct navigation to NYTimes URL") # Ensure URL has protocol url = query if query.startswith("http") else f"https://{query}" print(f"🌐 Navigating to: {url}") await page.goto(url, wait_until='domcontentloaded') print(f"✅ Page loaded") # Wait for page stabilization await asyncio.sleep(random.uniform(2, 3)) print(f"⏳ Waited for page stabilization") # Scroll to simulate reading await scroller.scroll(page, distance=1500, slowmo=200) print(f"📜 Scrolled down") await scroller.scroll(page, distance=500, slowmo=200) print(f"📜 Scrolled down again") return { "success": True, "method": "direct_navigation", "url": url, "page_url": page.url, "message": "Successfully navigated to NYTimes URL and scrolled" } else: # PATH 2: Search form interaction print(f"🔎 Using search form approach") await page.goto("https://www.nytimes.com/", wait_until='domcontentloaded') print(f"✅ NYTimes homepage loaded") await asyncio.sleep(random.uniform(0.3, 0.7)) # Click search button (handle mobile layout) print(f"🔘 Clicking search button") hamburger_selector = 'button:has(svg > rect[x="1"][y="3"])' lens_selector = '[data-testid="search-button"]:has(svg > path[d^="M11.3,9.2"])' hamburger = await page.query_selector(hamburger_selector) lens = await page.query_selector(lens_selector) hamburger_present = hamburger is not None and await hamburger.is_visible() lens_present = lens is not None and await lens.is_visible() if hamburger_present: print(f"📱 Detected mobile layout, clicking hamburger menu") await clicker.click_selector(managed_page, hamburger_selector) await asyncio.sleep(random.uniform(0.2, 1)) # Tab navigation for focus await tabber.press(page) await asyncio.sleep(random.uniform(0.2, 0.8)) await tabber.press(page, with_shift=True) await asyncio.sleep(random.uniform(0.2, 0.6)) else: await clicker.click_selector(managed_page, lens_selector) await asyncio.sleep(random.uniform(0.5, 1)) # Type or paste query print(f"⌨️ Entering search query") if random.random() < paste_probability: await paster.paste(page, query) else: await typer.type(page, query) # Submit search print(f"↵ Pressing enter") await presser.press(page) await asyncio.sleep(random.uniform(0.2, 0.9)) # Scroll and interact with results print(f"📜 Scrolling to see results") await scroller.scroll(page, distance=100, slowmo=100) await asyncio.sleep(random.uniform(0.2, 0.9)) # Click first result to ensure results loaded await clicker.click_selector(managed_page, 'main h4') await asyncio.sleep(random.uniform(0.2, 0.9)) await scroller.scroll(page, distance=100, slowmo=100) await asyncio.sleep(random.uniform(0.2, 0.9)) return { "success": True, "method": "search_form", "query": query, "page_url": page.url, "message": "Successfully performed search using search form" } except Exception as e: print(f"❌ NYTimes search failed: {e}") await managed_page.set_error(str(e)) raise finally: # Always return page to idle state await managed_page.set_idle()