File size: 4,391 Bytes
795f737 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | """DOM-first Playwright adapter with observation-scoped node handles.
Only use isolated browser contexts: this is research software, not a sandbox.
"""
from pathlib import Path
from uuid import uuid4
from urllib.parse import urlsplit
from .state import Element, State
from .authority import Rejected
DOM_SCRIPT = Path(__file__).with_name('dom.js').read_text(encoding='utf-8')
ELEMENT_SCRIPT = Path(__file__).with_name('element.js').read_text(encoding='utf-8')
def origin(url):
parsed = urlsplit(url)
return f"{parsed.scheme}://{parsed.netloc}" if parsed.netloc else parsed.scheme + ":"
class Browser:
def __init__(self, context, path_mode='traversal'):
if path_mode not in {'traversal', 'naive'}:
raise ValueError('invalid path mode')
self.path_mode = path_mode
self.context = context
self.page = context.new_page()
self.page.set_default_timeout(2000)
self.tabs = {"t0": self.page}
self._arrays = []
self._refs = {}
self._document = None
self._document_id = ""
self.revision = 0
self.state = None
def observe(self):
for arr in self._arrays:
try:
arr.dispose()
except Exception:
pass # Navigation already disposed the execution context.
self._arrays, self._refs = [], {}
try:
same = self.page.evaluate("old => old === document", self._document) if self._document else False
except Exception:
same = False
if not same:
if self._document:
try:
self._document.dispose()
except Exception:
pass
self._document = self.page.evaluate_handle("document")
self._document_id = uuid4().hex
self.revision += 1
elements = []
for frame_index, frame in enumerate(self.page.frames):
bundle = frame.evaluate_handle(DOM_SCRIPT)
path_arg = 'bundle.paths[i]' if self.path_mode == 'traversal' else 'undefined'
rows = bundle.evaluate(f"bundle => bundle.nodes.map((el, i) => ({ELEMENT_SCRIPT})(el, {path_arg}))")
arr = bundle.get_property('nodes')
bundle.dispose()
self._arrays.append(arr)
for index, row in enumerate(rows):
ref = f"e{len(elements)}"
row['path'] = f"{frame_index}:" + row['path']
element = Element(ref=ref, **row)
elements.append(element)
self._refs[ref] = (arr, index, element, frame_index)
self.state = State(self._document_id, self.revision, origin(self.page.url), tuple(elements))
return self.state
def resolve(self, ref):
if ref not in self._refs:
raise Rejected("UNKNOWN_REF")
arr, index, expected, frame_index = self._refs[ref]
try:
handle = arr.get_property(str(index)).as_element()
if not handle or not handle.evaluate("el => el.isConnected"):
if handle:
handle.dispose()
raise Rejected("DETACHED_TARGET")
row = handle.evaluate(ELEMENT_SCRIPT)
row['path'] = f"{frame_index}:" + row['path']
current = Element(ref=ref, **row)
if current != expected:
handle.dispose()
raise Rejected("CHANGED_TARGET")
if not current.visible or not current.enabled:
handle.dispose()
raise Rejected("INACTIVE_TARGET")
return handle, current
except Rejected:
raise
except Exception as exc:
raise Rejected("STALE_TARGET") from exc
def validate_document(self):
try:
if not self._document or not self.page.evaluate("old => old === document", self._document):
raise Rejected("CHANGED_DOCUMENT")
except Rejected:
raise
except Exception as exc:
raise Rejected("CHANGED_DOCUMENT") from exc
def sync_tabs(self):
for page in self.context.pages:
if page not in self.tabs.values():
self.tabs["t" + uuid4().hex[:12]] = page
self.tabs = {key: page for key, page in self.tabs.items() if not page.is_closed()}
return tuple(self.tabs)
|