#!/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": "awinmid=14379&awinaffid=661969",
"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 (
)
blocks = re.split(r' tag
name_match = re.search(r'([^<]+)', 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 33 , 99 €")
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.
Converts direct URLs to AWIN affiliate links (advertiser 14379)."""
AWIN_URL = "https://www.awin1.com/cread.php?awinmid=14379&awinaffid=661969&ued="
def _awin(url):
return AWIN_URL + urllib.parse.quote(url, safe='')
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": _awin(f"https://www.milan-spiele.de/advanced_search_result.php?keywords={query}"),
"price": None,
"shop": "Milan-Spiele.de",
}
# Parse product blocks: ...
blocks = re.split(r'