243bc9b6f5
Some checks are pending
Python application / build (push) Waiting to run
- Geocodes listing locations via Photon (Komoot) API - no rate limits - Shows distance in km next to location in all output formats - Uses geodesic distance (geopy) for accurate km calculation - Caches geocoded locations to avoid repeated API calls - Works with --top, --compare, --verdict, and default output - Example: --from "Farum" shows 'København V (20 km)', 'Korsør (94 km)' - Added geopy to requirements.txt
751 lines
28 KiB
Python
Executable File
751 lines
28 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""DBA vs PriceRunner price comparator with "should I buy it used?" verdicts.
|
||
|
||
Searches DBA (dba.dk) for used listings and PriceRunner for new prices,
|
||
then gives a side-by-side comparison with a buy/wait recommendation.
|
||
|
||
Usage:
|
||
python3 dba_pricerunner_scraper.py "playstation 5" --pricerunner --compare --verdict
|
||
python3 dba_pricerunner_scraper.py "iphone 15" --pricerunner --compare --format markdown --verdict
|
||
python3 dba_pricerunner_scraper.py "playstation 5" --top 10
|
||
"""
|
||
|
||
import sys
|
||
import time
|
||
import json
|
||
import re
|
||
import argparse
|
||
import statistics
|
||
import requests
|
||
from bs4 import BeautifulSoup
|
||
from urllib.parse import quote_plus, urljoin
|
||
|
||
try:
|
||
from geopy.distance import geodesic
|
||
import requests as _requests
|
||
_GEO_CACHE = {}
|
||
def _geocode(name):
|
||
if not name:
|
||
return None
|
||
if name in _GEO_CACHE:
|
||
return _GEO_CACHE[name]
|
||
try:
|
||
r = _requests.get(
|
||
'https://photon.komoot.io/api/',
|
||
params={'q': f'{name}, Danmark'},
|
||
headers={'User-Agent': 'dba_pricerunner/1.0'},
|
||
timeout=10
|
||
)
|
||
if r.status_code == 200:
|
||
data = r.json()
|
||
features = data.get('features', [])
|
||
if features:
|
||
coords = features[0]['geometry']['coordinates']
|
||
result = (coords[1], coords[0]) # [lon, lat] -> (lat, lon)
|
||
else:
|
||
result = None
|
||
else:
|
||
result = None
|
||
except Exception:
|
||
result = None
|
||
_GEO_CACHE[name] = result
|
||
return result
|
||
except ImportError:
|
||
def _geocode(name):
|
||
return None
|
||
|
||
try:
|
||
from selenium import webdriver
|
||
from selenium.webdriver.chrome.service import Service as ChromeService
|
||
from webdriver_manager.chrome import ChromeDriverManager
|
||
from selenium.webdriver.chrome.options import Options as ChromeOptions
|
||
except Exception:
|
||
webdriver = None
|
||
|
||
HEADERS = {
|
||
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0 Safari/537.36"
|
||
}
|
||
|
||
|
||
class ScraperError(Exception):
|
||
pass
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────
|
||
# Price parsing
|
||
# ──────────────────────────────────────────────────────────────
|
||
|
||
def normalize_price(price_str):
|
||
"""Extract a numeric price (float) from a string like 'kr. 1.234' or '1.234,00 kr'."""
|
||
if not price_str:
|
||
return None
|
||
s = price_str.replace("kr.", "").replace("kr", "").replace("DKK", "")
|
||
s = s.replace("\u00A0", " ")
|
||
s = s.strip()
|
||
cleaned = re.sub(r"[^0-9\.,\s]", "", s)
|
||
if not cleaned:
|
||
return None
|
||
|
||
# If there are separators followed by exactly three digits (e.g. '3.250' or '1 234'),
|
||
# it's very likely a thousands separator. Remove dots/spaces in that case.
|
||
if re.search(r'(?:[.\s]\d{3})', cleaned):
|
||
core = re.sub(r'[.\s]', '', cleaned)
|
||
core = core.replace(',', '.')
|
||
core = ''.join(ch for ch in core if (ch.isdigit() or ch == '.'))
|
||
try:
|
||
return float(core)
|
||
except Exception:
|
||
return None
|
||
|
||
# Decide which of '.' or ',' is the decimal separator by looking at the last occurrence
|
||
last_dot = cleaned.rfind('.')
|
||
last_comma = cleaned.rfind(',')
|
||
decimal_sep = None
|
||
if last_dot == -1 and last_comma == -1:
|
||
decimal_sep = None
|
||
elif last_dot > last_comma:
|
||
if len(cleaned) - last_dot - 1 in (1, 2, 3):
|
||
decimal_sep = '.'
|
||
else:
|
||
if len(cleaned) - last_comma - 1 in (1, 2, 3):
|
||
decimal_sep = ','
|
||
|
||
if decimal_sep is None:
|
||
digits = re.sub(r"[\s\.,]", "", cleaned)
|
||
try:
|
||
return float(digits)
|
||
except Exception:
|
||
return None
|
||
else:
|
||
if decimal_sep == '.':
|
||
core = re.sub(r"[\s,]", "", cleaned)
|
||
else:
|
||
core = re.sub(r"[.\s]", "", cleaned)
|
||
core = core.replace(',', '.')
|
||
core = ''.join(ch for ch in core if (ch.isdigit() or ch == '.'))
|
||
try:
|
||
return float(core)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def extract_price_string(text):
|
||
"""Return a short price string from a larger text blob, e.g. '3.999 kr.' or '150 kr.'"""
|
||
if not text:
|
||
return ''
|
||
m = re.search(r"(\d{1,3}(?:[.\s]\d{3})*(?:,\d{1,2})?)\s*(kr\.?|DKK)?", text, flags=re.IGNORECASE)
|
||
if not m:
|
||
m = re.search(r"(\d+[\d.\s,]*)", text)
|
||
if not m:
|
||
return ''
|
||
num = m.group(1)
|
||
cur = m.group(2) or 'kr.'
|
||
cur = cur.strip()
|
||
if cur.upper() == 'DKK':
|
||
cur = 'DKK'
|
||
elif not cur:
|
||
cur = 'kr.'
|
||
return f"{num} {cur}".strip()
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────
|
||
# DBA scraping
|
||
# ──────────────────────────────────────────────────────────────
|
||
|
||
def _parse_dba_article(el, origin_coords=None):
|
||
"""Extract title, price, url, location from a DBA <article> element.
|
||
|
||
If origin_coords is provided, also computes travel distance from origin.
|
||
"""
|
||
# Title: h2 text directly (the h2 has NO child <a> in current markup)
|
||
h2 = el.select_one('h2')
|
||
title = h2.get_text(strip=True) if h2 else ''
|
||
|
||
# Link: the sf-search-ad-link overlay (direct child of <article>)
|
||
link_el = el.select_one('a.sf-search-ad-link')
|
||
link = urljoin('https://www.dba.dk', link_el['href']) if link_el and link_el.get('href') else None
|
||
|
||
# Price: first span inside the flex justify-between div
|
||
price = ''
|
||
price_div = el.select_one('div.flex.justify-between')
|
||
if price_div:
|
||
price = price_div.get_text(strip=True)
|
||
if not price:
|
||
# fallback: scan all spans
|
||
for tag in el.find_all(['span', 'div', 'p']):
|
||
txt = tag.get_text(' ', strip=True)
|
||
if not txt:
|
||
continue
|
||
pstr = extract_price_string(txt)
|
||
if pstr:
|
||
price = pstr
|
||
break
|
||
|
||
# Location: span with whitespace-nowrap truncate mr-8
|
||
location = ''
|
||
loc_el = el.select_one('span.whitespace-nowrap.truncate')
|
||
if loc_el and loc_el.get_text(strip=True):
|
||
location = loc_el.get_text(strip=True)
|
||
else:
|
||
loc_block = el.select_one('.text-xs.s-text-subtle')
|
||
if loc_block:
|
||
sp = loc_block.select_one('span')
|
||
if sp and sp.get_text(strip=True):
|
||
location = sp.get_text(strip=True)
|
||
|
||
# Distance from origin (if provided)
|
||
distance_km = None
|
||
if origin_coords and location:
|
||
coords = _geocode(location)
|
||
if coords:
|
||
distance_km = geodesic(origin_coords, coords).km
|
||
|
||
price_num = normalize_price(price)
|
||
return {
|
||
"site": "dba",
|
||
"title": title,
|
||
"price": price,
|
||
"price_num": price_num,
|
||
"url": link,
|
||
"location": location,
|
||
"distance_km": distance_km,
|
||
}
|
||
|
||
|
||
def search_dba_requests(query, max_results=10, origin=None):
|
||
"""Search dba.dk (requests) and return list of dicts with title, price, url, location."""
|
||
origin_coords = None
|
||
if origin:
|
||
origin_coords = _geocode(origin)
|
||
q = quote_plus(query)
|
||
url = f"https://www.dba.dk/recommerce/forsale/search?q={q}"
|
||
r = requests.get(url, headers=HEADERS, timeout=15)
|
||
if r.status_code != 200:
|
||
raise ScraperError(f"DBA returned status {r.status_code}")
|
||
soup = BeautifulSoup(r.text, "lxml")
|
||
items = []
|
||
results = soup.select('article') or soup.select('.cAdList__item') or soup.select('.dba-result')
|
||
for el in results[:max_results]:
|
||
# skip skeleton/placeholder articles (no text content)
|
||
if not el.get_text(strip=True):
|
||
continue
|
||
item = _parse_dba_article(el, origin_coords)
|
||
if item['title']:
|
||
items.append(item)
|
||
time.sleep(1)
|
||
return items
|
||
|
||
|
||
def search_dba_selenium(query, max_results=10, headless=True, origin=None):
|
||
"""Render DBA with Selenium and extract the same fields as requests path."""
|
||
if webdriver is None:
|
||
raise ScraperError('Selenium or webdriver-manager not installed')
|
||
origin_coords = None
|
||
if origin:
|
||
origin_coords = _geocode(origin)
|
||
opts = ChromeOptions()
|
||
if headless:
|
||
opts.add_argument('--headless=new')
|
||
opts.add_argument('--no-sandbox')
|
||
opts.add_argument('--disable-dev-shm-usage')
|
||
driver = webdriver.Chrome(service=ChromeService(ChromeDriverManager().install()), options=opts)
|
||
try:
|
||
q = quote_plus(query)
|
||
url = f"https://www.dba.dk/soeg/?soegeord={q}"
|
||
driver.get(url)
|
||
time.sleep(2)
|
||
soup = BeautifulSoup(driver.page_source, 'lxml')
|
||
items = []
|
||
results = soup.select('article') or soup.select('.cAdList__item') or soup.select('.dba-result')
|
||
for el in results[:max_results]:
|
||
if not el.get_text(strip=True):
|
||
continue
|
||
item = _parse_dba_article(el, origin_coords)
|
||
if item['title']:
|
||
items.append(item)
|
||
time.sleep(1)
|
||
return items
|
||
finally:
|
||
try:
|
||
driver.quit()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────
|
||
# PriceRunner scraping
|
||
# ──────────────────────────────────────────────────────────────
|
||
|
||
def _extract_json_array(text, key):
|
||
"""Find a JSON array by key in a large HTML/JS blob and return the array text."""
|
||
needle = f'"{key}":['
|
||
idx = text.find(needle)
|
||
if idx == -1:
|
||
return None
|
||
i = text.find('[', idx)
|
||
if i == -1:
|
||
return None
|
||
in_str = False
|
||
esc = False
|
||
depth = 0
|
||
for j, ch in enumerate(text[i:], start=i):
|
||
if ch == '"' and not esc:
|
||
in_str = not in_str
|
||
if ch == '\\' and not esc:
|
||
esc = True
|
||
continue
|
||
esc = False
|
||
if not in_str:
|
||
if ch == '[':
|
||
depth += 1
|
||
elif ch == ']':
|
||
depth -= 1
|
||
if depth == 0:
|
||
return text[i:j+1]
|
||
return None
|
||
|
||
|
||
def search_pricerunner_requests(query, max_results=10):
|
||
"""Fetch PriceRunner search page and extract embedded product JSON (requests)."""
|
||
q = quote_plus(query)
|
||
url = f"https://www.pricerunner.dk/results?q={q}"
|
||
r = requests.get(url, headers=HEADERS, timeout=15)
|
||
if r.status_code != 200:
|
||
raise ScraperError(f"PriceRunner returned status {r.status_code}")
|
||
arr_text = _extract_json_array(r.text, 'products')
|
||
items = []
|
||
if not arr_text:
|
||
soup = BeautifulSoup(r.text, 'lxml')
|
||
cards = soup.select('.product, .product-item, .search-result')
|
||
for el in cards[:max_results]:
|
||
t = el.select_one('.product-title, h3, .title')
|
||
p = el.select_one('.price, .product-price')
|
||
a = el.select_one('a[href]')
|
||
title = t.get_text(strip=True) if t else ''
|
||
price = p.get_text(' ', strip=True) if p else ''
|
||
link = urljoin('https://www.pricerunner.dk', a['href']) if a and a.get('href') else None
|
||
price_num = normalize_price(price)
|
||
items.append({'site': 'pricerunner', 'title': title, 'price': price, 'price_num': price_num, 'url': link})
|
||
return items
|
||
try:
|
||
products = json.loads(arr_text)
|
||
except Exception:
|
||
return items
|
||
for p in products[:max_results]:
|
||
name = p.get('name')
|
||
price = None
|
||
lp = p.get('lowestPrice') or {}
|
||
if isinstance(lp, dict):
|
||
price = lp.get('amount')
|
||
price_str = f"{price} {lp.get('currency','')}".strip() if price else ''
|
||
path = p.get('url')
|
||
full = urljoin('https://www.pricerunner.dk', path) if path else None
|
||
price_num = normalize_price(price_str)
|
||
items.append({'site': 'pricerunner', 'title': name, 'price': price_str, 'price_num': price_num, 'url': full})
|
||
time.sleep(1)
|
||
return items
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────
|
||
# Verdict engine — "should I buy it used?"
|
||
# ──────────────────────────────────────────────────────────────
|
||
|
||
def compute_verdict(dba_items, pr_items):
|
||
"""Compare used (DBA) vs new (PriceRunner) prices and produce a recommendation.
|
||
|
||
Returns a dict with:
|
||
new_price, used_price, used_median, savings, savings_pct,
|
||
verdict (one of: 'buy_used', 'buy_new', 'wait', 'mixed'),
|
||
reason (human-readable explanation)
|
||
"""
|
||
new_prices = [it['price_num'] for it in pr_items if it.get('price_num') is not None]
|
||
used_prices = [it['price_num'] for it in dba_items if it.get('price_num') is not None]
|
||
|
||
if not new_prices or not used_prices:
|
||
return {
|
||
'new_price': None, 'used_price': None, 'used_median': None,
|
||
'savings': None, 'savings_pct': None,
|
||
'verdict': 'no_data',
|
||
'reason': 'Not enough price data to compare. Try a more specific search term.'
|
||
}
|
||
|
||
new_price = min(new_prices)
|
||
used_median = statistics.median(used_prices)
|
||
used_low = min(used_prices)
|
||
used_high = max(used_prices)
|
||
|
||
savings = new_price - used_median
|
||
savings_pct = (savings / new_price * 100) if new_price > 0 else 0
|
||
|
||
# Decide the verdict
|
||
if savings_pct >= 40:
|
||
verdict = 'buy_used'
|
||
reason = (f"Used is {savings_pct:.0f}% cheaper than new "
|
||
f"(~{used_median:,.0f} kr vs {new_price:,.0f} kr new). "
|
||
f"Great deal — buy used!")
|
||
elif savings_pct >= 20:
|
||
verdict = 'buy_used'
|
||
reason = (f"Used saves you {savings_pct:.0f}% "
|
||
f"(~{used_median:,.0f} kr vs {new_price:,.0f} kr new). "
|
||
f"Solid savings, worth considering used.")
|
||
elif savings_pct >= 5:
|
||
verdict = 'mixed'
|
||
reason = (f"Used is only {savings_pct:.0f}% cheaper "
|
||
f"(~{used_median:,.0f} kr vs {new_price:,.0f} kr new). "
|
||
f"Decent savings but you lose warranty & returns.")
|
||
elif savings_pct > 0:
|
||
verdict = 'buy_new'
|
||
reason = (f"Used barely saves anything ({savings_pct:.0f}%) "
|
||
f"(~{used_median:,.0f} kr vs {new_price:,.0f} kr new). "
|
||
f"New comes with warranty — probably worth the small premium.")
|
||
else:
|
||
verdict = 'wait'
|
||
reason = (f"Used prices ({used_low:,.0f}–{used_high:,.0f} kr) are at or above "
|
||
f"new price ({new_price:,.0f} kr). Something's off — wait for better deals.")
|
||
|
||
return {
|
||
'new_price': new_price,
|
||
'used_price': used_median,
|
||
'used_median': used_median,
|
||
'used_low': used_low,
|
||
'used_high': used_high,
|
||
'savings': savings,
|
||
'savings_pct': savings_pct,
|
||
'verdict': verdict,
|
||
'reason': reason,
|
||
}
|
||
|
||
|
||
def format_verdict(v, fmt='text'):
|
||
"""Format the verdict dict for display."""
|
||
if v['verdict'] == 'no_data':
|
||
return v['reason']
|
||
|
||
icon = {
|
||
'buy_used': '✅ BUY USED',
|
||
'buy_new': '🏷️ BUY NEW',
|
||
'mixed': '🤔 DECENT DEAL',
|
||
'wait': '⏳ WAIT',
|
||
}.get(v['verdict'], '❓')
|
||
|
||
if fmt == 'markdown':
|
||
lines = [
|
||
f"**{icon}** — {v['reason']}",
|
||
f"| New (PriceRunner) | Used (DBA median) | You save |",
|
||
f"|---|---|---|",
|
||
f"| {v['new_price']:,.0f} kr | {v['used_median']:,.0f} kr | {v['savings']:,.0f} kr ({v['savings_pct']:.0f}%) |",
|
||
]
|
||
return '\n'.join(lines)
|
||
|
||
# text / grid
|
||
bar_len = 30
|
||
ratio = min(v['savings_pct'] / 100, 1.0) if v['savings_pct'] > 0 else 0
|
||
bar = '█' * int(ratio * bar_len) + '░' * (bar_len - int(ratio * bar_len))
|
||
lines = [
|
||
f" {icon}",
|
||
f" {v['reason']}",
|
||
f" savings: [{bar}] {v['savings_pct']:.0f}%",
|
||
f" new: {v['new_price']:,.0f} kr used median: {v['used_median']:,.0f} kr "
|
||
f"range: {v['used_low']:,.0f}–{v['used_high']:,.0f} kr",
|
||
]
|
||
return '\n'.join(lines)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────
|
||
# Output formatting
|
||
# ──────────────────────────────────────────────────────────────
|
||
|
||
def sort_items_by_price(items):
|
||
def keyfn(it):
|
||
v = it.get('price_num')
|
||
return v if v is not None else float('inf')
|
||
return sorted(items, key=keyfn)
|
||
|
||
|
||
def _short(s, n=60):
|
||
if not s:
|
||
return ''
|
||
s = ' '.join(s.split())
|
||
return s if len(s) <= n else s[:n-3] + '...'
|
||
|
||
|
||
def print_top_results(items, n=5):
|
||
print(f"Top {n} DBA results (title — price — location):\n")
|
||
for it in items[:n]:
|
||
loc = it.get('location') or ''
|
||
dist = it.get('distance_km')
|
||
if dist is not None:
|
||
loc = f"{loc} ({dist:.0f} km)"
|
||
print(f"- {it.get('title','')}")
|
||
print(f" price: {it.get('price','')}\n location: {loc}\n url: {it.get('url','')}\n")
|
||
|
||
|
||
def print_comparison_table(dba_items, pr_items, n=5):
|
||
"""Print a simple comparison table between DBA and PriceRunner results."""
|
||
rows = []
|
||
maxrows = max(len(dba_items), len(pr_items), n)
|
||
for i in range(maxrows):
|
||
left = dba_items[i] if i < len(dba_items) else None
|
||
right = pr_items[i] if i < len(pr_items) else None
|
||
loc = left.get('location','') if left else ''
|
||
dist = left.get('distance_km') if left else None
|
||
if dist is not None:
|
||
loc = f"{loc} ({dist:.0f} km)"
|
||
rows.append((
|
||
str(i+1),
|
||
_short(left.get('title','')) if left else '',
|
||
left.get('price','') if left else '',
|
||
loc,
|
||
_short(right.get('title','')) if right else '',
|
||
right.get('price','') if right else '',
|
||
))
|
||
widths = [3, 50, 12, 18, 50, 12]
|
||
hdr = ('#', 'DBA title', 'DBA price', 'DBA loc', 'PriceRunner title', 'PR price')
|
||
sep = ' | '
|
||
def fmt(row):
|
||
return sep.join(row[i].ljust(widths[i]) for i in range(len(row)))
|
||
print('\nComparison table (DBA vs PriceRunner):')
|
||
print(fmt(hdr))
|
||
print('-' * (sum(widths) + len(sep) * (len(widths)-1)))
|
||
for r in rows[:n]:
|
||
print(fmt(r))
|
||
print()
|
||
|
||
|
||
def print_comparison_markdown(dba_items, pr_items, n=10):
|
||
"""Print a markdown table with two columns: DBA and PriceRunner."""
|
||
lines = []
|
||
lines.append("| DBA (used) | PriceRunner (new) |")
|
||
lines.append("|-----|------------|")
|
||
|
||
def cell(it):
|
||
if not it:
|
||
return ''
|
||
title = _short(it.get('title',''), 80)
|
||
url = it.get('url','') or ''
|
||
if url:
|
||
title_md = f"[{title}]({url})"
|
||
else:
|
||
title_md = title
|
||
price = it.get('price','')
|
||
loc = it.get('location','')
|
||
dist = it.get('distance_km')
|
||
if dist is not None:
|
||
loc = f"{loc} ({dist:.0f} km)"
|
||
parts = [title_md, price]
|
||
if loc:
|
||
parts.append(loc)
|
||
return '<br>'.join(p for p in parts if p)
|
||
|
||
for i in range(n):
|
||
left = dba_items[i] if i < len(dba_items) else None
|
||
right = pr_items[i] if i < len(pr_items) else None
|
||
lines.append(f"| {cell(left)} | {cell(right)} |")
|
||
|
||
print('\n'.join(lines))
|
||
print()
|
||
|
||
|
||
def print_comparison_grid(dba_items, pr_items, n=10):
|
||
"""Print a simple ASCII grid with two columns: DBA and PriceRunner."""
|
||
left_col = []
|
||
right_col = []
|
||
for i in range(n):
|
||
l = dba_items[i] if i < len(dba_items) else None
|
||
r = pr_items[i] if i < len(pr_items) else None
|
||
def cell_lines(it):
|
||
if not it:
|
||
return ['']
|
||
title = it.get('title','')
|
||
price = it.get('price','')
|
||
loc = it.get('location','')
|
||
dist = it.get('distance_km')
|
||
if dist is not None:
|
||
loc = f"{loc} ({dist:.0f} km)"
|
||
url = it.get('url','') or ''
|
||
if url and len(url) > 80:
|
||
url = url[:77] + '...'
|
||
lines = [title, price]
|
||
if loc:
|
||
lines.append(loc)
|
||
if url:
|
||
lines.append(url)
|
||
return lines
|
||
left_col.append(cell_lines(l))
|
||
right_col.append(cell_lines(r))
|
||
LEFT_MAX = 60
|
||
RIGHT_MAX = 80
|
||
|
||
def wrap_lines(block, width):
|
||
import textwrap
|
||
wrapped = []
|
||
for line in block:
|
||
if not line:
|
||
wrapped.append('')
|
||
else:
|
||
for w in textwrap.wrap(line, width=width) or ['']:
|
||
wrapped.append(w)
|
||
return wrapped
|
||
|
||
left_col = [wrap_lines(b, LEFT_MAX) for b in left_col]
|
||
right_col = [wrap_lines(b, RIGHT_MAX) for b in right_col]
|
||
|
||
left_w = min(max((len(line) for block in left_col for line in block), default=10), LEFT_MAX)
|
||
right_w = min(max((len(line) for block in right_col for line in block), default=10), RIGHT_MAX)
|
||
sep = ' | '
|
||
hor = '+' + '-'*(left_w+2) + '+' + '-'*(right_w+2) + '+'
|
||
for idx in range(n):
|
||
lblock = left_col[idx]
|
||
rblock = right_col[idx]
|
||
maxlines = max(len(lblock), len(rblock))
|
||
print(hor)
|
||
for i in range(maxlines):
|
||
lline = lblock[i] if i < len(lblock) else ''
|
||
rline = rblock[i] if i < len(rblock) else ''
|
||
print(f"| {lline.ljust(left_w)} | {rline.ljust(right_w)} |")
|
||
print(hor)
|
||
print()
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────
|
||
# Main
|
||
# ──────────────────────────────────────────────────────────────
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description='DBA vs PriceRunner price comparator — "should I buy it used?"'
|
||
)
|
||
parser.add_argument('query', nargs='+')
|
||
parser.add_argument('--engine', choices=['requests', 'selenium'], default='requests')
|
||
parser.add_argument('--max', type=int, default=15)
|
||
parser.add_argument('--json', action='store_true')
|
||
parser.add_argument('--top', type=int, default=0,
|
||
help='Print top N DBA results with price and location')
|
||
parser.add_argument('--pricerunner', action='store_true',
|
||
help='Fetch PriceRunner search results as well')
|
||
parser.add_argument('--compare', action='store_true',
|
||
help='Print a comparison table of DBA vs PriceRunner (uses --pricerunner)')
|
||
parser.add_argument('--format', choices=['text', 'markdown', 'grid'], default='text',
|
||
help='Output format for comparison table')
|
||
parser.add_argument('--verdict', action='store_true',
|
||
help='Add a "should I buy it used?" recommendation')
|
||
parser.add_argument('--min-price', type=str, default=None,
|
||
help='Filter out items below this price (e.g. 500 or "3.000")')
|
||
parser.add_argument('--max-price', type=str, default=None,
|
||
help='Filter out items above this price')
|
||
parser.add_argument('--from', dest='origin', type=str, default=None,
|
||
help='Your location for distance calculation (e.g. "Farum")')
|
||
args = parser.parse_args()
|
||
query = ' '.join(args.query)
|
||
print(f"Searching for: {query}\n")
|
||
|
||
# ── Fetch DBA ──
|
||
try:
|
||
if args.engine == 'requests':
|
||
dba = search_dba_requests(query, max_results=args.max, origin=args.origin)
|
||
else:
|
||
dba = search_dba_selenium(query, max_results=args.max, origin=args.origin)
|
||
except ScraperError as e:
|
||
print("Error while scraping DBA:", e)
|
||
sys.exit(1)
|
||
|
||
# ── Fetch PriceRunner ──
|
||
pr = []
|
||
if args.pricerunner or args.compare or args.verdict:
|
||
try:
|
||
pr = search_pricerunner_requests(query, max_results=args.max)
|
||
except ScraperError as e:
|
||
print('Error while scraping PriceRunner:', e)
|
||
|
||
# ── Price filters ──
|
||
def _parse_price_arg(s):
|
||
if s is None:
|
||
return None
|
||
v = normalize_price(s)
|
||
if v is None:
|
||
try:
|
||
s2 = s.replace('.', '').replace(',', '.')
|
||
return float(''.join(ch for ch in s2 if (ch.isdigit() or ch == '.')))
|
||
except Exception:
|
||
return None
|
||
return v
|
||
|
||
minp = _parse_price_arg(args.min_price)
|
||
maxp = _parse_price_arg(args.max_price)
|
||
|
||
if minp is not None or maxp is not None:
|
||
def in_range(it):
|
||
pn = it.get('price_num')
|
||
if pn is None:
|
||
return False
|
||
if minp is not None and pn < minp:
|
||
return False
|
||
if maxp is not None and pn > maxp:
|
||
return False
|
||
return True
|
||
before_d = len(dba)
|
||
before_p = len(pr)
|
||
dba = [it for it in dba if in_range(it)]
|
||
pr = [it for it in pr if in_range(it)]
|
||
print(f"Applied price filter: min={minp} max={maxp}. DBA: {before_d}->{len(dba)}, PR: {before_p}->{len(pr)}\n")
|
||
|
||
print(f"Found {len(dba)} items on DBA\n")
|
||
|
||
# ── JSON output ──
|
||
if args.json:
|
||
print(json.dumps(dba, ensure_ascii=False, indent=2))
|
||
return
|
||
|
||
# ── Top-N DBA only ──
|
||
if args.top and args.top > 0 and not args.compare:
|
||
print_top_results(dba, n=args.top)
|
||
return
|
||
|
||
# ── PriceRunner only (no compare) ──
|
||
if args.pricerunner and not args.compare and not args.verdict:
|
||
print(f"Found {len(pr)} items on PriceRunner\n")
|
||
if args.json:
|
||
print(json.dumps(pr, ensure_ascii=False, indent=2))
|
||
return
|
||
|
||
# ── Verdict ──
|
||
if args.verdict:
|
||
v = compute_verdict(dba, pr)
|
||
print(f"\n{'='*50}")
|
||
print(f" SHOULD I BUY IT USED? — {query}")
|
||
print(f"{'='*50}")
|
||
print(format_verdict(v, fmt=args.format))
|
||
print(f"{'='*50}\n")
|
||
|
||
# ── Comparison table ──
|
||
if args.compare:
|
||
d_sorted = sort_items_by_price(dba)
|
||
p_sorted = sort_items_by_price(pr)
|
||
n = args.top if args.top and args.top > 0 else 10
|
||
if args.format == 'markdown':
|
||
print_comparison_markdown(d_sorted, p_sorted, n=n)
|
||
elif args.format == 'grid':
|
||
print_comparison_grid(d_sorted, p_sorted, n=n)
|
||
else:
|
||
print_comparison_table(d_sorted, p_sorted, n=n)
|
||
return
|
||
|
||
# ── Top-N DBA only (no compare, no verdict) ──
|
||
if args.top and args.top > 0:
|
||
print_top_results(dba, n=args.top)
|
||
return
|
||
|
||
# ── Default: sorted DBA list ──
|
||
dba_sorted = sort_items_by_price(dba)
|
||
print(f"DBA — top {min(len(dba_sorted), args.max)} by price:")
|
||
for it in dba_sorted[:args.max]:
|
||
loc = it.get('location','')
|
||
dist = it.get('distance_km')
|
||
if dist is not None:
|
||
loc = f"{loc} ({dist:.0f} km)"
|
||
print(f"- {it.get('title','')} — {it.get('price','')} — {loc} — {it.get('url','')}")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|