1 Commits

Author SHA1 Message Date
=
b726f2f668 1.0.7 2025-07-06 18:52:41 +02:00
7 changed files with 21 additions and 50 deletions

2
.gitignore vendored
View File

@@ -181,7 +181,7 @@ cython_debug/
# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
# and can be added to the global gitignore or merged into this file. However, if you prefer,
# you could uncomment the following to ignore the enitre vscode folder
.vscode/
# .vscode/
# Ruff stuff:
.ruff_cache/

View File

@@ -1,16 +1,3 @@
## 1.0.10
### Fix
* KeyError when using shippable=True (#5)
## 1.0.9
### Fix
* Fixed SSL verification issue during Leboncoin cookie initialization.
## 1.0.8
### Added
* `max_retries` and `timeout` parameters to `Client`.
* `NotFoundError` exception raised when an ad or user is not found.
## 1.0.7
### Added
* Automatic rotation of browser impersonation when `impersonate` argument in `Client` is set to None.

View File

@@ -1,6 +1,6 @@
[project]
name = "lbc"
version = "1.0.10"
version = "1.0.7"
description = "Unofficial client for Leboncoin API"
readme = "README.md"
license = {text = "MIT"}

View File

@@ -1,14 +1,13 @@
from .session import Session
from .models import Proxy, Search, Category, AdType, OwnerType, Sort, Region, Department, City, User, Ad
from .exceptions import DatadomeError, RequestError, NotFoundError
from .exceptions import DatadomeError, RequestError
from .utils import build_search_payload_with_args, build_search_payload_with_url
from typing import Optional, List, Union
from curl_cffi import BrowserTypeLiteral
class Client(Session):
def __init__(self, proxy: Optional[Proxy] = None, impersonate: BrowserTypeLiteral = None,
request_verify: bool = True, timeout: int = 30, max_retries: int = 5):
def __init__(self, proxy: Optional[Proxy] = None, impersonate: BrowserTypeLiteral = None, request_verify: bool = True):
"""
Initializes a Leboncoin Client instance with optional proxy, browser impersonation, and SSL verification settings.
@@ -18,25 +17,20 @@ class Client(Session):
proxy (Optional[Proxy], optional): Proxy configuration to use for the client. If provided, it will be applied to all requests. Defaults to None.
impersonate (BrowserTypeLiteral, optional): Browser type to impersonate for requests (e.g., "firefox", "chrome", "edge", "safari", "safari_ios", "chrome_android"). If None, a random browser type will be chosen.
request_verify (bool, optional): Whether to verify SSL certificates when sending requests. Set to False to disable SSL verification (not recommended for production). Defaults to True.
timeout (int, optional): Maximum time in seconds to wait for a request before timing out. Defaults to 30.
max_retries (int, optional): Maximum number of times to retry a request in case of failure (403 error). Defaults to 5.
"""
super().__init__(proxy=proxy, impersonate=impersonate, request_verify=request_verify)
self.request_verify = request_verify
self.timeout = timeout
self.max_retries = max_retries
super().__init__(proxy=proxy, impersonate=impersonate)
def _fetch(self, method: str, url: str, payload: Optional[dict] = None, timeout: int = 30, max_retries: int = 5) -> Union[dict, None]:
self.request_verify = request_verify
def _fetch(self, method: str, url: str, payload: Optional[dict] = None, timeout: int = 30) -> Union[dict, None]:
"""
Internal method to send an HTTP request using the configured session.
Args:
method (str): HTTP method to use (e.g., "GET", "POST").
method (staticmethod): HTTP method to use (e.g., `GET`, `POST`).
url (str): Full URL of the API endpoint.
payload (Optional[dict], optional): JSON payload to send with the request. Used for POST/PUT methods. Defaults to None.
timeout (int, optional): Timeout for the request, in seconds. Defaults to 30.
max_retries (int, optional): Number of times to retry the request in case of failure. Defaults to 5.
Raises:
DatadomeError: Raised when the request is blocked by Datadome protection (HTTP 403).
@@ -55,15 +49,10 @@ class Client(Session):
if response.ok:
return response.json()
elif response.status_code == 403:
if max_retries > 0:
self.session = self._init_session(proxy=self._proxy, impersonate=self._impersonate, request_verify=self.request_verify) # Re-init session
return self._fetch(method=method, url=url, payload=payload, timeout=timeout, max_retries=max_retries - 1)
if self.proxy:
raise DatadomeError(f"Access blocked by Datadome: your proxy appears to have a poor reputation, try to change it.")
else:
raise DatadomeError(f"Access blocked by Datadome: your activity was flagged as suspicious. Please avoid sending excessive requests.")
elif response.status_code == 404 or response.status_code == 410:
raise NotFoundError(f"Unable to find ad or user.")
else:
raise RequestError(f"Request failed with status code {response.status_code}.")
@@ -119,7 +108,7 @@ class Client(Session):
owner_type=owner_type, shippable=shippable, search_in_title_only=search_in_title_only, **kwargs
)
body = self._fetch(method="POST", url="https://api.leboncoin.fr/finder/search", payload=payload, timeout=self.timeout, max_retries=self.max_retries)
body = self._fetch(method="POST", url="https://api.leboncoin.fr/finder/search", payload=payload)
return Search._build(raw=body, client=self)
def get_user(
@@ -138,13 +127,13 @@ class Client(Session):
Returns:
User: A `User` object containing the parsed user information.
"""
user_data = self._fetch(method="GET", url=f"https://api.leboncoin.fr/api/user-card/v2/{user_id}/infos", timeout=self.timeout, max_retries=self.max_retries)
user_data = self._fetch(method="GET", url=f"https://api.leboncoin.fr/api/user-card/v2/{user_id}/infos")
pro_data = None
if user_data.get("account_type") == "pro":
try:
pro_data = self._fetch(method="GET", url=f"https://api.leboncoin.fr/api/onlinestores/v2/users/{user_id}?fields=all", timeout=self.timeout, max_retries=self.max_retries)
except NotFoundError:
pro_data = self._fetch(method="GET", url=f"https://api.leboncoin.fr/api/onlinestores/v2/users/{user_id}?fields=all")
except Exception:
pass # Some professional users may not have a Leboncoin page.
return User._build(user_data=user_data, pro_data=pro_data)
@@ -152,7 +141,7 @@ class Client(Session):
def get_ad(
self,
ad_id: Union[str, int]
) -> Ad:
):
"""
Retrieve detailed information about a classified ad using its ID.
@@ -166,6 +155,6 @@ class Client(Session):
Returns:
Ad: An `Ad` object containing the parsed ad information.
"""
body = self._fetch(method="GET", url=f"https://api.leboncoin.fr/api/adfinder/v1/classified/{ad_id}", timeout=self.timeout, max_retries=self.max_retries)
body = self._fetch(method="GET", url=f"https://api.leboncoin.fr/api/adfinder/v1/classified/{ad_id}")
return Ad._build(raw=body, client=self)

