feat: keep login selector fallbacks close to auth flow (#855)

This commit is contained in:
Jens
2026-03-02 13:01:05 +01:00
committed by GitHub
parent c4a2d1c4f5
commit fa9df6fca4
3 changed files with 82 additions and 18 deletions

View File

@@ -7,7 +7,7 @@ import urllib.parse as urllib_parse
from datetime import datetime from datetime import datetime
from gettext import gettext as _ from gettext import gettext as _
from pathlib import Path from pathlib import Path
from typing import Any, Final, cast from typing import Any, Final, Sequence, cast
import certifi, colorama, nodriver # isort: skip import certifi, colorama, nodriver # isort: skip
from nodriver.core.connection import ProtocolException from nodriver.core.connection import ProtocolException
@@ -34,10 +34,19 @@ LOG.setLevel(loggers.INFO)
PUBLISH_MAX_RETRIES:Final[int] = 3 PUBLISH_MAX_RETRIES:Final[int] = 3
_NUMERIC_IDS_RE:Final[re.Pattern[str]] = re.compile(r"^\d+(,\d+)*$") _NUMERIC_IDS_RE:Final[re.Pattern[str]] = re.compile(r"^\d+(,\d+)*$")
_LOGIN_DETECTION_SELECTORS:Final[list[tuple["By", str]]] = [
(By.CLASS_NAME, "mr-medium"),
(By.ID, "user-email"),
]
_LOGIN_DETECTION_SELECTOR_LABELS:Final[tuple[str, ...]] = ("user_info_primary", "user_info_secondary")
colorama.just_fix_windows_console() colorama.just_fix_windows_console()
def _format_login_detection_selectors(selectors:Sequence[tuple["By", str]]) -> str:
return ", ".join(f"{selector_type.name}={selector_value}" for selector_type, selector_value in selectors)
class AdUpdateStrategy(enum.Enum): class AdUpdateStrategy(enum.Enum):
REPLACE = enum.auto() REPLACE = enum.auto()
MODIFY = enum.auto() MODIFY = enum.auto()
@@ -1245,38 +1254,39 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
login_check_timeout, login_check_timeout,
effective_timeout, effective_timeout,
) )
tried_login_selectors = _format_login_detection_selectors(_LOGIN_DETECTION_SELECTORS)
login_selectors = [
(By.CLASS_NAME, "mr-medium"),
(By.ID, "user-email"),
]
primary_selector_index = 0
try: try:
user_info, matched_selector = await self.web_text_first_available( user_info, matched_selector = await self.web_text_first_available(
login_selectors, _LOGIN_DETECTION_SELECTORS,
timeout = login_check_timeout, timeout = login_check_timeout,
key = "login_detection", key = "login_detection",
description = "login_detection(selector_group)", description = "login_detection(selector_group)",
) )
if username in user_info.lower(): if username in user_info.lower():
if matched_selector == primary_selector_index: matched_selector_label = (
LOG.debug("Login detected via .mr-medium element") _LOGIN_DETECTION_SELECTOR_LABELS[matched_selector]
else: if 0 <= matched_selector < len(_LOGIN_DETECTION_SELECTOR_LABELS)
LOG.debug("Login detected via #user-email element") else f"selector_index_{matched_selector}"
)
LOG.debug("Login detected via login detection selector '%s'", matched_selector_label)
return True return True
except TimeoutError: except TimeoutError:
LOG.debug("Timeout waiting for login detection selector group after %.1fs", effective_timeout) LOG.debug("Timeout waiting for login detection selector group after %.1fs", effective_timeout)
if not include_probe: if not include_probe:
LOG.debug("No login detected - neither .mr-medium nor #user-email found with username") LOG.debug("No login detected via configured login detection selectors (%s)", tried_login_selectors)
return False return False
state = await self._auth_probe_login_state() state = await self._auth_probe_login_state()
if state == LoginState.LOGGED_IN: if state == LoginState.LOGGED_IN:
return True return True
LOG.debug("No login detected - DOM elements not found and server probe returned %s", state.name) LOG.debug(
"No login detected - DOM login detection selectors (%s) did not confirm login and server probe returned %s",
tried_login_selectors,
state.name,
)
return False return False
async def _fetch_published_ads(self) -> list[dict[str, Any]]: async def _fetch_published_ads(self) -> list[dict[str, Any]]:

View File

