Update to itb_screenshot and friends
This commit is contained in:
@@ -0,0 +1,162 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys
|
||||
import os
|
||||
import time
|
||||
import argparse
|
||||
from selenium import webdriver
|
||||
from selenium.webdriver.common.by import By
|
||||
from selenium.webdriver.support.ui import WebDriverWait
|
||||
from selenium.webdriver.support import expected_conditions as EC
|
||||
|
||||
class RefreshController:
|
||||
def __init__(self, debug_port: int, verbose: bool = False):
|
||||
"""Initialize connection to Chrome instance"""
|
||||
self.verbose = verbose
|
||||
options = webdriver.ChromeOptions()
|
||||
options.add_experimental_option("debuggerAddress", f"127.0.0.1:{debug_port}")
|
||||
self.driver = webdriver.Chrome(options=options)
|
||||
|
||||
# Load JavaScript files
|
||||
js_dir = os.path.join(os.path.dirname(__file__), "..", "js")
|
||||
self.js_code = ""
|
||||
with open(os.path.join(js_dir, "viewport.js")) as f:
|
||||
self.js_code = f.read()
|
||||
|
||||
def log(self, msg: str):
|
||||
"""Log message if verbose mode is enabled"""
|
||||
if self.verbose:
|
||||
print(f"DEBUG: {msg}", file=sys.stderr)
|
||||
|
||||
def get_page_state(self) -> dict:
|
||||
"""Get current page state for comparison"""
|
||||
return {
|
||||
'url': self.driver.current_url,
|
||||
'title': self.driver.title,
|
||||
'scroll_position': self.driver.execute_script(
|
||||
self.js_code + "; return getViewportInfo();"
|
||||
)
|
||||
}
|
||||
|
||||
def wait_for_page_change(self, original_state: dict, timeout: int = 30) -> bool:
|
||||
"""Wait for page to refresh and stabilize"""
|
||||
try:
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < timeout:
|
||||
current_state = self.get_page_state()
|
||||
|
||||
# Check if page has reloaded
|
||||
if (current_state['url'] == original_state['url'] and
|
||||
current_state['title'] == original_state['title']):
|
||||
|
||||
# Check if DOM has finished updating
|
||||
is_stable = self.driver.execute_script("""
|
||||
return (typeof window._lastMutation === 'undefined' ||
|
||||
Date.now() - window._lastMutation > 1000);
|
||||
""")
|
||||
|
||||
if is_stable:
|
||||
return True
|
||||
|
||||
time.sleep(0.1)
|
||||
|
||||
self.log("Timeout waiting for page to stabilize")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
self.log(f"Error waiting for page change: {str(e)}")
|
||||
return False
|
||||
|
||||
def setup_mutation_tracking(self):
|
||||
"""Setup mutation observer to track DOM changes"""
|
||||
self.driver.execute_script("""
|
||||
// Track last mutation time
|
||||
window._lastMutation = Date.now();
|
||||
|
||||
// Create mutation observer
|
||||
const observer = new MutationObserver(() => {
|
||||
window._lastMutation = Date.now();
|
||||
});
|
||||
|
||||
// Observe entire document for any changes
|
||||
observer.observe(document, {
|
||||
childList: true,
|
||||
subtree: true,
|
||||
attributes: true,
|
||||
characterData: true
|
||||
});
|
||||
""")
|
||||
|
||||
def refresh(self, hard: bool = False, timeout: int = 30) -> bool:
|
||||
"""
|
||||
Refresh the current page
|
||||
|
||||
Args:
|
||||
hard (bool): If True, bypass cache with Ctrl+F5
|
||||
timeout (int): Maximum time to wait for refresh in seconds
|
||||
|
||||
Returns:
|
||||
bool: True if refresh was successful
|
||||
"""
|
||||
try:
|
||||
# Get initial page state
|
||||
original_state = self.get_page_state()
|
||||
self.log(f"Original page state: {original_state}")
|
||||
|
||||
# Setup mutation tracking
|
||||
self.setup_mutation_tracking()
|
||||
|
||||
if hard:
|
||||
# Force reload bypassing cache
|
||||
self.log("Executing hard refresh (Ctrl+F5)")
|
||||
self.driver.execute_script("""
|
||||
window.location.reload(true);
|
||||
""")
|
||||
else:
|
||||
# Normal reload
|
||||
self.log("Executing normal refresh")
|
||||
self.driver.refresh()
|
||||
|
||||
# Wait for page to stabilize
|
||||
success = self.wait_for_page_change(original_state, timeout)
|
||||
|
||||
if success:
|
||||
# Restore scroll position
|
||||
self.driver.execute_script(
|
||||
"window.scrollTo(arguments[0], arguments[1]);",
|
||||
original_state['scroll_position']['scroll_x'],
|
||||
original_state['scroll_position']['scroll_y']
|
||||
)
|
||||
|
||||
# Final verification
|
||||
new_state = self.get_page_state()
|
||||
self.log(f"New page state: {new_state}")
|
||||
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
self.log(f"Refresh failed: {str(e)}")
|
||||
return False
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description='Refresh page in Chrome')
|
||||
parser.add_argument('debug_port', type=int, help='Chrome debug port')
|
||||
parser.add_argument('--hard', action='store_true', help='Force refresh bypassing cache')
|
||||
parser.add_argument('--timeout', type=int, default=30, help='Refresh timeout in seconds')
|
||||
parser.add_argument('-v', '--verbose', action='store_true', help='Enable debug output')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
controller = RefreshController(args.debug_port, args.verbose)
|
||||
success = controller.refresh(args.hard, args.timeout)
|
||||
sys.exit(0 if success else 1)
|
||||
|
||||
except Exception as e:
|
||||
if args.verbose:
|
||||
print(f"Fatal error: {str(e)}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user