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: enable ruff preview rules

Browse Source
This commit is contained in:
sebthom
2025-04-28 00:33:58 +02:00
parent f98251ade3
commit bda0acf943
9 changed files with 70 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/kleinanzeigen_bot/__init__.py
View File

@@ -348,7 +348,7 @@ class KleinanzeigenBot(WebScrapingMixin):
for ad_file, ad_file_relative in sorted(ad_files.items()):
ad_cfg_orig = dicts.load_dict(ad_file, "ad")
ad_cfg = copy.deepcopy(ad_cfg_orig)
dicts.apply_defaults(ad_cfg, self.config["ad_defaults"], ignore = lambda k, _: k == "description", override = lambda _, v: v == "")
dicts.apply_defaults(ad_cfg, self.config["ad_defaults"], ignore = lambda k, _: k == "description", override = lambda _, v: not v)
dicts.apply_defaults(ad_cfg, ad_fields)
if ignore_inactive and not ad_cfg["active"]:

6
src/kleinanzeigen_bot/extract.py
View File

@@ -42,7 +42,7 @@ class AdExtractor(WebScrapingMixin):
os.mkdir(relative_directory)
LOG.info("Created ads directory at ./%s.", relative_directory)
new_base_dir = os.path.join(relative_directory, f'ad_{ad_id}')
new_base_dir = os.path.join(relative_directory, f"ad_{ad_id}")
if os.path.exists(new_base_dir):
LOG.info("Deleting current folder of ad %s...", ad_id)
shutil.rmtree(new_base_dir)
@@ -51,7 +51,7 @@ class AdExtractor(WebScrapingMixin):
# call extraction function
info = await self._extract_ad_page_info(new_base_dir, ad_id)
ad_file_path = new_base_dir + "/" + f'ad_{ad_id}.yaml'
ad_file_path = new_base_dir + "/" + f"ad_{ad_id}.yaml"
dicts.save_dict(ad_file_path, info)
async def _download_images_from_ad_page(self, directory:str, ad_id:int) -> list[str]:
@@ -96,7 +96,7 @@ class AdExtractor(WebScrapingMixin):
try:
# click next button, wait, and re-establish reference
await (await self.web_find(By.CLASS_NAME, "galleryimage--navigation--next")).click()
new_div = await self.web_find(By.CSS_SELECTOR, f'div.galleryimage-element:nth-child({img_nr + 1})')
new_div = await self.web_find(By.CSS_SELECTOR, f"div.galleryimage-element:nth-child({img_nr + 1})")
img_element = await self.web_find(By.TAG_NAME, "img", parent = new_div)
except TimeoutError:
LOG.error("NEXT button in image gallery somehow missing, aborting image fetching.")

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

@@ -61,7 +61,7 @@ def load_dict_if_exists(filepath:str, content_label:str = "") -> dict[str, Any]
LOG.info("Loading %s[%s]...", content_label and content_label + _(" from ") or "", abs_filepath)
__, file_ext = os.path.splitext(filepath)
if file_ext not in (".json", ".yaml", ".yml"):
if file_ext not in {".json", ".yaml", ".yml"}:
raise ValueError(_('Unsupported file type. The filename "%s" must end with *.json, *.yaml, or *.yml') % filepath)
if not os.path.exists(filepath):
@@ -78,7 +78,7 @@ def load_dict_from_module(module:ModuleType, filename:str, content_label:str = "
LOG.debug("Loading %s[%s.%s]...", content_label and content_label + " from " or "", module.__name__, filename)
__, file_ext = os.path.splitext(filename)
if file_ext not in (".json", ".yaml", ".yml"):
if file_ext not in {".json", ".yaml", ".yml"}:
raise ValueError(f'Unsupported file type. The filename "{filename}" must end with *.json, *.yaml, or *.yml')
content = get_resource_as_string(module, filename) # pylint: disable=deprecated-method

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

@@ -13,7 +13,12 @@ from . import i18n
T = TypeVar("T")
def ensure(condition:Any | bool | Callable[[], bool], error_message:str, timeout:float = 5, poll_requency:float = 0.5) -> None:
def ensure(
condition:Any | bool | Callable[[], bool], # noqa: FBT001 Boolean-typed positional argument in function definition
error_message:str,
timeout:float = 5,
poll_requency:float = 0.5
) -> None:
"""
:param timeout: timespan in seconds until when the condition must become `True`, default is 5 seconds
:param poll_requency: sleep interval between calls in seconds, default is 0.5 seconds
@@ -50,7 +55,7 @@ def is_frozen() -> bool:
async def ainput(prompt:str) -> str:
return await asyncio.to_thread(input, f'{prompt} ')
return await asyncio.to_thread(input, f"{prompt} ")
def parse_decimal(number:float | int | str) -> decimal.Decimal:

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

@@ -33,7 +33,7 @@ __all__ = [
LOG:Final[loggers.Logger] = loggers.get_logger(__name__)
# see https://api.jquery.com/category/selectors/
METACHAR_ESCAPER:Final[dict[int, str]] = str.maketrans({ch: f'\\{ch}' for ch in '!"#$%&\'()*+,./:;<=>?@[\\]^`{|}~'})
METACHAR_ESCAPER:Final[dict[int, str]] = str.maketrans({ch: f"\\{ch}" for ch in '!"#$%&\'()*+,./:;<=>?@[\\]^`{|}~'})
class By(enum.Enum):
@@ -535,13 +535,13 @@ class WebScrapingMixin:
bottom_y_pos:int = await self.web_execute("document.body.scrollHeight") # get bottom position
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
await self.web_execute(f"window.scrollTo(0, {current_y_pos})") # scroll one step
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})')
await self.web_execute(f"window.scrollTo(0, {current_y_pos})")
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: