| """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 |
| 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) |
|
|