@@ -97,11 +97,8 @@ kleinanzeigen_bot/__init__.py:
is_logged_in: is_logged_in:
"Starting login detection (timeout: %.1fs base, %.1fs effective with multiplier/backoff)": "Starte Login-Erkennung (Timeout: %.1fs Basis, %.1fs effektiv mit Multiplikator/Backoff)" "Starting login detection (timeout: %.1fs base, %.1fs effective with multiplier/backoff)": "Starte Login-Erkennung (Timeout: %.1fs Basis, %.1fs effektiv mit Multiplikator/Backoff)"
"Login detected via .mr-medium element": "Login erkannt über .mr-medium Element" "Login detected via login detection selector '%s'": "Login erkannt über Login-Erkennungs-Selektor '%s'"
"Login detected via #user-email element": "Login erkannt über #user-email Element"
"Timeout waiting for login detection selector group after %.1fs": "Timeout beim Warten auf die Login-Erkennungs-Selektorgruppe nach %.1fs" "Timeout waiting for login detection selector group after %.1fs": "Timeout beim Warten auf die Login-Erkennungs-Selektorgruppe nach %.1fs"
"No login detected - neither .mr-medium nor #user-email found with username": "Kein Login erkannt - weder .mr-medium noch #user-email mit Benutzername gefunden"
"No login detected - DOM elements not found and server probe returned %s": "Kein Login erkannt - DOM-Elemente nicht gefunden und Server-Probe ergab %s"
handle_after_login_logic: handle_after_login_logic:
"# Device verification message detected. Please follow the instruction displayed in the Browser.": "# Nachricht zur Geräteverifizierung erkannt. Bitte den Anweisungen im Browser folgen." "# Device verification message detected. Please follow the instruction displayed in the Browser.": "# Nachricht zur Geräteverifizierung erkannt. Bitte den Anweisungen im Browser folgen."

View File

@@ -483,6 +483,63 @@ class TestKleinanzeigenBotAuthentication:
assert call_args.kwargs["key"] == "login_detection" assert call_args.kwargs["key"] == "login_detection"
assert call_args.kwargs["timeout"] == test_bot._timeout("login_detection") assert call_args.kwargs["timeout"] == test_bot._timeout("login_detection")
@pytest.mark.asyncio
async def test_is_logged_in_logs_selector_label_without_raw_selector_literals(
self, test_bot:KleinanzeigenBot, caplog:pytest.LogCaptureFixture
) -> None:
"""Login detection logs should reference stable labels, not raw selector values."""
caplog.set_level("DEBUG")
with (
caplog.at_level("DEBUG"),
patch.object(test_bot, "web_text_first_available", new_callable = AsyncMock, return_value = ("angemeldet als: dummy_user", 1)),
):
assert await test_bot.is_logged_in(include_probe = False) is True
assert "Login detected via login detection selector 'user_info_secondary'" in caplog.text
for forbidden in (".mr-medium", "#user-email", "mr-medium", "user-email"):
assert forbidden not in caplog.text
@pytest.mark.asyncio
async def test_is_logged_in_logs_generic_message_when_selector_group_does_not_match(
self, test_bot:KleinanzeigenBot, caplog:pytest.LogCaptureFixture
) -> None:
"""Missing selector-group match should log the tried selectors when probe is disabled."""
caplog.set_level("DEBUG")
with (
caplog.at_level("DEBUG"),
patch.object(test_bot, "web_text_first_available", side_effect = TimeoutError),
):
assert await test_bot.is_logged_in(include_probe = False) is False
assert any(
record.message == "No login detected via configured login detection selectors (CLASS_NAME=mr-medium, ID=user-email)"
for record in caplog.records
)
@pytest.mark.asyncio
async def test_is_logged_in_logs_raw_selectors_when_probe_reports_logged_out(
self, test_bot:KleinanzeigenBot, caplog:pytest.LogCaptureFixture
) -> None:
"""Probe-based final failure should include the tried raw selectors for debugging."""
caplog.set_level("DEBUG")
with (
caplog.at_level("DEBUG"),
patch.object(test_bot, "web_text_first_available", side_effect = TimeoutError),
patch.object(test_bot, "_auth_probe_login_state", new_callable = AsyncMock, return_value = LoginState.LOGGED_OUT),
):
assert await test_bot.is_logged_in() is False
assert any(
record.message == (
"No login detected - DOM login detection selectors (CLASS_NAME=mr-medium, ID=user-email) "
"did not confirm login and server probe returned LOGGED_OUT"
)
for record in caplog.records
)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_get_login_state_prefers_dom_over_auth_probe(self, test_bot:KleinanzeigenBot) -> None: async def test_get_login_state_prefers_dom_over_auth_probe(self, test_bot:KleinanzeigenBot) -> None:
with ( with (