style: format lbc-finder with ruff

This commit is contained in:
etienne-hd
2026-04-29 19:44:17 +02:00
parent 73e33ab782
commit 90469d9025
12 changed files with 101 additions and 73 deletions

View File

@@ -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,
),
]
]

View File

@@ -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"<t:{int(timestamp)}:R>",
"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()
response.raise_for_status()

View File

@@ -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,
),
]
]

View File

@@ -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)

View File

@@ -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()
main()

View File

@@ -1,2 +1,4 @@
from .search import Search
from .parameters import Parameters
from .parameters import Parameters
__all__ = ["Search", "Parameters"]

View File

@@ -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
self._kwargs = kwargs

View File

@@ -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
proxy: Proxy | None = None

View File

@@ -1,2 +1,4 @@
from .searcher import Searcher
from .logger import logger
from .logger import logger
__all__ = ["Searcher", "logger"]

View File

@@ -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)

View File

@@ -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)
logger.addHandler(file_handler)

View File

@@ -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