#!/usr/bin/env python3 """ Competitive Intel — Autonomous competitor onboarding funnel analysis. ====================================================================== Navigates to a competitor's site, finds and completes signup autonomously, tracks every page transition with numbered screenshots, checks email for verification links, maps the full funnel, and generates a comprehensive JSON report with friction score, pricing discovery, and feature list. Usage (CLI): python testing/competitive_intel.py --url https://competitor.com python testing/competitive_intel.py --url https://competitor.com --screenshots --report python testing/competitive_intel.py --url https://competitor.com --email-provider mailslurp Programmatic: from testing.competitive_intel import CompetitiveIntelAgent agent = CompetitiveIntelAgent(url="https://competitor.com") report = await agent.run() """ from __future__ import annotations import argparse import asyncio import json import os import random import re import sys import time from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional from urllib.parse import urljoin, urlparse # --------------------------------------------------------------------------- # Path setup so module is importable from repo root or testing/ # --------------------------------------------------------------------------- _REPO_ROOT = Path(__file__).resolve().parent.parent if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) from testing.email_agent import ( EmailProvider, Inbox, extract_verification_link, get_email_provider, wait_for_email, ) from testing.scenarios.base import BaseScenario, ScenarioResult, StepResult SCREENSHOTS_BASE = Path("/mnt/e/genesis-system/testing/screenshots") REPORTS_DIR = Path("/mnt/e/genesis-system/testing/reports") # --------------------------------------------------------------------------- # Selector probes — tried in order for each element type # --------------------------------------------------------------------------- SIGNUP_LINK_SELECTORS = [ "a[href*=signup]", "a[href*=register]", "a[href*=sign-up]", "a[href*=create-account]", "a[href*=get-started]", "a[href*=free-trial]", "a[href*=start]", 'a:text-matches("sign up", "i")', 'a:text-matches("register", "i")', 'a:text-matches("get started", "i")', 'a:text-matches("start free", "i")', 'a:text-matches("try free", "i")', 'button:text-matches("sign up", "i")', 'button:text-matches("get started", "i")', 'button:text-matches("start free", "i")', ] EMAIL_FIELD_SELECTORS = [ "input[type=email]", "input[name=email]", "input[id*=email]", "input[placeholder*=email i]", "input[name=username]", "input[id*=username]", "input[autocomplete=email]", ] PASSWORD_FIELD_SELECTORS = [ "input[type=password]", "input[name=password]", "input[id*=password]", "input[placeholder*=password i]", ] NAME_FIELD_SELECTORS = [ "input[name=name]", "input[name=full_name]", "input[name=fullname]", "input[name=first_name]", "input[id*=name]", "input[placeholder*=name i]", "input[autocomplete=name]", ] SIGNUP_FORM_SELECTORS = [ "form[action*=register]", "form[action*=signup]", "form[action*=sign-up]", "form[action*=create]", "#signup", "#register", "#sign-up", "#create-account", "form", # last resort ] SUBMIT_BUTTON_SELECTORS = [ 'button[type=submit]', 'input[type=submit]', 'button:text-matches("sign up", "i")', 'button:text-matches("create account", "i")', 'button:text-matches("register", "i")', 'button:text-matches("get started", "i")', 'button:text-matches("continue", "i")', 'button:text-matches("next", "i")', "form button", ] PRICING_PAGE_SELECTORS = [ "a[href*=pricing]", "a[href*=plans]", "a[href*=subscribe]", 'a:text-matches("pricing", "i")', 'a:text-matches("plans", "i")', 'a:text-matches("upgrade", "i")', ] PRICE_AMOUNT_PATTERN = re.compile(r"[\$£€A\$]\s*[\d,]+(?:\.\d{1,2})?") FEATURE_BULLET_SELECTORS = [ "ul.features li", ".feature-list li", ".pricing-features li", ".plan-features li", '[class*=feature] li', '.benefits li', '[class*=benefit] li', ] # Login / authenticated state indicators DASHBOARD_INDICATORS = [ '[class*=dashboard]', '[class*=app-nav]', 'nav [href*=logout]', 'nav [href*=signout]', 'a[href*=logout]', 'a[href*=signout]', '[data-testid*=user]', '[class*=user-menu]', '[class*=account-menu]', '.sidebar', '#sidebar', '[class*=onboarding]', ] # Human-like first/last names for identity generation _FIRST_NAMES = [ "James", "Emma", "Oliver", "Ava", "William", "Sophia", "Noah", "Isabella", "Liam", "Mia", "Ethan", "Charlotte", "Mason", "Amelia", "Jacob", "Harper", "Michael", "Evelyn", "Jack", "Abigail", ] _LAST_NAMES = [ "Smith", "Johnson", "Williams", "Jones", "Brown", "Davis", "Miller", "Wilson", "Moore", "Taylor", "Anderson", "Thomas", "Jackson", "White", "Harris", "Martin", "Thompson", "Garcia", "Martinez", ] def _generate_identity(email_address: str) -> dict: """Generate a realistic human identity for signup forms.""" first = random.choice(_FIRST_NAMES) last = random.choice(_LAST_NAMES) suffix = random.randint(10, 99) password = f"Test{suffix}!Genesis{random.randint(100, 999)}" return { "first_name": first, "last_name": last, "full_name": f"{first} {last}", "email": email_address, "password": password, "company": f"{last} Consulting", "phone": f"+614{''.join([str(random.randint(0, 9)) for _ in range(8)])}", } def _domain_from_url(url: str) -> str: parsed = urlparse(url) return parsed.netloc.replace("www.", "") or url[:40] def _safe_dirname(domain: str) -> str: return re.sub(r"[^\w\-.]", "_", domain) # --------------------------------------------------------------------------- # Friction scoring # --------------------------------------------------------------------------- def compute_friction_score(funnel_steps: list[dict]) -> float: """ Score the onboarding friction on a 0–10 scale. 0 = frictionless, 10 = extremely painful. Factors: - Number of form fields encountered - Email verification required (adds friction) - Number of funnel steps before "first value" - Any errors/failures - Time to complete (seconds) """ score = 0.0 total_fields = sum(s.get("fields_filled", 0) for s in funnel_steps) verification_required = any(s.get("step_type") == "email_verification" for s in funnel_steps) failed_steps = sum(1 for s in funnel_steps if s.get("status") not in ("pass", "skipped")) step_count = len(funnel_steps) total_time_s = sum(s.get("duration_ms", 0) for s in funnel_steps) / 1000.0 # Field count: every 2 extra fields above 2 = +1 friction score += max(0, (total_fields - 2) / 2) # Verification required = +2 score += 2.0 if verification_required else 0.0 # Step count: every 2 steps above 3 = +0.5 score += max(0, (step_count - 3) / 2 * 0.5) # Failures = +1 each score += failed_steps * 1.0 # Time: >120s = +1, >300s = +2 if total_time_s > 300: score += 2.0 elif total_time_s > 120: score += 1.0 return round(min(score, 10.0), 2) # --------------------------------------------------------------------------- # CompetitiveIntelAgent # --------------------------------------------------------------------------- class CompetitiveIntelAgent(BaseScenario): """ Full autonomous competitor onboarding agent. Inherits BaseScenario for Playwright helpers + screenshot infrastructure. Adds: identity generation, autonomous form discovery, email verification, pricing extraction, feature extraction, funnel mapping, report generation. """ name = "competitive_intel" def __init__( self, url: str, email_provider: Optional[EmailProvider] = None, save_screenshots: bool = True, email_wait_timeout: int = 90, ) -> None: super().__init__() self.target_url = url.rstrip("/") self.domain = _domain_from_url(url) self.safe_domain = _safe_dirname(self.domain) self.save_screenshots = save_screenshots self.email_wait_timeout = email_wait_timeout self._email_provider: Optional[EmailProvider] = email_provider self._inbox: Optional[Inbox] = None self._identity: Optional[dict] = None self._funnel_steps: list[dict] = [] self._screenshot_counter: int = 0 self._screenshots_dir = SCREENSHOTS_BASE / f"competitor_{self.safe_domain}" self._screenshots_dir.mkdir(parents=True, exist_ok=True) REPORTS_DIR.mkdir(parents=True, exist_ok=True) self._start_time: float = time.time() # ------------------------------------------------------------------ # Entry point (called via execute() from BaseScenario) # ------------------------------------------------------------------ async def run(self, page: Any, config: dict) -> None: self._page = page self._config = config self._timeout = config.get("default_timeout", 20_000) # --- Phase 1: Create email inbox + identity await self._phase_create_identity() # --- Phase 2: Navigate to homepage await self._phase_homepage() # --- Phase 3: Find & navigate to signup await self._phase_find_signup() # --- Phase 4: Fill & submit signup form await self._phase_signup_form() # --- Phase 5: Handle post-signup page (may be dashboard or verification gate) await self._phase_post_signup() # --- Phase 6: Check email for verification await self._phase_email_verification() # --- Phase 7: Discover pricing await self._phase_discover_pricing() # --- Phase 8: Generate report await self._phase_generate_report() # ------------------------------------------------------------------ # Phase implementations # ------------------------------------------------------------------ async def _phase_create_identity(self) -> None: step = self._new_step("create_identity") try: provider = self._email_provider or get_email_provider() self._inbox = await provider.create_inbox( name=f"ci_{self.safe_domain}_{int(time.time())}" ) self._email_provider = provider self._identity = _generate_identity(self._inbox.email_address) self._finish_step( step, "pass", f"Identity: {self._identity['full_name']} <{self._inbox.email_address}>" ) self._record_funnel( step_type="identity_creation", step_name="create_identity", status="pass", details={"email": self._inbox.email_address, "name": self._identity["full_name"]}, ) except Exception as exc: self._finish_step(step, "error", f"Failed to create inbox: {exc}") self._record_funnel("identity_creation", "create_identity", "error", {"error": str(exc)}) async def _phase_homepage(self) -> None: step = self._new_step("navigate_homepage") t0 = time.time() try: await self._page.goto(self.target_url, timeout=self._timeout, wait_until="domcontentloaded") title = await self._page.title() screenshot = await self._numbered_screenshot("homepage") duration_ms = round((time.time() - t0) * 1000, 1) self._finish_step(step, "pass", f"Title: {title}") self._record_funnel( "homepage", "navigate_homepage", "pass", {"url": self.target_url, "title": title}, screenshot=screenshot, duration_ms=duration_ms, ) except Exception as exc: screenshot = await self._numbered_screenshot("homepage_error") self._finish_step(step, "error", str(exc)) self._record_funnel("homepage", "navigate_homepage", "error", {"error": str(exc)}, screenshot=screenshot) async def _phase_find_signup(self) -> None: step = self._new_step("find_signup_link") t0 = time.time() try: found_selector = None for sel in SIGNUP_LINK_SELECTORS: try: el = await self._page.query_selector(sel) if el: found_selector = sel break except Exception: continue if found_selector: href = await self._page.get_attribute(found_selector, "href") or "" self._log(f"Signup link found: {found_selector} → {href}") await self._page.click(found_selector, timeout=self._timeout) await self._page.wait_for_load_state("domcontentloaded", timeout=self._timeout) screenshot = await self._numbered_screenshot("signup_page") duration_ms = round((time.time() - t0) * 1000, 1) self._finish_step(step, "pass", f"Clicked signup link ({found_selector})") self._record_funnel( "find_signup", "find_signup_link", "pass", {"selector_used": found_selector, "href": href}, screenshot=screenshot, duration_ms=duration_ms, ) else: # Try navigating directly to common signup URLs base = self.target_url for path in ["/signup", "/register", "/sign-up", "/create-account", "/get-started"]: try: await self._page.goto(urljoin(base, path), timeout=self._timeout, wait_until="domcontentloaded") # Check if we got a real page (not 404) status_text = await self._page.title() if status_text and "404" not in status_text.lower(): screenshot = await self._numbered_screenshot("signup_page_direct") duration_ms = round((time.time() - t0) * 1000, 1) self._finish_step(step, "pass", f"Direct nav to {path} worked") self._record_funnel( "find_signup", "find_signup_link", "pass", {"method": "direct_nav", "path": path}, screenshot=screenshot, duration_ms=duration_ms, ) return except Exception: continue screenshot = await self._numbered_screenshot("signup_not_found") self._finish_step(step, "fail", "No signup link or form page found") self._record_funnel("find_signup", "find_signup_link", "fail", {"tried": SIGNUP_LINK_SELECTORS[:5]}, screenshot=screenshot) except Exception as exc: screenshot = await self._numbered_screenshot("find_signup_error") self._finish_step(step, "error", str(exc)) self._record_funnel("find_signup", "find_signup_link", "error", {"error": str(exc)}, screenshot=screenshot) async def _phase_signup_form(self) -> None: step = self._new_step("fill_signup_form") t0 = time.time() fields_filled = 0 try: # Wait briefly for form to settle await self._page.wait_for_load_state("domcontentloaded", timeout=self._timeout) # --- Name fields (optional — try but don't fail if missing) for sel in NAME_FIELD_SELECTORS: el = await self._page.query_selector(sel) if el: visible = await el.is_visible() if visible: # Detect first vs full name name_attr = await el.get_attribute("name") or "" if "first" in name_attr.lower(): await el.fill(self._identity["first_name"]) else: await el.fill(self._identity["full_name"]) fields_filled += 1 self._log(f"Filled name field: {sel}") break # --- Email field email_filled = False for sel in EMAIL_FIELD_SELECTORS: el = await self._page.query_selector(sel) if el and await el.is_visible(): await el.fill(self._identity["email"]) fields_filled += 1 email_filled = True self._log(f"Filled email: {sel}") break if not email_filled: screenshot = await self._numbered_screenshot("no_email_field") self._finish_step(step, "fail", "No email field found on signup form") self._record_funnel("signup_form", "fill_signup_form", "fail", {"reason": "no_email_field"}, screenshot=screenshot) return # --- Password field pw_filled = False for sel in PASSWORD_FIELD_SELECTORS: el = await self._page.query_selector(sel) if el and await el.is_visible(): await el.fill(self._identity["password"]) fields_filled += 1 pw_filled = True self._log(f"Filled password: {sel}") break # Some sites (Google, social login) have no password at signup step if not pw_filled: self._log("No password field found — may be OAuth or email-only flow") # --- Company/phone (try but don't fail) for try_sel, value in [ ("input[name*=company i]", self._identity["company"]), ("input[placeholder*=company i]", self._identity["company"]), ("input[name*=phone i]", self._identity["phone"]), ("input[placeholder*=phone i]", self._identity["phone"]), ]: el = await self._page.query_selector(try_sel) if el and await el.is_visible(): await el.fill(value) fields_filled += 1 self._log(f"Filled optional field: {try_sel}") # --- Human delay before submitting (avoids bot detection) await asyncio.sleep(random.uniform(0.5, 1.5)) # Screenshot pre-submit await self._numbered_screenshot("pre_submit") # --- Submit submitted = False for sel in SUBMIT_BUTTON_SELECTORS: el = await self._page.query_selector(sel) if el and await el.is_visible(): disabled = await el.is_disabled() if not disabled: await el.click(timeout=self._timeout) submitted = True self._log(f"Clicked submit: {sel}") break if not submitted: screenshot = await self._numbered_screenshot("no_submit_button") self._finish_step(step, "fail", "Could not find enabled submit button") self._record_funnel("signup_form", "fill_signup_form", "fail", {"fields_filled": fields_filled, "reason": "no_submit"}, screenshot=screenshot) return # Wait for navigation or confirmation try: await self._page.wait_for_load_state("domcontentloaded", timeout=self._timeout) except Exception: pass # Some SPAs don't trigger navigation events screenshot = await self._numbered_screenshot("post_submit") duration_ms = round((time.time() - t0) * 1000, 1) self._finish_step(step, "pass", f"Form submitted ({fields_filled} fields filled)") self._record_funnel( "signup_form", "fill_signup_form", "pass", {"fields_filled": fields_filled, "email": self._identity["email"]}, screenshot=screenshot, duration_ms=duration_ms, fields_filled=fields_filled, ) except Exception as exc: screenshot = await self._numbered_screenshot("signup_form_error") self._finish_step(step, "error", str(exc)) self._record_funnel("signup_form", "fill_signup_form", "error", {"error": str(exc), "fields_filled": fields_filled}, screenshot=screenshot, fields_filled=fields_filled) async def _phase_post_signup(self) -> None: step = self._new_step("post_signup_state") t0 = time.time() try: # Small wait for any redirect/animation await asyncio.sleep(2) current_url = self._page.url title = await self._page.title() page_text = await self._page.inner_text("body") # Detect state: dashboard, email verify gate, or error state = "unknown" for sel in DASHBOARD_INDICATORS: el = await self._page.query_selector(sel) if el and await el.is_visible(): state = "authenticated_dashboard" break if state == "unknown": verify_keywords = ["verify", "confirm", "check your email", "we sent", "activation", "spam"] page_lower = page_text.lower() if any(kw in page_lower for kw in verify_keywords): state = "email_verification_required" elif "error" in page_lower or "invalid" in page_lower: state = "signup_error" elif current_url != self.target_url: state = "redirected" screenshot = await self._numbered_screenshot(f"post_signup_{state}") duration_ms = round((time.time() - t0) * 1000, 1) self._finish_step(step, "pass", f"Post-signup state: {state}") self._record_funnel( "post_signup", "post_signup_state", "pass", {"state": state, "url": current_url, "title": title}, screenshot=screenshot, duration_ms=duration_ms, ) # Store state for email verification phase self._post_signup_state = state except Exception as exc: screenshot = await self._numbered_screenshot("post_signup_error") self._finish_step(step, "error", str(exc)) self._record_funnel("post_signup", "post_signup_state", "error", {"error": str(exc)}, screenshot=screenshot) self._post_signup_state = "error" async def _phase_email_verification(self) -> None: step = self._new_step("email_verification") t0 = time.time() if not self._inbox or not self._email_provider: self._finish_step(step, "skipped", "No inbox configured") self._record_funnel("email_verification", "email_verification", "skipped", {}) return try: self._log(f"Waiting for verification email (up to {self.email_wait_timeout}s)...") email = await wait_for_email( self._email_provider, self._inbox.id, subject_contains=None, # grab first email timeout=self.email_wait_timeout, poll_interval=5.0, ) subject = email.subject or "(no subject)" self._log(f"Email received: {subject}") # Extract verification link verify_url = None if email.html: verify_url = extract_verification_link(email.html) elif email.text: # Fallback: regex scan plain text urls = re.findall(r"https?://\S+", email.text) verify_url = next((u for u in urls if re.search( r"verif|confirm|activat", u, re.IGNORECASE )), urls[0] if urls else None) if verify_url: self._log(f"Verification link: {verify_url}") await self._page.goto(verify_url, timeout=self._timeout, wait_until="domcontentloaded") await asyncio.sleep(2) screenshot = await self._numbered_screenshot("after_email_verification") duration_ms = round((time.time() - t0) * 1000, 1) self._finish_step(step, "pass", f"Verified via: {verify_url[:80]}") self._record_funnel( "email_verification", "email_verification", "pass", { "subject": subject, "verify_url": verify_url, "step_type": "email_verification", }, screenshot=screenshot, duration_ms=duration_ms, ) else: screenshot = await self._numbered_screenshot("no_verify_link") self._finish_step(step, "fail", f"Email received but no verification link found (subject: {subject})") self._record_funnel( "email_verification", "email_verification", "fail", {"subject": subject, "reason": "no_link_found"}, screenshot=screenshot, ) except TimeoutError: self._log("No verification email received within timeout — continuing") self._finish_step(step, "skipped", "No email within timeout — product may not require it") self._record_funnel("email_verification", "email_verification", "skipped", {"reason": "timeout", "step_type": "email_verification"}) except Exception as exc: self._finish_step(step, "error", str(exc)) self._record_funnel("email_verification", "email_verification", "error", {"error": str(exc)}) async def _phase_discover_pricing(self) -> None: step = self._new_step("discover_pricing") t0 = time.time() try: # Try to navigate to pricing page from current state pricing_url = None for sel in PRICING_PAGE_SELECTORS: el = await self._page.query_selector(sel) if el: href = await el.get_attribute("href") or "" if href: pricing_url = urljoin(self._page.url, href) break prices: list[str] = [] features: list[str] = [] pricing_page_url = "" if pricing_url: await self._page.goto(pricing_url, timeout=self._timeout, wait_until="domcontentloaded") pricing_page_url = self._page.url page_text = await self._page.inner_text("body") # Extract price amounts prices = list(set(PRICE_AMOUNT_PATTERN.findall(page_text))) # Extract feature bullets for sel in FEATURE_BULLET_SELECTORS: items = await self._page.query_selector_all(sel) if items: for item in items[:30]: # cap at 30 try: text = (await item.inner_text()).strip() if text and len(text) < 200: features.append(text) except Exception: continue if features: break # Also check current page for any price signals if not prices: try: body = await self._page.inner_text("body") prices = list(set(PRICE_AMOUNT_PATTERN.findall(body))) except Exception: pass screenshot = await self._numbered_screenshot("pricing_page") duration_ms = round((time.time() - t0) * 1000, 1) self._finish_step( step, "pass", f"Pricing: {len(prices)} prices found, {len(features)} features" ) self._record_funnel( "pricing_discovery", "discover_pricing", "pass", { "pricing_url": pricing_page_url, "prices_found": prices, "features_found": features[:20], }, screenshot=screenshot, duration_ms=duration_ms, ) # Store for report self._pricing_data = {"prices": prices, "features": features[:30], "url": pricing_page_url} except Exception as exc: self._finish_step(step, "error", str(exc)) self._record_funnel("pricing_discovery", "discover_pricing", "error", {"error": str(exc)}) self._pricing_data = {"prices": [], "features": [], "url": ""} async def _phase_generate_report(self) -> None: step = self._new_step("generate_report") try: elapsed_s = round(time.time() - self._start_time, 1) friction = compute_friction_score(self._funnel_steps) ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") # Time-to-first-value: seconds until first "pass" after signup form ttfv_s = None passed_after_signup = [ s for s in self._funnel_steps if s.get("step_type") not in ("identity_creation", "homepage", "find_signup", "signup_form") and s.get("status") == "pass" ] if passed_after_signup: ttfv_s = round( sum(s.get("duration_ms", 0) for s in self._funnel_steps[: self._funnel_steps.index(passed_after_signup[0]) + 1 ]) / 1000.0, 1 ) report = { "meta": { "competitor_url": self.target_url, "domain": self.domain, "generated_at": ts, "agent_version": "1.0.0", "identity_used": { "name": self._identity["full_name"] if self._identity else None, "email": self._inbox.email_address if self._inbox else None, }, }, "summary": { "funnel_step_count": len(self._funnel_steps), "total_time_seconds": elapsed_s, "time_to_first_value_seconds": ttfv_s, "friction_score": friction, "friction_explanation": _friction_label(friction), "screenshots_taken": self._screenshot_counter, "email_verification_required": any( s.get("step_type") == "email_verification" and s.get("status") == "pass" for s in self._funnel_steps ), }, "pricing": getattr(self, "_pricing_data", {"prices": [], "features": [], "url": ""}), "funnel": self._funnel_steps, "screenshots_dir": str(self._screenshots_dir), } report_path = REPORTS_DIR / f"competitor_{self.safe_domain}_{ts}.json" report_path.write_text(json.dumps(report, indent=2)) self._finish_step(step, "pass", f"Report saved: {report_path}") self._log(f"Friction score: {friction}/10 ({_friction_label(friction)})") self._log(f"Funnel steps: {len(self._funnel_steps)}") self._log(f"Report: {report_path}") self._report_path = str(report_path) self._report = report except Exception as exc: self._finish_step(step, "error", f"Report generation failed: {exc}") # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ def _record_funnel( self, step_type: str, step_name: str, status: str, details: dict, screenshot: str = "", duration_ms: float = 0.0, fields_filled: int = 0, ) -> None: self._funnel_steps.append({ "step_number": len(self._funnel_steps) + 1, "step_type": step_type, "step_name": step_name, "status": status, "screenshot_path": screenshot, "duration_ms": duration_ms, "fields_filled": fields_filled, "details": details, }) async def _numbered_screenshot(self, label: str) -> str: """Capture screenshot with sequential number prefix.""" if not self.save_screenshots: return "" try: self._screenshot_counter += 1 filename = f"{self._screenshot_counter:03d}_{label}.png" path = self._screenshots_dir / filename await self._page.screenshot(path=str(path), full_page=True) self._log(f"[SHOT {self._screenshot_counter:03d}] {path.name}") return str(path) except Exception as exc: self._log(f"[SHOT-ERR] {exc}") return "" def _friction_label(score: float) -> str: if score <= 1.5: return "frictionless" if score <= 3.0: return "low_friction" if score <= 5.0: return "moderate_friction" if score <= 7.0: return "high_friction" return "very_high_friction" # --------------------------------------------------------------------------- # Standalone runner (programmatic + CLI) # --------------------------------------------------------------------------- async def run_competitive_intel( url: str, email_provider_name: Optional[str] = None, save_screenshots: bool = True, email_wait_timeout: int = 90, headless: bool = True, ) -> dict: """ Run competitive intel against a URL. Returns the full report dict. Args: url: Target competitor URL. email_provider_name: "mailslurp", "agentmail", "inbucket", or None (auto). save_screenshots: Save screenshots to testing/screenshots/competitor_{domain}/. email_wait_timeout: Seconds to wait for verification email. headless: Run browser headlessly. Returns: Report dict (same as written to testing/reports/). """ try: from playwright.async_api import async_playwright except ImportError: raise ImportError( "playwright is required: pip install playwright && playwright install chromium" ) provider = get_email_provider(email_provider_name) if email_provider_name else get_email_provider() agent = CompetitiveIntelAgent( url=url, email_provider=provider, save_screenshots=save_screenshots, email_wait_timeout=email_wait_timeout, ) async with async_playwright() as pw: browser = await pw.chromium.launch( headless=headless, args=[ "--disable-blink-features=AutomationControlled", "--no-sandbox", ], ) context = await browser.new_context( viewport={"width": 1280, "height": 800}, user_agent=( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/121.0.0.0 Safari/537.36" ), locale="en-AU", ) page = await context.new_page() config = { "product": agent.safe_domain, "default_timeout": 20_000, } result = await agent.execute(page, config) await browser.close() return getattr(agent, "_report", {"error": "report not generated", "result": result.to_dict()}) # --------------------------------------------------------------------------- # CLI entry point # --------------------------------------------------------------------------- def _build_cli_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Competitive Intel — autonomous competitor funnel analysis", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python testing/competitive_intel.py --url https://competitor.com python testing/competitive_intel.py --url https://competitor.com --screenshots --report python testing/competitive_intel.py --url https://competitor.com --email-provider mailslurp --no-headless """, ) parser.add_argument("--url", required=True, help="Competitor URL to analyze") parser.add_argument( "--email-provider", default=None, choices=["mailslurp", "agentmail", "inbucket"], help="Email provider (auto-detected from env if not set)", ) parser.add_argument( "--screenshots", action="store_true", default=True, help="Save screenshots (default: enabled)", ) parser.add_argument( "--no-screenshots", action="store_false", dest="screenshots", help="Disable screenshot capture", ) parser.add_argument( "--report", action="store_true", default=True, help="Generate JSON report (default: enabled)", ) parser.add_argument( "--no-headless", action="store_true", help="Run browser in visible (non-headless) mode", ) parser.add_argument( "--email-timeout", type=int, default=90, help="Seconds to wait for verification email (default: 90)", ) return parser async def _cli_main() -> None: parser = _build_cli_parser() args = parser.parse_args() print(f"\n[Competitive Intel] Target: {args.url}") print(f"[Competitive Intel] Email provider: {args.email_provider or 'auto'}") print(f"[Competitive Intel] Screenshots: {args.screenshots}") print(f"[Competitive Intel] Headless: {not args.no_headless}\n") report = await run_competitive_intel( url=args.url, email_provider_name=args.email_provider, save_screenshots=args.screenshots, email_wait_timeout=args.email_timeout, headless=not args.no_headless, ) # Print summary summary = report.get("summary", {}) pricing = report.get("pricing", {}) print("\n" + "=" * 60) print("COMPETITIVE INTEL REPORT") print("=" * 60) print(f"Domain: {report.get('meta', {}).get('domain', args.url)}") print(f"Funnel steps: {summary.get('funnel_step_count', '?')}") print(f"Total time: {summary.get('total_time_seconds', '?')}s") print(f"Time to first value: {summary.get('time_to_first_value_seconds', '?')}s") print(f"Friction score: {summary.get('friction_score', '?')}/10 " f"({summary.get('friction_explanation', '')})") print(f"Email verification: {summary.get('email_verification_required', '?')}") print(f"Screenshots taken: {summary.get('screenshots_taken', 0)}") print(f"Prices found: {pricing.get('prices', [])}") print(f"Features found: {len(pricing.get('features', []))} items") if pricing.get("features"): for f in pricing["features"][:5]: print(f" - {f}") if len(pricing["features"]) > 5: print(f" ... and {len(pricing['features']) - 5} more") print("=" * 60) # Print report path meta = report.get("meta", {}) domain = meta.get("domain", _safe_dirname(args.url)) ts = meta.get("generated_at", datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")) expected_path = REPORTS_DIR / f"competitor_{_safe_dirname(domain)}_{ts}.json" if expected_path.exists(): print(f"\nReport saved: {expected_path}") else: print(f"\nReport dir: {REPORTS_DIR}") if __name__ == "__main__": asyncio.run(_cli_main())