diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index adc00831..403b19ef 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -11,7 +11,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python-version: ["3.10", "3.11", "3.12"] + python-version: ["3.11", "3.12"] steps: - uses: actions/checkout@v4 diff --git a/pyproject.toml b/pyproject.toml index 8517010c..215d1045 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,7 +8,7 @@ version = "1.1.27" description = "NLP2CMD - Transforms natural language into domain-specific commands (SQL, Shell, Docker, Kubernetes) using a multi-layered detection pipeline and thermodynamic optimization." readme = "README.md" license = "Apache-2.0" -requires-python = ">=3.10" +requires-python = ">=3.11" authors = [ { name = "NLP2CMD Team", email = "tom@sapletta.com" } ] @@ -34,7 +34,6 @@ classifiers = [ "Intended Audience :: Developers", "Operating System :: OS Independent", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Topic :: Software Development :: Code Generators", diff --git a/src/nlp2cmd/cli/commands/doctor.py b/src/nlp2cmd/cli/commands/doctor.py index b990c193..1dd5aa7b 100644 --- a/src/nlp2cmd/cli/commands/doctor.py +++ b/src/nlp2cmd/cli/commands/doctor.py @@ -11,17 +11,11 @@ from __future__ import annotations +import json import os import sys -import json -import socket -import subprocess -import shutil -import time from pathlib import Path from typing import Optional -from dataclasses import dataclass, field -from enum import Enum try: import requests @@ -39,37 +33,17 @@ HAS_RICH = False Console = None -# Import modular browser token retriever -try: - from nlp2cmd.browser_token import HFTokenRetriever, TokenConfig - _BROWSER_TOKEN_AVAILABLE = True -except ImportError: - _BROWSER_TOKEN_AVAILABLE = False - -# Import modular browser manager -try: - from nlp2cmd.browser_manager import ExistingBrowserManager, BrowserConfig - _BROWSER_MANAGER_AVAILABLE = True -except ImportError: - _BROWSER_MANAGER_AVAILABLE = False - - -class Status(Enum): - OK = "ok" - WARNING = "warning" - ERROR = "error" - INFO = "info" - FIXED = "fixed" +from nlp2cmd.cli.commands.doctor_browser import get_hf_token_via_browser +from nlp2cmd.cli.commands.doctor_types import CheckResult, Status - -@dataclass -class CheckResult: - name: str - status: Status - message: str - details: dict = field(default_factory=dict) - fix_applied: bool = False - fix_command: Optional[str] = None +__all__ = [ + "CheckResult", + "NP2CMDDoctor", + "Status", + "doctor_command", + "get_hf_token_via_browser", + "run_doctor", +] class NP2CMDDoctor: @@ -590,680 +564,6 @@ def run_doctor(auto_fix: bool = False, output_json: bool = False, fix_script: Op return not has_errors -def get_hf_token_via_browser(console: Optional[Console] = None) -> Optional[str]: - """Open browser to help user get HF_TOKEN from Hugging Face. - - Browser priority: - 1. Connect to existing browser (Firefox/Chrome via CDP) - 2. Open new system browser (firefox/chrome commands) - 3. Use Playwright (requires manual login) - - Returns the token if successfully retrieved. - """ - if console: - console.print("[cyan]🌐 Opening Hugging Face in browser...[/cyan]") - else: - print("🌐 Opening Hugging Face in browser...") - - # Try priority 1: Connect to existing browser - token = _try_existing_browser(console) - if token: - return token - - # Try priority 2: Open new system browser - token = _try_system_browser(console) - if token: - return token - - # Try priority 3: Use Playwright - token = _try_playwright_browser(console) - if token: - return token - - return None - - -def _try_existing_browser(console: Optional[Console] = None) -> Optional[str]: - """Try to connect to existing browser via CDP with detailed logging.""" - import socket - - if console: - console.print("[dim] [Stage 1/3] Checking for existing browser...[/dim]") - - # Check common CDP ports - cdp_ports = [9222, 9223, 9224, 9333] - found_port = None - - for port in cdp_ports: - if console: - console.print(f"[dim] Checking port {port}...[/dim]") - - try: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(1) - result = sock.connect_ex(("localhost", port)) - sock.close() - - if result == 0: - found_port = port - if console: - console.print(f"[green] βœ“ Found browser on port {port}[/green]") - break - else: - if console: - console.print(f"[dim] Port {port}: not available[/dim]") - except Exception as e: - if console: - console.print(f"[dim] Port {port}: error - {e}[/dim]") - - if not found_port: - if console: - console.print("[dim] β„Ή No existing browser with CDP found[/dim]") - console.print("[dim] Tip: Run 'firefox --remote-debugging-port=9222' first[/dim]") - return None - - # Try to connect via Playwright CDP - if console: - console.print(f"[cyan] β†’ Connecting via Playwright to port {found_port}...[/cyan]") - - connection_success = False - browser = None - browser_type = None - - try: - from playwright.sync_api import sync_playwright - - with sync_playwright() as p: - # Try Chrome/Chromium first - try: - browser = p.chromium.connect_over_cdp(f"http://localhost:{found_port}") - browser_type = "Chrome/Chromium" - connection_success = True - if console: - console.print(f"[green] βœ“ Connected to {browser_type} via CDP[/green]") - except Exception as chrome_err: - if console: - console.print(f"[dim] Chromium CDP failed: {str(chrome_err)[:50]}...[/dim]") - - # Try Firefox - try: - browser = p.firefox.connect_over_cdp(f"http://localhost:{found_port}") - browser_type = "Firefox" - connection_success = True - if console: - console.print(f"[green] βœ“ Connected to Firefox via CDP[/green]") - except Exception as firefox_err: - if console: - console.print(f"[red] βœ— CDP connection failed for both browsers[/red]") - console.print(f"[dim] Chrome error: {str(chrome_err)[:30]}...[/dim]") - console.print(f"[dim] Firefox error: {str(firefox_err)[:30]}...[/dim]") - return None - - if not connection_success or not browser: - if console: - console.print(f"[red] βœ— Browser connection established but browser object is None[/red]") - return None - - # Create new context and page - if console: - console.print(f"[dim] Creating browser context...[/dim]") - - try: - context = browser.new_context() - page = context.new_page() - if console: - console.print(f"[green] βœ“ Browser context created[/green]") - except Exception as ctx_err: - if console: - console.print(f"[red] βœ— Failed to create browser context: {ctx_err}[/red]") - return None - - # Navigate - if console: - console.print(f"[cyan] β†’ Navigating to huggingface.co...[/cyan]") - - nav_success = False - expected_url_pattern = "huggingface.co/settings/tokens" - actual_url = None - - try: - page.goto("https://huggingface.co/settings/tokens", timeout=30000) - actual_url = page.url - - # Verify we reached the expected page (or at least HF domain) - if expected_url_pattern in actual_url: - nav_success = True - if console: - console.print(f"[green] βœ“ Page loaded at correct URL[/green]") - elif "huggingface.co/login" in actual_url: - # This is expected if not logged in, but we should warn - nav_success = True # Page loaded, just needs login - if console: - console.print(f"[yellow] ⚠ Page loaded but requires login first[/yellow]") - console.print(f"[dim] URL: {actual_url}[/dim]") - elif "huggingface.co" in actual_url: - nav_success = True - if console: - console.print(f"[yellow] ⚠ Page loaded on HF domain but different path[/yellow]") - console.print(f"[dim] URL: {actual_url}[/dim]") - else: - if console: - console.print(f"[red] βœ— Page loaded but unexpected URL[/red]") - console.print(f"[dim] Expected: {expected_url_pattern}[/dim]") - console.print(f"[dim] Actual: {actual_url}[/dim]") - - except Exception as e: - if console: - console.print(f"[red] βœ— Navigation failed: {e}[/red]") - if actual_url: - console.print(f"[dim] Last URL: {actual_url}[/dim]") - nav_success = False - - except ImportError: - if console: - console.print(f"[red] βœ— Playwright not installed[/red]") - return None - except Exception as e: - if console: - console.print(f"[red] βœ— CDP connection error: {e}[/red]") - return None - - -def _try_system_browser(console: Optional[Console] = None) -> Optional[str]: - """Try to open new system browser with detailed stage logging.""" - import subprocess - import time - import socket - - if console: - console.print("[dim] [Stage 2/3] Opening system browser...[/dim]") - - # Try to open Firefox or Chrome/Chromium - browsers_to_try = [ - ("firefox", ["firefox", "--new-window", "--remote-debugging-port=9222"]), - ("google-chrome", ["google-chrome", "--new-window", "--remote-debugging-port=9222"]), - ("chromium", ["chromium", "--new-window", "--remote-debugging-port=9222"]), - ("chromium-browser", ["chromium-browser", "--new-window", "--remote-debugging-port=9222"]), - ] - - for browser_name, cmd in browsers_to_try: - try: - # Stage 2.1: Check if browser binary exists - if console: - console.print(f"[dim] Checking {browser_name}...[/dim]") - - result = subprocess.run(["which", browser_name], capture_output=True, timeout=5) - if result.returncode != 0: - if console: - console.print(f"[dim] βœ— {browser_name} not found[/dim]") - continue - - if console: - console.print(f"[cyan] β†’ Launching {browser_name}...[/cyan]") - - # Stage 2.2: Launch browser with CDP enabled - try: - process = subprocess.Popen( - cmd + ["about:blank"], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - start_new_session=True - ) - if console: - console.print(f"[dim] PID: {process.pid}[/dim]") - except Exception as e: - if console: - console.print(f"[red] βœ— Failed to launch: {e}[/red]") - continue - - # Stage 2.3: Wait for browser to initialize - if console: - console.print("[dim] Waiting for browser to start (3s)...[/dim]") - time.sleep(3) - - # Stage 2.4: Try to connect via CDP (with actual protocol verification) - if console: - console.print("[dim] Checking CDP port 9222 (with protocol verification)...[/dim]") - - cdp_available = False - for attempt in range(5): - try: - # First: basic TCP check - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(2) - result = sock.connect_ex(("localhost", 9222)) - sock.close() - - if result == 0: - # Second: verify it's actually a CDP endpoint by making HTTP request - import urllib.request - try: - response = urllib.request.urlopen( - "http://localhost:9222/json/version", - timeout=3 - ) - cdp_info = response.read().decode('utf-8') - if 'Browser' in cdp_info or 'Protocol-Version' in cdp_info: - cdp_available = True - if console: - console.print(f"[green] βœ“ CDP port 9222 ready (verified protocol)[/green]") - break - else: - if console: - console.print(f"[dim] Attempt {attempt+1}/5: port open but not CDP protocol[/dim]") - except Exception as http_err: - if console: - console.print(f"[dim] Attempt {attempt+1}/5: port open but CDP check failed: {str(http_err)[:40]}[/dim]") - else: - if console: - console.print(f"[dim] Attempt {attempt+1}/5: port not ready (code: {result})[/dim]") - except Exception as e: - if console: - console.print(f"[dim] Attempt {attempt+1}/5: {str(e)[:40]}[/dim]") - time.sleep(1) - - if not cdp_available: - if console: - console.print(f"[yellow] βœ— CDP not available after 5 attempts[/yellow]") - console.print(f"[dim] Browser launched but CDP protocol not responding[/dim]") - continue - - # Stage 2.5: Connect with Playwright - if console: - console.print(f"[cyan] β†’ Connecting via Playwright CDP...[/cyan]") - - connection_success = False - browser = None - - try: - from playwright.sync_api import sync_playwright - - with sync_playwright() as p: - try: - browser = p.chromium.connect_over_cdp("http://localhost:9222") - connection_success = True - if console: - console.print(f"[green] βœ“ Connected to {browser_name} via CDP[/green]") - except Exception as e: - if console: - console.print(f"[yellow] ⚠ CDP connect failed: {str(e)[:50]}...[/yellow]") - console.print(f"[dim] Falling back to manual mode...[/dim]") - return _manual_browser_instructions(console, browser_name) - - if not connection_success or not browser: - if console: - console.print(f"[red] βœ— Connection reported success but browser is None[/red]") - return _manual_browser_instructions(console, browser_name) - - try: - context = browser.new_context() - page = context.new_page() - if console: - console.print(f"[green] βœ“ Browser context created[/green]") - except Exception as ctx_err: - if console: - console.print(f"[red] βœ— Failed to create context: {ctx_err}[/red]") - return _manual_browser_instructions(console, browser_name) - - # Stage 2.6: Navigate to HF - if console: - console.print(f"[cyan] β†’ Navigating to huggingface.co...[/cyan]") - - nav_success = False - expected_url_pattern = "huggingface.co/settings/tokens" - actual_url = None - - try: - page.goto("https://huggingface.co/settings/tokens", timeout=30000) - actual_url = page.url - - # Verify we reached the expected page (or at least HF domain) - if expected_url_pattern in actual_url: - nav_success = True - if console: - console.print(f"[green] βœ“ Page loaded at correct URL[/green]") - elif "huggingface.co/login" in actual_url: - nav_success = True # Page loaded, just needs login - if console: - console.print(f"[yellow] ⚠ Page loaded but requires login first[/yellow]") - console.print(f"[dim] URL: {actual_url}[/dim]") - elif "huggingface.co" in actual_url: - nav_success = True - if console: - console.print(f"[yellow] ⚠ Page loaded on HF domain but different path[/yellow]") - console.print(f"[dim] URL: {actual_url}[/dim]") - else: - if console: - console.print(f"[red] βœ— Page loaded but unexpected URL[/red]") - console.print(f"[dim] Expected: {expected_url_pattern}[/dim]") - console.print(f"[dim] Actual: {actual_url}[/dim]") - - except Exception as e: - if console: - console.print(f"[red] βœ— Navigation failed: {e}[/red]") - if actual_url: - console.print(f"[dim] Last URL: {actual_url}[/dim]") - nav_success = False - - # Stage 2.7: Get token from user (even if nav had issues, let user try) - if nav_success or actual_url: # Only proceed if we at least loaded something - return _navigate_and_get_token(page, console, browser_name) - else: - if console: - console.print(f"[red] βœ— Cannot proceed - page did not load[/red]") - return None - - except ImportError: - if console: - console.print(f"[red] βœ— Playwright not installed[/red]") - return _manual_browser_instructions(console, browser_name) - except Exception as e: - if console: - console.print(f"[red] βœ— Playwright error: {e}[/red]") - return _manual_browser_instructions(console, browser_name) - - except Exception as e: - if console: - console.print(f"[red] βœ— Error with {browser_name}: {e}[/red]") - continue - - if console: - console.print("[red] βœ— No system browser could be opened[/red]") - return None - - -def _try_playwright_browser(console: Optional[Console] = None) -> Optional[str]: - """Last resort: Use Playwright to launch browser with detailed logging.""" - try: - from playwright.sync_api import sync_playwright - - if console: - console.print("[dim] [Stage 3/3] Using Playwright (last resort)...[/dim]") - console.print("[yellow] ⚠ Note: You'll need to login manually[/yellow]") - else: - print(" [Stage 3/3] Using Playwright (you may need to login manually)...") - - with sync_playwright() as p: - # Try Firefox first (better privacy) - browser_type = "firefox" - try: - if console: - console.print("[dim] Launching Firefox...[/dim]") - browser = p.firefox.launch(headless=False) - if console: - console.print("[green] βœ“ Firefox launched[/green]") - except Exception as firefox_err: - if console: - console.print(f"[dim] Firefox failed: {firefox_err}[/dim]") - console.print("[dim] Trying Chromium...[/dim]") - - try: - browser = p.chromium.launch(headless=False) - browser_type = "chromium" - if console: - console.print("[green] βœ“ Chromium launched[/green]") - except Exception as chromium_err: - if console: - console.print(f"[red] βœ— Both browsers failed[/red]") - return None - - # Create context and page - if console: - console.print("[dim] Creating browser context...[/dim]") - - context = browser.new_context() - page = context.new_page() - - # Navigate - if console: - console.print(f"[cyan] β†’ Navigating to huggingface.co...[/cyan]") - - nav_success = False - expected_url_pattern = "huggingface.co/settings/tokens" - actual_url = None - - try: - page.goto("https://huggingface.co/settings/tokens", timeout=30000) - actual_url = page.url - - # Verify we reached the expected page (or at least HF domain) - if expected_url_pattern in actual_url: - nav_success = True - if console: - console.print(f"[green] βœ“ Page loaded at correct URL[/green]") - elif "huggingface.co/login" in actual_url: - nav_success = True - if console: - console.print(f"[yellow] ⚠ Page loaded but requires login first[/yellow]") - console.print(f"[dim] URL: {actual_url}[/dim]") - elif "huggingface.co" in actual_url: - nav_success = True - if console: - console.print(f"[yellow] ⚠ Page loaded on HF domain but different path[/yellow]") - console.print(f"[dim] URL: {actual_url}[/dim]") - else: - if console: - console.print(f"[red] βœ— Page loaded but unexpected URL[/red]") - console.print(f"[dim] Expected: {expected_url_pattern}[/dim]") - console.print(f"[dim] Actual: {actual_url}[/dim]") - - except Exception as e: - if console: - console.print(f"[red] βœ— Navigation failed: {e}[/red]") - if actual_url: - console.print(f"[dim] Last URL: {actual_url}[/dim]") - nav_success = False - - # Only proceed if page loaded - if nav_success or actual_url: - return _navigate_and_get_token(page, console, browser_type) - else: - if console: - console.print(f"[red] βœ— Cannot proceed - page did not load[/red]") - return None - - except ImportError: - if console: - console.print("[red] βœ— Playwright not installed[/red]") - else: - print(" βœ— Playwright not installed") - return None - except Exception as e: - if console: - console.print(f"[red] βœ— Playwright error: {e}[/red]") - else: - print(f" βœ— Playwright error: {e}") - return None - - -def _navigate_and_get_token(page, console: Optional[Console], browser_type: str) -> Optional[str]: - """Navigate to HuggingFace and get token from user.""" - - if console: - console.print(f"[dim] [Token Step 1/4] Already navigated to HF tokens page[/dim]") - - # Verify page loaded by checking URL - try: - current_url = page.url - if console: - console.print(f"[dim] Current URL: {current_url}[/dim]") - except Exception as e: - if console: - console.print(f"[yellow] ⚠ Could not get URL: {e}[/yellow]") - - # Show instructions - if console: - console.print(f"[cyan] [Token Step 2/4] Showing instructions:[/cyan]") - console.print(" 1. Login to Hugging Face if needed") - console.print(" 2. Click 'New token' button") - console.print(" 3. Set name: 'nlp2cmd'") - console.print(" 4. Select 'Read' role") - console.print(" 5. Click 'Generate token'") - console.print(" 6. Copy the token and paste it here") - else: - print("\nπŸ“‹ Instructions:") - print(" 1. Login to Hugging Face if needed") - print(" 2. Click 'New token' button") - print(" 3. Set name: 'nlp2cmd'") - print(" 4. Select 'Read' role") - print(" 5. Click 'Generate token'") - print(" 6. Copy the token and paste it here") - - # Interactive prompt for token - if console: - console.print(f"[cyan] [Token Step 3/4] Waiting for user input...[/cyan]") - console.print(f"[bold yellow] ⚠️ CHECK YOUR TERMINAL - waiting for token input![/bold yellow]") - console.print(f"[bold] The browser should be open.[/bold]") - console.print(f"[bold] After you create the token in the browser, come back here and paste it below.[/bold]") - - try: - # Print visible separator to catch attention - print("\n" + "="*60) - print("πŸ” ENTER YOUR HF_TOKEN BELOW πŸ”") - print("="*60) - - token = input("πŸ”‘ Paste HF_TOKEN here: ").strip() - - print("="*60) - - if console: - console.print(f"[dim] Input received: {'Yes' if token else 'No'}[/dim]") - - if token: - if console: - console.print(f"[cyan] [Token Step 4/4] Closing browser page...[/cyan]") - - try: - page.close() - if console: - console.print(f"[green] βœ“ Page closed[/green]") - except Exception as e: - if console: - console.print(f"[dim] Note: Could not close page: {e}[/dim]") - - return token - else: - if console: - console.print(f"[yellow] ⚠ No token entered[/yellow]") - except EOFError: - if console: - console.print(f"[red] βœ— EOFError (no input available)[/red]") - except KeyboardInterrupt: - if console: - console.print(f"[yellow] ⚠ User cancelled (KeyboardInterrupt)[/yellow]") - except Exception as e: - if console: - console.print(f"[red] βœ— Error getting input: {e}[/red]") - - # Cleanup on failure - if console: - console.print(f"[dim] Cleaning up...[/dim]") - - try: - page.close() - except Exception: - pass - - return None - - -def _try_existing_browser_dispatch(console: Optional[Console] = None) -> Optional[str]: - """New existing browser token retrieval using modular ExistingBrowserManager. - - This is the refactored version that uses the browser_manager package. - Falls back to legacy _try_existing_browser if modular version unavailable. - """ - if not _BROWSER_MANAGER_AVAILABLE: - return _try_existing_browser(console) - - try: - if console: - console.print("[dim] [Stage 1/3] Using modular browser manager...[/dim]") - - manager = ExistingBrowserManager() - result = manager.connect_and_navigate(verbose=True, console=console) - - if not result.success: - if console and result.error: - console.print(f"[dim] Modular manager failed: {result.error}[/dim]") - return _try_existing_browser(console) - - if result.page: - token = manager.get_token_interactive(result, verbose=True, console=console) - return token - - return None - - except Exception as e: - # Fall back to legacy implementation - if console: - console.print(f"[dim] Modular manager failed: {e}[/dim]") - console.print("[dim] Falling back to legacy implementation...[/dim]") - return _try_existing_browser(console) - - -def _try_playwright_browser_dispatch(console: Optional[Console] = None) -> Optional[str]: - """New browser token retrieval using modular HFTokenRetriever. - - This is the refactored version that uses the browser_token package. - Falls back to legacy _try_playwright_browser if modular version unavailable. - """ - if not _BROWSER_TOKEN_AVAILABLE: - return _try_playwright_browser(console) - - try: - if console: - console.print("[dim] [Stage 3/3] Using modular browser token retriever...[/dim]") - console.print("[yellow] ⚠ Note: You'll need to login manually[/yellow]") - - retriever = HFTokenRetriever() - result = retriever.retrieve() - - if result.success: - if console: - console.print(f"[green] βœ“ Token retrieved via {result.browser_type}[/green]") - return result.token - else: - if console: - if result.error: - console.print(f"[red] βœ— {result.error}[/red]") - else: - console.print(f"[yellow] ⚠ {result.message}[/yellow]") - return None - - except Exception as e: - # Fall back to legacy implementation - if console: - console.print(f"[dim] Modular retriever failed: {e}[/dim]") - console.print("[dim] Falling back to legacy implementation...[/dim]") - return _try_playwright_browser(console) - - -def _manual_browser_instructions(console: Optional[Console], browser_name: str) -> Optional[str]: - """Show manual instructions when browser automation fails.""" - if console: - console.print(f"\n[cyan]πŸ“‹ {browser_name} opened. Please:[/cyan]") - console.print(" 1. Go to: https://huggingface.co/settings/tokens") - console.print(" 2. Login if not logged in") - console.print(" 3. Create new token (name: nlp2cmd, role: read)") - console.print(" 4. Copy the token") - else: - print(f"\nπŸ“‹ {browser_name} opened. Please:") - print(" 1. Go to: https://huggingface.co/settings/tokens") - print(" 2. Login if not logged in") - print(" 3. Create new token (name: nlp2cmd, role: read)") - print(" 4. Copy the token") - - try: - token = input("\nπŸ”‘ Paste HF_TOKEN here: ").strip() - if token: - return token - except (EOFError, KeyboardInterrupt): - pass - - return None - - if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="NLP2CMD System Doctor") diff --git a/src/nlp2cmd/cli/commands/doctor_browser.py b/src/nlp2cmd/cli/commands/doctor_browser.py new file mode 100644 index 00000000..c21bb67f --- /dev/null +++ b/src/nlp2cmd/cli/commands/doctor_browser.py @@ -0,0 +1,704 @@ +"""Browser automation helpers for HF token retrieval.""" +from __future__ import annotations + +import os +import socket +import subprocess +import shutil +import time +from pathlib import Path +from typing import Optional + +try: + from rich.console import Console + HAS_RICH = True +except ImportError: + HAS_RICH = False + Console = None + +try: + from nlp2cmd.browser_token import HFTokenRetriever, TokenConfig + _BROWSER_TOKEN_AVAILABLE = True +except ImportError: + _BROWSER_TOKEN_AVAILABLE = False + +try: + from nlp2cmd.browser_manager import ExistingBrowserManager, BrowserConfig + _BROWSER_MANAGER_AVAILABLE = True +except ImportError: + _BROWSER_MANAGER_AVAILABLE = False + + +def get_hf_token_via_browser(console: Optional[Console] = None) -> Optional[str]: + """Open browser to help user get HF_TOKEN from Hugging Face. + + Browser priority: + 1. Connect to existing browser (Firefox/Chrome via CDP) + 2. Open new system browser (firefox/chrome commands) + 3. Use Playwright (requires manual login) + + Returns the token if successfully retrieved. + """ + if console: + console.print("[cyan]🌐 Opening Hugging Face in browser...[/cyan]") + else: + print("🌐 Opening Hugging Face in browser...") + + # Try priority 1: Connect to existing browser + token = _try_existing_browser(console) + if token: + return token + + # Try priority 2: Open new system browser + token = _try_system_browser(console) + if token: + return token + + # Try priority 3: Use Playwright + token = _try_playwright_browser(console) + if token: + return token + + return None + + +def _try_existing_browser(console: Optional[Console] = None) -> Optional[str]: + """Try to connect to existing browser via CDP with detailed logging.""" + import socket + + if console: + console.print("[dim] [Stage 1/3] Checking for existing browser...[/dim]") + + # Check common CDP ports + cdp_ports = [9222, 9223, 9224, 9333] + found_port = None + + for port in cdp_ports: + if console: + console.print(f"[dim] Checking port {port}...[/dim]") + + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(1) + result = sock.connect_ex(("localhost", port)) + sock.close() + + if result == 0: + found_port = port + if console: + console.print(f"[green] βœ“ Found browser on port {port}[/green]") + break + else: + if console: + console.print(f"[dim] Port {port}: not available[/dim]") + except Exception as e: + if console: + console.print(f"[dim] Port {port}: error - {e}[/dim]") + + if not found_port: + if console: + console.print("[dim] β„Ή No existing browser with CDP found[/dim]") + console.print("[dim] Tip: Run 'firefox --remote-debugging-port=9222' first[/dim]") + return None + + # Try to connect via Playwright CDP + if console: + console.print(f"[cyan] β†’ Connecting via Playwright to port {found_port}...[/cyan]") + + connection_success = False + browser = None + browser_type = None + + try: + from playwright.sync_api import sync_playwright + + with sync_playwright() as p: + # Try Chrome/Chromium first + try: + browser = p.chromium.connect_over_cdp(f"http://localhost:{found_port}") + browser_type = "Chrome/Chromium" + connection_success = True + if console: + console.print(f"[green] βœ“ Connected to {browser_type} via CDP[/green]") + except Exception as chrome_err: + if console: + console.print(f"[dim] Chromium CDP failed: {str(chrome_err)[:50]}...[/dim]") + + # Try Firefox + try: + browser = p.firefox.connect_over_cdp(f"http://localhost:{found_port}") + browser_type = "Firefox" + connection_success = True + if console: + console.print(f"[green] βœ“ Connected to Firefox via CDP[/green]") + except Exception as firefox_err: + if console: + console.print(f"[red] βœ— CDP connection failed for both browsers[/red]") + console.print(f"[dim] Chrome error: {str(chrome_err)[:30]}...[/dim]") + console.print(f"[dim] Firefox error: {str(firefox_err)[:30]}...[/dim]") + return None + + if not connection_success or not browser: + if console: + console.print(f"[red] βœ— Browser connection established but browser object is None[/red]") + return None + + # Create new context and page + if console: + console.print(f"[dim] Creating browser context...[/dim]") + + try: + context = browser.new_context() + page = context.new_page() + if console: + console.print(f"[green] βœ“ Browser context created[/green]") + except Exception as ctx_err: + if console: + console.print(f"[red] βœ— Failed to create browser context: {ctx_err}[/red]") + return None + + # Navigate + if console: + console.print(f"[cyan] β†’ Navigating to huggingface.co...[/cyan]") + + nav_success = False + expected_url_pattern = "huggingface.co/settings/tokens" + actual_url = None + + try: + page.goto("https://huggingface.co/settings/tokens", timeout=30000) + actual_url = page.url + + # Verify we reached the expected page (or at least HF domain) + if expected_url_pattern in actual_url: + nav_success = True + if console: + console.print(f"[green] βœ“ Page loaded at correct URL[/green]") + elif "huggingface.co/login" in actual_url: + # This is expected if not logged in, but we should warn + nav_success = True # Page loaded, just needs login + if console: + console.print(f"[yellow] ⚠ Page loaded but requires login first[/yellow]") + console.print(f"[dim] URL: {actual_url}[/dim]") + elif "huggingface.co" in actual_url: + nav_success = True + if console: + console.print(f"[yellow] ⚠ Page loaded on HF domain but different path[/yellow]") + console.print(f"[dim] URL: {actual_url}[/dim]") + else: + if console: + console.print(f"[red] βœ— Page loaded but unexpected URL[/red]") + console.print(f"[dim] Expected: {expected_url_pattern}[/dim]") + console.print(f"[dim] Actual: {actual_url}[/dim]") + + except Exception as e: + if console: + console.print(f"[red] βœ— Navigation failed: {e}[/red]") + if actual_url: + console.print(f"[dim] Last URL: {actual_url}[/dim]") + nav_success = False + + except ImportError: + if console: + console.print(f"[red] βœ— Playwright not installed[/red]") + return None + except Exception as e: + if console: + console.print(f"[red] βœ— CDP connection error: {e}[/red]") + return None + + +def _try_system_browser(console: Optional[Console] = None) -> Optional[str]: + """Try to open new system browser with detailed stage logging.""" + import subprocess + import time + import socket + + if console: + console.print("[dim] [Stage 2/3] Opening system browser...[/dim]") + + # Try to open Firefox or Chrome/Chromium + browsers_to_try = [ + ("firefox", ["firefox", "--new-window", "--remote-debugging-port=9222"]), + ("google-chrome", ["google-chrome", "--new-window", "--remote-debugging-port=9222"]), + ("chromium", ["chromium", "--new-window", "--remote-debugging-port=9222"]), + ("chromium-browser", ["chromium-browser", "--new-window", "--remote-debugging-port=9222"]), + ] + + for browser_name, cmd in browsers_to_try: + try: + # Stage 2.1: Check if browser binary exists + if console: + console.print(f"[dim] Checking {browser_name}...[/dim]") + + result = subprocess.run(["which", browser_name], capture_output=True, timeout=5) + if result.returncode != 0: + if console: + console.print(f"[dim] βœ— {browser_name} not found[/dim]") + continue + + if console: + console.print(f"[cyan] β†’ Launching {browser_name}...[/cyan]") + + # Stage 2.2: Launch browser with CDP enabled + try: + process = subprocess.Popen( + cmd + ["about:blank"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True + ) + if console: + console.print(f"[dim] PID: {process.pid}[/dim]") + except Exception as e: + if console: + console.print(f"[red] βœ— Failed to launch: {e}[/red]") + continue + + # Stage 2.3: Wait for browser to initialize + if console: + console.print("[dim] Waiting for browser to start (3s)...[/dim]") + time.sleep(3) + + # Stage 2.4: Try to connect via CDP (with actual protocol verification) + if console: + console.print("[dim] Checking CDP port 9222 (with protocol verification)...[/dim]") + + cdp_available = False + for attempt in range(5): + try: + # First: basic TCP check + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(2) + result = sock.connect_ex(("localhost", 9222)) + sock.close() + + if result == 0: + # Second: verify it's actually a CDP endpoint by making HTTP request + import urllib.request + try: + response = urllib.request.urlopen( + "http://localhost:9222/json/version", + timeout=3 + ) + cdp_info = response.read().decode('utf-8') + if 'Browser' in cdp_info or 'Protocol-Version' in cdp_info: + cdp_available = True + if console: + console.print(f"[green] βœ“ CDP port 9222 ready (verified protocol)[/green]") + break + else: + if console: + console.print(f"[dim] Attempt {attempt+1}/5: port open but not CDP protocol[/dim]") + except Exception as http_err: + if console: + console.print(f"[dim] Attempt {attempt+1}/5: port open but CDP check failed: {str(http_err)[:40]}[/dim]") + else: + if console: + console.print(f"[dim] Attempt {attempt+1}/5: port not ready (code: {result})[/dim]") + except Exception as e: + if console: + console.print(f"[dim] Attempt {attempt+1}/5: {str(e)[:40]}[/dim]") + time.sleep(1) + + if not cdp_available: + if console: + console.print(f"[yellow] βœ— CDP not available after 5 attempts[/yellow]") + console.print(f"[dim] Browser launched but CDP protocol not responding[/dim]") + continue + + # Stage 2.5: Connect with Playwright + if console: + console.print(f"[cyan] β†’ Connecting via Playwright CDP...[/cyan]") + + connection_success = False + browser = None + + try: + from playwright.sync_api import sync_playwright + + with sync_playwright() as p: + try: + browser = p.chromium.connect_over_cdp("http://localhost:9222") + connection_success = True + if console: + console.print(f"[green] βœ“ Connected to {browser_name} via CDP[/green]") + except Exception as e: + if console: + console.print(f"[yellow] ⚠ CDP connect failed: {str(e)[:50]}...[/yellow]") + console.print(f"[dim] Falling back to manual mode...[/dim]") + return _manual_browser_instructions(console, browser_name) + + if not connection_success or not browser: + if console: + console.print(f"[red] βœ— Connection reported success but browser is None[/red]") + return _manual_browser_instructions(console, browser_name) + + try: + context = browser.new_context() + page = context.new_page() + if console: + console.print(f"[green] βœ“ Browser context created[/green]") + except Exception as ctx_err: + if console: + console.print(f"[red] βœ— Failed to create context: {ctx_err}[/red]") + return _manual_browser_instructions(console, browser_name) + + # Stage 2.6: Navigate to HF + if console: + console.print(f"[cyan] β†’ Navigating to huggingface.co...[/cyan]") + + nav_success = False + expected_url_pattern = "huggingface.co/settings/tokens" + actual_url = None + + try: + page.goto("https://huggingface.co/settings/tokens", timeout=30000) + actual_url = page.url + + # Verify we reached the expected page (or at least HF domain) + if expected_url_pattern in actual_url: + nav_success = True + if console: + console.print(f"[green] βœ“ Page loaded at correct URL[/green]") + elif "huggingface.co/login" in actual_url: + nav_success = True # Page loaded, just needs login + if console: + console.print(f"[yellow] ⚠ Page loaded but requires login first[/yellow]") + console.print(f"[dim] URL: {actual_url}[/dim]") + elif "huggingface.co" in actual_url: + nav_success = True + if console: + console.print(f"[yellow] ⚠ Page loaded on HF domain but different path[/yellow]") + console.print(f"[dim] URL: {actual_url}[/dim]") + else: + if console: + console.print(f"[red] βœ— Page loaded but unexpected URL[/red]") + console.print(f"[dim] Expected: {expected_url_pattern}[/dim]") + console.print(f"[dim] Actual: {actual_url}[/dim]") + + except Exception as e: + if console: + console.print(f"[red] βœ— Navigation failed: {e}[/red]") + if actual_url: + console.print(f"[dim] Last URL: {actual_url}[/dim]") + nav_success = False + + # Stage 2.7: Get token from user (even if nav had issues, let user try) + if nav_success or actual_url: # Only proceed if we at least loaded something + return _navigate_and_get_token(page, console, browser_name) + else: + if console: + console.print(f"[red] βœ— Cannot proceed - page did not load[/red]") + return None + + except ImportError: + if console: + console.print(f"[red] βœ— Playwright not installed[/red]") + return _manual_browser_instructions(console, browser_name) + except Exception as e: + if console: + console.print(f"[red] βœ— Playwright error: {e}[/red]") + return _manual_browser_instructions(console, browser_name) + + except Exception as e: + if console: + console.print(f"[red] βœ— Error with {browser_name}: {e}[/red]") + continue + + if console: + console.print("[red] βœ— No system browser could be opened[/red]") + return None + + +def _try_playwright_browser(console: Optional[Console] = None) -> Optional[str]: + """Last resort: Use Playwright to launch browser with detailed logging.""" + try: + from playwright.sync_api import sync_playwright + + if console: + console.print("[dim] [Stage 3/3] Using Playwright (last resort)...[/dim]") + console.print("[yellow] ⚠ Note: You'll need to login manually[/yellow]") + else: + print(" [Stage 3/3] Using Playwright (you may need to login manually)...") + + with sync_playwright() as p: + # Try Firefox first (better privacy) + browser_type = "firefox" + try: + if console: + console.print("[dim] Launching Firefox...[/dim]") + browser = p.firefox.launch(headless=False) + if console: + console.print("[green] βœ“ Firefox launched[/green]") + except Exception as firefox_err: + if console: + console.print(f"[dim] Firefox failed: {firefox_err}[/dim]") + console.print("[dim] Trying Chromium...[/dim]") + + try: + browser = p.chromium.launch(headless=False) + browser_type = "chromium" + if console: + console.print("[green] βœ“ Chromium launched[/green]") + except Exception as chromium_err: + if console: + console.print(f"[red] βœ— Both browsers failed[/red]") + return None + + # Create context and page + if console: + console.print("[dim] Creating browser context...[/dim]") + + context = browser.new_context() + page = context.new_page() + + # Navigate + if console: + console.print(f"[cyan] β†’ Navigating to huggingface.co...[/cyan]") + + nav_success = False + expected_url_pattern = "huggingface.co/settings/tokens" + actual_url = None + + try: + page.goto("https://huggingface.co/settings/tokens", timeout=30000) + actual_url = page.url + + # Verify we reached the expected page (or at least HF domain) + if expected_url_pattern in actual_url: + nav_success = True + if console: + console.print(f"[green] βœ“ Page loaded at correct URL[/green]") + elif "huggingface.co/login" in actual_url: + nav_success = True + if console: + console.print(f"[yellow] ⚠ Page loaded but requires login first[/yellow]") + console.print(f"[dim] URL: {actual_url}[/dim]") + elif "huggingface.co" in actual_url: + nav_success = True + if console: + console.print(f"[yellow] ⚠ Page loaded on HF domain but different path[/yellow]") + console.print(f"[dim] URL: {actual_url}[/dim]") + else: + if console: + console.print(f"[red] βœ— Page loaded but unexpected URL[/red]") + console.print(f"[dim] Expected: {expected_url_pattern}[/dim]") + console.print(f"[dim] Actual: {actual_url}[/dim]") + + except Exception as e: + if console: + console.print(f"[red] βœ— Navigation failed: {e}[/red]") + if actual_url: + console.print(f"[dim] Last URL: {actual_url}[/dim]") + nav_success = False + + # Only proceed if page loaded + if nav_success or actual_url: + return _navigate_and_get_token(page, console, browser_type) + else: + if console: + console.print(f"[red] βœ— Cannot proceed - page did not load[/red]") + return None + + except ImportError: + if console: + console.print("[red] βœ— Playwright not installed[/red]") + else: + print(" βœ— Playwright not installed") + return None + except Exception as e: + if console: + console.print(f"[red] βœ— Playwright error: {e}[/red]") + else: + print(f" βœ— Playwright error: {e}") + return None + + +def _navigate_and_get_token(page, console: Optional[Console], browser_type: str) -> Optional[str]: + """Navigate to HuggingFace and get token from user.""" + + if console: + console.print(f"[dim] [Token Step 1/4] Already navigated to HF tokens page[/dim]") + + # Verify page loaded by checking URL + try: + current_url = page.url + if console: + console.print(f"[dim] Current URL: {current_url}[/dim]") + except Exception as e: + if console: + console.print(f"[yellow] ⚠ Could not get URL: {e}[/yellow]") + + # Show instructions + if console: + console.print(f"[cyan] [Token Step 2/4] Showing instructions:[/cyan]") + console.print(" 1. Login to Hugging Face if needed") + console.print(" 2. Click 'New token' button") + console.print(" 3. Set name: 'nlp2cmd'") + console.print(" 4. Select 'Read' role") + console.print(" 5. Click 'Generate token'") + console.print(" 6. Copy the token and paste it here") + else: + print("\nπŸ“‹ Instructions:") + print(" 1. Login to Hugging Face if needed") + print(" 2. Click 'New token' button") + print(" 3. Set name: 'nlp2cmd'") + print(" 4. Select 'Read' role") + print(" 5. Click 'Generate token'") + print(" 6. Copy the token and paste it here") + + # Interactive prompt for token + if console: + console.print(f"[cyan] [Token Step 3/4] Waiting for user input...[/cyan]") + console.print(f"[bold yellow] ⚠️ CHECK YOUR TERMINAL - waiting for token input![/bold yellow]") + console.print(f"[bold] The browser should be open.[/bold]") + console.print(f"[bold] After you create the token in the browser, come back here and paste it below.[/bold]") + + try: + # Print visible separator to catch attention + print("\n" + "="*60) + print("πŸ” ENTER YOUR HF_TOKEN BELOW πŸ”") + print("="*60) + + token = input("πŸ”‘ Paste HF_TOKEN here: ").strip() + + print("="*60) + + if console: + console.print(f"[dim] Input received: {'Yes' if token else 'No'}[/dim]") + + if token: + if console: + console.print(f"[cyan] [Token Step 4/4] Closing browser page...[/cyan]") + + try: + page.close() + if console: + console.print(f"[green] βœ“ Page closed[/green]") + except Exception as e: + if console: + console.print(f"[dim] Note: Could not close page: {e}[/dim]") + + return token + else: + if console: + console.print(f"[yellow] ⚠ No token entered[/yellow]") + except EOFError: + if console: + console.print(f"[red] βœ— EOFError (no input available)[/red]") + except KeyboardInterrupt: + if console: + console.print(f"[yellow] ⚠ User cancelled (KeyboardInterrupt)[/yellow]") + except Exception as e: + if console: + console.print(f"[red] βœ— Error getting input: {e}[/red]") + + # Cleanup on failure + if console: + console.print(f"[dim] Cleaning up...[/dim]") + + try: + page.close() + except Exception: + pass + + return None + + +def _try_existing_browser_dispatch(console: Optional[Console] = None) -> Optional[str]: + """New existing browser token retrieval using modular ExistingBrowserManager. + + This is the refactored version that uses the browser_manager package. + Falls back to legacy _try_existing_browser if modular version unavailable. + """ + if not _BROWSER_MANAGER_AVAILABLE: + return _try_existing_browser(console) + + try: + if console: + console.print("[dim] [Stage 1/3] Using modular browser manager...[/dim]") + + manager = ExistingBrowserManager() + result = manager.connect_and_navigate(verbose=True, console=console) + + if not result.success: + if console and result.error: + console.print(f"[dim] Modular manager failed: {result.error}[/dim]") + return _try_existing_browser(console) + + if result.page: + token = manager.get_token_interactive(result, verbose=True, console=console) + return token + + return None + + except Exception as e: + # Fall back to legacy implementation + if console: + console.print(f"[dim] Modular manager failed: {e}[/dim]") + console.print("[dim] Falling back to legacy implementation...[/dim]") + return _try_existing_browser(console) + + +def _try_playwright_browser_dispatch(console: Optional[Console] = None) -> Optional[str]: + """New browser token retrieval using modular HFTokenRetriever. + + This is the refactored version that uses the browser_token package. + Falls back to legacy _try_playwright_browser if modular version unavailable. + """ + if not _BROWSER_TOKEN_AVAILABLE: + return _try_playwright_browser(console) + + try: + if console: + console.print("[dim] [Stage 3/3] Using modular browser token retriever...[/dim]") + console.print("[yellow] ⚠ Note: You'll need to login manually[/yellow]") + + retriever = HFTokenRetriever() + result = retriever.retrieve() + + if result.success: + if console: + console.print(f"[green] βœ“ Token retrieved via {result.browser_type}[/green]") + return result.token + else: + if console: + if result.error: + console.print(f"[red] βœ— {result.error}[/red]") + else: + console.print(f"[yellow] ⚠ {result.message}[/yellow]") + return None + + except Exception as e: + # Fall back to legacy implementation + if console: + console.print(f"[dim] Modular retriever failed: {e}[/dim]") + console.print("[dim] Falling back to legacy implementation...[/dim]") + return _try_playwright_browser(console) + + +def _manual_browser_instructions(console: Optional[Console], browser_name: str) -> Optional[str]: + """Show manual instructions when browser automation fails.""" + if console: + console.print(f"\n[cyan]πŸ“‹ {browser_name} opened. Please:[/cyan]") + console.print(" 1. Go to: https://huggingface.co/settings/tokens") + console.print(" 2. Login if not logged in") + console.print(" 3. Create new token (name: nlp2cmd, role: read)") + console.print(" 4. Copy the token") + else: + print(f"\nπŸ“‹ {browser_name} opened. Please:") + print(" 1. Go to: https://huggingface.co/settings/tokens") + print(" 2. Login if not logged in") + print(" 3. Create new token (name: nlp2cmd, role: read)") + print(" 4. Copy the token") + + try: + token = input("\nπŸ”‘ Paste HF_TOKEN here: ").strip() + if token: + return token + except (EOFError, KeyboardInterrupt): + pass + + return None + diff --git a/src/nlp2cmd/cli/commands/doctor_types.py b/src/nlp2cmd/cli/commands/doctor_types.py new file mode 100644 index 00000000..76d175d9 --- /dev/null +++ b/src/nlp2cmd/cli/commands/doctor_types.py @@ -0,0 +1,25 @@ +"""Shared types for nlp2cmd doctor checks.""" +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from typing import Optional + + +class Status(Enum): + OK = "ok" + WARNING = "warning" + ERROR = "error" + INFO = "info" + FIXED = "fixed" + + +@dataclass +class CheckResult: + name: str + status: Status + message: str + details: dict = field(default_factory=dict) + fix_applied: bool = False + fix_command: Optional[str] = None + diff --git a/src/nlp2cmd/web_schema/site_explorer.py b/src/nlp2cmd/web_schema/site_explorer.py index 2a7cc092..37351f2b 100644 --- a/src/nlp2cmd/web_schema/site_explorer.py +++ b/src/nlp2cmd/web_schema/site_explorer.py @@ -15,164 +15,54 @@ import re import sys import time -from dataclasses import dataclass, field from typing import Any, Optional from urllib.parse import urljoin, urlparse from urllib.request import urlopen, Request -from urllib.error import URLError -import xml.etree.ElementTree as ET -# Import modular page analysis -try: - from nlp2cmd.page_analysis import PageAnalyzer, PageAnalysisResult - _PAGE_ANALYSIS_AVAILABLE = True -except ImportError: - _PAGE_ANALYSIS_AVAILABLE = False - -_DEBUG = os.environ.get("NLP2CMD_DEBUG", "").lower() in ("1", "true", "yes") - - -def _debug(msg: str) -> None: - """Print debug message to stderr when NLP2CMD_DEBUG=1.""" - if _DEBUG: - print(f"DEBUG [SiteExplorer] {msg}", file=sys.stderr, flush=True) - - -# ── Module-level helpers for platform URL resolution ────────────────── -def _github_readme_url(url: str) -> str: - """Convert github.com/owner/repo to raw README URL.""" - parsed = urlparse(url) - parts = [p for p in parsed.path.strip("/").split("/") if p] - if len(parts) >= 2: - return f"https://github.com/{parts[0]}/{parts[1]}" - return url - - -def _github_docs_url(url: str) -> str: - """Try to resolve GitHub repo docs (wiki, /docs, or README).""" - parsed = urlparse(url) - parts = [p for p in parsed.path.strip("/").split("/") if p] - if len(parts) >= 2: - return f"https://github.com/{parts[0]}/{parts[1]}/tree/main/docs" - return url - - -def _pypi_to_docs_url(url: str) -> Optional[str]: - """Convert pypi.org/project/X to readthedocs or homepage.""" - parsed = urlparse(url) - parts = [p for p in parsed.path.strip("/").split("/") if p] - if len(parts) >= 2 and parts[0] == "project": - pkg = parts[1].lower().replace("-", "").replace("_", "") - return f"https://{pkg}.readthedocs.io/en/latest/" - return None - - -@dataclass -class PageInfo: - """Information about a discovered page.""" - url: str - title: str = "" - links: list[str] = field(default_factory=list) - has_form: bool = False - form_count: int = 0 - contact_field_count: int = 0 - junk_field_count: int = 0 - score: float = 0.0 # Relevance score for form/contact pages - load_time_ms: float = 0.0 # Page load + analysis time - - -@dataclass -class ExplorationResult: - """Result of site exploration.""" - success: bool - form_url: Optional[str] = None - form_page: Optional[PageInfo] = None - explored_pages: list[PageInfo] = field(default_factory=list) - error: Optional[str] = None - - -class SiteExplorer: +from nlp2cmd.web_schema.site_explorer_page import SiteExplorerPageMixin +from nlp2cmd.web_schema.site_explorer_types import ( + ARTICLE_KEYWORDS, + BLOCKED_RESOURCE_PATTERNS, + CONTACT_KEYWORDS, + DOCS_FRAMEWORKS, + DOCS_KEYWORDS, + FORM_FIELD_KEYWORDS, + PLATFORM_DOCS_URLS, + PRODUCT_KEYWORDS, + ExplorationResult, + PageInfo, + debug, +) + +# Backward-compatible re-exports +__all__ = [ + "SiteExplorer", + "quick_find_form", + "quick_find_content", + "ExplorationResult", + "PageInfo", +] + + +class SiteExplorer(SiteExplorerPageMixin): """ Explores website to find forms, contact pages, and other content. - + Usage: explorer = SiteExplorer() result = explorer.find_form(url="https://example.com", intent="contact") if result.success: print(f"Found form at: {result.form_url}") """ - - # Keywords that suggest a page might contain a contact form - CONTACT_KEYWORDS = [ - "kontakt", "contact", "napisz do nas", "write to us", - "formularz", "form", "wiadomoΕ›Δ‡", "message", - "pomoc", "help", "support", "serwis", - "zapytaj", "ask", "biuro", "office", "dane", "info", - "obsΕ‚uga", "obsuga", "klienta", "customer", - ] - - # Keywords that suggest articles/content - ARTICLE_KEYWORDS = [ - "artykuΕ‚", "article", "blog", "news", "wiadomoΕ›ci", "aktualnoΕ›ci", - "publikacja", "publication", "post", "wpis", "treΕ›Δ‡", "content", - "poradnik", "guide", "tutorial", "instrukcja", "manual", - ] - - # Keywords that suggest products/services - PRODUCT_KEYWORDS = [ - "produkt", "product", "usΕ‚uga", "service", "oferta", "offer", - "sklep", "shop", "store", "cennik", "price", "cena", "buy", - "katalog", "catalog", "portfolio", "galeria", "gallery", - ] - - # Keywords that suggest documentation/help - DOCS_KEYWORDS = [ - "dokumentacja", "documentation", "docs", "pomoc", "help", - "faq", "pytania", "questions", "support", "wsparcie", - "manual", "instrukcja", "guide", "tutorial", "readme", - "wiki", "api", "reference", "examples", "przykΕ‚ady", - "github", "gitlab", "bitbucket", "repository", "repo" - ] - - # Keywords that suggest form fields - FORM_FIELD_KEYWORDS = [ - "email", "e-mail", "telefon", "phone", "imiΔ™", "name", - "nazwisko", "surname", "wiadomoΕ›Δ‡", "message", "temat", "subject", - ] - - # Resource types to block for faster loading - BLOCKED_RESOURCE_PATTERNS = ( - "**/*.png", "**/*.jpg", "**/*.jpeg", "**/*.gif", "**/*.svg", - "**/*.webp", "**/*.ico", "**/*.bmp", "**/*.tiff", - "**/*.woff", "**/*.woff2", "**/*.ttf", "**/*.eot", - "**/*.mp4", "**/*.webm", "**/*.ogg", "**/*.mp3", - ) - # Smart URL shortcuts for known platforms (Strategy 3) - PLATFORM_DOCS_URLS: dict[str, Any] = { - "github.com": { - "readme": lambda url: _github_readme_url(url), - "docs": lambda url: _github_docs_url(url), - }, - "readthedocs.io": { - "docs": lambda url: url if "/en/" in url else url.rstrip("/") + "/en/latest/", - }, - "docs.python.org": { - "docs": lambda _url: "https://docs.python.org/3/", - }, - "pypi.org": { - "docs": lambda url: _pypi_to_docs_url(url), - }, - } - - # Known documentation frameworks with predictable URL structures (Strategy 8) - DOCS_FRAMEWORKS: dict[str, list[str]] = { - "readthedocs": ["/en/latest/", "/en/stable/", "readthedocs.io"], - "mkdocs": ["/mkdocs.yml", "mkdocs-material", "/site/"], - "gitbook": ["gitbook.io", ".gitbook.io"], - "sphinx": ["/_static/sphinx", "searchindex.js", "genindex.html"], - "docusaurus": ["/docs/", "/blog/", "docusaurus"], - } + CONTACT_KEYWORDS = CONTACT_KEYWORDS + ARTICLE_KEYWORDS = ARTICLE_KEYWORDS + PRODUCT_KEYWORDS = PRODUCT_KEYWORDS + DOCS_KEYWORDS = DOCS_KEYWORDS + FORM_FIELD_KEYWORDS = FORM_FIELD_KEYWORDS + BLOCKED_RESOURCE_PATTERNS = BLOCKED_RESOURCE_PATTERNS + PLATFORM_DOCS_URLS = PLATFORM_DOCS_URLS + DOCS_FRAMEWORKS = DOCS_FRAMEWORKS def __init__( self, @@ -205,12 +95,12 @@ def _abort_heavy(route: Any) -> None: except Exception: pass - for pattern in SiteExplorer.BLOCKED_RESOURCE_PATTERNS: + for pattern in BLOCKED_RESOURCE_PATTERNS: try: context.route(pattern, _abort_heavy) except Exception: pass - _debug("Resource blocking enabled") + debug("Resource blocking enabled") # ── Strategy 3: Smart URL Patterns ───────────────────────────────── def _resolve_platform_url(self, url: str, content_type: str) -> Optional[str]: @@ -218,14 +108,14 @@ def _resolve_platform_url(self, url: str, content_type: str) -> Optional[str]: parsed = urlparse(url) netloc = parsed.netloc.lower() - for platform, handlers in self.PLATFORM_DOCS_URLS.items(): + for platform, handlers in PLATFORM_DOCS_URLS.items(): if platform in netloc: handler = handlers.get(content_type) or handlers.get("docs") if handler: try: resolved = handler(url) if resolved: - _debug(f"Platform shortcut: {platform} -> {resolved}") + debug(f"Platform shortcut: {platform} -> {resolved}") return resolved except Exception: pass @@ -250,7 +140,7 @@ def _goto_with_retry(self, page: Any, url: str) -> None: if not is_retriable: raise wait_ms = min(1000 * (2 ** attempt), 8000) - _debug(f"Retry {attempt + 1}/{self.max_retries} for {url} after {wait_ms}ms: {e}") + debug(f"Retry {attempt + 1}/{self.max_retries} for {url} after {wait_ms}ms: {e}") page.wait_for_timeout(wait_ms) if last_exc: raise last_exc @@ -275,10 +165,10 @@ def _try_github_api(url: str) -> Optional[str]: with urlopen(req, timeout=5) as resp: content = resp.read().decode("utf-8", errors="replace") if len(content) > 50: - _debug(f"GitHub API: fetched README ({len(content)} chars) for {owner}/{repo}") + debug(f"GitHub API: fetched README ({len(content)} chars) for {owner}/{repo}") return content except Exception as e: - _debug(f"GitHub API failed for {owner}/{repo}: {e}") + debug(f"GitHub API failed for {owner}/{repo}: {e}") return None # ── Strategy 8: Documentation Framework Detection ────────────────── @@ -288,9 +178,9 @@ def _detect_docs_framework(self, url: str, page_html: str = "") -> Optional[str] html_lower = page_html.lower() if page_html else "" combined = url_lower + " " + html_lower - for framework, indicators in self.DOCS_FRAMEWORKS.items(): + for framework, indicators in DOCS_FRAMEWORKS.items(): if any(ind in combined for ind in indicators): - _debug(f"Detected docs framework: {framework} at {url}") + debug(f"Detected docs framework: {framework} at {url}") return framework return None @@ -300,51 +190,13 @@ def _record_timing(self, url: str, phase: str, duration_ms: float) -> None: entry = {"url": url, "phase": phase, "duration_ms": round(duration_ms, 1)} self._timing_stats.append(entry) if duration_ms > 5000: - _debug(f"⚠ SLOW {phase}: {url} took {duration_ms:.0f}ms") + debug(f"⚠ SLOW {phase}: {url} took {duration_ms:.0f}ms") elif _DEBUG: - _debug(f"Timing {phase}: {url} = {duration_ms:.0f}ms") + debug(f"Timing {phase}: {url} = {duration_ms:.0f}ms") def get_timing_stats(self) -> list[dict[str, Any]]: """Return collected timing stats.""" return list(self._timing_stats) - - # ── Strategy 10: Graceful Degradation ────────────────────────────── - @staticmethod - def _fallback_static_scrape(url: str, timeout: int = 5) -> Optional[PageInfo]: - """Fallback: fetch page with urllib (no JS) when Playwright fails.""" - try: - req = Request(url, headers={"User-Agent": "nlp2cmd/1.0 (static fallback)"}) - with urlopen(req, timeout=timeout) as resp: - html = resp.read().decode("utf-8", errors="replace") - - info = PageInfo(url=url) - - # Extract title - m = re.search(r"]*>(.*?)", html, re.IGNORECASE | re.DOTALL) - if m: - info.title = m.group(1).strip() - - # Count form fields - inputs = len(re.findall(r' 0 - - # Extract links - for m in re.finditer(r'href=["\']([^"\']+)["\']', html): - href = m.group(1) - if href.startswith(("http://", "https://")): - info.links.append(href) - elif href.startswith("/"): - info.links.append(urljoin(url, href)) - info.links = info.links[:20] - - _debug(f"Static fallback OK: {url} title='{info.title[:40]}' forms={info.form_count} links={len(info.links)}") - return info - except Exception as e: - _debug(f"Static fallback failed for {url}: {e}") - return None - def find_content( self, url: str, @@ -410,7 +262,7 @@ def find_content( # Fast-path: try common contact URLs first (cheap and often works even # when menu extraction fails or homepage blocks link discovery). - if intent == "contact": + if content_type in ("contact", "form"): try: parsed_base = urlparse(url) base = f"{parsed_base.scheme}://{parsed_base.netloc}" if parsed_base.scheme and parsed_base.netloc else url @@ -430,7 +282,7 @@ def find_content( page=page, url=cand, depth=0, - intent=intent, + intent=content_type, explored_pages=explored_pages, base_domain=urlparse(url).netloc, ) @@ -455,11 +307,11 @@ def find_content( search_term=search_term, ) - _debug(f"Main page result: {main_page_result is not None}") + debug(f"Main page result: {main_page_result is not None}") if main_page_result: - _debug(f"Main page URL: {main_page_result.url}") - _debug(f"Main page has form: {main_page_result.has_form}, contact_fields: {main_page_result.contact_field_count}") - _debug(f"Main page links: {len(main_page_result.links)}") + debug(f"Main page URL: {main_page_result.url}") + debug(f"Main page has form: {main_page_result.has_form}, contact_fields: {main_page_result.contact_field_count}") + debug(f"Main page links: {len(main_page_result.links)}") # If main page has the target content AND it's not contact intent, return it if main_page_result and self._has_content_type(main_page_result, content_type) and content_type != "contact": @@ -472,7 +324,7 @@ def find_content( # For contact intent, always explore contact links first even if main page has form if main_page_result and main_page_result.links and len(explored_pages) < self.max_pages: - _debug("Exploring links from main page") + debug("Exploring links from main page") # Sort links by contact relevance contact_links = [] other_links = [] @@ -487,13 +339,13 @@ def find_content( else: other_links.append(link) - _debug(f"Found {len(contact_links)} contact links: {contact_links}") + debug(f"Found {len(contact_links)} contact links: {contact_links}") # Explore contact links first for link in contact_links[:5]: # Check up to 5 contact links if len(self._explored_urls) >= self.max_pages: break - _debug(f"Exploring contact link: {link}") + debug(f"Exploring contact link: {link}") result = self._explore_recursive( page=page, url=link, @@ -504,7 +356,7 @@ def find_content( search_term=search_term, ) if result and self._has_content_type(result, content_type): - _debug(f"Found contact form at: {link}") + debug(f"Found contact form at: {link}") return ExplorationResult( success=True, form_url=result.url, @@ -605,7 +457,7 @@ def find_form( # Fast-path: try common contact URLs first (cheap and often works even # when menu extraction fails or homepage blocks link discovery). - if intent == "contact": + if content_type in ("contact", "form"): try: parsed_base = urlparse(url) base = ( @@ -629,7 +481,7 @@ def find_form( page=page, url=cand, depth=0, - intent=intent, + intent=content_type, explored_pages=explored_pages, base_domain=urlparse(url).netloc, ) @@ -650,7 +502,7 @@ def find_form( if sitemap_urls: if intent == "contact": - _debug(f"Found {len(sitemap_urls)} sitemap URLs, prioritizing contact links") + debug(f"Found {len(sitemap_urls)} sitemap URLs, prioritizing contact links") # For contact intent, prioritize contact URLs from sitemap contact_sitemap_urls = [] other_sitemap_urls = [] @@ -677,7 +529,7 @@ def find_form( else: other_sitemap_urls.append(u) - _debug(f"Contact sitemap URLs: {contact_sitemap_urls[:3]}") + debug(f"Contact sitemap URLs: {contact_sitemap_urls[:3]}") # Explore contact URLs first for u in contact_sitemap_urls[:5]: @@ -741,7 +593,7 @@ def find_form( # Start exploration - use the same logic as find_content for contact intent if intent == "contact": - _debug(f"Using contact-aware exploration for {url}") + debug(f"Using contact-aware exploration for {url}") # Use the same logic as find_content for contact main_page_result = self._explore_recursive( page=page, @@ -752,15 +604,15 @@ def find_form( base_domain=urlparse(url).netloc, ) - _debug(f"Main page result: {main_page_result is not None}") + debug(f"Main page result: {main_page_result is not None}") if main_page_result: - _debug(f"Main page URL: {main_page_result.url}") - _debug(f"Main page has form: {main_page_result.has_form}, contact_fields: {main_page_result.contact_field_count}") - _debug(f"Main page links: {len(main_page_result.links)}") + debug(f"Main page URL: {main_page_result.url}") + debug(f"Main page has form: {main_page_result.has_form}, contact_fields: {main_page_result.contact_field_count}") + debug(f"Main page links: {len(main_page_result.links)}") # For contact intent, always explore contact links first even if main page has form if main_page_result and main_page_result.links and len(explored_pages) < self.max_pages: - _debug("Exploring links from main page") + debug("Exploring links from main page") # Sort links by contact relevance contact_links = [] other_links = [] @@ -775,23 +627,23 @@ def find_form( else: other_links.append(link) - _debug(f"Found {len(contact_links)} contact links: {contact_links}") + debug(f"Found {len(contact_links)} contact links: {contact_links}") # Explore contact links first for link in contact_links[:5]: # Check up to 5 contact links if len(self._explored_urls) >= self.max_pages: break - _debug(f"Exploring contact link: {link}") + debug(f"Exploring contact link: {link}") result = self._explore_recursive( page=page, url=link, depth=1, - intent=intent, + intent=content_type, explored_pages=explored_pages, base_domain=urlparse(url).netloc, ) if result and result.contact_field_count > 0: - _debug(f"Found contact form at: {link}") + debug(f"Found contact form at: {link}") return ExplorationResult( success=True, form_url=result.url, @@ -807,7 +659,7 @@ def find_form( page=page, url=link, depth=1, - intent=intent, + intent=content_type, explored_pages=explored_pages, base_domain=urlparse(url).netloc, ) @@ -948,532 +800,13 @@ def _explore_recursive( return None except Exception as e: - _debug(f"Playwright failed for {url}: {e}") + debug(f"Playwright failed for {url}: {e}") # Strategy 10: Graceful degradation β€” try static scrape fallback = self._fallback_static_scrape(url) if fallback: explored_pages.append(fallback) return fallback return None - - def _analyze_page(self, page: Any, url: str, console: Optional[Any] = None) -> PageInfo: - """Analyze a page for forms, iframes, and links.""" - info = PageInfo(url=url) - - try: - info.title = page.title() or "" - except Exception: - pass - - # Look for forms/fields - try: - inputs = page.query_selector_all('input:not([type="hidden"])') - textareas = page.query_selector_all('textarea') - selects = page.query_selector_all('select') - - info.form_count = len(inputs) + len(textareas) + len(selects) - info.has_form = info.form_count > 0 - except Exception: - inputs = [] - textareas = [] - selects = [] - info.form_count = 0 - info.has_form = False - - # Compute contact-like vs junk fields for contact intent. - # This helps avoid false positives (search boxes, cookie consent toggles, - # comment forms, captcha-only pages). - try: - field_nodes = [] - try: - field_nodes.extend(inputs[:30]) - except Exception: - field_nodes.extend(inputs) - try: - field_nodes.extend(textareas[:15]) - except Exception: - field_nodes.extend(textareas) - - def _is_junk_desc(field_type: str, name: str, fid: str, placeholder: str, aria: str) -> bool: - ft = (field_type or "").strip().lower() - n = (name or "").strip().lower() - i = (fid or "").strip().lower() - p = (placeholder or "").strip().lower() - a = (aria or "").strip().lower() - hay = " ".join([n, i, p, a]) - - if ft == "search" or n in {"s", "q", "search", "query"}: - return True - if "search" in hay or "szukaj" in hay or "wyszuki" in hay: - return True - - if "cookie" in hay or "consent" in hay: - return True - if i.startswith("cky") or "cky" in hay: - return True - if i.startswith("cmplz") or "cmplz" in hay: - return True - - if "captcha" in hay or "recaptcha" in hay or "g-recaptcha" in hay or "hcaptcha" in hay: - return True - - if n.startswith("apbct__") or "cleantalk" in hay: - return True - - if "comment" in hay or n in {"author", "email", "url"}: - return True - - return False - - def _is_contact_desc(field_type: str, name: str, fid: str, placeholder: str, aria: str) -> bool: - ft = (field_type or "").strip().lower() - n = (name or "").strip().lower() - i = (fid or "").strip().lower() - p = (placeholder or "").strip().lower() - a = (aria or "").strip().lower() - hay = " ".join([n, i, p, a]) - - if _is_junk_desc(field_type, name, fid, placeholder, aria): - return False - - if ft in {"email", "tel"}: - return True - if ft == "textarea": - return True - - tokens = [ - "email", - "e-mail", - "mail", - "telefon", - "phone", - "wiadomo", - "message", - "temat", - "subject", - "imi", - "name", - ] - return any(t in hay for t in tokens) - - for node in field_nodes: - try: - tag = (node.evaluate('el => el.tagName.toLowerCase()') or "").strip().lower() - except Exception: - tag = "" - try: - ftype = (node.get_attribute('type') or ("textarea" if tag == "textarea" else "text")) - except Exception: - ftype = "text" - try: - name = node.get_attribute('name') or "" - except Exception: - name = "" - try: - fid = node.get_attribute('id') or "" - except Exception: - fid = "" - try: - placeholder = node.get_attribute('placeholder') or "" - except Exception: - placeholder = "" - try: - aria = node.get_attribute('aria-label') or "" - except Exception: - aria = "" - - if _is_junk_desc(str(ftype), name, fid, placeholder, aria): - info.junk_field_count += 1 - if _is_contact_desc(str(ftype), name, fid, placeholder, aria): - info.contact_field_count += 1 - except Exception: - pass - - # Check for forms inside iframes (common for contact widgets) - if not info.has_form: - try: - iframes = page.query_selector_all('iframe') - for i, iframe in enumerate(iframes[:3]): # Check first 3 iframes - try: - frame = iframe.content_frame() - if frame: - # Count inputs in iframe - iframe_inputs = frame.query_selector_all('input:not([type="hidden"])') - iframe_textareas = frame.query_selector_all('textarea') - if len(iframe_inputs) > 0 or len(iframe_textareas) > 0: - info.has_form = True - info.form_count += len(iframe_inputs) + len(iframe_textareas) - break - except Exception: - continue - except Exception: - pass - - # Score page based on content - info.score = self._score_page(page, url, info) - - # Extract links for further exploration - try: - selector_groups = [ - 'nav a[href], header a[href], [role="navigation"] a[href]', - 'footer a[href]', - 'a[href]', - ] - for sel in selector_groups: - links = page.query_selector_all(sel) - for link in links: - try: - href = link.get_attribute('href') - if href: - absolute_url = urljoin(url, href) - parsed = urlparse(absolute_url) - if parsed.netloc == urlparse(url).netloc: - if not any(absolute_url.endswith(ext) for ext in ['.pdf', '.jpg', '.png', '.mp4']): - info.links.append(absolute_url) - except Exception: - continue - except Exception: - pass - - # Remove duplicates but preserve order - seen = set() - unique_links = [] - for link in info.links: - normalized = self._normalize_url(link) - if normalized not in seen: - seen.add(normalized) - unique_links.append(normalized) - info.links = unique_links[:10] # Limit links - - return info - - def _analyze_page_dispatch(self, page: Any, url: str, console: Optional[Any] = None) -> PageInfo: - """New page analysis using modular PageAnalyzer. - - This is the refactored version that uses the page_analysis package. - Falls back to legacy _analyze_page if modular version unavailable. - """ - if not _PAGE_ANALYSIS_AVAILABLE: - return self._analyze_page(page, url, console) - - try: - from nlp2cmd.page_analysis import PageAnalyzer - - analyzer = PageAnalyzer(max_links=10) - result = analyzer.analyze(page, url) - - # Convert PageAnalysisResult to PageInfo - info = PageInfo(url=url) - info.title = result.title - info.has_form = result.has_form - info.form_count = result.form_count - info.links = result.links - info.score = result.score - - # Copy field classification counts - info.contact_field_count = result.contact_field_count - info.junk_field_count = result.junk_field_count - - return info - - except Exception as e: - _debug(f"PageAnalyzer failed: {e}, falling back to legacy") - return self._analyze_page(page, url, console) - - def _dismiss_popups(self, page: Any) -> None: - """Try to dismiss common popups and cookie consents.""" - dismiss_selectors = [ - 'button:has-text("Accept all")', - 'button:has-text("Akceptuj wszystko")', - 'button:has-text("Zaakceptuj")', - 'button:has-text("Accept")', - 'button:has-text("Zgadzam siΔ™")', - 'button:has-text("Zgadzam sie")', - 'button:has-text("I agree")', - 'button:has-text("OK")', - 'button[aria-label*="Accept"]', - 'button[aria-label*="Akceptuj"]', - '[data-testid="cookie-accept"]', - '.cookie-accept', - '#onetrust-accept-btn-handler', - ] - - for selector in dismiss_selectors: - try: - page.wait_for_selector(selector, state="visible", timeout=1500) - page.click(selector, timeout=1500) - page.wait_for_timeout(500) - break - except Exception: - continue - - def _score_page(self, page: Any, url: str, info: PageInfo, intent: str = "contact") -> float: - """Score page relevance for finding content.""" - score = 0.0 - url_lower = url.lower() - title_lower = info.title.lower() - - # Choose keyword set based on intent - if intent == "contact": - keywords = self.CONTACT_KEYWORDS - elif intent == "article": - keywords = self.ARTICLE_KEYWORDS - elif intent == "product": - keywords = self.PRODUCT_KEYWORDS - elif intent == "docs": - keywords = self.DOCS_KEYWORDS - else: - keywords = self.CONTACT_KEYWORDS # Default - - # URL contains intent keywords - for kw in keywords: - if kw in url_lower: - score += 2.0 - - # Title contains intent keywords - for kw in keywords: - if kw in title_lower: - score += 1.5 - - # Special scoring for different content types - if intent == "contact" and info.has_form: - score += 4.0 # Increased boost for forms - # Check for email/phone fields (strong indicator of contact form) - try: - page_html = page.content().lower() - indicators = self.FORM_FIELD_KEYWORDS + ["required", "wyslij", "wyΕ›lij", "submit"] - for kw in indicators: - if kw in page_html: - score += 0.5 - except Exception: - pass - elif intent == "article": - # Check for article-like content - try: - page_html = page.content().lower() - article_indicators = [" bool: - """Check if a lowered URL looks like a contact/form page. - - Avoids false positives from words like 'informacje', 'platform', 'transform'. - """ - # Direct keyword hits - if any(kw in url_lower for kw in ["kontakt", "contact", "formularz"]): - return True - # Standalone "form" (not inside other words) - has_form_word = ( - "/form" in url_lower or url_lower.endswith("/form") - or "-form" in url_lower or "form-" in url_lower - ) and not any(w in url_lower for w in ["informacje", "platform", "transform", "perform", "reform"]) - return has_form_word - - def _find_best_form_candidate( - self, - pages: list[PageInfo], - intent: str, - ) -> Optional[PageInfo]: - """Find the best page with form based on scores.""" - # Filter pages with forms - form_pages = [p for p in pages if p.has_form] - if not form_pages: - return None - - # Prioritize pages with contact-related URLs for contact intent - if intent == "contact": - contact_urls = [p for p in form_pages if self._is_contact_url(p.url.lower())] - if contact_urls: - form_pages = contact_urls - - # Sort by score descending - form_pages.sort(key=lambda p: p.score, reverse=True) - return form_pages[0] - - @staticmethod - def _normalize_url(url: str) -> str: - """Normalize URL for comparison.""" - # Remove fragment - url = url.split('#')[0] - # Remove trailing slash - url = url.rstrip('/') - return url - - def _get_sitemap_urls(self, base_url: str) -> list[str]: - parsed = urlparse(base_url) - if not parsed.scheme or not parsed.netloc: - return [] - - sitemap_url = f"{parsed.scheme}://{parsed.netloc}/sitemap.xml" - - try: - with urlopen(sitemap_url, timeout=max(1, int(self.timeout_ms / 1000))) as resp: - raw = resp.read() - except Exception: - return [] - - try: - root = ET.fromstring(raw) - except Exception: - return [] - - ns = "" - if root.tag.startswith("{") and "}" in root.tag: - ns = root.tag.split("}", 1)[0] + "}" - - urls: list[str] = [] - - if root.tag.endswith("sitemapindex"): - for sm in root.findall(f"{ns}sitemap"): - loc = sm.find(f"{ns}loc") - if loc is None or not (loc.text or "").strip(): - continue - sm_url = (loc.text or "").strip() - try: - with urlopen(sm_url, timeout=max(1, int(self.timeout_ms / 1000))) as resp: - sm_raw = resp.read() - sm_root = ET.fromstring(sm_raw) - except Exception: - continue - - sm_ns = "" - if sm_root.tag.startswith("{") and "}" in sm_root.tag: - sm_ns = sm_root.tag.split("}", 1)[0] + "}" - - for u in sm_root.findall(f"{sm_ns}url"): - loc2 = u.find(f"{sm_ns}loc") - if loc2 is None: - continue - txt = (loc2.text or "").strip() - if not txt: - continue - if urlparse(txt).netloc != parsed.netloc: - continue - urls.append(txt) - if len(urls) >= self._max_sitemap_urls: - break - if len(urls) >= self._max_sitemap_urls: - break - else: - for u in root.findall(f"{ns}url"): - loc = u.find(f"{ns}loc") - if loc is None: - continue - txt = (loc.text or "").strip() - if not txt: - continue - if urlparse(txt).netloc != parsed.netloc: - continue - urls.append(txt) - if len(urls) >= self._max_sitemap_urls: - break - - if not urls: - return [] - - def _score_url(u: str) -> float: - ul = u.lower() - s = 0.0 - for kw in self.CONTACT_KEYWORDS: - if kw in ul: - s += 2.0 - if any(x in ul for x in ["kontakt", "contact", "formularz", "form", "wiadomosc", "wiadomoΕ›Δ‡"]): - s += 2.0 - if any(x in ul for x in ["tel", "email", "mail"]): - s += 1.0 - return s - - urls = sorted(urls, key=_score_url, reverse=True) - return urls[: self._max_sitemap_urls] - - def _has_content_type(self, page_info: PageInfo, content_type: str) -> bool: - """Check if page has specific content type.""" - if content_type in ["contact", "form"]: - return page_info.has_form - - intent_keywords = { - "article": self.ARTICLE_KEYWORDS, - "product": self.PRODUCT_KEYWORDS, - "docs": self.DOCS_KEYWORDS, - } - keywords = intent_keywords.get(content_type, []) - url_lower = page_info.url.lower() - title_lower = page_info.title.lower() - - for kw in keywords: - if kw in url_lower or kw in title_lower: - return True - return False - - def _find_best_content_candidate( - self, - pages: list[PageInfo], - content_type: str, - search_term: Optional[str] = None, - ) -> Optional[PageInfo]: - """Find best page for content type.""" - intent_keywords = { - "article": self.ARTICLE_KEYWORDS, - "product": self.PRODUCT_KEYWORDS, - "docs": self.DOCS_KEYWORDS, - } - keywords = intent_keywords.get(content_type, []) - - best_page = None - best_score = -1.0 - - for p in pages: - score = 0.0 - url_lower = p.url.lower() - title_lower = p.title.lower() - - for kw in keywords: - if kw in url_lower: - score += 2.0 - if kw in title_lower: - score += 1.5 - - if search_term: - st_lower = search_term.lower() - if st_lower in url_lower: - score += 3.0 - if st_lower in title_lower: - score += 2.0 - - if score > best_score: - best_score = score - best_page = p - - return best_page - - # ── Strategy 5: Two-Phase Exploration ───────────────────────────── def find_content_twophase( self, url: str, @@ -1486,7 +819,7 @@ def find_content_twophase( from playwright.sync_api import sync_playwright t0 = time.perf_counter() - _debug(f"Two-phase exploration: phase 1 (quick scan) for {url}") + debug(f"Two-phase exploration: phase 1 (quick scan) for {url}") # Phase 1: Quick scan β€” short timeouts, few pages quick_explorer = SiteExplorer( @@ -1506,11 +839,11 @@ def find_content_twophase( self._record_timing(url, "twophase_quick", phase1_ms) if quick_result.success: - _debug(f"Two-phase: found in phase 1 ({phase1_ms:.0f}ms)") + debug(f"Two-phase: found in phase 1 ({phase1_ms:.0f}ms)") return quick_result # Phase 2: Deep dive on discovered links - _debug(f"Two-phase: phase 2 (deep dive) on {len(quick_result.explored_pages)} candidates") + debug(f"Two-phase: phase 2 (deep dive) on {len(quick_result.explored_pages)} candidates") candidate_urls = [] for pg in quick_result.explored_pages: candidate_urls.extend(pg.links[:3]) @@ -1544,7 +877,7 @@ def find_content_twophase( if deep_result.success: total_ms = (time.perf_counter() - t0) * 1000 self._record_timing(url, "twophase_deep", total_ms) - _debug(f"Two-phase: found in phase 2 ({total_ms:.0f}ms)") + debug(f"Two-phase: found in phase 2 ({total_ms:.0f}ms)") return deep_result total_ms = (time.perf_counter() - t0) * 1000 @@ -1575,7 +908,7 @@ def _explore_links_parallel( results.append(info) except Exception: pass - _debug(f"Parallel scan: {len(results)}/{len(urls)} pages fetched") + debug(f"Parallel scan: {len(results)}/{len(urls)} pages fetched") return results def explore( diff --git a/src/nlp2cmd/web_schema/site_explorer_page.py b/src/nlp2cmd/web_schema/site_explorer_page.py new file mode 100644 index 00000000..df717902 --- /dev/null +++ b/src/nlp2cmd/web_schema/site_explorer_page.py @@ -0,0 +1,582 @@ +"""Page analysis, scoring, and sitemap helpers for SiteExplorer.""" + +from __future__ import annotations + +import re +import xml.etree.ElementTree as ET +from typing import Any, Optional +from urllib.parse import urljoin, urlparse +from urllib.request import urlopen, Request + +try: + from nlp2cmd.page_analysis import PageAnalyzer, PageAnalysisResult # noqa: F401 + PAGE_ANALYSIS_AVAILABLE = True +except ImportError: + PAGE_ANALYSIS_AVAILABLE = False + +from nlp2cmd.web_schema.site_explorer_types import ( + CONTACT_KEYWORDS, + ARTICLE_KEYWORDS, + PRODUCT_KEYWORDS, + DOCS_KEYWORDS, + FORM_FIELD_KEYWORDS, + PageInfo, + debug, +) + + +class SiteExplorerPageMixin: + """Page-level analysis helpers mixed into SiteExplorer.""" + + @staticmethod + def _fallback_static_scrape(url: str, timeout: int = 5) -> Optional[PageInfo]: + """Fallback: fetch page with urllib (no JS) when Playwright fails.""" + try: + req = Request(url, headers={"User-Agent": "nlp2cmd/1.0 (static fallback)"}) + with urlopen(req, timeout=timeout) as resp: + html = resp.read().decode("utf-8", errors="replace") + + info = PageInfo(url=url) + + # Extract title + m = re.search(r"]*>(.*?)", html, re.IGNORECASE | re.DOTALL) + if m: + info.title = m.group(1).strip() + + # Count form fields + inputs = len(re.findall(r' 0 + + # Extract links + for m in re.finditer(r'href=["\']([^"\']+)["\']', html): + href = m.group(1) + if href.startswith(("http://", "https://")): + info.links.append(href) + elif href.startswith("/"): + info.links.append(urljoin(url, href)) + info.links = info.links[:20] + + debug(f"Static fallback OK: {url} title='{info.title[:40]}' forms={info.form_count} links={len(info.links)}") + return info + except Exception as e: + debug(f"Static fallback failed for {url}: {e}") + return None + def _analyze_page(self, page: Any, url: str, console: Optional[Any] = None) -> PageInfo: + """Analyze a page for forms, iframes, and links.""" + info = PageInfo(url=url) + + try: + info.title = page.title() or "" + except Exception: + pass + + # Look for forms/fields + try: + inputs = page.query_selector_all('input:not([type="hidden"])') + textareas = page.query_selector_all('textarea') + selects = page.query_selector_all('select') + + info.form_count = len(inputs) + len(textareas) + len(selects) + info.has_form = info.form_count > 0 + except Exception: + inputs = [] + textareas = [] + selects = [] + info.form_count = 0 + info.has_form = False + + # Compute contact-like vs junk fields for contact intent. + # This helps avoid false positives (search boxes, cookie consent toggles, + # comment forms, captcha-only pages). + try: + field_nodes = [] + try: + field_nodes.extend(inputs[:30]) + except Exception: + field_nodes.extend(inputs) + try: + field_nodes.extend(textareas[:15]) + except Exception: + field_nodes.extend(textareas) + + def _is_junk_desc(field_type: str, name: str, fid: str, placeholder: str, aria: str) -> bool: + ft = (field_type or "").strip().lower() + n = (name or "").strip().lower() + i = (fid or "").strip().lower() + p = (placeholder or "").strip().lower() + a = (aria or "").strip().lower() + hay = " ".join([n, i, p, a]) + + if ft == "search" or n in {"s", "q", "search", "query"}: + return True + if "search" in hay or "szukaj" in hay or "wyszuki" in hay: + return True + + if "cookie" in hay or "consent" in hay: + return True + if i.startswith("cky") or "cky" in hay: + return True + if i.startswith("cmplz") or "cmplz" in hay: + return True + + if "captcha" in hay or "recaptcha" in hay or "g-recaptcha" in hay or "hcaptcha" in hay: + return True + + if n.startswith("apbct__") or "cleantalk" in hay: + return True + + if "comment" in hay or n in {"author", "email", "url"}: + return True + + return False + + def _is_contact_desc(field_type: str, name: str, fid: str, placeholder: str, aria: str) -> bool: + ft = (field_type or "").strip().lower() + n = (name or "").strip().lower() + i = (fid or "").strip().lower() + p = (placeholder or "").strip().lower() + a = (aria or "").strip().lower() + hay = " ".join([n, i, p, a]) + + if _is_junk_desc(field_type, name, fid, placeholder, aria): + return False + + if ft in {"email", "tel"}: + return True + if ft == "textarea": + return True + + tokens = [ + "email", + "e-mail", + "mail", + "telefon", + "phone", + "wiadomo", + "message", + "temat", + "subject", + "imi", + "name", + ] + return any(t in hay for t in tokens) + + for node in field_nodes: + try: + tag = (node.evaluate('el => el.tagName.toLowerCase()') or "").strip().lower() + except Exception: + tag = "" + try: + ftype = (node.get_attribute('type') or ("textarea" if tag == "textarea" else "text")) + except Exception: + ftype = "text" + try: + name = node.get_attribute('name') or "" + except Exception: + name = "" + try: + fid = node.get_attribute('id') or "" + except Exception: + fid = "" + try: + placeholder = node.get_attribute('placeholder') or "" + except Exception: + placeholder = "" + try: + aria = node.get_attribute('aria-label') or "" + except Exception: + aria = "" + + if _is_junk_desc(str(ftype), name, fid, placeholder, aria): + info.junk_field_count += 1 + if _is_contact_desc(str(ftype), name, fid, placeholder, aria): + info.contact_field_count += 1 + except Exception: + pass + + # Check for forms inside iframes (common for contact widgets) + if not info.has_form: + try: + iframes = page.query_selector_all('iframe') + for i, iframe in enumerate(iframes[:3]): # Check first 3 iframes + try: + frame = iframe.content_frame() + if frame: + # Count inputs in iframe + iframe_inputs = frame.query_selector_all('input:not([type="hidden"])') + iframe_textareas = frame.query_selector_all('textarea') + if len(iframe_inputs) > 0 or len(iframe_textareas) > 0: + info.has_form = True + info.form_count += len(iframe_inputs) + len(iframe_textareas) + break + except Exception: + continue + except Exception: + pass + + # Score page based on content + info.score = self._score_page(page, url, info) + + # Extract links for further exploration + try: + selector_groups = [ + 'nav a[href], header a[href], [role="navigation"] a[href]', + 'footer a[href]', + 'a[href]', + ] + for sel in selector_groups: + links = page.query_selector_all(sel) + for link in links: + try: + href = link.get_attribute('href') + if href: + absolute_url = urljoin(url, href) + parsed = urlparse(absolute_url) + if parsed.netloc == urlparse(url).netloc: + if not any(absolute_url.endswith(ext) for ext in ['.pdf', '.jpg', '.png', '.mp4']): + info.links.append(absolute_url) + except Exception: + continue + except Exception: + pass + + # Remove duplicates but preserve order + seen = set() + unique_links = [] + for link in info.links: + normalized = self._normalize_url(link) + if normalized not in seen: + seen.add(normalized) + unique_links.append(normalized) + info.links = unique_links[:10] # Limit links + + return info + + def _analyze_page_dispatch(self, page: Any, url: str, console: Optional[Any] = None) -> PageInfo: + """New page analysis using modular PageAnalyzer. + + This is the refactored version that uses the page_analysis package. + Falls back to legacy _analyze_page if modular version unavailable. + """ + if not PAGE_ANALYSIS_AVAILABLE: + return self._analyze_page(page, url, console) + + try: + from nlp2cmd.page_analysis import PageAnalyzer + + analyzer = PageAnalyzer(max_links=10) + result = analyzer.analyze(page, url) + + # Convert PageAnalysisResult to PageInfo + info = PageInfo(url=url) + info.title = result.title + info.has_form = result.has_form + info.form_count = result.form_count + info.links = result.links + info.score = result.score + + # Copy field classification counts + info.contact_field_count = result.contact_field_count + info.junk_field_count = result.junk_field_count + + return info + + except Exception as e: + debug(f"PageAnalyzer failed: {e}, falling back to legacy") + return self._analyze_page(page, url, console) + + def _dismiss_popups(self, page: Any) -> None: + """Try to dismiss common popups and cookie consents.""" + dismiss_selectors = [ + 'button:has-text("Accept all")', + 'button:has-text("Akceptuj wszystko")', + 'button:has-text("Zaakceptuj")', + 'button:has-text("Accept")', + 'button:has-text("Zgadzam siΔ™")', + 'button:has-text("Zgadzam sie")', + 'button:has-text("I agree")', + 'button:has-text("OK")', + 'button[aria-label*="Accept"]', + 'button[aria-label*="Akceptuj"]', + '[data-testid="cookie-accept"]', + '.cookie-accept', + '#onetrust-accept-btn-handler', + ] + + for selector in dismiss_selectors: + try: + page.wait_for_selector(selector, state="visible", timeout=1500) + page.click(selector, timeout=1500) + page.wait_for_timeout(500) + break + except Exception: + continue + + def _score_page(self, page: Any, url: str, info: PageInfo, intent: str = "contact") -> float: + """Score page relevance for finding content.""" + score = 0.0 + url_lower = url.lower() + title_lower = info.title.lower() + + # Choose keyword set based on intent + if intent == "contact": + keywords = CONTACT_KEYWORDS + elif intent == "article": + keywords = ARTICLE_KEYWORDS + elif intent == "product": + keywords = PRODUCT_KEYWORDS + elif intent == "docs": + keywords = DOCS_KEYWORDS + else: + keywords = CONTACT_KEYWORDS # Default + + # URL contains intent keywords + for kw in keywords: + if kw in url_lower: + score += 2.0 + + # Title contains intent keywords + for kw in keywords: + if kw in title_lower: + score += 1.5 + + # Special scoring for different content types + if intent == "contact" and info.has_form: + score += 4.0 # Increased boost for forms + # Check for email/phone fields (strong indicator of contact form) + try: + page_html = page.content().lower() + indicators = FORM_FIELD_KEYWORDS + ["required", "wyslij", "wyΕ›lij", "submit"] + for kw in indicators: + if kw in page_html: + score += 0.5 + except Exception: + pass + elif intent == "article": + # Check for article-like content + try: + page_html = page.content().lower() + article_indicators = [" bool: + """Check if a lowered URL looks like a contact/form page. + + Avoids false positives from words like 'informacje', 'platform', 'transform'. + """ + # Direct keyword hits + if any(kw in url_lower for kw in ["kontakt", "contact", "formularz"]): + return True + # Standalone "form" (not inside other words) + has_form_word = ( + "/form" in url_lower or url_lower.endswith("/form") + or "-form" in url_lower or "form-" in url_lower + ) and not any(w in url_lower for w in ["informacje", "platform", "transform", "perform", "reform"]) + return has_form_word + + def _find_best_form_candidate( + self, + pages: list[PageInfo], + intent: str, + ) -> Optional[PageInfo]: + """Find the best page with form based on scores.""" + # Filter pages with forms + form_pages = [p for p in pages if p.has_form] + if not form_pages: + return None + + # Prioritize pages with contact-related URLs for contact intent + if intent == "contact": + contact_urls = [p for p in form_pages if self._is_contact_url(p.url.lower())] + if contact_urls: + form_pages = contact_urls + + # Sort by score descending + form_pages.sort(key=lambda p: p.score, reverse=True) + return form_pages[0] + + @staticmethod + def _normalize_url(url: str) -> str: + """Normalize URL for comparison.""" + # Remove fragment + url = url.split('#')[0] + # Remove trailing slash + url = url.rstrip('/') + return url + + def _get_sitemap_urls(self, base_url: str) -> list[str]: + parsed = urlparse(base_url) + if not parsed.scheme or not parsed.netloc: + return [] + + sitemap_url = f"{parsed.scheme}://{parsed.netloc}/sitemap.xml" + + try: + with urlopen(sitemap_url, timeout=max(1, int(self.timeout_ms / 1000))) as resp: + raw = resp.read() + except Exception: + return [] + + try: + root = ET.fromstring(raw) + except Exception: + return [] + + ns = "" + if root.tag.startswith("{") and "}" in root.tag: + ns = root.tag.split("}", 1)[0] + "}" + + urls: list[str] = [] + + if root.tag.endswith("sitemapindex"): + for sm in root.findall(f"{ns}sitemap"): + loc = sm.find(f"{ns}loc") + if loc is None or not (loc.text or "").strip(): + continue + sm_url = (loc.text or "").strip() + try: + with urlopen(sm_url, timeout=max(1, int(self.timeout_ms / 1000))) as resp: + sm_raw = resp.read() + sm_root = ET.fromstring(sm_raw) + except Exception: + continue + + sm_ns = "" + if sm_root.tag.startswith("{") and "}" in sm_root.tag: + sm_ns = sm_root.tag.split("}", 1)[0] + "}" + + for u in sm_root.findall(f"{sm_ns}url"): + loc2 = u.find(f"{sm_ns}loc") + if loc2 is None: + continue + txt = (loc2.text or "").strip() + if not txt: + continue + if urlparse(txt).netloc != parsed.netloc: + continue + urls.append(txt) + if len(urls) >= self._max_sitemap_urls: + break + if len(urls) >= self._max_sitemap_urls: + break + else: + for u in root.findall(f"{ns}url"): + loc = u.find(f"{ns}loc") + if loc is None: + continue + txt = (loc.text or "").strip() + if not txt: + continue + if urlparse(txt).netloc != parsed.netloc: + continue + urls.append(txt) + if len(urls) >= self._max_sitemap_urls: + break + + if not urls: + return [] + + def _score_url(u: str) -> float: + ul = u.lower() + s = 0.0 + for kw in CONTACT_KEYWORDS: + if kw in ul: + s += 2.0 + if any(x in ul for x in ["kontakt", "contact", "formularz", "form", "wiadomosc", "wiadomoΕ›Δ‡"]): + s += 2.0 + if any(x in ul for x in ["tel", "email", "mail"]): + s += 1.0 + return s + + urls = sorted(urls, key=_score_url, reverse=True) + return urls[: self._max_sitemap_urls] + + def _has_content_type(self, page_info: PageInfo, content_type: str) -> bool: + """Check if page has specific content type.""" + if content_type in ["contact", "form"]: + return page_info.has_form + + intent_keywords = { + "article": ARTICLE_KEYWORDS, + "product": PRODUCT_KEYWORDS, + "docs": DOCS_KEYWORDS, + } + keywords = intent_keywords.get(content_type, []) + url_lower = page_info.url.lower() + title_lower = page_info.title.lower() + + for kw in keywords: + if kw in url_lower or kw in title_lower: + return True + return False + + def _find_best_content_candidate( + self, + pages: list[PageInfo], + content_type: str, + search_term: Optional[str] = None, + ) -> Optional[PageInfo]: + """Find best page for content type.""" + intent_keywords = { + "article": ARTICLE_KEYWORDS, + "product": PRODUCT_KEYWORDS, + "docs": DOCS_KEYWORDS, + } + keywords = intent_keywords.get(content_type, []) + + best_page = None + best_score = -1.0 + + for p in pages: + score = 0.0 + url_lower = p.url.lower() + title_lower = p.title.lower() + + for kw in keywords: + if kw in url_lower: + score += 2.0 + if kw in title_lower: + score += 1.5 + + if search_term: + st_lower = search_term.lower() + if st_lower in url_lower: + score += 3.0 + if st_lower in title_lower: + score += 2.0 + + if score > best_score: + best_score = score + best_page = p + + return best_page + diff --git a/src/nlp2cmd/web_schema/site_explorer_types.py b/src/nlp2cmd/web_schema/site_explorer_types.py new file mode 100644 index 00000000..d9c9c620 --- /dev/null +++ b/src/nlp2cmd/web_schema/site_explorer_types.py @@ -0,0 +1,134 @@ +"""Types, constants, and platform URL helpers for site exploration.""" + +from __future__ import annotations + +import os +import sys +from dataclasses import dataclass, field +from typing import Any, Optional +from urllib.parse import urlparse + +_DEBUG = os.environ.get("NLP2CMD_DEBUG", "").lower() in ("1", "true", "yes") + + +def debug(msg: str) -> None: + """Print debug message to stderr when NLP2CMD_DEBUG=1.""" + if _DEBUG: + print(f"DEBUG [SiteExplorer] {msg}", file=sys.stderr, flush=True) + + +def github_readme_url(url: str) -> str: + """Convert github.com/owner/repo to raw README URL.""" + parsed = urlparse(url) + parts = [p for p in parsed.path.strip("/").split("/") if p] + if len(parts) >= 2: + return f"https://github.com/{parts[0]}/{parts[1]}" + return url + + +def github_docs_url(url: str) -> str: + """Try to resolve GitHub repo docs (wiki, /docs, or README).""" + parsed = urlparse(url) + parts = [p for p in parsed.path.strip("/").split("/") if p] + if len(parts) >= 2: + return f"https://github.com/{parts[0]}/{parts[1]}/tree/main/docs" + return url + + +def pypi_to_docs_url(url: str) -> Optional[str]: + """Convert pypi.org/project/X to readthedocs or homepage.""" + parsed = urlparse(url) + parts = [p for p in parsed.path.strip("/").split("/") if p] + if len(parts) >= 2 and parts[0] == "project": + pkg = parts[1].lower().replace("-", "").replace("_", "") + return f"https://{pkg}.readthedocs.io/en/latest/" + return None + + +@dataclass +class PageInfo: + """Information about a discovered page.""" + url: str + title: str = "" + links: list[str] = field(default_factory=list) + has_form: bool = False + form_count: int = 0 + contact_field_count: int = 0 + junk_field_count: int = 0 + score: float = 0.0 + load_time_ms: float = 0.0 + + +@dataclass +class ExplorationResult: + """Result of site exploration.""" + success: bool + form_url: Optional[str] = None + form_page: Optional[PageInfo] = None + explored_pages: list[PageInfo] = field(default_factory=list) + error: Optional[str] = None + + +CONTACT_KEYWORDS = [ + "kontakt", "contact", "napisz do nas", "write to us", + "formularz", "form", "wiadomoΕ›Δ‡", "message", + "pomoc", "help", "support", "serwis", + "zapytaj", "ask", "biuro", "office", "dane", "info", + "obsΕ‚uga", "obsuga", "klienta", "customer", +] + +ARTICLE_KEYWORDS = [ + "artykuΕ‚", "article", "blog", "news", "wiadomoΕ›ci", "aktualnoΕ›ci", + "publikacja", "publication", "post", "wpis", "treΕ›Δ‡", "content", + "poradnik", "guide", "tutorial", "instrukcja", "manual", +] + +PRODUCT_KEYWORDS = [ + "produkt", "product", "usΕ‚uga", "service", "oferta", "offer", + "sklep", "shop", "store", "cennik", "price", "cena", "buy", + "katalog", "catalog", "portfolio", "galeria", "gallery", +] + +DOCS_KEYWORDS = [ + "dokumentacja", "documentation", "docs", "pomoc", "help", + "faq", "pytania", "questions", "support", "wsparcie", + "manual", "instrukcja", "guide", "tutorial", "readme", + "wiki", "api", "reference", "examples", "przykΕ‚ady", + "github", "gitlab", "bitbucket", "repository", "repo", +] + +FORM_FIELD_KEYWORDS = [ + "email", "e-mail", "telefon", "phone", "imiΔ™", "name", + "nazwisko", "surname", "wiadomoΕ›Δ‡", "message", "temat", "subject", +] + +BLOCKED_RESOURCE_PATTERNS = ( + "**/*.png", "**/*.jpg", "**/*.jpeg", "**/*.gif", "**/*.svg", + "**/*.webp", "**/*.ico", "**/*.bmp", "**/*.tiff", + "**/*.woff", "**/*.woff2", "**/*.ttf", "**/*.eot", + "**/*.mp4", "**/*.webm", "**/*.ogg", "**/*.mp3", +) + +PLATFORM_DOCS_URLS: dict[str, Any] = { + "github.com": { + "readme": github_readme_url, + "docs": github_docs_url, + }, + "readthedocs.io": { + "docs": lambda url: url if "/en/" in url else url.rstrip("/") + "/en/latest/", + }, + "docs.python.org": { + "docs": lambda _url: "https://docs.python.org/3/", + }, + "pypi.org": { + "docs": pypi_to_docs_url, + }, +} + +DOCS_FRAMEWORKS: dict[str, list[str]] = { + "readthedocs": ["/en/latest/", "/en/stable/", "readthedocs.io"], + "mkdocs": ["/mkdocs.yml", "mkdocs-material", "/site/"], + "gitbook": ["gitbook.io", ".gitbook.io"], + "sphinx": ["/_static/sphinx", "searchindex.js", "genindex.html"], + "docusaurus": ["/docs/", "/blog/", "docusaurus"], +} diff --git a/tests/unit/test_automation.py b/tests/unit/test_automation.py index e172daac..e0703eab 100644 --- a/tests/unit/test_automation.py +++ b/tests/unit/test_automation.py @@ -467,8 +467,11 @@ def test_is_configured_false(self): client.api_key = None assert client.is_configured is False - def test_is_configured_true(self): + def test_is_configured_true(self, monkeypatch): + from nlp2cmd.llm import openrouter as or_mod from nlp2cmd.llm.openrouter import OpenRouterClient + + monkeypatch.setattr(or_mod, "available_routes", lambda *args, **kwargs: True) client = OpenRouterClient(api_key="test-key") assert client.is_configured is True