Hermes Agent
144 lines
5.6 KiB
Python
144 lines
5.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Scrape mobile.de for Mazda 2 listings near Mönchengladbach using Playwright + proxy"""
|
|
import asyncio, json, re, sys
|
|
from playwright.async_api import async_playwright
|
|
|
|
MG_PLZ = "41061"
|
|
BUDGET = 10000
|
|
MAX_KM = 120000
|
|
MIN_YEAR = 2014
|
|
|
|
# mobile.de URL with filters
|
|
URL = (
|
|
f"https://suchen.mobile.de/fahrzeuge/search.html"
|
|
f"?dam=0&isSearchRequest=true&ms=1200;3;;&pMax={BUDGET}"
|
|
f"&mlMin={MIN_YEAR}&mlMax:{MAX_KM}&sf={MG_PLZ}&vc=Car&od=down&sb=rel"
|
|
)
|
|
|
|
PROXY = {
|
|
"server": "http://geo.iproyal.com:12321",
|
|
"username": "8B7M8nJOfXTNHPUw",
|
|
"password": "lR1VidJhN5RrRbeK_region-europe"
|
|
}
|
|
|
|
async def main():
|
|
listings = []
|
|
|
|
async with async_playwright() as p:
|
|
browser = await p.chromium.launch(
|
|
headless=True,
|
|
args=['--no-sandbox', '--disable-setuid-sandbox']
|
|
)
|
|
context = await browser.new_context(
|
|
proxy=PROXY,
|
|
user_agent='Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36',
|
|
locale='de-DE',
|
|
timezone_id='Europe/Berlin'
|
|
)
|
|
page = await context.new_page()
|
|
|
|
try:
|
|
print(f"Loading {URL}")
|
|
await page.goto(URL, wait_until='networkidle', timeout=30000)
|
|
|
|
content = await page.content()
|
|
print(f"Page loaded: {len(content)} bytes")
|
|
|
|
if 'Zugriff verweigert' in content or 'Access denied' in content:
|
|
print("BLOCKED - access denied with proxy too")
|
|
await browser.close()
|
|
return []
|
|
|
|
# Wait for results to load
|
|
try:
|
|
await page.wait_for_selector('[data-testid="result-list"], .result-list, .cfyXj, article', timeout=10000)
|
|
except:
|
|
print("No results selector found, trying to extract anyway...")
|
|
|
|
# Extract listings from page
|
|
content = await page.content()
|
|
|
|
# Try to find structured data via JSON-LD or embedded data
|
|
# mobile.de embeds listing data in <script> tags or in the HTML
|
|
|
|
# Method 1: Look for data attributes with listing info
|
|
listing_blocks = re.findall(
|
|
r'<article[^>]*>(.*?)</article>',
|
|
content, re.DOTALL
|
|
)
|
|
print(f"Found {len(listing_blocks)} article blocks")
|
|
|
|
if not listing_blocks:
|
|
# Try alternative selectors
|
|
listing_blocks = re.findall(
|
|
r'<(?:div|article)[^>]*class="[^"]*?(?:result-item|listing-item|cBoxBody)[^"]*"[^>]*>(.*?)</(?:div|article)>',
|
|
content, re.DOTALL
|
|
)
|
|
print(f"Found {len(listing_blocks)} result-item blocks")
|
|
|
|
for i, block in enumerate(listing_blocks[:25]):
|
|
# Extract title
|
|
title_m = re.search(r'<h[23][^>]*>(.*?)</h[23]>', block, re.DOTALL)
|
|
title = re.sub(r'<[^>]+>', '', title_m.group(1)).strip() if title_m else '?'
|
|
|
|
# Extract price
|
|
price_m = re.search(r'(\d{1,2}[\.\d]*\s*€)', block)
|
|
price = price_m.group(1) if price_m else '?'
|
|
|
|
# Extract mileage
|
|
km_m = re.search(r'(\d{1,3}[\.\d]*\s*km)', block, re.IGNORECASE)
|
|
km = km_m.group(1) if km_m else '?'
|
|
|
|
# Extract year
|
|
year_m = re.search(r'(EZ\s*\d{1,2}/\d{4}|(?:20[12]\d{1}))', block)
|
|
year = year_m.group(1) if year_m else '?'
|
|
|
|
# Extract location
|
|
loc_m = re.search(r'(?:4[0-9]{4}\s[A-Za-zäöüß\s-]+)', block)
|
|
loc = loc_m.group(0) if loc_m else '?'
|
|
|
|
# Extract link
|
|
link_m = re.search(r'href="(/fahrzeuge/detail/[^"]+)"', block)
|
|
link = f"https://suchen.mobile.de{link_m.group(1)}" if link_m else '?'
|
|
|
|
listings.append({
|
|
'title': title,
|
|
'price': price,
|
|
'km': km,
|
|
'year': year,
|
|
'location': loc,
|
|
'link': link
|
|
})
|
|
|
|
# If no blocks found, try extracting from full page text
|
|
if not listings:
|
|
# Try to find price/km/year patterns in the whole page
|
|
price_kms = re.findall(r'(\d{1,2}[\.\d]*\s*€)\s*.*?(\d{1,3}[\.\d]*\s*km)', content, re.DOTALL)
|
|
print(f"Found {len(price_kms)} price-km pairs in full text")
|
|
for pk in price_kms[:15]:
|
|
print(f" {pk[0]} | {pk[1]}")
|
|
|
|
# Print results
|
|
print(f"\n=== Extracted {len(listings)} listings ===")
|
|
for i, l in enumerate(listings):
|
|
print(f"{i+1}. {l['title']}")
|
|
print(f" {l['price']} | {l['km']} | {l['year']} | {l['location']}")
|
|
print(f" {l['link']}")
|
|
print()
|
|
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
await browser.close()
|
|
|
|
return listings
|
|
|
|
if __name__ == "__main__":
|
|
listings = asyncio.run(main())
|
|
# Write JSON output
|
|
with open("/tmp/mazda2_listings.json", "w") as f:
|
|
json.dump(listings, f, indent=2, ensure_ascii=False)
|
|
print(f"\nSaved {len(listings)} listings to /tmp/mazda2_listings.json")
|