"""Kühlfinder Live-Crawler — eigener Verfügbarkeits-/Preis-Scraper.
Muster gespiegelt vom AIDA Preisradar (FastAPI + httpx + APScheduler + /app/data + Coolify).
Scrapt echte, server-gerenderte Händlerseiten (OBI, Bauhaus), persistiert Snapshots +
Restock-Historie, liefert /api/live (CORS). Keine Drittdaten (kein braucheklima-Import)."""
import os, re, json, html, logging, asyncio
from datetime import datetime, timezone
import httpx
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import JSONResponse
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger
from sources import SOURCES
from discovery import discover
logging.basicConfig(level=logging.INFO); log = logging.getLogger("kf-crawler")
DATA = os.getenv("DATA_DIR","/app/data"); os.makedirs(DATA, exist_ok=True)
LIVE = os.path.join(DATA,"live.json"); HIST = os.path.join(DATA,"history.json")
INTERVAL_MIN = int(os.getenv("CRAWL_INTERVAL_MIN","3"))
CONCURRENCY = int(os.getenv("CRAWL_CONCURRENCY","5"))
DISCOVER = os.getenv("CRAWL_DISCOVER","1") not in ("0","false","")
DISCOVER_CAP = int(os.getenv("CRAWL_DISCOVER_CAP","20"))
UA = {
"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Accept":"text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"Accept-Language":"de-DE,de;q=0.9,en;q=0.6",
"sec-ch-ua":'"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"',
"sec-ch-ua-mobile":"?0","sec-ch-ua-platform":'"Windows"',
"sec-fetch-dest":"document","sec-fetch-mode":"navigate","sec-fetch-site":"none","sec-fetch-user":"?1",
"upgrade-insecure-requests":"1",
}
# Optionaler Proxy (DE-Residential empfohlen) – wie AidaBot. Env: HTTP_PROXY_URL
PROXY = os.getenv("HTTP_PROXY_URL","").strip() or None
def _num(s):
s=s.replace(".","").replace("\xa0"," ")
m=re.search(r"(\d{2,4})[,](\d{2})",s) or re.search(r"(\d{2,4}),-",s) or re.search(r"(\d{2,4})\s*€",s)
if not m: return None
g=m.groups()
return float(g[0]) + (float(g[1])/100 if len(g)>1 and g[1] else 0)
def clean_price(v):
v=(v or "").strip()
if not v: return None
if "," in v: v=v.replace(".","").replace(",",".")
m=re.search(r"\d+(?:\.\d+)?", v)
return float(m.group()) if m else None
def price_from_meta(h):
# Shopware/Microdata: itemprop="price" content="..."
mi=re.search(r'itemprop=["\']price["\'][^>]*content=["\']([^"\']+)', h, re.I) or re.search(r'content=["\']([^"\']+)["\'][^>]*itemprop=["\']price["\']', h, re.I)
if mi:
p=clean_price(mi.group(1));
if p: return p
for pat in ("product:price:amount","og:price:amount"):
m=re.search(r']+(?:property|name)=["\']'+pat+r'["\'][^>]+content=["\']([^"\']+)', h, re.I) or \
re.search(r']+content=["\']([^"\']+)["\'][^>]+(?:property|name)=["\']'+pat+r'["\']', h, re.I)
if m:
p=clean_price(m.group(1))
if p: return p
return None
def parse(htmltext):
# 1) JSON-LD Offer
price=None; avail=None
for m in re.finditer(r'', htmltext, re.S|re.I):
try: data=json.loads(m.group(1))
except Exception: continue
for node in (data if isinstance(data,list) else [data]):
graph = node.get("@graph",[node]) if isinstance(node,dict) else [node]
for g in graph:
if not isinstance(g,dict): continue
off=g.get("offers")
if off:
off=off[0] if isinstance(off,list) else off
try: price=float(off.get("price") or off.get("lowPrice"))
except Exception: pass
avail=str(off.get("availability") or "")
# 2) Heuristik
low=htmltext.lower()
in_store = ("vorrätig" in low or "verfügbar" in low) and not ("nicht vorrätig" in low or "nicht verfügbar" in low)
online = ("in den warenkorb" in low) or ("lieferung" in low and "nicht möglich" not in low) or (avail and "instock" in avail.lower())
if price is None:
price=price_from_meta(htmltext)
# Plausibilitäts-Schranke: Klimageräte kosten 3-stellig; verwirf Ausreißer (Raten/Versand)
if price is not None and price < 150:
price=None
if in_store: availability="in_stock"
elif online: availability="online_available"
elif price is not None: availability="out_of_stock"
else: availability="unknown" # erreichbar, aber Markup (noch) nicht geparst
return price, availability
async def fetch_one(client, src):
try:
r= await client.get(src["url"], timeout=25, follow_redirects=True)
if r.status_code in (404,410):
# Produktseite verschwunden -> als entfernt kennzeichnen (das kann braucheklima nicht)
return {"chain":src["chain"],"kind":src["kind"],"productId":src["productId"],"name":src["name"],
"url":src["url"],"price":None,"availability":"discontinued","status":r.status_code}
if r.status_code in (401,403,429):
return {"chain":src["chain"],"kind":src["kind"],"productId":src["productId"],"name":src["name"],
"url":src["url"],"price":None,"availability":"blocked","status":r.status_code}
if r.status_code!=200:
return {**src,"price":None,"availability":"unknown","status":r.status_code}
price, avail = parse(r.text)
return {"chain":src["chain"],"kind":src["kind"],"cc":src.get("cc","DE"),"productId":src["productId"],"name":src["name"],
"url":src["url"],"price":price,"availability":avail,"discovered":src.get("discovered",False)}
except Exception as e:
log.warning("fetch fail %s: %s", src["url"], e)
return {**src,"price":None,"availability":"unknown"}
def load(path,default):
try:
with open(path) as f: return json.load(f)
except Exception: return default
async def crawl():
now=datetime.now(timezone.utc).isoformat()
sem=asyncio.Semaphore(CONCURRENCY)
limits=httpx.Limits(max_connections=CONCURRENCY, max_keepalive_connections=CONCURRENCY)
async with httpx.AsyncClient(proxies=PROXY, headers=UA, http2=False, limits=limits) as client:
# Auto-Discovery: zusätzliche Produkte aus den Kategorieseiten (toom, OBI) finden
sources=list(SOURCES); discovered=0
if DISCOVER:
try:
curated_urls={s["url"] for s in SOURCES}
disc=[d for d in await discover(client, cap=DISCOVER_CAP) if d["url"] not in curated_urls]
sources+=disc; discovered=len(disc)
log.info("discovery: +%d Produkte aus Kategorieseiten", discovered)
except Exception as e:
log.warning("discovery failed: %s", e)
async def guarded(src):
async with sem:
await asyncio.sleep(0.2) # kleiner Jitter, trotzdem höflich
return await fetch_one(client, src)
# parallel (begrenzt) -> auch bei vielen Quellen schnell genug für kurzen Takt
results=list(await asyncio.gather(*[guarded(s) for s in sources]))
# Restock-Events
prev={ (o["chain"],o["productId"]): o for o in load(LIVE,{}).get("offers",[]) }
hist=load(HIST,{"events":[]})
for o in results:
p=prev.get((o["chain"],o["productId"]))
if p and p.get("availability")=="out_of_stock" and o.get("availability") in ("in_stock","online_available"):
hist["events"].append({"chain":o["chain"],"productId":o["productId"],"name":o["name"],"at":now})
if p and p.get("availability")!="discontinued" and o.get("availability")=="discontinued":
hist["events"].append({"chain":o["chain"],"productId":o["productId"],"name":o["name"],"at":now,"type":"discontinued"})
hist["events"]=hist["events"][-200:]
# Preis-Aggregat je Produkt
byp={}
for o in results:
if o.get("price"): byp.setdefault(o["productId"],[]).append(o["price"])
agg={k:{"min":min(v),"max":max(v),"n":len(v)} for k,v in byp.items()}
gone=[o for o in results if o.get("availability")=="discontinued"]
blocked=[o for o in results if o.get("availability")=="blocked"]
snap={"generatedAt":now,"source":"kuehlfinder-livecrawler","intervalMin":INTERVAL_MIN,"discontinued":len(gone),"blocked":len(blocked),"proxy":bool(PROXY),
"curatedCount":len(SOURCES),"discoveredCount":discovered,"totalSources":len(sources),
"offers":results,"priceByProduct":agg,"feed":hist["events"][-20:][::-1]}
json.dump(snap, open(LIVE,"w"), ensure_ascii=False)
json.dump(hist, open(HIST,"w"), ensure_ascii=False)
log.info("crawl done: %d offers, %d events", len(results), len(hist["events"]))
return snap
app=FastAPI(title="Kühlfinder Live-Crawler", version="0.1.0")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["GET"], allow_headers=["*"])
app.add_middleware(GZipMiddleware, minimum_size=400)
sched=AsyncIOScheduler()
@app.on_event("startup")
async def _start():
await crawl()
sched.add_job(crawl, IntervalTrigger(minutes=INTERVAL_MIN), id="crawl", max_instances=1, coalesce=True)
sched.start()
@app.get("/health")
def health(): return {"ok":True}
@app.get("/api/live")
def live(): return JSONResponse(load(LIVE,{"offers":[],"note":"warming up"}))