From e8c0046c51d7f1c2f344d071fcb01bd2c53f55ee Mon Sep 17 00:00:00 2001 From: Moritz Graf Date: Thu, 10 Sep 2026 13:43:29 +0200 Subject: [PATCH] feat(skills): add shopping skill with supermarket deals, geizhals, and ebay MCP servers --- AGENTS.md | 5 + skills/shopping/SKILL.md | 86 ++++ skills/shopping/scripts/ebay_mcp.py | 477 +++++++++++++++++++ skills/shopping/scripts/geizhals_mcp.py | 399 ++++++++++++++++ skills/shopping/scripts/supermarket_deals.py | 385 +++++++++++++++ 5 files changed, 1352 insertions(+) create mode 100644 skills/shopping/SKILL.md create mode 100755 skills/shopping/scripts/ebay_mcp.py create mode 100755 skills/shopping/scripts/geizhals_mcp.py create mode 100755 skills/shopping/scripts/supermarket_deals.py diff --git a/AGENTS.md b/AGENTS.md index 0121ecc..266a97a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,6 +41,11 @@ Seamless Jellyfin and Jellyseerr integration. Enables natural-language discovery - **Features:** Smart search cross-referenced with library availability, discover with genre & rating filters, direct stream & download links, Jellyfin active sessions & remote play, Jellyseerr requests. - **Tools:** Standalone MCP server and CLI at `scripts/jelly_media.py`. +### [shopping](skills/shopping/SKILL.md) +Smart shopping advisor and Vorrats-Concierge. +- **Features:** German supermarket & discounter weekly discounts (Edeka, Rewe, Aldi, Lidl, Kaufland, Netto, Penny, dm, Rossmann), Vorratskauf multi-brand comparison (Mutti / Oro di Parma), diaper piece-price calculator (ct/Windel), Geizhals price comparison & all-time-low price history, and eBay marketplace search. +- **Tools:** Standalone MCP servers and CLIs at `scripts/supermarket_deals.py`, `scripts/geizhals_mcp.py`, and `scripts/ebay_mcp.py`. + ### [skill-creator](skills/skill-creator/SKILL.md) Guidelines and scripts to help developers build high-quality, token-efficient Claude skills. diff --git a/skills/shopping/SKILL.md b/skills/shopping/SKILL.md new file mode 100644 index 0000000..46fa71d --- /dev/null +++ b/skills/shopping/SKILL.md @@ -0,0 +1,86 @@ +--- +name: shopping +description: Intelligent shopping advisory, price comparisons across German supermarkets (Edeka, Rewe, Aldi, Lidl, Kaufland, Netto, Penny, dm, Rossmann), Geizhals retail best prices & price history, and eBay marketplace searches. Use when the user asks for shopping advice, supermarket offers (e.g. Mutti/Oro Tomatensaucen, Windeln Größe X, Vorratskäufe), price checks, or second-hand/refurbished deals. +--- + +# 🛒 Smart Shopper & Vorrats-Concierge + +Comprehensive shopping advisor for the Graf family. Combines live German supermarket weekly brochures (Regensburg default), Geizhals retail price comparison & price histories, and eBay marketplace searches. + +--- + +## 🛠️ MCP Tools Overview + +### 1. Supermarket & Grocery Deals (`supermarket`) +- **`supermarket_search(query, city="Regensburg", lat=49.0134, lng=12.1016)`**: + - Searches live weekly offers and digital brochures across Edeka, Rewe, Aldi Süd, Lidl, Kaufland, Netto Marken-Discount, Penny, dm, Rossmann, V-Markt, Müller. + - Returns price in EUR, retailer, brand, validity dates, distance in km, and brochure page. +- **`supermarket_compare_staples(staples=["Mutti", "Oro di Parma"], city="Regensburg")`**: + - Batch search for multiple staple brands or product categories to compare discounts across all local supermarkets in one view. + +### 2. Retail Price Comparison & History (`geizhals`) +- **`geizhals_search(query, country="de", limit=8)`**: + - Full-text search across Geizhals.de (and .at). Returns the best price, average price, offer count across all web shops, cheapest shop name, star ratings, and test review count & metascore. +- **`geizhals_price_history(product_id, days=31, country="de")`**: + - Analyzes 31, 91, 183, or 365-day price trends. Reports `period_min_price`, `current_best_price`, and flags if it is an `is_all_time_low`. + +### 3. Marketplace & Second-Hand (`ebay`) +- **`ebay_search(query, condition, buying_format, min_price, max_price, limit=10)`**: + - Searches eBay.de using the official Browse API. Supports filtering by condition (`used`/`refurbished`/`new`) and buying format (`fixed_price`/`auction`). +- **`ebay_get_item(item_id)`**: + - Fetches listing details, photos, location, and seller reputation. + +--- + +## 🍅 Vorratskauf & Supermarkt Protocol + +When the user asks for groceries, weekly offers, or staples to stock up on: + +### 1. Multi-Brand / Equivalency Queries (e.g. „Mutti oder Oro Tomatensauce“) +- Always run a comparative search (using `supermarket_compare_staples` or parallel queries). +- Group results by retailer and rank by best price. +- **Price Benchmarks for Staples**: + - **Tomaten / Passata / Polpa (Mutti / Oro di Parma)**: + - Normal: 1,69 € – 1,99 € + - 🟢 **Top-Vorratspreis**: $\le$ 1,10 € (z.B. Netto 0,88 € oder Penny 1,19 €) + - **Marken-Pasta (Barilla / De Cecco)**: + - Normal: 1,89 € – 2,29 € + - 🟢 **Top-Vorratspreis**: $\le$ 0,99 € + - **Kaffee ganze Bohnen (Lavazza, Segafredo, Melitta)**: + - Normal: 14,99 € – 17,99 € / kg + - 🟢 **Top-Vorratspreis**: $\le$ 11,99 € / kg + +### 2. Windeln Größe X (Stückpreis-Berechnung) +Windel-Packungsgrößen variieren stark (Single-Pack vs. XXL-Box vs. Monatsbox). +- **Berechne immer den Cent-Preis pro Windel**: + $$\text{Stückpreis} = \frac{\text{Gesamtpreis}}{\text{Stückzahl}} \times 100\ \text{ct}$$ +- **Orientierungswerte**: + - **Eigenmarke** (Lupilu / babylove): **11 – 14 ct / Windel** + - **Markenwindeln** (Pampers Baby-Dry): **18 – 24 ct / Windel** (über 26 ct ist teuer) +- Beispiel-Antwort: *„Bei Lidl gibt es Lupilu Gr. 4 (84 Stück) für 10,35 € (12,3 ct/Windel). Bei V-Markt sind Pampers Pants für 6,99 € im Angebot.“* + +### 3. Location & Store Distance +- Default location is **Regensburg** (Lat: `49.0134`, Lng: `12.1016`). +- Mention the retailer and approximate distance (e.g. *„Netto Marken-Discount (0,5 km)“* oder *„Penny (1,4 km)“*). +- Mention the validity dates (z.B. *„Gültig bis Samstag, 12.09.“*). + +--- + +## 💻 Tech & Electronics Advisory Protocol + +When the user asks whether to buy a gadget, phone, laptop, or appliance: +1. **Model & Quality Check**: + - Inspect `test_reviews_count` and `test_reviews_metascore` in `geizhals_search`. + - If needed, run a quick web search on Stiftung Warentest / Chip / Reddit for common faults. +2. **Timing & Price History Check**: + - Run `geizhals_price_history(product_id)`. + - If `is_all_time_low` is true: *„🟢 Aktuell auf Tiefstpreis-Niveau!“* + - If significantly higher than `period_min_price`: advise waiting or setting a price alert. +3. **Refurbished / B-Ware Option**: + - Query `ebay_search` with `condition="refurbished"` or `"used"` to compare savings against brand-new retail. + +--- + +## 🔗 Presentation Rules +- Follow OpenClaw's SOUL protocol: Hide hyperlinks behind descriptive text (`[Händler (Preis €)](url)`). +- Never send raw search URLs or unverified Google links. diff --git a/skills/shopping/scripts/ebay_mcp.py b/skills/shopping/scripts/ebay_mcp.py new file mode 100755 index 0000000..1a5721a --- /dev/null +++ b/skills/shopping/scripts/ebay_mcp.py @@ -0,0 +1,477 @@ +#!/usr/bin/env python3 +""" +eBay Browse API MCP Server & CLI Tool +Connects to eBay's official Browse API using OAuth 2.0 Client Credentials Grant. +Allows AI assistants to search items on eBay.de, filter by condition (NEW, USED, REFURBISHED), +buying format (Sofort-Kaufen / Auktion), price range, and fetch listing details. +Runs standalone over MCP stdio or direct CLI commands. +Requires zero external dependencies (pure Python standard library). +""" + +import os +import sys +import json +import time +import base64 +import urllib.request +import urllib.parse +from typing import Dict, Any, List, Optional + +EBAY_TOKEN_URL = "https://api.ebay.com/identity/v1/oauth2/token" +EBAY_BROWSE_URL = "https://api.ebay.com/buy/browse/v1" +EBAY_OAUTH_SCOPE = "https://api.ebay.com/oauth/api_scope" + +EBAY_APP_ID = os.environ.get("EBAY_APP_ID", "").strip() +EBAY_CERT_ID = os.environ.get("EBAY_CERT_ID", "").strip() +EBAY_MARKETPLACE_ID = os.environ.get("EBAY_MARKETPLACE_ID", "EBAY_DE") + +# Condition mapping for user-friendly terms +CONDITION_MAP = { + "new": "NEW", + "neu": "NEW", + "used": "USED", + "gebraucht": "USED", + "refurbished": "CERTIFIED_REFURBISHED,EXCELLENT_REFURBISHED,VERY_GOOD_REFURBISHED,GOOD_REFURBISHED,SELLER_REFURBISHED", + "generalüberholt": "CERTIFIED_REFURBISHED,EXCELLENT_REFURBISHED,VERY_GOOD_REFURBISHED,GOOD_REFURBISHED,SELLER_REFURBISHED", +} + +BUYING_FORMAT_MAP = { + "fixed_price": "FIXED_PRICE", + "sofortkauf": "FIXED_PRICE", + "auction": "AUCTION", + "auktion": "AUCTION", +} + + +class EbayClient: + def __init__(self, app_id: Optional[str] = None, cert_id: Optional[str] = None, marketplace_id: Optional[str] = None): + self.app_id = app_id or EBAY_APP_ID + self.cert_id = cert_id or EBAY_CERT_ID + self.marketplace_id = marketplace_id or EBAY_MARKETPLACE_ID + self._access_token: Optional[str] = None + self._token_expires_at: float = 0.0 + + def is_configured(self) -> bool: + dummy_values = ["", "dummy-ebay-app-id", "dummy-ebay-cert-id", "your-ebay-app-id", "your-ebay-cert-id"] + return bool(self.app_id and self.cert_id and self.app_id not in dummy_values and self.cert_id not in dummy_values) + + def get_token(self) -> str: + """Retrieves or refreshes eBay OAuth 2.0 application access token.""" + if not self.is_configured(): + raise ValueError( + "eBay Developer credentials (EBAY_APP_ID and EBAY_CERT_ID) are not configured. " + "Please add them to openclaw.secret.yaml under ebay-app-id and ebay-cert-id." + ) + + # Reuse cached token if valid (with 60-second buffer) + if self._access_token and time.time() < self._token_expires_at - 60: + return self._access_token + + creds = f"{self.app_id}:{self.cert_id}".encode("utf-8") + basic_auth = base64.b64encode(creds).decode("utf-8") + + data = urllib.parse.urlencode({ + "grant_type": "client_credentials", + "scope": EBAY_OAUTH_SCOPE + }).encode("utf-8") + + req = urllib.request.Request( + EBAY_TOKEN_URL, + data=data, + headers={ + "Authorization": f"Basic {basic_auth}", + "Content-Type": "application/x-www-form-urlencoded" + } + ) + + with urllib.request.urlopen(req, timeout=15) as resp: + body = json.loads(resp.read().decode("utf-8")) + + self._access_token = body.get("access_token") + expires_in = body.get("expires_in", 7200) + self._token_expires_at = time.time() + expires_in + return self._access_token + + def search( + self, + query: str, + condition: Optional[str] = None, + buying_format: Optional[str] = None, + min_price: Optional[float] = None, + max_price: Optional[float] = None, + limit: int = 15 + ) -> Dict[str, Any]: + """Searches items on eBay via the Browse API.""" + if not self.is_configured(): + return { + "configured": False, + "error": "eBay credentials not configured. Please set EBAY_APP_ID and EBAY_CERT_ID in openclaw.secret.yaml.", + "query": query, + "results": [] + } + + try: + token = self.get_token() + except Exception as e: + return { + "configured": False, + "error": f"eBay authentication failed: {str(e)}", + "query": query, + "results": [] + } + + # Build query filters + filters = [] + if min_price is not None or max_price is not None: + min_str = f"{min_price:.2f}" if min_price is not None else "" + max_str = f"{max_price:.2f}" if max_price is not None else "" + filters.append(f"price:[{min_str}..{max_str}]") + + if condition: + cond_lower = condition.lower().strip() + ebay_cond = CONDITION_MAP.get(cond_lower, condition.upper()) + filters.append(f"conditions:{{{ebay_cond}}}") + + if buying_format: + fmt_lower = buying_format.lower().strip() + ebay_fmt = BUYING_FORMAT_MAP.get(fmt_lower, buying_format.upper()) + filters.append(f"buyingOptions:{{{ebay_fmt}}}") + + params = { + "q": query, + "limit": str(min(max(limit, 1), 50)) + } + if filters: + params["filter"] = ",".join(filters) + + url = f"{EBAY_BROWSE_URL}/item_summary/search?{urllib.parse.urlencode(params)}" + req = urllib.request.Request( + url, + headers={ + "Authorization": f"Bearer {token}", + "X-EBAY-C-MARKETPLACE-ID": self.marketplace_id, + "Accept": "application/json" + } + ) + + try: + with urllib.request.urlopen(req, timeout=15) as resp: + data = json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as e: + err_text = e.read().decode("utf-8", errors="replace") + return {"error": f"eBay API HTTP error {e.code}: {err_text}", "query": query, "results": []} + except Exception as e: + return {"error": f"eBay search error: {str(e)}", "query": query, "results": []} + + items = data.get("itemSummaries", []) + total = data.get("total", len(items)) + + results = [] + for it in items: + price_dict = it.get("price") or {} + price_val = price_dict.get("value") + price_curr = price_dict.get("currency", "EUR") + + price_str = f"{float(price_val):.2f} {price_curr}" if price_val is not None else "Kein Preis" + if price_curr == "EUR" and price_val is not None: + price_str = f"{float(price_val):.2f} €".replace(".", ",") + + # Shipping options + shipping_list = it.get("shippingOptions") or [] + shipping_info = "Versand nicht angegeben" + shipping_cost = None + if shipping_list: + s_cost_dict = shipping_list[0].get("shippingCost") or {} + s_val = s_cost_dict.get("value") + if s_val == "0.00" or s_val == "0" or s_val is None: + shipping_info = "Kostenloser Versand" + shipping_cost = 0.0 + else: + shipping_cost = float(s_val) + shipping_info = f"{shipping_cost:.2f} € Versand".replace(".", ",") + + # Buying options (Auktion vs Sofort-Kaufen) + buying_opts = it.get("buyingOptions") or [] + is_auction = "AUCTION" in buying_opts + is_fixed_price = "FIXED_PRICE" in buying_opts + format_label = "Sofort-Kaufen" if is_fixed_price else ("Auktion" if is_auction else ", ".join(buying_opts)) + + seller = it.get("seller") or {} + results.append({ + "item_id": it.get("itemId"), + "title": it.get("title"), + "price": price_str, + "price_numeric": float(price_val) if price_val is not None else None, + "shipping": shipping_info, + "shipping_cost": shipping_cost, + "condition": it.get("condition"), + "format": format_label, + "seller_username": seller.get("username"), + "seller_feedback_percent": seller.get("feedbackPercentage"), + "url": it.get("itemWebUrl"), + "image_url": it.get("image", {}).get("imageUrl") + }) + + return { + "query": query, + "marketplace": self.marketplace_id, + "total_found": total, + "returned_count": len(results), + "results": results + } + + def get_item(self, item_id: str) -> Dict[str, Any]: + """Fetches full details for a specific eBay listing.""" + if not self.is_configured(): + return {"configured": False, "error": "eBay credentials not configured."} + + try: + token = self.get_token() + except Exception as e: + return {"configured": False, "error": f"eBay authentication failed: {str(e)}"} + + url = f"{EBAY_BROWSE_URL}/item/{urllib.parse.quote(item_id)}" + req = urllib.request.Request( + url, + headers={ + "Authorization": f"Bearer {token}", + "X-EBAY-C-MARKETPLACE-ID": self.marketplace_id, + "Accept": "application/json" + } + ) + + try: + with urllib.request.urlopen(req, timeout=15) as resp: + item = json.loads(resp.read().decode("utf-8")) + except Exception as e: + return {"error": f"Failed to fetch eBay item {item_id}: {str(e)}"} + + images = [] + if item.get("image", {}).get("imageUrl"): + images.append(item["image"]["imageUrl"]) + for add_img in item.get("additionalImages", []): + if add_img.get("imageUrl"): + images.append(add_img["imageUrl"]) + + price_dict = item.get("price") or {} + price_val = price_dict.get("value") + price_curr = price_dict.get("currency", "EUR") + price_str = f"{float(price_val):.2f} €".replace(".", ",") if price_val else "Kein Preis" + + seller = item.get("seller") or {} + location = item.get("itemLocation") or {} + loc_str = ", ".join(filter(None, [location.get("city"), location.get("stateOrProvince"), location.get("country")])) + + return { + "item_id": item.get("itemId"), + "title": item.get("title"), + "price": price_str, + "condition": item.get("condition"), + "description": item.get("shortDescription") or (item.get("description", "")[:400] if item.get("description") else None), + "location": loc_str or None, + "seller": seller.get("username"), + "seller_feedback_score": seller.get("feedbackScore"), + "seller_feedback_percent": seller.get("feedbackPercentage"), + "images": images, + "url": item.get("itemWebUrl") + } + + +# ============================================================================== +# Model Context Protocol (MCP) stdio Server +# ============================================================================== + +MCP_TOOLS = [ + { + "name": "ebay_search", + "description": "Search items on eBay.de via official Browse API. Supports filtering by condition (NEW, USED, REFURBISHED), format (AUCTION, FIXED_PRICE/Sofort-Kaufen), and price bounds.", + "inputSchema": { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "Item query, e.g. 'ThinkPad T14', 'iPhone 13 gebraucht', 'Sonos One'" + }, + "condition": { + "type": "string", + "description": "Item condition: 'used'/'gebraucht', 'refurbished'/'generalüberholt', or 'new'/'neu'", + "enum": ["used", "gebraucht", "refurbished", "generalüberholt", "new", "neu"] + }, + "buying_format": { + "type": "string", + "description": "Buying format: 'fixed_price'/'sofortkauf' or 'auction'/'auktion'", + "enum": ["fixed_price", "sofortkauf", "auction", "auktion"] + }, + "min_price": { + "type": "number", + "description": "Minimum price in EUR" + }, + "max_price": { + "type": "number", + "description": "Maximum price in EUR" + }, + "limit": { + "type": "integer", + "description": "Max number of items to return (default 10)", + "default": 10 + } + }, + "required": ["query"] + } + }, + { + "name": "ebay_get_item", + "description": "Fetch detailed information for a specific eBay listing (description, item location, additional photos, return policy).", + "inputSchema": { + "type": "object", + "properties": { + "item_id": { + "type": "string", + "description": "eBay listing item ID, obtained from ebay_search" + } + }, + "required": ["item_id"] + } + } +] + + +def run_mcp_server(): + client = EbayClient() + + while True: + line = sys.stdin.readline() + if not line: + break + line = line.strip() + if not line: + continue + + try: + req = json.loads(line) + except Exception as e: + sys.stderr.write(f"Invalid JSON received: {e}\n") + continue + + msg_id = req.get("id") + method = req.get("method") + params = req.get("params", {}) + + if method == "initialize": + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "ebay-mcp", "version": "1.0.0"} + } + } + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + elif method == "notifications/initialized": + pass + + elif method == "ping": + resp = {"jsonrpc": "2.0", "id": msg_id, "result": {}} + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + elif method == "tools/list": + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": {"tools": MCP_TOOLS} + } + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + elif method == "tools/call": + tool_name = params.get("name") + args = params.get("arguments", {}) + try: + if tool_name == "ebay_search": + data = client.search( + query=args.get("query", ""), + condition=args.get("condition"), + buying_format=args.get("buying_format"), + min_price=args.get("min_price"), + max_price=args.get("max_price"), + limit=args.get("limit", 10) + ) + elif tool_name == "ebay_get_item": + data = client.get_item(item_id=args.get("item_id", "")) + else: + raise ValueError(f"Unknown tool: {tool_name}") + + text_result = json.dumps(data, indent=2, ensure_ascii=False) + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": { + "content": [{"type": "text", "text": text_result}] + } + } + except Exception as e: + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": { + "isError": True, + "content": [{"type": "text", "text": f"Error executing {tool_name}: {str(e)}"}] + } + } + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + else: + if msg_id is not None: + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "error": {"code": -32601, "message": f"Method not found: {method}"} + } + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + +def run_cli(): + import argparse + parser = argparse.ArgumentParser(description="eBay Browse API MCP & CLI") + subparsers = parser.add_subparsers(dest="command") + + sp_search = subparsers.add_parser("search", help="Search items on eBay") + sp_search.add_argument("query", help="Item search query") + sp_search.add_argument("--condition", choices=["used", "gebraucht", "refurbished", "new", "neu"]) + sp_search.add_argument("--format", choices=["fixed_price", "sofortkauf", "auction", "auktion"]) + sp_search.add_argument("--min-price", type=float) + sp_search.add_argument("--max-price", type=float) + sp_search.add_argument("--limit", type=int, default=10) + + sp_item = subparsers.add_parser("item", help="Get eBay item details") + sp_item.add_argument("item_id", help="eBay Item ID") + + args = parser.parse_args() + client = EbayClient() + + if args.command == "search": + print(json.dumps(client.search( + query=args.query, + condition=args.condition, + buying_format=args.format, + min_price=args.min_price, + max_price=args.max_price, + limit=args.limit + ), indent=2, ensure_ascii=False)) + elif args.command == "item": + print(json.dumps(client.get_item(args.item_id), indent=2, ensure_ascii=False)) + else: + parser.print_help() + + +if __name__ == "__main__": + if len(sys.argv) == 1 or (len(sys.argv) == 2 and sys.argv[1] == "--stdio"): + run_mcp_server() + else: + run_cli() diff --git a/skills/shopping/scripts/geizhals_mcp.py b/skills/shopping/scripts/geizhals_mcp.py new file mode 100755 index 0000000..b6c9ed8 --- /dev/null +++ b/skills/shopping/scripts/geizhals_mcp.py @@ -0,0 +1,399 @@ +#!/usr/bin/env python3 +""" +Geizhals MCP Server & CLI Tool +Searches Geizhals (and Skinflint) price comparison data: best prices across shops, +merchant counts, star ratings, test review counts, and price history trends. +Communicates directly with the official Geizhals mobile app API (api.geizhals.net/gh/v9) +using offline HMAC-SHA256 request signing (bypassing Cloudflare JS challenges). +Requires zero external Python dependencies (pure standard library). +""" + +import os +import sys +import json +import time +import hmac +import hashlib +import base64 +import urllib.request +import urllib.parse +from typing import Dict, Any, List, Optional + +BASE_URL = "https://api.geizhals.net/gh/v9" +API_HOST = "api.geizhals.net" +GH_TOKEN_ID = "mobileapp-2026-android-3.12.7" +GH_AUTH_ROLE = "mobileapp-2026-android-3.12.7" +GH_HMAC_SECRET = "2Uq.p2gD1z5Im,m73u5eSVzeuFZJpcQVtzx.M3OtyW.4LvzO" +GH_CLIENT = "Geizhals/3.13.1 (Android 14; sdk_gphone64_x86_64; emu64xa; x86_64; de)" + +DEFAULT_COUNTRY = os.environ.get("GEIZHALS_DEFAULT_LOC", "de") + + +def _b64url(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode() + + +def _auth_header(body_bytes: bytes, method: str = "POST", query_string: str = "") -> str: + now = int(time.time()) + if method == "POST": + s = f"POST|{API_HOST}||{body_bytes.decode('utf-8')}" + else: + s = f"{method}|{API_HOST}|{query_string}|" + + fp = hashlib.sha256(s.encode("utf-8")).hexdigest() + header = {"alg": "HS256", "typ": "JWT"} + payload = { + "token_id": GH_TOKEN_ID, + "timestamp": now, + "request_fingerprint": fp, + "iat": now, + } + + header_b64 = _b64url(json.dumps(header, separators=(",", ":")).encode()) + payload_b64 = _b64url(json.dumps(payload, separators=(",", ":")).encode()) + signing_input = f"{header_b64}.{payload_b64}" + + sig = hmac.new(GH_HMAC_SECRET.encode("utf-8"), signing_input.encode("utf-8"), hashlib.sha256).digest() + return f"Bearer {signing_input}.{_b64url(sig)}" + + +def _post(endpoint: str, payload: dict) -> dict: + body = json.dumps(payload, separators=(",", ":")).encode("utf-8") + req = urllib.request.Request( + f"{BASE_URL}/{endpoint}", + data=body, + headers={ + "User-Agent": GH_CLIENT, + "Authorization": _auth_header(body), + "auth-role": GH_AUTH_ROLE, + "Content-Type": "application/json" + } + ) + with urllib.request.urlopen(req, timeout=20) as resp: + return json.loads(resp.read().decode("utf-8")) + + +class GeizhalsClient: + def search(self, query: str, country: Optional[str] = None, limit: int = 8) -> Dict[str, Any]: + loc = (country or DEFAULT_COUNTRY).lower() + # Geizhals strictly requires pagesize to be in [1, 5, 10, 30, 100, 300, 1000] + valid_pagesizes = [1, 5, 10, 30, 100] + pagesize = 10 + for ps in valid_pagesizes: + if ps >= limit: + pagesize = ps + break + + payload = { + "query": query, + "params": { + "loc": loc, + "hloc": [loc], + "lang": "de", + "pagesize": pagesize, + "n_offers": 1 + } + } + + try: + raw = _post("search_product", payload) + except Exception as e: + return {"error": f"Geizhals search error: {str(e)}", "query": query, "results": []} + + resp = raw.get("response", {}) + products = resp.get("products", []) + total = raw.get("total", len(products)) + + results = [] + domain = "geizhals.de" if loc == "de" else "geizhals.at" + + for p in products[:limit]: + prices = p.get("prices") or {} + best_price = prices.get("best") + avg_price = prices.get("avg") + offers_list = p.get("offers") or [] + offer_count = p.get("offer_count", len(offers_list)) + + # If best_price is missing in prices dict, check first offer + cheapest_shop = None + if offers_list: + first_offer = offers_list[0] + cheapest_shop = first_offer.get("shop", {}).get("name") + if best_price is None: + best_price = first_offer.get("pricing", {}).get("eur_value") or first_offer.get("price", {}).get("value") + + # Determine product link + urls = p.get("urls") or {} + overview_path = urls.get("overview") + web_url = f"https://{domain}{overview_path}" if overview_path else None + + # Test review count & metascore + reviews = p.get("test_reviews") or [] + metascore = p.get("average_test_reviews_metascore") + + # Image handling (already starts with https://gzhls.at or relative) + raw_images = p.get("images") or [] + image_url = None + if raw_images: + first_img = raw_images[0] + if isinstance(first_img, str): + image_url = first_img if first_img.startswith("http") else f"https://gzhls.at/{first_img}" + elif isinstance(first_img, dict): + image_url = first_img.get("url") + + results.append({ + "id": p.get("gzhid"), + "name": p.get("product") or p.get("product_for_sort"), + "manufacturer": p.get("manufacturer_name"), + "best_price": float(best_price) if best_price is not None else None, + "best_price_formatted": f"{float(best_price):.2f} €".replace(".", ",") if best_price is not None else "Kein Live-Angebot", + "average_price": float(avg_price) if avg_price is not None else None, + "cheapest_shop": cheapest_shop, + "offer_count": offer_count, + "available": bool(best_price is not None and offer_count > 0), + "rating_stars": p.get("rating_stars"), + "rating_count": p.get("rating_count"), + "test_reviews_count": len(reviews), + "test_reviews_metascore": metascore, + "url": web_url, + "image_url": image_url + }) + + return { + "query": query, + "country": loc, + "total_found": total, + "returned_count": len(results), + "results": results + } + + def price_history(self, product_id: int, days: int = 31, country: Optional[str] = None) -> Dict[str, Any]: + loc = (country or DEFAULT_COUNTRY).lower() + # Days must be one of 31, 91, 183, 365 + valid_days = [31, 91, 183, 365] + closest_days = min(valid_days, key=lambda x: abs(x - days)) + + payload = { + "id": product_id, + "params": { + "days": closest_days, + "loc": loc + } + } + + try: + raw = _post("price_history", payload) + except Exception as e: + return {"error": f"Price history error: {str(e)}", "product_id": product_id} + + meta = raw.get("meta") or {} + series = raw.get("response") or [] + + min_price = meta.get("min") + max_price = meta.get("max") + current_best = meta.get("current_best") + + is_all_time_low = False + if current_best and min_price: + is_all_time_low = float(current_best) <= float(min_price) + 0.05 + + return { + "product_id": product_id, + "country": loc, + "period_days": closest_days, + "current_best_price": float(current_best) if current_best is not None else None, + "period_min_price": float(min_price) if min_price is not None else None, + "period_max_price": float(max_price) if max_price is not None else None, + "is_all_time_low": is_all_time_low, + "data_points_count": len(series) + } + + +# ============================================================================== +# Model Context Protocol (MCP) stdio Server +# ============================================================================== + +MCP_TOOLS = [ + { + "name": "geizhals_search", + "description": "Search products on Geizhals (Germany/Austria price comparison). Returns lowest price, offer count across all stores, rating, and product link.", + "inputSchema": { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "Product search query, e.g. 'MacBook Air M3', 'Sony WH-1000XM5', 'Mutti Tomaten'" + }, + "country": { + "type": "string", + "description": "Country code ('de' for Germany, 'at' for Austria). Defaults to 'de'.", + "default": "de" + }, + "limit": { + "type": "integer", + "description": "Max number of items to return (default 8)", + "default": 8 + } + }, + "required": ["query"] + } + }, + { + "name": "geizhals_price_history", + "description": "Get price history trends for a Geizhals product ID over 31, 91, 183, or 365 days to check if the current price is a good deal or all-time low.", + "inputSchema": { + "type": "object", + "properties": { + "product_id": { + "type": "integer", + "description": "Geizhals product ID (gzhid) obtained from geizhals_search" + }, + "days": { + "type": "integer", + "description": "History timeframe in days (31, 91, 183, or 365). Defaults to 31.", + "default": 31 + }, + "country": { + "type": "string", + "description": "Country code ('de' or 'at'). Defaults to 'de'.", + "default": "de" + } + }, + "required": ["product_id"] + } + } +] + + +def run_mcp_server(): + client = GeizhalsClient() + + while True: + line = sys.stdin.readline() + if not line: + break + line = line.strip() + if not line: + continue + + try: + req = json.loads(line) + except Exception as e: + sys.stderr.write(f"Invalid JSON received: {e}\n") + continue + + msg_id = req.get("id") + method = req.get("method") + params = req.get("params", {}) + + if method == "initialize": + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "geizhals-mcp", "version": "1.0.0"} + } + } + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + elif method == "notifications/initialized": + pass + + elif method == "ping": + resp = {"jsonrpc": "2.0", "id": msg_id, "result": {}} + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + elif method == "tools/list": + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": {"tools": MCP_TOOLS} + } + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + elif method == "tools/call": + tool_name = params.get("name") + args = params.get("arguments", {}) + try: + if tool_name == "geizhals_search": + data = client.search( + query=args.get("query", ""), + country=args.get("country"), + limit=args.get("limit", 8) + ) + elif tool_name == "geizhals_price_history": + data = client.price_history( + product_id=int(args.get("product_id")), + days=int(args.get("days", 31)), + country=args.get("country") + ) + else: + raise ValueError(f"Unknown tool: {tool_name}") + + text_result = json.dumps(data, indent=2, ensure_ascii=False) + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": { + "content": [{"type": "text", "text": text_result}] + } + } + except Exception as e: + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": { + "isError": True, + "content": [{"type": "text", "text": f"Error executing {tool_name}: {str(e)}"}] + } + } + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + else: + if msg_id is not None: + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "error": {"code": -32601, "message": f"Method not found: {method}"} + } + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + +def run_cli(): + import argparse + parser = argparse.ArgumentParser(description="Geizhals Price Comparison MCP & CLI") + subparsers = parser.add_subparsers(dest="command") + + sp_search = subparsers.add_parser("search", help="Search products on Geizhals") + sp_search.add_argument("query", help="Product query") + sp_search.add_argument("--country", default="de", help="Country (de or at)") + sp_search.add_argument("--limit", type=int, default=8, help="Max results") + + sp_hist = subparsers.add_parser("history", help="Get price history for product ID") + sp_hist.add_argument("product_id", type=int, help="Geizhals product ID (gzhid)") + sp_hist.add_argument("--days", type=int, default=31, help="History days (31, 91, 183, 365)") + sp_hist.add_argument("--country", default="de", help="Country (de or at)") + + args = parser.parse_args() + client = GeizhalsClient() + + if args.command == "search": + print(json.dumps(client.search(args.query, country=args.country, limit=args.limit), indent=2, ensure_ascii=False)) + elif args.command == "history": + print(json.dumps(client.price_history(args.product_id, days=args.days, country=args.country), indent=2, ensure_ascii=False)) + else: + parser.print_help() + + +if __name__ == "__main__": + if len(sys.argv) == 1 or (len(sys.argv) == 2 and sys.argv[1] == "--stdio"): + run_mcp_server() + else: + run_cli() diff --git a/skills/shopping/scripts/supermarket_deals.py b/skills/shopping/scripts/supermarket_deals.py new file mode 100755 index 0000000..995cf69 --- /dev/null +++ b/skills/shopping/scripts/supermarket_deals.py @@ -0,0 +1,385 @@ +#!/usr/bin/env python3 +""" +Supermarket Deals MCP Server & CLI Tool +Fetches live supermarket discounts and brochures across German retailers +(Edeka, Rewe, Aldi Süd/Nord, Lidl, Kaufland, Netto, Penny, dm, Rossmann, V-Markt). +Optimized for staple goods (Mutti, Oro di Parma, Barilla, coffee) and diapers (Windeln piece-price calculation). +Runs standalone over MCP stdio or direct CLI commands. +""" + +import os +import sys +import json +import re +import urllib.request +import urllib.parse +from typing import Dict, Any, List, Optional + +# Default Coordinates: Regensburg, Germany +DEFAULT_LAT = float(os.environ.get("SUPERMARKET_DEFAULT_LAT", "49.0134")) +DEFAULT_LNG = float(os.environ.get("SUPERMARKET_DEFAULT_LNG", "12.1016")) +DEFAULT_CITY = os.environ.get("SUPERMARKET_DEFAULT_CITY", "Regensburg") + +CITY_COORDINATES = { + "regensburg": (49.0134, 12.1016), + "münchen": (48.1351, 11.5820), + "munich": (48.1351, 11.5820), + "nürnberg": (49.4521, 11.0767), + "berlin": (52.5200, 13.4050), + "hamburg": (53.5511, 9.9937), + "frankfurt": (50.1109, 8.6821), + "köln": (50.9375, 6.9603), + "stuttgart": (48.7758, 9.1829), +} + +HEADERS = { + "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", + "Accept": "application/json", + "Accept-Language": "de-DE,de;q=0.9,en;q=0.8" +} + + +def resolve_coordinates(city: Optional[str] = None, lat: Optional[float] = None, lng: Optional[float] = None) -> tuple: + if lat is not None and lng is not None: + return (lat, lng) + if city: + norm = city.lower().strip() + if norm in CITY_COORDINATES: + return CITY_COORDINATES[norm] + return (DEFAULT_LAT, DEFAULT_LNG) + + +def calculate_piece_price(title: str, description: Optional[str], price_float: float) -> Optional[Dict[str, Any]]: + """Calculates piece price (e.g. ct/Windel or ct/Stück) from title/description.""" + if price_float <= 0: + return None + + text = f"{title} {description or ''}" + # Match patterns like: "84 Stück", "84 Stk.", "84-St.-Packg.", "114er", "24 ct" + match = re.search(r"(\d+)\s*(?:stück|stk\.?|st\.-packg\.?|ct\b|pants\b|windeln\b|er\s*packung)", text, re.IGNORECASE) + if not match: + # Also check for "Größe 3/4/5, Je 92/84/72 Stück" + multi_match = re.search(r"(\d+)/(\d+)/(\d+)\s*stück", text, re.IGNORECASE) + if multi_match: + counts = [int(g) for g in multi_match.groups()] + avg_count = sum(counts) / len(counts) + cent_piece = round((price_float / avg_count) * 100, 1) + return { + "piece_counts": counts, + "approx_cents_per_piece": cent_piece, + "formatted": f"ca. {cent_piece} ct / Stück" + } + return None + + count = int(match.group(1)) + if count <= 0 or count > 500: + return None + + cent_piece = round((price_float / count) * 100, 1) + return { + "pieces": count, + "cents_per_piece": cent_piece, + "formatted": f"{cent_piece} ct / Stück" + } + + +def evaluate_deal(brand: str, title: str, price: float) -> str: + """Evaluates whether an offer is a standout bargain for known staple brands.""" + norm = f"{brand} {title}".lower() + if any(k in norm for k in ["mutti", "oro di parma", "oro"]): + if any(t in norm for t in ["polpa", "stückig", "passata", "geschält", "pelati", "tomaten"]): + if price <= 1.05: + return "TOP_DEAL_VORRAT" + elif price <= 1.29: + return "GUTER_ANGEBOTSPREIS" + return "NORMALER_PREIS" + if "barilla" in norm: + if price <= 0.89: + return "TOP_DEAL_VORRAT" + elif price <= 1.09: + return "GUTER_ANGEBOTSPREIS" + return "ANGEBOT" + + +class SupermarketClient: + def search(self, query: str, city: Optional[str] = None, lat: Optional[float] = None, lng: Optional[float] = None) -> Dict[str, Any]: + use_lat, use_lng = resolve_coordinates(city, lat, lng) + url = f"https://www.kaufda.de/api/search?lat={use_lat}&lng={use_lng}&query={urllib.parse.quote(query)}" + req = urllib.request.Request(url, headers=HEADERS) + + try: + with urllib.request.urlopen(req, timeout=12) as resp: + data = json.loads(resp.read().decode("utf-8")) + except Exception as e: + return { + "query": query, + "city": city or DEFAULT_CITY, + "coordinates": {"lat": use_lat, "lng": use_lng}, + "error": f"Failed to fetch offers: {str(e)}", + "offers": [], + "brochures": [] + } + + search_results = data.get("searchResults", {}) + contents = search_results.get("contents", {}) + raw_offers = contents.get("offers", []) + raw_brochures = contents.get("brochures", []) + + formatted_offers = [] + for o in raw_offers: + prices = o.get("prices", {}) + price_fmt = prices.get("mainPriceFormatted") or "" + price_num = float(prices.get("mainPrice") or 0.0) + if not price_fmt and price_num > 0: + price_fmt = f"{price_num:.2f} €".replace(".", ",") + + title = o.get("title") or "" + brand = o.get("brand") or "" + desc = o.get("description") or "" + retailer = o.get("publisherName") or "Supermarkt" + + piece_info = calculate_piece_price(title, desc, price_num) + deal_rating = evaluate_deal(brand, title, price_num) + + offer_dict = { + "retailer": retailer, + "brand": brand, + "title": title, + "price": price_fmt, + "price_numeric": price_num, + "description": desc, + "valid_from": (o.get("validFrom") or "")[:10], + "valid_until": (o.get("validUntil") or "")[:10], + "deal_rating": deal_rating, + "image_url": o.get("offerImages", {}).get("url", {}).get("normal") + } + if piece_info: + offer_dict["piece_price"] = piece_info + + formatted_offers.append(offer_dict) + + # Sort offers: top deals first, then by price + formatted_offers.sort(key=lambda x: ( + 0 if x.get("deal_rating") == "TOP_DEAL_VORRAT" else (1 if x.get("deal_rating") == "GUTER_ANGEBOTSPREIS" else 2), + x.get("price_numeric", 9999) if x.get("price_numeric", 0) > 0 else 9999 + )) + + formatted_brochures = [] + for b in raw_brochures: + bc = b.get("content", {}) + formatted_brochures.append({ + "retailer": bc.get("publisher", {}).get("name"), + "brochure_title": bc.get("title"), + "page": bc.get("page"), + "distance_km": round(bc.get("distance", 0), 1), + "valid_until": (bc.get("validUntil") or "")[:10] + }) + + return { + "query": query, + "city": city or DEFAULT_CITY, + "coordinates": {"lat": use_lat, "lng": use_lng}, + "total_offers": len(formatted_offers), + "offers": formatted_offers, + "matching_brochures": formatted_brochures[:5] + } + + def compare_staples(self, staples: List[str], city: Optional[str] = None) -> Dict[str, Any]: + """Searches multiple staple products and aggregates results by product and store.""" + summary = {} + for s in staples: + s_clean = s.strip() + if s_clean: + summary[s_clean] = self.search(s_clean, city=city) + + return { + "city": city or DEFAULT_CITY, + "staples_searched": list(summary.keys()), + "results": summary + } + + +# ============================================================================== +# Model Context Protocol (MCP) stdio Server +# ============================================================================== + +MCP_TOOLS = [ + { + "name": "supermarket_search", + "description": "Search current regional supermarket, discounter, and drugstore discounts (Edeka, Rewe, Aldi Süd, Lidl, Kaufland, Netto, Penny, dm, Rossmann, V-Markt). Ideal for brand staples (Mutti, Oro di Parma, Barilla, Lavazza) and baby essentials (Windeln, Pampers). Defaults to Regensburg.", + "inputSchema": { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "Product or brand name to search (e.g. 'Mutti', 'Oro di Parma', 'Windeln', 'Pampers', 'Kaffee')" + }, + "city": { + "type": "string", + "description": "City name for regional offers (e.g. 'Regensburg', 'München', 'Nürnberg'). Defaults to Regensburg." + }, + "lat": { + "type": "number", + "description": "Optional latitude coordinate override" + }, + "lng": { + "type": "number", + "description": "Optional longitude coordinate override" + } + }, + "required": ["query"] + } + }, + { + "name": "supermarket_compare_staples", + "description": "Batch search multiple staple goods (e.g. ['Mutti', 'Oro di Parma'] or ['Windeln', 'Pampers']) to compare discounts across all local supermarkets in one call.", + "inputSchema": { + "type": "object", + "properties": { + "staples": { + "type": "array", + "items": {"type": "string"}, + "description": "List of staple brands or product categories to search, e.g. ['Mutti', 'Oro di Parma']" + }, + "city": { + "type": "string", + "description": "City name for regional offers. Defaults to Regensburg." + } + }, + "required": ["staples"] + } + } +] + + +def run_mcp_server(): + client = SupermarketClient() + + while True: + line = sys.stdin.readline() + if not line: + break + line = line.strip() + if not line: + continue + + try: + req = json.loads(line) + except Exception as e: + sys.stderr.write(f"Invalid JSON received: {e}\n") + continue + + msg_id = req.get("id") + method = req.get("method") + params = req.get("params", {}) + + if method == "initialize": + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "supermarket-deals-mcp", "version": "1.0.0"} + } + } + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + elif method == "notifications/initialized": + pass + + elif method == "ping": + resp = {"jsonrpc": "2.0", "id": msg_id, "result": {}} + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + elif method == "tools/list": + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": {"tools": MCP_TOOLS} + } + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + elif method == "tools/call": + tool_name = params.get("name") + args = params.get("arguments", {}) + try: + if tool_name == "supermarket_search": + data = client.search( + query=args.get("query", ""), + city=args.get("city"), + lat=args.get("lat"), + lng=args.get("lng") + ) + elif tool_name == "supermarket_compare_staples": + data = client.compare_staples( + staples=args.get("staples", []), + city=args.get("city") + ) + else: + raise ValueError(f"Unknown tool: {tool_name}") + + text_result = json.dumps(data, indent=2, ensure_ascii=False) + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": { + "content": [{"type": "text", "text": text_result}] + } + } + except Exception as e: + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": { + "isError": True, + "content": [{"type": "text", "text": f"Error executing {tool_name}: {str(e)}"}] + } + } + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + else: + if msg_id is not None: + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "error": {"code": -32601, "message": f"Method not found: {method}"} + } + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + +def run_cli(): + import argparse + parser = argparse.ArgumentParser(description="Supermarket Deals Tool & MCP Server") + subparsers = parser.add_subparsers(dest="command") + + sp_search = subparsers.add_parser("search", help="Search supermarket offers") + sp_search.add_argument("query", help="Search query (e.g. Mutti, Pampers, Windeln)") + sp_search.add_argument("--city", default="Regensburg", help="City name (default: Regensburg)") + + sp_comp = subparsers.add_parser("compare", help="Compare multiple staple items") + sp_comp.add_argument("items", help="Comma-separated items, e.g. 'Mutti,Oro di Parma'") + sp_comp.add_argument("--city", default="Regensburg", help="City name (default: Regensburg)") + + args = parser.parse_args() + client = SupermarketClient() + + if args.command == "search": + print(json.dumps(client.search(args.query, city=args.city), indent=2, ensure_ascii=False)) + elif args.command == "compare": + items = [i.strip() for i in args.items.split(",") if i.strip()] + print(json.dumps(client.compare_staples(items, city=args.city), indent=2, ensure_ascii=False)) + else: + parser.print_help() + + +if __name__ == "__main__": + if len(sys.argv) == 1 or (len(sys.argv) == 2 and sys.argv[1] == "--stdio"): + run_mcp_server() + else: + run_cli()