Mirrors/kleinanzeigen-bot
1
0
Fork 0
mirror of https://github.com/Second-Hand-Friends/kleinanzeigen-bot.git synced 2026-03-12 10:31:50 +01:00
Code Issues Packages Projects Releases Wiki Activity

refact: use ruff instead of autopep8,bandit,pylint for linting

Browse Source
This commit is contained in:
sebthom
2025-04-28 12:51:51 +02:00
parent f0b84ab335
commit 376ec76226
27 changed files with 437 additions and 605 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/kleinanzeigen_bot/utils/dicts.py
View File

@@ -1,16 +1,15 @@
"""
SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
SPDX-License-Identifier: AGPL-3.0-or-later
SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
"""
import copy, json, os
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import copy, json, os # isort: skip
from collections.abc import Callable
from importlib.resources import read_text as get_resource_as_string
from gettext import gettext as _
from importlib.resources import read_text as get_resource_as_string
from types import ModuleType
from typing import Any, Final
from ruamel.yaml import YAML
from . import files, loggers # pylint: disable=cyclic-import
LOG:Final[loggers.Logger] = loggers.get_logger(__name__)
@@ -112,9 +111,9 @@ def safe_get(a_map:dict[Any, Any], *keys:str) -> Any:
'some_value'
"""
if a_map:
for key in keys:
try:
try:
for key in keys:
a_map = a_map[key]
except (KeyError, TypeError):
return None
except (KeyError, TypeError):
return None
return a_map

12
src/kleinanzeigen_bot/utils/error_handlers.py
View File

@@ -1,11 +1,9 @@
"""
SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
SPDX-License-Identifier: AGPL-3.0-or-later
SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
"""
import sys, traceback
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import sys, traceback # isort: skip
from types import FrameType, TracebackType
from typing import Final
from typing import Any, Final
from . import loggers

10
src/kleinanzeigen_bot/utils/exceptions.py
View File

@@ -1,8 +1,6 @@
"""
SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
SPDX-License-Identifier: AGPL-3.0-or-later
SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
"""
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
from datetime import timedelta
@@ -13,6 +11,6 @@ class KleinanzeigenBotError(RuntimeError):
class CaptchaEncountered(KleinanzeigenBotError):
"""Raised when a Captcha was detected and auto-restart is enabled."""
def __init__(self, restart_delay: timedelta):
def __init__(self, restart_delay: timedelta) -> None:
super().__init__()
self.restart_delay = restart_delay

8
src/kleinanzeigen_bot/utils/files.py
View File

@@ -1,8 +1,6 @@
"""
SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
SPDX-License-Identifier: AGPL-3.0-or-later
SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
"""
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import os

27
src/kleinanzeigen_bot/utils/i18n.py
View File

@@ -1,13 +1,12 @@
"""
SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
SPDX-License-Identifier: AGPL-3.0-or-later
SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
"""
import ctypes, gettext, inspect, locale, logging, os, sys
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import ctypes, gettext, inspect, locale, logging, os, sys # isort: skip
from collections.abc import Sized
from typing import Any, Final, NamedTuple
from kleinanzeigen_bot import resources
from . import dicts, reflect
__all__ = [
@@ -96,7 +95,7 @@ def translate(text:object, caller: inspect.FrameInfo | None) -> str:
if not caller:
return text
global _TRANSLATIONS
global _TRANSLATIONS # noqa: PLW0603 Using the global statement to update `...` is discouraged
if _TRANSLATIONS is None:
try:
_TRANSLATIONS = dicts.load_dict_from_module(resources, f"translations.{_CURRENT_LOCALE[0]}.yaml")
@@ -125,10 +124,10 @@ gettext.gettext = lambda message: translate(_original_gettext(message), reflect.
for module_name, module in sys.modules.items():
if module is None or module_name in sys.builtin_module_names:
continue
if hasattr(module, '_') and getattr(module, '_') is _original_gettext:
setattr(module, '_', gettext.gettext)
if hasattr(module, 'gettext') and getattr(module, 'gettext') is _original_gettext:
setattr(module, 'gettext', gettext.gettext)
if hasattr(module, '_') and module._ is _original_gettext:
module._ = gettext.gettext # type: ignore[attr-defined]
if hasattr(module, 'gettext') and module.gettext is _original_gettext:
module.gettext = gettext.gettext # type: ignore[attr-defined]
def get_current_locale() -> Locale:
@@ -136,13 +135,13 @@ def get_current_locale() -> Locale:
def set_current_locale(new_locale:Locale) -> None:
global _CURRENT_LOCALE, _TRANSLATIONS
global _CURRENT_LOCALE, _TRANSLATIONS # noqa: PLW0603 Using the global statement to update `...` is discouraged
if new_locale.language != _CURRENT_LOCALE.language:
_TRANSLATIONS = None
_CURRENT_LOCALE = new_locale
def pluralize(noun:str, count:int | Sized, prefix_with_count:bool = True) -> str:
def pluralize(noun:str, count:int | Sized, *, prefix_with_count:bool = True) -> str:
"""
>>> set_current_locale(Locale("en")) # Setup for doctests
>>> pluralize("field", 1)
@@ -189,7 +188,7 @@ def pluralize(noun:str, count:int | Sized, prefix_with_count:bool = True) -> str
return f"{prefix}{noun}e" # Hund -> Hunde
# English
if len(noun) < 2:
if len(noun) < 2: # noqa: PLR2004 Magic value used in comparison
return f"{prefix}{noun}s"
if noun.endswith(('s', 'sh', 'ch', 'x', 'z')):
return f"{prefix}{noun}es"

