diff --git a/examples/discord/config/__init__.py b/examples/discord/config/__init__.py index 2031e3a..82f080d 100644 --- a/examples/discord/config/__init__.py +++ b/examples/discord/config/__init__.py @@ -3,11 +3,11 @@ import lbc from .handler import handle -location = lbc.City( +location = lbc.City( lat=48.85994982004764, lng=2.33801967847424, - radius=10_000, # 10 km - city="Paris" + radius=10_000, # 10 km + city="Paris", ) CONFIG = [ @@ -18,9 +18,9 @@ CONFIG = [ locations=[location], category=lbc.Category.IMMOBILIER, square=[200, 400], - price=[300_000, 700_000] + price=[300_000, 700_000], ), - delay=60 * 5, # Check every 5 minutes - handler=handle + delay=60 * 5, # Check every 5 minutes + handler=handle, ), -] \ No newline at end of file +] diff --git a/examples/discord/config/handler.py b/examples/discord/config/handler.py index 8fbca5b..31cc6f2 100644 --- a/examples/discord/config/handler.py +++ b/examples/discord/config/handler.py @@ -1,50 +1,43 @@ import lbc import requests from datetime import datetime -from typing import Final -WEBHOOK_URL: Final[str] = ... +WEBHOOK_URL: str = ... + def handle(ad: lbc.Ad, search_name: str) -> None: timestamp = datetime.strptime(ad.index_date, "%Y-%m-%d %H:%M:%S").timestamp() - + payload = { "content": None, "embeds": [ { "title": ad.title, - "description": f"```{ad.body[:4087]}...```" if len(ad.body) >= 4090 else f"```{ad.body[:4090]}```", + "description": f"```{ad.body[:4087]}...```" + if len(ad.body) >= 4090 + else f"```{ad.body[:4090]}```", "url": ad.url, "color": 14381568, - "author": { - "name": ad.user.name, - "icon_url": ad.user.profile_picture - }, - "image": { - "url": ad.images[0] if ad.images else None - }, + "author": {"name": ad.user.name, "icon_url": ad.user.profile_picture}, + "image": {"url": ad.images[0] if ad.images else None}, "fields": [ { "name": "🕒 Publication", "value": f"", - "inline": True - }, - { - "name": "💰 Price", - "value": f"`{ad.price}€`", - "inline": True + "inline": True, }, + {"name": "💰 Price", "value": f"`{ad.price}€`", "inline": True}, { "name": "📍 Location", "value": f"`{ad.location.city_label}`", - "inline": True - } + "inline": True, + }, ], } ], "username": search_name, - "attachments": [] + "attachments": [], } response = requests.post(WEBHOOK_URL, json=payload) - response.raise_for_status() \ No newline at end of file + response.raise_for_status() diff --git a/lbc-finder/config/__init__.py b/lbc-finder/config/__init__.py index 2031e3a..82f080d 100644 --- a/lbc-finder/config/__init__.py +++ b/lbc-finder/config/__init__.py @@ -3,11 +3,11 @@ import lbc from .handler import handle -location = lbc.City( +location = lbc.City( lat=48.85994982004764, lng=2.33801967847424, - radius=10_000, # 10 km - city="Paris" + radius=10_000, # 10 km + city="Paris", ) CONFIG = [ @@ -18,9 +18,9 @@ CONFIG = [ locations=[location], category=lbc.Category.IMMOBILIER, square=[200, 400], - price=[300_000, 700_000] + price=[300_000, 700_000], ), - delay=60 * 5, # Check every 5 minutes - handler=handle + delay=60 * 5, # Check every 5 minutes + handler=handle, ), -] \ No newline at end of file +] diff --git a/lbc-finder/config/handler.py b/lbc-finder/config/handler.py index 1869d72..4c10aee 100644 --- a/lbc-finder/config/handler.py +++ b/lbc-finder/config/handler.py @@ -1,9 +1,9 @@ import lbc + def handle(ad: lbc.Ad, search_name: str): print(f"[{search_name}] New ads!") print(f"Title : {ad.subject}") print(f"Price : {ad.price} €") print(f"URL : {ad.url}") print("-" * 40) - diff --git a/lbc-finder/main.py b/lbc-finder/main.py index 0c239c0..c1d9de5 100644 --- a/lbc-finder/main.py +++ b/lbc-finder/main.py @@ -1,9 +1,11 @@ from searcher import Searcher from config import CONFIG + def main() -> None: searcher = Searcher(searches=CONFIG) searcher.start() + if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/lbc-finder/model/__init__.py b/lbc-finder/model/__init__.py index 14b8f9a..a1c9259 100644 --- a/lbc-finder/model/__init__.py +++ b/lbc-finder/model/__init__.py @@ -1,2 +1,4 @@ from .search import Search -from .parameters import Parameters \ No newline at end of file +from .parameters import Parameters + +__all__ = ["Search", "Parameters"] diff --git a/lbc-finder/model/parameters.py b/lbc-finder/model/parameters.py index 2e1d40e..8816f35 100644 --- a/lbc-finder/model/parameters.py +++ b/lbc-finder/model/parameters.py @@ -1,24 +1,28 @@ -from typing import Optional, Union, List from lbc import Category, Region, Department, City, OwnerType from typing import overload + class Parameters: @overload def __init__( self, - url: Optional[str] = None, - text: Optional[str] = None, + url: str | None = None, + text: str | None = None, category: Category = Category.TOUTES_CATEGORIES, - locations: Optional[Union[List[Union[Region, Department, City]], Union[Region, Department, City]]] = None, + locations: list[Region | Department | City] + | Region + | Department + | City + | None = None, limit: int = 35, limit_alu: int = 3, page: int = 1, - owner_type: Optional[OwnerType] = None, - shippable: Optional[bool] = None, + owner_type: OwnerType | None = None, + shippable: bool | None = None, search_in_title_only: bool = False, - **kwargs + **kwargs, ): ... - + def __init__(self, **kwargs): - self._kwargs = kwargs \ No newline at end of file + self._kwargs = kwargs diff --git a/lbc-finder/model/search.py b/lbc-finder/model/search.py index 417290d..6d8f41b 100644 --- a/lbc-finder/model/search.py +++ b/lbc-finder/model/search.py @@ -1,7 +1,8 @@ from lbc import Proxy, Ad from .parameters import Parameters from dataclasses import dataclass -from typing import Callable, Optional +from typing import Callable + @dataclass class Search: @@ -9,4 +10,4 @@ class Search: parameters: Parameters delay: float handler: Callable[[Ad, str], None] - proxy: Optional[Proxy] = None \ No newline at end of file + proxy: Proxy | None = None diff --git a/lbc-finder/searcher/__init__.py b/lbc-finder/searcher/__init__.py index f9f57b2..c90d7cf 100644 --- a/lbc-finder/searcher/__init__.py +++ b/lbc-finder/searcher/__init__.py @@ -1,2 +1,4 @@ from .searcher import Searcher -from .logger import logger \ No newline at end of file +from .logger import logger + +__all__ = ["Searcher", "logger"] diff --git a/lbc-finder/searcher/id.py b/lbc-finder/searcher/id.py index 2a3831c..43e2663 100644 --- a/lbc-finder/searcher/id.py +++ b/lbc-finder/searcher/id.py @@ -1,21 +1,21 @@ from .logger import logger -from typing import List, Final import os import json -MAX_ID: Final[int] = 10_000 +MAX_ID: int = 10_000 + class ID: def __init__(self): - self._ids: List[str] = self._get_ids() + self._ids: list[str] = self._get_ids() @property - def ids(self) -> List[str]: + def ids(self) -> list[str]: return self._ids - def _get_ids(self) -> List[str]: - ids: List[str] = [] + def _get_ids(self) -> list[str]: + ids: list[str] = [] id_path = os.path.join("data", "id.json") if os.path.exists(id_path): with open(id_path, "r") as f: @@ -23,8 +23,10 @@ class ID: ids = json.load(f) except json.JSONDecodeError: os.remove(id_path) - except: - logger.exception("An error occurred while attempting to open the id.json file.") + except Exception: + logger.exception( + "An error occurred while attempting to open the id.json file." + ) return ids def contains(self, id_: str) -> bool: @@ -32,7 +34,7 @@ class ID: def add(self, id_: str) -> bool: id_path = os.path.join("data", "id.json") - if not id_ in self._ids: + if id_ not in self._ids: self._ids.append(id_) with open(id_path, "w") as f: json.dump(self._ids[-MAX_ID:], f, indent=3) diff --git a/lbc-finder/searcher/logger.py b/lbc-finder/searcher/logger.py index 788acde..73dfe18 100644 --- a/lbc-finder/searcher/logger.py +++ b/lbc-finder/searcher/logger.py @@ -10,14 +10,17 @@ os.makedirs(os.path.join("data", "logs"), exist_ok=True) # Config logging logger = logging.getLogger("lbc-finder") -formatter = logging.Formatter('[%(asctime)s] [%(levelname)s] [%(threadName)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S') +formatter = logging.Formatter( + "[%(asctime)s] [%(levelname)s] [%(threadName)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) stream_handler = logging.StreamHandler() stream_handler.setFormatter(formatter) logger.addHandler(stream_handler) logger.setLevel(logging.INFO) # Log File -file_handler = logging.FileHandler(file_path, mode='w', encoding='utf-8') +file_handler = logging.FileHandler(file_path, mode="w", encoding="utf-8") file_handler.setLevel(logging.WARNING) file_handler.setFormatter(formatter) -logger.addHandler(file_handler) \ No newline at end of file +logger.addHandler(file_handler) diff --git a/lbc-finder/searcher/searcher.py b/lbc-finder/searcher/searcher.py index 86b7288..2f71465 100644 --- a/lbc-finder/searcher/searcher.py +++ b/lbc-finder/searcher/searcher.py @@ -5,12 +5,19 @@ from .logger import logger import time import threading -from typing import List, Union + class Searcher: - def __init__(self, searches: Union[List[Search], Search], request_verify: bool = True, - handler_max_attempts: int = 3, handler_initial_backoff: float = 2.0): - self._searches: List[Search] = searches if isinstance(searches, list) else [searches] + def __init__( + self, + searches: list[Search] | Search, + request_verify: bool = True, + handler_max_attempts: int = 3, + handler_initial_backoff: float = 2.0, + ): + self._searches: list[Search] = ( + searches if isinstance(searches, list) else [searches] + ) self._request_verify = request_verify self._handler_max_attempts = handler_max_attempts self._handler_initial_backoff = handler_initial_backoff @@ -42,10 +49,14 @@ class Searcher: before = time.time() try: response = client.search(**search.parameters._kwargs, sort=Sort.NEWEST) - logger.debug(f"Successfully found {response.total} ad{'s' if response.total > 1 else ''}.") + logger.debug( + f"Successfully found {response.total} ad{'s' if response.total > 1 else ''}." + ) ads = [ad for ad in response.ads if not self._id.contains(ad.id)] if len(ads): - logger.info(f"Successfully found {len(ads)} new ad{'s' if len(ads) > 1 else ''}!") + logger.info( + f"Successfully found {len(ads)} new ad{'s' if len(ads) > 1 else ''}!" + ) notified = 0 for ad in ads: @@ -57,15 +68,23 @@ class Searcher: f"[{search.name}] {len(ads) - notified} ad{'s were' if len(ads) - notified > 1 else ' was'} not marked as seen and will be retried." ) except Exception: - logger.exception(f"An error occured.") - time.sleep(search.delay - (time.time() - before) if search.delay - (time.time() - before) > 0 else 0) + logger.exception("An error occured.") + time.sleep( + search.delay - (time.time() - before) + if search.delay - (time.time() - before) > 0 + else 0 + ) def start(self) -> bool: if not len(self._searches): - logger.warning("No search rules have been set. Please create search rules in config.py (see example in README.md).") + logger.warning( + "No search rules have been set. Please create search rules in config.py (see example in README.md)." + ) return False for search in self._searches: - threading.Thread(target=self._search, args=(search,), name=search.name).start() - time.sleep(5) # Add latency between each thread to prevent spam + threading.Thread( + target=self._search, args=(search,), name=search.name + ).start() + time.sleep(5) # Add latency between each thread to prevent spam return True