#!/usr/bin/env python3
"""
Fantasywelt.de Neuheiten-Tracker v9
====================================
Starten:
python3 fantasywelt_neuheiten.py # Server auf Port 5544
python3 fantasywelt_neuheiten.py --port 8080
python3 fantasywelt_neuheiten.py --once # Einmal ohne Server
"""
import csv, io, json, os, sys, urllib.request, urllib.error
from datetime import datetime, timedelta
from pathlib import Path
from html import escape
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
# ─── KONFIGURATION ───────────────────────────────────────────────────────────
CSV_URL = "https://www.fantasywelt.de/export/brettspiele.csv"
DATA_DIR = Path("/home/hermes/workspace/fantasywelt_data")
HISTORY_DIR = DATA_DIR / "history"
OUTPUT_FILE = DATA_DIR / "neuheiten.html"
SNAPSHOT_FILE = DATA_DIR / "last_snapshot.json"
CHANGES_FILE = DATA_DIR / "changes.json"
DEFAULT_PORT = 5544
DEFAULT_DAYS_NEW = 14
DEFAULT_DAYS_PREORDER = 7
AFFILIATE_TAG = "bsn=ig1UxQ0L2cNMNslzFhyr"
ZUBEHOER_KEYWORDS = [
"schutzhüll","sleeve","hülle","hüllen","deckbox","deck box","deck-box",
"schaumstoff","feldherr","portfolio","zipfolio","flexxfolio","play-mat",
"playmat","play mat","toploader","card holder","card divider",
"pages ","page ","pocket","kartenbox","card case","card box",
"faltkarton","lagerbox","gaming case","gaming box",
"pro-matte","pro-fit","ultra pro","ultra-snug",
"dragon shield sleeves","dragon shield: sleeves","dragon shield schutzhüll",
"board game sleeves","board games sleeves","kmc small","kmc standard",
"precise-fit","soft sleeves",
"9-pocket","4-pocket","8-pocket","12-pocket","16-pocket","18-pocket",
"e-raptor spieler-tableau","organizer","würfel","dice tower",
"raster schaumstoff","schaumstoffeinlage","schaumstoffzuschnitt",
"schaumstoff-set","schaumstofeinlage","schaumstoff abdeckung",
"half-size schaumstoff","full-size schaumstoff","tasche für","medium tasche",
]
EXCLUDE_KEYWORDS = ["mängelexemplar","mängelexemplare","mangelexemplar","%sale%"]
# ─── HILFSFUNKTIONEN ─────────────────────────────────────────────────────────
def should_skip(name):
n = name.lower()
return any(kw in n for kw in ZUBEHOER_KEYWORDS) or any(kw in n for kw in EXCLUDE_KEYWORDS)
def add_affiliate(url):
if not url: return url
return url + ("&" if "?" in url else "?") + AFFILIATE_TAG
def download_csv():
req = urllib.request.Request(CSV_URL, headers={"User-Agent": "BSN-Tracker/9.0"})
with urllib.request.urlopen(req, timeout=60) as resp:
data = resp.read()
try: return data.decode("utf-8")
except UnicodeDecodeError: return data.decode("latin-1")
def parse_csv(raw):
reader = csv.DictReader(io.StringIO(raw), delimiter=";", quotechar='"')
products = []
for row in reader:
name = row.get("Name", "").strip()
if not name: continue
products.append({
"name": name, "sku": row.get("Artikelnummer","").strip(),
"ean": row.get("EAN","").strip(), "price": row.get("VK/Sonderpreis","0").strip(),
"available": row.get("Verfügbarkeit","").strip(),
"supply": row.get("Lieferstatus","").strip(),
"created": row.get("Erstellt am","").strip(),
"release": row.get("Erscheint am","").strip(),
"url": row.get("Produkt-URL","").strip(),
"image": row.get("Bild-URL","").strip(),
"bestprice": row.get("Best-Price_Garantie!","N").strip(),
})
return products
def format_price(s):
try: return f"{float(s.replace(',','.')):.2f}".replace(".",",")
except: return s
def detect_language(name):
n = name.upper()
if "(DE|EN)" in n or "(DEEN)" in n or "(DE/EN)" in n: return "DE/EN"
elif "(DE)" in n or "(DEUTSCH)" in n: return "DE"
elif "(EN)" in n or "(ENGLISH)" in n or "(ENGLISCH)" in n: return "EN"
elif "MULTILINGUAL" in n: return "Multi"
return ""
# ─── SNAPSHOT & ÄNDERUNGSLOG ─────────────────────────────────────────────────
def load_snapshot():
if not SNAPSHOT_FILE.exists(): return None
try:
with open(SNAPSHOT_FILE, "r", encoding="utf-8") as f: return json.load(f)
except: return None
def save_snapshot(products, timestamp):
DATA_DIR.mkdir(parents=True, exist_ok=True); HISTORY_DIR.mkdir(parents=True, exist_ok=True)
snap = {"timestamp": timestamp, "products": {
p["sku"]: {"name":p["name"],"available":p["available"],"price":p["price"],"supply":p["supply"]}
for p in products if p["sku"]}}
with open(SNAPSHOT_FILE, "w", encoding="utf-8") as f: json.dump(snap, f, ensure_ascii=False, indent=2)
safe = timestamp.replace(":","-").replace(" ","_")
with open(HISTORY_DIR / f"snapshot_{safe}.json", "w", encoding="utf-8") as f: json.dump(snap, f, ensure_ascii=False, indent=2)
def load_changes():
if not CHANGES_FILE.exists(): return []
try:
with open(CHANGES_FILE, "r", encoding="utf-8") as f: return json.load(f)
except: return []
def save_changes(changes):
DATA_DIR.mkdir(parents=True, exist_ok=True)
with open(CHANGES_FILE, "w", encoding="utf-8") as f: json.dump(changes, f, ensure_ascii=False, indent=2)
def cleanup_changes(changes, keep_days=30):
cutoff = (datetime.now() - timedelta(days=keep_days)).isoformat()
return [c for c in changes if c.get("detected_at","") >= cutoff]
def cleanup_old_snapshots(keep_days=30):
if not HISTORY_DIR.exists(): return
cutoff = datetime.now() - timedelta(days=keep_days)
for f in sorted(HISTORY_DIR.glob("snapshot_*.json")):
try:
with open(f) as fh: ts = json.load(fh).get("timestamp","")
if datetime.strptime(ts[:19], "%Y-%m-%d %H:%M:%S") < cutoff: f.unlink()
except: continue
# ─── NEUHEITEN-ERKENNUNG ────────────────────────────────────────────────────
def detect_changes(products, old_snapshot, timestamp):
old_products = old_snapshot.get("products", {})
nc = []
for p in products:
if should_skip(p["name"]): continue
sku = p["sku"]
if not sku: continue
old = old_products.get(sku)
ct = None
if p["available"] == "auf Lager":
if old is None:
# Prüfe Erscheinungsdatum: Spiele älter als 120 Tage sind
# keine echten Neuheiten, sondern nur neu im Fantasywelt-Katalog
release_date = None
if p.get("release"):
try: release_date = datetime.strptime(p["release"], "%Y-%m-%d")
except: pass
if release_date and release_date < datetime.now() - timedelta(days=120):
ct = "wieder_im_lager" # altes Spiel, neu im Katalog → wie Restock
else:
ct = "neu_im_lager" # wirklich neu (oder kein release date)
elif old.get("available") != "auf Lager":
ct = "wieder_im_lager" # war schon im Katalog, jetzt wieder da
if p["supply"] == "Vorbestellung":
if old is None: ct = "neue_vorbestellung"
elif old.get("supply") != "Vorbestellung": ct = "neue_vorbestellung"
if ct: nc.append({"sku": sku, "change_type": ct, "detected_at": timestamp,
"name": p["name"], "price": p["price"],
"url": p.get("url",""), "image": p.get("image",""),
"release": p.get("release",""), "bestprice": p.get("bestprice","N")})
return nc
def initial_seed(products, days_new, days_preorder, timestamp):
cutoff_new = datetime.now() - timedelta(days=days_new)
cutoff_pre = datetime.now() - timedelta(days=days_preorder)
seed, seen = [], set()
for p in products:
if should_skip(p["name"]): continue
sku = p["sku"]
if not sku or sku in seen: continue
try: created = datetime.strptime(p["created"], "%Y-%m-%d")
except: continue
if p["supply"] == "Vorbestellung" and created >= cutoff_pre:
seen.add(sku); seed.append({"sku":sku,"change_type":"neue_vorbestellung","detected_at":timestamp,
"name":p["name"],"price":p["price"],"url":p.get("url",""),
"image":p.get("image",""),"release":p.get("release",""),
"bestprice":p.get("bestprice","N")})
elif p["available"] == "auf Lager" and created >= cutoff_new:
seen.add(sku); seed.append({"sku":sku,"change_type":"neu_im_lager","detected_at":timestamp,
"name":p["name"],"price":p["price"],"url":p.get("url",""),
"image":p.get("image",""),"release":p.get("release",""),
"bestprice":p.get("bestprice","N")})
return seed
def build_neuheiten(products, changes, days_new, days_preorder):
cutoff_new = (datetime.now() - timedelta(days=days_new)).isoformat()
cutoff_pre = (datetime.now() - timedelta(days=days_preorder)).isoformat()
pm = {p["sku"]: p for p in products if p["sku"]}
neu_lager, wieder_lager, vorb = {}, {}, {}
for c in changes:
sku, ct, det = c["sku"], c["change_type"], c.get("detected_at","")
if ct == "neu_im_lager" and det >= cutoff_new:
if sku not in neu_lager or det > neu_lager[sku]["detected_at"]: neu_lager[sku] = c
if ct == "wieder_im_lager" and det >= cutoff_new:
if sku not in wieder_lager or det > wieder_lager[sku]["detected_at"]: wieder_lager[sku] = c
if ct == "neue_vorbestellung" and det >= cutoff_pre:
if sku not in vorb or det > vorb[sku]["detected_at"]: vorb[sku] = c
neu = []
for sku, ch in neu_lager.items():
p = pm.get(sku)
if p is None:
# Fallback: Produkt nicht mehr in CSV (verkauft), nutze gespeicherte Daten
if "name" not in ch or should_skip(ch.get("name","")):
continue
p = {"name": ch["name"], "price": ch["price"], "url": ch.get("url",""),
"image": ch.get("image",""), "release": ch.get("release",""),
"bestprice": ch.get("bestprice","N"), "available": "ausverkauft",
"supply": "", "sku": sku}
elif should_skip(p["name"]):
continue
p["change_reason"] = "Neu im Shop"; p["detected_at"] = ch["detected_at"]; neu.append(p)
wieder = []
wieder_skipped = 0
for sku, ch in wieder_lager.items():
if sku in neu_lager: continue
p = pm.get(sku)
if p is None:
if "name" not in ch or should_skip(ch.get("name","")):
continue
p = {"name": ch["name"], "price": ch["price"], "url": ch.get("url",""),
"image": ch.get("image",""), "release": ch.get("release",""),
"bestprice": ch.get("bestprice","N"), "available": "ausverkauft",
"supply": "", "sku": sku}
elif should_skip(p["name"]):
continue
# "Wieder da" nur DE-Versionen — sonst ufern EN-Restocks aus
if detect_language(p["name"]) != "DE":
wieder_skipped += 1
continue
p["change_reason"] = "Wieder auf Lager"; p["detected_at"] = ch["detected_at"]; wieder.append(p)
vl = []
for sku, ch in vorb.items():
if sku in neu_lager or sku in wieder_lager: continue
p = pm.get(sku)
if p is None:
if "name" not in ch or should_skip(ch.get("name","")):
continue
p = {"name": ch["name"], "price": ch["price"], "url": ch.get("url",""),
"image": ch.get("image",""), "release": ch.get("release",""),
"bestprice": ch.get("bestprice","N"), "available": "ausverkauft",
"supply": "Vorbestellung", "sku": sku}
elif should_skip(p["name"]):
continue
p["change_reason"] = "Neue Vorbestellung"; p["detected_at"] = ch["detected_at"]; vl.append(p)
neu.sort(key=lambda p: p.get("detected_at",""), reverse=True)
wieder.sort(key=lambda p: p.get("detected_at",""), reverse=True)
vl.sort(key=lambda p: p.get("detected_at",""), reverse=True)
return neu, wieder, vl
# ─── TRACKER-KERN ────────────────────────────────────────────────────────────
def run_tracker(days_new, days_preorder):
log = []
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
raw = download_csv()
products = parse_csv(raw)
games = [p for p in products if not should_skip(p["name"])]
log.append(f"✅ {len(games)} Spiele geladen")
old = load_snapshot(); changes = load_changes()
if old:
mode = "diff"; nc = detect_changes(products, old, timestamp)
log.append(f"🔍 {len(nc)} neue Änderungen"); changes.extend(nc)
else:
mode = "initial"; seed = initial_seed(products, days_new, days_preorder, timestamp)
log.append(f"📅 Erster Lauf: {len(seed)} Seed"); changes.extend(seed)
changes = cleanup_changes(changes, keep_days=max(days_new, days_preorder)+7)
save_changes(changes); save_snapshot(products, timestamp)
cleanup_old_snapshots(keep_days=max(days_new, days_preorder)+7)
neu, wieder, vorb = build_neuheiten(products, changes, days_new, days_preorder)
log.append(f"🆕 {len(neu)} Neuheiten, {len(wieder)} wieder da (DE), {len(vorb)} Vorbestellungen")
return neu, wieder, vorb, mode, days_new, days_preorder, log
# ─── HTML-GENERIERUNG ────────────────────────────────────────────────────────
def render_deal_cards(items, accent="#20228a"):
"""Amazon-Stil: Flexbox-Cards mit Bild, Name, Preis, Badges."""
if not items:
return 'Keine Neuheiten in diesem Zeitraum.
'
cards = []
for p in items:
name = escape(p["name"])
price = format_price(p["price"])
url = escape(add_affiliate(p.get("url", "#")))
image = p.get("image", "")
lang = detect_language(p["name"])
bestprice = p.get("bestprice", "N")
release_str = ""
release_future = False
if p.get("release"):
try:
rd = datetime.strptime(p["release"], "%Y-%m-%d")
release_str = rd.strftime("%d.%m.%Y")
if rd > datetime.now():
release_future = True
except: pass
# Badges
badges = []
if lang:
badges.append(f'{escape(lang)}')
if bestprice == "Y":
badges.append('Best-Price')
if release_str:
if release_future:
badges.append(f'ET: {release_str} ⚠')
else:
badges.append(f'Ersch. {release_str}')
badge_html = "".join(badges)
# Ausverkauft-Badge
if p.get("available") == "ausverkauft":
badge_html += ' Ausverkauft'
# Image — Cloudflare schützt /media/ vor Hotlinking.
# Bild-URL wird als data-fallback gespeichert, aber nicht als ![]()
eingebunden.
# Im Browser-Kontext (gleiche Session) würden sie laden, cross-origin nicht.
img_html = ""
cards.append(f'''''')
return "".join(cards)
def _render_embed_cards(items):
"""Reine Cards ohne Container (für Copy-Button)."""
return render_deal_cards(items)
def generate_html(neu, wieder, vorb, days_new, days_preorder, mode, sections="all"):
"""Generate HTML. sections: 'all', 'neu', 'neu_wieder', 'vorbestellung'"""
now_str = datetime.now().strftime("%d.%m.%Y")
show_neu = sections in ("all", "neu", "neu_wieder")
show_wieder = sections in ("all", "neu_wieder")
show_vorb = sections in ("all", "vorbestellung")
neu_cards = render_deal_cards(neu, accent="#20228a") if (show_neu and neu) else ""
wieder_cards = render_deal_cards(wieder, accent="#20228a") if (show_wieder and wieder) else ""
vorb_cards = render_deal_cards(vorb, accent="#20228a") if (show_vorb and vorb) else ""
sections_html = []
if show_neu:
if neu:
sections_html.append(f'Neu im Shop ({len(neu)})
')
sections_html.append(f'Brandneue Spiele · Stand: {now_str}
')
sections_html.append(neu_cards)
else:
sections_html.append('Keine brandneuen Spiele in diesem Zeitraum.
')
if show_wieder:
if wieder:
sections_html.append(f'')
sections_html.append(f'
Wieder auf Lager ({len(wieder)})
')
sections_html.append(f'
Nachfrage hat sich gelohnt — diese Spiele sind zurück
')
sections_html.append(wieder_cards)
sections_html.append(f'
Keine Nachschub-Entdeckungen in diesem Zeitraum.
')
sections_html.append(f'
Neue Vorbestellungen ({len(vorb)})
')
sections_html.append(f'
Demnächst erscheinend · Stand: {now_str}
')
sections_html.append(vorb_cards)
sections_html.append(f'
Keine neuen Vorbestellungen in diesem Zeitraum.