This is page 2 of 3. Use http://codebase.md/vibheksoni/stealth-browser-mcp?page={x} to view the full context.
# Directory Structure
```
├── .dockerignore
├── .github
│ ├── FUNDING.yml
│ ├── ISSUE_TEMPLATE
│ │ ├── bug_report.md
│ │ ├── config.yml
│ │ ├── feature_request.md
│ │ └── showcase.yml
│ ├── labeler.yml
│ ├── pull_request_template.md
│ └── workflows
│ └── ci.yml
├── .gitignore
├── CHANGELOG.md
├── Checklist.md
├── CODE_OF_CONDUCT.md
├── CODEOWNERS
├── COMPARISON.md
├── CONTRIBUTING.md
├── demo
│ ├── augment-hero-clone.md
│ ├── augment-hero-recreation.html
│ └── README.md
├── Dockerfile
├── examples
│ └── claude_prompts.md
├── HALL_OF_FAME.md
├── LICENSE
├── media
│ ├── AugmentHeroClone.PNG
│ ├── Showcase Stealth Browser Mcp.mp4
│ ├── showcase-demo-full.gif
│ ├── showcase-demo.gif
│ └── UndetectedStealthBrowser.png
├── pyproject.toml
├── README.md
├── requirements.txt
├── ROADMAP.md
├── run_server.bat
├── run_server.sh
├── SECURITY.md
├── smithery.yaml
└── src
├── __init__.py
├── browser_manager.py
├── cdp_element_cloner.py
├── cdp_function_executor.py
├── comprehensive_element_cloner.py
├── debug_logger.py
├── dom_handler.py
├── dynamic_hook_ai_interface.py
├── dynamic_hook_system.py
├── element_cloner.py
├── file_based_element_cloner.py
├── hook_learning_system.py
├── js
│ ├── comprehensive_element_extractor.js
│ ├── extract_animations.js
│ ├── extract_assets.js
│ ├── extract_events.js
│ ├── extract_related_files.js
│ ├── extract_structure.js
│ └── extract_styles.js
├── models.py
├── network_interceptor.py
├── persistent_storage.py
├── platform_utils.py
├── process_cleanup.py
├── progressive_element_cloner.py
├── response_handler.py
├── response_stage_hooks.py
└── server.py
```
# Files
--------------------------------------------------------------------------------
/src/debug_logger.py:
--------------------------------------------------------------------------------
```python
import json
import traceback
from datetime import datetime
from typing import Dict, List, Any, Optional
from collections import defaultdict
import threading
import pickle
import gzip
import os
import asyncio
from concurrent.futures import ThreadPoolExecutor, TimeoutError
class DebugLogger:
"""Centralized debug logging system for the MCP server."""
def __init__(self):
"""
Initializes the DebugLogger.
Variables:
self._errors (List[Dict[str, Any]]): Stores error logs.
self._warnings (List[Dict[str, Any]]): Stores warning logs.
self._info (List[Dict[str, Any]]): Stores info logs.
self._stats (Dict[str, int]): Stores statistics for errors, warnings, and calls.
self._lock (threading.Lock): Ensures thread safety for logging.
self._enabled (bool): Indicates if logging is enabled.
self._seen_errors (set): Track error signatures to prevent duplicates.
"""
self._errors: List[Dict[str, Any]] = []
self._warnings: List[Dict[str, Any]] = []
self._info: List[Dict[str, Any]] = []
self._stats: Dict[str, int] = defaultdict(int)
self._lock = threading.Lock()
self._enabled = True
self._lock_owner = "none"
import time
self._lock_acquired_time = 0
self._seen_errors: set = set()
def log_error(self, component: str, method: str, error: Exception, context: Optional[Dict[str, Any]] = None):
"""
Log an error with full context.
Args:
component (str): Name of the component where the error occurred.
method (str): Name of the method where the error occurred.
error (Exception): The exception instance.
context (Optional[Dict[str, Any]]): Additional context for the error.
"""
if not self._enabled:
return
with self._lock:
error_signature = f"{component}.{method}.{type(error).__name__}.{str(error)}"
if error_signature in self._seen_errors:
self._stats[f'{component}.{method}.errors'] += 1
return
self._seen_errors.add(error_signature)
error_entry = {
'timestamp': datetime.now().isoformat(),
'component': component,
'method': method,
'error_type': type(error).__name__,
'error_message': str(error),
'traceback': traceback.format_exc(),
'context': context or {}
}
self._errors.append(error_entry)
self._stats[f'{component}.{method}.errors'] += 1
print(f"[DEBUG ERROR] {component}.{method}: {error}")
def log_warning(self, component: str, method: str, message: str, context: Optional[Dict[str, Any]] = None):
"""
Log a warning.
Args:
component (str): Name of the component where the warning occurred.
method (str): Name of the method where the warning occurred.
message (str): Warning message.
context (Optional[Dict[str, Any]]): Additional context for the warning.
"""
if not self._enabled:
return
with self._lock:
warning_entry = {
'timestamp': datetime.now().isoformat(),
'component': component,
'method': method,
'message': message,
'context': context or {}
}
self._warnings.append(warning_entry)
self._stats[f'{component}.{method}.warnings'] += 1
print(f"[DEBUG WARN] {component}.{method}: {message}")
def log_info(self, component: str, method: str, message: str, data: Optional[Any] = None):
"""
Log information for debugging.
Args:
component (str): Name of the component where the info is logged.
method (str): Name of the method where the info is logged.
message (str): Info message.
data (Optional[Any]): Additional data for the info log.
"""
if not self._enabled:
return
with self._lock:
info_entry = {
'timestamp': datetime.now().isoformat(),
'component': component,
'method': method,
'message': message,
'data': data
}
self._info.append(info_entry)
self._stats[f'{component}.{method}.calls'] += 1
print(f"[DEBUG INFO] {component}.{method}: {message}")
if data:
print(f" Data: {data}")
def get_debug_view(self) -> Dict[str, Any]:
"""
Get comprehensive debug view of all logged data.
Returns:
Dict[str, Any]: Dictionary containing summary, recent errors/warnings, all errors/warnings, and component breakdown.
"""
return self.get_debug_view_paginated()
def get_debug_view_paginated(
self,
max_errors: Optional[int] = None,
max_warnings: Optional[int] = None,
max_info: Optional[int] = None
) -> Dict[str, Any]:
"""
Get paginated debug view of logged data with size limits.
Args:
max_errors (Optional[int]): Maximum number of errors to include. None for all.
max_warnings (Optional[int]): Maximum number of warnings to include. None for all.
max_info (Optional[int]): Maximum number of info logs to include. None for all.
Returns:
Dict[str, Any]: Dictionary containing summary, recent errors/warnings, limited errors/warnings, and component breakdown.
"""
with self._lock:
if max_errors is not None:
limited_errors = self._errors[-max_errors:] if self._errors else []
all_errors = limited_errors
else:
limited_errors = self._errors[-10:] if self._errors else []
all_errors = self._errors
if max_warnings is not None:
limited_warnings = self._warnings[-max_warnings:] if self._warnings else []
all_warnings = limited_warnings
else:
limited_warnings = self._warnings[-10:] if self._warnings else []
all_warnings = self._warnings
if max_info is not None:
limited_info = self._info[-max_info:] if self._info else []
all_info = limited_info
else:
limited_info = self._info[-10:] if self._info else []
all_info = self._info
return {
'summary': {
'total_errors': len(self._errors),
'total_warnings': len(self._warnings),
'total_info': len(self._info),
'returned_errors': len(all_errors),
'returned_warnings': len(all_warnings),
'returned_info': len(all_info),
'error_types': self._get_error_summary(),
'stats': dict(self._stats)
},
'recent_errors': limited_errors,
'recent_warnings': limited_warnings,
'recent_info': limited_info,
'all_errors': all_errors,
'all_warnings': all_warnings,
'all_info': all_info,
'component_breakdown': self._get_component_breakdown()
}
def _get_error_summary(self) -> Dict[str, int]:
"""
Get summary of error types.
Returns:
Dict[str, int]: Dictionary mapping error type names to their counts.
"""
error_types = defaultdict(int)
for error in self._errors:
error_types[error['error_type']] += 1
return dict(error_types)
def _get_component_breakdown(self) -> Dict[str, Dict[str, int]]:
"""
Get breakdown by component.
Returns:
Dict[str, Dict[str, int]]: Dictionary mapping component names to their error, warning, and call counts.
"""
breakdown = defaultdict(lambda: {'errors': 0, 'warnings': 0, 'calls': 0})
for error in self._errors:
breakdown[error['component']]['errors'] += 1
for warning in self._warnings:
breakdown[warning['component']]['warnings'] += 1
for info in self._info:
breakdown[info['component']]['calls'] += 1
return dict(breakdown)
def clear_debug_view(self):
"""
Clear all debug logs with timeout protection.
Variables:
self._errors (List[Dict[str, Any]]): Cleared.
self._warnings (List[Dict[str, Any]]): Cleared.
self._info (List[Dict[str, Any]]): Cleared.
self._stats (Dict[str, int]): Cleared.
"""
try:
if self._lock.acquire(timeout=5.0):
try:
self._errors.clear()
self._warnings.clear()
self._info.clear()
self._stats.clear()
print("[DEBUG] Debug logs cleared")
finally:
self._lock.release()
else:
print("[DEBUG] Failed to clear logs - timeout acquiring lock")
except Exception as e:
print(f"[DEBUG] Error clearing logs: {e}")
def clear_debug_view_safe(self):
"""
Safe version that recreates data structures if lock fails.
"""
try:
self.clear_debug_view()
except:
self._errors = []
self._warnings = []
self._info = []
self._stats = defaultdict(int)
print("[DEBUG] Debug logs force-cleared (lock bypass)")
def enable(self):
"""
Enable debug logging.
Variables:
self._enabled (bool): Set to True.
"""
self._enabled = True
print("[DEBUG] Debug logging enabled")
def disable(self):
"""
Disable debug logging.
Variables:
self._enabled (bool): Set to False.
"""
self._enabled = False
print("[DEBUG] Debug logging disabled")
def get_lock_status(self) -> Dict[str, Any]:
"""Get current lock status for debugging."""
import time
return {
"lock_owner": self._lock_owner,
"lock_held_duration": time.time() - self._lock_acquired_time if self._lock_acquired_time > 0 else 0,
"lock_acquired": self._lock.locked() if hasattr(self._lock, 'locked') else "unknown"
}
def export_to_file(self, filepath: str = "debug_log.json"):
"""
Export debug logs to a JSON file.
Args:
filepath (str): Path to the file where logs will be exported.
Returns:
str: The filepath where logs were exported.
"""
return self.export_to_file_paginated(filepath)
def export_to_file_paginated(
self,
filepath: str = "debug_log.json",
max_errors: Optional[int] = None,
max_warnings: Optional[int] = None,
max_info: Optional[int] = None,
format: str = "auto"
):
"""
Export paginated debug logs to a file using fastest method available.
Args:
filepath (str): Path to the file where logs will be exported.
max_errors (Optional[int]): Maximum number of errors to export. None for all.
max_warnings (Optional[int]): Maximum number of warnings to export. None for all.
max_info (Optional[int]): Maximum number of info logs to export. None for all.
format (str): Export format: 'json', 'pickle', 'gzip-pickle', 'auto' (default: 'auto').
Returns:
str: The filepath where logs were exported.
"""
import time
try:
print(f"[DEBUG] export_debug_logs attempting lock acquisition...")
current_status = self.get_lock_status()
print(f"[DEBUG] Current lock status: {current_status}")
acquired = self._lock.acquire(timeout=5.0)
if not acquired:
print("[DEBUG] Lock timeout - falling back to lock-free export")
return self._export_lockfree(filepath, max_errors, max_warnings, max_info, format)
self._lock_owner = "export_debug_logs"
self._lock_acquired_time = time.time()
print("[DEBUG] Lock acquired by export_debug_logs")
try:
debug_data = self.get_debug_view_paginated(
max_errors=max_errors,
max_warnings=max_warnings,
max_info=max_info
)
finally:
self._lock_owner = "none"
self._lock_acquired_time = 0
self._lock.release()
print("[DEBUG] Lock released by export_debug_logs")
except Exception as e:
print(f"[DEBUG] Exception in export: {e}")
return self._export_lockfree(filepath, max_errors, max_warnings, max_info, format)
if format == "auto":
total_items = (debug_data['summary']['returned_errors'] +
debug_data['summary']['returned_warnings'] +
debug_data['summary']['returned_info'])
if total_items > 1000:
format = "gzip-pickle"
elif total_items > 100:
format = "pickle"
else:
format = "json"
if format == "gzip-pickle":
return self._export_gzip_pickle(debug_data, filepath)
elif format == "pickle":
return self._export_pickle(debug_data, filepath)
else:
return self._export_json(debug_data, filepath)
def _export_lockfree(self, filepath: str, max_errors: Optional[int], max_warnings: Optional[int], max_info: Optional[int], format: str) -> str:
"""
Lock-free export method that creates a snapshot without acquiring locks.
"""
errors_snapshot = list(self._errors)
warnings_snapshot = list(self._warnings)
info_snapshot = list(self._info)
if max_errors is not None:
errors_snapshot = errors_snapshot[:max_errors]
if max_warnings is not None:
warnings_snapshot = warnings_snapshot[:max_warnings]
if max_info is not None:
info_snapshot = info_snapshot[:max_info]
debug_data = {
'summary': {
'total_errors': len(self._errors),
'total_warnings': len(self._warnings),
'total_info': len(self._info),
'returned_errors': len(errors_snapshot),
'returned_warnings': len(warnings_snapshot),
'returned_info': len(info_snapshot)
},
'all_errors': errors_snapshot,
'all_warnings': warnings_snapshot,
'all_info': info_snapshot
}
if format == "auto":
total_items = len(errors_snapshot) + len(warnings_snapshot) + len(info_snapshot)
if total_items > 1000:
format = "gzip-pickle"
elif total_items > 100:
format = "pickle"
else:
format = "json"
if format == "gzip-pickle":
return self._export_gzip_pickle(debug_data, filepath)
elif format == "pickle":
return self._export_pickle(debug_data, filepath)
else:
return self._export_json(debug_data, filepath)
def _export_gzip_pickle(self, debug_data: Dict[str, Any], filepath: str) -> str:
if not filepath.endswith('.pkl.gz'):
filepath = filepath.replace('.json', '.pkl.gz')
with gzip.open(filepath, 'wb') as f:
pickle.dump(debug_data, f, protocol=pickle.HIGHEST_PROTOCOL)
file_size = os.path.getsize(filepath)
print(f"[DEBUG] Exported {debug_data['summary']['returned_errors']} errors, "
f"{debug_data['summary']['returned_warnings']} warnings, "
f"{debug_data['summary']['returned_info']} info logs to {filepath} "
f"({file_size} bytes, gzip-pickle format)")
return filepath
def _export_pickle(self, debug_data: Dict[str, Any], filepath: str) -> str:
"""Export using pickle (fast for medium data)."""
if not filepath.endswith('.pkl'):
filepath = filepath.replace('.json', '.pkl')
with open(filepath, 'wb') as f:
pickle.dump(debug_data, f, protocol=pickle.HIGHEST_PROTOCOL)
file_size = os.path.getsize(filepath)
print(f"[DEBUG] Exported {debug_data['summary']['returned_errors']} errors, "
f"{debug_data['summary']['returned_warnings']} warnings, "
f"{debug_data['summary']['returned_info']} info logs to {filepath} "
f"({file_size} bytes, pickle format)")
return filepath
def _export_json(self, debug_data: Dict[str, Any], filepath: str) -> str:
"""Export using JSON (human readable but slower)."""
with open(filepath, 'w') as f:
json.dump(debug_data, f, separators=(',', ':'), default=str)
file_size = os.path.getsize(filepath)
print(f"[DEBUG] Exported {debug_data['summary']['returned_errors']} errors, "
f"{debug_data['summary']['returned_warnings']} warnings, "
f"{debug_data['summary']['returned_info']} info logs to {filepath} "
f"({file_size} bytes, JSON format)")
return filepath
debug_logger = DebugLogger()
```
--------------------------------------------------------------------------------
/src/browser_manager.py:
--------------------------------------------------------------------------------
```python
"""Browser instance management with nodriver."""
import asyncio
import uuid
from typing import Dict, Optional, List
from datetime import datetime, timedelta
import nodriver as uc
from nodriver import Browser, Tab
from debug_logger import debug_logger
from models import BrowserInstance, BrowserState, BrowserOptions, PageState
from persistent_storage import persistent_storage
from dynamic_hook_system import dynamic_hook_system
from platform_utils import get_platform_info
from process_cleanup import process_cleanup
class BrowserManager:
"""Manages multiple browser instances."""
def __init__(self):
self._instances: Dict[str, dict] = {}
self._lock = asyncio.Lock()
async def spawn_browser(self, options: BrowserOptions) -> BrowserInstance:
"""
Spawn a new browser instance with given options.
Args:
options (BrowserOptions): Options for browser configuration.
Returns:
BrowserInstance: The spawned browser instance.
"""
instance_id = str(uuid.uuid4())
instance = BrowserInstance(
instance_id=instance_id,
headless=options.headless,
user_agent=options.user_agent,
viewport={"width": options.viewport_width, "height": options.viewport_height}
)
try:
platform_info = get_platform_info()
debug_logger.log_info(
"browser_manager",
"spawn_browser",
f"Platform info: {platform_info['system']} | Root: {platform_info['is_root']} | Container: {platform_info['is_container']} | Sandbox: {options.sandbox}"
)
config = uc.Config(
headless=options.headless,
user_data_dir=options.user_data_dir,
sandbox=options.sandbox
)
browser = await uc.start(config=config)
tab = browser.main_tab
if hasattr(browser, '_process') and browser._process:
process_cleanup.track_browser_process(instance_id, browser._process)
else:
debug_logger.log_warning("browser_manager", "spawn_browser",
f"Browser {instance_id} has no process to track")
if options.user_agent:
await tab.send(uc.cdp.emulation.set_user_agent_override(
user_agent=options.user_agent
))
if options.extra_headers:
await tab.send(uc.cdp.network.set_extra_http_headers(
headers=options.extra_headers
))
await tab.set_window_size(
left=0,
top=0,
width=options.viewport_width,
height=options.viewport_height
)
print(f"[DEBUG] Set viewport to {options.viewport_width}x{options.viewport_height}")
await self._setup_dynamic_hooks(tab, instance_id)
async with self._lock:
self._instances[instance_id] = {
'browser': browser,
'tab': tab,
'instance': instance,
'options': options,
'network_data': []
}
instance.state = BrowserState.READY
instance.update_activity()
persistent_storage.store_instance(instance_id, {
'state': instance.state.value,
'created_at': instance.created_at.isoformat(),
'current_url': getattr(tab, 'url', ''),
'title': 'Browser Instance'
})
except Exception as e:
instance.state = BrowserState.ERROR
raise Exception(f"Failed to spawn browser: {str(e)}")
return instance
async def _setup_dynamic_hooks(self, tab: Tab, instance_id: str):
"""Setup dynamic hook system for browser instance."""
try:
dynamic_hook_system.add_instance(instance_id)
await dynamic_hook_system.setup_interception(tab, instance_id)
debug_logger.log_info("browser_manager", "_setup_dynamic_hooks", f"Dynamic hook system setup complete for instance {instance_id}")
except Exception as e:
debug_logger.log_error("browser_manager", "_setup_dynamic_hooks", f"Failed to setup dynamic hooks for {instance_id}: {e}")
async def get_instance(self, instance_id: str) -> Optional[dict]:
"""
Get browser instance by ID.
Args:
instance_id (str): The ID of the browser instance.
Returns:
Optional[dict]: The browser instance data if found, else None.
"""
async with self._lock:
return self._instances.get(instance_id)
async def list_instances(self) -> List[BrowserInstance]:
"""
List all browser instances.
Returns:
List[BrowserInstance]: List of all browser instances.
"""
async with self._lock:
return [data['instance'] for data in self._instances.values()]
async def close_instance(self, instance_id: str) -> bool:
"""
Close and remove a browser instance.
Args:
instance_id (str): The ID of the browser instance to close.
Returns:
bool: True if closed successfully, False otherwise.
"""
import asyncio
async def _do_close():
async with self._lock:
if instance_id not in self._instances:
return False
data = self._instances[instance_id]
browser = data['browser']
instance = data['instance']
try:
if hasattr(browser, 'tabs') and browser.tabs:
for tab in browser.tabs[:]:
try:
await tab.close()
except Exception:
pass
except Exception:
pass
try:
import asyncio
if hasattr(browser, 'connection') and browser.connection:
asyncio.get_event_loop().create_task(browser.connection.disconnect())
debug_logger.log_info("browser_manager", "close_connection", "closed connection using get_event_loop().create_task()")
except RuntimeError:
try:
import asyncio
if hasattr(browser, 'connection') and browser.connection:
await asyncio.wait_for(browser.connection.disconnect(), timeout=2.0)
debug_logger.log_info("browser_manager", "close_connection", "closed connection with direct await and timeout")
except (asyncio.TimeoutError, Exception) as e:
debug_logger.log_info("browser_manager", "close_connection", f"connection disconnect failed or timed out: {e}")
pass
except Exception as e:
debug_logger.log_info("browser_manager", "close_connection", f"connection disconnect failed: {e}")
pass
try:
import nodriver.cdp.browser as cdp_browser
if hasattr(browser, 'connection') and browser.connection:
await browser.connection.send(cdp_browser.close())
except Exception:
pass
try:
process_cleanup.kill_browser_process(instance_id)
except Exception as e:
debug_logger.log_warning("browser_manager", "close_instance",
f"Process cleanup failed for {instance_id}: {e}")
try:
await browser.stop()
except Exception:
pass
if hasattr(browser, '_process') and browser._process and browser._process.returncode is None:
import os
for attempt in range(3):
try:
browser._process.terminate()
debug_logger.log_info("browser_manager", "terminate_process", f"terminated browser with pid {browser._process.pid} successfully on attempt {attempt + 1}")
break
except Exception:
try:
browser._process.kill()
debug_logger.log_info("browser_manager", "kill_process", f"killed browser with pid {browser._process.pid} successfully on attempt {attempt + 1}")
break
except Exception:
try:
if hasattr(browser, '_process_pid') and browser._process_pid:
os.kill(browser._process_pid, 15)
debug_logger.log_info("browser_manager", "kill_process", f"killed browser with pid {browser._process_pid} using signal 15 successfully on attempt {attempt + 1}")
break
except (PermissionError, ProcessLookupError) as e:
debug_logger.log_info("browser_manager", "kill_process", f"browser already stopped or no permission to kill: {e}")
break
except Exception as e:
if attempt == 2:
debug_logger.log_error("browser_manager", "kill_process", e)
try:
if hasattr(browser, '_process'):
browser._process = None
if hasattr(browser, '_process_pid'):
browser._process_pid = None
instance.state = BrowserState.CLOSED
except Exception:
pass
del self._instances[instance_id]
persistent_storage.remove_instance(instance_id)
return True
try:
return await asyncio.wait_for(_do_close(), timeout=5.0)
except asyncio.TimeoutError:
debug_logger.log_info("browser_manager", "close_instance", f"Close timeout for {instance_id}, forcing cleanup")
try:
async with self._lock:
if instance_id in self._instances:
data = self._instances[instance_id]
data['instance'].state = BrowserState.CLOSED
del self._instances[instance_id]
persistent_storage.remove_instance(instance_id)
except Exception:
pass
return True
except Exception as e:
debug_logger.log_error("browser_manager", "close_instance", e)
return False
async def get_tab(self, instance_id: str) -> Optional[Tab]:
"""
Get the main tab for a browser instance.
Args:
instance_id (str): The ID of the browser instance.
Returns:
Optional[Tab]: The main tab if found, else None.
"""
data = await self.get_instance(instance_id)
if data:
return data['tab']
return None
async def get_browser(self, instance_id: str) -> Optional[Browser]:
"""
Get the browser object for an instance.
Args:
instance_id (str): The ID of the browser instance.
Returns:
Optional[Browser]: The browser object if found, else None.
"""
data = await self.get_instance(instance_id)
if data:
return data['browser']
return None
async def list_tabs(self, instance_id: str) -> List[Dict[str, str]]:
"""
List all tabs for a browser instance.
Args:
instance_id (str): The ID of the browser instance.
Returns:
List[Dict[str, str]]: List of tab information dictionaries.
"""
browser = await self.get_browser(instance_id)
if not browser:
return []
await browser.update_targets()
tabs = []
for tab in browser.tabs:
await tab
tabs.append({
'tab_id': str(tab.target.target_id),
'url': getattr(tab, 'url', '') or '',
'title': getattr(tab.target, 'title', '') or 'Untitled',
'type': getattr(tab.target, 'type_', 'page')
})
return tabs
async def switch_to_tab(self, instance_id: str, tab_id: str) -> bool:
"""
Switch to a specific tab by bringing it to front.
Args:
instance_id (str): The ID of the browser instance.
tab_id (str): The target ID of the tab to switch to.
Returns:
bool: True if switched successfully, False otherwise.
"""
browser = await self.get_browser(instance_id)
if not browser:
return False
await browser.update_targets()
target_tab = None
for tab in browser.tabs:
if str(tab.target.target_id) == tab_id:
target_tab = tab
break
if not target_tab:
return False
try:
await target_tab.bring_to_front()
async with self._lock:
if instance_id in self._instances:
self._instances[instance_id]['tab'] = target_tab
return True
except Exception:
return False
async def get_active_tab(self, instance_id: str) -> Optional[Tab]:
"""
Get the currently active tab.
Args:
instance_id (str): The ID of the browser instance.
Returns:
Optional[Tab]: The active tab if found, else None.
"""
return await self.get_tab(instance_id)
async def close_tab(self, instance_id: str, tab_id: str) -> bool:
"""
Close a specific tab.
Args:
instance_id (str): The ID of the browser instance.
tab_id (str): The target ID of the tab to close.
Returns:
bool: True if closed successfully, False otherwise.
"""
browser = await self.get_browser(instance_id)
if not browser:
return False
target_tab = None
for tab in browser.tabs:
if str(tab.target.target_id) == tab_id:
target_tab = tab
break
if not target_tab:
return False
try:
await target_tab.close()
return True
except Exception:
return False
async def update_instance_state(self, instance_id: str, url: str = None, title: str = None):
"""
Update instance state after navigation or action.
Args:
instance_id (str): The ID of the browser instance.
url (str, optional): The current URL to update.
title (str, optional): The title to update.
"""
async with self._lock:
if instance_id in self._instances:
instance = self._instances[instance_id]['instance']
if url:
instance.current_url = url
if title:
instance.title = title
instance.update_activity()
async def get_page_state(self, instance_id: str) -> Optional[PageState]:
"""
Get complete page state for an instance.
Args:
instance_id (str): The ID of the browser instance.
Returns:
Optional[PageState]: The page state if available, else None.
"""
tab = await self.get_tab(instance_id)
if not tab:
return None
try:
url = await tab.evaluate("window.location.href")
title = await tab.evaluate("document.title")
ready_state = await tab.evaluate("document.readyState")
cookies = await tab.send(uc.cdp.network.get_cookies())
local_storage = {}
session_storage = {}
try:
local_storage_keys = await tab.evaluate("Object.keys(localStorage)")
for key in local_storage_keys:
value = await tab.evaluate(f"localStorage.getItem('{key}')")
local_storage[key] = value
session_storage_keys = await tab.evaluate("Object.keys(sessionStorage)")
for key in session_storage_keys:
value = await tab.evaluate(f"sessionStorage.getItem('{key}')")
session_storage[key] = value
except Exception:
pass
viewport = await tab.evaluate("""
({
width: window.innerWidth,
height: window.innerHeight,
devicePixelRatio: window.devicePixelRatio
})
""")
return PageState(
instance_id=instance_id,
url=url,
title=title,
ready_state=ready_state,
cookies=cookies.get('cookies', []),
local_storage=local_storage,
session_storage=session_storage,
viewport=viewport
)
except Exception as e:
raise Exception(f"Failed to get page state: {str(e)}")
async def cleanup_inactive(self, timeout_minutes: int = 30):
"""
Clean up inactive browser instances.
Args:
timeout_minutes (int, optional): Timeout in minutes to consider an instance inactive. Defaults to 30.
"""
now = datetime.now()
timeout = timedelta(minutes=timeout_minutes)
to_close = []
async with self._lock:
for instance_id, data in self._instances.items():
instance = data['instance']
if now - instance.last_activity > timeout:
to_close.append(instance_id)
for instance_id in to_close:
await self.close_instance(instance_id)
async def close_all(self):
"""
Close all browser instances.
Closes all currently managed browser instances.
"""
instance_ids = list(self._instances.keys())
for instance_id in instance_ids:
await self.close_instance(instance_id)
```
--------------------------------------------------------------------------------
/demo/augment-hero-recreation.html:
--------------------------------------------------------------------------------
```html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Augment Code Hero Recreation</title>
<style>
@import url('https://fonts.googleapis.com/css2?family=Inter:wght@300;400;500;600;700;800;900&display=swap');
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: "Inter", system-ui, -apple-system, "Segoe UI", "Roboto", sans-serif;
color: #fafaf9;
background: #000000;
line-height: 1.5;
font-size: 16px;
font-weight: 400;
-webkit-font-smoothing: antialiased;
-moz-osx-font-smoothing: grayscale;
}
/* Navigation Bar */
.navbar {
position: fixed;
top: 0;
left: 0;
right: 0;
z-index: 1000;
padding: 1rem 1.5rem;
background: rgba(0, 0, 0, 0.8);
backdrop-filter: blur(20px);
border-bottom: 1px solid rgba(255, 255, 255, 0.05);
}
.nav-container {
max-width: 1200px;
margin: 0 auto;
display: flex;
align-items: center;
justify-content: space-between;
}
.logo {
display: flex;
align-items: center;
gap: 0.5rem;
font-size: 1.1rem;
font-weight: 600;
color: #fafaf9;
text-decoration: none;
}
.nav-links {
display: none;
gap: 2rem;
list-style: none;
}
@media (min-width: 768px) {
.nav-links {
display: flex;
}
}
.nav-links a {
color: #a1a1aa;
text-decoration: none;
font-weight: 500;
transition: color 0.3s ease;
}
.nav-links a:hover {
color: #fafaf9;
}
.nav-buttons {
display: flex;
gap: 0.5rem;
}
.nav-btn {
padding: 0.5rem 1rem;
border-radius: 0.375rem;
font-weight: 500;
font-size: 0.875rem;
text-decoration: none;
transition: all 0.3s ease;
}
.nav-btn.secondary {
color: #fafaf9;
background: transparent;
border: 1px solid rgba(255, 255, 255, 0.1);
}
.nav-btn.secondary:hover {
background: rgba(255, 255, 255, 0.05);
}
.nav-btn.primary {
color: #000;
background: #fafaf9;
border: 1px solid #fafaf9;
}
.nav-btn.primary:hover {
background: #f4f4f5;
}
/* Hero Section */
.hero-section {
position: relative;
min-height: 100vh;
overflow: hidden;
padding: 0 1rem;
background:
radial-gradient(ellipse 50% 80% at 20% 40%, rgba(120, 119, 198, 0.3), transparent),
radial-gradient(ellipse 50% 80% at 80% 50%, rgba(120, 119, 198, 0.15), transparent),
radial-gradient(ellipse 50% 80% at 40% 80%, rgba(120, 119, 198, 0.1), transparent),
#000000;
display: flex;
align-items: center;
}
.hero-container {
position: relative;
z-index: 10;
margin: 0 auto;
display: flex;
max-width: 1200px;
width: 100%;
flex-direction: column;
align-items: center;
justify-content: center;
text-align: center;
gap: 3rem;
padding: 6rem 0 4rem 0;
}
/* Announcement Banner */
.announcement {
animation: slideInFromTop 0.8s ease-out;
}
.announcement a {
text-decoration: none;
color: inherit;
}
.announcement-banner {
display: inline-flex;
align-items: center;
gap: 0.75rem;
padding: 0.5rem 1.25rem;
border-radius: 50px;
border: 1px solid rgba(255, 255, 255, 0.08);
background: rgba(0, 0, 0, 0.4);
backdrop-filter: blur(10px);
font-size: 0.875rem;
font-weight: 500;
letter-spacing: 0.5px;
text-transform: uppercase;
transition: all 0.3s ease;
cursor: pointer;
}
.announcement-banner:hover {
background: rgba(255, 255, 255, 0.05);
border-color: rgba(255, 255, 255, 0.15);
transform: translateY(-1px);
}
/* Main Headlines */
.main-headlines {
animation: slideInFromBottom 0.8s ease-out 0.2s both;
}
.headline-large {
font-size: clamp(2.5rem, 8vw, 6rem);
font-weight: 800;
line-height: 1.1;
letter-spacing: -0.02em;
margin-bottom: 0.5rem;
background: linear-gradient(135deg, #fafaf9 0%, #d4d4d8 100%);
background-clip: text;
-webkit-background-clip: text;
-webkit-text-fill-color: transparent;
text-align: center;
}
@media (min-width: 640px) {
.headline-large {
font-size: clamp(3rem, 10vw, 7rem);
}
}
/* Subtitle */
.subtitle {
max-width: 42rem;
font-size: 1.25rem;
line-height: 1.6;
color: #a1a1aa;
font-weight: 400;
margin: 0 auto;
animation: slideInFromBottom 0.8s ease-out 0.4s both;
}
@media (min-width: 768px) {
.subtitle {
font-size: 1.375rem;
line-height: 1.7;
}
}
/* CTA Button */
.cta-section {
animation: slideInFromBottom 0.8s ease-out 0.6s both;
}
.install-button {
display: inline-block;
text-decoration: none;
background: linear-gradient(135deg, rgba(120, 119, 198, 0.15) 0%, rgba(120, 119, 198, 0.05) 100%);
border: 1px solid rgba(120, 119, 198, 0.2);
border-radius: 8px;
padding: 1px;
transition: all 0.3s ease;
position: relative;
overflow: hidden;
}
.install-button::before {
content: '';
position: absolute;
inset: 0;
background: linear-gradient(135deg, rgba(120, 119, 198, 0.1) 0%, rgba(120, 119, 198, 0.02) 100%);
opacity: 0;
transition: opacity 0.3s ease;
border-radius: 7px;
}
.install-button:hover::before {
opacity: 1;
}
.install-button:hover {
border-color: rgba(120, 119, 198, 0.3);
transform: translateY(-2px);
box-shadow: 0 20px 40px rgba(120, 119, 198, 0.1);
}
.button-content {
position: relative;
z-index: 1;
display: flex;
align-items: center;
gap: 1.5rem;
padding: 1rem 1.5rem;
background: rgba(0, 0, 0, 0.6);
border-radius: 7px;
backdrop-filter: blur(10px);
}
.button-text {
font-size: 1.25rem;
font-weight: 600;
color: #fafaf9;
}
.button-divider {
width: 1px;
height: 2rem;
background: rgba(255, 255, 255, 0.1);
}
.button-icons {
display: flex;
gap: 1rem;
}
.icon-wrapper {
width: 2.5rem;
height: 2.5rem;
display: flex;
align-items: center;
justify-content: center;
border-radius: 6px;
background: rgba(255, 255, 255, 0.05);
transition: all 0.3s ease;
cursor: pointer;
}
.icon-wrapper:hover {
background: rgba(255, 255, 255, 0.1);
transform: scale(1.05);
}
.icon-wrapper svg {
width: 1.5rem;
height: 1.5rem;
}
/* Video Container */
.video-showcase {
position: relative;
max-width: 900px;
width: 100%;
animation: slideInFromBottom 0.8s ease-out 0.8s both;
}
.video-frame {
position: relative;
background: linear-gradient(135deg, rgba(120, 119, 198, 0.1) 0%, rgba(120, 119, 198, 0.02) 100%);
border: 1px solid rgba(255, 255, 255, 0.08);
border-radius: 16px;
padding: 1rem;
backdrop-filter: blur(20px);
}
.video-container {
position: relative;
overflow: hidden;
border-radius: 12px;
background: #000;
box-shadow:
0 25px 50px rgba(0, 0, 0, 0.5),
0 0 0 1px rgba(255, 255, 255, 0.05);
}
.hero-video {
width: 100%;
height: auto;
display: block;
}
/* Animations */
@keyframes slideInFromTop {
from {
opacity: 0;
transform: translateY(-30px);
}
to {
opacity: 1;
transform: translateY(0);
}
}
@keyframes slideInFromBottom {
from {
opacity: 0;
transform: translateY(30px);
}
to {
opacity: 1;
transform: translateY(0);
}
}
/* Floating Elements (Desktop Only) */
.floating-ui {
position: absolute;
pointer-events: none;
opacity: 0;
transition: all 0.8s ease;
}
@media (min-width: 1200px) {
.video-showcase {
animation: slideInFromBottom 0.8s ease-out 0.8s both, floatIn 1s ease-out 1.6s both;
}
.floating-ui {
opacity: 0.8;
}
}
.floating-terminal {
top: -10rem;
right: -15rem;
z-index: 5;
}
.floating-panel {
bottom: -10rem;
left: -15rem;
z-index: 5;
}
.floating-ide {
top: 50%;
right: -18rem;
transform: translateY(-50%);
z-index: 5;
}
.floating-element {
width: 200px;
height: 300px;
background: linear-gradient(135deg, rgba(120, 119, 198, 0.05) 0%, rgba(120, 119, 198, 0.01) 100%);
border: 1px solid rgba(255, 255, 255, 0.05);
border-radius: 12px;
padding: 0.5rem;
backdrop-filter: blur(20px);
}
.floating-element img {
width: 100%;
height: 100%;
object-fit: cover;
border-radius: 8px;
opacity: 0.9;
}
@keyframes floatIn {
from {
opacity: 0;
transform: scale(0.8);
}
to {
opacity: 0.8;
transform: scale(1);
}
}
/* Responsive Adjustments */
@media (max-width: 768px) {
.hero-container {
gap: 2rem;
padding: 4rem 0 2rem 0;
}
.button-content {
flex-direction: column;
gap: 1rem;
text-align: center;
}
.button-divider {
display: none;
}
.button-text {
font-size: 1.125rem;
}
.subtitle {
font-size: 1.125rem;
}
}
</style>
</head>
<body>
<!-- Navigation -->
<nav class="navbar">
<div class="nav-container">
<a href="/" class="logo">
<svg width="20" height="20" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<rect x="3" y="3" width="6" height="6" fill="currentColor"/>
<rect x="15" y="3" width="6" height="6" fill="currentColor"/>
<rect x="3" y="15" width="6" height="6" fill="currentColor"/>
<rect x="15" y="15" width="6" height="6" fill="currentColor"/>
</svg>
augment code
</a>
<ul class="nav-links">
<li><a href="/product">Product</a></li>
<li><a href="/pricing">Pricing</a></li>
<li><a href="/docs">Docs</a></li>
<li><a href="/blog">Blog</a></li>
</ul>
<div class="nav-buttons">
<a href="/signin" class="nav-btn secondary">Sign in</a>
<a href="/install" class="nav-btn primary">Install</a>
</div>
</div>
</nav>
<!-- Hero Section -->
<section class="hero-section">
<div class="hero-container">
<!-- Announcement -->
<div class="announcement">
<a href="/blog/gpt-5-is-here-and-we-now-have-a-model-picker">
<div class="announcement-banner">
<span>Now supporting GPT-5 and Sonnet 4</span>
<span>→</span>
</div>
</a>
</div>
<!-- Main Headlines -->
<div class="main-headlines">
<h1 class="headline-large">Better Context. Better Agent.</h1>
<h1 class="headline-large">Better Code.</h1>
</div>
<!-- Subtitle -->
<p class="subtitle">
The most powerful AI software development platform backed by the industry-leading context engine.
</p>
<!-- CTA Button -->
<div class="cta-section">
<a href="/signup" class="install-button">
<div class="button-content">
<span class="button-text">Install now</span>
<div class="button-divider"></div>
<div class="button-icons">
<div class="icon-wrapper">
<svg viewBox="0 0 50 48" fill="none" xmlns="http://www.w3.org/2000/svg">
<path d="M2.355 17.08C2.355 17.08 1.2012 16.2498 2.58576 15.1412L5.8116 12.2617C5.8116 12.2617 6.73465 11.2922 7.71057 12.1369L37.4787 34.6354V45.4239C37.4787 45.4239 37.4643 47.118 35.2865 46.9309L2.355 17.08Z" fill="#2489CA"/>
<path d="M10.0252 24.0346L2.35237 30.9982C2.35237 30.9982 1.56394 31.5837 2.35237 32.6299L5.91473 35.8646C5.91473 35.8646 6.76086 36.7716 8.01081 35.7398L16.1451 29.5824L10.0252 24.0346Z" fill="#1070B3"/>
<path d="M23.4933 24.0917L37.5649 13.3655L37.4735 2.63458C37.4735 2.63458 36.8726 0.292582 34.8678 1.51157L16.1426 18.5246L23.4933 24.0917Z" fill="#0877B9"/>
<path d="M35.2826 46.9455C36.0999 47.7806 37.0902 47.507 37.0902 47.507L48.0561 42.1127C49.4599 41.1577 49.2628 39.9723 49.2628 39.9723V7.76029C49.2628 6.34453 47.811 5.85502 47.811 5.85502L38.3065 1.28141C36.2297 2.88486e-05 34.8691 1.51177 34.8691 1.51177C34.8691 1.51177 36.6191 0.254385 37.4748 2.63477V45.2274C37.4748 45.5202 37.4123 45.8081 37.2873 46.0673C37.0373 46.5712 36.4941 47.0415 35.1912 46.8447L35.2826 46.9455Z" fill="#3C99D4"/>
</svg>
</div>
<div class="icon-wrapper">
<svg viewBox="0 0 64 64">
<defs>
<linearGradient id="jetbrains-gradient" x1=".8" x2="62.6" y1="3.3" y2="64.2" gradientTransform="matrix(1 0 0 -1 0 66)" gradientUnits="userSpaceOnUse">
<stop offset="0" stop-color="#ff9419"/>
<stop offset=".4" stop-color="#ff021d"/>
<stop offset="1" stop-color="#e600ff"/>
</linearGradient>
</defs>
<path d="M20.3 3.7 3.7 20.3C1.4 22.6 0 25.8 0 29.1v29.8c0 2.8 2.2 5 5 5h29.8c3.3 0 6.5-1.3 8.8-3.7l16.7-16.7c2.3-2.3 3.7-5.5 3.7-8.8V5c0-2.8-2.2-5-5-5H29.2c-3.3 0-6.5 1.3-8.8 3.7Z" fill="url(#jetbrains-gradient)"/>
<path d="M48 16H8v40h40V16Z" fill="#000"/>
<path d="M30 47H13v4h17v-4Z" fill="#fff"/>
</svg>
</div>
</div>
</div>
</a>
</div>
<!-- Video Showcase -->
<div class="video-showcase">
<div class="video-frame">
<div class="video-container">
<video class="hero-video" autoplay loop muted playsinline poster="data:image/svg+xml;base64,PHN2ZyB3aWR0aD0iODAwIiBoZWlnaHQ9IjQ1MCIgeG1sbnM9Imh0dHA6Ly93d3cudzMub3JnLzIwMDAvc3ZnIj48cmVjdCB3aWR0aD0iMTAwJSIgaGVpZ2h0PSIxMDAlIiBmaWxsPSIjMDAwIi8+PHRleHQgeD0iNTAlIiB5PSI1MCUiIGZvbnQtZmFtaWx5PSJBcmlhbCwgc2Fucy1zZXJpZiIgZm9udC1zaXplPSIxOCIgZmlsbD0iIzk5OSIgdGV4dC1hbmNob3I9Im1pZGRsZSIgZHk9Ii4zZW0iPkF1Z21lbnQgQ29kZSBEZW1vPC90ZXh0Pjwvc3ZnPg==">
<source src="https://augment-assets.com/video.hevc.mp4" type="video/mp4; codecs=hvc1">
<source src="https://augment-assets.com/video.h264.mp4" type="video/mp4; codecs=avc1.4D401E">
</video>
</div>
</div>
<!-- Floating UI Elements -->
<div class="floating-ui floating-terminal">
<div class="floating-element">
<div style="width: 100%; height: 100%; background: linear-gradient(135deg, #1a1a1a 0%, #000 100%); border-radius: 8px; display: flex; align-items: center; justify-content: center; color: #00ff88; font-family: monospace; font-size: 0.75rem;">Terminal</div>
</div>
</div>
<div class="floating-ui floating-panel">
<div class="floating-element">
<div style="width: 100%; height: 100%; background: linear-gradient(135deg, #2d2d2d 0%, #1a1a1a 100%); border-radius: 8px; display: flex; align-items: center; justify-content: center; color: #60a5fa; font-family: sans-serif; font-size: 0.75rem;">Augment Panel</div>
</div>
</div>
<div class="floating-ui floating-ide">
<div class="floating-element">
<div style="width: 100%; height: 100%; background: linear-gradient(135deg, #ff6b35 0%, #f7931e 100%); border-radius: 8px; display: flex; align-items: center; justify-content: center; color: #fff; font-family: sans-serif; font-size: 0.75rem;">IntelliJ IDE</div>
</div>
</div>
</div>
</div>
</section>
</body>
</html>
```
--------------------------------------------------------------------------------
/src/hook_learning_system.py:
--------------------------------------------------------------------------------
```python
"""
Hook Learning System - AI Training and Examples
This system provides examples, documentation, and learning materials for AI
to understand how to create effective hook functions.
"""
from typing import Dict, List, Any
import ast
class HookLearningSystem:
"""System to help AI learn how to create hook functions."""
@staticmethod
def get_request_object_documentation() -> Dict[str, Any]:
"""Get comprehensive documentation of the request object structure."""
return {
"request_object": {
"description": "The request object passed to hook functions",
"type": "dict",
"fields": {
"request_id": {
"type": "str",
"description": "Unique identifier for this request",
"example": "fetch-12345-abcde"
},
"instance_id": {
"type": "str",
"description": "Browser instance ID that made the request",
"example": "8e226b0c-3879-4d5e-96b3-db1805bfd4c4"
},
"url": {
"type": "str",
"description": "Full URL of the request",
"example": "https://example.com/api/data?param=value"
},
"method": {
"type": "str",
"description": "HTTP method (GET, POST, PUT, DELETE, etc.)",
"example": "GET"
},
"headers": {
"type": "dict[str, str]",
"description": "Request headers as key-value pairs",
"example": {
"User-Agent": "Mozilla/5.0...",
"Accept": "application/json",
"Authorization": "Bearer token123"
}
},
"post_data": {
"type": "str or None",
"description": "POST/PUT body data (None for GET requests)",
"example": '{"username": "user", "password": "pass"}'
},
"resource_type": {
"type": "str or None",
"description": "Type of resource (Document, Script, Image, XHR, etc.)",
"example": "Document"
},
"stage": {
"type": "str",
"description": "Request stage (request or response)",
"example": "request"
}
}
},
"hook_action": {
"description": "Return value from hook functions",
"type": "HookAction or dict",
"actions": {
"continue": {
"description": "Allow request to proceed normally",
"example": 'HookAction(action="continue")'
},
"block": {
"description": "Block the request entirely",
"example": 'HookAction(action="block")'
},
"redirect": {
"description": "Redirect request to a different URL",
"fields": ["url"],
"example": 'HookAction(action="redirect", url="https://httpbin.org/get")'
},
"modify": {
"description": "Modify request parameters",
"fields": ["url", "method", "headers", "post_data"],
"example": 'HookAction(action="modify", headers={"X-Custom": "value"})'
},
"fulfill": {
"description": "Return custom response without sending request",
"fields": ["status_code", "headers", "body"],
"example": 'HookAction(action="fulfill", status_code=200, body="Custom response")'
}
}
}
}
@staticmethod
def get_hook_examples() -> List[Dict[str, Any]]:
"""Get example hook functions for AI learning."""
return [
{
"name": "Simple URL Blocker",
"description": "Block all requests to doubleclick.net (ad blocker)",
"requirements": {
"url_pattern": "*doubleclick.net*"
},
"function": '''
def process_request(request):
# Block any request to doubleclick.net
return HookAction(action="block")
''',
"explanation": "This hook blocks all requests matching the URL pattern. No conditions needed since we always want to block ads."
},
{
"name": "Simple Redirect",
"description": "Redirect example.com to httpbin.org for testing",
"requirements": {
"url_pattern": "*example.com*"
},
"function": '''
def process_request(request):
# Redirect to httpbin for testing
return HookAction(action="redirect", url="https://httpbin.org/get")
''',
"explanation": "This hook redirects any request to example.com to httpbin.org for testing purposes."
},
{
"name": "Header Modifier",
"description": "Add custom headers to API requests",
"requirements": {
"url_pattern": "*/api/*"
},
"function": '''
def process_request(request):
# Add API key header to all API requests
new_headers = request["headers"].copy()
new_headers["X-API-Key"] = "secret-api-key-123"
new_headers["X-Custom-Client"] = "Browser-Hook-System"
return HookAction(
action="modify",
headers=new_headers
)
''',
"explanation": "This hook adds custom headers to API requests. It copies existing headers and adds new ones."
},
{
"name": "Method Converter",
"description": "Convert GET requests to POST for specific endpoints",
"requirements": {
"url_pattern": "*/convert-to-post*",
"method": "GET"
},
"function": '''
def process_request(request):
# Convert GET to POST and add JSON body
return HookAction(
action="modify",
method="POST",
headers={
**request["headers"],
"Content-Type": "application/json"
},
post_data='{"converted": true, "original_url": "' + request["url"] + '"}'
)
''',
"explanation": "This hook converts GET requests to POST, adds JSON content-type header, and includes original URL in body."
},
{
"name": "Custom Response Generator",
"description": "Return custom JSON response for API endpoints",
"requirements": {
"url_pattern": "*/mock-api/*"
},
"function": '''
def process_request(request):
# Return mock API response
mock_data = {
"status": "success",
"data": {
"message": "This is a mocked response",
"request_url": request["url"],
"timestamp": datetime.now().isoformat()
}
}
return HookAction(
action="fulfill",
status_code=200,
headers={"Content-Type": "application/json"},
body=str(mock_data).replace("'", '"') # Convert to JSON string
)
''',
"explanation": "This hook intercepts API requests and returns custom JSON responses without hitting the real server."
},
{
"name": "Conditional Blocker",
"description": "Block requests based on multiple conditions",
"requirements": {
"url_pattern": "*" # Match all URLs
},
"function": '''
def process_request(request):
# Block requests to social media trackers during work hours
social_trackers = ["facebook.com", "twitter.com", "linkedin.com", "instagram.com"]
# Check if URL contains social tracker
is_social_tracker = any(tracker in request["url"] for tracker in social_trackers)
# Check if it's tracking related
is_tracker = "/track" in request["url"] or "/analytics" in request["url"]
if is_social_tracker and is_tracker:
return HookAction(action="block")
# Otherwise continue normally
return HookAction(action="continue")
''',
"explanation": "This hook uses conditional logic to block social media trackers based on URL patterns and content."
},
{
"name": "Dynamic URL Rewriter",
"description": "Rewrite URLs based on patterns and parameters",
"requirements": {
"url_pattern": "*old-domain.com*"
},
"function": '''
def process_request(request):
original_url = request["url"]
# Replace domain but keep path and parameters
new_url = original_url.replace("old-domain.com", "new-domain.com")
# Add cache-busting parameter
separator = "&" if "?" in new_url else "?"
new_url += f"{separator}cache_bust=hook_modified"
return HookAction(action="redirect", url=new_url)
''',
"explanation": "This hook rewrites URLs by replacing domains and adding parameters, useful for domain migrations."
},
{
"name": "Request Logger",
"description": "Log specific requests without modifying them",
"requirements": {
"url_pattern": "*important-api*"
},
"function": '''
def process_request(request):
# Log important API calls for debugging
print(f"[API LOG] {request['method']} {request['url']}")
# Log headers if they contain auth info
if "authorization" in str(request["headers"]).lower():
print(f"[API LOG] Has Authorization header")
# Always continue the request
return HookAction(action="continue")
''',
"explanation": "This hook logs request details for debugging/monitoring purposes but doesn't modify the request."
},
{
"name": "Security Header Injector",
"description": "Add security headers to outgoing requests",
"requirements": {
"url_pattern": "*",
"custom_condition": "request['method'] in ['POST', 'PUT', 'PATCH']"
},
"function": '''
def process_request(request):
# Add security headers to modification requests
security_headers = request["headers"].copy()
security_headers.update({
"X-Requested-With": "XMLHttpRequest",
"X-CSRF-Protection": "enabled",
"X-Custom-Security": "browser-hook-system"
})
return HookAction(
action="modify",
headers=security_headers
)
''',
"explanation": "This hook adds security headers to POST/PUT/PATCH requests using custom conditions in requirements."
},
{
"name": "Response Time Simulator",
"description": "Add artificial delays by fulfilling with delayed responses",
"requirements": {
"url_pattern": "*slow-api*"
},
"function": '''
def process_request(request):
# Simulate slow API by returning custom response immediately
# (In real implementation, you'd add actual delays)
return HookAction(
action="fulfill",
status_code=200,
headers={"Content-Type": "application/json"},
body='{"message": "Simulated slow response", "delay": "3000ms"}'
)
''',
"explanation": "This hook simulates slow APIs by immediately returning responses instead of waiting for real server."
},
{
"name": "Response Content Modifier",
"description": "Modify response content at response stage",
"requirements": {
"url_pattern": "*api/*",
"stage": "response"
},
"function": '''
def process_request(request):
# Only process responses (not requests)
if request.get("stage") != "response":
return HookAction(action="continue")
# Get response body
response_body = request.get("response_body", "")
if "user_data" in response_body:
# Replace sensitive data in API responses
modified_body = response_body.replace(
'"email":', '"email_redacted":'
).replace(
'"phone":', '"phone_redacted":'
)
return HookAction(
action="fulfill",
status_code=200,
headers={"Content-Type": "application/json"},
body=modified_body
)
# Continue normally if no modification needed
return HookAction(action="continue")
''',
"explanation": "This response-stage hook modifies API response content to redact sensitive user data."
},
{
"name": "Response Header Injector",
"description": "Add security headers to responses at response stage",
"requirements": {
"url_pattern": "*",
"stage": "response"
},
"function": '''
def process_request(request):
# Only process responses
if request.get("stage") != "response":
return HookAction(action="continue")
# Add security headers to all responses
security_headers = {
"X-Content-Type-Options": "nosniff",
"X-Frame-Options": "DENY",
"X-XSS-Protection": "1; mode=block",
"Strict-Transport-Security": "max-age=31536000"
}
# Merge with existing headers
current_headers = request.get("response_headers", {})
merged_headers = {**current_headers, **security_headers}
return HookAction(
action="modify",
headers=merged_headers
)
''',
"explanation": "This response-stage hook adds security headers to all responses for better protection."
},
{
"name": "API Response Faker",
"description": "Replace API responses with fake data for testing",
"requirements": {
"url_pattern": "*api/users*",
"stage": "response"
},
"function": '''
def process_request(request):
# Only process responses
if request.get("stage") != "response":
return HookAction(action="continue")
# Generate fake user data for testing
fake_response = {
"users": [
{"id": 1, "name": "Test User 1", "email": "[email protected]"},
{"id": 2, "name": "Test User 2", "email": "[email protected]"},
{"id": 3, "name": "Test User 3", "email": "[email protected]"}
],
"total": 3,
"fake": True
}
return HookAction(
action="fulfill",
status_code=200,
headers={"Content-Type": "application/json"},
body=str(fake_response).replace("'", '"')
)
''',
"explanation": "This response-stage hook replaces real API responses with fake data for testing environments."
}
]
@staticmethod
def get_requirements_documentation() -> Dict[str, Any]:
"""Get documentation on hook requirements/matching criteria."""
return {
"requirements": {
"description": "Criteria that determine when a hook should trigger",
"fields": {
"url_pattern": {
"type": "str",
"description": "Wildcard pattern to match URLs (* = any characters, ? = single character)",
"examples": [
"*example.com*", # Any URL containing example.com
"https://api.*.com/*", # Any subdomain of .com domains
"*api/v*/users*", # API versioned endpoints
"*.jpg", # Image files
"*doubleclick*" # Ad networks
]
},
"method": {
"type": "str",
"description": "HTTP method to match (GET, POST, PUT, DELETE, etc.)",
"examples": ["GET", "POST", "PUT", "DELETE"]
},
"resource_type": {
"type": "str",
"description": "Type of resource to match",
"examples": ["Document", "Script", "Image", "XHR", "Fetch", "WebSocket"]
},
"stage": {
"type": "str",
"description": "Stage of request processing (request = before sending, response = after receiving headers/body)",
"examples": ["request", "response"],
"note": "Response stage hooks can access response_body, response_status_code, and response_headers"
},
"custom_condition": {
"type": "str",
"description": "Python expression evaluated with 'request' variable",
"examples": [
"len(request['headers']) > 10",
"'json' in request['headers'].get('Content-Type', '')",
"request['method'] in ['POST', 'PUT']",
"'auth' in request['url'].lower()"
]
}
}
},
"best_practices": [
"Use specific URL patterns to avoid over-matching",
"Include method filters for POST/PUT hooks to avoid affecting GET requests",
"Use custom conditions for complex matching logic",
"Test hooks with console logging before deploying",
"Always return a HookAction object",
"Handle exceptions gracefully",
"Use priority (lower = higher priority) to control hook execution order"
]
}
@staticmethod
def get_common_patterns() -> List[Dict[str, Any]]:
"""Get common hook patterns and use cases."""
return [
{
"pattern": "Ad Blocker",
"requirements": {"url_pattern": "*ads*|*analytics*|*tracking*"},
"action": "block",
"use_case": "Block advertising and tracking requests"
},
{
"pattern": "API Proxy",
"requirements": {"url_pattern": "*api.old-site.com*"},
"action": "redirect",
"use_case": "Redirect API calls to new endpoints"
},
{
"pattern": "Authentication Injector",
"requirements": {"url_pattern": "*api/*", "method": "GET|POST"},
"action": "modify",
"use_case": "Add authentication headers to API requests"
},
{
"pattern": "Mock Server",
"requirements": {"url_pattern": "*mock/*"},
"action": "fulfill",
"use_case": "Return custom responses for testing"
},
{
"pattern": "Request Logger",
"requirements": {"url_pattern": "*"},
"action": "continue",
"use_case": "Log requests for debugging without modification"
},
{
"pattern": "Security Headers",
"requirements": {"method": "POST|PUT|PATCH"},
"action": "modify",
"use_case": "Add security headers to modification requests"
}
]
@staticmethod
def validate_hook_function(function_code: str) -> Dict[str, Any]:
"""Validate hook function code for common issues."""
issues = []
warnings = []
try:
# Parse the function code
parsed = ast.parse(function_code)
# Check for required function
has_process_request = False
for node in ast.walk(parsed):
if isinstance(node, ast.FunctionDef) and node.name == "process_request":
has_process_request = True
# Check function parameters
if len(node.args.args) != 1:
issues.append("process_request function must take exactly one parameter (request)")
elif node.args.args[0].arg != "request":
warnings.append("First parameter should be named 'request' for clarity")
if not has_process_request:
issues.append("Function must define 'process_request(request)' function")
# Check for dangerous operations
dangerous_nodes = []
for node in ast.walk(parsed):
if isinstance(node, ast.Import) or isinstance(node, ast.ImportFrom):
warnings.append(f"Imports may not work in hook context: {ast.dump(node)}")
elif isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
if node.func.id in ['eval', 'exec', 'open', 'input']:
issues.append(f"Dangerous function call: {node.func.id}")
return {
"valid": len(issues) == 0,
"issues": issues,
"warnings": warnings
}
except SyntaxError as e:
return {
"valid": False,
"issues": [f"Syntax error: {e}"],
"warnings": []
}
except Exception as e:
return {
"valid": False,
"issues": [f"Parse error: {e}"],
"warnings": []
}
# Global instance
hook_learning_system = HookLearningSystem()
```
--------------------------------------------------------------------------------
/src/dynamic_hook_system.py:
--------------------------------------------------------------------------------
```python
"""
Dynamic Hook System - AI-Generated Request Hooks
This system allows AI to create custom hook functions that process network requests
in real-time with no pending state. Hooks are Python functions generated by AI
that can modify, block, redirect, or fulfill requests dynamically.
"""
import asyncio
import uuid
import fnmatch
from datetime import datetime
from typing import Dict, List, Any, Callable, Optional, Union
from dataclasses import dataclass, asdict
import nodriver as uc
from debug_logger import debug_logger
import ast
import sys
from io import StringIO
import contextlib
@dataclass
class RequestInfo:
"""Request information passed to hook functions."""
request_id: str
instance_id: str
url: str
method: str
headers: Dict[str, str]
post_data: Optional[str] = None
resource_type: Optional[str] = None
stage: str = "request" # "request" or "response"
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for AI function processing."""
return asdict(self)
@dataclass
class HookAction:
"""Action returned by hook functions."""
action: str # "continue", "block", "redirect", "fulfill", "modify"
url: Optional[str] = None # For redirect/modify
method: Optional[str] = None # For modify
headers: Optional[Dict[str, str]] = None # For modify/fulfill
body: Optional[str] = None # For fulfill
status_code: Optional[int] = None # For fulfill
post_data: Optional[str] = None # For modify
class DynamicHook:
"""A dynamic hook with AI-generated function."""
def __init__(self, hook_id: str, name: str, requirements: Dict[str, Any],
function_code: str, priority: int = 100):
self.hook_id = hook_id
self.name = name
self.requirements = requirements
self.function_code = function_code
self.priority = priority # Lower number = higher priority
self.created_at = datetime.now()
self.trigger_count = 0
self.last_triggered: Optional[datetime] = None
self.status = "active"
self.request_stage = requirements.get('stage', 'request') # 'request' or 'response'
self._compiled_function = self._compile_function()
def _compile_function(self) -> Callable:
"""Compile the AI-generated function."""
try:
namespace = {
'HookAction': HookAction,
'datetime': datetime,
'fnmatch': fnmatch,
'__builtins__': {
'len': len, 'str': str, 'int': int, 'float': float,
'bool': bool, 'dict': dict, 'list': list, 'tuple': tuple,
'print': lambda *args: debug_logger.log_info("hook_function", self.name, " ".join(map(str, args)))
}
}
exec(self.function_code, namespace)
if 'process_request' not in namespace:
raise ValueError("Function must define 'process_request(request)'")
return namespace['process_request']
except Exception as e:
debug_logger.log_error("dynamic_hook", "compile_function", f"Failed to compile function for hook {self.name}: {e}")
return lambda request: HookAction(action="continue")
def matches(self, request: RequestInfo) -> bool:
"""Check if this hook matches the request."""
try:
# Check URL pattern
if 'url_pattern' in self.requirements:
if not fnmatch.fnmatch(request.url, self.requirements['url_pattern']):
return False
# Check method
if 'method' in self.requirements:
if request.method.upper() != self.requirements['method'].upper():
return False
# Check resource type
if 'resource_type' in self.requirements:
if request.resource_type != self.requirements['resource_type']:
return False
# Check stage
if 'stage' in self.requirements:
if request.stage != self.requirements['stage']:
return False
# Check custom conditions (if any)
if 'custom_condition' in self.requirements:
condition_code = self.requirements['custom_condition']
namespace = {'request': request, '__builtins__': {'len': len, 'str': str}}
try:
result = eval(condition_code, namespace)
if not result:
return False
except:
return False
return True
except Exception as e:
debug_logger.log_error("dynamic_hook", "matches", f"Error matching hook {self.name}: {e}")
return False
def process(self, request: RequestInfo) -> HookAction:
"""Execute the hook function."""
try:
self.trigger_count += 1
self.last_triggered = datetime.now()
debug_logger.log_info("dynamic_hook", "process", f"Processing request {request.url} with hook {self.name}")
result = self._compiled_function(request.to_dict())
if isinstance(result, dict):
result = HookAction(**result)
elif not isinstance(result, HookAction):
debug_logger.log_error("dynamic_hook", "process", f"Hook {self.name} returned invalid type: {type(result)}")
return HookAction(action="continue")
debug_logger.log_info("dynamic_hook", "process", f"Hook {self.name} returned action: {result.action}")
return result
except Exception as e:
debug_logger.log_error("dynamic_hook", "process", f"Error executing hook {self.name}: {e}")
return HookAction(action="continue")
class DynamicHookSystem:
"""Real-time dynamic hook processing system."""
def __init__(self):
self.hooks: Dict[str, DynamicHook] = {}
self.instance_hooks: Dict[str, List[str]] = {} # instance_id -> list of hook_ids
self._lock = asyncio.Lock()
async def setup_interception(self, tab, instance_id: str):
"""Set up request and response interception for a browser tab."""
try:
all_hooks = []
instance_hook_ids = self.instance_hooks.get(instance_id, [])
for hook_id in instance_hook_ids:
hook = self.hooks.get(hook_id)
if hook and hook.status == "active":
all_hooks.append(hook)
for hook_id, hook in self.hooks.items():
if hook.status == "active" and hook_id not in instance_hook_ids:
if not hasattr(hook, 'instance_ids') or not hook.instance_ids:
all_hooks.append(hook)
request_patterns = []
response_patterns = []
for hook in all_hooks:
url_pattern = hook.requirements.get('url_pattern', '*')
resource_type = hook.requirements.get('resource_type')
stage = hook.request_stage
if stage == 'response':
pattern = uc.cdp.fetch.RequestPattern(
url_pattern=url_pattern,
resource_type=getattr(uc.cdp.network.ResourceType, resource_type.upper()) if resource_type else None,
request_stage=uc.cdp.fetch.RequestStage.RESPONSE
)
response_patterns.append(pattern)
else:
pattern = uc.cdp.fetch.RequestPattern(
url_pattern=url_pattern,
resource_type=getattr(uc.cdp.network.ResourceType, resource_type.upper()) if resource_type else None,
request_stage=uc.cdp.fetch.RequestStage.REQUEST
)
request_patterns.append(pattern)
all_patterns = request_patterns + response_patterns
if not all_patterns:
all_patterns = [
uc.cdp.fetch.RequestPattern(url_pattern='*', request_stage=uc.cdp.fetch.RequestStage.REQUEST),
uc.cdp.fetch.RequestPattern(url_pattern='*', request_stage=uc.cdp.fetch.RequestStage.RESPONSE)
]
await tab.send(uc.cdp.fetch.enable(patterns=all_patterns))
tab.add_handler(
uc.cdp.fetch.RequestPaused,
lambda event: asyncio.create_task(self._on_request_paused(tab, event, instance_id))
)
debug_logger.log_info("dynamic_hook_system", "setup_interception", f"Set up interception for instance {instance_id} with {len(all_patterns)} patterns ({len(request_patterns)} request, {len(response_patterns)} response)")
except Exception as e:
debug_logger.log_error("dynamic_hook_system", "setup_interception", f"Failed to setup interception: {e}")
async def _on_request_paused(self, tab, event, instance_id: str):
"""Handle intercepted requests and responses - process hooks immediately."""
try:
# Determine if this is request stage or response stage
# According to nodriver docs: "The stage of the request can be determined by presence of responseErrorReason
# and responseStatusCode -- the request is at the response stage if either of these fields is present"
is_response_stage = (hasattr(event, 'response_status_code') and event.response_status_code is not None) or \
(hasattr(event, 'response_error_reason') and event.response_error_reason is not None)
stage = "response" if is_response_stage else "request"
request = RequestInfo(
request_id=str(event.request_id),
instance_id=instance_id,
url=event.request.url,
method=event.request.method,
headers=dict(event.request.headers) if hasattr(event.request, 'headers') else {},
post_data=event.request.post_data if hasattr(event.request, 'post_data') else None,
resource_type=str(event.resource_type) if hasattr(event, 'resource_type') else None,
stage=stage
)
debug_logger.log_info("dynamic_hook_system", "_on_request_paused", f"Intercepted {stage}: {request.method} {request.url}")
if is_response_stage and hasattr(event, 'response_status_code'):
debug_logger.log_info("dynamic_hook_system", "_on_request_paused", f"Response status: {event.response_status_code}")
await self._process_request_hooks(tab, request, event)
except Exception as e:
debug_logger.log_error("dynamic_hook_system", "_on_request_paused", f"Error processing {stage if 'stage' in locals() else 'request'}: {e}")
try:
await tab.send(uc.cdp.fetch.continue_request(request_id=event.request_id))
except:
pass
async def _process_request_hooks(self, tab, request: RequestInfo, event=None):
"""Process hooks for a request/response in real-time with priority chain processing."""
try:
instance_hook_ids = self.instance_hooks.get(request.instance_id, [])
matching_hooks = []
for hook_id in instance_hook_ids:
hook = self.hooks.get(hook_id)
if hook and hook.status == "active" and hook.request_stage == request.stage and hook.matches(request):
matching_hooks.append(hook)
matching_hooks.sort(key=lambda h: h.priority)
if not matching_hooks:
debug_logger.log_info("dynamic_hook_system", "_process_request_hooks", f"No matching hooks for {request.stage} stage: {request.url}")
if request.stage == "response":
await tab.send(uc.cdp.fetch.continue_response(request_id=uc.cdp.fetch.RequestId(request.request_id)))
else:
await tab.send(uc.cdp.fetch.continue_request(request_id=uc.cdp.fetch.RequestId(request.request_id)))
return
debug_logger.log_info("dynamic_hook_system", "_process_request_hooks", f"Found {len(matching_hooks)} matching hooks for {request.stage} stage: {request.url}")
response_body = None
if request.stage == "response" and event:
try:
body_result = await tab.send(uc.cdp.fetch.get_response_body(request_id=uc.cdp.fetch.RequestId(request.request_id)))
response_body = body_result[0] # body content
debug_logger.log_info("dynamic_hook_system", "_process_request_hooks", f"Retrieved response body ({len(response_body)} chars)")
except Exception as e:
debug_logger.log_error("dynamic_hook_system", "_process_request_hooks", f"Failed to get response body: {e}")
hook = matching_hooks[0]
request_data = request.to_dict()
if response_body:
request_data['response_body'] = response_body
request_data['response_status_code'] = getattr(event, 'response_status_code', None)
response_headers = {}
if hasattr(event, 'response_headers') and event.response_headers:
try:
if isinstance(event.response_headers, dict):
response_headers = event.response_headers
elif hasattr(event.response_headers, 'items'):
for header in event.response_headers:
if hasattr(header, 'name') and hasattr(header, 'value'):
response_headers[header.name] = header.value
else:
response_headers = {}
except Exception:
response_headers = {}
request_data['response_headers'] = response_headers
action = hook._compiled_function(request_data)
if isinstance(action, dict):
action = HookAction(**action)
hook.trigger_count += 1
hook.last_triggered = datetime.now()
debug_logger.log_info("dynamic_hook_system", "_process_request_hooks", f"Hook {hook.name} returned action: {action.action}")
await self._execute_hook_action(tab, request, action, event if request.stage == "response" else None)
except Exception as e:
debug_logger.log_error("dynamic_hook_system", "_process_request_hooks", f"Error processing hooks: {e}")
try:
if request.stage == "response":
await tab.send(uc.cdp.fetch.continue_response(request_id=uc.cdp.fetch.RequestId(request.request_id)))
else:
await tab.send(uc.cdp.fetch.continue_request(request_id=uc.cdp.fetch.RequestId(request.request_id)))
except:
pass
async def create_hook(self, name: str, requirements: Dict[str, Any], function_code: str,
instance_ids: Optional[List[str]] = None, priority: int = 100) -> str:
"""Create a new dynamic hook."""
try:
hook_id = str(uuid.uuid4())
hook = DynamicHook(hook_id, name, requirements, function_code, priority)
async with self._lock:
self.hooks[hook_id] = hook
if instance_ids:
for instance_id in instance_ids:
if instance_id not in self.instance_hooks:
self.instance_hooks[instance_id] = []
self.instance_hooks[instance_id].append(hook_id)
else:
for instance_id in self.instance_hooks:
self.instance_hooks[instance_id].append(hook_id)
debug_logger.log_info("dynamic_hook_system", "create_hook", f"Created hook {name} with ID {hook_id}")
return hook_id
except Exception as e:
debug_logger.log_error("dynamic_hook_system", "create_hook", f"Failed to create hook {name}: {e}")
raise
def list_hooks(self) -> List[Dict[str, Any]]:
"""List all hooks."""
return [
{
"hook_id": hook.hook_id,
"name": hook.name,
"requirements": hook.requirements,
"priority": hook.priority,
"status": hook.status,
"trigger_count": hook.trigger_count,
"last_triggered": hook.last_triggered.isoformat() if hook.last_triggered else None,
"created_at": hook.created_at.isoformat()
}
for hook in self.hooks.values()
]
def get_hook_details(self, hook_id: str) -> Optional[Dict[str, Any]]:
"""Get detailed hook information."""
hook = self.hooks.get(hook_id)
if not hook:
return None
return {
"hook_id": hook.hook_id,
"name": hook.name,
"requirements": hook.requirements,
"function_code": hook.function_code,
"priority": hook.priority,
"status": hook.status,
"trigger_count": hook.trigger_count,
"last_triggered": hook.last_triggered.isoformat() if hook.last_triggered else None,
"created_at": hook.created_at.isoformat()
}
async def remove_hook(self, hook_id: str) -> bool:
"""Remove a hook."""
try:
async with self._lock:
if hook_id in self.hooks:
del self.hooks[hook_id]
for instance_id in self.instance_hooks:
if hook_id in self.instance_hooks[instance_id]:
self.instance_hooks[instance_id].remove(hook_id)
debug_logger.log_info("dynamic_hook_system", "remove_hook", f"Removed hook {hook_id}")
return True
return False
except Exception as e:
debug_logger.log_error("dynamic_hook_system", "remove_hook", f"Failed to remove hook {hook_id}: {e}")
return False
def add_instance(self, instance_id: str):
"""Add a new browser instance."""
if instance_id not in self.instance_hooks:
self.instance_hooks[instance_id] = []
async def _execute_hook_action(self, tab, request: RequestInfo, action: HookAction, event=None):
"""Execute a hook action for either request or response stage."""
try:
request_id = uc.cdp.fetch.RequestId(request.request_id)
if action.action == "block":
await tab.send(uc.cdp.fetch.fail_request(
request_id=request_id,
error_reason=uc.cdp.network.ErrorReason.BLOCKED_BY_CLIENT
))
debug_logger.log_info("dynamic_hook_system", "_execute_hook_action", f"Blocked {request.stage} {request.url}")
elif action.action == "fulfill":
headers = []
if action.headers:
for name, value in action.headers.items():
headers.append(uc.cdp.fetch.HeaderEntry(name=name, value=value))
import base64
body_bytes = (action.body or "").encode('utf-8')
body_base64 = base64.b64encode(body_bytes).decode('ascii')
await tab.send(uc.cdp.fetch.fulfill_request(
request_id=request_id,
response_code=action.status_code or 200,
response_headers=headers,
body=body_base64
))
debug_logger.log_info("dynamic_hook_system", "_execute_hook_action", f"Fulfilled {request.stage} {request.url}")
elif action.action == "redirect" and request.stage == "request":
await tab.send(uc.cdp.fetch.continue_request(
request_id=request_id,
url=action.url
))
debug_logger.log_info("dynamic_hook_system", "_execute_hook_action", f"Redirected request {request.url} to {action.url}")
elif action.action == "modify":
if request.stage == "response":
response_headers = []
if action.headers:
for name, value in action.headers.items():
response_headers.append(uc.cdp.fetch.HeaderEntry(name=name, value=value))
await tab.send(uc.cdp.fetch.continue_response(
request_id=request_id,
response_code=action.status_code,
response_headers=response_headers if response_headers else None
))
debug_logger.log_info("dynamic_hook_system", "_execute_hook_action", f"Modified response for {request.url}")
else:
headers = []
if action.headers:
for name, value in action.headers.items():
headers.append(uc.cdp.fetch.HeaderEntry(name=name, value=value))
await tab.send(uc.cdp.fetch.continue_request(
request_id=request_id,
url=action.url or request.url,
method=action.method or request.method,
headers=headers if headers else None,
post_data=action.post_data
))
debug_logger.log_info("dynamic_hook_system", "_execute_hook_action", f"Modified request {request.url}")
else:
if request.stage == "response":
await tab.send(uc.cdp.fetch.continue_response(request_id=request_id))
debug_logger.log_info("dynamic_hook_system", "_execute_hook_action", f"Continued response {request.url}")
else:
await tab.send(uc.cdp.fetch.continue_request(request_id=request_id))
debug_logger.log_info("dynamic_hook_system", "_execute_hook_action", f"Continued request {request.url}")
except Exception as e:
debug_logger.log_error("dynamic_hook_system", "_execute_hook_action", f"Error executing {request.stage} action: {e}")
try:
if request.stage == "response":
await tab.send(uc.cdp.fetch.continue_response(request_id=uc.cdp.fetch.RequestId(request.request_id)))
else:
await tab.send(uc.cdp.fetch.continue_request(request_id=uc.cdp.fetch.RequestId(request.request_id)))
except:
pass
dynamic_hook_system = DynamicHookSystem()
```
--------------------------------------------------------------------------------
/src/dom_handler.py:
--------------------------------------------------------------------------------
```python
"""DOM manipulation and element interaction utilities."""
import asyncio
import time
from typing import List, Optional, Dict, Any
from nodriver import Tab, Element
from models import ElementInfo, ElementAction
from debug_logger import debug_logger
class DOMHandler:
"""Handles DOM queries and element interactions."""
@staticmethod
async def query_elements(
tab: Tab,
selector: str,
text_filter: Optional[str] = None,
visible_only: bool = True,
limit: Optional[Any] = None
) -> List[ElementInfo]:
"""
Query elements with advanced filtering.
Args:
tab (Tab): The browser tab object.
selector (str): CSS or XPath selector for elements.
text_filter (Optional[str]): Filter elements by text content.
visible_only (bool): Only include visible elements.
limit (Optional[Any]): Limit the number of results.
Returns:
List[ElementInfo]: List of element information objects.
"""
processed_limit = None
if limit is not None:
try:
if isinstance(limit, int):
processed_limit = limit
elif isinstance(limit, str) and limit.isdigit():
processed_limit = int(limit)
elif isinstance(limit, str) and limit.strip() == '':
processed_limit = None
else:
debug_logger.log_warning('DOMHandler', 'query_elements',
f'Invalid limit parameter: {limit} (type: {type(limit)})')
processed_limit = None
except (ValueError, TypeError) as e:
debug_logger.log_error('DOMHandler', 'query_elements', e,
{'limit_value': limit, 'limit_type': type(limit)})
processed_limit = None
debug_logger.log_info('DOMHandler', 'query_elements',
f'Starting query with selector: {selector}',
{'text_filter': text_filter, 'visible_only': visible_only,
'limit': limit, 'processed_limit': processed_limit})
try:
if selector.startswith('//'):
elements = await tab.select_all(f'xpath={selector}')
debug_logger.log_info('DOMHandler', 'query_elements',
f'XPath query returned {len(elements)} elements')
else:
elements = await tab.select_all(selector)
debug_logger.log_info('DOMHandler', 'query_elements',
f'CSS query returned {len(elements)} elements')
results = []
for idx, elem in enumerate(elements):
try:
debug_logger.log_info('DOMHandler', 'query_elements',
f'Processing element {idx+1}/{len(elements)}')
if hasattr(elem, 'update'):
await elem.update()
debug_logger.log_info('DOMHandler', 'query_elements',
f'Element {idx+1} updated')
tag_name = elem.tag_name if hasattr(elem, 'tag_name') else 'unknown'
text_content = elem.text_all if hasattr(elem, 'text_all') else ''
attrs = elem.attrs if hasattr(elem, 'attrs') else {}
debug_logger.log_info('DOMHandler', 'query_elements',
f'Element {idx+1}: tag={tag_name}, text_len={len(text_content)}, attrs={len(attrs)}')
if text_filter and text_filter.lower() not in text_content.lower():
continue
is_visible = True
if visible_only:
try:
is_visible = await elem.apply(
"""(elem) => {
var style = window.getComputedStyle(elem);
return style.display !== 'none' &&
style.visibility !== 'hidden' &&
style.opacity !== '0';
}"""
)
if not is_visible:
continue
except:
pass
bbox = None
try:
position = await elem.get_position()
if position:
bbox = {
'x': position.x,
'y': position.y,
'width': position.width,
'height': position.height
}
debug_logger.log_info('DOMHandler', 'query_elements',
f'Element {idx+1} position: {bbox}')
except Exception as pos_error:
debug_logger.log_warning('DOMHandler', 'query_elements',
f'Could not get position for element {idx+1}: {pos_error}')
is_clickable = False
children_count = 0
try:
if hasattr(elem, 'children'):
children = elem.children
children_count = len(children) if children else 0
except Exception:
pass
element_info = ElementInfo(
selector=selector,
tag_name=tag_name,
text=text_content[:500] if text_content else None,
attributes=attrs or {},
is_visible=is_visible,
is_clickable=is_clickable,
bounding_box=bbox,
children_count=children_count
)
results.append(element_info)
if processed_limit and len(results) >= processed_limit:
debug_logger.log_info('DOMHandler', 'query_elements',
f'Reached limit of {processed_limit} results')
break
except Exception as elem_error:
debug_logger.log_error('DOMHandler', 'query_elements',
elem_error,
{'element_index': idx, 'selector': selector})
continue
debug_logger.log_info('DOMHandler', 'query_elements',
f'Returning {len(results)} results')
return results
except Exception as e:
debug_logger.log_error('DOMHandler', 'query_elements', e,
{'selector': selector, 'tab': str(tab)})
return []
@staticmethod
async def click_element(
tab: Tab,
selector: str,
text_match: Optional[str] = None,
timeout: int = 10000
) -> bool:
"""
Click an element with smart retry logic.
Args:
tab (Tab): The browser tab object.
selector (str): CSS selector for the element.
text_match (Optional[str]): Match element by text content.
timeout (int): Timeout in milliseconds.
Returns:
bool: True if click succeeded, False otherwise.
"""
try:
element = None
if text_match:
element = await tab.find(text_match, best_match=True)
else:
element = await tab.select(selector, timeout=timeout/1000)
if not element:
raise Exception(f"Element not found: {selector}")
await element.scroll_into_view()
await asyncio.sleep(0.5)
try:
await element.click()
except Exception:
await element.mouse_click()
return True
except Exception as e:
raise Exception(f"Failed to click element: {str(e)}")
@staticmethod
async def type_text(
tab: Tab,
selector: str,
text: str,
clear_first: bool = True,
delay_ms: int = 50,
parse_newlines: bool = False,
shift_enter: bool = False
) -> bool:
"""
Type text with human-like delays and optional newline parsing.
Args:
tab (Tab): The browser tab object.
selector (str): CSS selector for the input element.
text (str): Text to type.
clear_first (bool): Clear input before typing.
delay_ms (int): Delay between keystrokes in milliseconds.
parse_newlines (bool): If True, parse \n as Enter key presses.
shift_enter (bool): If True, use Shift+Enter instead of Enter (for chat apps).
Returns:
bool: True if typing succeeded, False otherwise.
"""
try:
element = await tab.select(selector)
if not element:
raise Exception(f"Element not found: {selector}")
await element.focus()
await asyncio.sleep(0.1)
if clear_first:
try:
await element.apply("(elem) => { elem.value = ''; }")
except:
await element.send_keys('\ue009' + 'a')
await element.send_keys('\ue017')
await asyncio.sleep(0.1)
if parse_newlines:
from nodriver import cdp
lines = text.split('\n')
for i, line in enumerate(lines):
for char in line:
await element.send_keys(char)
await asyncio.sleep(delay_ms / 1000)
if i < len(lines) - 1:
if shift_enter:
await element.apply('''(elem) => {
const start = elem.selectionStart;
const end = elem.selectionEnd;
const value = elem.value;
elem.value = value.substring(0, start) + '\\n' + value.substring(end);
elem.selectionStart = elem.selectionEnd = start + 1;
elem.dispatchEvent(new KeyboardEvent('keydown', {
key: 'Enter',
code: 'Enter',
shiftKey: true,
bubbles: true
}));
elem.dispatchEvent(new Event('input', { bubbles: true }));
}''')
else:
await element.apply('''(elem) => {
const start = elem.selectionStart;
const end = elem.selectionEnd;
const value = elem.value;
elem.value = value.substring(0, start) + '\\n' + value.substring(end);
elem.selectionStart = elem.selectionEnd = start + 1;
elem.dispatchEvent(new KeyboardEvent('keydown', {
key: 'Enter',
code: 'Enter',
bubbles: true
}));
elem.dispatchEvent(new Event('input', { bubbles: true }));
}''')
await asyncio.sleep(delay_ms / 1000)
else:
for char in text:
await element.send_keys(char)
await asyncio.sleep(delay_ms / 1000)
return True
except Exception as e:
raise Exception(f"Failed to type text: {str(e)}")
@staticmethod
async def paste_text(
tab: Tab,
selector: str,
text: str,
clear_first: bool = True
) -> bool:
"""
Paste text instantly using nodriver's insert_text method.
This is much faster than typing character by character.
Args:
tab (Tab): The browser tab object.
selector (str): CSS selector for the input element.
text (str): Text to paste.
clear_first (bool): Clear input before pasting.
Returns:
bool: True if pasting succeeded, False otherwise.
"""
from nodriver import cdp
try:
element = await tab.select(selector)
if not element:
raise Exception(f"Element not found: {selector}")
await element.focus()
await asyncio.sleep(0.1)
if clear_first:
try:
await element.apply("(elem) => { elem.value = ''; }")
except:
await tab.send(cdp.input_.dispatch_key_event(
"rawKeyDown",
modifiers=2, # Ctrl
key="a",
code="KeyA",
windows_virtual_key_code=65
))
await tab.send(cdp.input_.dispatch_key_event(
"keyUp",
modifiers=2, # Ctrl
key="a",
code="KeyA",
windows_virtual_key_code=65
))
await tab.send(cdp.input_.dispatch_key_event(
"rawKeyDown",
key="Delete",
code="Delete",
windows_virtual_key_code=46
))
await tab.send(cdp.input_.dispatch_key_event(
"keyUp",
key="Delete",
code="Delete",
windows_virtual_key_code=46
))
await asyncio.sleep(0.1)
await tab.send(cdp.input_.insert_text(text))
return True
except Exception as e:
raise Exception(f"Failed to paste text: {str(e)}")
@staticmethod
async def select_option(
tab: Tab,
selector: str,
value: Optional[str] = None,
text: Optional[str] = None,
index: Optional[int] = None
) -> bool:
"""
Select option from dropdown using nodriver's native methods.
Args:
tab (Tab): The browser tab object.
selector (str): CSS selector for the select element.
value (Optional[str]): Option value to select.
text (Optional[str]): Option text to select.
index (Optional[int]): Option index to select.
Returns:
bool: True if option selected, False otherwise.
"""
try:
select_element = await tab.select(selector)
if not select_element:
raise Exception(f"Select element not found: {selector}")
if text is not None:
await select_element.send_keys(text)
return True
if value is not None:
await tab.evaluate(f"""
const select = document.querySelector('{selector}');
if (select) {{
select.value = '{value}';
select.dispatchEvent(new Event('change', {{bubbles: true}}));
}}
""")
return True
elif index is not None:
await tab.evaluate(f"""
const select = document.querySelector('{selector}');
if (select && {index} >= 0 && {index} < select.options.length) {{
select.selectedIndex = {index};
select.dispatchEvent(new Event('change', {{bubbles: true}}));
}}
""")
return True
raise Exception("No selection criteria provided (value, text, or index)")
except Exception as e:
raise Exception(f"Failed to select option: {str(e)}")
@staticmethod
async def get_element_state(
tab: Tab,
selector: str
) -> Dict[str, Any]:
"""
Get complete state of an element.
Args:
tab (Tab): The browser tab object.
selector (str): CSS selector for the element.
Returns:
Dict[str, Any]: Dictionary of element state properties.
"""
try:
element = await tab.select(selector)
if not element:
raise Exception(f"Element not found: {selector}")
if hasattr(element, 'update'):
await element.update()
state = {
'tag_name': element.tag_name if hasattr(element, 'tag_name') else 'unknown',
'text': element.text if hasattr(element, 'text') else '',
'text_all': element.text_all if hasattr(element, 'text_all') else '',
'attributes': element.attrs if hasattr(element, 'attrs') else {},
'is_visible': True,
'is_clickable': False,
'is_enabled': True,
'value': element.attrs.get('value') if hasattr(element, 'attrs') else None,
'href': element.attrs.get('href') if hasattr(element, 'attrs') else None,
'src': element.attrs.get('src') if hasattr(element, 'attrs') else None,
'class': element.attrs.get('class') if hasattr(element, 'attrs') else None,
'id': element.attrs.get('id') if hasattr(element, 'attrs') else None,
'position': await element.get_position() if hasattr(element, 'get_position') else None,
'computed_style': {},
'children_count': len(element.children) if hasattr(element, 'children') and element.children else 0,
'parent_tag': None
}
return state
except Exception as e:
raise Exception(f"Failed to get element state: {str(e)}")
@staticmethod
async def wait_for_element(
tab: Tab,
selector: str,
timeout: int = 30000,
visible: bool = True,
text_content: Optional[str] = None
) -> bool:
"""
Wait for element to appear and match conditions.
Args:
tab (Tab): The browser tab object.
selector (str): CSS selector for the element.
timeout (int): Timeout in milliseconds.
visible (bool): Wait for element to be visible.
text_content (Optional[str]): Wait for element to contain text.
Returns:
bool: True if element matches conditions, False otherwise.
"""
start_time = time.time()
timeout_seconds = timeout / 1000
while time.time() - start_time < timeout_seconds:
try:
element = await tab.select(selector)
if element:
if visible:
try:
is_visible = await element.apply(
"""(elem) => {
var style = window.getComputedStyle(elem);
return style.display !== 'none' &&
style.visibility !== 'hidden' &&
style.opacity !== '0';
}"""
)
if not is_visible:
await asyncio.sleep(0.5)
continue
except:
pass
if text_content:
text = element.text_all
if text_content not in text:
await asyncio.sleep(0.5)
continue
return True
except Exception:
pass
await asyncio.sleep(0.5)
return False
@staticmethod
async def execute_script(
tab: Tab,
script: str,
args: Optional[List[Any]] = None
) -> Any:
"""
Execute JavaScript in page context.
Args:
tab (Tab): The browser tab object.
script (str): JavaScript code to execute.
args (Optional[List[Any]]): Arguments for the script.
Returns:
Any: Result of script execution.
"""
try:
if args:
result = await tab.evaluate(f'(function() {{ {script} }})({",".join(map(str, args))})')
else:
result = await tab.evaluate(script)
return result
except Exception as e:
raise Exception(f"Failed to execute script: {str(e)}")
@staticmethod
async def get_page_content(
tab: Tab,
include_frames: bool = False
) -> Dict[str, str]:
"""
Get page HTML and text content.
Args:
tab (Tab): The browser tab object.
include_frames (bool): Include iframe contents.
Returns:
Dict[str, str]: Dictionary with page content.
"""
try:
html = await tab.get_content()
text = await tab.evaluate("document.body.innerText")
content = {
'html': html,
'text': text,
'url': await tab.evaluate("window.location.href"),
'title': await tab.evaluate("document.title")
}
if include_frames:
frames = []
iframe_elements = await tab.select_all('iframe')
for i, iframe in enumerate(iframe_elements):
try:
src = iframe.attrs.get('src') if hasattr(iframe, 'attrs') else None
if src:
frames.append({
'index': i,
'src': src,
'id': iframe.attrs.get('id') if hasattr(iframe, 'attrs') else None,
'name': iframe.attrs.get('name') if hasattr(iframe, 'attrs') else None
})
except Exception:
continue
content['frames'] = frames
return content
except Exception as e:
raise Exception(f"Failed to get page content: {str(e)}")
@staticmethod
async def scroll_page(
tab: Tab,
direction: str = "down",
amount: int = 500,
smooth: bool = True
) -> bool:
"""
Scroll the page in specified direction.
Args:
tab (Tab): The browser tab object.
direction (str): Direction to scroll ('down', 'up', 'right', 'left', 'top', 'bottom').
amount (int): Amount to scroll in pixels.
smooth (bool): Use smooth scrolling.
Returns:
bool: True if scroll succeeded, False otherwise.
"""
try:
if direction == "down":
script = f"window.scrollBy(0, {amount})"
elif direction == "up":
script = f"window.scrollBy(0, -{amount})"
elif direction == "right":
script = f"window.scrollBy({amount}, 0)"
elif direction == "left":
script = f"window.scrollBy(-{amount}, 0)"
elif direction == "top":
script = "window.scrollTo(0, 0)"
elif direction == "bottom":
script = "window.scrollTo(0, document.body.scrollHeight)"
else:
raise ValueError(f"Invalid scroll direction: {direction}")
if smooth:
script = script.replace("scrollBy", "scrollBy({behavior: 'smooth'}, ")
script = script.replace("scrollTo", "scrollTo({behavior: 'smooth', top: ")
if "scrollTo" in script:
script = script.replace(")", "})")
await tab.evaluate(script)
await asyncio.sleep(0.5 if smooth else 0.1)
return True
except Exception as e:
raise Exception(f"Failed to scroll page: {str(e)}")
```
--------------------------------------------------------------------------------
/src/file_based_element_cloner.py:
--------------------------------------------------------------------------------
```python
import asyncio
import json
import os
import sys
import uuid
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Any, Optional
try:
from .debug_logger import debug_logger
except ImportError:
from debug_logger import debug_logger
project_root = Path(__file__).parent.parent
sys.path.append(str(project_root))
from comprehensive_element_cloner import ComprehensiveElementCloner
from element_cloner import element_cloner
class FileBasedElementCloner:
"""Element cloner that saves data to files and returns file paths."""
def __init__(self, output_dir: str = "element_clones"):
"""
Initialize with output directory for clone files.
Args:
output_dir (str): Directory to save clone files.
"""
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True)
self.comprehensive_cloner = ComprehensiveElementCloner()
def _safe_process_framework_handlers(self, framework_handlers):
"""Safely process framework handlers that might be dict or list."""
if isinstance(framework_handlers, dict):
return {k: len(v) if isinstance(v, list) else str(v) for k, v in framework_handlers.items()}
elif isinstance(framework_handlers, list):
return {"handlers": len(framework_handlers)}
else:
return {"value": str(framework_handlers)}
def _generate_filename(self, prefix: str, extension: str = "json") -> str:
"""
Generate unique filename with timestamp.
Args:
prefix (str): Prefix for the filename.
extension (str): File extension.
Returns:
str: Generated filename.
"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
unique_id = str(uuid.uuid4())[:8]
return f"{prefix}_{timestamp}_{unique_id}.{extension}"
async def extract_element_styles_to_file(
self,
tab,
selector: str,
include_computed: bool = True,
include_css_rules: bool = True,
include_pseudo: bool = True,
include_inheritance: bool = False
) -> Dict[str, Any]:
"""
Extract element styles and save to file, returning file path.
Args:
tab: Browser tab instance
selector (str): CSS selector for the element
include_computed (bool): Include computed styles
include_css_rules (bool): Include matching CSS rules
include_pseudo (bool): Include pseudo-element styles
include_inheritance (bool): Include style inheritance chain
Returns:
Dict[str, Any]: File path and summary of extracted styles
"""
try:
debug_logger.log_info("file_element_cloner", "extract_styles_to_file",
f"Starting style extraction for selector: {selector}")
# Extract styles using element_cloner
style_data = await element_cloner.extract_element_styles(
tab,
selector=selector,
include_computed=include_computed,
include_css_rules=include_css_rules,
include_pseudo=include_pseudo,
include_inheritance=include_inheritance
)
# Generate filename and save
filename = self._generate_filename("styles")
file_path = self._save_to_file(style_data, filename)
# Create summary
summary = {
"file_path": str(file_path),
"extraction_type": "styles",
"selector": selector,
"url": getattr(tab, 'url', 'unknown'),
"components": {
"computed_styles_count": len(style_data.get('computed_styles', {})),
"css_rules_count": len(style_data.get('css_rules', [])),
"pseudo_elements_count": len(style_data.get('pseudo_elements', {})),
"custom_properties_count": len(style_data.get('custom_properties', {}))
}
}
debug_logger.log_info("file_element_cloner", "extract_styles_to_file",
f"Styles saved to {file_path}")
return summary
except Exception as e:
debug_logger.log_error("file_element_cloner", "extract_styles_to_file", e)
return {"error": str(e)}
def _save_to_file(self, data: Dict[str, Any], filename: str) -> str:
"""
Save data to file and return absolute path.
Args:
data (Dict[str, Any]): Data to save.
filename (str): Name of the file.
Returns:
str: Absolute path to the saved file.
"""
file_path = self.output_dir / filename
with open(file_path, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
return str(file_path.absolute())
async def extract_complete_element_to_file(
self,
tab,
selector: str,
include_children: bool = True
) -> Dict[str, Any]:
"""
Extract complete element using working comprehensive cloner and save to file.
Args:
tab: Browser tab object.
selector (str): CSS selector for the element.
include_children (bool): Whether to include children.
Returns:
Dict[str, Any]: Summary of extraction and file path.
"""
try:
complete_data = await self.comprehensive_cloner.extract_complete_element(
tab, selector, include_children
)
complete_data['_metadata'] = {
'extraction_type': 'complete_comprehensive',
'selector': selector,
'timestamp': datetime.now().isoformat(),
'include_children': include_children
}
filename = self._generate_filename("complete_comprehensive")
file_path = self._save_to_file(complete_data, filename)
debug_logger.log_info("file_element_cloner", "extract_complete_to_file",
f"Saved complete element data to {file_path}")
summary = {
"file_path": file_path,
"extraction_type": "complete_comprehensive",
"selector": selector,
"url": complete_data.get('url', 'unknown'),
"summary": {
"tag_name": complete_data.get('html', {}).get('tagName', 'unknown'),
"computed_styles_count": len(complete_data.get('styles', {})),
"attributes_count": len(complete_data.get('html', {}).get('attributes', [])),
"event_listeners_count": len(complete_data.get('eventListeners', [])),
"children_count": len(complete_data.get('children', [])) if include_children else 0,
"has_pseudo_elements": bool(complete_data.get('pseudoElements')),
"css_rules_count": len(complete_data.get('cssRules', [])),
"animations_count": len(complete_data.get('animations', [])),
"file_size_kb": round(len(json.dumps(complete_data)) / 1024, 2)
}
}
return summary
except Exception as e:
debug_logger.log_error("file_element_cloner", "extract_complete_to_file", e)
return {"error": str(e)}
async def extract_element_structure_to_file(
self,
tab,
element=None,
selector: str = None,
include_children: bool = False,
include_attributes: bool = True,
include_data_attributes: bool = True,
max_depth: int = 3
) -> Dict[str, str]:
"""
Extract structure and save to file, return file path.
Args:
tab: Browser tab object.
element: DOM element object.
selector (str): CSS selector for the element.
include_children (bool): Whether to include children.
include_attributes (bool): Whether to include attributes.
include_data_attributes (bool): Whether to include data attributes.
max_depth (int): Maximum depth for extraction.
Returns:
Dict[str, str]: Summary of extraction and file path.
"""
try:
structure_data = await element_cloner.extract_element_structure(
tab, element, selector, include_children,
include_attributes, include_data_attributes, max_depth
)
structure_data['_metadata'] = {
'extraction_type': 'structure',
'selector': selector,
'timestamp': datetime.now().isoformat(),
'options': {
'include_children': include_children,
'include_attributes': include_attributes,
'include_data_attributes': include_data_attributes,
'max_depth': max_depth
}
}
filename = self._generate_filename("structure")
file_path = self._save_to_file(structure_data, filename)
debug_logger.log_info("file_element_cloner", "extract_structure_to_file",
f"Saved structure data to {file_path}")
return {
"file_path": file_path,
"extraction_type": "structure",
"selector": selector,
"summary": {
"tag_name": structure_data.get('tag_name'),
"attributes_count": len(structure_data.get('attributes', {})),
"data_attributes_count": len(structure_data.get('data_attributes', {})),
"children_count": len(structure_data.get('children', [])),
"dom_path": structure_data.get('dom_path')
}
}
except Exception as e:
debug_logger.log_error("file_element_cloner", "extract_structure_to_file", e)
return {"error": str(e)}
async def extract_element_events_to_file(
self,
tab,
element=None,
selector: str = None,
include_inline: bool = True,
include_listeners: bool = True,
include_framework: bool = True,
analyze_handlers: bool = True
) -> Dict[str, str]:
"""
Extract events and save to file, return file path.
Args:
tab: Browser tab object.
element: DOM element object.
selector (str): CSS selector for the element.
include_inline (bool): Include inline event handlers.
include_listeners (bool): Include event listeners.
include_framework (bool): Include framework event handlers.
analyze_handlers (bool): Analyze event handlers.
Returns:
Dict[str, str]: Summary of extraction and file path.
"""
try:
event_data = await element_cloner.extract_element_events(
tab, element, selector, include_inline,
include_listeners, include_framework, analyze_handlers
)
event_data['_metadata'] = {
'extraction_type': 'events',
'selector': selector,
'timestamp': datetime.now().isoformat(),
'options': {
'include_inline': include_inline,
'include_listeners': include_listeners,
'include_framework': include_framework,
'analyze_handlers': analyze_handlers
}
}
filename = self._generate_filename("events")
file_path = self._save_to_file(event_data, filename)
debug_logger.log_info("file_element_cloner", "extract_events_to_file",
f"Saved events data to {file_path}")
return {
"file_path": file_path,
"extraction_type": "events",
"selector": selector,
"summary": {
"inline_handlers_count": len(event_data.get('inline_handlers', [])),
"event_listeners_count": len(event_data.get('event_listeners', [])),
"detected_frameworks": event_data.get('detected_frameworks', []),
"framework_handlers": self._safe_process_framework_handlers(event_data.get('framework_handlers', {}))
}
}
except Exception as e:
debug_logger.log_error("file_element_cloner", "extract_events_to_file", e)
return {"error": str(e)}
async def extract_element_animations_to_file(
self,
tab,
element=None,
selector: str = None,
include_css_animations: bool = True,
include_transitions: bool = True,
include_transforms: bool = True,
analyze_keyframes: bool = True
) -> Dict[str, str]:
"""
Extract animations and save to file, return file path.
Args:
tab: Browser tab object.
element: DOM element object.
selector (str): CSS selector for the element.
include_css_animations (bool): Include CSS animations.
include_transitions (bool): Include transitions.
include_transforms (bool): Include transforms.
analyze_keyframes (bool): Analyze keyframes.
Returns:
Dict[str, str]: Summary of extraction and file path.
"""
try:
animation_data = await element_cloner.extract_element_animations(
tab, element, selector, include_css_animations,
include_transitions, include_transforms, analyze_keyframes
)
animation_data['_metadata'] = {
'extraction_type': 'animations',
'selector': selector,
'timestamp': datetime.now().isoformat(),
'options': {
'include_css_animations': include_css_animations,
'include_transitions': include_transitions,
'include_transforms': include_transforms,
'analyze_keyframes': analyze_keyframes
}
}
filename = self._generate_filename("animations")
file_path = self._save_to_file(animation_data, filename)
debug_logger.log_info("file_element_cloner", "extract_animations_to_file",
f"Saved animations data to {file_path}")
return {
"file_path": file_path,
"extraction_type": "animations",
"selector": selector,
"summary": {
"has_animations": animation_data.get('animations', {}).get('animation_name', 'none') != 'none',
"has_transitions": animation_data.get('transitions', {}).get('transition_property', 'none') != 'none',
"has_transforms": animation_data.get('transforms', {}).get('transform', 'none') != 'none',
"keyframes_count": len(animation_data.get('keyframes', []))
}
}
except Exception as e:
debug_logger.log_error("file_element_cloner", "extract_animations_to_file", e)
return {"error": str(e)}
async def extract_element_assets_to_file(
self,
tab,
element=None,
selector: str = None,
include_images: bool = True,
include_backgrounds: bool = True,
include_fonts: bool = True,
fetch_external: bool = False
) -> Dict[str, str]:
"""
Extract assets and save to file, return file path.
Args:
tab: Browser tab object.
element: DOM element object.
selector (str): CSS selector for the element.
include_images (bool): Include images.
include_backgrounds (bool): Include background images.
include_fonts (bool): Include fonts.
fetch_external (bool): Fetch external assets.
Returns:
Dict[str, str]: Summary of extraction and file path.
"""
try:
asset_data = await element_cloner.extract_element_assets(
tab, element, selector, include_images,
include_backgrounds, include_fonts, fetch_external
)
asset_data['_metadata'] = {
'extraction_type': 'assets',
'selector': selector,
'timestamp': datetime.now().isoformat(),
'options': {
'include_images': include_images,
'include_backgrounds': include_backgrounds,
'include_fonts': include_fonts,
'fetch_external': fetch_external
}
}
filename = self._generate_filename("assets")
file_path = self._save_to_file(asset_data, filename)
debug_logger.log_info("file_element_cloner", "extract_assets_to_file",
f"Saved assets data to {file_path}")
return {
"file_path": file_path,
"extraction_type": "assets",
"selector": selector,
"summary": {
"images_count": len(asset_data.get('images', [])),
"background_images_count": len(asset_data.get('background_images', [])),
"font_family": asset_data.get('fonts', {}).get('family'),
"custom_fonts_count": len(asset_data.get('fonts', {}).get('custom_fonts', [])),
"icons_count": len(asset_data.get('icons', [])),
"videos_count": len(asset_data.get('videos', [])),
"audio_count": len(asset_data.get('audio', []))
}
}
except Exception as e:
debug_logger.log_error("file_element_cloner", "extract_assets_to_file", e)
return {"error": str(e)}
async def extract_related_files_to_file(
self,
tab,
element=None,
selector: str = None,
analyze_css: bool = True,
analyze_js: bool = True,
follow_imports: bool = False,
max_depth: int = 2
) -> Dict[str, str]:
"""
Extract related files and save to file, return file path.
Args:
tab: Browser tab object.
element: DOM element object.
selector (str): CSS selector for the element.
analyze_css (bool): Analyze CSS files.
analyze_js (bool): Analyze JS files.
follow_imports (bool): Follow imports.
max_depth (int): Maximum depth for import following.
Returns:
Dict[str, str]: Summary of extraction and file path.
"""
try:
file_data = await element_cloner.extract_related_files(
tab, element, selector, analyze_css, analyze_js, follow_imports, max_depth
)
file_data['_metadata'] = {
'extraction_type': 'related_files',
'selector': selector,
'timestamp': datetime.now().isoformat(),
'options': {
'analyze_css': analyze_css,
'analyze_js': analyze_js,
'follow_imports': follow_imports,
'max_depth': max_depth
}
}
filename = self._generate_filename("related_files")
file_path = self._save_to_file(file_data, filename)
debug_logger.log_info("file_element_cloner", "extract_related_files_to_file",
f"Saved related files data to {file_path}")
return {
"file_path": file_path,
"extraction_type": "related_files",
"selector": selector,
"summary": {
"stylesheets_count": len(file_data.get('stylesheets', [])),
"scripts_count": len(file_data.get('scripts', [])),
"imports_count": len(file_data.get('imports', [])),
"modules_count": len(file_data.get('modules', []))
}
}
except Exception as e:
debug_logger.log_error("file_element_cloner", "extract_related_files_to_file", e)
return {"error": str(e)}
async def clone_element_complete_to_file(
self,
tab,
element=None,
selector: str = None,
extraction_options: Dict[str, Any] = None
) -> Dict[str, Any]:
"""
Master function that extracts all element data and saves to file.
Returns file path instead of full data.
Args:
tab: Browser tab object.
element: DOM element object.
selector (str): CSS selector for the element.
extraction_options (Dict[str, Any]): Extraction options.
Returns:
Dict[str, Any]: Summary of extraction and file path.
"""
try:
complete_data = await element_cloner.clone_element_complete(
tab, element, selector, extraction_options
)
if 'error' in complete_data:
return complete_data
complete_data['_metadata'] = {
'extraction_type': 'complete_clone',
'selector': selector,
'timestamp': datetime.now().isoformat(),
'extraction_options': extraction_options
}
filename = self._generate_filename("complete_clone")
file_path = self._save_to_file(complete_data, filename)
summary = {
"file_path": file_path,
"extraction_type": "complete_clone",
"selector": selector,
"url": complete_data.get('url'),
"components": {}
}
if 'styles' in complete_data:
styles = complete_data['styles']
summary['components']['styles'] = {
'computed_styles_count': len(styles.get('computed_styles', {})),
'css_rules_count': len(styles.get('css_rules', [])),
'pseudo_elements_count': len(styles.get('pseudo_elements', {}))
}
if 'structure' in complete_data:
structure = complete_data['structure']
summary['components']['structure'] = {
'tag_name': structure.get('tag_name'),
'attributes_count': len(structure.get('attributes', {})),
'children_count': len(structure.get('children', []))
}
if 'events' in complete_data:
events = complete_data['events']
summary['components']['events'] = {
'inline_handlers_count': len(events.get('inline_handlers', [])),
'detected_frameworks': events.get('detected_frameworks', [])
}
if 'animations' in complete_data:
animations = complete_data['animations']
summary['components']['animations'] = {
'has_animations': animations.get('animations', {}).get('animation_name', 'none') != 'none',
'keyframes_count': len(animations.get('keyframes', []))
}
if 'assets' in complete_data:
assets = complete_data['assets']
summary['components']['assets'] = {
'images_count': len(assets.get('images', [])),
'background_images_count': len(assets.get('background_images', []))
}
if 'related_files' in complete_data:
files = complete_data['related_files']
summary['components']['related_files'] = {
'stylesheets_count': len(files.get('stylesheets', [])),
'scripts_count': len(files.get('scripts', []))
}
debug_logger.log_info("file_element_cloner", "clone_complete_to_file",
f"Saved complete clone data to {file_path}")
return summary
except Exception as e:
debug_logger.log_error("file_element_cloner", "clone_complete_to_file", e)
return {"error": str(e)}
def list_clone_files(self) -> List[Dict[str, Any]]:
"""
List all clone files in the output directory.
Returns:
List[Dict[str, Any]]: List of file info dictionaries.
"""
files = []
for file_path in self.output_dir.glob("*.json"):
try:
file_info = {
"file_path": str(file_path.absolute()),
"filename": file_path.name,
"size": file_path.stat().st_size,
"created": datetime.fromtimestamp(file_path.stat().st_ctime).isoformat(),
"modified": datetime.fromtimestamp(file_path.stat().st_mtime).isoformat()
}
try:
with open(file_path, 'r', encoding='utf-8') as f:
data = json.load(f)
if '_metadata' in data:
file_info['metadata'] = data['_metadata']
except:
pass
files.append(file_info)
except Exception as e:
debug_logger.log_warning("file_element_cloner", "list_files", f"Error reading {file_path}: {e}")
files.sort(key=lambda x: x['created'], reverse=True)
return files
def cleanup_old_files(self, max_age_hours: int = 24) -> int:
"""
Clean up clone files older than specified hours.
Args:
max_age_hours (int): Maximum age of files in hours.
Returns:
int: Number of deleted files.
"""
import time
cutoff_time = time.time() - (max_age_hours * 3600)
deleted_count = 0
for file_path in self.output_dir.glob("*.json"):
try:
if file_path.stat().st_ctime < cutoff_time:
file_path.unlink()
deleted_count += 1
debug_logger.log_info("file_element_cloner", "cleanup", f"Deleted old file: {file_path.name}")
except Exception as e:
debug_logger.log_warning("file_element_cloner", "cleanup", f"Error deleting {file_path}: {e}")
return deleted_count
file_based_element_cloner = FileBasedElementCloner()
```
--------------------------------------------------------------------------------
/src/element_cloner.py:
--------------------------------------------------------------------------------
```python
"""Advanced element cloning system with complete styling and JS extraction."""
import asyncio
import json
import re
from typing import Dict, List, Any, Optional, Set, Union
from urllib.parse import urljoin, urlparse
from pathlib import Path
import requests
try:
from .debug_logger import debug_logger
except ImportError:
from debug_logger import debug_logger
class ElementCloner:
"""Advanced element cloning with full fidelity extraction."""
def __init__(self):
self.extracted_files = {}
self.framework_patterns = {
'react': [r'_react', r'__reactInternalInstance', r'__reactFiber'],
'vue': [r'__vue__', r'_vnode', r'$el'],
'angular': [r'ng-', r'__ngContext__', r'ɵ'],
'jquery': [r'jQuery', r'\$\.', r'__jquery']
}
async def extract_element_styles(
self,
tab,
element=None,
selector: str = None,
include_computed: bool = True,
include_css_rules: bool = True,
include_pseudo: bool = True,
include_inheritance: bool = False
) -> Dict[str, Any]:
"""
Extract complete styling information from an element.
Args:
tab (Any): Browser tab instance
element (Any): Element object or None to use selector
selector (str): CSS selector if element is None
include_computed (bool): Include computed styles
include_css_rules (bool): Include matching CSS rules
include_pseudo (bool): Include pseudo-element styles
include_inheritance (bool): Include style inheritance chain
Returns:
Dict[str, Any]: Dict with styling data
"""
try:
return await self.extract_element_styles_cdp(
tab=tab,
element=element,
selector=selector,
include_computed=include_computed,
include_css_rules=include_css_rules,
include_pseudo=include_pseudo,
include_inheritance=include_inheritance
)
except Exception as e:
debug_logger.log_error("element_cloner", "extract_styles", e)
return {"error": str(e)}
def _load_js_file(self, filename: str, selector: str, options: dict) -> str:
"""Load and prepare JavaScript file with template substitution"""
js_dir = Path(__file__).parent / "js"
js_file = js_dir / filename
if not js_file.exists():
raise FileNotFoundError(f"JavaScript file not found: {js_file}")
with open(js_file, 'r', encoding='utf-8') as f:
js_code = f.read()
js_code = js_code.replace('$SELECTOR$', selector)
js_code = js_code.replace('$SELECTOR', selector)
js_code = js_code.replace('$OPTIONS$', json.dumps(options))
js_code = js_code.replace('$OPTIONS', json.dumps(options))
for key, value in options.items():
placeholder_key = f'${key.upper()}'
placeholder_value = 'true' if value else 'false'
js_code = js_code.replace(placeholder_key, placeholder_value)
return js_code
def _convert_nodriver_result(self, data):
"""Convert nodriver's array format back to dict"""
if isinstance(data, list) and len(data) > 0 and isinstance(data[0], list):
result = {}
for item in data:
if isinstance(item, list) and len(item) == 2:
key = item[0]
value_obj = item[1]
if isinstance(value_obj, dict) and 'type' in value_obj:
if value_obj['type'] == 'string':
result[key] = value_obj.get('value', '')
elif value_obj['type'] == 'number':
result[key] = value_obj.get('value', 0)
elif value_obj['type'] == 'null':
result[key] = None
elif value_obj['type'] == 'array':
result[key] = value_obj.get('value', [])
elif value_obj['type'] == 'object':
result[key] = self._convert_nodriver_result(value_obj.get('value', []))
else:
result[key] = value_obj.get('value')
else:
result[key] = value_obj
return result
return data
async def extract_element_structure(
self,
tab,
element=None,
selector: str = None,
include_children: bool = False,
include_attributes: bool = True,
include_data_attributes: bool = True,
max_depth: int = 3
) -> Dict[str, Any]:
"""
Extract complete HTML structure and DOM information.
Args:
tab (Any): Browser tab instance
element (Any): Element object or None to use selector
selector (str): CSS selector if element is None
include_children (bool): Include child elements
include_attributes (bool): Include all attributes
include_data_attributes (bool): Include data-* attributes specifically
max_depth (int): Maximum depth for children extraction
Returns:
Dict[str, Any]: Dict with structure data
"""
try:
if not selector:
return {"error": "Selector is required"}
options = {
'include_children': include_children,
'include_attributes': include_attributes,
'include_data_attributes': include_data_attributes,
'max_depth': max_depth
}
js_code = self._load_js_file('extract_structure.js', selector, options)
structure_data = await tab.evaluate(js_code)
if hasattr(structure_data, 'exception_details'):
return {"error": f"JavaScript error: {structure_data.exception_details}"}
elif isinstance(structure_data, dict):
debug_logger.log_info("element_cloner", "extract_structure", f"Extracted structure for {structure_data.get('tag_name', 'unknown')} element")
return structure_data
elif isinstance(structure_data, list):
result = self._convert_nodriver_result(structure_data)
debug_logger.log_info("element_cloner", "extract_structure", f"Extracted structure for {result.get('tag_name', 'unknown')} element")
return result
else:
debug_logger.log_warning("element_cloner", "extract_structure", f"Got unexpected type: {type(structure_data)}")
return {"error": f"Unexpected return type: {type(structure_data)}", "raw_data": str(structure_data)}
except Exception as e:
debug_logger.log_error("element_cloner", "extract_structure", e)
return {"error": str(e)}
async def extract_element_events(
self,
tab,
element=None,
selector: str = None,
include_inline: bool = True,
include_listeners: bool = True,
include_framework: bool = True,
analyze_handlers: bool = True
) -> Dict[str, Any]:
"""
Extract complete event listener and JavaScript handler information.
Args:
tab (Any): Browser tab instance
element (Any): Element object or None to use selector
selector (str): CSS selector if element is None
include_inline (bool): Include inline event handlers (onclick, etc.)
include_listeners (bool): Include addEventListener attached handlers
include_framework (bool): Include framework-specific handlers (React, Vue, etc.)
analyze_handlers (bool): Analyze handler functions for details
Returns:
Dict[str, Any]: Dict with event data
"""
try:
if not selector:
return {"error": "Selector is required"}
options = {
'include_inline': include_inline,
'include_listeners': include_listeners,
'include_framework': include_framework,
'analyze_handlers': analyze_handlers
}
js_code = self._load_js_file('extract_events.js', selector, options)
event_data = await tab.evaluate(js_code)
if hasattr(event_data, 'exception_details'):
return {"error": f"JavaScript error: {event_data.exception_details}"}
elif isinstance(event_data, dict):
debug_logger.log_info("element_cloner", "extract_events", f"Extracted events for element")
return event_data
elif isinstance(event_data, list):
result = self._convert_nodriver_result(event_data)
debug_logger.log_info("element_cloner", "extract_events", f"Extracted events for element")
return result
else:
debug_logger.log_warning("element_cloner", "extract_events", f"Got unexpected type: {type(event_data)}")
return {"error": f"Unexpected return type: {type(event_data)}", "raw_data": str(event_data)}
except Exception as e:
debug_logger.log_error("element_cloner", "extract_events", e)
return {"error": str(e)}
async def extract_element_animations(
self,
tab,
element=None,
selector: str = None,
include_css_animations: bool = True,
include_transitions: bool = True,
include_transforms: bool = True,
analyze_keyframes: bool = True
) -> Dict[str, Any]:
"""
Extract CSS animations, transitions, and transforms.
Args:
tab (Any): Browser tab instance
element (Any): Element object or None to use selector
selector (str): CSS selector if element is None
include_css_animations (bool): Include CSS @keyframes animations
include_transitions (bool): Include CSS transitions
include_transforms (bool): Include CSS transforms
analyze_keyframes (bool): Analyze keyframe rules
Returns:
Dict[str, Any]: Dict with animation data
"""
try:
if not selector:
return {"error": "Selector is required"}
options = {
'include_css_animations': include_css_animations,
'include_transitions': include_transitions,
'include_transforms': include_transforms,
'analyze_keyframes': analyze_keyframes
}
js_code = self._load_js_file('extract_animations.js', selector, options)
animation_data = await tab.evaluate(js_code)
if hasattr(animation_data, 'exception_details'):
return {"error": f"JavaScript error: {animation_data.exception_details}"}
elif isinstance(animation_data, dict):
debug_logger.log_info("element_cloner", "extract_animations", f"Extracted animations for element")
return animation_data
elif isinstance(animation_data, list):
result = self._convert_nodriver_result(animation_data)
debug_logger.log_info("element_cloner", "extract_animations", f"Extracted animations for element")
return result
else:
debug_logger.log_warning("element_cloner", "extract_animations", f"Got unexpected type: {type(animation_data)}")
return {"error": f"Unexpected return type: {type(animation_data)}", "raw_data": str(animation_data)}
except Exception as e:
debug_logger.log_error("element_cloner", "extract_animations", e)
return {"error": str(e)}
async def extract_element_assets(
self,
tab,
element=None,
selector: str = None,
include_images: bool = True,
include_backgrounds: bool = True,
include_fonts: bool = True,
fetch_external: bool = False
) -> Dict[str, Any]:
"""
Extract all assets related to an element (images, fonts, etc.).
Args:
tab (Any): Browser tab instance
element (Any): Element object or None to use selector
selector (str): CSS selector if element is None
include_images (bool): Include img src and related images
include_backgrounds (bool): Include background images
include_fonts (bool): Include font information
fetch_external (bool): Whether to fetch external assets for analysis
Returns:
Dict[str, Any]: Dict with asset data
"""
try:
if not selector:
return {"error": "Selector is required"}
js_dir = Path(__file__).parent / "js"
js_file = js_dir / "extract_assets.js"
if not js_file.exists():
return {"error": f"JavaScript file not found: {js_file}"}
with open(js_file, 'r', encoding='utf-8') as f:
js_code = f.read()
js_code = js_code.replace('$SELECTOR', selector)
js_code = js_code.replace('$INCLUDE_IMAGES', 'true' if include_images else 'false')
js_code = js_code.replace('$INCLUDE_BACKGROUNDS', 'true' if include_backgrounds else 'false')
js_code = js_code.replace('$INCLUDE_FONTS', 'true' if include_fonts else 'false')
js_code = js_code.replace('$FETCH_EXTERNAL', 'true' if fetch_external else 'false')
asset_data = await tab.evaluate(js_code)
if hasattr(asset_data, 'exception_details'):
return {"error": f"JavaScript error: {asset_data.exception_details}"}
elif isinstance(asset_data, dict):
pass
elif isinstance(asset_data, list):
# Convert nodriver's array format back to dict
asset_data = self._convert_nodriver_result(asset_data)
else:
debug_logger.log_warning("element_cloner", "extract_assets", f"Got unexpected type: {type(asset_data)}")
return {"error": f"Unexpected return type: {type(asset_data)}", "raw_data": str(asset_data)}
if fetch_external and isinstance(asset_data, dict):
asset_data['external_assets'] = {}
for bg_img in asset_data.get('background_images', []):
try:
url = bg_img.get('url', '')
if url.startswith('http'):
response = requests.get(url, timeout=5)
asset_data['external_assets'][url] = {
'content_type': response.headers.get('content-type'),
'size': len(response.content),
'status': response.status_code
}
except Exception as e:
debug_logger.log_warning("element_cloner", "extract_assets", f"Could not fetch asset {url}: {e}")
debug_logger.log_info("element_cloner", "extract_assets", f"Extracted assets for element")
return asset_data
except Exception as e:
debug_logger.log_error("element_cloner", "extract_assets", e)
return {"error": str(e)}
async def extract_related_files(
self,
tab,
element=None,
selector: str = None,
analyze_css: bool = True,
analyze_js: bool = True,
follow_imports: bool = False,
max_depth: int = 2
) -> Dict[str, Any]:
"""
Discover and analyze related CSS/JS files for context.
Args:
tab (Any): Browser tab instance
element (Any): Element object or None to use selector
selector (str): CSS selector if element is None
analyze_css (bool): Analyze linked CSS files
analyze_js (bool): Analyze linked JS files
follow_imports (bool): Follow @import and module imports
max_depth (int): Maximum depth for following imports
Returns:
Dict[str, Any]: Dict with related file data
"""
try:
js_dir = Path(__file__).parent / "js"
js_file = js_dir / "extract_related_files.js"
if not js_file.exists():
return {"error": f"JavaScript file not found: {js_file}"}
with open(js_file, 'r', encoding='utf-8') as f:
js_code = f.read()
js_code = js_code.replace('$ANALYZE_CSS', 'true' if analyze_css else 'false')
js_code = js_code.replace('$ANALYZE_JS', 'true' if analyze_js else 'false')
js_code = js_code.replace('$FOLLOW_IMPORTS', 'true' if follow_imports else 'false')
js_code = js_code.replace('$MAX_DEPTH', str(max_depth))
file_data = await tab.evaluate(js_code)
if hasattr(file_data, 'exception_details'):
return {"error": f"JavaScript error: {file_data.exception_details}"}
elif isinstance(file_data, dict):
pass
elif isinstance(file_data, list):
file_data = self._convert_nodriver_result(file_data)
else:
debug_logger.log_warning("element_cloner", "extract_related_files", f"Got unexpected type: {type(file_data)}")
return {"error": f"Unexpected return type: {type(file_data)}", "raw_data": str(file_data)}
if follow_imports and max_depth > 0 and isinstance(file_data, dict):
await self._fetch_and_analyze_files(file_data, tab.url, max_depth)
debug_logger.log_info("element_cloner", "extract_related_files", f"Found related files")
return file_data
except Exception as e:
debug_logger.log_error("element_cloner", "extract_related_files", e)
return {"error": str(e)}
async def _fetch_and_analyze_files(self, file_data: Dict, base_url: str, max_depth: int) -> None:
"""
Fetch and analyze external CSS/JS files for additional context.
Args:
file_data (Dict): Data structure containing file info
base_url (str): Base URL for resolving relative paths
max_depth (int): Maximum depth for following imports
Returns:
None
"""
for stylesheet in file_data['stylesheets']:
if stylesheet.get('href') and stylesheet['href'] not in self.extracted_files:
try:
response = requests.get(stylesheet['href'], timeout=10)
if response.status_code == 200:
content = response.text
self.extracted_files[stylesheet['href']] = content
imports = re.findall(r'@import\s+["\']([^"\']+)["\']', content)
stylesheet['imports'] = []
for imp in imports:
absolute_url = urljoin(stylesheet['href'], imp)
stylesheet['imports'].append(absolute_url)
css_vars = re.findall(r'--[\w-]+:\s*[^;]+', content)
stylesheet['custom_properties'] = css_vars
except Exception as e:
debug_logger.log_warning("element_cloner", "fetch_css", f"Could not fetch CSS file {stylesheet.get('href')}: {e}")
for script in file_data['scripts']:
if script.get('src') and script['src'] not in self.extracted_files:
try:
response = requests.get(script['src'], timeout=10)
if response.status_code == 200:
content = response.text
self.extracted_files[script['src']] = content
script['detected_frameworks'] = []
for framework, patterns in self.framework_patterns.items():
for pattern in patterns:
if re.search(pattern, content, re.IGNORECASE):
if framework not in script['detected_frameworks']:
script['detected_frameworks'].append(framework)
imports = re.findall(r'import.*from\s+["\']([^"\']+)["\']', content)
script['module_imports'] = imports
except Exception as e:
debug_logger.log_warning("element_cloner", "fetch_js", f"Could not fetch JS file {script.get('src')}: {e}")
async def clone_element_complete(
self,
tab,
element=None,
selector: str = None,
extraction_options: Dict[str, Any] = None
) -> Dict[str, Any]:
"""
Master function that extracts all element data using specialized functions.
Args:
tab (Any): Browser tab instance
element (Any): Element object or None to use selector
selector (str): CSS selector if element is None
extraction_options (Dict[str, Any]): Dict specifying what to extract and options for each
Example: {
'styles': {'include_computed': True, 'include_pseudo': True},
'structure': {'include_children': True, 'max_depth': 2},
'events': {'include_framework': True, 'analyze_handlers': True},
'animations': {'analyze_keyframes': True},
'assets': {'fetch_external': True},
'related_files': {'follow_imports': True, 'max_depth': 1}
}
Returns:
Dict[str, Any]: Complete element clone data
"""
try:
default_options = {
'styles': {'include_computed': True, 'include_css_rules': True, 'include_pseudo': True},
'structure': {'include_children': False, 'include_attributes': True},
'events': {'include_framework': True, 'analyze_handlers': False},
'animations': {'analyze_keyframes': True},
'assets': {'fetch_external': False},
'related_files': {'follow_imports': False}
}
if extraction_options:
for key, value in extraction_options.items():
if key in default_options:
default_options[key].update(value)
else:
default_options[key] = value
if element is None and selector:
element = await tab.select(selector)
if not element:
return {"error": "Element not found"}
result = {
"url": tab.url,
"timestamp": asyncio.get_event_loop().time(),
"selector": selector,
"extraction_options": default_options
}
tasks = []
if 'styles' in default_options:
tasks.append(('styles', self.extract_element_styles(tab, element, **default_options['styles'])))
if 'structure' in default_options:
tasks.append(('structure', self.extract_element_structure(tab, element, **default_options['structure'])))
if 'events' in default_options:
tasks.append(('events', self.extract_element_events(tab, element, **default_options['events'])))
if 'animations' in default_options:
tasks.append(('animations', self.extract_element_animations(tab, element, **default_options['animations'])))
if 'assets' in default_options:
tasks.append(('assets', self.extract_element_assets(tab, element, **default_options['assets'])))
if 'related_files' in default_options:
tasks.append(('related_files', self.extract_related_files(tab, **default_options['related_files'])))
results = await asyncio.gather(*[task[1] for task in tasks], return_exceptions=True)
for i, (name, _) in enumerate(tasks):
if isinstance(results[i], Exception):
result[name] = {"error": str(results[i])}
else:
result[name] = results[i]
debug_logger.log_info("element_cloner", "clone_complete", f"Complete element clone extracted with {len(tasks)} data types")
return result
except Exception as e:
debug_logger.log_error("element_cloner", "clone_complete", e)
return {"error": str(e)}
async def extract_element_styles_cdp(
self,
tab,
element=None,
selector: str = None,
include_computed: bool = True,
include_css_rules: bool = True,
include_pseudo: bool = True,
include_inheritance: bool = False
) -> Dict[str, Any]:
"""
Extract complete styling information using direct CDP calls (no JavaScript evaluation).
This prevents hanging issues by using nodriver's native CDP methods.
Args:
tab (Any): Browser tab instance
element (Any): Element object or None to use selector
selector (str): CSS selector if element is None
include_computed (bool): Include computed styles
include_css_rules (bool): Include matching CSS rules
include_pseudo (bool): Include pseudo-element styles
include_inheritance (bool): Include style inheritance chain
Returns:
Dict[str, Any]: Dict with styling data
"""
try:
import nodriver.cdp as cdp
await tab.send(cdp.dom.enable())
await tab.send(cdp.css.enable())
if element is None and selector:
element = await tab.select(selector)
if not element:
return {"error": "Element not found"}
if hasattr(element, 'node_id'):
node_id = element.node_id
elif hasattr(element, 'backend_node_id'):
node_info = await tab.send(cdp.dom.describe_node(backend_node_id=element.backend_node_id))
node_id = node_info.node.node_id
else:
return {"error": "Could not get node ID from element"}
result = {"method": "cdp_direct"}
if include_computed:
debug_logger.log_info("element_cloner", "extract_styles_cdp", "Getting computed styles via CDP")
computed_styles_list = await tab.send(cdp.css.get_computed_style_for_node(node_id))
result["computed_styles"] = {prop.name: prop.value for prop in computed_styles_list}
if include_css_rules:
debug_logger.log_info("element_cloner", "extract_styles_cdp", "Getting matched styles via CDP")
matched_styles = await tab.send(cdp.css.get_matched_styles_for_node(node_id))
# Extract CSS rules from matched styles
result["css_rules"] = []
if matched_styles[2]: # matchedCSSRules
for rule_match in matched_styles[2]:
if rule_match.rule and rule_match.rule.style:
result["css_rules"].append({
"selector": rule_match.rule.selector_list.text if rule_match.rule.selector_list else "unknown",
"css_text": rule_match.rule.style.css_text or "",
"source": rule_match.rule.origin.value if rule_match.rule.origin else "unknown"
})
# Add inline styles if present
if matched_styles[0]: # inlineStyle
result["inline_style"] = {
"css_text": matched_styles[0].css_text or "",
"properties": len(matched_styles[0].css_properties) if matched_styles[0].css_properties else 0
}
# Add attribute styles if present
if matched_styles[1]: # attributesStyle
result["attributes_style"] = {
"css_text": matched_styles[1].css_text or "",
"properties": len(matched_styles[1].css_properties) if matched_styles[1].css_properties else 0
}
# Handle pseudo elements (if available in matched_styles)
if include_pseudo and len(matched_styles) > 3 and matched_styles[3]:
result["pseudo_elements"] = {}
for pseudo_match in matched_styles[3]:
if pseudo_match.pseudo_type:
result["pseudo_elements"][pseudo_match.pseudo_type.value] = {
"matches": len(pseudo_match.matches) if pseudo_match.matches else 0
}
# Handle inheritance (if available in matched_styles)
if include_inheritance and len(matched_styles) > 4 and matched_styles[4]:
result["inheritance_chain"] = []
for inherited_entry in matched_styles[4]:
if inherited_entry.inline_style:
result["inheritance_chain"].append({
"inline_css": inherited_entry.inline_style.css_text or "",
"properties": len(inherited_entry.inline_style.css_properties) if inherited_entry.inline_style.css_properties else 0
})
debug_logger.log_info("element_cloner", "extract_styles_cdp", f"CDP extraction completed with {len(result.get('css_rules', []))} CSS rules")
return result
except Exception as e:
debug_logger.log_error("element_cloner", "extract_styles_cdp", e)
return {"error": f"CDP extraction failed: {str(e)}"}
element_cloner = ElementCloner()
```
--------------------------------------------------------------------------------
/src/cdp_function_executor.py:
--------------------------------------------------------------------------------
```python
"""
CDP Function Executor - Direct JavaScript function execution via Chrome DevTools Protocol
This module provides comprehensive function execution capabilities using nodriver's CDP access:
1. Direct CDP command execution
2. JavaScript function discovery and execution
3. Dynamic script injection and execution
4. Python-JavaScript bridge functionality
"""
import asyncio
import json
import uuid
import inspect
from typing import Dict, List, Any, Optional, Callable, Union
from datetime import datetime
import nodriver as uc
from nodriver import Tab
from debug_logger import debug_logger
class ExecutionContext:
"""Represents a JavaScript execution context."""
def __init__(self, id: str, name: str, origin: str, unique_id: str, aux_data: dict = None):
"""
Args:
id (str): Execution context identifier.
name (str): Name of the context.
origin (str): Origin URL of the context.
unique_id (str): Unique identifier for the context.
aux_data (dict, optional): Auxiliary data for the context.
"""
self.id = id
self.name = name
self.origin = origin
self.unique_id = unique_id
self.aux_data = aux_data or {}
class FunctionInfo:
"""Information about a discovered JavaScript function."""
def __init__(self, name: str, path: str, signature: str = None, description: str = None):
"""
Args:
name (str): Function name.
path (str): Path to the function (e.g., "window.document.getElementById").
signature (str, optional): Function signature.
description (str, optional): Description of the function.
"""
self.name = name
self.path = path
self.signature = signature
self.description = description
class FunctionCall:
"""Represents a function call to be executed."""
def __init__(self, function_path: str, args: List[Any] = None, context_id: str = None):
"""
Args:
function_path (str): Path to the function.
args (List[Any], optional): Arguments to pass to the function.
context_id (str, optional): Execution context identifier.
"""
self.function_path = function_path
self.args = args or []
self.context_id = context_id
class CDPFunctionExecutor:
"""Main class for CDP-based function execution."""
def __init__(self):
"""
Initializes the CDPFunctionExecutor instance.
"""
self._python_bindings: Dict[str, Callable] = {}
self._persistent_functions: Dict[str, Dict[str, str]] = {}
async def enable_runtime(self, tab: Tab) -> bool:
"""
Enables CDP Runtime domain for a tab.
Args:
tab (Tab): The browser tab.
Returns:
bool: True if enabled, False otherwise.
"""
try:
await tab.send(uc.cdp.runtime.enable())
debug_logger.log_info("cdp_function_executor", "enable_runtime", f"Runtime enabled for tab")
return True
except Exception as e:
debug_logger.log_error("cdp_function_executor", "enable_runtime", e)
return False
async def list_cdp_commands(self) -> List[str]:
"""
Lists all available CDP Runtime commands.
Returns:
List[str]: List of command names.
"""
commands = [
"evaluate", "callFunctionOn", "addBinding", "removeBinding",
"compileScript", "runScript", "awaitPromise", "getProperties",
"getExceptionDetails", "globalLexicalScopeNames", "queryObjects",
"releaseObject", "releaseObjectGroup", "terminateExecution",
"setAsyncCallStackDepth", "setCustomObjectFormatterEnabled",
"setMaxCallStackSizeToCapture", "runIfWaitingForDebugger",
"discardConsoleEntries", "getHeapUsage", "getIsolateId"
]
return commands
async def execute_cdp_command(self, tab: Tab, command: str, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Executes any CDP Runtime command with given parameters.
Args:
tab (Tab): The browser tab.
command (str): CDP command name.
params (Dict[str, Any]): Parameters for the command.
Returns:
Dict[str, Any]: Result of the command execution.
"""
try:
await self.enable_runtime(tab)
cdp_method = getattr(uc.cdp.runtime, command, None)
if not cdp_method:
raise ValueError(f"Unknown CDP command: {command}")
result = await tab.send(cdp_method(**params))
debug_logger.log_info("cdp_function_executor", "execute_cdp_command", f"Executed {command} with params: {params}")
return {
"success": True,
"result": result,
"command": command,
"params": params
}
except Exception as e:
debug_logger.log_error("cdp_function_executor", "execute_cdp_command", e)
return {
"success": False,
"error": str(e),
"command": command,
"params": params
}
async def get_execution_contexts(self, tab: Tab) -> List[ExecutionContext]:
"""
Gets all available execution contexts.
Args:
tab (Tab): The browser tab.
Returns:
List[ExecutionContext]: List of execution contexts.
"""
try:
await self.enable_runtime(tab)
script = """
(function() {
return {
location: window.location.href,
title: document.title,
readyState: document.readyState,
contexts: [{
name: 'main',
origin: window.location.origin,
url: window.location.href
}]
};
})()
"""
result = await tab.send(uc.cdp.runtime.evaluate(
expression=script,
return_by_value=True,
await_promise=True
))
if result and result[0] and result[0].value:
context_data = result[0].value
contexts = []
for i, ctx in enumerate(context_data.get('contexts', [])):
contexts.append(ExecutionContext(
id=str(i),
name=ctx['name'],
origin=ctx['origin'],
unique_id=f"{ctx['origin']}_{i}"
))
return contexts
return []
except Exception as e:
debug_logger.log_error("cdp_function_executor", "get_execution_contexts", e)
return []
async def discover_global_functions(self, tab: Tab, context_id: str = None) -> List[FunctionInfo]:
"""
Discovers all global JavaScript functions.
Args:
tab (Tab): The browser tab.
context_id (str, optional): Execution context identifier.
Returns:
List[FunctionInfo]: List of discovered functions.
"""
try:
await self.enable_runtime(tab)
discovery_script = """
(function() {
const functions = [];
function isFunction(obj) {
return typeof obj === 'function';
}
function discoverFunctions(obj, path = '', depth = 0) {
if (depth > 3) return;
try {
for (const key of Object.getOwnPropertyNames(obj)) {
if (key.startsWith('_') || key === 'constructor') continue;
try {
const value = obj[key];
const fullPath = path ? `${path}.${key}` : key;
if (isFunction(value)) {
functions.push({
name: key,
path: fullPath,
signature: value.toString().split('{')[0].trim(),
description: `Function at ${fullPath}`
});
} else if (typeof value === 'object' && value !== null && depth < 2) {
discoverFunctions(value, fullPath, depth + 1);
}
} catch (e) {
}
}
} catch (e) {
}
}
discoverFunctions(window, 'window');
discoverFunctions(document, 'document');
discoverFunctions(console, 'console');
const globalFuncs = ['setTimeout', 'setInterval', 'clearTimeout', 'clearInterval',
'fetch', 'alert', 'confirm', 'prompt', 'parseInt', 'parseFloat'];
for (const funcName of globalFuncs) {
if (typeof window[funcName] === 'function') {
functions.push({
name: funcName,
path: funcName,
signature: window[funcName].toString().split('{')[0].trim(),
description: `Global function ${funcName}`
});
}
}
return functions;
})()
"""
result = await tab.send(uc.cdp.runtime.evaluate(
expression=discovery_script,
return_by_value=True,
await_promise=True
))
if result and result[0] and result[0].value:
functions_data = result[0].value
functions = []
for func_data in functions_data:
functions.append(FunctionInfo(
name=func_data['name'],
path=func_data['path'],
signature=func_data.get('signature'),
description=func_data.get('description')
))
return functions
return []
except Exception as e:
debug_logger.log_error("cdp_function_executor", "discover_global_functions", e)
return []
async def discover_object_methods(self, tab: Tab, object_path: str) -> List[FunctionInfo]:
"""
Discovers methods of a specific JavaScript object.
Args:
tab (Tab): The browser tab.
object_path (str): Path to the JavaScript object.
Returns:
List[FunctionInfo]: List of discovered methods.
"""
try:
await self.enable_runtime(tab)
object_result = await tab.send(uc.cdp.runtime.evaluate(
expression=object_path,
return_by_value=False
))
if not object_result or not object_result[0] or not object_result[0].object_id:
debug_logger.log_warning("cdp_function_executor", "discover_object_methods", f"Could not get object reference for {object_path}")
return []
object_id = object_result[0].object_id
properties_result = await tab.send(uc.cdp.runtime.get_properties(
object_id=object_id,
own_properties=False,
accessor_properties_only=False
))
if not properties_result or not properties_result[0]:
debug_logger.log_warning("cdp_function_executor", "discover_object_methods", f"No properties returned for {object_path}")
return []
properties = properties_result[0]
methods = []
for prop in properties:
try:
if prop.value and prop.value.type_ == "function":
methods.append(FunctionInfo(
name=prop.name,
path=f'{object_path}.{prop.name}',
signature=prop.value.description or f"function {prop.name}()",
description=f"Method {prop.name} of {object_path}"
))
except Exception as e:
debug_logger.log_warning("cdp_function_executor", "discover_object_methods", f"Error processing property {prop.name}: {e}")
continue
debug_logger.log_info("cdp_function_executor", "discover_object_methods", f"Found {len(methods)} methods for {object_path}")
return methods
except Exception as e:
debug_logger.log_error("cdp_function_executor", "discover_object_methods", e)
return []
async def call_discovered_function(self, tab: Tab, function_path: str, args: List[Any]) -> Dict[str, Any]:
"""
Calls a discovered JavaScript function with arguments.
Args:
tab (Tab): The browser tab.
function_path (str): Path to the function.
args (List[Any]): Arguments to pass.
Returns:
Dict[str, Any]: Result of the function call.
"""
try:
await self.enable_runtime(tab)
js_args = json.dumps(args) if args else '[]'
call_script = f"""
(function() {{
try {{
const pathParts = '{function_path}'.split('.');
let context = window;
let func = window;
for (let i = 0; i < pathParts.length; i++) {{
if (i === pathParts.length - 1) {{
func = context[pathParts[i]];
}} else {{
context = context[pathParts[i]];
func = context;
}}
}}
if (typeof func !== 'function') {{
throw new Error('Not a function: {function_path}');
}}
const args = {js_args};
const result = func.apply(context, args);
return {{
success: true,
result: result,
function_path: '{function_path}',
args: args
}};
}} catch (error) {{
return {{
success: false,
error: error.message,
function_path: '{function_path}',
args: {js_args}
}};
}}
}})()
"""
result = await tab.send(uc.cdp.runtime.evaluate(
expression=call_script,
return_by_value=True,
await_promise=True
))
if result and result[0] and result[0].value:
return result[0].value
elif result and result[1]:
return {
"success": False,
"error": f"Runtime exception: {result[1].text}",
"function_path": function_path,
"args": args
}
return {
"success": False,
"error": "No result returned",
"function_path": function_path,
"args": args
}
except Exception as e:
debug_logger.log_error("cdp_function_executor", "call_discovered_function", e)
return {
"success": False,
"error": str(e),
"function_path": function_path,
"args": args
}
async def inspect_function_signature(self, tab: Tab, function_path: str) -> Dict[str, Any]:
"""
Inspects a function's signature and details.
Args:
tab (Tab): The browser tab.
function_path (str): Path to the function.
Returns:
Dict[str, Any]: Signature and details of the function.
"""
try:
await self.enable_runtime(tab)
inspect_script = f"""
(function() {{
try {{
const func = {function_path};
if (typeof func !== 'function') {{
return {{
success: false,
error: 'Not a function: {function_path}'
}};
}}
return {{
success: true,
name: func.name || 'anonymous',
path: '{function_path}',
signature: func.toString(),
length: func.length,
is_async: func.constructor.name === 'AsyncFunction',
is_generator: func.constructor.name === 'GeneratorFunction'
}};
}} catch (error) {{
return {{
success: false,
error: error.message
}};
}}
}})()
"""
result = await tab.send(uc.cdp.runtime.evaluate(
expression=inspect_script,
return_by_value=True,
await_promise=True
))
if result and result[0] and result[0].value:
return result[0].value
return {"success": False, "error": "No result returned"}
except Exception as e:
debug_logger.log_error("cdp_function_executor", "inspect_function_signature", e)
return {"success": False, "error": str(e)}
async def inject_and_execute_script(self, tab: Tab, script_code: str, context_id: str = None) -> Dict[str, Any]:
"""
Injects and executes custom JavaScript code.
Args:
tab (Tab): The browser tab.
script_code (str): JavaScript code to execute.
context_id (str, optional): Execution context identifier.
Returns:
Dict[str, Any]: Result of script execution.
"""
try:
await self.enable_runtime(tab)
wrapped_script = f"""
(function() {{
try {{
const result = (function() {{
{script_code}
}})();
return {{
success: true,
result: result,
executed_at: new Date().toISOString()
}};
}} catch (error) {{
return {{
success: false,
error: error.message,
stack: error.stack,
executed_at: new Date().toISOString()
}};
}}
}})()
"""
result = await tab.send(uc.cdp.runtime.evaluate(
expression=wrapped_script,
return_by_value=True,
await_promise=True,
allow_unsafe_eval_blocked_by_csp=True
))
if result and result[0] and result[0].value:
return result[0].value
elif result and result[1]:
return {
"success": False,
"error": f"Runtime exception: {result[1].text}",
"line_number": result[1].line_number,
"column_number": result[1].column_number
}
return {"success": False, "error": "No result returned"}
except Exception as e:
debug_logger.log_error("cdp_function_executor", "inject_and_execute_script", e)
return {"success": False, "error": str(e)}
async def create_persistent_function(self, tab: Tab, function_name: str, function_code: str, instance_id: str) -> Dict[str, Any]:
"""
Creates a persistent JavaScript function that survives page reloads.
Args:
tab (Tab): The browser tab.
function_name (str): Name of the function.
function_code (str): JavaScript code for the function.
instance_id (str): Instance identifier.
Returns:
Dict[str, Any]: Result of function creation.
"""
try:
await self.enable_runtime(tab)
if instance_id not in self._persistent_functions:
self._persistent_functions[instance_id] = {}
self._persistent_functions[instance_id][function_name] = function_code
create_script = f"""
(function() {{
try {{
window.{function_name} = {function_code};
return {{
success: true,
function_name: '{function_name}',
created_at: new Date().toISOString(),
available_as: 'window.{function_name}'
}};
}} catch (error) {{
return {{
success: false,
error: error.message,
function_name: '{function_name}'
}};
}}
}})()
"""
result = await tab.send(uc.cdp.runtime.evaluate(
expression=create_script,
return_by_value=True,
await_promise=True
))
if result and result[0] and result[0].value:
return result[0].value
return {"success": False, "error": "Failed to create function"}
except Exception as e:
debug_logger.log_error("cdp_function_executor", "create_persistent_function", e)
return {"success": False, "error": str(e)}
async def execute_function_sequence(self, tab: Tab, function_calls: List[FunctionCall]) -> List[Dict[str, Any]]:
"""
Executes a sequence of function calls.
Args:
tab (Tab): The browser tab.
function_calls (List[FunctionCall]): List of function calls to execute.
Returns:
List[Dict[str, Any]]: Results of each function call.
"""
results = []
for i, func_call in enumerate(function_calls):
try:
debug_logger.log_info("cdp_function_executor", "execute_function_sequence", f"Executing call {i+1}/{len(function_calls)}: {func_call.function_path}")
result = await self.call_discovered_function(
tab,
func_call.function_path,
func_call.args
)
results.append({
"sequence_index": i,
"function_call": {
"function_path": func_call.function_path,
"args": func_call.args,
"context_id": func_call.context_id
},
"result": result
})
except Exception as e:
debug_logger.log_error("cdp_function_executor", "execute_function_sequence", e)
results.append({
"sequence_index": i,
"function_call": {
"function_path": func_call.function_path,
"args": func_call.args,
"context_id": func_call.context_id
},
"result": {
"success": False,
"error": str(e)
}
})
return results
async def create_python_binding(self, tab: Tab, binding_name: str, python_function: Callable) -> Dict[str, Any]:
"""
Creates a binding that allows JavaScript to call Python functions.
Args:
tab (Tab): The browser tab.
binding_name (str): Name of the binding.
python_function (Callable): Python function to bind.
Returns:
Dict[str, Any]: Result of binding creation.
"""
try:
await self.enable_runtime(tab)
self._python_bindings[binding_name] = python_function
await tab.send(uc.cdp.runtime.add_binding(name=binding_name))
wrapper_script = f"""
(function() {{
if (!window.{binding_name}) {{
window.{binding_name} = function(...args) {{
return new Promise((resolve, reject) => {{
const callId = Math.random().toString(36).substr(2, 9);
window.addEventListener(`{binding_name}_response_${{callId}}`, function(event) {{
if (event.detail.success) {{
resolve(event.detail.result);
}} else {{
reject(new Error(event.detail.error));
}}
}}, {{ once: true }});
window.chrome.runtime.sendMessage({{
binding: '{binding_name}',
args: args,
callId: callId
}});
}});
}};
}}
return {{
success: true,
binding_name: '{binding_name}',
available_as: 'window.{binding_name}'
}};
}})()
"""
result = await tab.send(uc.cdp.runtime.evaluate(
expression=wrapper_script,
return_by_value=True,
await_promise=True
))
if result and result[0] and result[0].value:
return result[0].value
return {"success": False, "error": "Failed to create binding"}
except Exception as e:
debug_logger.log_error("cdp_function_executor", "create_python_binding", e)
return {"success": False, "error": str(e)}
async def execute_python_in_browser(self, tab: Tab, python_code: str) -> Dict[str, Any]:
"""
Executes Python code by translating it to JavaScript with timeout protection.
Args:
tab (Tab): The browser tab.
python_code (str): Python code to execute.
Returns:
Dict[str, Any]: Result of execution.
"""
try:
js_code = self._translate_python_to_js(python_code)
debug_logger.log_info("cdp_function_executor", "execute_python_in_browser", f"Translated JS: {js_code}")
import asyncio
result = await asyncio.wait_for(
self.inject_and_execute_script(tab, js_code),
timeout=10.0
)
return result
except asyncio.TimeoutError:
return {"success": False, "error": "Python execution timeout - code may have infinite loop or syntax error"}
except Exception as e:
debug_logger.log_error("cdp_function_executor", "execute_python_in_browser", e)
return {"success": False, "error": str(e)}
def _translate_python_to_js(self, python_code: str) -> str:
"""
Professional Python to JavaScript translation using py2js library.
Args:
python_code (str): Python code to translate.
Returns:
str: Translated JavaScript code.
"""
try:
import py2js
js_code = py2js.convert(python_code)
debug_logger.log_info("cdp_function_executor", "_translate_python_to_js", f"py2js generated: {js_code}")
lines = python_code.strip().split('\n')
last_line = lines[-1].strip() if lines else ""
if (last_line and
'=' not in last_line and
not last_line.startswith(('def ', 'class ', 'if ', 'for ', 'while ', 'try:', 'with ', 'import ', 'from '))):
wrapped_code = f"(() => {{ {js_code}; return {last_line}; }})()"
return wrapped_code
else:
return f"(() => {{ {js_code}; }})()"
except ImportError:
debug_logger.log_warning("cdp_function_executor", "_translate_python_to_js", "py2js not available, using fallback")
return self._fallback_python_to_js(python_code)
except Exception as e:
debug_logger.log_error("cdp_function_executor", "_translate_python_to_js", e, {"python_code": python_code})
return self._fallback_python_to_js(python_code)
def _fallback_python_to_js(self, python_code: str) -> str:
"""
Fallback Python to JavaScript translation for basic cases.
Args:
python_code (str): Python code to translate.
Returns:
str: Basic translated JavaScript code.
"""
import re
lines = python_code.strip().split('\n')
js_lines = []
for line in lines:
js_line = line
replacements = {
"True": "true",
"False": "false",
"None": "null",
"print(": "console.log(",
".append(": ".push(",
}
for py_syntax, js_syntax in replacements.items():
js_line = js_line.replace(py_syntax, js_syntax)
if '=' in js_line and not js_line.strip().startswith('//'):
if re.match(r'^\s*[a-zA-Z_][a-zA-Z0-9_]*\s*=', js_line):
js_line = re.sub(r'^(\s*)([a-zA-Z_][a-zA-Z0-9_]*\s*=)', r'\1let \2', js_line)
js_lines.append(js_line)
js_code = ";\n".join(js_lines) + ";"
last_line = lines[-1].strip() if lines else ""
if last_line and '=' not in last_line and not last_line.endswith(':'):
js_code = js_code.rsplit(';', 2)[0] + f"; return {last_line};"
wrapped_code = f"(function() {{ {js_code} }})()"
return wrapped_code
async def call_python_from_js(self, binding_name: str, args: List[Any]) -> Dict[str, Any]:
"""
Handles JavaScript calls to Python functions.
Args:
binding_name (str): Name of the Python binding.
args (List[Any]): Arguments to pass to the Python function.
Returns:
Dict[str, Any]: Result of the Python function call.
"""
try:
if binding_name not in self._python_bindings:
return {"success": False, "error": f"Unknown binding: {binding_name}"}
python_function = self._python_bindings[binding_name]
if asyncio.iscoroutinefunction(python_function):
result = await python_function(*args)
else:
result = python_function(*args)
return {
"success": True,
"result": result,
"binding_name": binding_name,
"args": args
}
except Exception as e:
debug_logger.log_error("cdp_function_executor", "call_python_from_js", e)
return {
"success": False,
"error": str(e),
"binding_name": binding_name,
"args": args
}
async def get_function_executor_info(self, instance_id: str = None) -> Dict[str, Any]:
"""
Gets information about the function executor state.
Args:
instance_id (str, optional): Instance identifier.
Returns:
Dict[str, Any]: Information about the executor.
"""
return {
"python_bindings": list(self._python_bindings.keys()),
"persistent_functions": self._persistent_functions.get(instance_id, {}) if instance_id else self._persistent_functions,
"available_commands": await self.list_cdp_commands(),
"executor_version": "1.0.0",
"capabilities": [
"direct_cdp_execution",
"function_discovery",
"dynamic_script_injection",
"python_js_bridge"
]
}
```