mirror of
https://github.com/etienne-hd/lbc.git
synced 2026-04-22 15:02:41 +02:00
Compare commits
27 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a44472e757 | ||
|
|
6669950f82 | ||
|
|
bbbdd1652f | ||
|
|
c891e593c5 | ||
|
|
e8cb0a9f5d | ||
|
|
48b83c0c9d | ||
|
|
671fb291f1 | ||
|
|
81d1799ea8 | ||
|
|
a927bd7cf5 | ||
|
|
8d4d2da64f | ||
|
|
4e69194821 | ||
|
|
00cf534191 | ||
|
|
8a74cde3b1 | ||
|
|
d24b1cc0e6 | ||
|
|
6ef01383f0 | ||
|
|
b9ac610b04 | ||
|
|
672204dd95 | ||
|
|
4fa9409f78 | ||
|
|
8eb8a96e8f | ||
|
|
07d200b653 | ||
|
|
feebd85591 | ||
|
|
76dda0cb76 | ||
|
|
2888942ecc | ||
|
|
fb31b43fd6 | ||
|
|
887ad99959 | ||
|
|
f5b08f0890 | ||
|
|
5841f55b20 |
2
.gitignore
vendored
2
.gitignore
vendored
@@ -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/
|
||||
|
||||
64
CHANGELOG.md
64
CHANGELOG.md
@@ -1,3 +1,65 @@
|
||||
## 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.
|
||||
|
||||
## 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.
|
||||
* Ability to choose which browser to impersonate via the `impersonate` argument in `Client`.
|
||||
* Option to disable SSL verification for requests by setting `request_verify` to `False` in `Client`.
|
||||
|
||||
## 1.0.6
|
||||
### Fixed
|
||||
* "Unknown location type" error when searching with a URL containing a zipcode.
|
||||
|
||||
## 1.0.5
|
||||
### Fixed
|
||||
* 404 error when fetching a pro user who doesn't have a public page
|
||||
|
||||
## 1.0.4
|
||||
### Added
|
||||
* A lot of new user information can be retrieved (feedback, badges & professional info).
|
||||
* New [examples](examples/) directory with practical usage cases.
|
||||
* `get_ad` function to retrieve ad information.
|
||||
* `get_user` function to retrieve user information (with pro info such as siret).
|
||||
* Automatic cookies initialization during session setup to prevent HTTP 403 errors.
|
||||
|
||||
### Changed
|
||||
* Codebase refactored: models are now split by functionality (e.g., all user dataclasses are in `user.py`).
|
||||
* Proxies can now be updated after `Client` creation via `client.proxy = ...`.
|
||||
* The session can also be changed dynamically via `client.session = ...`.
|
||||
|
||||
### Removed
|
||||
* Removed the `test/` folder (migrated to `examples/`).
|
||||
|
||||
## 1.0.3
|
||||
### Fixed
|
||||
* Incorrect raw data extraction for location and owner in `Search.build` function
|
||||
@@ -15,4 +77,4 @@
|
||||
* Realistic `Sec-Fetch-*` headers to prevent 403 errors
|
||||
|
||||
## 1.0.0
|
||||
* Initial release
|
||||
* Initial release
|
||||
|
||||
34
README.md
34
README.md
@@ -1,5 +1,6 @@
|
||||
# lbc
|
||||
[](https://pypi.org/project/lbc)
|
||||

|
||||
[](https://github.com/etienne-hd/lbc/blob/master/LICENSE)
|
||||
|
||||
**Unofficial client for Leboncoin API**
|
||||
@@ -34,12 +35,15 @@ for ad in result.ads:
|
||||
*lbc is not affiliated with, endorsed by, or in any way associated with Leboncoin or its services. Use at your own risk.*
|
||||
|
||||
## Installation
|
||||
Required Python 3.9+
|
||||
Required **Python 3.9+**
|
||||
```bash
|
||||
pip install lbc
|
||||
```
|
||||
|
||||
## Usage
|
||||
|
||||
Start with the [examples](examples/) to quickly understand how to use the library in real-world scenarios.
|
||||
|
||||
### Client
|
||||
To create client you need to use `lbc.Client` class
|
||||
```python
|
||||
@@ -51,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
|
||||
```
|
||||
|
||||
|
||||
|
||||
22
examples/get_ad.py
Normal file
22
examples/get_ad.py
Normal file
@@ -0,0 +1,22 @@
|
||||
"""Get detailed information about an ad on Leboncoin using its ID."""
|
||||
|
||||
import lbc
|
||||
|
||||
def main() -> None:
|
||||
# Initialize the Leboncoin API client
|
||||
client = lbc.Client()
|
||||
|
||||
# Fetch an ad by its Leboncoin ID (replace with a real one for testing)
|
||||
ad = client.get_ad("0123456789")
|
||||
|
||||
# Print basic information about the ad
|
||||
print("Title:", ad.subject)
|
||||
print("Price:", ad.price)
|
||||
print("Favorites:", ad.favorites)
|
||||
print("First published on:", ad.first_publication_date)
|
||||
|
||||
# Print information about the user who posted the ad
|
||||
print("User info:", ad.user)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
20
examples/get_user.py
Normal file
20
examples/get_user.py
Normal file
@@ -0,0 +1,20 @@
|
||||
"""Get detailed information about a Leboncoin user using their user ID."""
|
||||
|
||||
import lbc
|
||||
|
||||
def main() -> None:
|
||||
# Initialize the Leboncoin API client
|
||||
client = lbc.Client()
|
||||
|
||||
# Fetch a user by their Leboncoin user ID
|
||||
# Replace the ID with a real one for testing
|
||||
user = client.get_user("01234567-89ab-cdef-0123-456789abcdef")
|
||||
|
||||
# Print raw user attributes
|
||||
print("User ID:", user.id)
|
||||
print("Name:", user.name)
|
||||
print("Pro status:", user.is_pro)
|
||||
print("Ads count:", user.total_ads)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
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()
|
||||
38
examples/search_with_args.py
Normal file
38
examples/search_with_args.py
Normal file
@@ -0,0 +1,38 @@
|
||||
"""Search for ads on Leboncoin by location and filters (example: real estate in Paris)."""
|
||||
|
||||
import lbc
|
||||
|
||||
def main() -> None:
|
||||
# Initialize the Leboncoin API client
|
||||
client = lbc.Client()
|
||||
|
||||
# Define the search location: Paris with a 10 km radius
|
||||
location = lbc.City(
|
||||
lat=48.85994982004764,
|
||||
lng=2.33801967847424,
|
||||
radius=10_000, # 10 km
|
||||
city="Paris"
|
||||
)
|
||||
|
||||
# Perform a search with various filters
|
||||
result = client.search(
|
||||
text="maison", # Search for houses
|
||||
locations=[location], # Only in Paris
|
||||
page=1,
|
||||
limit=35, # Max results per page
|
||||
limit_alu=0, # No auto-suggestions
|
||||
sort=lbc.Sort.NEWEST, # Sort by newest ads
|
||||
ad_type=lbc.AdType.OFFER, # Only offers, not searches
|
||||
category=lbc.Category.IMMOBILIER, # Real estate category
|
||||
owner_type=lbc.OwnerType.ALL, # All types of sellers
|
||||
search_in_title_only=True, # Only search in titles
|
||||
square=(200, 400), # Surface between 200 and 400 m²
|
||||
price=[300_000, 700_000] # Price range in euros
|
||||
)
|
||||
|
||||
# Display summary of each ad
|
||||
for ad in result.ads:
|
||||
print(f"{ad.id} | {ad.url} | {ad.subject} | {ad.price}€ | Seller: {ad.user}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
43
examples/search_with_args_pro.py
Normal file
43
examples/search_with_args_pro.py
Normal file
@@ -0,0 +1,43 @@
|
||||
"""Search for professional ads on Leboncoin (example: real estate in Paris)."""
|
||||
|
||||
import lbc
|
||||
|
||||
def main() -> None:
|
||||
# Initialize the Leboncoin API client
|
||||
client = lbc.Client()
|
||||
|
||||
# Define search location: Paris with a 10 km radius
|
||||
location = lbc.City(
|
||||
lat=48.85994982004764,
|
||||
lng=2.33801967847424,
|
||||
radius=10_000, # 10 km
|
||||
city="Paris"
|
||||
)
|
||||
|
||||
# Perform a search with specific filters
|
||||
result = client.search(
|
||||
text="maison", # Search for houses
|
||||
locations=[location], # Only in Paris
|
||||
page=1,
|
||||
limit=35, # Max results per page
|
||||
limit_alu=0, # No auto-suggestions
|
||||
sort=lbc.Sort.NEWEST, # Sort by newest ads
|
||||
ad_type=lbc.AdType.OFFER, # Only offers, not searches
|
||||
category=lbc.Category.IMMOBILIER, # Real estate category
|
||||
owner_type=lbc.OwnerType.ALL, # All types of sellers
|
||||
search_in_title_only=True, # Only search in titles
|
||||
square=(200, 400), # Surface between 200 and 400 m²
|
||||
price=[300_000, 700_000] # Price range in euros
|
||||
)
|
||||
|
||||
# Display only professional ads with their legal/professional data
|
||||
for ad in result.ads:
|
||||
if ad.user.pro:
|
||||
print(
|
||||
f"Store: {ad.user.pro.online_store_name} | "
|
||||
f"SIRET: {ad.user.pro.siret} | "
|
||||
f"Website: {ad.user.pro.website_url or 'N/A'}"
|
||||
)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
21
examples/search_with_url.py
Normal file
21
examples/search_with_url.py
Normal file
@@ -0,0 +1,21 @@
|
||||
"""Search for ads on Leboncoin using a full search URL."""
|
||||
|
||||
import lbc
|
||||
|
||||
def main() -> None:
|
||||
# Initialize the Leboncoin API client
|
||||
client = lbc.Client()
|
||||
|
||||
# Perform a search using a prebuilt Leboncoin URL
|
||||
result = client.search(
|
||||
url="https://www.leboncoin.fr/recherche?category=10&text=maison&locations=Paris__48.86023250788424_2.339006433295173_9256_30000",
|
||||
page=1,
|
||||
limit=35
|
||||
)
|
||||
|
||||
# Print basic info about each ad
|
||||
for ad in result.ads:
|
||||
print(f"{ad.id} | {ad.url} | {ad.subject} | {ad.price}€ | Seller: {ad.user}")
|
||||
|
||||
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,14 +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)
|
||||
@@ -1,47 +1,9 @@
|
||||
from .session import Session
|
||||
from .models import Proxy, Search, Category, AdType, OwnerType, Sort, Region, Department, City
|
||||
from .exceptions import DatadomeError, RequestError
|
||||
from .utils import build_search_payload_with_args, build_search_payload_with_url
|
||||
from typing import Optional, Union, List
|
||||
|
||||
from typing import Optional, List, Union
|
||||
|
||||
class Client(Session):
|
||||
def __init__(self, proxy: Optional[Proxy] = None):
|
||||
super().__init__(proxy=proxy)
|
||||
|
||||
def _fetch(self, method: str, url: str, payload: Optional[dict] = None, timeout: int = 30) -> dict:
|
||||
"""
|
||||
Internal method to send an HTTP request using the configured session.
|
||||
|
||||
Args:
|
||||
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.
|
||||
|
||||
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
|
||||
)
|
||||
if response.ok:
|
||||
return response.json()
|
||||
elif response.status_code == 403:
|
||||
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.")
|
||||
else:
|
||||
raise RequestError(f"Request failed with status code {response.status_code}.")
|
||||
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,
|
||||
@@ -95,4 +57,4 @@ class Client(Session):
|
||||
)
|
||||
|
||||
body = self._fetch(method="POST", url="https://api.leboncoin.fr/finder/search", payload=payload)
|
||||
return Search.build(raw=body)
|
||||
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,4 +1,6 @@
|
||||
from .proxy import Proxy
|
||||
from .search import Search
|
||||
from .ad import Ad
|
||||
from .user import User
|
||||
from .enums import *
|
||||
from .city import City
|
||||
125
lbc/model/ad.py
Normal file
125
lbc/model/ad.py
Normal file
@@ -0,0 +1,125 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Any, Optional
|
||||
|
||||
from .user import User
|
||||
|
||||
@dataclass
|
||||
class Location:
|
||||
country_id: str
|
||||
region_id: str
|
||||
region_name: str
|
||||
department_id: str
|
||||
department_name: str
|
||||
city_label: str
|
||||
city: str
|
||||
zipcode: str
|
||||
lat: float
|
||||
lng: float
|
||||
source: str
|
||||
provider: str
|
||||
is_shape: bool
|
||||
|
||||
@dataclass
|
||||
class Attribute:
|
||||
key: str
|
||||
key_label: Optional[str]
|
||||
value: str
|
||||
value_label: str
|
||||
values: List[str]
|
||||
values_label: Optional[List[str]]
|
||||
value_label_reader: Optional[str]
|
||||
generic: bool
|
||||
|
||||
@dataclass
|
||||
class Ad:
|
||||
id: int
|
||||
first_publication_date: str
|
||||
expiration_date: str
|
||||
index_date: str
|
||||
status: str
|
||||
category_id: str
|
||||
category_name: str
|
||||
subject: str
|
||||
body: str
|
||||
brand: str
|
||||
ad_type: str
|
||||
url: str
|
||||
price: float
|
||||
images: List[str]
|
||||
attributes: List[Attribute]
|
||||
location: Location
|
||||
has_phone: bool
|
||||
favorites: int # Unvailaible on Ad from Search
|
||||
|
||||
_client: Any
|
||||
_user_id: str
|
||||
_user: User
|
||||
|
||||
@staticmethod
|
||||
def _build(raw: dict, client: Any) -> "Ad":
|
||||
attributes: List[Attribute] = []
|
||||
for raw_attribute in raw.get("attributes", []):
|
||||
attributes.append(
|
||||
Attribute(
|
||||
key=raw_attribute.get("key"),
|
||||
key_label=raw_attribute.get("key_label"),
|
||||
value=raw_attribute.get("value"),
|
||||
value_label=raw_attribute.get("value_label"),
|
||||
values=raw_attribute.get("values"),
|
||||
values_label=raw_attribute.get("values_label"),
|
||||
value_label_reader=raw_attribute.get("value_label_reader"),
|
||||
generic=raw_attribute.get("generic")
|
||||
)
|
||||
)
|
||||
|
||||
raw_location: dict = raw.get("location", {})
|
||||
location = Location(
|
||||
country_id=raw_location.get("country_id"),
|
||||
region_id=raw_location.get("region_id"),
|
||||
region_name=raw_location.get("region_name"),
|
||||
department_id=raw_location.get("department_id"),
|
||||
department_name=raw_location.get("department_name"),
|
||||
city_label=raw_location.get("city_label"),
|
||||
city=raw_location.get("city"),
|
||||
zipcode=raw_location.get("zipcode"),
|
||||
lat=raw_location.get("lat"),
|
||||
lng=raw_location.get("lng"),
|
||||
source=raw_location.get("source"),
|
||||
provider=raw_location.get("provider"),
|
||||
is_shape=raw_location.get("is_shape")
|
||||
)
|
||||
|
||||
raw_owner: dict = raw.get("owner", {})
|
||||
return Ad(
|
||||
id=raw.get("list_id"),
|
||||
first_publication_date=raw.get("first_publication_date"),
|
||||
expiration_date=raw.get("expiration_date"),
|
||||
index_date=raw.get("index_date"),
|
||||
status=raw.get("status"),
|
||||
category_id=raw.get("category_id"),
|
||||
category_name=raw.get("category_name"),
|
||||
subject=raw.get("subject"),
|
||||
body=raw.get("body"),
|
||||
brand=raw.get("brand"),
|
||||
ad_type=raw.get("ad_type"),
|
||||
url=raw.get("url"),
|
||||
price=raw.get("price_cents") / 100 if raw.get("price_cents") else None,
|
||||
images=raw.get("images", {}).get("urls_large"),
|
||||
attributes=attributes,
|
||||
location=location,
|
||||
has_phone=raw.get("has_phone"),
|
||||
favorites=raw.get("counters", {}).get("favorites"),
|
||||
_client=client,
|
||||
_user_id=raw_owner.get("user_id"),
|
||||
_user=None
|
||||
)
|
||||
|
||||
@property
|
||||
def title(self) -> str:
|
||||
return self.subject
|
||||
|
||||
@property
|
||||
def user(self) -> User:
|
||||
if self._user is None:
|
||||
self._user = self._client.get_user(user_id=self._user_id)
|
||||
return self._user
|
||||
@@ -1,5 +1,4 @@
|
||||
from enum import Enum
|
||||
from typing import Union, Tuple
|
||||
|
||||
class OwnerType(Enum):
|
||||
PRO = "pro"
|
||||
@@ -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}"
|
||||
35
lbc/model/search.py
Normal file
35
lbc/model/search.py
Normal file
@@ -0,0 +1,35 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Any
|
||||
|
||||
from .ad import Ad
|
||||
|
||||
@dataclass
|
||||
class Search:
|
||||
total: int
|
||||
total_all: int
|
||||
total_pro: int
|
||||
total_private: int
|
||||
total_active: int
|
||||
total_inactive: int
|
||||
total_shippable: int
|
||||
max_pages: int
|
||||
ads: List[Ad]
|
||||
|
||||
@staticmethod
|
||||
def _build(raw: dict, client: Any) -> "Search":
|
||||
ads: List[Ad] = [
|
||||
Ad._build(raw=ad, client=client)
|
||||
for ad in raw.get("ads", [])
|
||||
]
|
||||
|
||||
return Search(
|
||||
total=raw.get("total"),
|
||||
total_all=raw.get("total_all"),
|
||||
total_pro=raw.get("total_pro"),
|
||||
total_private=raw.get("total_private"),
|
||||
total_active=raw.get("total_active"),
|
||||
total_inactive=raw.get("total_inactive"),
|
||||
total_shippable=raw.get("total_shippable"),
|
||||
max_pages=raw.get("max_pages"),
|
||||
ads=ads,
|
||||
)
|
||||
235
lbc/model/user.py
Normal file
235
lbc/model/user.py
Normal file
@@ -0,0 +1,235 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional
|
||||
|
||||
@dataclass
|
||||
class Reply:
|
||||
in_minutes: int
|
||||
text: str
|
||||
rate_text: str
|
||||
rate: int
|
||||
reply_time_text: str
|
||||
|
||||
@dataclass
|
||||
class Presence:
|
||||
status: str
|
||||
presence_text: str
|
||||
last_activity: str
|
||||
enabled: bool
|
||||
|
||||
@dataclass
|
||||
class Badge:
|
||||
type: str
|
||||
name: str
|
||||
|
||||
@dataclass
|
||||
class Feedback:
|
||||
overall_score: float
|
||||
cleanness: float
|
||||
communication: float
|
||||
conformity: float
|
||||
package: float
|
||||
product: float
|
||||
recommendation: float
|
||||
respect: float
|
||||
transaction: float
|
||||
user_attention: float
|
||||
received_count: float
|
||||
|
||||
@property
|
||||
def score(self) -> float:
|
||||
return self.overall_score * 5 if self.overall_score else None
|
||||
|
||||
@dataclass
|
||||
class Location:
|
||||
address: str
|
||||
district: str
|
||||
city: str
|
||||
label: str
|
||||
lat: float
|
||||
lng: float
|
||||
zipcode: str
|
||||
geo_source: str
|
||||
geo_provider: str
|
||||
region: str
|
||||
region_label: str
|
||||
department: str
|
||||
department_label: str
|
||||
country: str
|
||||
|
||||
@dataclass
|
||||
class Review:
|
||||
author_name: str
|
||||
rating_value: int
|
||||
text: str
|
||||
review_time: str
|
||||
|
||||
@dataclass
|
||||
class Rating:
|
||||
rating_value: int
|
||||
user_ratings_total: int
|
||||
source: str
|
||||
source_display: str
|
||||
retrieval_time: str
|
||||
url: str
|
||||
reviews: List[Review]
|
||||
|
||||
@dataclass
|
||||
class Pro:
|
||||
online_store_id: int
|
||||
online_store_name: str
|
||||
activity_sector_id: int
|
||||
activity_sector: str
|
||||
category_id: int
|
||||
siren: str
|
||||
siret: str
|
||||
store_id: int
|
||||
active_since: str
|
||||
location: Location
|
||||
logo: str
|
||||
cover: str
|
||||
slogan: str
|
||||
description: str
|
||||
opening_hours: str
|
||||
website_url: str
|
||||
rating: Rating
|
||||
|
||||
@dataclass
|
||||
class User:
|
||||
id: str
|
||||
name: str
|
||||
registered_at: str
|
||||
location: str
|
||||
feedback: Feedback
|
||||
profile_picture: str
|
||||
reply: Reply
|
||||
presence: Presence
|
||||
badges: List[Badge]
|
||||
total_ads: int
|
||||
store_id: int
|
||||
account_type: str
|
||||
description: str
|
||||
pro: Optional[Pro]
|
||||
|
||||
@staticmethod
|
||||
def _build(user_data: dict, pro_data: Optional[dict]) -> "User":
|
||||
raw_feedback = user_data.get("feedback", {})
|
||||
feedback = Feedback(
|
||||
overall_score=raw_feedback.get("overall_score"),
|
||||
cleanness=raw_feedback.get("category_scores", {}).get("CLEANNESS"),
|
||||
communication=raw_feedback.get("category_scores", {}).get("COMMUNICATION"),
|
||||
conformity=raw_feedback.get("category_scores", {}).get("CONFORMITY"),
|
||||
package=raw_feedback.get("category_scores", {}).get("PACKAGE"),
|
||||
product=raw_feedback.get("category_scores", {}).get("PRODUCT"),
|
||||
recommendation=raw_feedback.get("category_scores", {}).get("RECOMMENDATION"),
|
||||
respect=raw_feedback.get("category_scores", {}).get("RESPECT"),
|
||||
transaction=raw_feedback.get("category_scores", {}).get("TRANSACTION"),
|
||||
user_attention=raw_feedback.get("category_scores", {}).get("USER_ATTENTION"),
|
||||
received_count=raw_feedback.get("received_count")
|
||||
)
|
||||
|
||||
raw_reply = user_data.get("reply", {})
|
||||
reply = Reply(
|
||||
in_minutes=raw_reply.get("in_minutes"),
|
||||
text=raw_reply.get("text"),
|
||||
rate_text=raw_reply.get("rate_text"),
|
||||
rate=raw_reply.get("rate"),
|
||||
reply_time_text=raw_reply.get("reply_time_text")
|
||||
)
|
||||
|
||||
raw_presence = user_data.get("presence", {})
|
||||
presence = Presence(
|
||||
status=raw_presence.get("status"),
|
||||
presence_text=raw_presence.get("presence_text"),
|
||||
last_activity=raw_presence.get("last_activity"),
|
||||
enabled=raw_presence.get("enabled")
|
||||
)
|
||||
|
||||
badges = [
|
||||
Badge(type=badge.get("type"), name=badge.get("name"))
|
||||
for badge in user_data.get("badges", [])
|
||||
]
|
||||
|
||||
pro = None
|
||||
if pro_data:
|
||||
raw_pro_location = pro_data.get("location", {})
|
||||
pro_location = Location(
|
||||
address=raw_pro_location.get("address"),
|
||||
district=raw_pro_location.get("district"),
|
||||
city=raw_pro_location.get("city"),
|
||||
label=raw_pro_location.get("label"),
|
||||
lat=raw_pro_location.get("lat"),
|
||||
lng=raw_pro_location.get("lng"),
|
||||
zipcode=raw_pro_location.get("zipcode"),
|
||||
geo_source=raw_pro_location.get("geo_source"),
|
||||
geo_provider=raw_pro_location.get("geo_provider"),
|
||||
region=raw_pro_location.get("region"),
|
||||
region_label=raw_pro_location.get("region_label"),
|
||||
department=raw_pro_location.get("department"),
|
||||
department_label=raw_pro_location.get("dpt_label"),
|
||||
country=raw_pro_location.get("country")
|
||||
)
|
||||
|
||||
raw_pro_rating = pro_data.get("rating", {})
|
||||
pro_rating_reviews = [
|
||||
Review(
|
||||
author_name=review.get("author_name"),
|
||||
rating_value=review.get("rating_value"),
|
||||
text=review.get("text"),
|
||||
review_time=review.get("review_time")
|
||||
)
|
||||
for review in raw_pro_rating.get("reviews", [])
|
||||
]
|
||||
|
||||
pro_rating = Rating(
|
||||
rating_value=raw_pro_rating.get("rating_value"),
|
||||
user_ratings_total=raw_pro_rating.get("user_ratings_total"),
|
||||
source=raw_pro_rating.get("source"),
|
||||
source_display=raw_pro_rating.get("source_display"),
|
||||
retrieval_time=raw_pro_rating.get("retrieval_time"),
|
||||
url=raw_pro_rating.get("url"),
|
||||
reviews=pro_rating_reviews
|
||||
)
|
||||
|
||||
pro_owner = pro_data.get("owner", {})
|
||||
pro_brand = pro_data.get("brand", {})
|
||||
pro_information = pro_data.get("information", {})
|
||||
pro = Pro(
|
||||
online_store_id=pro_data.get("online_store_id"),
|
||||
online_store_name=pro_data.get("online_store_name"),
|
||||
activity_sector_id=pro_owner.get("activitySectorID"),
|
||||
activity_sector=pro_owner.get("activitySector"),
|
||||
category_id=pro_owner.get("categoryId"),
|
||||
siren=pro_owner.get("siren"),
|
||||
siret=pro_owner.get("siret"),
|
||||
store_id=pro_owner.get("storeId"),
|
||||
active_since=pro_owner.get("activeSince"),
|
||||
location=pro_location,
|
||||
logo=pro_brand.get("logo", {}).get("large"),
|
||||
cover=pro_brand.get("cover", {}).get("large"),
|
||||
slogan=pro_brand.get("slogan"),
|
||||
description=pro_information.get("description"),
|
||||
opening_hours=pro_information.get("opening_hours"),
|
||||
website_url=pro_information.get("website_url"),
|
||||
rating=pro_rating
|
||||
)
|
||||
|
||||
return User(
|
||||
id=user_data.get("user_id"),
|
||||
name=user_data.get("name"),
|
||||
registered_at=user_data.get("registered_at"),
|
||||
location=user_data.get("location"),
|
||||
feedback=feedback,
|
||||
profile_picture=user_data.get("profile_picture", {}).get("extra_large_url"),
|
||||
reply=reply,
|
||||
presence=presence,
|
||||
badges=badges,
|
||||
total_ads=user_data.get("total_ads"),
|
||||
store_id=user_data.get("store_id"),
|
||||
account_type=user_data.get("account_type"),
|
||||
description=user_data.get("description"),
|
||||
pro=pro
|
||||
)
|
||||
|
||||
@property
|
||||
def is_pro(self):
|
||||
return self.account_type == "pro"
|
||||
@@ -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,
|
||||
@@ -55,7 +55,7 @@ def build_search_payload_with_url(
|
||||
location_parts = location.split("__") # City ['Paris', '48.86023250788424_2.339006433295173_9256'], Department ['d_69'], Region ['r_18'] or Place ['p_give a star if you like it!', '0.1234567891234_-0.1234567891234567_5000_5500']
|
||||
|
||||
prefix_parts = location_parts[0].split("_")
|
||||
if len(prefix_parts) == 2: # Department ['d', '1'], Region ['r', '1'], or Place ['p', 'give a star if you like it!']
|
||||
if len(prefix_parts[0]) == 1: # Department ['d', '1'], Region ['r', '1'], or Place ['p', 'give a star if you like it!']
|
||||
location_id = prefix_parts[1] # Department '1', Region '1' or Place 'give a star if you like it!'
|
||||
match prefix_parts[0]:
|
||||
case "d": # Department
|
||||
@@ -90,7 +90,7 @@ def build_search_payload_with_url(
|
||||
payload["filters"]["location"]["locations"].append(
|
||||
{
|
||||
"locationType": "city",
|
||||
"city": location_parts[0],
|
||||
#"city": location_parts[0],
|
||||
"area": build_area(area_values)
|
||||
}
|
||||
)
|
||||
@@ -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.3"
|
||||
version = "1.1.2"
|
||||
description = "Unofficial client for Leboncoin API"
|
||||
readme = "README.md"
|
||||
license = {text = "MIT"}
|
||||
|
||||
@@ -1,32 +0,0 @@
|
||||
from .attribute import Attribute
|
||||
from .location import Location
|
||||
from .owner import Owner
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from typing import List
|
||||
|
||||
@dataclass
|
||||
class Ad:
|
||||
id: int
|
||||
first_publication_date: datetime
|
||||
expiration_date: datetime
|
||||
index_date: datetime
|
||||
status: str
|
||||
category_id: str
|
||||
category_name: str
|
||||
subject: str
|
||||
body: str
|
||||
brand: str
|
||||
ad_type: str
|
||||
url: str
|
||||
price: float
|
||||
images: List[str]
|
||||
attributes: List[Attribute]
|
||||
location: Location
|
||||
owner: Owner
|
||||
has_phone: bool
|
||||
|
||||
@property
|
||||
def title(self) -> str:
|
||||
return self.subject
|
||||
@@ -1,13 +0,0 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional, List
|
||||
|
||||
@dataclass
|
||||
class Attribute:
|
||||
key: str
|
||||
key_label: Optional[str]
|
||||
value: str
|
||||
value_label: str
|
||||
values: List[str]
|
||||
values_label: Optional[List[str]]
|
||||
value_label_reader: Optional[str]
|
||||
generic: bool
|
||||
@@ -1,17 +0,0 @@
|
||||
from dataclasses import dataclass
|
||||
|
||||
@dataclass
|
||||
class Location:
|
||||
country_id: str
|
||||
region_id: str
|
||||
region_name: str
|
||||
department_id: str
|
||||
department_name: str
|
||||
city_label: str
|
||||
city: str
|
||||
zipcode: str
|
||||
lat: float
|
||||
lng: float
|
||||
source: str
|
||||
provider: str
|
||||
is_shape: bool
|
||||
@@ -1,9 +0,0 @@
|
||||
from dataclasses import dataclass
|
||||
|
||||
@dataclass
|
||||
class Owner:
|
||||
store_id: str
|
||||
user_id: str
|
||||
type: str
|
||||
name: str
|
||||
no_salesmen: bool
|
||||
@@ -1,101 +0,0 @@
|
||||
from .ad import Ad
|
||||
from .attribute import Attribute
|
||||
from .location import Location
|
||||
from .owner import Owner
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import List
|
||||
from datetime import datetime
|
||||
|
||||
@dataclass
|
||||
class Search:
|
||||
total: int
|
||||
total_all: int
|
||||
total_pro: int
|
||||
total_private: int
|
||||
total_active: int
|
||||
total_inactive: int
|
||||
total_shippable: int
|
||||
max_pages: int
|
||||
ads: List[Ad]
|
||||
|
||||
@staticmethod
|
||||
def build(raw: dict) -> "Search":
|
||||
ads: List[Ad] = []
|
||||
|
||||
for raw_ad in raw.get("ads", []):
|
||||
attributes: List[Attribute] = []
|
||||
for raw_attribute in raw_ad.get("attributes", []):
|
||||
attributes.append(
|
||||
Attribute(
|
||||
key=raw_attribute.get("key"),
|
||||
key_label=raw_attribute.get("key_label"),
|
||||
value=raw_attribute.get("value"),
|
||||
value_label=raw_attribute.get("value_label"),
|
||||
values=raw_attribute.get("values"),
|
||||
values_label=raw_attribute.get("values_label"),
|
||||
value_label_reader=raw_attribute.get("value_label_reader"),
|
||||
generic=raw_attribute.get("generic")
|
||||
)
|
||||
)
|
||||
|
||||
raw_location: dict = raw_ad.get("location", {})
|
||||
location = Location(
|
||||
country_id=raw_location.get("country_id"),
|
||||
region_id=raw_location.get("region_id"),
|
||||
region_name=raw_location.get("region_name"),
|
||||
department_id=raw_location.get("department_id"),
|
||||
department_name=raw_location.get("department_name"),
|
||||
city_label=raw_location.get("city_label"),
|
||||
city=raw_location.get("city"),
|
||||
zipcode=raw_location.get("zipcode"),
|
||||
lat=raw_location.get("lat"),
|
||||
lng=raw_location.get("lng"),
|
||||
source=raw_location.get("source"),
|
||||
provider=raw_location.get("provider"),
|
||||
is_shape=raw_location.get("is_shape")
|
||||
)
|
||||
|
||||
raw_owner: dict = raw_ad.get("owner", {})
|
||||
owner = Owner(
|
||||
store_id=raw_owner.get("store_id"),
|
||||
user_id=raw_owner.get("user_id"),
|
||||
type=raw_owner.get("type"),
|
||||
name=raw_owner.get("name"),
|
||||
no_salesmen=raw_owner.get("no_salesmen")
|
||||
)
|
||||
|
||||
ads.append(
|
||||
Ad(
|
||||
id=raw_ad.get("list_id"),
|
||||
first_publication_date=datetime.strptime(raw_ad.get("first_publication_date"), "%Y-%m-%d %H:%M:%S") if raw_ad.get("first_publication_date") else None,
|
||||
expiration_date=datetime.strptime(raw_ad.get("expiration_date"), "%Y-%m-%d %H:%M:%S") if raw_ad.get("expiration_date") else None,
|
||||
index_date=datetime.strptime(raw_ad.get("index_date"), "%Y-%m-%d %H:%M:%S") if raw_ad.get("index_date") else None,
|
||||
status=raw_ad.get("status"),
|
||||
category_id=raw_ad.get("category_id"),
|
||||
category_name=raw_ad.get("category_name"),
|
||||
subject=raw_ad.get("subject"),
|
||||
body=raw_ad.get("body"),
|
||||
brand=raw_ad.get("brand"),
|
||||
ad_type=raw_ad.get("ad_type"),
|
||||
url=raw_ad.get("url"),
|
||||
price=raw_ad.get("price_cents") / 100 if raw_ad.get("price_cents") else None,
|
||||
images=raw_ad.get("images", {}).get("urls_large"),
|
||||
attributes=attributes,
|
||||
location=location,
|
||||
owner=owner,
|
||||
has_phone=raw_ad.get("has_phone")
|
||||
)
|
||||
)
|
||||
|
||||
return Search(
|
||||
total=raw.get("total"),
|
||||
total_all=raw.get("total_all"),
|
||||
total_pro=raw.get("total_pro"),
|
||||
total_private=raw.get("total_private"),
|
||||
total_active=raw.get("total_active"),
|
||||
total_inactive=raw.get("total_inactive"),
|
||||
total_shippable=raw.get("total_shippable"),
|
||||
max_pages=raw.get("max_pages"),
|
||||
ads=ads
|
||||
)
|
||||
@@ -1,47 +0,0 @@
|
||||
from .models import Proxy
|
||||
|
||||
from curl_cffi import requests
|
||||
from typing import Optional
|
||||
|
||||
class Session:
|
||||
def __init__(self, proxy: Optional[Proxy] = None):
|
||||
self._session = self._init_session(proxy=proxy)
|
||||
self._proxy = proxy
|
||||
|
||||
def _init_session(self, proxy: Optional[Proxy] = None) -> requests.Session:
|
||||
"""
|
||||
Initializes an HTTP session with optional proxy and browser impersonation.
|
||||
|
||||
Args:
|
||||
proxy (Optional[Proxy], optional): Proxy configuration to use for the session. If provided, it will be applied to both HTTP and HTTPS traffic.
|
||||
|
||||
Returns:
|
||||
requests.Session: A configured session instance ready to send requests.
|
||||
"""
|
||||
session = requests.Session(
|
||||
impersonate="firefox",
|
||||
)
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
return session
|
||||
|
||||
@property
|
||||
def session(self):
|
||||
return self._session
|
||||
|
||||
@property
|
||||
def proxy(self):
|
||||
return self._proxy
|
||||
@@ -1,33 +0,0 @@
|
||||
import lbc
|
||||
|
||||
def main() -> None:
|
||||
client = lbc.Client()
|
||||
|
||||
# Paris
|
||||
location = lbc.City(
|
||||
lat=48.85994982004764,
|
||||
lng=2.33801967847424,
|
||||
radius=10_000, # 10 km
|
||||
city="Paris"
|
||||
)
|
||||
|
||||
result = client.search(
|
||||
text="maison",
|
||||
locations=[location],
|
||||
page=1,
|
||||
limit=35,
|
||||
limit_alu=0,
|
||||
sort=lbc.Sort.NEWEST,
|
||||
ad_type=lbc.AdType.OFFER,
|
||||
category=lbc.Category.IMMOBILIER,
|
||||
owner_type=lbc.OwnerType.ALL,
|
||||
search_in_title_only=True,
|
||||
square=(200, 400),
|
||||
price=[300_000, 700_000]
|
||||
)
|
||||
|
||||
for ad in result.ads:
|
||||
print(ad.id, ad.url, ad.subject, ad.price)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,16 +0,0 @@
|
||||
import lbc
|
||||
|
||||
def main() -> None:
|
||||
client = lbc.Client()
|
||||
|
||||
result = client.search(
|
||||
url="https://www.leboncoin.fr/recherche?category=9&text=maison&locations=Paris__48.86023250788424_2.339006433295173_9256&square=100-200&price=500000-1000000&rooms=1-6&bedrooms=3-6&outside_access=garden,terrace&orientation=south_west&owner_type=private",
|
||||
page=1,
|
||||
limit=35
|
||||
)
|
||||
|
||||
for ad in result.ads:
|
||||
print(ad.id, ad.url, ad.subject, ad.price)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -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