View File

@@ -12,7 +12,3 @@ class RequestError(LBCError):
class DatadomeError(RequestError):
"""Raised when access is blocked by Datadome anti-bot protection."""
class NotFoundError(LBCError):
"""Raised when a user or ad is not found."""

View File

@@ -5,12 +5,12 @@ from typing import Optional
import random
class Session:
def __init__(self, proxy: Optional[Proxy] = None, impersonate: BrowserTypeLiteral = None, request_verify: bool = True):
self._session = self._init_session(proxy=proxy, impersonate=impersonate, request_verify=request_verify)
def __init__(self, proxy: Optional[Proxy] = None, impersonate: BrowserTypeLiteral = None):
self._session = self._init_session(proxy=proxy, impersonate=impersonate)
self._proxy = proxy
self._impersonate = impersonate
def _init_session(self, proxy: Optional[Proxy] = None, impersonate: BrowserTypeLiteral = None, request_verify: bool = True) -> requests.Session:
def _init_session(self, proxy: Optional[Proxy] = None, impersonate: BrowserTypeLiteral = None) -> requests.Session:
"""
Initializes an HTTP session with optional proxy configuration and browser impersonation.
@@ -19,8 +19,7 @@ class Session:
Args:
proxy (Optional[Proxy], optional): Proxy configuration to use for the session. If provided, it will be applied to both HTTP and HTTPS traffic. Defaults to None.
impersonate (BrowserTypeLiteral, optional): Browser type to impersonate for requests (e.g., "firefox", "chrome", "edge", "safari", "safari_ios", "chrome_android"). If None, a random browser type will be chosen.
request_verify (bool, optional): Whether to verify SSL certificates for HTTPS requests. Defaults to True.
Returns:
requests.Session: A configured session instance ready to send requests.
"""
@@ -54,7 +53,7 @@ class Session:
"https": proxy.url
}
session.get("https://www.leboncoin.fr/", verify=request_verify) # Init cookies
session.get("https://www.leboncoin.fr/") # Init cookies
return session

View File

@@ -251,7 +251,7 @@ def build_search_payload_with_args(
payload["filters"]["keywords"]["type"] = "subject"
if shippable:
payload["filters"]["location"]["shippable"] = True
payload["filters"]["locations"]["shippable"] = True
if kwargs:
for key, value in kwargs.items():