fix: harden publishing against DOM variants

src/kleinanzeigen_bot/download_flow.py

@@ -78,7 +78,7 @@ async def _download_ad_with_resolved_state(
         published_ad = published_ads_by_id.get(ad_id, {})
         LOG.debug("Ad %d has state '%s'. Saving as inactive.", ad_id, published_ad.get("state", "unknown"))
 
-    await ad_extractor.download_ad(ad_id, active = resolved.active)
+    await ad_extractor.download_ad(ad_id, active = resolved.active, owned_overview = True)
 
 
 async def _fetch_published_ads_by_id(

src/kleinanzeigen_bot/extract.py

@@ -41,6 +41,10 @@
 _LOG_SNIPPET_LIMIT:Final[int] = 120
 _ELLIPSIS:Final[str] = "..."
 _ELLIPSIS_LEN:Final[int] = len(_ELLIPSIS)
+_OWNED_AD_TITLE_DECORATION_PREFIXES:Final[tuple[str, ...]] = ("Gelöscht",)
+_OWNED_AD_TITLE_DECORATION_RE:Final[re.Pattern[str]] = re.compile(
+    rf"^(?:{'|'.join(re.escape(prefix) for prefix in _OWNED_AD_TITLE_DECORATION_PREFIXES)})\s*•\s*"
+)
 _CONDITION_DISPLAY_TO_API:Final[dict[str, str]] = {
     "neu": "new",
     "sehr gut": "like_new",
@@ -282,20 +286,32 @@ def _render_download_ad_file_stem(self, ad_id:int, title:str) -> str:
     def _render_download_folder_name(self, ad_id:int, title:str) -> str:
         return self._render_download_name_with_budget(self.config.download.folder_name_template, ad_id, title, self.config.download.folder_name_max_length)
 
-    async def download_ad(self, ad_id:int, *, active:bool | None = None) -> None:
+    async def download_ad(
+        self,
+        ad_id:int,
+        *,
+        active:bool | None = None,
+        owned_overview:bool = False,
+    ) -> None:
         """
         Downloads an ad to a specific location, specified by config and ad ID.
         NOTE: Requires that the driver session currently is on the ad page.
 
         :param ad_id: the ad ID
         :param active: optional override for ad activity state
+        :param owned_overview: whether the ad was discovered in the user's overview
         """
 
         download_dir = self.download_dir
         LOG.info("Using download directory: %s", download_dir)
 
         # Extract ad info into a staging directory and determine final target directory
-        ad_cfg, staging_dir, final_dir, ad_file_stem = await self._extract_ad_page_info_with_directory_handling(download_dir, ad_id, active_override = active)
+        ad_cfg, staging_dir, final_dir, ad_file_stem = await self._extract_ad_page_info_with_directory_handling(
+            download_dir,
+            ad_id,
+            active_override = active,
+            owned_overview = owned_overview,
+        )
 
         # Preserve local-only settings when re-downloading an existing ad
         await self._preserve_local_settings_from_existing_ad(ad_cfg, final_dir, ad_file_stem, ad_id)
@@ -614,15 +630,23 @@ async def _extract_title_from_ad_page(self) -> str:
         """
         return await self.web_text(By.ID, "viewad-title")
 
-    async def _resolve_download_title(self, ad_id:int) -> str:
+    async def _resolve_download_title(self, ad_id:int, *, owned_overview:bool = False) -> str:
         """Return the canonical title for a downloaded ad."""
         cached_ad = self.published_ads_by_id.get(ad_id)
         if cached_ad is not None:
             title = cached_ad.get("title")
             if isinstance(title, str) and title.strip():
                 return html.unescape(title.strip())
 
-        return await self._extract_title_from_ad_page()
+        page_title = await self._extract_title_from_ad_page()
+        if not owned_overview:
+            return page_title
+
+        cleaned_title = _OWNED_AD_TITLE_DECORATION_RE.sub("", page_title, count = 1).strip()
+        if cleaned_title and cleaned_title != page_title:
+            LOG.info("Removed owned-ad status decoration from downloaded ad %s title.", ad_id)
+            return cleaned_title
+        return page_title
 
     async def _extract_ad_page_info(
         self,
@@ -716,16 +740,18 @@ async def _extract_ad_page_info_with_directory_handling(
         ad_id:int,
         *,
         active_override:bool | None = None,
+        owned_overview:bool = False,
     ) -> tuple[AdPartial, Path, Path, str]:
         """
         Extracts ad information and handles directory creation/renaming.
 
         :param relative_directory: Base directory for downloads
         :param ad_id: The ad ID
         :param active_override: optional override for ad activity state
+        :param owned_overview: whether the ad was discovered in the user's overview
         :return: AdPartial with staging/final directory information and rendered ad file stem
         """
-        title = await self._resolve_download_title(ad_id)
+        title = await self._resolve_download_title(ad_id, owned_overview = owned_overview)
         LOG.info('Resolved title for ad %s: "%s"', ad_id, title)
 
         # Determine the final directory path

src/kleinanzeigen_bot/publishing_form.py

@@ -35,11 +35,17 @@
 from .utils.web_scraping_mixin import By, Element, Is, WebScrapingMixin
 
 LOG:Final[_loggers.Logger] = _loggers.get_logger(__name__)
+_OPEN_SHIPPING_DIALOG_XPATH:Final[str] = '//*[self::dialog[@open] or (@role="dialog" and not(@aria-hidden="true"))]'
 _OTHER_SHIPPING_METHODS_XPATH:Final[str] = (
-    '//*[self::dialog[@open] or (@role="dialog" and not(@aria-hidden="true"))]'
-    '//*[contains(normalize-space(.), "Andere Versandmethoden")'
+    _OPEN_SHIPPING_DIALOG_XPATH
+    + '//*[contains(normalize-space(.), "Andere Versandmethoden")'
     ' and not(.//*[contains(normalize-space(.), "Andere Versandmethoden")])]'
 )
+_SHIPPING_SIZE_RADIO_XPATH:Final[str] = (
+    f'{_OPEN_SHIPPING_DIALOG_XPATH}//input[@type="radio" and '
+    '(@value="SMALL" or @value="MEDIUM" or @value="LARGE")]'
+)
+_SHIPPING_BACK_XPATH:Final[str] = f'{_OPEN_SHIPPING_DIALOG_XPATH}//button[contains(., "Zurück")]'
 
 
 async def set_category(web:WebScrapingMixin, *, root_url:str, category:str | None, ad_file:str) -> None:
@@ -730,23 +736,53 @@ async def _set_configured_shipping_options(
         LOG.debug("Shipping enabled toggle not found before options dialog: %s", ex)
     await web.web_click(By.ID, "ad-shipping-options")
 
-    if mode == AdUpdateStrategy.MODIFY:
-        try:
-            await web.web_find(By.XPATH, _OTHER_SHIPPING_METHODS_XPATH, timeout = short_timeout)
-        except TimeoutError:
-            await web.web_click(By.XPATH, '//button[contains(., "Zurück")]', timeout = short_timeout)
-            try:
-                await web.web_find(By.XPATH, _OTHER_SHIPPING_METHODS_XPATH, timeout = short_timeout)
-            except TimeoutError:
-                await web.web_click(By.XPATH, '//button[contains(., "Zurück")]', timeout = short_timeout)
-
-    # The redesign has rendered this action as different element types. Click
-    # its deepest text-bearing node so the event bubbles to whichever clickable
-    # ancestor the current dialog uses.
-    await web.web_click(By.XPATH, _OTHER_SHIPPING_METHODS_XPATH, timeout = short_timeout)
+    await _open_shipping_size_selection(web, short_timeout)
     await set_shipping_options(web, ad_cfg, mode)
 
 
+async def _open_shipping_size_selection(
+    web:WebScrapingMixin,
+    short_timeout:int | float,
+) -> None:
+    """Reach a supported shipping size pane across Kleinanzeigen A/B variants.
+
+    A size pane may be visible directly or behind ``Andere Versandmethoden``.
+    Nested panes are unwound by at most two back steps; unsupported dialog
+    states fail with a clear timeout error.
+
+    Examples:
+        A direct size pane returns immediately. An alternate-methods action is
+        opened before returning. A nested carrier pane is unwound with bounded
+        back navigation. A dialog exposing none of these routes times out.
+    """
+    max_back_steps = 2
+    try:
+        for back_steps in range(max_back_steps + 1):
+            size_radio = await web.web_probe(By.XPATH, _SHIPPING_SIZE_RADIO_XPATH, timeout = short_timeout)
+            if size_radio is not None:
+                LOG.debug("Shipping dialog route: direct size selection (%s back step(s)).", back_steps)
+                return
+
+            other_methods = await web.web_probe(By.XPATH, _OTHER_SHIPPING_METHODS_XPATH, timeout = short_timeout)
+            if other_methods is not None:
+                await web.web_click(By.XPATH, _OTHER_SHIPPING_METHODS_XPATH, timeout = short_timeout)
+                await web.web_find(By.XPATH, _SHIPPING_SIZE_RADIO_XPATH, timeout = short_timeout)
+                LOG.debug("Shipping dialog route: Andere Versandmethoden (%s back step(s)).", back_steps)
+                return
+
+            if back_steps == max_back_steps:
+                break
+            back_button = await web.web_probe(By.XPATH, _SHIPPING_BACK_XPATH, timeout = short_timeout)
+            if back_button is None:
+                break
+            await web.web_click(By.XPATH, _SHIPPING_BACK_XPATH, timeout = short_timeout)
+            await web.web_sleep(300, 500)
+    except TimeoutError as ex:
+        raise TimeoutError(_("Failed to configure shipping options in dialog!")) from ex
+
+    raise TimeoutError(_("Failed to configure shipping options in dialog!"))
+
+
 async def _enable_platform_default_shipping(
     web:WebScrapingMixin, short_timeout:int | float
 ) -> None:
@@ -782,7 +818,7 @@ async def set_shipping_options(web:WebScrapingMixin, ad_cfg:Ad, mode:AdUpdateStr
     all_codes_for_size = CARRIER_CODES_BY_SIZE[shipping_size]
 
     short_timeout = web.timeout("quick_dom")
-    dialog = '//*[self::dialog or @role="dialog"]'
+    dialog = _OPEN_SHIPPING_DIALOG_XPATH
 
     try:
         # Select the size group via radio button value (e.g. "SMALL", "MEDIUM", "LARGE")

src/kleinanzeigen_bot/publishing_submission.py

@@ -11,10 +11,11 @@
 import re
 import urllib.parse as urllib_parse
 from gettext import gettext as _
+from typing import Final
 
 from nodriver.core.connection import ProtocolException
 
-from . import captcha_flow
+from . import captcha_flow, published_ads
 from .model.ad_model import Ad, AdUpdateStrategy
 from .model.config_model import CaptchaConfig
 from .utils import loggers as _loggers
@@ -23,6 +24,78 @@
 from .utils.web_scraping_mixin import By, WebScrapingMixin
 
 LOG = _loggers.get_logger(__name__)
+_PUBLISHED_AD_RECOVERY_DELAYS_MS:Final[tuple[int, ...]] = (0, 1_000, 2_000, 4_000)
+
+
+async def _is_idless_publish_success_page(web:WebScrapingMixin) -> bool:
+    """Detect the redesigned successful publish page that exposes no ad ID."""
+    try:
+        result = await web.web_execute(r"""
+(() => {
+    const bodyText = (document.body?.innerText || '').replace(/\s+/g, ' ').trim();
+    const hasManageAdsControl = [...document.querySelectorAll('a, button')].some((element) =>
+        (element.innerText || '').replace(/\s+/g, ' ').trim().includes('Zu meinen Anzeigen')
+    );
+    return bodyText.includes('Geschafft!') && hasManageAdsControl;
+})()
+""")
+    except (TimeoutError, ProtocolException):
+        return False
+    return result is True
+
+
+async def _try_recover_ad_id_from_published_ads(
+    web:WebScrapingMixin,
+    *,
+    root_url:str,
+    title:str,
+    known_published_ad_ids:frozenset[int] | None,
+) -> int | None:
+    """Recover one newly published exact-title ad from a complete list.
+
+    Recovery makes four strict fetch attempts, after delays of 0, 1, 2, and 4
+    seconds. It returns ``None`` when the pre-submit baseline is unavailable,
+    no unique match appears, or multiple matching candidates make the result
+    ambiguous.
+    """
+    if known_published_ad_ids is None:
+        LOG.warning("Published-ad ID recovery skipped because the pre-submit list was incomplete")
+        return None
+
+    for delay_ms in _PUBLISHED_AD_RECOVERY_DELAYS_MS:
+        if delay_ms:
+            await web.web_sleep(delay_ms)
+        try:
+            current_ads = await published_ads.fetch_published_ads(web, root_url, strict = True)
+        except published_ads.PublishedAdsFetchIncompleteError as ex:
+            LOG.debug("Strict published-ad recovery fetch failed: %s", ex)
+            continue
+
+        candidates:set[int] = set()
+        for published_ad in current_ads:
+            if published_ad.get("title") != title:
+                continue
+            raw_id = published_ad.get("id")
+            if raw_id is None:
+                continue
+            try:
+                candidate_id = int(raw_id)
+            except (TypeError, ValueError):
+                continue
+            if candidate_id not in known_published_ad_ids:
+                candidates.add(candidate_id)
+
+        if len(candidates) == 1:
+            return next(iter(candidates))
+        if len(candidates) > 1:
+            LOG.warning(
+                "Published-ad ID recovery was ambiguous for '%s'; refusing candidate IDs: %s",
+                title,
+                sorted(candidates),
+            )
+            return None
+
+    return None
 
 
 async def _try_recover_ad_id_from_redirect(
@@ -91,6 +164,8 @@ async def submit_and_confirm_ad(
     mode:AdUpdateStrategy,
     *,
     captcha_config:CaptchaConfig,
+    root_url:str,
+    known_published_ad_ids:frozenset[int] | None = None,
 ) -> int:
     """Submit the ad form, handle post-submit dialogs, wait for confirmation,
     and extract the published ad ID.
@@ -130,6 +205,7 @@ async def submit_and_confirm_ad(
 
     # Everything after the first click is uncertain: the ad may already have been submitted.
     ad_id:int | None = None
+    idless_success_detected = False
     try:
         quick_dom = web.timeout("quick_dom")
 
@@ -171,34 +247,65 @@ async def submit_and_confirm_ad(
 
         confirmation_timeout = web.timeout("publishing_confirmation")
 
-        async def _check_confirmation_url() -> bool:
+        async def _check_confirmation_state() -> bool:
+            nonlocal idless_success_detected
             url = str(await web.web_execute("window.location.href"))
-            return "p-anzeige-aufgeben-bestaetigung.html?adId=" in url
-
-        await web.web_await(_check_confirmation_url, timeout = confirmation_timeout)
-
-        # extract the ad id from the URL's query parameter (use JS for fresh URL, not stale page url)
-        current_url = str(await web.web_execute("window.location.href"))
-        current_url_query_params = urllib_parse.parse_qs(urllib_parse.urlparse(current_url).query)
-        ad_id = int(current_url_query_params.get("adId", [])[0])
+            if "p-anzeige-aufgeben-bestaetigung.html?adId=" in url:
+                return True
+            if mode == AdUpdateStrategy.REPLACE and await _is_idless_publish_success_page(web):
+                idless_success_detected = True
+                return True
+            return False
+
+        await web.web_await(_check_confirmation_state, timeout = confirmation_timeout)
+
+        if idless_success_detected:
+            try:
+                ad_id = await _try_recover_ad_id_from_published_ads(
+                    web,
+                    root_url = root_url,
+                    title = ad_cfg.title,
+                    known_published_ad_ids = known_published_ad_ids,
+                )
+            except Exception as recovery_ex:  # noqa: BLE001
+                LOG.debug("Published-ad list fallback failed: %s", recovery_ex)
+                raise PublishSubmissionUncertainError(
+                    "publish succeeded but no ad ID could be recovered"
+                ) from recovery_ex
+            if ad_id is None:
+                raise PublishSubmissionUncertainError(
+                    "publish succeeded but no ad ID could be recovered"
+                )
+            LOG.warning(
+                "Confirmation page exposed no ad ID; recovered ad ID %s from the published ads list",
+                ad_id,
+            )
+        else:
+            # Use the live URL because the page object URL may be stale after redirects.
+            current_url = str(await web.web_execute("window.location.href"))
+            current_url_query_params = urllib_parse.parse_qs(urllib_parse.urlparse(current_url).query)
+            ad_id = int(current_url_query_params.get("adId", [])[0])
 
     except (TimeoutError, ProtocolException, IndexError, ValueError, TypeError) as ex:
         # The confirmation page may have auto-redirected before we could poll it,
         # or the URL was redirected between polling and extraction (race condition).
         # Try to recover the ad ID from tracking data on the current page.
         LOG.debug("Confirmation URL polling or extraction failed (%s), attempting tracking data fallback...", type(ex).__name__)
+        recovered_from_tracking = False
         try:
             ad_id = await _try_recover_ad_id_from_redirect(web, pre_submit_referrer = pre_submit_referrer)
+            recovered_from_tracking = ad_id is not None
         except Exception as fallback_ex:  # noqa: BLE001
             LOG.debug("Tracking data fallback failed: %s", fallback_ex)
 
         if ad_id is None:
             raise PublishSubmissionUncertainError("submission may have succeeded before failure") from ex
 
-        LOG.warning(
-            "Confirmation page redirected too fast; extracted ad ID %s from page tracking data",
-            ad_id,
-        )
+        if recovered_from_tracking:
+            LOG.warning(
+                "Confirmation page redirected too fast; extracted ad ID %s from page tracking data",
+                ad_id,
+            )
 
     # Defensive guard: ad_id must be set by now — either from the confirmation URL
     # (try block) or the tracking fallback (except block). The except block always