29
src/kleinanzeigen_bot/utils/loggers.py
View File

@@ -1,15 +1,14 @@
"""
SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
SPDX-License-Identifier: AGPL-3.0-or-later
SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
"""
import copy, logging, re, sys
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import copy, logging, re, sys # isort: skip
from gettext import gettext as _
from logging import Logger, DEBUG, INFO, WARNING, ERROR, CRITICAL
from logging import CRITICAL, DEBUG, ERROR, INFO, WARNING, Logger
from logging.handlers import RotatingFileHandler
from typing import Any, Final # @UnusedImport
import colorama
from . import i18n, reflect
__all__ = [
@@ -27,6 +26,16 @@ __all__ = [
LOG_ROOT:Final[logging.Logger] = logging.getLogger()
class _MaxLevelFilter(logging.Filter):
def __init__(self, level: int) -> None:
super().__init__()
self.level = level
def filter(self, record: logging.LogRecord) -> bool:
return record.levelno <= self.level
def configure_console_logging() -> None:
# if a StreamHandler already exists, do not append it again
if any(isinstance(h, logging.StreamHandler) for h in LOG_ROOT.handlers):
@@ -82,9 +91,7 @@ def configure_console_logging() -> None:
stdout_log = logging.StreamHandler(sys.stderr)
stdout_log.setLevel(DEBUG)
stdout_log.addFilter(type("", (logging.Filter,), {
"filter": lambda rec: rec.levelno <= INFO
}))
stdout_log.addFilter(_MaxLevelFilter(INFO))
stdout_log.setFormatter(formatter)
LOG_ROOT.addHandler(stdout_log)
@@ -97,7 +104,7 @@ def configure_console_logging() -> None:
class LogFileHandle:
"""Encapsulates a log file handler with close and status methods."""
def __init__(self, file_path: str, handler: RotatingFileHandler, logger: logging.Logger):
def __init__(self, file_path: str, handler: RotatingFileHandler, logger: logging.Logger) -> None:
self.file_path = file_path
self._handler:RotatingFileHandler | None = handler
self._logger = logger

50
src/kleinanzeigen_bot/utils/misc.py
View File

@@ -1,11 +1,9 @@
"""
SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
SPDX-License-Identifier: AGPL-3.0-or-later
SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
"""
import asyncio, decimal, re, sys, time
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import asyncio, decimal, re, sys, time # isort: skip
from collections.abc import Callable
from datetime import datetime, timedelta
from datetime import datetime, timedelta, timezone
from gettext import gettext as _
from typing import Any, TypeVar
@@ -39,6 +37,10 @@ def ensure(condition:Any | bool | Callable[[], bool], error_message:str, timeout
time.sleep(poll_requency)
def now() -> datetime:
return datetime.now(timezone.utc)
def is_frozen() -> bool:
"""
>>> is_frozen()
@@ -81,12 +83,27 @@ def parse_decimal(number:float | int | str) -> decimal.Decimal:
raise decimal.DecimalException(f"Invalid number format: {number}") from ex
def parse_datetime(date:datetime | str | None) -> datetime | None:
def parse_datetime(
date: datetime | str | None,
*,
add_timezone_if_missing: bool = True,
use_local_timezone: bool = True
) -> datetime | None:
"""
>>> parse_datetime(datetime(2020, 1, 1, 0, 0))
Parses a datetime object or ISO-formatted string.
Args:
date: The input datetime object or ISO string.
add_timezone_if_missing: If True, add timezone info if missing.
use_local_timezone: If True, use local timezone; otherwise UTC if adding timezone.
Returns:
A timezone-aware or naive datetime object, depending on parameters.
>>> parse_datetime(datetime(2020, 1, 1, 0, 0), add_timezone_if_missing = False)
datetime.datetime(2020, 1, 1, 0, 0)
>>> parse_datetime("2020-01-01T00:00:00")
>>> parse_datetime("2020-01-01T00:00:00", add_timezone_if_missing = False)
datetime.datetime(2020, 1, 1, 0, 0)
>>> parse_datetime(None)
@@ -94,9 +111,16 @@ def parse_datetime(date:datetime | str | None) -> datetime | None:
"""
if date is None:
return None
if isinstance(date, datetime):
return date
return datetime.fromisoformat(date)
dt = date if isinstance(date, datetime) else datetime.fromisoformat(date)
if dt.tzinfo is None and add_timezone_if_missing:
dt = (
dt.astimezone() if use_local_timezone
else dt.replace(tzinfo = timezone.utc)
)
return dt
def parse_duration(text:str) -> timedelta:

8
src/kleinanzeigen_bot/utils/net.py
View File

@@ -1,8 +1,6 @@
"""
SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
SPDX-License-Identifier: AGPL-3.0-or-later
SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
"""
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import socket

8
src/kleinanzeigen_bot/utils/reflect.py
View File

@@ -1,8 +1,6 @@
"""
SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
SPDX-License-Identifier: AGPL-3.0-or-later
SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
"""
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import inspect
from typing import Any

65
src/kleinanzeigen_bot/utils/web_scraping_mixin.py
View File

@@ -1,26 +1,24 @@
"""
SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
SPDX-License-Identifier: AGPL-3.0-or-later
SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
"""
import asyncio, enum, inspect, json, os, platform, secrets, shutil, time
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import asyncio, enum, inspect, json, os, platform, secrets, shutil # isort: skip
from collections.abc import Callable, Coroutine, Iterable
from gettext import gettext as _
from typing import cast, Any, Final
from typing import Any, Final, cast
try:
from typing import Never # type: ignore[attr-defined,unused-ignore] # mypy
except ImportError:
from typing import NoReturn as Never # Python <3.11
import nodriver, psutil
import nodriver, psutil # isort: skip
from nodriver.core.browser import Browser
from nodriver.core.config import Config
from nodriver.core.element import Element
from nodriver.core.tab import Tab as Page
from . import loggers, net
from .misc import ensure, T
from .misc import T, ensure
__all__ = [
"Browser",
@@ -70,8 +68,8 @@ class WebScrapingMixin:
def __init__(self) -> None:
self.browser_config:Final[BrowserConfig] = BrowserConfig()
self.browser:Browser = None # pyright: ignore
self.page:Page = None # pyright: ignore
self.browser:Browser = None # pyright: ignore[reportAttributeAccessIssue]
self.page:Page = None # pyright: ignore[reportAttributeAccessIssue]
async def create_browser_session(self) -> None:
LOG.info("Creating Browser session...")
@@ -96,7 +94,7 @@ class WebScrapingMixin:
if remote_port > 0:
LOG.info("Using existing browser process at %s:%s", remote_host, remote_port)
ensure(net.is_port_open(remote_host, remote_port),
f"Browser process not reachable at {remote_host}:{remote_port}. " +
f"Browser process not reachable at {remote_host}:{remote_port}. "
f"Start the browser with --remote-debugging-port={remote_port} or remove this port from your config.yaml")
cfg = Config(
browser_executable_path = self.browser_config.binary_location # actually not necessary but nodriver fails without
@@ -208,14 +206,14 @@ class WebScrapingMixin:
def close_browser_session(self) -> None:
if self.browser:
LOG.debug("Closing Browser session...")
self.page = None # pyright: ignore
browser_process = psutil.Process(self.browser._process_pid) # pylint: disable=protected-access
self.page = None # pyright: ignore[reportAttributeAccessIssue]
browser_process = psutil.Process(self.browser._process_pid) # noqa: SLF001 Private member accessed
browser_children:list[psutil.Process] = browser_process.children()
self.browser.stop()
for p in browser_children:
if p.is_running():
p.kill() # terminate orphaned browser processes
self.browser = None # pyright: ignore
self.browser = None # pyright: ignore[reportAttributeAccessIssue]
def get_compatible_browser(self) -> str:
match platform.system():
@@ -236,15 +234,15 @@ class WebScrapingMixin:
case "Windows":
browser_paths = [
os.environ.get("ProgramFiles", "C:\\Program Files") + r'\Microsoft\Edge\Application\msedge.exe',
os.environ.get("ProgramFiles(x86)", "C:\\Program Files (x86)") + r'\Microsoft\Edge\Application\msedge.exe',
os.environ.get("PROGRAMFILES", "C:\\Program Files") + r'\Microsoft\Edge\Application\msedge.exe',
os.environ.get("PROGRAMFILES(X86)", "C:\\Program Files (x86)") + r'\Microsoft\Edge\Application\msedge.exe',
os.environ["ProgramFiles"] + r'\Chromium\Application\chrome.exe',
os.environ["ProgramFiles(x86)"] + r'\Chromium\Application\chrome.exe',
os.environ["PROGRAMFILES"] + r'\Chromium\Application\chrome.exe',
os.environ["PROGRAMFILES(X86)"] + r'\Chromium\Application\chrome.exe',
os.environ["LOCALAPPDATA"] + r'\Chromium\Application\chrome.exe',
os.environ["ProgramFiles"] + r'\Chrome\Application\chrome.exe',
os.environ["ProgramFiles(x86)"] + r'\Chrome\Application\chrome.exe',
os.environ["PROGRAMFILES"] + r'\Chrome\Application\chrome.exe',
os.environ["PROGRAMFILES(X86)"] + r'\Chrome\Application\chrome.exe',
os.environ["LOCALAPPDATA"] + r'\Chrome\Application\chrome.exe',
shutil.which("msedge.exe"),
@@ -277,7 +275,7 @@ class WebScrapingMixin:
ex:Exception | None = None
try:
result_raw = condition()
result:T = (await result_raw) if inspect.isawaitable(result_raw) else result_raw
result:T = cast(T, await result_raw if inspect.isawaitable(result_raw) else result_raw)
if result:
return result
except Exception as ex1:
@@ -359,11 +357,11 @@ class WebScrapingMixin:
_prev_jscode:str = getattr(self.__class__.web_execute, "_prev_jscode", "")
if not (jscode == _prev_jscode or (jscode.startswith("window.scrollTo") and _prev_jscode.startswith("window.scrollTo"))):
LOG.debug("web_execute(`%s`) = `%s`", jscode, result)
self.__class__.web_execute._prev_jscode = jscode # type: ignore[attr-defined] # pylint: disable=protected-access
self.__class__.web_execute._prev_jscode = jscode # type: ignore[attr-defined] # noqa: SLF001 Private member accessed
return result
async def web_find(self, selector_type:By, selector_value:str, *, parent:Element = None, timeout:int | float = 5) -> Element:
async def web_find(self, selector_type:By, selector_value:str, *, parent:Element | None = None, timeout:int | float = 5) -> Element:
"""
Locates an HTML element by the given selector type and value.
@@ -408,7 +406,7 @@ class WebScrapingMixin:
raise AssertionError(_("Unsupported selector type: %s") % selector_type)
async def web_find_all(self, selector_type:By, selector_value:str, *, parent:Element = None, timeout:int | float = 5) -> list[Element]:
async def web_find_all(self, selector_type:By, selector_value:str, *, parent:Element | None = None, timeout:int | float = 5) -> list[Element]:
"""
Locates an HTML element by ID.
@@ -460,7 +458,7 @@ class WebScrapingMixin:
await self.web_sleep()
return input_field
async def web_open(self, url:str, *, timeout:int | float = 15000, reload_if_already_open:bool = False) -> None:
async def web_open(self, url:str, *, timeout:int | float = 15_000, reload_if_already_open:bool = False) -> None:
"""
:param url: url to open in browser
:param timeout: timespan in seconds within the page needs to be loaded
@@ -475,7 +473,7 @@ class WebScrapingMixin:
await self.web_await(lambda: self.web_execute("document.readyState == 'complete'"), timeout = timeout,
timeout_error_message = f"Page did not finish loading within {timeout} seconds.")
async def web_text(self, selector_type:By, selector_value:str, *, parent:Element = None, timeout:int | float = 5) -> str:
async def web_text(self, selector_type:By, selector_value:str, *, parent:Element | None = None, timeout:int | float = 5) -> str:
return str(await (await self.web_find(selector_type, selector_value, parent = parent, timeout = timeout)).apply("""
function (elem) {
let sel = window.getSelection()
@@ -489,10 +487,11 @@ class WebScrapingMixin:
}
"""))
async def web_sleep(self, min_ms:int = 1000, max_ms:int = 2500) -> None:
async def web_sleep(self, min_ms:int = 1_000, max_ms:int = 2_500) -> None:
duration = max_ms <= min_ms and min_ms or secrets.randbelow(max_ms - min_ms) + min_ms
LOG.log(loggers.INFO if duration > 1500 else loggers.DEBUG, " ... pausing for %d ms ...", duration)
await self.page.sleep(duration / 1000)
LOG.log(loggers.INFO if duration > 1_500 else loggers.DEBUG, # noqa: PLR2004 Magic value used in comparison
" ... pausing for %d ms ...", duration)
await self.page.sleep(duration / 1_000)
async def web_request(self, url:str, method:str = "GET", valid_response_codes:int | Iterable[int] = 200,
headers:dict[str, str] | None = None) -> dict[str, Any]:
@@ -524,7 +523,7 @@ class WebScrapingMixin:
return response
# pylint: enable=dangerous-default-value
async def web_scroll_page_down(self, scroll_length: int = 10, scroll_speed: int = 10000, scroll_back_top: bool = False) -> None:
async def web_scroll_page_down(self, scroll_length: int = 10, scroll_speed: int = 10_000, *, scroll_back_top: bool = False) -> None:
"""
Smoothly scrolls the current web page down.
@@ -537,13 +536,13 @@ class WebScrapingMixin:
while current_y_pos < bottom_y_pos: # scroll in steps until bottom reached
current_y_pos += scroll_length
await self.web_execute(f'window.scrollTo(0, {current_y_pos})') # scroll one step
time.sleep(scroll_length / scroll_speed)
await asyncio.sleep(scroll_length / scroll_speed)
if scroll_back_top: # scroll back to top in same style
while current_y_pos > 0:
current_y_pos -= scroll_length
await self.web_execute(f'window.scrollTo(0, {current_y_pos})')
time.sleep(scroll_length / scroll_speed / 2) # double speed
await asyncio.sleep(scroll_length / scroll_speed / 2) # double speed
async def web_select(self, selector_type:By, selector_value:str, selected_value:Any, timeout:int | float = 5) -> Element:
"""