Hermes Agent
775 lines
28 KiB
Python
775 lines
28 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Partner-Shop-Suche — finde Brettspiele in allen Affiliate-Shops.
|
||
Generiert HTML für den „Hier kaufen"-Abschnitt in Artikeln.
|
||
|
||
Shops:
|
||
- Spiele-Offensive.de (pid=505)
|
||
- Fantasywelt.de (bsn=ig1UxQ0L2cNMNslzFhyr)
|
||
- AllGames4You.de (bsn=1)
|
||
- Amazon.de (tag=60pro05-21 — via CreatorAPI)
|
||
- Milan-Spiele.de (Scraping-Übergangslösung)
|
||
- Thalia.de (AWIN 14158)
|
||
- Osiander.de (AWIN 30799)
|
||
- idee+spiel.de (AWIN 75052)
|
||
"""
|
||
|
||
import re
|
||
import sys
|
||
import os
|
||
import csv
|
||
import urllib.request
|
||
import urllib.parse
|
||
from html import escape, unescape
|
||
|
||
USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36"
|
||
|
||
# ---- Shop-Konfigurationen ----
|
||
|
||
SHOPS = {
|
||
"Spiele-Offensive.de": {
|
||
"search_url": "https://www.spiele-offensive.de/index.php?cmd=suchergebnis&suchwort={query}&pid=505",
|
||
"affix": "pid=505",
|
||
"scrape": True,
|
||
},
|
||
"Fantasywelt.de": {
|
||
"search_url": "https://www.fantasywelt.de/search?q={query}&bsn=ig1UxQ0L2cNMNslzFhyr",
|
||
"affix": "bsn=ig1UxQ0L2cNMNslzFhyr",
|
||
"scrape": True,
|
||
},
|
||
"AllGames4You.de": {
|
||
"search_url": "https://www.allgames4you.de/search?q={query}&bsn=1",
|
||
"affix": "bsn=1",
|
||
"scrape": False,
|
||
},
|
||
"Amazon.de": {
|
||
"search_url": "https://www.amazon.de/s?k={query}&tag=60pro05-21",
|
||
"affix": "tag=60pro05-21",
|
||
"scrape_api": True,
|
||
},
|
||
"Milan-Spiele.de": {
|
||
"search_url": "https://www.milan-spiele.de/advanced_search_result.php?keywords={query}",
|
||
"affix": None,
|
||
"scrape": True,
|
||
},
|
||
"Thalia.de": {
|
||
"search_url": "https://www.awin1.com/cread.php?awinmid=14158&awinaffid=661969&ued={encoded_url}",
|
||
"affix": None,
|
||
"scrape": False,
|
||
},
|
||
"Osiander.de": {
|
||
"search_url": "https://www.awin1.com/cread.php?awinmid=30799&awinaffid=661969&ued={encoded_url}",
|
||
"affix": None,
|
||
"scrape": False,
|
||
},
|
||
"idee+spiel.de": {
|
||
"search_url": "https://www.awin1.com/cread.php?awinmid=75052&awinaffid=661969&ued={encoded_url}",
|
||
"affix": None,
|
||
"scrape": False,
|
||
},
|
||
}
|
||
|
||
|
||
def fetch(url, timeout=10):
|
||
"""Fetch URL, return decoded text."""
|
||
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
raw = resp.read()
|
||
try:
|
||
return raw.decode("utf-8")
|
||
except UnicodeDecodeError:
|
||
return raw.decode("iso-8859-1")
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def clean_price(text):
|
||
"""Extract first EUR price from text."""
|
||
m = re.search(r"(\d+)[,.](\d{2})\s*(?:€|€|EUR)", text)
|
||
if m:
|
||
return f"{m.group(1)},{m.group(2)}"
|
||
# Try comma format
|
||
m = re.search(r"(\d+,\d{2})\s*(?:€|€|EUR)", text)
|
||
if m:
|
||
return m.group(1)
|
||
return None
|
||
|
||
|
||
def match_score(query, title):
|
||
"""Score how well a product title matches the search query (0-100).
|
||
|
||
Returns 0 if it's clearly not the right product.
|
||
Returns 100 for an exact match.
|
||
"""
|
||
q = query.lower().strip()
|
||
t = title.lower().strip()
|
||
|
||
# Exact match
|
||
if q == t:
|
||
return 100
|
||
|
||
# Query is fully contained in title as a word-boundary match
|
||
if re.search(r'\b' + re.escape(q) + r'\b', t):
|
||
t_clean = t.replace(q, '').strip(' -:,.|')
|
||
|
||
# If the remainder is generic qualifiers, it's the base game → high score
|
||
generic_terms = ['spiel des jahres', 'kennerspiel', 'kinderspiel',
|
||
'deutsche edition', 'de', 'basis', 'grundspiel',
|
||
'brettspiel', 'kartenspiel', 'gesellschaftsspiel']
|
||
t_clean_lower = t_clean.lower()
|
||
# Remove trailing year numbers (e.g. "spiel des jahres 2022" → "spiel des jahres")
|
||
t_clean_normalized = re.sub(r'\s+\d{4}$', '', t_clean_lower)
|
||
if not t_clean or any(t_clean_normalized == s for s in generic_terms):
|
||
return 98
|
||
|
||
# If remainder contains expansion/variant keywords → lower score
|
||
expansion_words = ['erweiterung', 'expansion', 'insert', 'würfelturm',
|
||
'deluxe', 'big box', 'sammler', 'collector']
|
||
if any(w in t_clean_lower for w in expansion_words):
|
||
return 75
|
||
|
||
# If remainder is a subtitle after colon (e.g. "Cascadia: Rolling Rivers")
|
||
# → likely a variant, not the base game
|
||
if ':' in t and t.index(':') >= len(q):
|
||
return 78 # Lower than 88 but still a match
|
||
|
||
# Generic good match (e.g. "Cascadia Junior")
|
||
return 88
|
||
|
||
# Title is fully contained in query (e.g. "Cascadia" query matches "Cascadia: Rolling Rivers")
|
||
if re.search(r'\b' + re.escape(t) + r'\b', q):
|
||
return 70
|
||
|
||
# Word overlap scoring
|
||
q_words = set(re.findall(r'\w+', q))
|
||
t_words = set(re.findall(r'\w+', t))
|
||
common = q_words & t_words
|
||
if not common:
|
||
return 0
|
||
|
||
# At least the main word(s) must match
|
||
score = len(common) / max(len(q_words), len(t_words)) * 50 + 25
|
||
return min(int(score), 85)
|
||
|
||
|
||
def search_spiele_offensive(game_name):
|
||
"""Search Spiele-Offensive and find the best matching result with price."""
|
||
query = urllib.parse.quote(game_name)
|
||
search_url = f"https://www.spiele-offensive.de/index.php?cmd=suchergebnis&suchwort={query}"
|
||
html = fetch(search_url, timeout=15)
|
||
if not html:
|
||
return None
|
||
|
||
# Split into article blocks (<li class="ala">)
|
||
blocks = re.split(r'<li\s+class="ala"', html)[1:]
|
||
if not blocks:
|
||
return None
|
||
|
||
# Parse each block
|
||
candidates = []
|
||
for block in blocks:
|
||
# Name from <font> tag
|
||
name_match = re.search(r'<font>([^<]+)</font>', block)
|
||
if not name_match:
|
||
continue
|
||
name = name_match.group(1).strip()
|
||
name = name.replace("ü", "ü").replace("ö", "ö").replace("ä", "ä").replace("ß", "ß")
|
||
|
||
# URL and article ID
|
||
url_match = re.search(r'href="(/Spiel/[^"]+-(\d+))\.html"', block)
|
||
if not url_match:
|
||
continue
|
||
url = url_match.group(1)
|
||
art_id = url_match.group(2)
|
||
|
||
# Price — strip HTML tags and find "nur X,XX €" or just "X,XX €"
|
||
text = re.sub(r'<[^>]+>', ' ', block)
|
||
text = re.sub(r'\s+', ' ', text)
|
||
|
||
# Try "nur X , XX" pattern (prices split across spans: "nur <span>33</span> , <span>99</span> €")
|
||
nur_match = re.search(r'nur\s+(\d+)\s*,\s*(\d{2})\s*(?:€|€)', text)
|
||
# Try "nur X,XX" (without span splitting)
|
||
if not nur_match:
|
||
nur_match = re.search(r'nur\s+(\d+,\d{2})\s*(?:€|€)', text)
|
||
# Fallback: any price
|
||
price_match = re.search(r'(\d+)\s*,\s*(\d{2})\s*(?:€|€)', text)
|
||
|
||
price = None
|
||
if nur_match:
|
||
price = f"{nur_match.group(1)},{nur_match.group(2)}"
|
||
elif price_match:
|
||
price = f"{price_match.group(1)},{price_match.group(2)}"
|
||
|
||
# Score against query
|
||
score = match_score(game_name, name)
|
||
if score >= 60: # Minimum confidence threshold
|
||
candidates.append({
|
||
"name": name, "art_id": art_id, "price": price,
|
||
"url": url, "score": score,
|
||
})
|
||
|
||
if not candidates:
|
||
# No good match — return search link
|
||
return {
|
||
"url": f"https://www.spiele-offensive.de/index.php?cmd=suchergebnis&suchwort={query}&pid=505",
|
||
"price": None,
|
||
"shop": "Spiele-Offensive.de",
|
||
}
|
||
|
||
# Pick highest-scoring match
|
||
best = max(candidates, key=lambda c: (c["score"], 1 if c["price"] else 0))
|
||
|
||
aff_url = f"https://www.spiele-offensive.de/index.php?cmd=artikel_anzeigen&aid={best['art_id']}&pid=505"
|
||
|
||
return {"url": aff_url, "price": best["price"], "shop": "Spiele-Offensive.de"}
|
||
|
||
|
||
def _is_english_version(product_name):
|
||
"""Detect if a product name indicates an English-language version."""
|
||
name = product_name.lower()
|
||
# Standalone "EN" as a word (not part of another word like "Catan")
|
||
if re.search(r'\ben\b', name):
|
||
return True
|
||
if 'engl.' in name or 'english' in name or 'englisch' in name:
|
||
return True
|
||
return False
|
||
|
||
|
||
def search_fantasywelt(game_name):
|
||
"""Search Fantasywelt.de via local CSV export for price."""
|
||
csv_path = "/tmp/fantasywelt_export.csv"
|
||
|
||
# First run: download CSV if not present or older than 24h
|
||
import os
|
||
if not os.path.exists(csv_path) or (os.path.getmtime(csv_path) < __import__("time").time() - 86400):
|
||
print(" Lade CSV-Export...", file=sys.stderr, flush=True)
|
||
csv_html = fetch("http://www.fantasywelt.de/export/brettspiele.csv", timeout=30)
|
||
if csv_html and csv_html.startswith("Name;"):
|
||
with open(csv_path, "w") as f:
|
||
f.write(csv_html)
|
||
|
||
if not os.path.exists(csv_path):
|
||
query = urllib.parse.quote(game_name)
|
||
return {
|
||
"url": f"https://www.fantasywelt.de/search?q={query}&bsn=ig1UxQ0L2cNMNslzFhyr",
|
||
"price": None,
|
||
"shop": "Fantasywelt.de",
|
||
}
|
||
|
||
# Search CSV
|
||
import csv
|
||
game_lower = game_name.lower().strip()
|
||
matches = []
|
||
with open(csv_path, newline="", encoding="utf-8") as f:
|
||
reader = csv.DictReader(f, delimiter=";")
|
||
for row in reader:
|
||
name = (row.get("Name") or "").lower()
|
||
if game_lower in name or name in game_lower:
|
||
matches.append(row)
|
||
elif len(matches) < 5:
|
||
# Fuzzy: check if a significant portion of words match
|
||
game_words = set(game_lower.split())
|
||
name_words = set(name.replace(",", "").replace("(", "").replace(")", "").replace(":", "").split())
|
||
common = game_words & name_words
|
||
if len(common) >= min(2, len(game_words)):
|
||
matches.append(row)
|
||
|
||
if not matches:
|
||
query = urllib.parse.quote(game_name)
|
||
return {
|
||
"url": f"https://www.fantasywelt.de/search?q={query}&bsn=ig1UxQ0L2cNMNslzFhyr",
|
||
"price": None,
|
||
"shop": "Fantasywelt.de",
|
||
}
|
||
|
||
# Best match: prefer "auf Lager" over others
|
||
matches.sort(key=lambda r: 0 if "auf Lager" in (r.get("Verfügbarkeit") or "") else 1)
|
||
best = matches[0]
|
||
|
||
price = best.get("VK/Sonderpreis", "").replace(".", ",")
|
||
url = (best.get("Produkt-URL") or "").strip()
|
||
if not url:
|
||
query = urllib.parse.quote(game_name)
|
||
url = f"https://www.fantasywelt.de/search?q={query}"
|
||
|
||
# Append affiliate bsn
|
||
sep = "&" if "?" in url else "?"
|
||
aff_url = f"{url}{sep}bsn=ig1UxQ0L2cNMNslzFhyr"
|
||
|
||
stock = best.get("Verfügbarkeit", "")
|
||
|
||
# Detect English version
|
||
name = (best.get("Name") or "").strip()
|
||
extra = ""
|
||
if _is_english_version(name):
|
||
extra = " (englische Version)"
|
||
|
||
return {"url": aff_url, "price": price, "shop": f"Fantasywelt.de ({stock}){extra}"}
|
||
|
||
|
||
def search_amazon(game_name):
|
||
"""Search Amazon via CreatorAPI for price."""
|
||
try:
|
||
from amazon_creatorsapi import AmazonCreatorsApi, Country
|
||
from amazon_creatorsapi import models as m
|
||
|
||
api = AmazonCreatorsApi(
|
||
"amzn1.application-oa2-client.91799968d0744e66950ea413659197d4",
|
||
"amzn1.oa2-cs.v1.4c41154ecfd1835e00879ad5c8060ef5cd3b209b06077c0af1d868b343cab955",
|
||
"3.2", "60pro05-21", country=Country.DE,
|
||
)
|
||
|
||
resp = api.search_items(
|
||
keywords=game_name,
|
||
browse_node_id="1290250031",
|
||
item_count=3,
|
||
resources=[
|
||
m.SearchItemsResource.ITEM_INFO_DOT_TITLE,
|
||
m.SearchItemsResource.OFFERS_V2_DOT_LISTINGS_DOT_PRICE,
|
||
],
|
||
)
|
||
|
||
if not resp or not resp.items:
|
||
return None
|
||
|
||
item = resp.items[0]
|
||
asin = str(item.asin)
|
||
|
||
price = None
|
||
if item.offers_v2 and item.offers_v2.listings:
|
||
listing = item.offers_v2.listings[0]
|
||
if listing.price and listing.price.money:
|
||
price = f"{float(listing.price.money.amount):.2f}".replace(".", ",")
|
||
|
||
return {
|
||
"url": f"https://www.amazon.de/dp/{asin}?tag=60pro05-21",
|
||
"price": price,
|
||
"shop": "Amazon.de",
|
||
}
|
||
except Exception:
|
||
# Fallback: search link
|
||
query = urllib.parse.quote(game_name)
|
||
return {
|
||
"url": f"https://www.amazon.de/s?k={query}&tag=60pro05-21",
|
||
"price": None,
|
||
"shop": "Amazon.de",
|
||
}
|
||
|
||
|
||
def search_allgames4you(game_name):
|
||
"""AllGames4You.de — via Angebote-CSV-Export mit Preis & Verfügbarkeit."""
|
||
csv_path = "/tmp/allgames4you_angebote.csv"
|
||
csv_url = "https://www.allgames4you.de/export/brettspielangebote.csv"
|
||
|
||
import os
|
||
if not os.path.exists(csv_path) or (os.path.getmtime(csv_path) < __import__("time").time() - 86400):
|
||
csv_html = fetch(csv_url, timeout=30)
|
||
if csv_html and csv_html.startswith('"BARCODE"'):
|
||
with open(csv_path, "w") as f:
|
||
f.write(csv_html)
|
||
|
||
if not os.path.exists(csv_path):
|
||
query = urllib.parse.quote(game_name)
|
||
return {
|
||
"url": f"https://www.allgames4you.de/search?q={query}&bsn=1",
|
||
"price": None,
|
||
"shop": "AllGames4You.de",
|
||
}
|
||
|
||
# Search CSV with match_score()
|
||
candidates = []
|
||
with open(csv_path, newline="", encoding="utf-8") as f:
|
||
reader = csv.DictReader(f, delimiter=";")
|
||
for row in reader:
|
||
name = (row.get("BEZEICHNUNG") or "").strip()
|
||
if not name:
|
||
continue
|
||
score = match_score(game_name, name)
|
||
if score < 60:
|
||
continue
|
||
|
||
price_raw = row.get("VKBRUTTO", "").replace(",", ".")
|
||
try:
|
||
price = f"{float(price_raw):.2f}".replace(".", ",")
|
||
except ValueError:
|
||
price = None
|
||
|
||
url = (row.get("URL") or "").strip()
|
||
in_stock = (row.get("LIEFERBAR") or "").upper() == "Y"
|
||
|
||
candidates.append({
|
||
"name": name, "url": url, "price": price,
|
||
"score": score, "in_stock": in_stock,
|
||
})
|
||
|
||
if not candidates:
|
||
query = urllib.parse.quote(game_name)
|
||
return {
|
||
"url": f"https://www.allgames4you.de/search?q={query}&bsn=1",
|
||
"price": None,
|
||
"shop": "AllGames4You.de",
|
||
}
|
||
|
||
# Sort: score desc, then in_stock first, then has_price
|
||
candidates.sort(key=lambda c: (-c["score"], not c["in_stock"], 0 if c["price"] else 1))
|
||
best = candidates[0]
|
||
|
||
sep = "&" if "?" in (best["url"] or "") else "?"
|
||
aff_url = f"{best['url']}{sep}bsn=1"
|
||
|
||
stock = "auf Lager" if best["in_stock"] else "nicht verfügbar"
|
||
|
||
# Build shop label — add product name clarification when score < 100
|
||
shop_label = f"AllGames4You.de ({stock})"
|
||
if best["score"] < 100:
|
||
# Show the actual matched product name for transparency
|
||
shop_label += f" – {best['name']}"
|
||
|
||
return {"url": aff_url, "price": best["price"], "shop": shop_label}
|
||
|
||
|
||
def search_milan_spiele(game_name):
|
||
"""Milan-Spiele — scrape product listing for price and availability."""
|
||
query = urllib.parse.quote(game_name)
|
||
search_url = f"https://www.milan-spiele.de/advanced_search_result.php?keywords={query}"
|
||
html = fetch(search_url, timeout=15)
|
||
if not html:
|
||
return {
|
||
"url": f"https://www.milan-spiele.de/advanced_search_result.php?keywords={query}",
|
||
"price": None,
|
||
"shop": "Milan-Spiele.de",
|
||
}
|
||
|
||
# Parse product blocks: <div class="product"> ... </div>
|
||
blocks = re.split(r'<div class="product">', html)[1:]
|
||
if not blocks:
|
||
return {
|
||
"url": f"https://www.milan-spiele.de/advanced_search_result.php?keywords={query}",
|
||
"price": None,
|
||
"shop": "Milan-Spiele.de",
|
||
}
|
||
|
||
candidates = []
|
||
for block in blocks:
|
||
# Extract product URL and name from <span class="title"><a ...>
|
||
title_match = re.search(
|
||
r'<span class="title">.*?<a[^>]*href="([^"]*)"[^>]*><nobr>([^<]+)</nobr>',
|
||
block, re.DOTALL
|
||
)
|
||
if not title_match:
|
||
continue
|
||
prod_url = title_match.group(1)
|
||
name = title_match.group(2).strip()
|
||
|
||
# Extract price from <div class="priceBox"><span class="price"><strong><nobr> PRICE €
|
||
price_match = re.search(
|
||
r'<span class="price">\s*<strong><nobr>\s*([\d,.]+)\s*€',
|
||
block
|
||
)
|
||
price = price_match.group(1) if price_match else None
|
||
|
||
# Extract availability
|
||
avail_match = re.search(
|
||
r'<span[^>]*class="delivery\s+(\w+)"[^>]*><nobr>([^<]+)</nobr>',
|
||
block
|
||
)
|
||
availability = avail_match.group(2).strip() if avail_match else ""
|
||
|
||
# Score match
|
||
score = match_score(game_name, name)
|
||
if score >= 60:
|
||
candidates.append({
|
||
"name": name, "url": prod_url, "price": price,
|
||
"score": score, "availability": availability,
|
||
})
|
||
|
||
if not candidates:
|
||
return {
|
||
"url": f"https://www.milan-spiele.de/advanced_search_result.php?keywords={query}",
|
||
"price": None,
|
||
"shop": "Milan-Spiele.de",
|
||
}
|
||
|
||
# Prefer: high score first, then in stock, then has price
|
||
def sort_key(c):
|
||
in_stock = 0 if "lieferbar" in c.get("availability", "").lower() else 1
|
||
has_price = 0 if c.get("price") else 1
|
||
return (-c["score"], in_stock, has_price)
|
||
|
||
candidates.sort(key=sort_key)
|
||
best = candidates[0]
|
||
|
||
shop_label = "Milan-Spiele.de"
|
||
if best.get("availability"):
|
||
# Translate common availability phrases
|
||
avail = best["availability"]
|
||
if "lieferbar" in avail.lower():
|
||
if "sofort" in avail.lower():
|
||
avail = "auf Lager"
|
||
elif "nicht" in avail.lower():
|
||
avail = "nicht lieferbar"
|
||
shop_label = f"Milan-Spiele.de ({avail})"
|
||
|
||
# Detect English version
|
||
if _is_english_version(best["name"]):
|
||
shop_label += " (englische Version)"
|
||
|
||
return {
|
||
"url": best["url"],
|
||
"price": best["price"],
|
||
"shop": shop_label,
|
||
}
|
||
|
||
|
||
def search_thalia(game_name):
|
||
"""Thalia.de — AWIN CSV-Feed (Spielwaren)."""
|
||
feed_url = (
|
||
"https://productdata.awin.com/datafeed/download/apikey/"
|
||
"c1f65d88cf133fc5fbdeb7c6785857b8/fid/37855/format/csv/language/de/"
|
||
"delimiter/%2C/compression/gzip/columns/"
|
||
"data_feed_id%2Cmerchant_id%2Cmerchant_name%2Caw_product_id%2C"
|
||
"aw_deep_link%2Caw_image_url%2Caw_thumb_url%2Ccategory_id%2C"
|
||
"category_name%2Cbrand_id%2Cbrand_name%2Cmerchant_product_id%2C"
|
||
"merchant_category%2Cean%2Cisbn%2Cproduct_name%2Cdescription%2C"
|
||
"specifications%2Clanguage%2Cmerchant_deep_link%2Cmerchant_image_url%2C"
|
||
"delivery_time%2Ccurrency%2Csearch_price%2Cdelivery_cost%2Cpre_order%2C"
|
||
"in_stock%2Ccondition%2Cproduct_type%2Cparent_product_id%2Cdimensions%2C"
|
||
"colour%2Ccustom_1%2Ccustom_2%2Ccustom_3%2Ccustom_4%2Ccustom_5%2C"
|
||
"rating%2Cmerchant_product_category_path%2Cmerchant_product_second_category%2C"
|
||
"merchant_product_third_category%2Cbase_price%2Cbase_price_amount%2C"
|
||
"base_price_text%2Ccustom_6%2Ccustom_7%2Ccustom_8%2Ccustom_9/"
|
||
)
|
||
cache_path = "/tmp/thalia_spielwaren.csv"
|
||
_download_csv_feed(feed_url, cache_path, "Thalia Spielwaren")
|
||
result = _search_awin_feed(game_name, cache_path, ",", "Thalia.de",
|
||
"product_name", "search_price", "aw_deep_link", "in_stock")
|
||
if result:
|
||
return result
|
||
# Fallback
|
||
return {
|
||
"url": f"https://www.awin1.com/cread.php?awinmid=14158&awinaffid=661969&ued="
|
||
f"{urllib.parse.quote('https://www.thalia.de/suche?filterPATHROOT=brettspiel&q=' + urllib.parse.quote(game_name))}",
|
||
"price": None,
|
||
"shop": "Thalia.de",
|
||
}
|
||
|
||
|
||
def search_osiander(game_name):
|
||
"""Osiander.de — AWIN CSV-Feed (Spielwaren)."""
|
||
feed_url = (
|
||
"https://productdata.awin.com/datafeed/download/apikey/"
|
||
"c1f65d88cf133fc5fbdeb7c6785857b8/fid/75513/format/csv/language/de/"
|
||
"delimiter/%2C/compression/gzip/columns/"
|
||
"data_feed_id%2Cmerchant_id%2Cmerchant_name%2Caw_product_id%2C"
|
||
"aw_deep_link%2Caw_image_url%2Caw_thumb_url%2Ccategory_id%2C"
|
||
"category_name%2Cbrand_id%2Cbrand_name%2Cmerchant_product_id%2C"
|
||
"merchant_category%2Cean%2Cisbn%2Cproduct_name%2Cdescription%2C"
|
||
"specifications%2Clanguage%2Cmerchant_deep_link%2Cmerchant_image_url%2C"
|
||
"currency%2Csearch_price%2Cdelivery_cost%2Cpre_order%2Cin_stock%2C"
|
||
"condition%2Cparent_product_id%2Cdimensions%2Ccolour%2Ccustom_1%2C"
|
||
"custom_2%2Ccustom_3%2Ccustom_4%2Caverage_rating%2C"
|
||
"merchant_product_category_path%2Cmerchant_product_second_category%2C"
|
||
"merchant_product_third_category%2Cbase_price%2Cbase_price_amount%2C"
|
||
"base_price_text%2Csize_stock_amount/"
|
||
)
|
||
cache_path = "/tmp/osiander_spielwaren.csv"
|
||
_download_csv_feed(feed_url, cache_path, "Osiander Spielwaren")
|
||
result = _search_awin_feed(game_name, cache_path, ",", "Osiander.de",
|
||
"product_name", "search_price", "aw_deep_link", "in_stock")
|
||
if result:
|
||
return result
|
||
return {
|
||
"url": f"https://www.awin1.com/cread.php?awinmid=30799&awinaffid=661969&ued="
|
||
f"{urllib.parse.quote('https://www.osiander.de/suche?suchbegriff=' + urllib.parse.quote(game_name))}",
|
||
"price": None,
|
||
"shop": "Osiander.de",
|
||
}
|
||
|
||
|
||
def search_idee_spiel(game_name):
|
||
"""idee+spiel.de — AWIN CSV-Feed (Productfeed)."""
|
||
feed_url = (
|
||
"https://productdata.awin.com/datafeed/download/apikey/"
|
||
"c1f65d88cf133fc5fbdeb7c6785857b8/fid/91824/format/csv/language/de/"
|
||
"delimiter/%2C/compression/gzip/columns/"
|
||
"data_feed_id%2Cmerchant_id%2Cmerchant_name%2Caw_product_id%2C"
|
||
"aw_deep_link%2Caw_image_url%2Caw_thumb_url%2Ccategory_id%2C"
|
||
"category_name%2Cbrand_id%2Cbrand_name%2Cmerchant_product_id%2C"
|
||
"mpn%2Cproduct_name%2Cdescription%2Cmerchant_deep_link%2C"
|
||
"merchant_image_url%2Cdelivery_time%2Ccurrency%2Csearch_price%2C"
|
||
"delivery_cost%2Cin_stock%2Ccondition%2Ccustom_1%2Ccustom_2%2C"
|
||
"custom_3%2Ccustom_4%2Ccustom_5%2Cbase_price%2Cbase_price_amount%2C"
|
||
"base_price_text%2Ccustom_6%2Cproduct_GTIN/"
|
||
)
|
||
cache_path = "/tmp/ideespiel_feed.csv"
|
||
_download_csv_feed(feed_url, cache_path, "idee+spiel Feed")
|
||
result = _search_awin_feed(game_name, cache_path, ",", "idee+spiel.de",
|
||
"product_name", "search_price", "aw_deep_link", "in_stock")
|
||
if result:
|
||
return result
|
||
return {
|
||
"url": f"https://www.awin1.com/cread.php?awinmid=75052&awinaffid=661969&ued="
|
||
f"{urllib.parse.quote('https://www.ideeundspiel.de/suche?q=' + urllib.parse.quote(game_name))}",
|
||
"price": None,
|
||
"shop": "idee+spiel.de",
|
||
}
|
||
|
||
|
||
SEARCHERS = {
|
||
"Spiele-Offensive.de": search_spiele_offensive,
|
||
"Fantasywelt.de": search_fantasywelt,
|
||
"AllGames4You.de": search_allgames4you,
|
||
"Amazon.de": search_amazon,
|
||
"Milan-Spiele.de": search_milan_spiele,
|
||
"Thalia.de": search_thalia,
|
||
"Osiander.de": search_osiander,
|
||
"idee+spiel.de": search_idee_spiel,
|
||
}
|
||
|
||
|
||
# ── CSV-Cache-Helfer ────────────────────────────────────────────────
|
||
|
||
def _download_csv_feed(url, cache_path, label, max_age=86400):
|
||
"""Download and cache a CSV feed, return cached path or None."""
|
||
if os.path.exists(cache_path):
|
||
age = __import__("time").time() - os.path.getmtime(cache_path)
|
||
if age < max_age:
|
||
return cache_path
|
||
print(f" Feed {label} älter als 24h, lade neu...", file=sys.stderr)
|
||
else:
|
||
print(f" Lade {label}...", file=sys.stderr)
|
||
|
||
try:
|
||
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
|
||
with urllib.request.urlopen(req, timeout=120) as resp:
|
||
raw = resp.read()
|
||
# Check for gzip magic bytes
|
||
if raw[:2] == b'\x1f\x8b':
|
||
import gzip
|
||
raw = gzip.decompress(raw)
|
||
with open(cache_path, "wb") as f:
|
||
f.write(raw)
|
||
return cache_path
|
||
except Exception as e:
|
||
print(f" Fehler beim Download von {label}: {e}", file=sys.stderr)
|
||
return cache_path if os.path.exists(cache_path) else None
|
||
|
||
|
||
def _search_awin_feed(game_name, cache_path, delimiter, shop_label, col_name, col_price, col_link, col_stock=None):
|
||
"""Generic search in an AWIN CSV feed.
|
||
|
||
Returns dict with url, price, shop or None.
|
||
"""
|
||
if not os.path.exists(cache_path):
|
||
return None
|
||
|
||
game_lower = game_name.lower().strip()
|
||
candidates = []
|
||
|
||
with open(cache_path, "r", encoding="utf-8") as f:
|
||
reader = csv.DictReader(f, delimiter=delimiter)
|
||
for row in reader:
|
||
name = (row.get(col_name) or "").strip()
|
||
if not name:
|
||
continue
|
||
|
||
name_lower = name.lower()
|
||
# Quick filter: game name must be substantially present
|
||
if game_lower not in name_lower and not any(
|
||
w in name_lower for w in game_lower.split() if len(w) > 3
|
||
):
|
||
continue
|
||
|
||
score = match_score(game_name, name)
|
||
if score < 60:
|
||
continue
|
||
|
||
price_raw = (row.get(col_price) or "").strip()
|
||
try:
|
||
price_num = float(price_raw)
|
||
price = f"{price_num:.2f}".replace(".", ",")
|
||
except ValueError:
|
||
price = None
|
||
|
||
link = (row.get(col_link) or "").strip()
|
||
in_stock = (row.get(col_stock, "") or "").strip().lower() if col_stock else ""
|
||
|
||
candidates.append({
|
||
"name": name, "url": link, "price": price,
|
||
"score": score, "in_stock": in_stock,
|
||
})
|
||
|
||
if not candidates:
|
||
return None
|
||
|
||
# Sort: score desc, then in_stock, then has_price
|
||
candidates.sort(key=lambda c: (
|
||
-c["score"],
|
||
0 if c["in_stock"] in ("yes", "true", "1", "in stock") else 1,
|
||
0 if c["price"] else 1,
|
||
))
|
||
|
||
best = candidates[0]
|
||
stock_text = ""
|
||
if best.get("in_stock") in ("yes", "true", "1", "in stock"):
|
||
stock_text = " (auf Lager)"
|
||
elif best.get("in_stock") in ("no", "false", "0", "out of stock"):
|
||
stock_text = " (nicht auf Lager)"
|
||
|
||
return {
|
||
"url": best["url"],
|
||
"price": best["price"],
|
||
"shop": f"{shop_label}{stock_text}",
|
||
}
|
||
|
||
|
||
def search_all(game_name, shops=None):
|
||
"""Search all (or specified) shops for a game."""
|
||
if shops is None:
|
||
shops = SEARCHERS.keys()
|
||
|
||
results = []
|
||
for shop_name in shops:
|
||
if shop_name not in SEARCHERS:
|
||
continue
|
||
print(f" Suche {shop_name}...", file=sys.stderr, flush=True)
|
||
try:
|
||
result = SEARCHERS[shop_name](game_name)
|
||
if result:
|
||
results.append(result)
|
||
except Exception as e:
|
||
print(f" Fehler: {e}", file=sys.stderr)
|
||
|
||
return results
|
||
|
||
|
||
def generate_html_section(game_name, results):
|
||
"""Generate HTML for the article 'Hier kaufen' section."""
|
||
if not results:
|
||
return ""
|
||
|
||
items = []
|
||
for r in results:
|
||
price_str = f" – {r['price']} €" if r.get("price") else ""
|
||
items.append(
|
||
f'<li><a href="{r["url"]}" target="_blank" rel="nofollow sponsored">'
|
||
f'{r["shop"]}{price_str}</a></li>'
|
||
)
|
||
|
||
return f"""<div class="affiliate-box" style="background:#f5f0eb;border:1px solid #d4c5b2;border-radius:8px;padding:1rem 1.5rem;margin:1.5rem 0;">
|
||
<strong>🛒 {escape(game_name)} kaufen:</strong>
|
||
<ul style="margin:0.5rem 0 0 1.2rem;padding:0;">
|
||
{chr(10).join(items)}
|
||
</ul>
|
||
<small style="color:#999;">Affiliate-Links – der Preis für dich bleibt gleich, wir erhalten eine kleine Provision.</small>
|
||
</div>"""
|
||
|
||
|
||
if __name__ == "__main__":
|
||
game = sys.argv[1] if len(sys.argv) > 1 else "Cascadia"
|
||
print(f"Suche '{game}' in Partner-Shops...", file=sys.stderr)
|
||
results = search_all(game)
|
||
html = generate_html_section(game, results)
|
||
print(html)
|