Hermes Agent
276 lines
9.9 KiB
Python
276 lines
9.9 KiB
Python
#!/usr/bin/env python3
|
||
"""Fetch Amazon board game deals and compare with brettspielpreise.de price histories.
|
||
Final version with validated matches."""
|
||
|
||
import re, json, time, urllib.request, urllib.parse, ssl
|
||
|
||
# ---- Step 1: Parse HTML ----
|
||
html_path = "/home/hermes/workspace/amazon_deals_2026-06-05.html"
|
||
with open(html_path, "r", encoding="utf-8") as f:
|
||
html = f.read()
|
||
|
||
deal_blocks = re.findall(
|
||
r'<div style="display:flex;gap:1rem;padding:1rem 0;border-bottom:1px solid #eee;">'
|
||
r'(.*?)</div></div>',
|
||
html, re.DOTALL
|
||
)
|
||
print(f"Found {len(deal_blocks)} deal blocks")
|
||
|
||
deals = []
|
||
for block in deal_blocks:
|
||
name_match = re.search(r'<strong>(.*?)</strong>', block)
|
||
name = name_match.group(1).strip() if name_match else "Unknown"
|
||
price_match = re.search(r'font-weight:bold;">(\d+[.,]\d+) €', block)
|
||
price = float(price_match.group(1).replace(",", ".")) if price_match else None
|
||
uvp_match = re.search(r'UVP (\d+[.,]\d+) €', block)
|
||
uvp = float(uvp_match.group(1).replace(",", ".")) if uvp_match else None
|
||
asin_match = re.search(r'/dp/([A-Z0-9]+)', block)
|
||
asin = asin_match.group(1) if asin_match else None
|
||
url_match = re.search(r'href="(https://www\.amazon\.de/dp/[^"]+)"', block)
|
||
url = url_match.group(1) if url_match else f"https://www.amazon.de/dp/{asin}?tag=60pro05-21"
|
||
deals.append({"name": name, "amazon_price": price, "uvp": uvp, "asin": asin, "url": url})
|
||
|
||
# ---- Step 2: Filter ----
|
||
FILTER_KEYWORDS = [
|
||
"schwarzer peter", "skat", "rommé", "romme", "bridge canasta",
|
||
"top trumps", "wetfussballstars",
|
||
"uno ", "uno flip", "uno show", "uno teams", "phase 10",
|
||
]
|
||
|
||
filtered = []
|
||
for d in deals:
|
||
name_lower = d["name"].lower()
|
||
if not any(kw in name_lower for kw in FILTER_KEYWORDS):
|
||
filtered.append(d)
|
||
else:
|
||
kw = next(kw for kw in FILTER_KEYWORDS if kw in name_lower)
|
||
print(f" SKIP: {d['name'][:70]}... [{kw}]")
|
||
|
||
print(f"After filtering: {len(filtered)} deals")
|
||
|
||
# ---- Step 3: HTTP helpers ----
|
||
ctx = ssl.create_default_context()
|
||
ctx.check_hostname = False
|
||
ctx.verify_mode = ssl.CERT_NONE
|
||
|
||
def http_get(url):
|
||
req = urllib.request.Request(url, headers={
|
||
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36",
|
||
})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=15, context=ctx) as resp:
|
||
return resp.read().decode("utf-8", errors="replace")
|
||
except Exception:
|
||
return None
|
||
|
||
# ---- Search terms with specific overrides ----
|
||
SEARCH_CONFIG = [
|
||
# (amazon_name_keyword, search_term)
|
||
("Maus reiß aus", "Maus reiß aus"),
|
||
("Die Verräter", "Die Verräter Brettspiel"),
|
||
("Jumanji", "Jumanji Familienspiel"),
|
||
("Monsterjäger", "Monsterjäger Schmidt"),
|
||
("Sweet Takes", "Sweet Takes"),
|
||
("My City", "My City Kosmos"),
|
||
("Evacuation", "Evacuation"),
|
||
("STILLE POST EXTREM", "Stille Post Extrem"),
|
||
("Dog", "Dog Royal"),
|
||
("THE MIND", "The Mind"),
|
||
("Catan", "Catan"),
|
||
("Leben in Reterra", "Leben in Reterra"),
|
||
("Sagaland", "Sagaland"),
|
||
("Hitster", "Hitster"),
|
||
("Outsmarted", "Outsmarted"),
|
||
("Monopoly Harry Potter", "Monopoly Harry Potter"),
|
||
("Die Magischen Schlüssel", "Die Magischen Schlüssel"),
|
||
("Scrabble Zwei in Eins", "Scrabble"),
|
||
("Das verrückte Labyrinth", "Das verrückte Labyrinth"),
|
||
("Funkelschatz", "Funkelschatz"),
|
||
("Leiterspiel", "Leiterspiel Schmidt"),
|
||
("Ligretto", "Ligretto"),
|
||
]
|
||
|
||
FALLBACK_SEARCH = {
|
||
"Die Verräter": ["the traitors die verrater"],
|
||
"Dog": ["Dog Den letzten beissen die Hunde"],
|
||
"Sweet Takes": ["Sweet Takes Hasbro"],
|
||
"Die Magischen Schlüssel": ["Magischen Schlüssel Game Factory"],
|
||
}
|
||
|
||
def get_search_terms(name):
|
||
terms = []
|
||
for key, val in SEARCH_CONFIG:
|
||
if key.lower() in name.lower():
|
||
terms.append(val)
|
||
break
|
||
if not terms:
|
||
words = name.replace("-", " ").replace("–", " ").replace(",", " ").split()
|
||
stop = {"das", "die", "der", "den", "für", "und", "mit", "von", "ein", "eine", "einen", "ab", "des", "dem"}
|
||
meaningful = [w for w in words if w.lower() not in stop and len(w) > 1 and not w.isdigit()]
|
||
terms.append(" ".join(meaningful[:3]))
|
||
|
||
for key, fbs in FALLBACK_SEARCH.items():
|
||
if key.lower() in name.lower():
|
||
for fb in fbs:
|
||
if fb not in terms:
|
||
terms.append(fb)
|
||
return terms
|
||
|
||
def search_brettspielpreise(search_term):
|
||
url = f"https://brettspielpreise.de/item/search?search={urllib.parse.quote(search_term)}"
|
||
body = http_get(url)
|
||
if not body:
|
||
return []
|
||
show_links = re.findall(r'/item/show/(\d+)/([^"]+)', body)
|
||
items, seen = [], set()
|
||
for item_id, slug in show_links:
|
||
if item_id not in seen:
|
||
seen.add(item_id)
|
||
history_url = f"https://brettspielpreise.de/item/history/{item_id}/{slug}"
|
||
items.append((item_id, urllib.parse.unquote(slug), history_url))
|
||
return items[:5]
|
||
|
||
def get_price_history(history_url):
|
||
body = http_get(history_url)
|
||
if not body:
|
||
return None
|
||
idx = body.find('datasets')
|
||
if idx == -1:
|
||
return None
|
||
s = body.rfind('<script', 0, idx)
|
||
e = body.find('</script>', idx)
|
||
if s == -1 or e == -1:
|
||
return None
|
||
script = body[s:e]
|
||
all_data, pos = [], 0
|
||
while True:
|
||
di = script.find('data: [', pos)
|
||
if di == -1:
|
||
break
|
||
start = di + 6
|
||
depth, end = 0, start
|
||
for i in range(start, len(script)):
|
||
if script[i] == '[':
|
||
depth += 1
|
||
elif script[i] == ']':
|
||
depth -= 1
|
||
if depth == 0:
|
||
end = i + 1
|
||
break
|
||
try:
|
||
pts = json.loads(script[start:end])
|
||
all_data.append(pts)
|
||
except json.JSONDecodeError:
|
||
pass
|
||
pos = end
|
||
for pts in all_data:
|
||
if pts and isinstance(pts, list) and len(pts) > 0 and isinstance(pts[0], dict) and "y" in pts[0]:
|
||
prices = []
|
||
for pt in pts:
|
||
try:
|
||
p = float(pt.get("y", 0))
|
||
if p > 0:
|
||
prices.append(p)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
if len(prices) >= 3:
|
||
return prices
|
||
return None
|
||
|
||
def validate_match(current, low, high, datapoints):
|
||
"""Check if current price reasonably fits this game's history."""
|
||
if low <= 0:
|
||
return False
|
||
dist = (current - low) / low * 100
|
||
|
||
# For sparse data (<15 points), allow wider range (tracking may miss deals)
|
||
if datapoints < 15:
|
||
if dist < -65: # very sparse data could miss deals entirely
|
||
return False
|
||
else:
|
||
if dist < -35: # current shouldn't be way below tracked history
|
||
return False
|
||
|
||
if dist > 300: # way too expensive = wrong match
|
||
return False
|
||
|
||
return True
|
||
|
||
# ---- Step 4: Process ----
|
||
results = []
|
||
skipped_wrong_match = []
|
||
|
||
for i, d in enumerate(filtered):
|
||
print(f"\n[{i+1}/{len(filtered)}] {d['name'][:75]}")
|
||
print(f" Amazon: {d['amazon_price']}€ | UVP: {d['uvp']}€")
|
||
|
||
search_terms = get_search_terms(d["name"])
|
||
matched = False
|
||
|
||
for term in search_terms:
|
||
if matched:
|
||
break
|
||
candidates = search_brettspielpreise(term)
|
||
if not candidates:
|
||
continue
|
||
|
||
for c_idx, (item_id, item_name, history_url) in enumerate(candidates):
|
||
prices = get_price_history(history_url)
|
||
if not prices:
|
||
continue
|
||
|
||
low, high = round(min(prices), 2), round(max(prices), 2)
|
||
current = d["amazon_price"]
|
||
|
||
if not validate_match(current, low, high, len(prices)):
|
||
dist = (current - low) / low * 100
|
||
print(f" #{c_idx+1} ID {item_id} '{item_name[:45]}': low={low}€, dist={dist:.1f}% → SKIP")
|
||
continue
|
||
|
||
dist_pct = round((current - low) / low * 100, 1)
|
||
sav_pct = round((d["uvp"] - current) / d["uvp"] * 100, 1) if d["uvp"] and d["uvp"] > 0 else 0
|
||
|
||
print(f" ✓ MATCH: '{item_name[:50]}' (ID {item_id})")
|
||
print(f" History: {len(prices)}pts, low={low}€, high={high}€, current={current}€ ({dist_pct:+.1f}%)")
|
||
|
||
results.append({
|
||
"name": d["name"],
|
||
"amazon_price": current,
|
||
"uvp": d["uvp"],
|
||
"savings_pct": sav_pct,
|
||
"history_low": low,
|
||
"history_high": high,
|
||
"distance_from_low_pct": dist_pct,
|
||
"url": d["url"],
|
||
"history_url": history_url,
|
||
"datapoints": len(prices),
|
||
})
|
||
matched = True
|
||
break
|
||
|
||
if not matched:
|
||
print(f" ✗ No valid match on brettspielpreise.de")
|
||
|
||
# Sort: closer to best price first (lowest distance_from_low_pct)
|
||
results.sort(key=lambda x: x["distance_from_low_pct"])
|
||
|
||
print(f"\n{'='*70}")
|
||
print(f"RESULTS: {len(results)} games with price histories (sorted by proximity to best price)")
|
||
print(f"{'='*70}")
|
||
for i, r in enumerate(results):
|
||
star = "⭐" if r["distance_from_low_pct"] <= 10 else " "
|
||
print(f" {i+1:2d}. {star} {r['distance_from_low_pct']:>+6.1f}% | {r['amazon_price']:>7.2f}€ "
|
||
f"(low: {r['history_low']:>6.2f}€, high: {r['history_high']:>6.2f}€, {r['datapoints']}pts) "
|
||
f"| {r['name'][:50]}")
|
||
|
||
# Save JSON
|
||
output_path = "/tmp/amazon_near_bestprice.json"
|
||
with open(output_path, "w", encoding="utf-8") as f:
|
||
json.dump(results, f, ensure_ascii=False, indent=2)
|
||
print(f"\nSaved {len(results)} results to {output_path}")
|
||
|
||
# Summary stats
|
||
close = [r for r in results if r["distance_from_low_pct"] <= 15]
|
||
print(f"\n⭐ {len(close)} games within 15% of their historical best price")
|
||
print(f" {len([r for r in results if r['distance_from_low_pct'] <= 0])} games at or below historical best price (new lows!)")
|