mirror of
https://github.com/etienne-hd/lbc.git
synced 2026-04-25 00:05:37 +02:00
Compare commits
22 Commits
1.0.0
...
00cf534191
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
00cf534191 | ||
|
|
8a74cde3b1 | ||
|
|
d24b1cc0e6 | ||
|
|
6ef01383f0 | ||
|
|
b9ac610b04 | ||
|
|
672204dd95 | ||
|
|
4fa9409f78 | ||
|
|
8eb8a96e8f | ||
|
|
07d200b653 | ||
|
|
feebd85591 | ||
|
|
76dda0cb76 | ||
|
|
2888942ecc | ||
|
|
fb31b43fd6 | ||
|
|
887ad99959 | ||
|
|
f5b08f0890 | ||
|
|
5841f55b20 | ||
|
|
241645baae | ||
|
|
0320e2575c | ||
|
|
fb9bca527d | ||
|
|
47580f9321 | ||
|
|
e7668b5d89 | ||
|
|
a42332e4ef |
5
.gitignore
vendored
5
.gitignore
vendored
@@ -1,3 +1,6 @@
|
||||
# lbc
|
||||
publish.sh
|
||||
|
||||
# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
@@ -178,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/
|
||||
|
||||
61
CHANGELOG.md
61
CHANGELOG.md
@@ -1,2 +1,61 @@
|
||||
## v1.0.0
|
||||
## 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
|
||||
|
||||
## 1.0.2
|
||||
### Added
|
||||
* Support for full Leboncoin URL in `client.search(url=...)`
|
||||
* New `shippable` argument in the `search` function
|
||||
|
||||
### Fixed
|
||||
* Incorrect enum key assignment in the `build_search_payload_with_args` function
|
||||
|
||||
## 1.0.1
|
||||
### Added
|
||||
* Realistic `Sec-Fetch-*` headers to prevent 403 errors
|
||||
|
||||
## 1.0.0
|
||||
* Initial release
|
||||
22
README.md
22
README.md
@@ -35,14 +35,18 @@ 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
|
||||
**Full documentation will be available soon.**
|
||||
|
||||
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
|
||||
To create client you need to use `lbc.Client` class
|
||||
```python
|
||||
import lbc
|
||||
|
||||
@@ -98,6 +102,20 @@ client.search(
|
||||
)
|
||||
```
|
||||
|
||||
#### Alternatively
|
||||
|
||||
You can also perform search using a full Leboncoin URL:
|
||||
|
||||
```python
|
||||
client.search(
|
||||
url="https://www.leboncoin.fr/recherche?category=9&text=maison&locations=Paris__48.86023250788424_2.339006433295173_9256&square=100-200price=500000-1000000&rooms=1-6&bedrooms=3-6&outside_access=garden,terrace&orientation=south_west&owner_type=private",
|
||||
page=1,
|
||||
limit=35
|
||||
)
|
||||
```
|
||||
|
||||
If `url` is provided, it overrides other keyword parameters such as `text`, `category`, `locations`, etc. However, pagination parameters like `page`, `limit`, and `limit_alu` are still applied.
|
||||
|
||||
### Location
|
||||
|
||||
The `locations` parameter accepts a list of one or more location objects. You can use one of the following:
|
||||
|
||||
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()
|
||||
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,6 +1,6 @@
|
||||
[project]
|
||||
name = "lbc"
|
||||
version = "1.0.0"
|
||||
version = "1.0.10"
|
||||
description = "Unofficial client for Leboncoin API"
|
||||
readme = "README.md"
|
||||
license = {text = "MIT"}
|
||||
|
||||
@@ -1,23 +1,42 @@
|
||||
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
|
||||
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):
|
||||
super().__init__(proxy=proxy)
|
||||
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.
|
||||
|
||||
def _fetch(self, method: str, url: str, payload: Optional[dict] = None, timeout: int = 30) -> dict:
|
||||
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 (staticmethod): HTTP method to use (e.g., `GET`, `POST`).
|
||||
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).
|
||||
@@ -30,20 +49,27 @@ class Client(Session):
|
||||
method=method,
|
||||
url=url,
|
||||
json=payload,
|
||||
timeout=timeout
|
||||
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,
|
||||
@@ -53,13 +79,19 @@ class Client(Session):
|
||||
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.
|
||||
@@ -69,17 +101,71 @@ class Client(Session):
|
||||
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.
|
||||
"""
|
||||
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, search_in_title_only=search_in_title_only, **kwargs
|
||||
)
|
||||
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)
|
||||
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)
|
||||
@@ -12,3 +12,7 @@ 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."""
|
||||
|
||||
@@ -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
|
||||
@@ -1,17 +1,41 @@
|
||||
from .attribute import Attribute
|
||||
from .location import Location
|
||||
from .owner import Owner
|
||||
from .user import User
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from typing import List
|
||||
from typing import List, Any, Optional
|
||||
|
||||
@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: datetime
|
||||
expiration_date: datetime
|
||||
index_date: datetime
|
||||
first_publication_date: str
|
||||
expiration_date: str
|
||||
index_date: str
|
||||
status: str
|
||||
category_id: str
|
||||
category_name: str
|
||||
@@ -24,9 +48,78 @@ class Ad:
|
||||
images: List[str]
|
||||
attributes: List[Attribute]
|
||||
location: Location
|
||||
owner: Owner
|
||||
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
|
||||
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,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,5 +1,4 @@
|
||||
from enum import Enum
|
||||
from typing import Union, Tuple
|
||||
|
||||
class OwnerType(Enum):
|
||||
PRO = "pro"
|
||||
|
||||
@@ -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,11 +1,7 @@
|
||||
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
|
||||
from typing import List, Any
|
||||
|
||||
@dataclass
|
||||
class Search:
|
||||
@@ -20,73 +16,11 @@ class Search:
|
||||
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.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", {})
|
||||
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")
|
||||
)
|
||||
)
|
||||
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"),
|
||||
@@ -97,5 +31,5 @@ class Search:
|
||||
total_inactive=raw.get("total_inactive"),
|
||||
total_shippable=raw.get("total_shippable"),
|
||||
max_pages=raw.get("max_pages"),
|
||||
ads=ads
|
||||
ads=ads,
|
||||
)
|
||||
235
src/lbc/models/user.py
Normal file
235
src/lbc/models/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,25 +1,51 @@
|
||||
from .models import Proxy
|
||||
|
||||
from curl_cffi import requests
|
||||
from curl_cffi import requests, BrowserTypeLiteral
|
||||
from typing import Optional
|
||||
import random
|
||||
|
||||
class Session:
|
||||
def __init__(self, proxy: Optional[Proxy] = None):
|
||||
self._session = self._init_session(proxy=proxy)
|
||||
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) -> requests.Session:
|
||||
def _init_session(self, proxy: Optional[Proxy] = None, impersonate: BrowserTypeLiteral = None, request_verify: bool = True) -> requests.Session:
|
||||
"""
|
||||
Initializes an HTTP session with optional proxy and browser impersonation.
|
||||
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.
|
||||
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="firefox",
|
||||
impersonate=impersonate,
|
||||
)
|
||||
|
||||
session.headers.update(
|
||||
{
|
||||
'Sec-Fetch-Dest': 'empty',
|
||||
'Sec-Fetch-Mode': 'cors',
|
||||
'Sec-Fetch-Site': 'same-site',
|
||||
}
|
||||
)
|
||||
|
||||
if proxy:
|
||||
@@ -28,12 +54,31 @@ class Session:
|
||||
"https": proxy.url
|
||||
}
|
||||
|
||||
session.get("https://www.leboncoin.fr/", verify=request_verify) # Init cookies
|
||||
|
||||
return session
|
||||
|
||||
@property
|
||||
def session(self):
|
||||
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):
|
||||
return self._proxy
|
||||
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")
|
||||
|
||||
160
src/lbc/utils.py
160
src/lbc/utils.py
@@ -3,6 +3,154 @@ from .exceptions import InvalidValue
|
||||
|
||||
from typing import Optional, Union, List
|
||||
|
||||
def build_search_payload_with_url(
|
||||
url: str,
|
||||
limit: int = 35,
|
||||
limit_alu: int = 3,
|
||||
page: int = 1
|
||||
):
|
||||
def build_area(area_values: list[str]) -> dict:
|
||||
area = {
|
||||
"lat": float(area_values[0]),
|
||||
"lng": float(area_values[1])
|
||||
}
|
||||
if len(area_values) >= 3:
|
||||
area["default_radius"] = int(area_values[2])
|
||||
if len(area_values) >= 4:
|
||||
area["radius"] = int(area_values[3])
|
||||
return area
|
||||
|
||||
payload = {
|
||||
"filters": {},
|
||||
"limit": limit,
|
||||
"limit_alu": limit_alu,
|
||||
"offset": limit * (page - 1),
|
||||
"disable_total": True,
|
||||
"extend": True,
|
||||
"listing_source": "direct-search" if page == 1 else "pagination"
|
||||
}
|
||||
|
||||
args: List[str] = url.split("?")[1].split("&")
|
||||
for arg in args:
|
||||
key, value = arg.split("=") # e.g: real_estate_type 3,4 / square 300-400 / category 9
|
||||
|
||||
match key:
|
||||
case "text":
|
||||
payload["filters"]["keywords"] = {
|
||||
"text": value
|
||||
}
|
||||
|
||||
case "category":
|
||||
payload["filters"]["category"] = {
|
||||
"id": value
|
||||
}
|
||||
|
||||
case "locations":
|
||||
payload["filters"]["location"] = {
|
||||
"locations": []
|
||||
}
|
||||
|
||||
locations = value.split(",")
|
||||
for location in locations:
|
||||
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[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
|
||||
payload["filters"]["location"]["locations"].append(
|
||||
{
|
||||
"locationType": "department",
|
||||
"department_id": location_id
|
||||
}
|
||||
)
|
||||
case "r": # Region
|
||||
payload["filters"]["location"]["locations"].append(
|
||||
{
|
||||
"locationType": "region",
|
||||
"region_id": location_id
|
||||
}
|
||||
)
|
||||
case "p": # Place
|
||||
area_values = location_parts[1].split("_") # lat, lng, default_radius, radius
|
||||
payload["filters"]["location"]["locations"].append(
|
||||
{
|
||||
"locationType": "place",
|
||||
"place": location_id,
|
||||
"label": location_id,
|
||||
"area": build_area(area_values)
|
||||
}
|
||||
)
|
||||
case _:
|
||||
raise InvalidValue(f"Unknown location type: {prefix_parts[0]}")
|
||||
|
||||
else: # City
|
||||
area_values = location_parts[1].split("_") # lat, lng, default_radius, radius
|
||||
payload["filters"]["location"]["locations"].append(
|
||||
{
|
||||
"locationType": "city",
|
||||
#"city": location_parts[0],
|
||||
"area": build_area(area_values)
|
||||
}
|
||||
)
|
||||
|
||||
case "order":
|
||||
payload["sort_order"] = value
|
||||
|
||||
case "sort":
|
||||
payload["sort_by"] = value
|
||||
|
||||
case "owner_type":
|
||||
payload["owner_type"] = value
|
||||
|
||||
case "shippable":
|
||||
if value == "1":
|
||||
payload["filters"]["location"]["shippable"] = True
|
||||
|
||||
case _:
|
||||
if value in ["page"]: # Pass
|
||||
continue
|
||||
|
||||
# Range or Enum
|
||||
elif len(value.split("-")) == 2: # Range
|
||||
range_values = value.split("-", 1)
|
||||
if len(range_values) == 2:
|
||||
min_val, max_val = range_values
|
||||
|
||||
try:
|
||||
min_val = int(min_val)
|
||||
except ValueError:
|
||||
min_val = None
|
||||
|
||||
try:
|
||||
max_val = int(max_val)
|
||||
except ValueError:
|
||||
max_val = None
|
||||
|
||||
if not payload["filters"].get("ranges"):
|
||||
payload["filters"]["ranges"] = {}
|
||||
|
||||
if not payload["filters"].get("ranges"):
|
||||
payload["filters"]["ranges"] = {}
|
||||
|
||||
ranges = {}
|
||||
if min_val is not None:
|
||||
ranges["min"] = min_val
|
||||
if max_val is not None:
|
||||
ranges["max"] = max_val
|
||||
|
||||
if ranges:
|
||||
payload["filters"]["ranges"][key] = ranges
|
||||
|
||||
else: # Enum
|
||||
if not payload["filters"].get("enums"):
|
||||
payload["filters"]["enums"] = {}
|
||||
|
||||
payload["filters"]["enums"][key] = value.split(",")
|
||||
|
||||
return payload
|
||||
|
||||
def build_search_payload_with_args(
|
||||
text: Optional[str] = None,
|
||||
category: Category = Category.TOUTES_CATEGORIES,
|
||||
@@ -13,6 +161,7 @@ def build_search_payload_with_args(
|
||||
page: int = 1,
|
||||
ad_type: AdType = AdType.OFFER,
|
||||
owner_type: Optional[OwnerType] = None,
|
||||
shippable: Optional[bool] = False,
|
||||
search_in_title_only: bool = False,
|
||||
**kwargs
|
||||
) -> dict:
|
||||
@@ -99,7 +248,10 @@ def build_search_payload_with_args(
|
||||
# Search in title only
|
||||
if text:
|
||||
if search_in_title_only:
|
||||
payload["filters"]["keywords"]["type"] = "subject"
|
||||
payload["filters"]["keywords"]["type"] = "subject"
|
||||
|
||||
if shippable:
|
||||
payload["filters"]["location"]["shippable"] = True
|
||||
|
||||
if kwargs:
|
||||
for key, value in kwargs.items():
|
||||
@@ -107,7 +259,7 @@ def build_search_payload_with_args(
|
||||
raise InvalidValue(f"The value of '{key}' must be a list or a tuple.")
|
||||
# Range
|
||||
if all(isinstance(x, int) for x in value):
|
||||
if len(value) == 1:
|
||||
if len(value) <= 1:
|
||||
raise InvalidValue(f"The value of '{key}' must be a list or tuple with at least two elements.")
|
||||
|
||||
if not "ranges" in payload["filters"]:
|
||||
@@ -119,8 +271,8 @@ def build_search_payload_with_args(
|
||||
}
|
||||
# Enum
|
||||
elif all(isinstance(x, str) for x in value):
|
||||
payload["filters"]["enums"]["key"] = value
|
||||
payload["filters"]["enums"][key] = value
|
||||
else:
|
||||
raise InvalidValue(f"The value of '{key}' must be a list or tuple containing only integers or only strings.")
|
||||
|
||||
|
||||
return payload
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user