mirror of
https://github.com/Second-Hand-Friends/kleinanzeigen-bot.git
synced 2026-03-12 02:31:45 +01:00
Replace deprecated type hints and add more pylint rules
This commit is contained in:
@@ -3,10 +3,11 @@ Copyright (C) 2022 Sebastian Thomschke and contributors
|
|||||||
SPDX-License-Identifier: AGPL-3.0-or-later
|
SPDX-License-Identifier: AGPL-3.0-or-later
|
||||||
"""
|
"""
|
||||||
import atexit, copy, getopt, glob, json, logging, os, signal, sys, textwrap, time, urllib
|
import atexit, copy, getopt, glob, json, logging, os, signal, sys, textwrap, time, urllib
|
||||||
|
from collections.abc import Iterable
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
import importlib.metadata
|
import importlib.metadata
|
||||||
from logging.handlers import RotatingFileHandler
|
from logging.handlers import RotatingFileHandler
|
||||||
from typing import Any, Dict, Final, Iterable
|
from typing import Any, Final
|
||||||
|
|
||||||
from ruamel.yaml import YAML
|
from ruamel.yaml import YAML
|
||||||
from selenium.common.exceptions import NoSuchElementException
|
from selenium.common.exceptions import NoSuchElementException
|
||||||
@@ -29,10 +30,10 @@ class KleinanzeigenBot(SeleniumMixin):
|
|||||||
|
|
||||||
self.root_url = "https://www.ebay-kleinanzeigen.de"
|
self.root_url = "https://www.ebay-kleinanzeigen.de"
|
||||||
|
|
||||||
self.config:Dict[str, Any] = {}
|
self.config:dict[str, Any] = {}
|
||||||
self.config_file_path = os.path.join(os.getcwd(), "config.yaml")
|
self.config_file_path = os.path.join(os.getcwd(), "config.yaml")
|
||||||
|
|
||||||
self.categories:Dict[str, str] = {}
|
self.categories:dict[str, str] = {}
|
||||||
|
|
||||||
self.file_log:logging.FileHandler = None
|
self.file_log:logging.FileHandler = None
|
||||||
if is_frozen():
|
if is_frozen():
|
||||||
@@ -67,15 +68,15 @@ class KleinanzeigenBot(SeleniumMixin):
|
|||||||
case "publish":
|
case "publish":
|
||||||
self.configure_file_logging()
|
self.configure_file_logging()
|
||||||
self.load_config()
|
self.load_config()
|
||||||
ads = self.load_ads()
|
if ads := self.load_ads():
|
||||||
if len(ads) == 0:
|
|
||||||
LOG.info("############################################")
|
|
||||||
LOG.info("No ads to (re-)publish found.")
|
|
||||||
LOG.info("############################################")
|
|
||||||
else:
|
|
||||||
self.create_webdriver_session()
|
self.create_webdriver_session()
|
||||||
self.login()
|
self.login()
|
||||||
self.publish_ads(ads)
|
self.publish_ads(ads)
|
||||||
|
else:
|
||||||
|
LOG.info("############################################")
|
||||||
|
LOG.info("No ads to (re-)publish found.")
|
||||||
|
LOG.info("############################################")
|
||||||
|
|
||||||
case _:
|
case _:
|
||||||
LOG.error("Unknown command: %s", self.command)
|
LOG.error("Unknown command: %s", self.command)
|
||||||
sys.exit(2)
|
sys.exit(2)
|
||||||
@@ -145,7 +146,7 @@ class KleinanzeigenBot(SeleniumMixin):
|
|||||||
|
|
||||||
LOG.info("App version: %s", self.get_version())
|
LOG.info("App version: %s", self.get_version())
|
||||||
|
|
||||||
def load_ads(self, exclude_inactive = True, exclude_undue = True) -> Iterable[Dict[str, Any]]:
|
def load_ads(self, exclude_inactive = True, exclude_undue = True) -> Iterable[dict[str, Any]]:
|
||||||
LOG.info("Searching for ad files...")
|
LOG.info("Searching for ad files...")
|
||||||
|
|
||||||
ad_files = set()
|
ad_files = set()
|
||||||
@@ -222,7 +223,7 @@ class KleinanzeigenBot(SeleniumMixin):
|
|||||||
for image_pattern in ad_cfg["images"]:
|
for image_pattern in ad_cfg["images"]:
|
||||||
for image_file in glob.glob(image_pattern, root_dir = os.path.dirname(ad_file), recursive = True):
|
for image_file in glob.glob(image_pattern, root_dir = os.path.dirname(ad_file), recursive = True):
|
||||||
_, image_file_ext = os.path.splitext(image_file)
|
_, image_file_ext = os.path.splitext(image_file)
|
||||||
ensure(image_file_ext.lower() in (".gif", ".jpg", ".jpeg", ".png"), f"Unsupported image file type [{image_file}]")
|
ensure(image_file_ext.lower() in {".gif", ".jpg", ".jpeg", ".png"}, f"Unsupported image file type [{image_file}]")
|
||||||
if os.path.isabs(image_file):
|
if os.path.isabs(image_file):
|
||||||
images.add(image_file)
|
images.add(image_file)
|
||||||
else:
|
else:
|
||||||
@@ -290,7 +291,7 @@ class KleinanzeigenBot(SeleniumMixin):
|
|||||||
self.web_await(lambda _: self.webdriver.find_element(By.ID, "recaptcha-anchor").get_attribute("aria-checked") == "true", timeout = 5 * 60)
|
self.web_await(lambda _: self.webdriver.find_element(By.ID, "recaptcha-anchor").get_attribute("aria-checked") == "true", timeout = 5 * 60)
|
||||||
self.webdriver.switch_to.default_content()
|
self.webdriver.switch_to.default_content()
|
||||||
|
|
||||||
def delete_ad(self, ad_cfg: Dict[str, Any]) -> bool:
|
def delete_ad(self, ad_cfg: dict[str, Any]) -> bool:
|
||||||
LOG.info("Deleting ad '%s' if already present...", ad_cfg["title"])
|
LOG.info("Deleting ad '%s' if already present...", ad_cfg["title"])
|
||||||
|
|
||||||
self.web_open(f"{self.root_url}/m-meine-anzeigen.html")
|
self.web_open(f"{self.root_url}/m-meine-anzeigen.html")
|
||||||
@@ -314,7 +315,7 @@ class KleinanzeigenBot(SeleniumMixin):
|
|||||||
ad_cfg["id"] = None
|
ad_cfg["id"] = None
|
||||||
return True
|
return True
|
||||||
|
|
||||||
def publish_ads(self, ad_cfgs:Iterable[Dict[str, Any]]) -> None:
|
def publish_ads(self, ad_cfgs:Iterable[dict[str, Any]]) -> None:
|
||||||
count = 0
|
count = 0
|
||||||
|
|
||||||
for (ad_file, ad_cfg, ad_cfg_orig) in ad_cfgs:
|
for (ad_file, ad_cfg, ad_cfg_orig) in ad_cfgs:
|
||||||
@@ -327,7 +328,7 @@ class KleinanzeigenBot(SeleniumMixin):
|
|||||||
LOG.info("(Re-)published %s", pluralize("ad", count))
|
LOG.info("(Re-)published %s", pluralize("ad", count))
|
||||||
LOG.info("############################################")
|
LOG.info("############################################")
|
||||||
|
|
||||||
def publish_ad(self, ad_file, ad_cfg: Dict[str, Any], ad_cfg_orig: Dict[str, Any]) -> None:
|
def publish_ad(self, ad_file, ad_cfg: dict[str, Any], ad_cfg_orig: dict[str, Any]) -> None:
|
||||||
self.delete_ad(ad_cfg)
|
self.delete_ad(ad_cfg)
|
||||||
|
|
||||||
LOG.info("Publishing ad '%s'...", ad_cfg["title"])
|
LOG.info("Publishing ad '%s'...", ad_cfg["title"])
|
||||||
|
|||||||
@@ -3,7 +3,8 @@ Copyright (C) 2022 Sebastian Thomschke and contributors
|
|||||||
SPDX-License-Identifier: AGPL-3.0-or-later
|
SPDX-License-Identifier: AGPL-3.0-or-later
|
||||||
"""
|
"""
|
||||||
import logging, os, shutil, sys
|
import logging, os, shutil, sys
|
||||||
from typing import Any, Callable, Dict, Final, Iterable, Tuple
|
from collections.abc import Callable, Iterable
|
||||||
|
from typing import Any, Final
|
||||||
|
|
||||||
from selenium import webdriver
|
from selenium import webdriver
|
||||||
from selenium.common.exceptions import NoSuchElementException, TimeoutException
|
from selenium.common.exceptions import NoSuchElementException, TimeoutException
|
||||||
@@ -107,7 +108,7 @@ class SeleniumMixin:
|
|||||||
|
|
||||||
LOG.info("New WebDriver session is: %s %s", self.webdriver.session_id, self.webdriver.command_executor._url) # pylint: disable=protected-access
|
LOG.info("New WebDriver session is: %s %s", self.webdriver.session_id, self.webdriver.command_executor._url) # pylint: disable=protected-access
|
||||||
|
|
||||||
def get_browser_version(self, executable_path: str) -> Tuple[ChromeType, str]:
|
def get_browser_version(self, executable_path: str) -> tuple[ChromeType, str]:
|
||||||
if sys.platform == "win32":
|
if sys.platform == "win32":
|
||||||
import win32api # pylint: disable=import-outside-toplevel,import-error
|
import win32api # pylint: disable=import-outside-toplevel,import-error
|
||||||
# pylint: disable=no-member
|
# pylint: disable=no-member
|
||||||
@@ -136,7 +137,7 @@ class SeleniumMixin:
|
|||||||
return (ChromeType.MSEDGE, version)
|
return (ChromeType.MSEDGE, version)
|
||||||
return (ChromeType.GOOGLE, version)
|
return (ChromeType.GOOGLE, version)
|
||||||
|
|
||||||
def get_browser_version_from_os(self) -> Tuple[ChromeType, str]:
|
def get_browser_version_from_os(self) -> tuple[ChromeType, str]:
|
||||||
version = ChromeDriverManagerUtils.get_browser_version_from_os(ChromeType.CHROMIUM)
|
version = ChromeDriverManagerUtils.get_browser_version_from_os(ChromeType.CHROMIUM)
|
||||||
if version != "UNKNOWN":
|
if version != "UNKNOWN":
|
||||||
return (ChromeType.CHROMIUM, version)
|
return (ChromeType.CHROMIUM, version)
|
||||||
@@ -211,7 +212,7 @@ class SeleniumMixin:
|
|||||||
WebDriverWait(self.webdriver, timeout).until(lambda _: self.web_execute("return document.readyState") == "complete")
|
WebDriverWait(self.webdriver, timeout).until(lambda _: self.web_execute("return document.readyState") == "complete")
|
||||||
|
|
||||||
# pylint: disable=dangerous-default-value
|
# pylint: disable=dangerous-default-value
|
||||||
def web_request(self, url:str, method:str = "GET", valid_response_codes:Iterable[int] = [200], headers:Dict[str, str] = None) -> Dict[str, Any]:
|
def web_request(self, url:str, method:str = "GET", valid_response_codes:Iterable[int] = [200], headers:dict[str, str] = None) -> dict[str, Any]:
|
||||||
method = method.upper()
|
method = method.upper()
|
||||||
LOG.debug(" -> HTTP %s [%s]...", method, url)
|
LOG.debug(" -> HTTP %s [%s]...", method, url)
|
||||||
response = self.webdriver.execute_async_script(f"""
|
response = self.webdriver.execute_async_script(f"""
|
||||||
|
|||||||
@@ -4,8 +4,9 @@ SPDX-License-Identifier: AGPL-3.0-or-later
|
|||||||
"""
|
"""
|
||||||
import copy, json, logging, os, secrets, sys, traceback, time
|
import copy, json, logging, os, secrets, sys, traceback, time
|
||||||
from importlib.resources import read_text as get_resource_as_string
|
from importlib.resources import read_text as get_resource_as_string
|
||||||
|
from collections.abc import Iterable
|
||||||
from types import ModuleType
|
from types import ModuleType
|
||||||
from typing import Any, Dict, Final, Iterable, Optional, Union
|
from typing import Any, Final
|
||||||
|
|
||||||
import coloredlogs, inflect
|
import coloredlogs, inflect
|
||||||
from ruamel.yaml import YAML
|
from ruamel.yaml import YAML
|
||||||
@@ -30,7 +31,7 @@ def is_frozen() -> bool:
|
|||||||
return getattr(sys, "frozen", False)
|
return getattr(sys, "frozen", False)
|
||||||
|
|
||||||
|
|
||||||
def apply_defaults(target:Dict[Any, Any], defaults:Dict[Any, Any], ignore = lambda _k, _v: False, override = lambda _k, _v: False) -> Dict[Any, Any]:
|
def apply_defaults(target:dict[Any, Any], defaults:dict[Any, Any], ignore = lambda _k, _v: False, override = lambda _k, _v: False) -> dict[Any, Any]:
|
||||||
"""
|
"""
|
||||||
>>> apply_defaults({}, {"foo": "bar"})
|
>>> apply_defaults({}, {"foo": "bar"})
|
||||||
{'foo': 'bar'}
|
{'foo': 'bar'}
|
||||||
@@ -47,17 +48,16 @@ def apply_defaults(target:Dict[Any, Any], defaults:Dict[Any, Any], ignore = lamb
|
|||||||
"""
|
"""
|
||||||
for key, default_value in defaults.items():
|
for key, default_value in defaults.items():
|
||||||
if key in target:
|
if key in target:
|
||||||
if isinstance(target[key], Dict) and isinstance(default_value, Dict):
|
if isinstance(target[key], dict) and isinstance(default_value, dict):
|
||||||
apply_defaults(target[key], default_value, ignore = ignore)
|
apply_defaults(target[key], default_value, ignore = ignore)
|
||||||
elif override(key, target[key]):
|
elif override(key, target[key]):
|
||||||
target[key] = copy.deepcopy(default_value)
|
target[key] = copy.deepcopy(default_value)
|
||||||
else:
|
elif not ignore(key, default_value):
|
||||||
if not ignore(key, default_value):
|
|
||||||
target[key] = copy.deepcopy(default_value)
|
target[key] = copy.deepcopy(default_value)
|
||||||
return target
|
return target
|
||||||
|
|
||||||
|
|
||||||
def safe_get(a_map:Dict[Any, Any], *keys:str) -> Any:
|
def safe_get(a_map:dict[Any, Any], *keys:str) -> Any:
|
||||||
"""
|
"""
|
||||||
>>> safe_get({"foo": {}}, "foo", "bar") is None
|
>>> safe_get({"foo": {}}, "foo", "bar") is None
|
||||||
True
|
True
|
||||||
@@ -119,7 +119,7 @@ def pause(min_ms:int = 200, max_ms:int = 2000) -> None:
|
|||||||
time.sleep(duration / 1000)
|
time.sleep(duration / 1000)
|
||||||
|
|
||||||
|
|
||||||
def pluralize(word:str, count:Union[int, Iterable], prefix = True):
|
def pluralize(word:str, count:int | Iterable, prefix = True):
|
||||||
"""
|
"""
|
||||||
>>> pluralize("field", 1)
|
>>> pluralize("field", 1)
|
||||||
'1 field'
|
'1 field'
|
||||||
@@ -138,7 +138,7 @@ def pluralize(word:str, count:Union[int, Iterable], prefix = True):
|
|||||||
return plural
|
return plural
|
||||||
|
|
||||||
|
|
||||||
def load_dict(filepath:str, content_label:str = "", must_exist = True) -> Optional[Dict[str, Any]]:
|
def load_dict(filepath:str, content_label:str = "", must_exist = True) -> dict[str, Any] | None:
|
||||||
filepath = os.path.abspath(filepath)
|
filepath = os.path.abspath(filepath)
|
||||||
LOG.info("Loading %s[%s]...", content_label and content_label + " from " or "", filepath)
|
LOG.info("Loading %s[%s]...", content_label and content_label + " from " or "", filepath)
|
||||||
|
|
||||||
@@ -155,7 +155,7 @@ def load_dict(filepath:str, content_label:str = "", must_exist = True) -> Option
|
|||||||
return json.load(file) if filepath.endswith(".json") else YAML().load(file)
|
return json.load(file) if filepath.endswith(".json") else YAML().load(file)
|
||||||
|
|
||||||
|
|
||||||
def load_dict_from_module(module:ModuleType, filename:str, content_label:str = "", must_exist = True) -> Optional[Dict[str, Any]]:
|
def load_dict_from_module(module:ModuleType, filename:str, content_label:str = "", must_exist = True) -> dict[str, Any] | None:
|
||||||
LOG.debug("Loading %s[%s.%s]...", content_label and content_label + " from " or "", module.__name__, filename)
|
LOG.debug("Loading %s[%s.%s]...", content_label and content_label + " from " or "", module.__name__, filename)
|
||||||
|
|
||||||
_, file_ext = os.path.splitext(filename)
|
_, file_ext = os.path.splitext(filename)
|
||||||
@@ -172,7 +172,7 @@ def load_dict_from_module(module:ModuleType, filename:str, content_label:str = "
|
|||||||
return json.loads(content) if filename.endswith(".json") else YAML().load(content)
|
return json.loads(content) if filename.endswith(".json") else YAML().load(content)
|
||||||
|
|
||||||
|
|
||||||
def save_dict(filepath:str, content:Dict[str, Any]) -> None:
|
def save_dict(filepath:str, content:dict[str, Any]) -> None:
|
||||||
filepath = os.path.abspath(filepath)
|
filepath = os.path.abspath(filepath)
|
||||||
LOG.info("Saving [%s]...", filepath)
|
LOG.info("Saving [%s]...", filepath)
|
||||||
with open(filepath, "w", encoding = "utf-8") as file:
|
with open(filepath, "w", encoding = "utf-8") as file:
|
||||||
|
|||||||
@@ -70,9 +70,9 @@ dev = [
|
|||||||
app = "python -m kleinanzeigen_bot"
|
app = "python -m kleinanzeigen_bot"
|
||||||
compile = "python -O -m PyInstaller pyinstaller.spec --clean"
|
compile = "python -O -m PyInstaller pyinstaller.spec --clean"
|
||||||
format = "autopep8 --recursive --in-place kleinanzeigen_bot tests"
|
format = "autopep8 --recursive --in-place kleinanzeigen_bot tests"
|
||||||
lint = "pylint kleinanzeigen_bot"
|
lint = "pylint -v kleinanzeigen_bot tests"
|
||||||
scan = "bandit -c pyproject.toml -r kleinanzeigen_bot"
|
scan = "bandit -c pyproject.toml -r kleinanzeigen_bot"
|
||||||
test = "python -m pytest -v"
|
test = "python -m pytest --capture=tee-sys -v"
|
||||||
|
|
||||||
|
|
||||||
#####################
|
#####################
|
||||||
@@ -109,6 +109,19 @@ extension-pkg-whitelist = "win32api"
|
|||||||
ignore = "version.py"
|
ignore = "version.py"
|
||||||
jobs = 4
|
jobs = 4
|
||||||
persistent = "no"
|
persistent = "no"
|
||||||
|
load-plugins = [
|
||||||
|
"pylint.extensions.bad_builtin",
|
||||||
|
"pylint.extensions.comparetozero",
|
||||||
|
"pylint.extensions.check_elif",
|
||||||
|
"pylint.extensions.code_style",
|
||||||
|
"pylint.extensions.comparison_placement",
|
||||||
|
"pylint.extensions.empty_comment",
|
||||||
|
"pylint.extensions.for_any_all",
|
||||||
|
"pylint.extensions.overlapping_exceptions",
|
||||||
|
"pylint.extensions.redefined_variable_type",
|
||||||
|
"pylint.extensions.set_membership",
|
||||||
|
"pylint.extensions.typing",
|
||||||
|
]
|
||||||
|
|
||||||
[tool.pylint.basic]
|
[tool.pylint.basic]
|
||||||
good-names = ["i", "j", "k", "v", "by", "ex", "fd", "_"]
|
good-names = ["i", "j", "k", "v", "by", "ex", "fd", "_"]
|
||||||
@@ -124,6 +137,7 @@ logging-modules = "logging"
|
|||||||
# https://pylint.pycqa.org/en/latest/technical_reference/features.html#messages-control-options
|
# https://pylint.pycqa.org/en/latest/technical_reference/features.html#messages-control-options
|
||||||
disable= [
|
disable= [
|
||||||
"broad-except",
|
"broad-except",
|
||||||
|
"consider-using-assignment-expr",
|
||||||
"missing-docstring",
|
"missing-docstring",
|
||||||
"multiple-imports",
|
"multiple-imports",
|
||||||
"multiple-statements",
|
"multiple-statements",
|
||||||
|
|||||||
Reference in New Issue
Block a user