375 lines
17 KiB
Python
375 lines
17 KiB
Python
"""Collect unnormalized Leboncoin search and individual listing observations."""
|
|
|
|
import argparse
|
|
import gzip
|
|
import json
|
|
import math
|
|
import sys
|
|
import time
|
|
import tomllib
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from urllib.parse import urlencode
|
|
|
|
|
|
FIELDS = {
|
|
"list_id", "first_publication_date", "expiration_date", "index_date",
|
|
"status", "category_id", "category_name", "ad_type", "price", "price_cents",
|
|
"subject", "body", "url", "brand", "is_boosted", "options", "counters",
|
|
}
|
|
EXCLUDED_ATTRIBUTES = {
|
|
"phone", "phone_number", "email", "name", "first_name", "last_name",
|
|
"siren", "siret", "licence_plate", "license_plate", "registration_number", "vin",
|
|
"vehicle_history_report_public_url",
|
|
}
|
|
EXCLUDED_ATTRIBUTE_PREFIXES = ("owner_", "seller_", "profile_", "rating_", "contact_")
|
|
|
|
|
|
class SourceAccessError(RuntimeError):
|
|
def __init__(self, status, capture=None):
|
|
super().__init__("Source access failed; stopping without retries")
|
|
self.status = status
|
|
self.capture = capture
|
|
|
|
|
|
def now():
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def search_url(text, page):
|
|
return "https://www.leboncoin.fr/recherche?" + urlencode(
|
|
{"category": "2", "text": text, "page": page}
|
|
)
|
|
|
|
|
|
def validate_ad(ad):
|
|
if not isinstance(ad, dict):
|
|
raise ValueError("Invalid listing object")
|
|
identifier = str(ad.get("list_id", ""))
|
|
if not identifier.isascii() or not identifier.isdigit() or int(identifier) < 1:
|
|
raise ValueError("Invalid listing ID")
|
|
if not isinstance(ad.get("location", {}), dict):
|
|
raise ValueError("Invalid location object")
|
|
attributes = ad.get("attributes", [])
|
|
if not isinstance(attributes, list) or any(
|
|
not isinstance(a, dict) or not isinstance(a.get("key"), str) for a in attributes
|
|
):
|
|
raise ValueError("Invalid attributes")
|
|
|
|
|
|
def parse_search(text):
|
|
"""Validate the source envelope, not the vehicle values."""
|
|
try:
|
|
data = json.loads(text)["props"]["pageProps"]["searchData"]
|
|
ads, total, pages = data["ads"], data["total"], data["max_pages"]
|
|
if not isinstance(ads, list):
|
|
raise ValueError()
|
|
if type(total) is not int or type(pages) is not int or total < 0 or pages < 0:
|
|
raise ValueError()
|
|
if (total > 0 and (not ads or pages == 0)) or (total == 0 and ads):
|
|
raise ValueError()
|
|
for ad in ads:
|
|
validate_ad(ad)
|
|
return data
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
raise ValueError("Missing or changed Leboncoin search payload") from exc
|
|
|
|
|
|
def parse_detail(text, expected_id):
|
|
try:
|
|
ad = json.loads(text)["props"]["pageProps"]["ad"]
|
|
validate_ad(ad)
|
|
if str(ad["list_id"]) != str(expected_id):
|
|
raise ValueError()
|
|
return ad
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
raise ValueError("Missing, changed, or mismatched Leboncoin detail payload") from exc
|
|
|
|
|
|
def minimize(ad):
|
|
result = {key: value for key, value in ad.items() if key in FIELDS}
|
|
if "attributes" in ad:
|
|
result["attributes"] = [
|
|
a for a in ad["attributes"]
|
|
if a["key"].lower() not in EXCLUDED_ATTRIBUTES
|
|
and not a["key"].lower().startswith(EXCLUDED_ATTRIBUTE_PREFIXES)
|
|
]
|
|
if "location" in ad:
|
|
result["location"] = ad["location"]
|
|
return result
|
|
|
|
|
|
def write_json(path, value):
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
temporary.replace(path)
|
|
|
|
|
|
def write_records(path, records):
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
with gzip.open(temporary, "wt", encoding="utf-8") as stream:
|
|
for record in records:
|
|
stream.write(json.dumps(record, ensure_ascii=False) + "\n")
|
|
temporary.replace(path)
|
|
|
|
|
|
def collect(fetch, searches, output, max_pages=10, delay=5, search_only=False, max_details=None, detail_policy=None):
|
|
run_id = str(uuid.uuid4())
|
|
directory = output / datetime.now(timezone.utc).strftime("%Y-%m-%d") / run_id
|
|
directory.mkdir(mode=0o700, parents=True, exist_ok=False)
|
|
manifest = {
|
|
"schema_version": 2, "source": "leboncoin", "run_id": run_id,
|
|
"started_at": now(), "status": "running", "max_pages": max_pages,
|
|
"delay_seconds": delay, "searches": [],
|
|
"privacy_policy": "v2-text-location-filtered-attributes",
|
|
"planned_searches": searches,
|
|
"details": {
|
|
"enabled": not search_only, "max_details": max_details,
|
|
"policy": "incremental" if detail_policy is not None else "always",
|
|
"refresh_days": getattr(detail_policy, "refresh_days", None),
|
|
"status": "disabled" if search_only else "running", "listings": {},
|
|
},
|
|
}
|
|
path = directory / "manifest.json"
|
|
write_json(path, manifest)
|
|
requests = 0
|
|
detail_attempts = 0
|
|
target = None
|
|
|
|
def fetch_page(url, item=None):
|
|
nonlocal requests, detail_attempts
|
|
if requests:
|
|
time.sleep(delay)
|
|
if item is not None:
|
|
detail_attempts += 1
|
|
item.update(status="running", attempted_at=now())
|
|
write_json(path, manifest)
|
|
requests += 1
|
|
return fetch(url)
|
|
|
|
try:
|
|
for index, search in enumerate(searches):
|
|
scope = {**search, "status": "running", "pages": [], "records": 0}
|
|
manifest["searches"].append(scope)
|
|
write_json(path, manifest)
|
|
seen_pages = set()
|
|
for number in range(1, max_pages + 1):
|
|
target = scope
|
|
url = search_url(search["text"], number)
|
|
data = parse_search(fetch_page(url))
|
|
observed_at = now()
|
|
signature = frozenset(str(ad["list_id"]) for ad in data["ads"])
|
|
if signature and signature in seen_pages:
|
|
raise ValueError("Repeated result page; pagination did not advance")
|
|
seen_pages.add(signature)
|
|
filename = f"search-{index + 1:03d}-page-{number:04d}.jsonl.gz"
|
|
write_records(directory / filename, ({
|
|
"schema_version": 2, "source": "leboncoin", "run_id": run_id,
|
|
"record_type": "search", "observed_at": observed_at,
|
|
"family": search["family"], "search_text": search["text"],
|
|
"search_url": url, "page": number, "payload": minimize(ad),
|
|
} for ad in data["ads"]))
|
|
scope["pages"].append({
|
|
"page": number, "file": filename, "observed_at": observed_at,
|
|
"reported_total": data["total"], "reported_max_pages": data["max_pages"],
|
|
"records": len(data["ads"]),
|
|
})
|
|
scope["records"] += len(data["ads"])
|
|
if number >= data["max_pages"]:
|
|
scope["status"] = "complete"
|
|
elif number == max_pages:
|
|
scope["status"] = "partial"
|
|
scope["reason"] = "page_limit"
|
|
|
|
details = manifest["details"]["listings"]
|
|
if not search_only:
|
|
for ad in data["ads"]:
|
|
details.setdefault(str(ad["list_id"]), {"status": "pending"})
|
|
# Finalize discovery before visiting details so enrichment cannot lose it.
|
|
write_json(path, manifest)
|
|
if not search_only:
|
|
for ad in data["ads"]:
|
|
identifier = str(ad["list_id"])
|
|
item = details[identifier]
|
|
if item["status"] != "pending":
|
|
continue
|
|
if detail_policy is not None:
|
|
decision = detail_policy(ad)
|
|
item["detail_decision"] = decision
|
|
if decision == "fresh_unchanged":
|
|
item.update(status="not_attempted", reason="fresh_unchanged")
|
|
continue
|
|
if max_details is not None and detail_attempts >= max_details:
|
|
item.update(status="not_attempted", reason="detail_limit")
|
|
continue
|
|
target = item
|
|
# Construct a same-source URL rather than navigate untrusted payload URLs.
|
|
detail_url = f"https://www.leboncoin.fr/ad/voitures/{identifier}"
|
|
item["url"] = detail_url
|
|
try:
|
|
detail = parse_detail(fetch_page(detail_url, item), identifier)
|
|
except SourceAccessError as exc:
|
|
if exc.status not in (404, 410):
|
|
raise
|
|
item.update(status="unavailable", http_status=exc.status)
|
|
write_json(path, manifest)
|
|
continue
|
|
detail_time = now()
|
|
detail_file = f"detail-{identifier}.jsonl.gz"
|
|
write_records(directory / detail_file, [{
|
|
"schema_version": 2, "source": "leboncoin", "run_id": run_id,
|
|
"record_type": "detail", "observed_at": detail_time,
|
|
"listing_url": detail_url, "payload": minimize(detail),
|
|
}])
|
|
item.update(status="complete", file=detail_file, observed_at=detail_time)
|
|
write_json(path, manifest)
|
|
if scope["status"] == "complete":
|
|
break
|
|
write_json(path, manifest)
|
|
except (Exception, KeyboardInterrupt) as exc:
|
|
manifest["error_type"] = type(exc).__name__
|
|
if isinstance(exc, SourceAccessError) and exc.status == 403 and exc.capture is not None:
|
|
try:
|
|
diagnostic = directory / "diagnostics/http-403"
|
|
report = exc.capture(diagnostic)
|
|
manifest["diagnostics"] = {
|
|
"directory": "diagnostics/http-403", "http_status": 403,
|
|
"capture_status": "partial" if report.get("errors") or report.get("truncated") else "complete",
|
|
}
|
|
except Exception:
|
|
# Preserve the access error and finalized records if capture fails.
|
|
manifest["diagnostics"] = {"http_status": 403, "capture_status": "failed"}
|
|
if target is not None:
|
|
if target["status"] == "pending":
|
|
target["status"] = "not_attempted"
|
|
else:
|
|
target["status"] = "partial" if target.get("pages") else "failed"
|
|
# Exception bodies may contain page contents or personal information.
|
|
target["reason"] = type(exc).__name__
|
|
if isinstance(exc, SourceAccessError):
|
|
target["http_status"] = exc.status
|
|
finally:
|
|
for scope in manifest["searches"]:
|
|
if scope["status"] == "running":
|
|
scope["status"] = "partial" if scope["pages"] else "failed"
|
|
details = manifest["details"]
|
|
if details["enabled"]:
|
|
for item in details["listings"].values():
|
|
if item["status"] == "pending":
|
|
item.update(status="not_attempted", reason="run_stopped")
|
|
details["status"] = "complete" if all(
|
|
item["status"] == "complete" for item in details["listings"].values()
|
|
) else "partial"
|
|
if not details["listings"] and "error_type" in manifest:
|
|
details["status"] = "not_attempted"
|
|
complete = (
|
|
"error_type" not in manifest
|
|
and len(manifest["searches"]) == len(searches)
|
|
and all(s["status"] == "complete" for s in manifest["searches"])
|
|
and details["status"] in ("complete", "disabled")
|
|
)
|
|
manifest["status"] = "complete" if complete else (
|
|
"partial" if any(s["pages"] for s in manifest["searches"]) else "failed"
|
|
)
|
|
details["attempted"] = detail_attempts
|
|
manifest["finished_at"] = now()
|
|
write_json(path, manifest)
|
|
return path, manifest["status"]
|
|
|
|
|
|
def parse_collection_args(argv=None, description=None):
|
|
parser = argparse.ArgumentParser(description=description or __doc__)
|
|
parser.add_argument("--config", type=Path, default=Path("searches.toml"))
|
|
parser.add_argument("--output", type=Path, default=Path("data/raw/leboncoin"))
|
|
parser.add_argument("--family", help="Collect only this configured family")
|
|
parser.add_argument("--max-pages", type=int, default=10)
|
|
parser.add_argument("--delay", type=float, default=5)
|
|
parser.add_argument("--headed", action="store_true")
|
|
parser.add_argument("--search-only", action="store_true", help="Skip individual listing pages")
|
|
parser.add_argument("--max-details", type=int, help="Maximum unique detail requests per run")
|
|
parser.add_argument("--detail-policy", choices=("always", "incremental"), default="always")
|
|
parser.add_argument("--detail-refresh-days", type=int, default=7)
|
|
args = parser.parse_args(argv)
|
|
if args.max_pages < 1 or not math.isfinite(args.delay) or args.delay < 1:
|
|
parser.error("max-pages must be positive and delay must be finite and >= 1 second")
|
|
if args.max_details is not None and args.max_details < 1:
|
|
parser.error("max-details must be positive")
|
|
if args.detail_refresh_days < 1:
|
|
parser.error("detail-refresh-days must be positive")
|
|
with args.config.open("rb") as stream:
|
|
searches = tomllib.load(stream).get("searches", [])
|
|
if not isinstance(searches, list) or not searches or any(
|
|
not isinstance(s, dict) or set(s) != {"family", "text"}
|
|
or any(not isinstance(v, str) or not v.strip() for v in s.values())
|
|
for s in searches
|
|
):
|
|
parser.error("config needs searches with nonempty family and text strings")
|
|
if args.family:
|
|
searches = [s for s in searches if s["family"] == args.family]
|
|
if not searches:
|
|
parser.error("no matching searches")
|
|
args.searches = searches
|
|
return args
|
|
|
|
|
|
def collect_browser(args):
|
|
from playwright.sync_api import sync_playwright
|
|
from playwright_stealth import Stealth
|
|
|
|
result = None
|
|
try:
|
|
with Stealth(navigator_languages_override=("fr-FR", "fr")).use_sync(sync_playwright()) as playwright:
|
|
browser = playwright.chromium.launch(headless=not args.headed)
|
|
try:
|
|
context = browser.new_context(locale="fr-FR")
|
|
page = context.new_page()
|
|
|
|
def fetch(url):
|
|
response = page.goto(url, wait_until="domcontentloaded", timeout=60_000)
|
|
if response is None or response.status != 200:
|
|
capture = None
|
|
if response is not None and response.status == 403:
|
|
from .diagnostics import capture_403
|
|
|
|
def capture(directory):
|
|
return capture_403(directory, response, page, browser, {
|
|
"headed": args.headed, "locale": "fr-FR", "delay_seconds": args.delay,
|
|
"max_pages": args.max_pages, "max_details": args.max_details,
|
|
"stealth": True,
|
|
})
|
|
raise SourceAccessError(response.status if response else None, capture=capture)
|
|
return page.locator("script#__NEXT_DATA__").text_content(timeout=15_000)
|
|
|
|
result = collect(
|
|
fetch, args.searches, args.output, args.max_pages, args.delay,
|
|
args.search_only, args.max_details,
|
|
detail_policy=getattr(args, "detail_decider", None),
|
|
)
|
|
finally:
|
|
browser.close()
|
|
except KeyboardInterrupt:
|
|
if result is not None:
|
|
print(f"Finalized manifest before interruption: {result[0]}", file=sys.stderr)
|
|
raise
|
|
except Exception as exc:
|
|
if result is None:
|
|
raise
|
|
# Browser teardown must not discard an already-finalized raw run.
|
|
print(f"Browser cleanup failed ({type(exc).__name__}); raw run is finalized.", file=sys.stderr)
|
|
try:
|
|
write_json(result[0].parent / "browser_cleanup.json", {"error_type": type(exc).__name__})
|
|
except OSError:
|
|
print("Could not record browser cleanup failure.", file=sys.stderr)
|
|
return result
|
|
|
|
|
|
def main():
|
|
args = parse_collection_args()
|
|
if args.detail_policy == "incremental":
|
|
print("Incremental details require car-tracker-pipeline and imported database history.", file=sys.stderr)
|
|
return 3
|
|
path, status = collect_browser(args)
|
|
print(f"{status}: {path}")
|
|
return 0 if status == "complete" else 1
|