mirror of
https://github.com/etienne-hd/lbc.git
synced 2026-04-25 08:15:38 +02:00
Compare commits
15 Commits
b9ac610b04
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a44472e757 | ||
|
|
6669950f82 | ||
|
|
bbbdd1652f | ||
|
|
c891e593c5 | ||
|
|
e8cb0a9f5d | ||
|
|
48b83c0c9d | ||
|
|
671fb291f1 | ||
|
|
81d1799ea8 | ||
|
|
a927bd7cf5 | ||
|
|
8d4d2da64f | ||
|
|
4e69194821 | ||
|
|
00cf534191 | ||
|
|
8a74cde3b1 | ||
|
|
d24b1cc0e6 | ||
|
|
6ef01383f0 |
25
CHANGELOG.md
25
CHANGELOG.md
@@ -1,3 +1,26 @@
|
||||
## 1.1.2
|
||||
### Changed
|
||||
* Proxy handling, you can now remove proxy using `client.proxy = None`
|
||||
* `client._session` is now public -> `client.session`
|
||||
|
||||
### Added
|
||||
* Proxy example at [examples/proxy.py](examples/proxy.py)
|
||||
|
||||
## 1.1.1
|
||||
### Added
|
||||
* Optional `scheme` attribute on the Proxy dataclass, allowing custom proxy URL schemes (#7)
|
||||
|
||||
## 1.1.0
|
||||
### Changed
|
||||
* Project structure reorganized: features such as **search**, **user**, and **ad** are now separated using mixins.
|
||||
|
||||
### Added
|
||||
* Realistic dynamic mobile User-Agent generation.
|
||||
|
||||
## 1.0.10
|
||||
### Fix
|
||||
* KeyError when using shippable=True (#5)
|
||||
|
||||
## 1.0.9
|
||||
### Fix
|
||||
* Fixed SSL verification issue during Leboncoin cookie initialization.
|
||||
@@ -54,4 +77,4 @@
|
||||
* Realistic `Sec-Fetch-*` headers to prevent 403 errors
|
||||
|
||||
## 1.0.0
|
||||
* Initial release
|
||||
* Initial release
|
||||
|
||||
29
README.md
29
README.md
@@ -41,7 +41,6 @@ pip install lbc
|
||||
```
|
||||
|
||||
## Usage
|
||||
**Full documentation will be available soon.**
|
||||
|
||||
Start with the [examples](examples/) to quickly understand how to use the library in real-world scenarios.
|
||||
|
||||
@@ -56,13 +55,29 @@ client = lbc.Client()
|
||||
#### Proxy
|
||||
You can also configure the client to use a proxy by providing a `Proxy` object:
|
||||
```python
|
||||
proxy = lbc.Proxy(
|
||||
host=...,
|
||||
port=...,
|
||||
username=...,
|
||||
password=...
|
||||
# Setup proxy1
|
||||
proxy1 = lbc.Proxy(
|
||||
host="127.0.0.1",
|
||||
port=12345,
|
||||
username="username",
|
||||
password="password",
|
||||
scheme="http"
|
||||
)
|
||||
client = lbc.Client(proxy=proxy)
|
||||
|
||||
# Initialize client with proxy1
|
||||
client = lbc.Client(proxy=proxy1)
|
||||
|
||||
# Setup proxy2
|
||||
proxy2 = lbc.Proxy(
|
||||
host="127.0.0.1",
|
||||
port=23456,
|
||||
)
|
||||
|
||||
# Change client proxy to proxy2
|
||||
client.proxy = proxy2
|
||||
|
||||
# Remove proxy
|
||||
client.proxy = None
|
||||
```
|
||||
|
||||
|
||||
|
||||
29
examples/proxy.py
Normal file
29
examples/proxy.py
Normal file
@@ -0,0 +1,29 @@
|
||||
import lbc
|
||||
|
||||
def main() -> None:
|
||||
# Setup proxy1
|
||||
proxy1 = lbc.Proxy(
|
||||
host="127.0.0.1",
|
||||
port=12345,
|
||||
username="username",
|
||||
password="password",
|
||||
scheme="http"
|
||||
)
|
||||
|
||||
# Initialize client with proxy1
|
||||
client = lbc.Client(proxy=proxy1)
|
||||
|
||||
# Setup proxy2
|
||||
proxy2 = lbc.Proxy(
|
||||
host="127.0.0.1",
|
||||
port=23456,
|
||||
)
|
||||
|
||||
# Change client proxy to proxy2
|
||||
client.proxy = proxy2
|
||||
|
||||
# Remove proxy
|
||||
client.proxy = None
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,2 +1,2 @@
|
||||
from .client import Client
|
||||
from .models import *
|
||||
from .model import *
|
||||
80
lbc/client.py
Normal file
80
lbc/client.py
Normal file
@@ -0,0 +1,80 @@
|
||||
from typing import Optional, Union
|
||||
from curl_cffi import BrowserTypeLiteral
|
||||
|
||||
from .mixin import (
|
||||
SessionMixin,
|
||||
SearchMixin,
|
||||
UserMixin,
|
||||
AdMixin
|
||||
)
|
||||
from .model import Proxy
|
||||
from .exceptions import DatadomeError, RequestError, NotFoundError
|
||||
|
||||
class Client(
|
||||
SessionMixin,
|
||||
SearchMixin,
|
||||
UserMixin,
|
||||
AdMixin
|
||||
):
|
||||
def __init__(self, proxy: Optional[Proxy] = None, impersonate: BrowserTypeLiteral = None,
|
||||
request_verify: bool = True, timeout: float = 30.0, max_retries: int = 5):
|
||||
"""
|
||||
Initializes a Leboncoin Client instance with optional proxy, browser impersonation, and SSL verification settings.
|
||||
|
||||
If no `impersonate` value is provided, a random browser type will be selected among common options.
|
||||
|
||||
Args:
|
||||
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
|
||||
|
||||
def _fetch(self, method: str, url: str, payload: Optional[dict] = None, max_retries: int = -1) -> dict:
|
||||
"""
|
||||
Internal method to send an HTTP request using the configured session.
|
||||
|
||||
Args:
|
||||
method (str): 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).
|
||||
RequestError: Raised for any other non-successful HTTP response.
|
||||
|
||||
Returns:
|
||||
dict: Parsed JSON response from the server.
|
||||
"""
|
||||
if max_retries == -1:
|
||||
max_retries = self.max_retries
|
||||
|
||||
response = self.session.request(
|
||||
method=method,
|
||||
url=url,
|
||||
json=payload,
|
||||
verify=self.request_verify,
|
||||
timeout=self.timeout
|
||||
)
|
||||
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, 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}.")
|
||||
@@ -1,18 +1,14 @@
|
||||
class LBCError(Exception):
|
||||
"""Base exception for all errors raised by the LBC client."""
|
||||
|
||||
|
||||
class InvalidValue(LBCError):
|
||||
"""Raised when a provided value is invalid or improperly formatted."""
|
||||
|
||||
|
||||
class RequestError(LBCError):
|
||||
"""Raised when an HTTP request fails with a non-success status code."""
|
||||
|
||||
|
||||
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."""
|
||||
4
lbc/mixin/__init__.py
Normal file
4
lbc/mixin/__init__.py
Normal file
@@ -0,0 +1,4 @@
|
||||
from .session import SessionMixin
|
||||
from .search import SearchMixin
|
||||
from .user import UserMixin
|
||||
from .ad import AdMixin
|
||||
21
lbc/mixin/ad.py
Normal file
21
lbc/mixin/ad.py
Normal file
@@ -0,0 +1,21 @@
|
||||
from typing import Union
|
||||
|
||||
from ..model import Ad
|
||||
|
||||
class AdMixin:
|
||||
def get_ad(self, ad_id: Union[str, int]) -> Ad:
|
||||
"""
|
||||
Retrieve detailed information about a classified ad using its ID.
|
||||
|
||||
This method fetches the full content of an ad, including its description,
|
||||
pricing, location, and other relevant metadata made
|
||||
available through the public Leboncoin ad API.
|
||||
|
||||
Args:
|
||||
ad_id (Union[str, int]): The unique identifier of the ad on Leboncoin. Can be found in the ad URL.
|
||||
|
||||
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}")
|
||||
return Ad._build(raw=body, client=self)
|
||||
60
lbc/mixin/search.py
Normal file
60
lbc/mixin/search.py
Normal file
@@ -0,0 +1,60 @@
|
||||
from typing import Optional, Union, List
|
||||
|
||||
from ..model import Category, Sort, Region, Department, City, AdType, OwnerType, Search
|
||||
from ..utils import build_search_payload_with_args, build_search_payload_with_url
|
||||
|
||||
class SearchMixin:
|
||||
def search(
|
||||
self,
|
||||
url: Optional[str] = None,
|
||||
text: Optional[str] = None,
|
||||
category: Category = Category.TOUTES_CATEGORIES,
|
||||
sort: Sort = Sort.RELEVANCE,
|
||||
locations: Optional[Union[List[Union[Region, Department, City]], Union[Region, Department, City]]] = None,
|
||||
limit: int = 35,
|
||||
limit_alu: int = 3,
|
||||
page: int = 1,
|
||||
ad_type: AdType = AdType.OFFER,
|
||||
owner_type: Optional[OwnerType] = None,
|
||||
shippable: Optional[bool] = None,
|
||||
search_in_title_only: bool = False,
|
||||
**kwargs
|
||||
) -> Search:
|
||||
"""
|
||||
Perform a classified ads search on Leboncoin with the specified criteria.
|
||||
|
||||
You can either:
|
||||
- Provide a full `url` from a Leboncoin search to replicate the search directly.
|
||||
- Or use the individual parameters (`text`, `category`, `locations`, etc.) to construct a custom search.
|
||||
|
||||
Args:
|
||||
url (Optional[str], optional): A full Leboncoin search URL. If provided, all other parameters will be ignored and the search will replicate the results from the URL.
|
||||
text (Optional[str], optional): Search keywords. If None, returns all matching ads without filtering by keyword. Defaults to None.
|
||||
category (Category, optional): Category to search in. Defaults to Category.TOUTES_CATEGORIES.
|
||||
sort (Sort, optional): Sorting method for results (e.g., relevance, date, price). Defaults to Sort.RELEVANCE.
|
||||
locations (Optional[Union[List[Union[Region, Department, City]], Union[Region, Department, City]]], optional): One or multiple locations (region, department, or city) to filter results. Defaults to None.
|
||||
limit (int, optional): Maximum number of results to return. Defaults to 35.
|
||||
limit_alu (int, optional): Number of ALU (Annonces Lu / similar ads) suggestions to include. Defaults to 3.
|
||||
page (int, optional): Page number to retrieve for paginated results. Defaults to 1.
|
||||
ad_type (AdType, optional): Type of ad (offer or request). Defaults to AdType.OFFER.
|
||||
owner_type (Optional[OwnerType], optional): Filter by seller type (individual, professional, or all). Defaults to None.
|
||||
shippable (Optional[bool], optional): If True, only includes ads that offer shipping. Defaults to None.
|
||||
search_in_title_only (bool, optional): If True, search will only be performed on ad titles. Defaults to False.
|
||||
**kwargs: Additional advanced filters such as price range (`price=(min, max)`), surface area (`square=(min, max)`), property type, and more.
|
||||
|
||||
Returns:
|
||||
Search: A `Search` object containing the parsed search results.
|
||||
"""
|
||||
if url:
|
||||
payload = build_search_payload_with_url(
|
||||
url=url, limit=limit, page=page
|
||||
)
|
||||
else:
|
||||
payload = build_search_payload_with_args(
|
||||
text=text, category=category, sort=sort, locations=locations,
|
||||
limit=limit, limit_alu=limit_alu, page=page, ad_type=ad_type,
|
||||
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)
|
||||
return Search._build(raw=body, client=self)
|
||||
98
lbc/mixin/session.py
Normal file
98
lbc/mixin/session.py
Normal file
@@ -0,0 +1,98 @@
|
||||
from curl_cffi import requests, BrowserTypeLiteral
|
||||
from typing import Optional
|
||||
import random
|
||||
import uuid
|
||||
|
||||
from ..model import Proxy
|
||||
|
||||
class SessionMixin:
|
||||
def __init__(self, proxy: Optional[Proxy] = None, impersonate: BrowserTypeLiteral = None,
|
||||
request_verify: bool = True, **kwargs):
|
||||
self.session = self._init_session(proxy=proxy, impersonate=impersonate, request_verify=request_verify)
|
||||
self._proxy = proxy
|
||||
self._impersonate = impersonate
|
||||
super().__init__(**kwargs)
|
||||
|
||||
def _generate_user_agent(self) -> str:
|
||||
# LBC;iOS;26.2;iPhone;phone;01234567-89AB-CDEF-0123-456789ABCDEF;wifi;101.44.0
|
||||
# LBC;Android;11;Android SDK built for arm64;phone;0123456789ABCDEF;wifi;100.85.2
|
||||
# LBC;<OS>;<OS_VERSION>;<MODEL>;phone;<DEVICE_ID>;wifi;<APP_VERSION>
|
||||
os = random.choice(["iOS", "Android"])
|
||||
if os == "iOS":
|
||||
os_version = random.choice(["18.0", "18.1", "18.2", "18.3", "18.4", "18.5", "18.6", "18.7", "18.7.3",
|
||||
"26.0", "26.1", "26.2"])
|
||||
model = "iPhone"
|
||||
device_id = str(uuid.uuid4())
|
||||
app_version = random.choice(["101.45.0", "101.44.0", "101.43.1", "101.43.0", "101.42.1", "101.42.0", "101.41.0", "101.40.0", "101.39.0", "101.38.0"])
|
||||
else:
|
||||
os_version = random.choice(["11", "12", "13", "14", "15"])
|
||||
model = random.choice(["SM-G991B", "SM-G996B", "SM-G998B", "SM-S911B", "SM-S916B", "SM-S918B", "SM-A505F", "SM-A546B", "SM-A137F", "SM-M336B",
|
||||
"Pixel 5", "Pixel 6", "Pixel 6a", "Pixel 7", "Pixel 7 Pro", "Pixel 8", "Pixel 8 Pro",
|
||||
"Mi 10", "Mi 11", "Mi 11 Lite", "Redmi Note 10", "Redmi Note 11", "Redmi Note 12", "POCO F3", "POCO F4", "POCO X3 Pro",
|
||||
"ONEPLUS A6003", "ONEPLUS A6013", "ONEPLUS A5000", "ONEPLUS A5010", "OnePlus 8", "OnePlus 9", "OnePlus 10 Pro", "OnePlus Nord"])
|
||||
device_id = uuid.uuid4().hex[:16]
|
||||
app_version = random.choice(["100.85.2", "100.84.1", "100.83.1", "100.82.0", "100.81.1"])
|
||||
return f"LBC;{os};{os_version};{model};phone;{device_id};wifi;{app_version}"
|
||||
|
||||
def _init_session(self, proxy: Optional[Proxy] = None, impersonate: BrowserTypeLiteral = None, request_verify: bool = True) -> requests.Session:
|
||||
"""
|
||||
Initializes an HTTP session with optional proxy configuration and browser impersonation.
|
||||
|
||||
If no `impersonate` value is provided, a random browser type will be selected among common options.
|
||||
|
||||
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.
|
||||
"""
|
||||
if impersonate == None: # Pick a random browser client
|
||||
impersonate: BrowserTypeLiteral = random.choice(
|
||||
[
|
||||
"safari",
|
||||
"safari_ios",
|
||||
"chrome_android",
|
||||
"firefox"
|
||||
]
|
||||
)
|
||||
|
||||
session = requests.Session(
|
||||
impersonate=impersonate
|
||||
)
|
||||
|
||||
session.headers.update(
|
||||
{
|
||||
'User-Agent': self._generate_user_agent(),
|
||||
'Sec-Fetch-Dest': 'empty',
|
||||
'Sec-Fetch-Mode': 'cors',
|
||||
'Sec-Fetch-Site': 'same-site',
|
||||
}
|
||||
)
|
||||
if proxy:
|
||||
session.proxies = {
|
||||
"http": proxy.url,
|
||||
"https": proxy.url
|
||||
}
|
||||
|
||||
session.get("https://www.leboncoin.fr/", verify=request_verify) # Init cookies
|
||||
return session
|
||||
|
||||
@property
|
||||
def proxy(self) -> Proxy:
|
||||
return self._proxy
|
||||
|
||||
@proxy.setter
|
||||
def proxy(self, value: Proxy):
|
||||
if value:
|
||||
if isinstance(value, Proxy):
|
||||
self.session.proxies = {
|
||||
"http": value.url,
|
||||
"https": value.url
|
||||
}
|
||||
else:
|
||||
raise TypeError("Proxy must be an instance of the lbc.Proxy")
|
||||
else:
|
||||
self.session.proxies = {}
|
||||
self._proxy = value
|
||||
26
lbc/mixin/user.py
Normal file
26
lbc/mixin/user.py
Normal file
@@ -0,0 +1,26 @@
|
||||
from ..model import User
|
||||
from ..exceptions import NotFoundError
|
||||
|
||||
class UserMixin:
|
||||
def get_user(self, user_id: str) -> User:
|
||||
"""
|
||||
Retrieve information about a user based on their user ID.
|
||||
|
||||
This method fetches detailed user data such as their profile, professional status,
|
||||
and other relevant metadata available through the public user API.
|
||||
|
||||
Args:
|
||||
user_id (str): The unique identifier of the user on Leboncoin. Usually found in the url (e.g 57f99bb6-0446-4b82-b05d-a44ea7bcd2cc).
|
||||
|
||||
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")
|
||||
|
||||
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")
|
||||
except NotFoundError:
|
||||
pass # Some professional users may not have a Leboncoin page.
|
||||
return User._build(user_data=user_data, pro_data=pro_data)
|
||||
@@ -1,8 +1,8 @@
|
||||
from .user import User
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Any, Optional
|
||||
|
||||
from .user import User
|
||||
|
||||
@dataclass
|
||||
class Location:
|
||||
country_id: str
|
||||
@@ -7,10 +7,11 @@ class Proxy:
|
||||
port: Union[str, int]
|
||||
username: Optional[str] = None
|
||||
password: Optional[str] = None
|
||||
scheme: str = "http"
|
||||
|
||||
@property
|
||||
def url(self):
|
||||
if self.username and self.password:
|
||||
return f"http://{self.username}:{self.password}@{self.host}:{self.port}"
|
||||
return f"{self.scheme}://{self.username}:{self.password}@{self.host}:{self.port}"
|
||||
else:
|
||||
return f"http://{self.host}:{self.port}"
|
||||
return f"{self.scheme}://{self.host}:{self.port}"
|
||||
@@ -1,8 +1,8 @@
|
||||
from .ad import Ad
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Any
|
||||
|
||||
from .ad import Ad
|
||||
|
||||
@dataclass
|
||||
class Search:
|
||||
total: int
|
||||
@@ -1,8 +1,8 @@
|
||||
from .models import Category, AdType, OwnerType, Sort, Region, Department, City
|
||||
from .exceptions import InvalidValue
|
||||
|
||||
from typing import Optional, Union, List
|
||||
|
||||
from .model import Category, AdType, OwnerType, Sort, Region, Department, City
|
||||
from .exceptions import InvalidValue
|
||||
|
||||
def build_search_payload_with_url(
|
||||
url: str,
|
||||
limit: int = 35,
|
||||
@@ -251,7 +251,7 @@ def build_search_payload_with_args(
|
||||
payload["filters"]["keywords"]["type"] = "subject"
|
||||
|
||||
if shippable:
|
||||
payload["filters"]["locations"]["shippable"] = True
|
||||
payload["filters"]["location"]["shippable"] = True
|
||||
|
||||
if kwargs:
|
||||
for key, value in kwargs.items():
|
||||
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "lbc"
|
||||
version = "1.0.9"
|
||||
version = "1.1.2"
|
||||
description = "Unofficial client for Leboncoin API"
|
||||
readme = "README.md"
|
||||
license = {text = "MIT"}
|
||||
|
||||
@@ -1,171 +0,0 @@
|
||||
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 .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):
|
||||
"""
|
||||
Initializes a Leboncoin Client instance with optional proxy, browser impersonation, and SSL verification settings.
|
||||
|
||||
If no `impersonate` value is provided, a random browser type will be selected among common options.
|
||||
|
||||
Args:
|
||||
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
|
||||
|
||||
def _fetch(self, method: str, url: str, payload: Optional[dict] = None, timeout: int = 30, max_retries: int = 5) -> 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").
|
||||
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).
|
||||
RequestError: Raised for any other non-successful HTTP response.
|
||||
|
||||
Returns:
|
||||
dict: Parsed JSON response from the server.
|
||||
"""
|
||||
response = self.session.request(
|
||||
method=method,
|
||||
url=url,
|
||||
json=payload,
|
||||
timeout=timeout,
|
||||
verify=self.request_verify,
|
||||
)
|
||||
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}.")
|
||||
|
||||
def search(
|
||||
self,
|
||||
url: Optional[str] = None,
|
||||
text: Optional[str] = None,
|
||||
category: Category = Category.TOUTES_CATEGORIES,
|
||||
sort: Sort = Sort.RELEVANCE,
|
||||
locations: Optional[Union[List[Union[Region, Department, City]], Union[Region, Department, City]]] = None,
|
||||
limit: int = 35,
|
||||
limit_alu: int = 3,
|
||||
page: int = 1,
|
||||
ad_type: AdType = AdType.OFFER,
|
||||
owner_type: Optional[OwnerType] = None,
|
||||
shippable: Optional[bool] = None,
|
||||
search_in_title_only: bool = False,
|
||||
**kwargs
|
||||
) -> Search:
|
||||
"""
|
||||
Perform a classified ads search on Leboncoin with the specified criteria.
|
||||
|
||||
You can either:
|
||||
- Provide a full `url` from a Leboncoin search to replicate the search directly.
|
||||
- Or use the individual parameters (`text`, `category`, `locations`, etc.) to construct a custom search.
|
||||
|
||||
Args:
|
||||
url (Optional[str], optional): A full Leboncoin search URL. If provided, all other parameters will be ignored and the search will replicate the results from the URL.
|
||||
text (Optional[str], optional): Search keywords. If None, returns all matching ads without filtering by keyword. Defaults to None.
|
||||
category (Category, optional): Category to search in. Defaults to Category.TOUTES_CATEGORIES.
|
||||
sort (Sort, optional): Sorting method for results (e.g., relevance, date, price). Defaults to Sort.RELEVANCE.
|
||||
locations (Optional[Union[List[Union[Region, Department, City]], Union[Region, Department, City]]], optional): One or multiple locations (region, department, or city) to filter results. Defaults to None.
|
||||
limit (int, optional): Maximum number of results to return. Defaults to 35.
|
||||
limit_alu (int, optional): Number of ALU (Annonces Lu / similar ads) suggestions to include. Defaults to 3.
|
||||
page (int, optional): Page number to retrieve for paginated results. Defaults to 1.
|
||||
ad_type (AdType, optional): Type of ad (offer or request). Defaults to AdType.OFFER.
|
||||
owner_type (Optional[OwnerType], optional): Filter by seller type (individual, professional, or all). Defaults to None.
|
||||
shippable (Optional[bool], optional): If True, only includes ads that offer shipping. Defaults to None.
|
||||
search_in_title_only (bool, optional): If True, search will only be performed on ad titles. Defaults to False.
|
||||
**kwargs: Additional advanced filters such as price range (`price=(min, max)`), surface area (`square=(min, max)`), property type, and more.
|
||||
|
||||
Returns:
|
||||
Search: A `Search` object containing the parsed search results.
|
||||
"""
|
||||
if url:
|
||||
payload = build_search_payload_with_url(
|
||||
url=url, limit=limit, page=page
|
||||
)
|
||||
else:
|
||||
payload = build_search_payload_with_args(
|
||||
text=text, category=category, sort=sort, locations=locations,
|
||||
limit=limit, limit_alu=limit_alu, page=page, ad_type=ad_type,
|
||||
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)
|
||||
return Search._build(raw=body, client=self)
|
||||
|
||||
def get_user(
|
||||
self,
|
||||
user_id: str
|
||||
) -> User:
|
||||
"""
|
||||
Retrieve information about a user based on their user ID.
|
||||
|
||||
This method fetches detailed user data such as their profile, professional status,
|
||||
and other relevant metadata available through the public user API.
|
||||
|
||||
Args:
|
||||
user_id (str): The unique identifier of the user on Leboncoin. Usually found in the url (e.g 57f99bb6-0446-4b82-b05d-a44ea7bcd2cc).
|
||||
|
||||
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)
|
||||
|
||||
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:
|
||||
pass # Some professional users may not have a Leboncoin page.
|
||||
|
||||
return User._build(user_data=user_data, pro_data=pro_data)
|
||||
|
||||
def get_ad(
|
||||
self,
|
||||
ad_id: Union[str, int]
|
||||
) -> Ad:
|
||||
"""
|
||||
Retrieve detailed information about a classified ad using its ID.
|
||||
|
||||
This method fetches the full content of an ad, including its description,
|
||||
pricing, location, and other relevant metadata made
|
||||
available through the public Leboncoin ad API.
|
||||
|
||||
Args:
|
||||
ad_id (Union[str, int]): The unique identifier of the ad on Leboncoin. Can be found in the ad URL.
|
||||
|
||||
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)
|
||||
|
||||
return Ad._build(raw=body, client=self)
|
||||
@@ -1,84 +0,0 @@
|
||||
from .models import Proxy
|
||||
|
||||
from curl_cffi import requests, BrowserTypeLiteral
|
||||
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)
|
||||
self._proxy = proxy
|
||||
self._impersonate = impersonate
|
||||
|
||||
def _init_session(self, proxy: Optional[Proxy] = None, impersonate: BrowserTypeLiteral = None, request_verify: bool = True) -> requests.Session:
|
||||
"""
|
||||
Initializes an HTTP session with optional proxy configuration and browser impersonation.
|
||||
|
||||
If no `impersonate` value is provided, a random browser type will be selected among common options.
|
||||
|
||||
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.
|
||||
"""
|
||||
if impersonate == None: # Pick a random browser client
|
||||
impersonate: BrowserTypeLiteral = random.choice(
|
||||
[
|
||||
"chrome",
|
||||
"edge",
|
||||
"safari",
|
||||
"safari_ios",
|
||||
"chrome_android",
|
||||
"firefox"
|
||||
]
|
||||
)
|
||||
|
||||
session = requests.Session(
|
||||
impersonate=impersonate,
|
||||
)
|
||||
|
||||
session.headers.update(
|
||||
{
|
||||
'Sec-Fetch-Dest': 'empty',
|
||||
'Sec-Fetch-Mode': 'cors',
|
||||
'Sec-Fetch-Site': 'same-site',
|
||||
}
|
||||
)
|
||||
|
||||
if proxy:
|
||||
session.proxies = {
|
||||
"http": proxy.url,
|
||||
"https": proxy.url
|
||||
}
|
||||
|
||||
session.get("https://www.leboncoin.fr/", verify=request_verify) # Init cookies
|
||||
|
||||
return session
|
||||
|
||||
@property
|
||||
def session(self) -> requests.Session:
|
||||
return self._session
|
||||
|
||||
@session.setter
|
||||
def session(self, value: requests.Session):
|
||||
if isinstance(value, requests.Session):
|
||||
self._session = value
|
||||
else:
|
||||
raise TypeError("Session must be an instance of the curl_cffi.requests.Session")
|
||||
|
||||
@property
|
||||
def proxy(self) -> Proxy:
|
||||
return self._proxy
|
||||
|
||||
@proxy.setter
|
||||
def proxy(self, value: Proxy):
|
||||
if isinstance(value, Proxy):
|
||||
self._session.proxies = {
|
||||
"http": value.url,
|
||||
"https": value.url
|
||||
}
|
||||
else:
|
||||
raise TypeError("Proxy must be an instance of the Proxy class")
|
||||
@@ -11,7 +11,7 @@ def print_category(category_data: dict, category_name: Optional[str] = None) ->
|
||||
print(f'{f"{category_name}_" if category_name else ""}{label} = "{category_data['catId']}"')
|
||||
|
||||
def main() -> None:
|
||||
client = lbc.Client()
|
||||
client = lbc.Client(impersonate="chrome_android")
|
||||
body = client._fetch(method="GET", url="https://api.leboncoin.fr/api/frontend/v1/data/v7/fdata")
|
||||
|
||||
for category in body["categories"]:
|
||||
|
||||
@@ -9,7 +9,7 @@ def print_department(department_data: dict, region: dict) -> None:
|
||||
print(f'{name} = ("{region['rId']}", "{transform_str(region['rName'])}", "{department_data["dId"]}", "{name}")')
|
||||
|
||||
def main() -> None:
|
||||
client = lbc.Client()
|
||||
client = lbc.Client(impersonate="chrome_android")
|
||||
body = client._fetch(method="GET", url="https://api.leboncoin.fr/api/frontend/v1/data/v7/fdata")
|
||||
|
||||
for region in body["regions"]:
|
||||
|
||||
@@ -9,7 +9,7 @@ def print_region(region_data: dict) -> None:
|
||||
print(f'{name} = ("{region_data['rId']}", "{name}")')
|
||||
|
||||
def main() -> None:
|
||||
client = lbc.Client()
|
||||
client = lbc.Client(impersonate="chrome_android")
|
||||
body = client._fetch(method="GET", url="https://api.leboncoin.fr/api/frontend/v1/data/v7/fdata")
|
||||
|
||||
for region in body["regions"]:
|
||||
|
||||
Reference in New Issue
Block a user