Moritz Graf
9d890a290f
Introduce the marktguru agent skill and MCP integration for searching supermarket deals, comparing product prices, and querying supported retail chains: - Add Marktguru API client and MCP server script - Add skill definition (SKILL.md) and supported supermarkets reference - Add test suite covering Marktguru API integration and offer queries
578 lines
22 KiB
Python
Executable File
578 lines
22 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Marktguru API Client (Production Grade)
|
|
Fetches promotional offers and discounts from German supermarkets via the Marktguru API.
|
|
|
|
Key Design Principles:
|
|
- Zero external dependencies (uses standard library only: urllib, json, os, time, datetime).
|
|
- Resilient: Retries with exponential backoff on transient errors.
|
|
- Low Maintenance: Configurable keys and postal codes via environment variables.
|
|
- Fail-Safe: Caching with stale fallback if upstream API is temporarily unreachable.
|
|
- Defensive: Robust parsing protecting against upstream schema shifts and malformed payloads.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import json
|
|
import logging
|
|
import hashlib
|
|
import tempfile
|
|
import urllib.request
|
|
import urllib.parse
|
|
import urllib.error
|
|
from datetime import datetime, timezone
|
|
from typing import List, Dict, Any, Optional, Set, Tuple
|
|
|
|
# Logger writing to stderr to keep stdout pure for JSON / CLI
|
|
logger = logging.getLogger("marktguru_client")
|
|
if not logger.handlers:
|
|
handler = logging.StreamHandler(sys.stderr)
|
|
handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s"))
|
|
logger.addHandler(handler)
|
|
logger.setLevel(logging.INFO)
|
|
|
|
API_BASE = "https://api.marktguru.de/api/v1"
|
|
|
|
# Environment variable overrides for zero-code configuration changes
|
|
DEFAULT_ZIP_CODE = os.environ.get("MARKTGURU_ZIP_CODE", "93059")
|
|
CLIENT_KEY = os.environ.get("MARKTGURU_CLIENT_KEY", "WU/RH+PMGDi+gkZer3WbMelt6zcYHSTytNB7VpTia90=")
|
|
API_KEY = os.environ.get("MARKTGURU_API_KEY", "8Kk+pmbf7TgJ9nVj2cXeA7P5zBGv8iuutVVMRfOfvNE=")
|
|
CACHE_TTL = int(os.environ.get("MARKTGURU_CACHE_TTL", "1800")) # 30 minutes
|
|
CACHE_ENABLED = os.environ.get("MARKTGURU_CACHE_ENABLED", "true").lower() in ("true", "1", "yes")
|
|
|
|
HEADERS = {
|
|
"x-clientkey": CLIENT_KEY,
|
|
"x-apikey": API_KEY,
|
|
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko)",
|
|
"Accept": "application/json",
|
|
}
|
|
|
|
# Major supermarket chains and their Marktguru retailer IDs
|
|
MAJOR_RETAILERS: Dict[str, Dict[str, Any]] = {
|
|
"ALDI SÜD": {"id": 127153, "type": "Discounter", "major": True},
|
|
"ALDI NORD": {"id": 127139, "type": "Discounter", "major": True},
|
|
"Lidl": {"id": 126679, "type": "Discounter", "major": True},
|
|
"EDEKA": {"id": 127093, "type": "Supermarkt", "major": True},
|
|
"E center": {"id": 126699, "type": "Supermarkt", "major": True},
|
|
"REWE": {"id": 126802, "type": "Supermarkt", "major": True},
|
|
"REWE Center": {"id": 126804, "type": "Supermarkt", "major": True},
|
|
"Kaufland": {"id": 126654, "type": "Großfläche/Hypermarkt", "major": True},
|
|
"PENNY": {"id": 126765, "type": "Discounter", "major": True},
|
|
"Netto Marken-Discount": {"id": 126735, "type": "Discounter", "major": True},
|
|
"Netto (mit Hund)": {"id": 126734, "type": "Discounter", "major": False},
|
|
"NORMA": {"id": 126745, "type": "Discounter", "major": True},
|
|
"Globus": {"id": 127177, "type": "Großfläche/Hypermarkt", "major": True},
|
|
"HIT": {"id": 126598, "type": "Supermarkt", "major": False},
|
|
"tegut": {"id": 126866, "type": "Supermarkt", "major": False},
|
|
"nahkauf": {"id": 137320, "type": "Supermarkt", "major": False},
|
|
"dm-drogerie markt": {"id": 127057, "type": "Drogerie", "major": True},
|
|
"ROSSMANN": {"id": 126811, "type": "Drogerie", "major": True},
|
|
"Müller Drogeriemarkt": {"id": 126720, "type": "Drogerie", "major": True},
|
|
}
|
|
|
|
# Aliases mapping user query strings to Marktguru retailer IDs
|
|
RETAILER_ALIASES: Dict[str, List[int]] = {
|
|
"aldi": [127153, 127139],
|
|
"aldi süd": [127153],
|
|
"aldi sued": [127153],
|
|
"aldi nord": [127139],
|
|
"lidl": [126679],
|
|
"edeka": [127093, 126699],
|
|
"e center": [126699],
|
|
"ecenter": [126699],
|
|
"rewe": [126802, 126804],
|
|
"kaufland": [126654],
|
|
"penny": [126765],
|
|
"netto": [126735, 126734],
|
|
"norma": [126745],
|
|
"globus": [127177],
|
|
"hit": [126598],
|
|
"tegut": [126866],
|
|
"nahkauf": [137320],
|
|
"dm": [127057],
|
|
"rossmann": [126811],
|
|
"müller": [126720],
|
|
"mueller": [126720],
|
|
}
|
|
|
|
|
|
def _get_cache_dir() -> str:
|
|
"""Returns directory path for local cached responses."""
|
|
cache_dir = os.path.join(tempfile.gettempdir(), "marktguru_cache")
|
|
try:
|
|
os.makedirs(cache_dir, exist_ok=True)
|
|
except Exception:
|
|
pass
|
|
return cache_dir
|
|
|
|
|
|
def _read_cache(cache_key: str, max_age_seconds: int = CACHE_TTL) -> Tuple[Optional[Dict[str, Any]], bool]:
|
|
"""
|
|
Reads cache file. Returns (data, is_fresh).
|
|
If cache is expired, returns (data, False) so caller can use it as emergency fallback.
|
|
"""
|
|
if not CACHE_ENABLED:
|
|
return None, False
|
|
|
|
cache_file = os.path.join(_get_cache_dir(), f"{cache_key}.json")
|
|
if not os.path.isfile(cache_file):
|
|
return None, False
|
|
|
|
try:
|
|
with open(cache_file, "r", encoding="utf-8") as f:
|
|
payload = json.load(f)
|
|
cached_time = payload.get("_cached_at", 0)
|
|
age = time.time() - cached_time
|
|
data = payload.get("data")
|
|
is_fresh = age < max_age_seconds
|
|
return data, is_fresh
|
|
except Exception as e:
|
|
logger.debug(f"Failed to read cache {cache_key}: {e}")
|
|
return None, False
|
|
|
|
|
|
def _write_cache(cache_key: str, data: Dict[str, Any]) -> None:
|
|
"""Writes data to cache file atomically."""
|
|
if not CACHE_ENABLED:
|
|
return
|
|
|
|
cache_file = os.path.join(_get_cache_dir(), f"{cache_key}.json")
|
|
temp_file = f"{cache_file}.tmp.{os.getpid()}"
|
|
try:
|
|
payload = {
|
|
"_cached_at": time.time(),
|
|
"data": data,
|
|
}
|
|
with open(temp_file, "w", encoding="utf-8") as f:
|
|
json.dump(payload, f, ensure_ascii=False)
|
|
os.replace(temp_file, cache_file)
|
|
except Exception as e:
|
|
logger.debug(f"Failed to write cache {cache_key}: {e}")
|
|
if os.path.exists(temp_file):
|
|
try:
|
|
os.remove(temp_file)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def resolve_supermarket_ids(supermarkets: Optional[List[str]]) -> Optional[str]:
|
|
"""Resolves human supermarket names/aliases to comma-separated Marktguru retailer IDs."""
|
|
if not supermarkets:
|
|
return None
|
|
|
|
# Handle string passed instead of list (e.g. "Aldi, Lidl")
|
|
if isinstance(supermarkets, str):
|
|
supermarkets = [s.strip() for s in supermarkets.split(",") if s.strip()]
|
|
|
|
resolved_ids: Set[int] = set()
|
|
for name in supermarkets:
|
|
if not isinstance(name, str):
|
|
continue
|
|
cleaned = name.strip().lower()
|
|
if not cleaned:
|
|
continue
|
|
|
|
if cleaned in RETAILER_ALIASES:
|
|
resolved_ids.update(RETAILER_ALIASES[cleaned])
|
|
elif cleaned.isdigit():
|
|
resolved_ids.add(int(cleaned))
|
|
else:
|
|
found = False
|
|
for r_name, r_info in MAJOR_RETAILERS.items():
|
|
if cleaned in r_name.lower():
|
|
resolved_ids.add(r_info["id"])
|
|
found = True
|
|
if not found:
|
|
logger.warning(f"Unknown supermarket name '{name}' skipped.")
|
|
|
|
return ",".join(str(i) for i in sorted(resolved_ids)) if resolved_ids else None
|
|
|
|
|
|
def format_iso_date(iso_str: Optional[str]) -> str:
|
|
"""Formats an ISO-8601 date string to human-readable DD.MM.YYYY format."""
|
|
if not iso_str or not isinstance(iso_str, str):
|
|
return "-"
|
|
try:
|
|
dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
|
|
return dt.strftime("%d.%m.%Y")
|
|
except Exception:
|
|
return iso_str[:10] if len(iso_str) >= 10 else iso_str
|
|
|
|
|
|
def parse_offer(raw: Any, zip_code: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Defensively parses and standardizes an offer payload from Marktguru.
|
|
Protects against null fields, malformed dates, and missing keys.
|
|
"""
|
|
if not isinstance(raw, dict):
|
|
return None
|
|
|
|
try:
|
|
offer_id = raw.get("id")
|
|
price = raw.get("price")
|
|
if isinstance(price, (int, float)):
|
|
price = round(float(price), 2)
|
|
else:
|
|
price = None
|
|
|
|
old_price = raw.get("oldPrice")
|
|
if isinstance(old_price, (int, float)):
|
|
old_price = round(float(old_price), 2)
|
|
else:
|
|
old_price = None
|
|
|
|
ref_price = raw.get("referencePrice")
|
|
if isinstance(ref_price, (int, float)):
|
|
ref_price = round(float(ref_price), 2)
|
|
else:
|
|
ref_price = None
|
|
|
|
discount_pct = None
|
|
if old_price and price and old_price > price:
|
|
discount_pct = round(((old_price - price) / old_price) * 100)
|
|
|
|
# Validity dates
|
|
validity = raw.get("validityDates")
|
|
valid_from_raw = None
|
|
valid_to_raw = None
|
|
is_valid_now = True
|
|
days_remaining = None
|
|
|
|
if isinstance(validity, list) and len(validity) > 0 and isinstance(validity[0], dict):
|
|
valid_from_raw = validity[0].get("from")
|
|
valid_to_raw = validity[0].get("to")
|
|
|
|
try:
|
|
now = datetime.now(timezone.utc)
|
|
if valid_to_raw and isinstance(valid_to_raw, str):
|
|
dt_to = datetime.fromisoformat(valid_to_raw.replace("Z", "+00:00"))
|
|
diff = (dt_to - now).days
|
|
days_remaining = max(0, diff)
|
|
if dt_to < now:
|
|
is_valid_now = False
|
|
|
|
if valid_from_raw and isinstance(valid_from_raw, str):
|
|
dt_from = datetime.fromisoformat(valid_from_raw.replace("Z", "+00:00"))
|
|
if dt_from > now:
|
|
is_valid_now = False
|
|
except Exception:
|
|
pass
|
|
|
|
# Supermarket / advertiser extraction
|
|
supermarket = "Unbekannt"
|
|
supermarket_id = None
|
|
advertisers = raw.get("advertisers")
|
|
if isinstance(advertisers, list) and len(advertisers) > 0 and isinstance(advertisers[0], dict):
|
|
supermarket = advertisers[0].get("name") or "Unbekannt"
|
|
supermarket_id = advertisers[0].get("id")
|
|
|
|
# High resolution CDN image
|
|
image_url = None
|
|
if offer_id:
|
|
image_url = f"https://mg2de.b-cdn.net/api/v1/offers/{offer_id}/images/default/0/medium.webp"
|
|
|
|
# Unit & volume
|
|
unit_info = raw.get("unit")
|
|
unit_name = None
|
|
if isinstance(unit_info, dict):
|
|
unit_name = unit_info.get("shortName") or unit_info.get("name")
|
|
|
|
product_info = raw.get("product")
|
|
product_name = None
|
|
if isinstance(product_info, dict):
|
|
product_name = product_info.get("name")
|
|
|
|
brand_info = raw.get("brand")
|
|
brand_name = None
|
|
if isinstance(brand_info, dict):
|
|
brand_name = brand_info.get("name")
|
|
|
|
return {
|
|
"id": offer_id,
|
|
"supermarket": supermarket,
|
|
"supermarket_id": supermarket_id,
|
|
"product_name": product_name,
|
|
"brand": brand_name,
|
|
"description": raw.get("description"),
|
|
"price": price,
|
|
"old_price": old_price,
|
|
"reference_price": ref_price,
|
|
"discount_percent": discount_pct,
|
|
"unit": unit_name,
|
|
"volume": raw.get("volume"),
|
|
"loyalty_app_required": bool(raw.get("requiresLoyalityMembership", False)),
|
|
"is_valid_now": is_valid_now,
|
|
"days_remaining": days_remaining,
|
|
"valid_from": format_iso_date(valid_from_raw),
|
|
"valid_to": format_iso_date(valid_to_raw),
|
|
"image_url": image_url,
|
|
"zip_code": str(zip_code),
|
|
}
|
|
except Exception as e:
|
|
logger.debug(f"Error parsing single offer {raw.get('id') if isinstance(raw, dict) else raw}: {e}")
|
|
return None
|
|
|
|
|
|
def _fetch_with_retry(url: str, max_retries: int = 2) -> Dict[str, Any]:
|
|
"""
|
|
Executes HTTP request with exponential backoff on transient errors (5xx / network blips).
|
|
"""
|
|
last_err: Optional[Exception] = None
|
|
|
|
for attempt in range(max_retries + 1):
|
|
req = urllib.request.Request(url, headers=HEADERS)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
raw_bytes = resp.read()
|
|
return json.loads(raw_bytes.decode("utf-8"))
|
|
except urllib.error.HTTPError as e:
|
|
last_err = e
|
|
# Retry only on transient server errors (500, 502, 503, 504) or rate limits (429)
|
|
if e.code in (429, 500, 502, 503, 504) and attempt < max_retries:
|
|
backoff = 0.5 * (2 ** attempt)
|
|
logger.warning(f"Marktguru HTTP {e.code}. Retrying in {backoff:.1f}s (attempt {attempt+1}/{max_retries})...")
|
|
time.sleep(backoff)
|
|
continue
|
|
raise
|
|
except (urllib.error.URLError, TimeoutError, OSError) as e:
|
|
last_err = e
|
|
if attempt < max_retries:
|
|
backoff = 0.5 * (2 ** attempt)
|
|
logger.warning(f"Connection issue ({e}). Retrying in {backoff:.1f}s...")
|
|
time.sleep(backoff)
|
|
continue
|
|
raise
|
|
|
|
raise last_err or RuntimeError("Max retries exceeded")
|
|
|
|
|
|
def search_offers(
|
|
query: str,
|
|
zip_code: str = DEFAULT_ZIP_CODE,
|
|
supermarkets: Optional[List[str]] = None,
|
|
limit: int = 24,
|
|
offset: int = 0,
|
|
force_refresh: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Searches current supermarket offers in Germany via the Marktguru API.
|
|
|
|
Args:
|
|
query: Search term (e.g. 'Butter', 'Kaffee', 'Bier', 'Milch')
|
|
zip_code: German postal code (default: '93059' Regensburg)
|
|
supermarkets: List of supermarket names or aliases (e.g. ['Aldi', 'Lidl', 'Edeka'])
|
|
limit: Max number of items to return (default: 24, max: 50)
|
|
offset: Result pagination offset
|
|
force_refresh: If True, bypasses local TTL cache
|
|
"""
|
|
cleaned_query = (query or "").strip()
|
|
cleaned_zip = str(zip_code or DEFAULT_ZIP_CODE).strip()
|
|
limit = max(1, min(int(limit), 50))
|
|
offset = max(0, int(offset))
|
|
|
|
retailer_ids = resolve_supermarket_ids(supermarkets)
|
|
|
|
# Compute deterministic cache key
|
|
cache_raw = f"{cleaned_query.lower()}:{cleaned_zip}:{retailer_ids or ''}:{limit}:{offset}"
|
|
cache_key = hashlib.sha256(cache_raw.encode("utf-8")).hexdigest()[:16]
|
|
|
|
if not force_refresh:
|
|
cached_data, is_fresh = _read_cache(cache_key)
|
|
if is_fresh and cached_data:
|
|
return cached_data
|
|
|
|
params: Dict[str, Any] = {
|
|
"as": "web",
|
|
"limit": limit,
|
|
"offset": offset,
|
|
"q": cleaned_query,
|
|
"zipCode": cleaned_zip,
|
|
}
|
|
if retailer_ids:
|
|
params["retailers"] = retailer_ids
|
|
|
|
url = f"{API_BASE}/offers/search?{urllib.parse.urlencode(params)}"
|
|
|
|
try:
|
|
data = _fetch_with_retry(url)
|
|
except Exception as e:
|
|
logger.error(f"Marktguru API query failed: {e}")
|
|
# Emergency fail-safe: Check if stale cache is available
|
|
cached_data, _ = _read_cache(cache_key, max_age_seconds=86400 * 7)
|
|
if cached_data:
|
|
logger.warning("Serving stale cached data due to API failure.")
|
|
cached_data["warning"] = "Stale cache returned due to temporary upstream API issue."
|
|
return cached_data
|
|
|
|
return {
|
|
"error": f"API request failed: {str(e)}",
|
|
"query": cleaned_query,
|
|
"zip_code": cleaned_zip,
|
|
"total_results": 0,
|
|
"returned_count": 0,
|
|
"offers": [],
|
|
"available_retailers": [],
|
|
}
|
|
|
|
raw_results = data.get("results") or []
|
|
parsed_offers: List[Dict[str, Any]] = []
|
|
for item in raw_results:
|
|
parsed = parse_offer(item, cleaned_zip)
|
|
if parsed is not None:
|
|
parsed_offers.append(parsed)
|
|
|
|
filters = data.get("filters") or {}
|
|
raw_retailers = filters.get("retailers") or []
|
|
available_retailers = [
|
|
{"name": r.get("name"), "id": r.get("id"), "count": r.get("resultsCount")}
|
|
for r in raw_retailers if isinstance(r, dict) and r.get("name")
|
|
]
|
|
|
|
result = {
|
|
"query": cleaned_query,
|
|
"zip_code": cleaned_zip,
|
|
"total_results": data.get("totalResults", len(parsed_offers)),
|
|
"returned_count": len(parsed_offers),
|
|
"available_retailers": available_retailers,
|
|
"offers": parsed_offers,
|
|
}
|
|
|
|
_write_cache(cache_key, result)
|
|
return result
|
|
|
|
|
|
def compare_prices(
|
|
query: str,
|
|
zip_code: str = DEFAULT_ZIP_CODE,
|
|
supermarkets: Optional[List[str]] = None,
|
|
force_refresh: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Compares prices for a given product query across supermarkets, sorted from cheapest to most expensive.
|
|
"""
|
|
data = search_offers(query, zip_code=zip_code, supermarkets=supermarkets, limit=50, force_refresh=force_refresh)
|
|
offers = data.get("offers", [])
|
|
|
|
# Filter out items without price and sort ascending
|
|
priced_offers = [o for o in offers if o.get("price") is not None]
|
|
priced_offers.sort(key=lambda x: x["price"])
|
|
|
|
return {
|
|
"query": query,
|
|
"zip_code": str(zip_code),
|
|
"total_compared": len(priced_offers),
|
|
"cheapest_offer": priced_offers[0] if priced_offers else None,
|
|
"all_offers_sorted": priced_offers,
|
|
}
|
|
|
|
|
|
def list_supported_supermarkets() -> Dict[str, Any]:
|
|
"""Returns a structured catalog of all major supermarket chains supported by Marktguru."""
|
|
categories: Dict[str, List[Dict[str, Any]]] = {
|
|
"Discounter": [],
|
|
"Supermarkt": [],
|
|
"Großfläche / Hypermarkt": [],
|
|
"Drogerie": [],
|
|
}
|
|
|
|
for name, info in MAJOR_RETAILERS.items():
|
|
cat = info.get("type", "Sonstige")
|
|
if cat not in categories:
|
|
categories[cat] = []
|
|
categories[cat].append({
|
|
"name": name,
|
|
"id": info["id"],
|
|
"major": info.get("major", False)
|
|
})
|
|
|
|
return {
|
|
"default_zip_code": DEFAULT_ZIP_CODE,
|
|
"cache_ttl_seconds": CACHE_TTL,
|
|
"supported_chains_count": len(MAJOR_RETAILERS),
|
|
"categories": categories,
|
|
"supported_aliases": list(RETAILER_ALIASES.keys()),
|
|
}
|
|
|
|
|
|
def main():
|
|
import argparse
|
|
parser = argparse.ArgumentParser(description="Query German supermarket flyers/discounts via Marktguru")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
# Search
|
|
search_p = subparsers.add_parser("search", help="Search offers for a product")
|
|
search_p.add_argument("query", help="Product to search (e.g. Butter, Kaffee)")
|
|
search_p.add_argument("--zip", default=DEFAULT_ZIP_CODE, help=f"Postal code (default: {DEFAULT_ZIP_CODE})")
|
|
search_p.add_argument("--supermarket", help="Comma-separated supermarkets (e.g. 'Aldi,Lidl,Edeka')")
|
|
search_p.add_argument("--limit", type=int, default=15, help="Max results (default 15)")
|
|
search_p.add_argument("--refresh", action="store_true", help="Bypass local cache")
|
|
search_p.add_argument("--json", action="store_true", help="Output raw JSON")
|
|
|
|
# Compare
|
|
comp_p = subparsers.add_parser("compare", help="Compare prices across supermarkets")
|
|
comp_p.add_argument("query", help="Product to compare")
|
|
comp_p.add_argument("--zip", default=DEFAULT_ZIP_CODE, help=f"Postal code (default: {DEFAULT_ZIP_CODE})")
|
|
comp_p.add_argument("--supermarket", help="Comma-separated supermarkets")
|
|
comp_p.add_argument("--refresh", action="store_true", help="Bypass local cache")
|
|
comp_p.add_argument("--json", action="store_true", help="Output raw JSON")
|
|
|
|
# Supermarkets
|
|
subparsers.add_parser("supermarkets", help="List all supported supermarkets and chains")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "supermarkets":
|
|
data = list_supported_supermarkets()
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
return
|
|
|
|
sm_list = [s.strip() for s in args.supermarket.split(",")] if getattr(args, "supermarket", None) else None
|
|
|
|
if args.command == "search":
|
|
result = search_offers(
|
|
args.query,
|
|
zip_code=args.zip,
|
|
supermarkets=sm_list,
|
|
limit=args.limit,
|
|
force_refresh=args.refresh
|
|
)
|
|
if args.json:
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
else:
|
|
if "error" in result and not result.get("offers"):
|
|
print(f"❌ Fehler: {result['error']}")
|
|
return
|
|
|
|
print(f"\n🔍 Angebote für '{args.query}' in PLZ {args.zip} (Gefunden: {result['total_results']}):\n")
|
|
for o in result["offers"]:
|
|
disc_str = f" (-{o['discount_percent']}%)" if o.get("discount_percent") else ""
|
|
loyalty_str = " 📱 [App Coupon]" if o.get("loyalty_app_required") else ""
|
|
timing_str = f"noch {o['days_remaining']} Tage" if o.get("days_remaining") is not None else f"bis {o['valid_to']}"
|
|
print(f" • [{o['supermarket']}] {o['description'] or o['product_name'] or 'Angebot'}")
|
|
print(f" Preis: {o['price']:.2f} €{disc_str}{loyalty_str} | Gültig: {timing_str}")
|
|
if o.get("reference_price"):
|
|
print(f" Grundpreis: {o['reference_price']:.2f} € / {o.get('unit') or 'Einheit'}")
|
|
print()
|
|
|
|
elif args.command == "compare":
|
|
result = compare_prices(
|
|
args.query,
|
|
zip_code=args.zip,
|
|
supermarkets=sm_list,
|
|
force_refresh=args.refresh
|
|
)
|
|
if args.json:
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
else:
|
|
print(f"\n🏷️ Preisvergleich für '{args.query}' in PLZ {args.zip}:\n")
|
|
for idx, o in enumerate(result["all_offers_sorted"], 1):
|
|
disc_str = f" (UVP {o['old_price']:.2f}€)" if o.get("old_price") else ""
|
|
loyalty_str = " [App erforderlich]" if o.get("loyalty_app_required") else ""
|
|
print(f" {idx:2d}. {o['price']:.2f} € - [{o['supermarket']}] {o['description'] or o['product_name']}{disc_str}{loyalty_str}")
|
|
print()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|