diff --git a/.gitignore b/.gitignore
index e6829408..8dd1ec66 100644
--- a/.gitignore
+++ b/.gitignore
@@ -179,3 +179,5 @@ tests/adata/visual/reference/reference_with_rerun.png
# AI
.claude
+
+# Self-healing snapshots
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 37ce270c..e9857833 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -23,6 +23,38 @@ types are active, preventing cross-driver method overwrites on shared Element/Pa
---
+## v4.0.0rc1
+
+*Release date: 2026-07-14*
+
+### Added
+- **Self-healing locators** — automatic recovery of broken element locators at runtime
+ - `mops/self_healing/` package with config, healer, locator generator, and snapshot storage
+ - `@healing` decorator — applied to action methods (`click`, `type_text`, `get_attribute`, `text`, etc.),
+ catches `NoSuchElementException`, heals the locator via similarity scoring, and retries the method
+ - `@healing_after_wait` decorator — applied to `wait_visibility` and `wait_availability`;
+ triggers healing after the wait condition times out
+ - `@no_healing` decorator — temporarily disables healing for specific methods (`is_displayed`, `wait_hidden`)
+ - `SelfHealingConfig` with `configure()` / `get_config()` for global configuration
+ - Snapshot capture on successful element lookup (tag, attributes, text, parent, up to 5 siblings)
+ - Candidate search — JS script collects all same-tag elements, scores each against the saved snapshot
+ - Similarity scoring with tunable per-attribute weights (`ScoringWeights`)
+ - Locator generation pipeline
+ - `SnapshotStorage` ABC for pluggable backends (Redis, S3, PostgreSQL, etc.)
+ - `JsonFileSnapshotStorage` — default JSON file-based backend
+ - Snapshot normalization rules to strip dynamic data (CSS hashes, state classes, etc.)
+ - Callbacks: `on_healing_success` / `on_healing_failure` for external integrations (metrics, alerting)
+ - Full documentation in `docs/source/other/self_healing.md`
+
+### Changed
+- Selenium & Playwright action methods now decorated with `@healing` for transparent self-healing
+- `wait_visibility` refactored into `_wait_visibility_base` to properly stack `@healing_after_wait` with `@wait_condition` / `@wait_continuous`
+- `wait_availability` decorated with `@healing_after_wait` for wait-timeout healing
+- Playwright element methods (`click`, `type_text`, `type_slowly`, `clear_text`, `hover`) now convert `playwright.Error` to `NoSuchElementException` for consistent healing trigger
+- `wait_visibility_without_error` — corrected log message from "hidden" to "visible" for non-continuous mode
+
+---
+
### v3.5.2
#### Added
diff --git a/docs/source/other/self_healing.md b/docs/source/other/self_healing.md
new file mode 100644
index 00000000..1c43c4b9
--- /dev/null
+++ b/docs/source/other/self_healing.md
@@ -0,0 +1,393 @@
+# Self-Healing Locators
+
+## Overview
+
+Self-healing locators automatically recover broken element locators at runtime.
+When an element is successfully located, its DOM context (tag, attributes, text, parent, siblings) is saved as a snapshot.
+If the original locator later fails, the system scans all elements with the same tag on the page,
+finds the most similar one, generates new stable locators, and updates the element transparently.
+
+Supported for both **Selenium** and **Playwright** backends.
+
+```{attention}
+Snapshots are saved per unique locator key per session. If a page changes significantly between runs,
+the snapshot should be refreshed by calling `wait_visibility()` or any other element-interaction method
+that triggers snapshot saving.
+```
+
+
+
+## Quick Start
+
+```python
+from mops.self_healing import configure, JsonFileSnapshotStorage
+
+# Minimal setup — save snapshots and heal broken locators
+configure(
+ save_snapshots=True,
+ heal_locators=True,
+ storage=JsonFileSnapshotStorage('snapshots'),
+)
+
+# Use elements normally — snapshots are saved automatically on first successful lookup
+my_element = page.my_button
+my_element.click() # snapshot saved here
+
+# If the locator breaks (e.g., after a deploy), healing kicks in transparently
+my_element.click() # heals automatically
+```
+
+
+
+## Configuration
+
+```{eval-rst}
+.. autoclass:: mops.self_healing.config.SelfHealingConfig
+ :members:
+ :undoc-members:
+```
+
+### Storage Backends
+
+**`JsonFileSnapshotStorage`** (default) — stores each snapshot as a JSON file on disk:
+
+```python
+from mops.self_healing import JsonFileSnapshotStorage
+
+configure(storage=JsonFileSnapshotStorage('my_snapshots'))
+```
+
+**Custom backends** — implement `SnapshotStorage` ABC:
+
+```python
+from mops.self_healing.snapshot import SnapshotStorage, ElementSnapshot
+
+class MyRedisStorage(SnapshotStorage):
+ def save(self, locator_key, snapshot):
+ ...
+
+ def load(self, locator_key):
+ ...
+```
+
+### Callbacks
+
+```python
+def on_success(result):
+ # result.healed_locator — the new working locator
+ # result.score — similarity score (0-1)
+ # result.original_locator — the broken locator
+ # result.timestamp — when the heal happened (ISO-8601, UTC)
+ # result.breakdown — SimilarityBreakdown of the best candidate
+ metrics.increment('healing.success', tags={'element': result.element_name})
+
+def on_failure(result):
+ # result.reason — 'no-snapshot', 'below-threshold', 'no-verified-locator'
+ # result.locator — the broken locator
+ # result.best_score — highest similarity score found before failure
+ # (0-1), or None if no candidate could be scored
+ # result.score_threshold — the configured acceptance threshold
+ # result.candidates_count — how many DOM candidates were compared
+ # result.breakdown — SimilarityBreakdown of the best candidate, or None
+ alerting.send(f'Healing failed for {result.element_name}: {result.reason} (best score: {result.best_score})')
+
+configure(
+ save_snapshots=True,
+ heal_locators=True,
+ storage=JsonFileSnapshotStorage(),
+ on_healing_success=on_success,
+ on_healing_failure=on_failure,
+)
+```
+
+Both callbacks receive a `SimilarityBreakdown` in `result.breakdown` for the **best candidate** —
+regardless of whether healing succeeded or failed. It tells you exactly which attributes
+matched the snapshot and which did not:
+
+```python
+def on_healing_result(result):
+ b = result.breakdown
+ print(f'matched: {b.matched_attributes}') # ['id', 'name']
+ print(f'mismatched: {b.mismatched_attributes}') # ['class']
+
+ # Inspect a specific attribute to spot dynamic data
+ id_match = b.attributes['id']
+ print(id_match.snapshot_value, '->', id_match.candidate_value) # 'user-123' -> 'user-456'
+ print(id_match.matched, id_match.score) # False 0.0
+
+ # The raw DOM snapshot of the best candidate as found on the page
+ snap = b.candidate_snapshot
+ print(snap.attributes, snap.text, snap.parent_tag)
+```
+
+* `AttributeMatch` — one per attribute: `attribute`, `snapshot_value`, `candidate_value`,
+ `matched`, `score` (0-1), `weight` (0.0 for attributes that do not affect scoring).
+* `SimilarityBreakdown` — `score`, `attributes` (dict), `text_snapshot`, `text_candidate`,
+ `text_score`, `parent_tag_*`, `parent_attrs_score`, `siblings_score`,
+ `siblings_snapshot_count`, `siblings_candidate_count`, `candidate_snapshot`
+ (a raw `ElementSnapshot` of the best candidate — reflects the actual DOM state,
+ unlike the normalized reference stored in the storage), plus convenience
+ `matched_attributes` / `mismatched_attributes` lists.
+
+### Scoring Weights
+
+Tune similarity scoring per attribute type — externally, from any project that
+uses the framework. Pass your own `ScoringWeights` through `configure()`:
+
+```python
+from mops.self_healing import configure, ScoringWeights, JsonFileSnapshotStorage
+
+configure(
+ save_snapshots=True,
+ heal_locators=True,
+ storage=JsonFileSnapshotStorage(),
+ scoring_weights=ScoringWeights(
+ attribute={'id': 1.0, 'class': 0.3, 'name': 0.7},
+ text=0.5,
+ parent=0.2,
+ siblings=0.1,
+ ),
+)
+```
+
+`attribute` maps an attribute name to its weight; keys not present are ignored
+during scoring (weight `0.0`). Weights apply to every element globally. Changing
+them at runtime is picked up on the next heal: the `Healer` singleton
+re-initialises when `scoring_weights`, `score_threshold`, `storage`, or
+`on_healing_failure` change. Mutating the same `ScoringWeights` object in place
+(e.g. `get_config().scoring_weights.attribute['id'] = 0.9`) also works because
+the `Healer` keeps a reference to it.
+
+### Healing Statistics
+
+Process-wide counters for end-of-run reporting (e.g. in `pytest_sessionfinish`):
+
+```python
+from mops.self_healing import get_healing_stats
+
+def pytest_sessionfinish(session, exitstatus):
+ stats = get_healing_stats()
+ print(f'healing: {stats.healed} ok, {stats.failed} failed of {stats.attempts}')
+ print(f'average best score: {stats.avg_best_score}')
+ print(f'failure reasons: {stats.failed_reasons}')
+```
+
+`HealingStats` fields: `attempts`, `healed`, `failed`, `failed_reasons` (per-reason breakdown),
+and `avg_best_score` (average similarity of successful heals).
+Note that `healed` counts candidates found by the Healer — a candidate can still fail DOM
+verification afterwards (`no-verified-locator`), which is reported via `on_healing_failure`
+but not counted here.
+
+### Snapshot Normalization
+
+Remove dynamic data (CSS hashes, state classes) from snapshots before saving:
+
+```python
+import re
+
+storage = JsonFileSnapshotStorage()
+storage.set_normalization_rules([
+ *storage._normalization_rules, # keep defaults
+ ('data-track', re.compile(r'.*'), ''), # remove data-track entirely
+ (None, re.compile(r'-\d+'), ''), # strip numeric suffixes from any attr
+])
+```
+
+The same rules are applied on **both sides** of the similarity comparison:
+snapshots are normalized when saved, and every DOM candidate is normalized with
+`SnapshotStorage.normalize_snapshot()` before scoring. This means dynamic data
+covered by the rules (e.g. a CSS-module hash in `class` or a numeric id suffix)
+does not unfairly penalise a candidate. The `candidate_snapshot` attached to a
+`SimilarityBreakdown` is the **raw** (non-normalized) candidate — so you can still
+see the actual DOM values for diagnostics.
+
+
+
+## Architecture
+
+### Decorators
+
+```{eval-rst}
+.. autofunction:: mops.utils.decorators.healing
+.. autofunction:: mops.utils.decorators.healing_after_wait
+.. autofunction:: mops.utils.decorators.no_healing
+```
+
+**`@healing`** — applied on action methods (`click`, `type_text`, `get_attribute`, etc.).
+Catches `NoSuchElementException`, attempts healing, retries the method once with the healed locator.
+
+**`@healing_after_wait`** — applied on `wait_visibility` and `wait_availability`.
+Triggers healing after the wait condition times out.
+
+**`@no_healing`** — applied on methods that must never trigger healing (`is_displayed`, `wait_hidden`).
+Temporarily disables `heal_locators` on the global config.
+
+### Healing Flow
+
+1. **Snapshot capture** — on first successful element lookup, DOM context is extracted via JS and saved to storage.
+ The storage key is the element's hierarchical locator key (`name::locator -> parent::locator`).
+
+2. **Healing trigger** — when an action method raises `NoSuchElementException`, `@healing` catches it,
+ or when `@healing_after_wait` catches a wait timeout.
+
+3. **Candidate search** — JS script collects all elements with the same tag, extracts their attributes,
+ text, parent, and up to 5 siblings.
+
+4. **Similarity scoring** — each candidate is scored against the saved snapshot.
+ Attributes (`id`, `name`, `class`, `aria-label`, etc.) have per-key weights.
+ `class` is compared **canonically** — kebab/snake/camel/Pascal variants of the
+ same words are treated as equal (`user-profile-card` == `UserProfileCard`,
+ `checkout_form_submit` == `checkoutFormSubmit`, `modal__close-btn` ==
+ `ModalCloseBtn`), for the element, its parent, and its
+ siblings. Text matching rejects short (1–2 char) substring false-positives and
+ only counts whole-word matches. Parent tag/attributes and sibling structure
+ contribute additional weighted scores. The best candidate must exceed
+ `score_threshold` (default `0.7`).
+
+ After a candidate is picked, the resolved DOM element is re-snapshotted and
+ compared against the best candidate — protecting against DOM changes (e.g.
+ React re-renders) that shift element indices between the candidate scan and
+ element resolution. If they differ, healing fails with
+ `dom-changed-during-healing` instead of silently healing the wrong element.
+
+5. **Locator generation** — stable XPath locators are generated for the best candidate:
+ `@id` → `data-testid` → `@name` → `@aria-label` → `@type` → stable `@class` → visible text → positional XPath.
+ The caller tries each in order until one resolves.
+
+6. **Locator persistence** — the first working locator is written to the element's `locator` attribute.
+ Subsequent lookups use the healed locator directly.
+
+### Key Classes
+
+```{eval-rst}
+.. autoclass:: mops.self_healing.healer.Healer
+ :members: heal
+ :undoc-members:
+
+.. autoclass:: mops.self_healing.healer.SuccessHealingResult
+ :members:
+ :undoc-members:
+
+.. autoclass:: mops.self_healing.healer.FailedHealingResult
+ :members:
+ :undoc-members:
+
+.. autoclass:: mops.self_healing.snapshot.SnapshotStorage
+ :members: save, load, save_from_element, extract_full_locator_key, set_normalization_rules
+ :undoc-members:
+
+.. autoclass:: mops.self_healing.snapshot.ElementSnapshot
+ :members:
+ :undoc-members:
+
+.. autoclass:: mops.self_healing.snapshot.JsonFileSnapshotStorage
+ :members:
+ :undoc-members:
+
+.. autoclass:: mops.self_healing.healer.ScoringWeights
+ :members:
+ :undoc-members:
+
+.. autoclass:: mops.self_healing.healer.AttributeMatch
+ :members:
+ :undoc-members:
+
+.. autoclass:: mops.self_healing.healer.SimilarityBreakdown
+ :members:
+ :undoc-members:
+
+.. autoclass:: mops.self_healing.healer.HealingStats
+ :members:
+ :undoc-members:
+
+.. autofunction:: mops.self_healing.healer.get_healing_stats
+```
+
+
+
+## Disabling Healing
+
+### Per-method: `@no_healing`
+
+```python
+from mops.utils.decorators import no_healing
+
+class MyElement(Element):
+ @no_healing
+ def is_displayed(self, silent=False):
+ ...
+```
+
+### Globally
+
+```python
+from mops.self_healing import configure
+
+# Save snapshots but don't heal (data collection mode)
+configure(save_snapshots=True, heal_locators=False)
+
+# Disable entirely
+configure(save_snapshots=False, heal_locators=False)
+```
+
+
+
+## Error Handling
+
+Healing failures are non-fatal — the original exception is re-raised if healing cannot find a working locator.
+
+| Phase | Failure Reason | Callback |
+|-------|---------------|----------|
+| No snapshot saved | `no-snapshot` | `on_healing_failure` |
+| JS script error | `candidates-script-error` | `on_healing_failure` |
+| No matching candidates on page | `no-candidates` | `on_healing_failure` |
+| Best score below threshold | `below-threshold` | `on_healing_failure` |
+| Best index out of bounds | `index-out-of-bounds` | `on_healing_failure` |
+| Locator generation error | `generate-locator-error` | `on_healing_failure` |
+| DOM changed between candidate scan and element resolution | `dom-changed-during-healing` | `on_healing_failure` |
+| No candidate passes DOM verification | `no-verified-locator` | `on_healing_failure` |
+| Candidate passes DOM verification | — | `on_healing_success` |
+
+`FailedHealingResult` exposes diagnostics for external monitoring
+(`SuccessHealingResult` also carries a `timestamp`):
+
+* `best_score` — highest similarity score found before the failure
+ (`below-threshold`, `index-out-of-bounds`, `generate-locator-error`,
+ `dom-changed-during-healing`, `no-verified-locator`),
+ or `None` when no candidate could be scored (`no-snapshot`, `candidates-script-error`, `no-candidates`).
+* `score_threshold` — the configured acceptance threshold.
+* `candidates_count` — how many DOM candidates were compared, or `None` when candidates were never collected.
+* `breakdown` — a `SimilarityBreakdown` of the best candidate whenever one was scored
+ (`below-threshold`, `index-out-of-bounds`, `generate-locator-error`,
+ `dom-changed-during-healing`, `no-verified-locator`),
+ or `None` when no candidate could be scored. The same breakdown is available on
+ `SuccessHealingResult` after a successful heal.
+
+
+
+## Methods with Healing
+
+| Method | `@healing` | `@healing_after_wait` |
+|--------|-----------|----------------------|
+| `click` | ✓ | |
+| `type_text` | ✓ | |
+| `type_slowly` | ✓ | |
+| `clear_text` | ✓ | |
+| `check` | ✓ | |
+| `uncheck` | ✓ | |
+| `hover` | ✓ | |
+| `get_attribute` | ✓ | |
+| `is_enabled` | ✓ | |
+| `is_checked` | ✓ | |
+| `screenshot_image` | ✓ | |
+| `screenshot_base` | ✓ | |
+| `text` | ✓ | |
+| `inner_text` | ✓ | |
+| `value` | ✓ | |
+| `scroll_into_view` | ✓ | |
+| `hide` | ✓ | |
+| `show` | ✓ | |
+| `wait_visibility` | | ✓ |
+| `wait_availability` | | ✓ |
+
+Methods without these decorators (`is_displayed`, `is_available`, `is_hidden`, `wait_hidden`, `wait_hidden_without_error`) do not trigger healing.
diff --git a/docs/source/toc.md b/docs/source/toc.md
index 20c8221c..fc439619 100644
--- a/docs/source/toc.md
+++ b/docs/source/toc.md
@@ -30,6 +30,7 @@ caption: Other
other/objects_initialisation
other/visual_comparison
+other/self_healing
```
```{toctree}
diff --git a/mops/__init__.py b/mops/__init__.py
index 13a042dd..4dafab96 100644
--- a/mops/__init__.py
+++ b/mops/__init__.py
@@ -1,4 +1,4 @@
"""Wrapper of Selenium, Appium and Playwright with a single API."""
-__version__ = '3.5.4'
+__version__ = '4.0.0rc2'
__project_name__ = 'mops'
diff --git a/mops/abstraction/element_abc.py b/mops/abstraction/element_abc.py
index 9d649078..3b62fb18 100644
--- a/mops/abstraction/element_abc.py
+++ b/mops/abstraction/element_abc.py
@@ -371,6 +371,17 @@ def is_available(self) -> bool:
"""
raise NotImplementedError
+ def _apply_healing(self) -> bool:
+ """
+ Attempt self-healing and persist the first working locator.
+
+ Called by the :func:`@healing ` decorator
+ and :func:`@healing_after_wait `.
+
+ :return: :obj:`True` if a healed locator was found and applied.
+ """
+ raise NotImplementedError
+
def is_displayed(self, silent: bool = False) -> bool:
"""
Check if the element is displayed.
diff --git a/mops/base/element.py b/mops/base/element.py
index 93518a75..d4246592 100644
--- a/mops/base/element.py
+++ b/mops/base/element.py
@@ -31,7 +31,7 @@
from mops.playwright.play_element import PlayElement
from mops.selenium.elements.mobile_element import MobileElement
from mops.selenium.elements.web_element import WebElement
-from mops.utils.decorators import wait_condition, wait_continuous
+from mops.utils.decorators import healing, healing_after_wait, wait_condition, wait_continuous
from mops.utils.internal_utils import (
QUARTER_WAIT_EL,
WAIT_EL,
@@ -263,8 +263,7 @@ def send_keyboard_action(self, action: str | KeyboardKeys) -> Element:
# Elements waits
- @wait_continuous
- @wait_condition
+ @healing_after_wait
def wait_visibility(
self,
*,
@@ -299,11 +298,7 @@ def wait_visibility(
:type continuous: typing.Union[int, float, bool]
:return: :class:`Element`
"""
- return Result(
- execution_result=self.is_displayed(silent=True),
- log=f'Wait until "{self.name}" becomes visible',
- exc=TimeoutException(f'"{self.name}" not visible', info=self),
- )
+ return self._wait_visibility_base(timeout=timeout, silent=silent, continuous=continuous)
def wait_visibility_without_error(
self,
@@ -340,11 +335,11 @@ def wait_visibility_without_error(
:return: :class:`Element`
"""
if not silent:
- strategy = 'continuous visible' if continuous else 'hidden'
+ strategy = 'continuous visible' if continuous else 'visible'
self.log(f'Wait until "{self.name}" becomes {strategy} without error exception')
try:
- self.wait_visibility(timeout=timeout, silent=True, continuous=continuous)
+ self._wait_visibility_base(timeout=timeout, silent=True, continuous=continuous)
except (TimeoutException, WebDriverException, ContinuousWaitException) as exception:
if not silent:
self.log(f'Ignored exception: "{exception.msg}"')
@@ -437,6 +432,7 @@ def wait_hidden_without_error(
self.log(f'Ignored exception: "{exception.msg}"')
return self
+ @healing_after_wait
@wait_condition
def wait_availability(self, *, timeout: int = WAIT_EL, silent: bool = False) -> Element:
r"""
@@ -766,6 +762,7 @@ def is_fully_visible(self, check_displaying: bool = True, silent: bool = False)
return is_visible
+ @healing
def scroll_into_view(
self,
block: ScrollTo = ScrollTo.CENTER,
@@ -834,6 +831,7 @@ def save_screenshot(
return image_object
+ @healing
def hide(self, silent: bool = False) -> Element:
"""
Make the element invisible by setting its opacity to 0.
@@ -848,6 +846,7 @@ def hide(self, silent: bool = False) -> Element:
self.execute_script('arguments[0].style.opacity = "0";')
return self
+ @healing
def show(self, silent: bool = False) -> Element:
"""
Make the element visible by setting its opacity to 1.
@@ -1072,3 +1071,26 @@ def _element_cls(self) -> type[Element]:
:return: :obj:`typing.Type` [:class:`Element`]
"""
return Element
+
+ def _wait_visibility_base(
+ self, *, timeout: int = WAIT_EL, silent: bool = False, continuous: bool | float = False
+ ) -> Element:
+ """Wait for element visibility with polling, used by wait_visibility."""
+
+ def _visibility_check(
+ self: Element,
+ *,
+ timeout: int = WAIT_EL, # noqa: ARG001
+ silent: bool = False, # noqa: ARG001
+ continuous: bool | float = False, # noqa: ARG001
+ ) -> Element:
+ return Result(
+ execution_result=self.is_displayed(silent=True),
+ log=f'Wait until "{self.name}" becomes visible',
+ exc=TimeoutException(f'"{self.name}" not visible', info=self),
+ )
+
+ # Give the check a stable name so decorators report "wait_visibility"
+ _visibility_check.__name__ = 'wait_visibility'
+ _check = wait_continuous(wait_condition(_visibility_check))
+ return _check(self, timeout=timeout, silent=silent, continuous=continuous)
diff --git a/mops/playwright/play_element.py b/mops/playwright/play_element.py
index f0c483cc..24ab213f 100644
--- a/mops/playwright/play_element.py
+++ b/mops/playwright/play_element.py
@@ -6,11 +6,21 @@
from playwright.sync_api import Error, Locator, Page as PlaywrightPage
from mops.abstraction.element_abc import ElementABC
-from mops.exceptions import InvalidSelectorException, NotInitializedException
+from mops.exceptions import (
+ DriverWrapperException,
+ InvalidSelectorException,
+ NoSuchElementException,
+ NoSuchParentException,
+ NotInitializedException,
+)
from mops.mixins.objects.location import Location
from mops.mixins.objects.size import Size
+from mops.self_healing.config import get_config
+from mops.self_healing.healer import MAX_HEALING_DEPTH, FailedHealingResult, SuccessHealingResult
+from mops.self_healing.healer_factory import get_healer
+from mops.self_healing.locator_generator import generate_locator_pw
from mops.shared_utils import cut_log_data, get_image
-from mops.utils.decorators import retry
+from mops.utils.decorators import healing, retry
from mops.utils.internal_utils import (
calculate_coordinate_to_click,
is_element,
@@ -77,6 +87,7 @@ def all_elements(self) -> list[PlayElement] | list[Any]:
# Element interaction
+ @healing
def click(self, *, force_wait: bool = True, **kwargs: Any) -> PlayElement:
"""
Clicks on the element.
@@ -101,10 +112,13 @@ def click(self, *, force_wait: bool = True, **kwargs: Any) -> PlayElement:
if force_wait:
self.wait_visibility(silent=True)
- if self.driver_wrapper.is_mobile_resolution:
- self._first_element.tap(**kwargs)
- else:
- self._first_element.click(**kwargs)
+ try:
+ if self.driver_wrapper.is_mobile_resolution:
+ self._first_element.tap(**kwargs)
+ else:
+ self._first_element.click(**kwargs)
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
return self
@@ -142,6 +156,7 @@ def click_into_center(self, silent: bool = False) -> PlayElement:
self.driver_wrapper.click_by_coordinates(x=x, y=y, silent=True)
return self
+ @healing
def type_text(self, text: str | KeyboardKeys, silent: bool = False) -> PlayElement:
"""
Types text into the element.
@@ -157,9 +172,13 @@ def type_text(self, text: str | KeyboardKeys, silent: bool = False) -> PlayEleme
if not silent:
self.log(f'Type text "{cut_log_data(text)}" into "{self.name}"')
- self._first_element.type(text=text)
+ try:
+ self._first_element.type(text=text)
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
return self
+ @healing
def type_slowly(self, text: str, sleep_gap: float = 0.05, silent: bool = False) -> PlayElement:
"""
Types text into the element slowly with a delay between keystrokes.
@@ -175,9 +194,13 @@ def type_slowly(self, text: str, sleep_gap: float = 0.05, silent: bool = False)
if not silent:
self.log(f'Type text {cut_log_data(text)} into "{self.name}"')
- self._first_element.type(text=text, delay=sleep_gap)
+ try:
+ self._first_element.type(text=text, delay=sleep_gap)
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
return self
+ @healing
def clear_text(self, silent: bool = False) -> PlayElement:
"""
Clear the text of the element.
@@ -189,9 +212,13 @@ def clear_text(self, silent: bool = False) -> PlayElement:
if not silent:
self.log(f'Clear text in "{self.name}"')
- self._first_element.fill('')
+ try:
+ self._first_element.fill('')
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
return self
+ @healing
def hover(self, silent: bool = False) -> PlayElement:
"""
Hover the mouse over the current element.
@@ -203,7 +230,10 @@ def hover(self, silent: bool = False) -> PlayElement:
if not silent:
self.log(f'Hover over "{self.name}"')
- self._first_element.hover()
+ try:
+ self._first_element.hover()
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
return self
def hover_outside(self, x: int = 0, y: int = -5) -> PlayElement:
@@ -220,28 +250,35 @@ def hover_outside(self, x: int = 0, y: int = -5) -> PlayElement:
self._first_element.hover(position={'x': float(x), 'y': float(y)}, force=True)
return self
+ @healing
def check(self) -> PlayElement:
"""
Check the checkbox element.
:return: :class:`PlayElement`
"""
- self._first_element.check()
-
+ try:
+ self._first_element.check()
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
return self
+ @healing
def uncheck(self) -> PlayElement:
"""
Unchecks the checkbox element.
:return: :class:`PlayElement`
"""
- self._first_element.uncheck()
-
+ try:
+ self._first_element.uncheck()
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
return self
# Element state
+ @healing
def screenshot_image(self, screenshot_base: bytes | None = None) -> Image:
"""
Return a :class:`PIL.Image.Image` object representing the screenshot of the web element.
@@ -256,40 +293,56 @@ def screenshot_image(self, screenshot_base: bytes | None = None) -> Image:
return get_image(screenshot_base)
@property
+ @healing
def screenshot_base(self) -> bytes:
"""
Returns the binary screenshot data of the element.
:return: :class:`bytes` - screenshot binary
"""
- return self._first_element.screenshot()
+ try:
+ return self._first_element.screenshot()
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
@property
+ @healing
def text(self) -> str:
"""
Returns the text of the element.
:return: :class:`str` - element text
"""
- return self.inner_text
+ try:
+ return self.inner_text
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
@property
+ @healing
def inner_text(self) -> str:
"""
Returns the inner text of the element.
:return: :class:`str` - element inner text
"""
- return self._first_element.inner_text()
+ try:
+ return self._first_element.inner_text()
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
@property
+ @healing
def value(self) -> str:
"""
Returns the value of the element.
:return: :class:`str` - element value
"""
- return self._first_element.input_value()
+ try:
+ return self._first_element.input_value()
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
def is_available(self) -> bool:
"""
@@ -297,7 +350,10 @@ def is_available(self) -> bool:
:return: :class:`bool` - :obj:`True` if present in DOM
"""
- return bool(len(self.element.element_handles()))
+ result = bool(len(self.element.element_handles()))
+ if result:
+ self._save_snapshot(self._first_element)
+ return result
def is_displayed(self, silent: bool = False) -> bool:
"""
@@ -311,9 +367,13 @@ def is_displayed(self, silent: bool = False) -> bool:
self.log(f'Check visibility of "{self.name}"')
try:
- return self._first_element.is_visible()
+ result = self._first_element.is_visible()
except Error as exc:
raise InvalidSelectorException(exc.message) from exc
+ else:
+ if result:
+ self._save_snapshot(self._first_element)
+ return result
def is_hidden(self, silent: bool = False) -> bool:
"""
@@ -328,6 +388,7 @@ def is_hidden(self, silent: bool = False) -> bool:
return self._first_element.is_hidden()
+ @healing
def get_attribute(self, attribute: str, silent: bool = False) -> str:
"""
Retrieve a specific attribute from the current element.
@@ -341,7 +402,10 @@ def get_attribute(self, attribute: str, silent: bool = False) -> str:
if not silent:
self.log(f'Get "{attribute}" from "{self.name}"')
- return self._first_element.get_attribute(attribute)
+ try:
+ return self._first_element.get_attribute(attribute)
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
def get_all_texts(self, silent: bool = False) -> list:
"""
@@ -401,6 +465,7 @@ def location(self) -> Location:
box = self.element.first.bounding_box()
return Location(x=box['x'], y=box['y'])
+ @healing
def is_enabled(self, silent: bool = False) -> bool:
"""
Check if the current element is enabled.
@@ -412,25 +477,38 @@ def is_enabled(self, silent: bool = False) -> bool:
if not silent:
self.log(f'Check is element "{self.name}" enabled')
- return self._first_element.is_enabled()
+ try:
+ return self._first_element.is_enabled()
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
+ @healing
def is_checked(self) -> bool:
"""
Check if a checkbox or radio button is selected.
:return: :class:`bool` - :obj:`True` if the checkbox or radio button is checked, :obj:`False` otherwise.
"""
- return self._first_element.is_checked()
+ try:
+ return self._first_element.is_checked()
+ except Error as exc:
+ raise NoSuchElementException(str(exc)) from exc
# Mixin
- def _get_base(self) -> PlaywrightPage | Locator:
+ def _get_base(self, wait_strategy: bool = True) -> PlaywrightPage | Locator:
"""
Get driver depends on parent element if available
+ :param wait_strategy: Compatibility parameter for the Selenium signature.
+ Playwright Locators are lazy so this has no effect.
:return: driver
"""
base = self.driver
+ if not base:
+ msg = "Can't find driver"
+ raise DriverWrapperException(msg)
+
if self.parent:
self.log(f'Get element "{self.name}" from parent element "{self.parent.name}"', level='debug')
@@ -452,3 +530,171 @@ def _set_locator(self) -> None:
self.locator = get_platform_locator(self)
set_playwright_locator(self)
self._is_locator_configured = True
+
+ # Self-healing
+
+ @staticmethod
+ def _parse_healed_locator_pw(healed_locator: str) -> str:
+ """Strip the ``xpath=`` prefix from a healed locator for Playwright use."""
+ if healed_locator.startswith('xpath='):
+ return healed_locator[len('xpath=') :]
+ return healed_locator
+
+ def _save_snapshot(self, locator_element: Locator) -> None:
+ """Save a DOM snapshot for the current element if snapshot saving is enabled."""
+ config = get_config()
+ if config.save_snapshots and config.storage:
+ try:
+ config.storage.save_from_element(self, locator_element, self.driver_wrapper)
+ except Exception as exc: # noqa: BLE001
+ self.log(f'Failed to save snapshot for "{self.name}": {exc}', level='debug')
+
+ def _attempt_healing(self) -> SuccessHealingResult | None:
+ """Attempt to heal a failed element lookup using the self-healing subsystem.
+
+ :return: :class:`SuccessHealingResult` if a suitable candidate was found,
+ :obj:`None` otherwise.
+ """
+ try:
+ healer = get_healer()
+ storage = get_config().storage
+ locator_key = storage.extract_full_locator_key(self)
+ result = healer.heal(
+ element_name=self.name,
+ locator_key=locator_key,
+ locator=self.locator,
+ driver_wrapper=self.driver_wrapper,
+ find_elements_fn=lambda tag: self.driver.locator(tag).all(),
+ generate_locator_fn=generate_locator_pw,
+ )
+ if type(result) is SuccessHealingResult:
+ return result
+ except Exception as exc: # noqa: BLE001
+ self.log(f'Self-healing failed with unexpected exception: {exc}', level='warning')
+ return None
+
+ def _resolve_base(self, depth: int) -> tuple[PlaywrightPage | Locator | None, str | None]:
+ """Resolve the base (parent) context, healing a missing parent first.
+
+ Returns a ``(base, error)`` pair; *base* is ``None`` (with an error message)
+ when the parent could not be resolved even after healing.
+ """
+ try:
+ return self._get_base(wait_strategy=False), None
+ except (NoSuchElementException, NoSuchParentException) as exc:
+ # Parent is missing — heal it first, then retry inside the healed context.
+ if depth < MAX_HEALING_DEPTH and self.parent is not None and self.parent._apply_healing(depth + 1):
+ try:
+ return self._get_base(wait_strategy=False), None
+ except (NoSuchElementException, NoSuchParentException) as exc2:
+ return None, str(exc2)
+ return None, str(exc)
+
+ def _try_healed_locators(self, result: SuccessHealingResult, depth: int = 0) -> None:
+ """Try each healed locator candidate and persist the first working one.
+
+ A healing attempt always ends with a terminal callback: ``on_healing_success``
+ when a candidate passes DOM verification, or ``on_healing_failure`` otherwise
+ (including when verification itself fails or cannot be performed).
+
+ If the base (parent) context is stale and cannot be resolved, the parent is
+ healed first so the sub-element locators are verified inside the healed parent.
+
+ :param result: The healing result containing candidate locators.
+ :param depth: Recursion depth guard for healing missing parents.
+ :raises NoSuchElementException: If no candidate locator resolves to an element.
+ """
+ config = get_config()
+ base, error = self._resolve_base(depth)
+
+ if base is not None:
+ # Try the ORIGINAL locator first — the DOM may have settled while
+ # healing was running (e.g. async re-render) and it could work again.
+ # If it does, no healing actually happened, so no metrics are emitted.
+ try:
+ original = base.locator(self.locator)
+ if original.count() > 0:
+ self._element = original
+ self.log(f'Self-healing: original locator "{self.locator}" works again for "{self.name}"')
+ return
+ except Error:
+ pass
+
+ try:
+ healed = self._verify_healed_locators(base, result, config)
+ except Exception as exc: # noqa: BLE001
+ healed = None
+ error = str(exc)
+ if healed is not None:
+ return
+ if error is None:
+ error = 'All healed locator candidates failed DOM verification'
+
+ # Terminal failure — none of the candidates passed DOM verification
+ if config.on_healing_failure:
+ config.on_healing_failure(
+ FailedHealingResult(
+ element_name=result.element_name,
+ locator_key=result.locator_key,
+ locator=result.original_locator,
+ reason='no-verified-locator',
+ error=error,
+ best_score=result.score,
+ score_threshold=config.score_threshold,
+ candidates_count=result.candidates_count,
+ breakdown=result.breakdown,
+ )
+ )
+
+ msg = error or 'Healed locator candidates did not match any element'
+ raise NoSuchElementException(msg)
+
+ def _verify_healed_locators(
+ self,
+ base: PlaywrightPage | Locator,
+ result: SuccessHealingResult,
+ config: Any,
+ ) -> Locator | None:
+ """Try each healed locator against *base*; persist and callback the first hit."""
+ for locator_str in result.healed_locators_candidates:
+ selector = self._parse_healed_locator_pw(locator_str)
+ try:
+ candidate = base.locator(selector)
+ if candidate.count() > 0:
+ result.healed_locator = locator_str
+ self.locator = selector
+ self._element = candidate
+ self.log(f'Self-healing: healed "{self.name}" with locator "{locator_str}"')
+ # Fire success callback AFTER locator is verified against DOM
+ if config.on_healing_success:
+ config.on_healing_success(result)
+ return candidate
+ except Error:
+ continue
+ return None
+
+ def _apply_healing(self, depth: int = 0) -> bool:
+ """Attempt healing and persist the first working locator.
+
+ Called by :func:`@healing ` and
+ :func:`@healing_after_wait `.
+
+ :param depth: Recursion depth guard used when a parent needs to be healed
+ first so the element can be verified inside the healed parent context.
+ :return: :obj:`True` if a healed locator was found and applied.
+ """
+ result = self._attempt_healing()
+ if not result:
+ return False
+ try:
+ self._try_healed_locators(result, depth=depth)
+ except NoSuchElementException:
+ return False
+ return True
+
+ def _heal_after_wait(self) -> bool:
+ """Attempt healing after a wait condition timed out.
+
+ :return: :obj:`True` if a working locator was found.
+ """
+ return self._apply_healing()
diff --git a/mops/selenium/core/core_element.py b/mops/selenium/core/core_element.py
index 8e95c2a2..a0fa5bbd 100644
--- a/mops/selenium/core/core_element.py
+++ b/mops/selenium/core/core_element.py
@@ -11,6 +11,7 @@
StaleElementReferenceException as SeleniumStaleElementReferenceException,
WebDriverException as SeleniumWebDriverException,
)
+from selenium.webdriver.common.by import By
from selenium.webdriver.support.wait import WebDriverWait
from mops.abstraction.element_abc import ElementABC
@@ -27,8 +28,12 @@
from mops.mixins.objects.location import Location
from mops.mixins.objects.size import Size
from mops.selenium.sel_utils import ActionChains
+from mops.self_healing.config import get_config
+from mops.self_healing.healer import MAX_HEALING_DEPTH, FailedHealingResult, SuccessHealingResult
+from mops.self_healing.healer_factory import get_healer
+from mops.self_healing.locator_generator import generate_locator
from mops.shared_utils import _scaled_screenshot, cut_log_data
-from mops.utils.decorators import retry
+from mops.utils.decorators import healing, retry
from mops.utils.internal_utils import WAIT_EL, get_dict, is_group, safe_call
if TYPE_CHECKING:
@@ -43,6 +48,14 @@
from mops.keyboard_keys import KeyboardKeys
+def _parse_healed_locator(healed_locator: str) -> tuple[str, str]:
+ """Convert a ``xpath=//...`` prefixed locator into a ``(By, value)`` tuple."""
+ if healed_locator.startswith('xpath='):
+ return By.XPATH, healed_locator[len('xpath=') :]
+ # Default fallback: treat as XPath
+ return By.XPATH, healed_locator
+
+
class CoreElement(ElementABC, ABC):
parent: Element | CoreElement
@@ -88,6 +101,7 @@ def all_elements(self) -> list[CoreElement] | list[Any]:
# Element interaction
+ @healing
@retry(ElementNotInteractableException)
def click(self, *, force_wait: bool = True, **kwargs: Any) -> CoreElement:
"""
@@ -123,6 +137,7 @@ def click(self, *, force_wait: bool = True, **kwargs: Any) -> CoreElement:
msg = f'Element "{self.name}" not interactable. {self.get_element_info()}. Original error: {selenium_exc_msg}'
raise ElementNotInteractableException(msg)
+ @healing
def type_text(self, text: str | KeyboardKeys, silent: bool = False) -> CoreElement:
"""
Types text into the element.
@@ -142,6 +157,7 @@ def type_text(self, text: str | KeyboardKeys, silent: bool = False) -> CoreEleme
return self
+ @healing
def type_slowly(self, text: str, sleep_gap: float = 0.05, silent: bool = False) -> CoreElement:
"""
Types text into the element slowly with a delay between keystrokes.
@@ -166,6 +182,7 @@ def type_slowly(self, text: str, sleep_gap: float = 0.05, silent: bool = False)
return self
+ @healing
def clear_text(self, silent: bool = False) -> CoreElement:
"""
Clear the text of the element.
@@ -181,6 +198,7 @@ def clear_text(self, silent: bool = False) -> CoreElement:
return self
+ @healing
def check(self) -> CoreElement:
"""
Check the checkbox element.
@@ -197,6 +215,7 @@ def check(self) -> CoreElement:
return self
+ @healing
def uncheck(self) -> CoreElement:
"""
Unchecks the checkbox element.
@@ -215,6 +234,7 @@ def uncheck(self) -> CoreElement:
# Element state
+ @healing
def screenshot_image(self, screenshot_base: bytes | None = None) -> Image:
"""
Return a :class:`PIL.Image.Image` object representing the screenshot of the web element.
@@ -230,6 +250,7 @@ def screenshot_image(self, screenshot_base: bytes | None = None) -> Image:
return _scaled_screenshot(screenshot_base, element_size)
@property
+ @healing
def screenshot_base(self) -> bytes:
"""
Returns the binary screenshot data of the element.
@@ -241,6 +262,7 @@ def screenshot_base(self) -> bytes:
return self.element.screenshot_as_png
@property
+ @healing
@retry(SeleniumStaleElementReferenceException)
def text(self) -> str:
"""
@@ -324,6 +346,7 @@ def is_hidden(self, silent: bool = False) -> bool:
return status
+ @healing
@retry(SeleniumStaleElementReferenceException)
def get_attribute(self, attribute: str, silent: bool = False) -> str:
"""
@@ -397,6 +420,7 @@ def location(self) -> Location:
"""
return Location(**self.execute_script(get_element_position_on_screen_js))
+ @healing
def is_enabled(self, silent: bool = False) -> bool:
"""
Check if the current element is enabled.
@@ -410,6 +434,7 @@ def is_enabled(self, silent: bool = False) -> bool:
return self.element.is_enabled()
+ @healing
def is_checked(self) -> bool:
"""
Check if a checkbox or radio button is selected.
@@ -479,6 +504,8 @@ def _get_base(self, wait_strategy: bool | Callable = True) -> SeleniumWebDriver
"""
Get driver with depends on parent element if available
+ :param wait_strategy: wait strategy for the parent element before it is used
+ as a base for the sub-element lookup. ``True`` waits for the parent to be visible.
:return: driver
"""
base = self.driver
@@ -508,6 +535,10 @@ def _find_element(self, wait_parent: bool = False) -> SeleniumWebElement | Appiu
try:
element = base.find_element(self.locator_type, self.locator)
self._cached_element = element
+ # Save snapshot for future healing
+ config = get_config()
+ if config.save_snapshots and config.storage:
+ config.storage.save_from_element(self, element, self.driver)
except (SeleniumInvalidArgumentException, SeleniumInvalidSelectorException) as exc:
self._raise_invalid_selector_exception(exc)
except SeleniumNoSuchElementException as exc:
@@ -515,6 +546,151 @@ def _find_element(self, wait_parent: bool = False) -> SeleniumWebElement | Appiu
else:
return element
+ def _attempt_healing(self) -> SuccessHealingResult | None:
+ """
+ Attempt to heal a failed element lookup using the self-healing subsystem.
+
+ :return: :class:`SuccessHealingResult` if a suitable candidate was found, :obj:`None` otherwise.
+ """
+ try:
+ healer = get_healer()
+ locator_key = get_config().storage.extract_full_locator_key(self)
+ result = healer.heal(
+ element_name=self.name,
+ locator_key=locator_key,
+ locator=self.locator,
+ driver_wrapper=self.driver_wrapper,
+ find_elements_fn=lambda tag: self.driver.find_elements(By.TAG_NAME, tag),
+ generate_locator_fn=generate_locator,
+ )
+ if type(result) is SuccessHealingResult:
+ return result
+ except Exception as exc: # noqa: BLE001
+ self.log(f'Self-healing failed with unexpected exception: {exc}', level='warning')
+ return None
+
+ def _resolve_base(self, depth: int) -> tuple[Any, str | None]:
+ """Resolve the base (parent) context, healing a missing parent first.
+
+ Returns a ``(base, error)`` pair; *base* is ``None`` (with an error message)
+ when the parent could not be resolved even after healing.
+ """
+ try:
+ return self._get_base(wait_strategy=False), None
+ except (NoSuchElementException, NoSuchParentException) as exc:
+ # Parent is missing — heal it first, then retry inside the healed context.
+ if depth < MAX_HEALING_DEPTH and self.parent is not None and self.parent._apply_healing(depth + 1):
+ try:
+ return self._get_base(wait_strategy=False), None
+ except (NoSuchElementException, NoSuchParentException) as exc2:
+ return None, str(exc2)
+ return None, str(exc)
+
+ def _try_healed_locators(self, result: SuccessHealingResult, depth: int = 0) -> SeleniumWebElement:
+ """Try each healed locator and persist the first working one.
+
+ A healing attempt always ends with a terminal callback: ``on_healing_success``
+ when a candidate passes DOM verification, or ``on_healing_failure`` otherwise
+ (including when verification itself fails or cannot be performed).
+
+ If the base (parent) context is stale and cannot be resolved, the parent is
+ healed first so the sub-element locators are verified inside the healed parent.
+ """
+ config = get_config()
+ base, error = self._resolve_base(depth)
+
+ if base is not None:
+ # Try the ORIGINAL locator first — the DOM may have settled while
+ # healing was running (e.g. async re-render) and it could work again.
+ # If it does, no healing actually happened, so no metrics are emitted.
+ try:
+ original = base.find_element(self.locator_type, self.locator)
+ except SeleniumNoSuchElementException:
+ original = None
+ if original is not None:
+ self._cached_element = original
+ self.log(f'Self-healing: original locator "{self.locator}" works again for "{self.name}"')
+ return original
+
+ try:
+ healed = self._verify_healed_locators(base, result, config)
+ except Exception as exc: # noqa: BLE001
+ healed = None
+ error = exc.msg if isinstance(exc, SeleniumWebDriverException) else str(exc)
+ if healed is not None:
+ return healed
+ if error is None:
+ error = 'All healed locator candidates failed find_element()'
+
+ # Terminal failure — none of the candidates passed DOM verification
+ if config.on_healing_failure:
+ config.on_healing_failure(
+ FailedHealingResult(
+ element_name=result.element_name,
+ locator_key=result.locator_key,
+ locator=result.original_locator,
+ reason='no-verified-locator',
+ error=error,
+ best_score=result.score,
+ score_threshold=config.score_threshold,
+ candidates_count=result.candidates_count,
+ breakdown=result.breakdown,
+ )
+ )
+ msg = error or 'Healed locator candidates did not match any element'
+ raise NoSuchElementException(msg)
+
+ def _verify_healed_locators(
+ self,
+ base: SeleniumWebDriver | SeleniumWebElement,
+ result: SuccessHealingResult,
+ config: Any,
+ ) -> SeleniumWebElement | None:
+ """Try each healed locator against *base*; persist and callback the first hit."""
+ for locator in result.healed_locators_candidates:
+ healed_locator_type, healed_locator_value = _parse_healed_locator(locator)
+ try:
+ healed = base.find_element(healed_locator_type, healed_locator_value)
+ except SeleniumNoSuchElementException:
+ continue
+ result.healed_locator = locator
+ self.locator_type = healed_locator_type
+ self.locator = healed_locator_value
+ self._cached_element = healed
+ self.log(f'Self-healing: healed "{self.name}" with locator "{locator}"')
+ # Fire success callback AFTER locator is verified against DOM
+ if config.on_healing_success:
+ config.on_healing_success(result)
+ return healed
+ return None
+
+ def _apply_healing(self, depth: int = 0) -> bool:
+ """Attempt healing and persist the first working locator.
+
+ Called by :func:`@healing ` and
+ :func:`@healing_after_wait `.
+
+ :param depth: Recursion depth guard used when a parent needs to be healed
+ first so the element can be verified inside the healed parent context.
+ :return: :obj:`True` if a healed locator was found and applied.
+ """
+ result = self._attempt_healing()
+ if not result:
+ return False
+ try:
+ self._try_healed_locators(result, depth=depth)
+ except NoSuchElementException:
+ return False
+ return True
+
+ def _heal_after_wait(self) -> bool:
+ """Attempt healing after a wait condition timed out.
+
+ Persists the first working healed locator so subsequent lookups
+ use it directly. Returns ``True`` if a working locator was found.
+ """
+ return self._apply_healing()
+
def _find_elements(self, wait_parent: bool = False) -> list[SeleniumWebElement | AppiumWebElement]:
"""
Find all selenium/appium elements
diff --git a/mops/selenium/elements/web_element.py b/mops/selenium/elements/web_element.py
index 98052e3f..d747347c 100644
--- a/mops/selenium/elements/web_element.py
+++ b/mops/selenium/elements/web_element.py
@@ -7,7 +7,7 @@
from mops.js_scripts import js_click
from mops.selenium.core.core_element import CoreElement
-from mops.utils.decorators import retry
+from mops.utils.decorators import healing, retry
from mops.utils.internal_utils import calculate_coordinate_to_click
from mops.utils.selector_synchronizer import get_platform_locator, set_selenium_selector
@@ -40,6 +40,7 @@ def click(self, *, force_wait: bool = True, **kwargs: Any) -> WebElement:
return self
+ @healing
@retry(JavascriptException)
def hover(self, silent: bool = False) -> WebElement:
"""
diff --git a/mops/self_healing/TODO.md b/mops/self_healing/TODO.md
new file mode 100644
index 00000000..b9dfec3b
--- /dev/null
+++ b/mops/self_healing/TODO.md
@@ -0,0 +1,55 @@
+# Self-Healing TODO
+
+Backlog of not-yet-implemented improvements for the self-healing feature.
+
+## 1. Multiple ancestors comparison (P4)
+
+**Status:** not implemented
+
+**Problem:** scoring compares only the immediate DOM parent. When a new wrapper
+(e.g. a React `InputContainer__childrenWrapper`) is inserted, the old snapshot's
+parent context (e.g. `parent-studios-add-modal` modal container) is no longer
+the direct parent, and the parent similarity drops.
+
+**Goal:** compare several ancestors up the DOM and accept a candidate if its
+ancestor chain contains a matching container (e.g. the modal), even when an
+intermediate wrapper was added.
+
+**What it touches:**
+- JS: `_GET_CANDIDATES_JS` / `_GET_ELEMENT_SNAPSHOT_JS` — collect an ancestor
+ chain (3–5 levels) instead of only the immediate parent
+- `ElementSnapshot` + snapshot JSON storage format — new `ancestors` field
+ (keep backward compatibility with old snapshots)
+- `_score_similarity` — match each snapshot ancestor against the candidate's
+ ancestor chain ("search for the modal container anywhere in the chain")
+- Breakdown: expose ancestor-level scores
+
+**Reference case (currently passing via P1–P3, would be made robust):**
+`input` inside a modal (`parent-studios-add-modal`) got wrapped by
+`InputContainer__childrenWrapper` — placeholder anchor + HTML defaults + optional
+siblings push the score above threshold, but the immediate-parent signal is lost.
+
+## 2. DOM index drift guard (temporarily removed)
+
+**Status:** removed, tests stubbed out in
+`tests/static_tests/unit/test_self_healing_scoring.py`
+
+**Problem:** the resolved element at `best_index` may no longer match the best
+candidate if the DOM changed between the candidates JS scan and `find_elements`
+(React re-renders).
+
+**Re-add:** re-snapshot the resolved element and compare with the best candidate;
+on mismatch fail with a `dom-changed-during-healing` reason.
+
+## 3. Placeholder-aware text matching (P2)
+
+**Status:** not implemented
+
+**Problem:** `{username}` in a snapshot text never matches a real value
+(`m_john_456`), and naive "normalize any text to `{username}`" risks merging
+distinct elements.
+
+**Goal:** per-placeholder matchers with regex patterns (`{username}`, `{id}`,
+`{date}`, `{token}`); `snapshot_text == '{username}'` should match only values
+that fit the username pattern, and ideally work for partial templates too
+(`"User: {username}"`).
diff --git a/mops/self_healing/__init__.py b/mops/self_healing/__init__.py
new file mode 100644
index 00000000..84981430
--- /dev/null
+++ b/mops/self_healing/__init__.py
@@ -0,0 +1,49 @@
+"""Self-healing locators for MOPS.
+
+Quick start::
+
+ from mops.self_healing import configure, JsonFileSnapshotStorage
+ configure(
+ save_snapshots=True,
+ heal_locators=True,
+ score_threshold=0.75,
+ storage=JsonFileSnapshotStorage(),
+ )
+
+Healing requires a :class:`SnapshotStorage` to be configured. The quickest way is::
+
+ configure(
+ save_snapshots=True,
+ heal_locators=True,
+ storage=JsonFileSnapshotStorage('my_snapshots'),
+ )
+"""
+
+from mops.self_healing.config import configure, get_config
+from mops.self_healing.healer import (
+ AttributeMatch,
+ FailedHealingResult,
+ Healer,
+ HealingStats,
+ ScoringWeights,
+ SimilarityBreakdown,
+ SuccessHealingResult,
+ get_healing_stats,
+)
+from mops.self_healing.snapshot import ElementSnapshot, JsonFileSnapshotStorage, SnapshotStorage
+
+__all__ = [
+ 'AttributeMatch',
+ 'ElementSnapshot',
+ 'FailedHealingResult',
+ 'Healer',
+ 'HealingStats',
+ 'JsonFileSnapshotStorage',
+ 'ScoringWeights',
+ 'SimilarityBreakdown',
+ 'SnapshotStorage',
+ 'SuccessHealingResult',
+ 'configure',
+ 'get_config',
+ 'get_healing_stats',
+]
diff --git a/mops/self_healing/config.py b/mops/self_healing/config.py
new file mode 100644
index 00000000..b399adec
--- /dev/null
+++ b/mops/self_healing/config.py
@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import dataclasses
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from mops.self_healing.healer import FailedHealingResult, ScoringWeights, SuccessHealingResult
+ from mops.self_healing.snapshot import SnapshotStorage
+
+
+@dataclass
+class SelfHealingConfig:
+ """Configuration for self-healing locators.
+
+ :param save_snapshots: When :obj:`True`, snapshots of successfully located elements
+ are saved to storage for future healing.
+ :param heal_locators: When :obj:`True`, the system attempts to heal broken locators
+ by loading saved snapshots and searching for matching elements.
+ :param score_threshold: Minimum similarity score (0-1) to accept a healed locator.
+ :param scoring_weights: Tunable :class:`ScoringWeights` for element similarity scoring.
+ Controls per-attribute, text, parent, and sibling contributions.
+ When not set, built-in defaults are used.
+ :param storage: A :class:`SnapshotStorage` instance. When not set, storage
+ remains uninitialised and neither snapshots nor healing will work.
+ External projects can pass their own backend (Redis, S3, PostgreSQL, etc.) here.
+ :param on_healing_success: Optional callback invoked when a broken locator has been
+ healed. Receives the :class:`SuccessHealingResult`.
+ :param on_healing_failure: Optional callback invoked when healing was attempted
+ but couldn't find a matching element. Receives the :class:`FailedHealingResult`.
+ """
+
+ save_snapshots: bool = False
+ heal_locators: bool = False
+ score_threshold: float = 0.7
+ scoring_weights: ScoringWeights | None = None
+ storage: SnapshotStorage | None = None
+ on_healing_success: Callable[[SuccessHealingResult], None] | None = None
+ on_healing_failure: Callable[[FailedHealingResult], None] | None = None
+
+
+_config = SelfHealingConfig()
+
+
+def configure(**kwargs: object) -> None:
+ """Update the global self-healing config.
+
+ Example::
+
+ from mops.self_healing import configure, JsonFileSnapshotStorage
+
+ # Save snapshots but don't heal (data collection)
+ configure(save_snapshots=True)
+
+ # Full healing: save snapshots AND heal broken locators
+ configure(
+ save_snapshots=True,
+ heal_locators=True,
+ score_threshold=0.75,
+ storage=JsonFileSnapshotStorage('my_snapshots'),
+ )
+
+ # Custom backend
+ configure(storage=MyCustomStorage())
+
+ # Callbacks for external integrations
+ def on_success(result: HealingResult) -> None:
+ metrics.send(...)
+
+ def on_failure(**kwargs: object) -> None:
+ # kwargs: element_name, locator_key, locator
+ post_comment_on_pr(kwargs['element_name'], kwargs['locator'])
+
+ configure(
+ save_snapshots=True,
+ heal_locators=True,
+ on_healing_success=on_success,
+ on_healing_failure=on_failure,
+ )
+ """
+ valid_fields = {f.name for f in dataclasses.fields(SelfHealingConfig)}
+ for key, value in kwargs.items():
+ if key not in valid_fields:
+ msg = f"Unknown configuration key '{key}'. Valid fields: {sorted(valid_fields)}"
+ raise ValueError(msg)
+ setattr(_config, key, value)
+
+
+def get_config() -> SelfHealingConfig:
+ """Return the current global self-healing config."""
+ return _config
diff --git a/mops/self_healing/healer.py b/mops/self_healing/healer.py
new file mode 100644
index 00000000..f43cf20c
--- /dev/null
+++ b/mops/self_healing/healer.py
@@ -0,0 +1,761 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+import logging
+import re
+from typing import TYPE_CHECKING, Any
+
+from selenium.common.exceptions import WebDriverException
+
+from mops.self_healing.snapshot import ElementSnapshot
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from mops.self_healing.snapshot import SnapshotStorage
+
+_MIN_SUBSTRING_LENGTH = 3
+_CAMEL_BOUNDARY_RE = re.compile(r'([a-z0-9])([A-Z])')
+_NON_WORD_RE = re.compile(r'[^a-zA-Z0-9]+')
+
+# Recursion guard when an element's healing needs to heal missing parents first.
+MAX_HEALING_DEPTH = 3
+
+logger = logging.getLogger('mops.self_healing')
+
+_GET_CANDIDATES_JS = """
+return (function(tag) {
+ function getAttrs(node) {
+ var attrs = {};
+ for (var i = 0; i < node.attributes.length; i++) {
+ attrs[node.attributes[i].name] = node.attributes[i].value;
+ }
+ return attrs;
+ }
+ function getSiblings(el) {
+ var parent = el.parentElement;
+ if (!parent) return [];
+ var children = parent.children;
+ var siblings = [];
+ for (var i = 0; i < children.length && siblings.length < 5; i++) {
+ if (children[i] !== el) {
+ siblings.push({
+ tag: children[i].tagName.toLowerCase(),
+ text: (children[i].textContent || '').trim().substring(0, 50),
+ attrs: getAttrs(children[i])
+ });
+ }
+ }
+ return siblings;
+ }
+ var elements = document.getElementsByTagName(tag);
+ var result = [];
+ for (var i = 0; i < elements.length; i++) {
+ var el = elements[i];
+ var parent = el.parentElement;
+ result.push({
+ index: i,
+ attrs: getAttrs(el),
+ text: (el.textContent || '').trim().substring(0, 100),
+ parentTag: parent ? parent.tagName.toLowerCase() : null,
+ parentAttrs: parent ? getAttrs(parent) : {},
+ siblings: getSiblings(el)
+ });
+ }
+ return result;
+})(arguments[0]);
+"""
+
+
+@dataclass
+class ScoringWeights:
+ """Tunable weights for the similarity scoring function.
+
+ Each weight controls how much a signal contributes to the final 0-1 score.
+ ``attribute`` is a per-attribute dict; the rest are scalar multipliers.
+ """
+
+ attribute: dict[str, float] = field(
+ default_factory=lambda: {
+ 'id': 1.0,
+ 'name': 0.7,
+ 'placeholder': 0.5,
+ 'type': 0.4,
+ 'role': 0.3,
+ 'href': 0.3,
+ 'title': 0.2,
+ 'class': 0.15,
+ }
+ )
+ text: float = 0.3
+ parent: float = 0.2
+ siblings: float = 0.15
+
+
+@dataclass
+class AttributeMatch:
+ """Comparison of a single attribute between the snapshot and a DOM candidate.
+
+ :param attribute: Attribute name (e.g. ``id``, ``class``).
+ :param snapshot_value: Value saved in the snapshot, or ``None`` when the
+ attribute is absent from the snapshot.
+ :param candidate_value: Value found on the candidate element, or ``None``
+ when the candidate does not have this attribute.
+ :param matched: ``True`` when both values exist and are exactly equal.
+ :param score: Similarity of the values — ``1.0`` on exact match, partial
+ token overlap for weighted attributes, ``0.0`` otherwise.
+ :param weight: Configured :class:`ScoringWeights` weight. ``0.0`` when the
+ attribute does not participate in scoring (diagnostics only).
+ """
+
+ attribute: str
+ snapshot_value: str | None
+ candidate_value: str | None
+ matched: bool
+ score: float
+ weight: float
+
+
+@dataclass
+class SimilarityBreakdown:
+ """Per-signal similarity breakdown of one DOM candidate vs the snapshot.
+
+ Exposed on :class:`SuccessHealingResult` and :class:`FailedHealingResult`
+ for the best-scoring candidate so callers can see exactly which attributes
+ matched and which did not — useful for spotting dynamic data (e.g. a
+ changing ``id`` or a CSS-module hash in ``class``).
+
+ :param candidate_snapshot: Raw DOM snapshot of the best candidate as found
+ on the page. Unlike the normalized reference snapshot stored in the
+ storage, this reflects the actual current state of the element — useful
+ for comparing dynamic values side-by-side.
+ """
+
+ score: float
+ attributes: dict[str, AttributeMatch]
+ text_snapshot: str | None
+ text_candidate: str | None
+ text_score: float | None
+ parent_tag_snapshot: str | None
+ parent_tag_candidate: str | None
+ parent_tag_matched: bool | None
+ parent_attrs_score: float | None
+ siblings_score: float | None
+ siblings_snapshot_count: int
+ siblings_candidate_count: int
+ candidate_snapshot: ElementSnapshot | None = None
+
+ @property
+ def matched_attributes(self) -> list[str]:
+ """Names of attributes that matched exactly."""
+ return [attr for attr, match in self.attributes.items() if match.matched]
+
+ @property
+ def mismatched_attributes(self) -> list[str]:
+ """Names of attributes that did not match exactly."""
+ return [attr for attr, match in self.attributes.items() if not match.matched]
+
+
+@dataclass
+class SuccessHealingResult:
+ element_name: str
+ original_locator: str
+ healed_locator: str | None
+ healed_locators_candidates: list[str]
+ score: float
+ locator_key: str = ''
+ candidates_count: int | None = None
+ timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+ breakdown: SimilarityBreakdown | None = None
+
+
+@dataclass
+class FailedHealingResult:
+ element_name: str
+ locator_key: str
+ locator: str
+ reason: str
+ error: str | None = None
+ best_score: float | None = None
+ score_threshold: float | None = None
+ candidates_count: int | None = None
+ breakdown: SimilarityBreakdown | None = None
+
+
+@dataclass
+class HealingStats:
+ """Process-wide self-healing counters.
+
+ Updated automatically by the :class:`Healer` on every :meth:`Healer.heal` call.
+ Read the current values with :func:`get_healing_stats`.
+
+ .. note::
+ ``healed`` counts candidates found by the Healer. A candidate can still
+ fail DOM verification afterwards (``no-verified-locator``), which is
+ reported via ``on_healing_failure`` but is not reflected in these
+ counters — it happens outside the :class:`Healer`.
+ """
+
+ attempts: int = 0
+ healed: int = 0
+ failed: int = 0
+ failed_reasons: dict[str, int] = field(default_factory=dict)
+ _score_sum: float = 0.0
+ _score_count: int = 0
+
+ @property
+ def avg_best_score(self) -> float | None:
+ """Average best similarity score of successfully healed elements."""
+ if self._score_count == 0:
+ return None
+ return self._score_sum / self._score_count
+
+
+_stats = HealingStats()
+
+
+def get_healing_stats() -> HealingStats:
+ """Return process-wide self-healing statistics.
+
+ Useful for end-of-run reporting, e.g. in ``pytest_sessionfinish``::
+
+ from mops.self_healing import get_healing_stats
+
+ def pytest_sessionfinish(session, exitstatus):
+ stats = get_healing_stats()
+ print(f'healing: {stats.healed} ok, {stats.failed} failed of {stats.attempts}')
+ """
+ return _stats
+
+
+def _record_attempt() -> None:
+ _stats.attempts += 1
+
+
+def _record_heal_success(score: float) -> None:
+ _stats.healed += 1
+ _stats._score_sum += score
+ _stats._score_count += 1
+
+
+def _record_heal_failure(reason: str) -> None:
+ _stats.failed += 1
+ _stats.failed_reasons[reason] = _stats.failed_reasons.get(reason, 0) + 1
+
+
+class Healer:
+ """Orchestrates the self-healing process for a failed element lookup."""
+
+ def __init__(
+ self,
+ storage: SnapshotStorage,
+ score_threshold: float,
+ scoring_weights: ScoringWeights | None = None,
+ on_healing_failure: Callable[[FailedHealingResult], None] | None = None,
+ ) -> None:
+ self._storage = storage
+ self._score_threshold = score_threshold
+ self._scoring_weights = scoring_weights or ScoringWeights()
+ self._on_healing_failure = on_healing_failure
+
+ def _fail(
+ self,
+ reason: str,
+ element_name: str,
+ locator_key: str,
+ locator: str,
+ exc: BaseException | None = None,
+ best_score: float | None = None,
+ candidates_count: int | None = None,
+ breakdown: SimilarityBreakdown | None = None,
+ ) -> None:
+ """Fire failure callback, record stats, and return None.
+
+ :param best_score: Highest similarity score found before the failure,
+ or ``None`` when no candidate was scored (e.g. no snapshot,
+ script error, or empty candidate list).
+ :param candidates_count: Number of DOM candidates scored, or ``None``
+ when candidates were never collected.
+ :param breakdown: Similarity breakdown of the best candidate, or
+ ``None`` when no candidate was scored.
+ """
+ error: str | None = None
+ if exc:
+ error = exc.msg if isinstance(exc, WebDriverException) else str(exc)
+ result = FailedHealingResult(
+ element_name=element_name,
+ locator_key=locator_key,
+ locator=locator,
+ reason=reason,
+ error=error,
+ best_score=best_score,
+ score_threshold=self._score_threshold,
+ candidates_count=candidates_count,
+ breakdown=breakdown,
+ )
+ _record_heal_failure(reason)
+ if self._on_healing_failure:
+ self._on_healing_failure(result)
+
+ def heal(
+ self,
+ element_name: str,
+ locator_key: str,
+ locator: str,
+ driver_wrapper: Any,
+ find_elements_fn: Callable[[str], list[Any]],
+ generate_locator_fn: Callable[[Any, Any], list[str]],
+ ) -> SuccessHealingResult | None:
+ """Try to find a healed locator for a failed element lookup.
+
+ :param element_name: Human-readable element name for logging.
+ :param locator_key: Storage key used to load the saved snapshot.
+ :param locator: The original locator string (for the result record).
+ :param driver_wrapper: Driver wrapper with an ``execute_script(script, *args)`` method.
+ :param find_elements_fn: Callback ``(tag: str) -> list`` to find all
+ elements with a given tag name (backend-specific).
+ :param generate_locator_fn: Callback ``(element, driver_wrapper) -> list[str]``
+ to generate candidate locators from a live element (backend-specific).
+ :return: :class:`SuccessHealingResult` if healed, ``None`` otherwise.
+ """
+ _record_attempt()
+ snapshot = self._storage.load(locator_key)
+
+ if not snapshot:
+ logger.info('Self-healing: no snapshot for "%s", skipping', element_name)
+ return self._fail('no-snapshot', element_name, locator_key, locator)
+
+ candidates = self._collect_candidates(driver_wrapper, snapshot.tag, element_name, locator_key, locator)
+ if candidates is None:
+ return None
+
+ scored = self._score_candidates(candidates, snapshot, element_name, locator_key, locator)
+ if scored is None:
+ return None
+
+ best_score, best_index, best_breakdown = scored
+ healed_locators = self._resolve_element(
+ driver_wrapper,
+ find_elements_fn,
+ generate_locator_fn,
+ snapshot.tag,
+ best_index,
+ best_score,
+ best_breakdown,
+ candidates,
+ element_name,
+ locator_key,
+ locator,
+ )
+ if healed_locators is None:
+ return None
+
+ _record_heal_success(best_score)
+ result = SuccessHealingResult(
+ element_name=element_name,
+ original_locator=locator,
+ healed_locator=None,
+ healed_locators_candidates=healed_locators,
+ score=best_score,
+ locator_key=locator_key,
+ candidates_count=len(candidates),
+ breakdown=best_breakdown,
+ )
+
+ logger.info(
+ 'Self-healing: candidate locators generated for "%s": %s (score=%.2f)',
+ element_name,
+ healed_locators,
+ best_score,
+ )
+
+ return result
+
+ def _collect_candidates(
+ self,
+ driver_wrapper: Any,
+ tag: str,
+ element_name: str,
+ locator_key: str,
+ locator: str,
+ ) -> list[dict] | None:
+ """Collect DOM candidates of the same tag; returns ``None`` on failure."""
+ try:
+ candidates: list[dict] = driver_wrapper.execute_script(_GET_CANDIDATES_JS, tag)
+ except WebDriverException as exc:
+ logger.info('Self-healing: failed to get candidates for "%s": %s', element_name, exc)
+ return self._fail('candidates-script-error', element_name, locator_key, locator, exc=exc)
+
+ if not candidates:
+ return self._fail('no-candidates', element_name, locator_key, locator, candidates_count=0)
+
+ return candidates
+
+ def _score_candidates(
+ self,
+ candidates_data: list[dict],
+ snapshot: ElementSnapshot,
+ element_name: str,
+ locator_key: str,
+ locator: str,
+ ) -> tuple[float, int, SimilarityBreakdown] | None:
+ """Score all candidates and return the best ``(score, index, breakdown)``.
+
+ Returns ``None`` when the best score is below the threshold.
+ """
+ best_score = -1.0
+ best_index = -1
+ best_breakdown: SimilarityBreakdown | None = None
+
+ for item in candidates_data:
+ # Normalize the candidate with the same rules applied to snapshots,
+ # so both sides of the comparison are cleaned equally.
+ raw_candidate_snapshot = _candidate_to_snapshot(item, snapshot.tag)
+ normalized_candidate_snapshot = self._storage.normalize_snapshot(raw_candidate_snapshot)
+
+ breakdown = _compute_similarity_breakdown(
+ normalized_candidate_snapshot,
+ snapshot,
+ self._scoring_weights,
+ candidate_snapshot=raw_candidate_snapshot,
+ )
+ if breakdown.score > best_score:
+ best_score = breakdown.score
+ best_index = item['index']
+ best_breakdown = breakdown
+
+ if best_score < self._score_threshold or best_index < 0:
+ logger.info(
+ 'Self-healing: best score %.2f below threshold %.2f for "%s"',
+ best_score,
+ self._score_threshold,
+ element_name,
+ )
+ self._fail(
+ 'below-threshold',
+ element_name,
+ locator_key,
+ locator,
+ best_score=best_score,
+ candidates_count=len(candidates_data),
+ breakdown=best_breakdown,
+ )
+ return None
+
+ return best_score, best_index, best_breakdown
+
+ def _resolve_element(
+ self,
+ driver_wrapper: Any,
+ find_elements_fn: Callable[[str], list[Any]],
+ generate_locator_fn: Callable[[Any, Any], list[str]],
+ tag: str,
+ best_index: int,
+ best_score: float,
+ best_breakdown: SimilarityBreakdown | None,
+ candidates_data: list[dict],
+ element_name: str,
+ locator_key: str,
+ locator: str,
+ ) -> list[str] | None:
+ """Resolve the best-index element and generate locators for it.
+
+ Returns the generated locator list, or ``None`` on failure (the failure
+ callback is fired inside).
+ """
+ try:
+ web_elements = find_elements_fn(tag)
+ if best_index >= len(web_elements):
+ self._fail(
+ 'index-out-of-bounds',
+ element_name,
+ locator_key,
+ locator,
+ best_score=best_score,
+ candidates_count=len(candidates_data),
+ breakdown=best_breakdown,
+ )
+ return None
+ healed_locators = generate_locator_fn(web_elements[best_index], driver_wrapper)
+ except Exception as exc: # noqa: BLE001
+ logger.info('Self-healing: failed to generate locator for "%s": %s', element_name, exc)
+ self._fail(
+ 'generate-locator-error',
+ element_name,
+ locator_key,
+ locator,
+ exc=exc,
+ best_score=best_score,
+ candidates_count=len(candidates_data),
+ breakdown=best_breakdown,
+ )
+ return None
+
+ if healed_locators is None:
+ _record_heal_failure('no-generated-locator')
+ return None
+
+ return healed_locators
+
+
+def _candidate_to_snapshot(candidate: dict[str, Any], tag: str) -> ElementSnapshot:
+ """Convert a raw DOM candidate dict (from the candidates JS) into an ElementSnapshot."""
+ return ElementSnapshot(
+ tag=tag,
+ attributes=candidate.get('attrs', {}),
+ text=candidate.get('text', ''),
+ parent_tag=candidate.get('parentTag'),
+ parent_attributes=candidate.get('parentAttrs', {}),
+ siblings=candidate.get('siblings', []),
+ )
+
+
+def _compute_similarity_breakdown( # noqa: PLR0912
+ candidate: ElementSnapshot,
+ snapshot: ElementSnapshot,
+ weights: ScoringWeights | None = None,
+ candidate_snapshot: ElementSnapshot | None = None,
+) -> SimilarityBreakdown:
+ """Compute a similarity score and a per-signal breakdown for one candidate.
+
+ The returned :class:`SimilarityBreakdown` contains the aggregate 0-1 ``score``
+ plus an ``AttributeMatch`` for every snapshot attribute (and every weighted
+ attribute) so callers can see which attributes matched and which did not.
+
+ :param candidate: The normalized candidate snapshot to score — already cleaned
+ with the same rules as *snapshot* (see :meth:`SnapshotStorage.normalize_snapshot`).
+ :param candidate_snapshot: Optional raw (non-normalized) snapshot of the
+ candidate as found on the page, attached to the breakdown for diagnostics.
+ Defaults to *candidate*.
+ """
+ w = weights or ScoringWeights()
+ score = 0.0
+ total_weight = 0.0
+ attributes: dict[str, AttributeMatch] = {}
+
+ # Attribute matching — all snapshot attributes plus weighted ones present on
+ # the candidate. Only weighted attributes contribute to the total score.
+ weighted_attrs = set(w.attribute)
+ for attr in set(snapshot.attributes) | weighted_attrs:
+ snap_val = snapshot.attributes.get(attr)
+ cand_val = candidate.attributes.get(attr)
+ weight = w.attribute.get(attr, 0.0)
+
+ if snap_val is None and cand_val is None:
+ continue
+
+ if weight:
+ total_weight += weight
+ attr_score = _attribute_similarity(attr, snap_val, cand_val, tag=snapshot.tag)
+ score += weight * attr_score
+ else:
+ # Unweighted attribute — diagnostics only, binary match indicator
+ attr_score = 1.0 if snap_val is not None and snap_val == cand_val else 0.0
+
+ attributes[attr] = AttributeMatch(
+ attribute=attr,
+ snapshot_value=snap_val,
+ candidate_value=cand_val,
+ matched=snap_val is not None and snap_val == cand_val,
+ score=attr_score,
+ weight=weight,
+ )
+
+ # Text similarity
+ snap_text = snapshot.text
+ cand_text = candidate.text
+ text_score: float | None = None
+ if snap_text:
+ total_weight += w.text
+ if snap_text == cand_text:
+ text_score = 1.0
+ elif snap_text and cand_text:
+ text_score = _text_similarity(snap_text, cand_text)
+ else:
+ text_score = 0.0
+ score += w.text * text_score
+
+ # Parent tag match. An empty ``snapshot.parent_attributes`` means the snapshot
+ # carries no parent attribute info — that is an absence of data, not a
+ # mismatch, so it is never penalized.
+ parent_tag_matched: bool | None = None
+ parent_attrs_score: float | None = None
+ if snapshot.parent_tag and candidate.parent_tag:
+ total_weight += w.parent
+ parent_tag_matched = candidate.parent_tag == snapshot.parent_tag
+ if parent_tag_matched:
+ score += w.parent * 0.5
+ if snapshot.parent_attributes:
+ parent_attrs_score = _attrs_overlap(snapshot.parent_attributes, candidate.parent_attributes)
+ score += w.parent * 0.5 * parent_attrs_score
+
+ # Sibling similarity is optional: a candidate may lose its siblings entirely
+ # (e.g. a new React wrapper), so an empty candidate sibling list is treated as
+ # missing data rather than a mismatch.
+ snap_siblings = snapshot.siblings
+ cand_siblings = candidate.siblings
+ siblings_score: float | None = None
+ if snap_siblings and cand_siblings:
+ total_weight += w.siblings
+ siblings_score = _siblings_similarity(snap_siblings, cand_siblings)
+ score += w.siblings * siblings_score
+
+ final_score = 0.0 if total_weight == 0 else score / total_weight
+
+ if candidate_snapshot is None:
+ candidate_snapshot = candidate
+
+ return SimilarityBreakdown(
+ score=final_score,
+ attributes=attributes,
+ text_snapshot=snap_text,
+ text_candidate=cand_text,
+ text_score=text_score,
+ parent_tag_snapshot=snapshot.parent_tag,
+ parent_tag_candidate=candidate.parent_tag,
+ parent_tag_matched=parent_tag_matched,
+ parent_attrs_score=parent_attrs_score,
+ siblings_score=siblings_score,
+ siblings_snapshot_count=len(snap_siblings),
+ siblings_candidate_count=len(cand_siblings),
+ candidate_snapshot=candidate_snapshot,
+ )
+
+
+def _token_overlap(a: str, b: str) -> float:
+ """Jaccard token overlap for strings (e.g. space-separated value lists)."""
+ a_tokens = set(a.split())
+ b_tokens = set(b.split())
+ if not a_tokens or not b_tokens:
+ return 0.0
+ intersection = a_tokens & b_tokens
+ union = a_tokens | b_tokens
+ return len(intersection) / len(union)
+
+
+def _class_to_tokens(class_str: str) -> set[str]:
+ """Split a CSS class into canonical lowercase word tokens.
+
+ Handles kebab-case, snake_case, camelCase, PascalCase and BEM/CSS-module
+ separators (``_``, ``__``, ``--``)::
+
+ 'user-profile-card' -> {'user', 'profile', 'card'}
+ 'UserProfileCard' -> {'user', 'profile', 'card'}
+ 'checkout_form_submit' -> {'checkout', 'form', 'submit'}
+ 'checkoutFormSubmit' -> {'checkout', 'form', 'submit'}
+ 'modal__close-btn' -> {'modal', 'close', 'btn'}
+ 'ModalCloseBtn' -> {'modal', 'close', 'btn'}
+
+ So classes that only differ in case or word separators compare as equal.
+ """
+ # split camelCase / PascalCase word boundaries first
+ split_case = _CAMEL_BOUNDARY_RE.sub(r'\1 \2', class_str)
+ # then split on any non-alphanumeric separator (kebab, snake, __, ...)
+ tokens = _NON_WORD_RE.split(split_case)
+ return {token.lower() for token in tokens if token}
+
+
+def _class_similarity(a: str, b: str) -> float:
+ """Jaccard similarity of two CSS classes after canonical word tokenization."""
+ a_tokens = _class_to_tokens(a)
+ b_tokens = _class_to_tokens(b)
+ if not a_tokens or not b_tokens:
+ return 0.0
+ if a_tokens == b_tokens:
+ return 1.0
+ intersection = a_tokens & b_tokens
+ union = a_tokens | b_tokens
+ return len(intersection) / len(union)
+
+
+_HTML_DEFAULT_ATTRS: dict[str, dict[str, str]] = {
+ 'input': {'type': 'text'},
+ 'button': {'type': 'submit'},
+ 'script': {'type': 'text/javascript'},
+ 'style': {'type': 'text/css'},
+ 'link': {'type': 'text/css'},
+ 'ol': {'type': '1'},
+}
+
+
+def _attribute_similarity(
+ attr: str,
+ snap_val: str | None,
+ cand_val: str | None,
+ tag: str | None = None,
+) -> float:
+ """0-1 similarity of a single attribute value between snapshot and candidate.
+
+ ``class`` uses canonical word-token comparison; everything else falls back to
+ exact match (``1.0``) or token overlap for partial matches.
+
+ HTML defaults are applied when *tag* is known: e.g. an ```` without a
+ ``type`` attribute is ``type="text"``, a ``