From 5376ddbc20ecb6e3e794edd1d87c6986affdedfe Mon Sep 17 00:00:00 2001 From: Moritz Graf Date: Wed, 16 Sep 2026 08:22:40 +0200 Subject: [PATCH] feat(skills): add regensburg_ris_mcp.py and universal_product_extractor.py --- .../scripts/regensburg_ris_mcp.py | 233 ++++++++++++++++++ .../scripts/universal_product_extractor.py | 160 ++++++++++++ 2 files changed, 393 insertions(+) create mode 100755 skills/regensburg-stadtrat/scripts/regensburg_ris_mcp.py create mode 100755 skills/shopping/scripts/universal_product_extractor.py diff --git a/skills/regensburg-stadtrat/scripts/regensburg_ris_mcp.py b/skills/regensburg-stadtrat/scripts/regensburg_ris_mcp.py new file mode 100755 index 0000000..987f735 --- /dev/null +++ b/skills/regensburg-stadtrat/scripts/regensburg_ris_mcp.py @@ -0,0 +1,233 @@ +#!/usr/bin/env python3 +# /// script +# dependencies = [ +# "beautifulsoup4>=4.12.0", +# "requests>=2.31.0", +# ] +# /// +""" +Regensburg Stadtrat (ALLRIS) MCP Server +Exposes municipal council documents, upcoming meetings, agenda items (TOPs), +and binding resolutions as Model Context Protocol (MCP) tools over stdio. +""" + +import os +import sys +import json +import traceback + +# Ensure current directory is on sys.path to import AllrisClient +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from regensburg_ris import AllrisClient + +TOOLS = [ + { + "name": "stadtrat_search", + "description": "Search Regensburg city council documents, resolutions (Beschlüsse), and motions (Vorlagen) by keyword (e.g. 'Kaufhof', 'Protzenweiherbrücke', 'Stadtbahn', 'Haushalt').", + "inputSchema": { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "Keyword to search in ALLRIS net (e.g. 'Brücke', 'Kaufhof', 'Verkehr')." + }, + "limit": { + "type": "integer", + "description": "Maximum number of results to return (default: 10).", + "default": 10 + } + }, + "required": ["query"] + } + }, + { + "name": "stadtrat_calendar", + "description": "Fetch upcoming municipal council and committee meetings (Stadtrat, Ausschüsse) in Regensburg.", + "inputSchema": { + "type": "object", + "properties": { + "days": { + "type": "integer", + "description": "Number of days in the future to search (default: 30).", + "default": 30 + }, + "from_date": { + "type": "string", + "description": "Start date in DD.MM.YYYY format (default: today)." + }, + "to_date": { + "type": "string", + "description": "End date in DD.MM.YYYY format." + } + } + } + }, + { + "name": "stadtrat_session", + "description": "Fetch detailed meeting agenda items (Tagesordnungspunkte / TOPs) and linked Vorlagen IDs for a meeting ID (SILFDNR).", + "inputSchema": { + "type": "object", + "properties": { + "session_id": { + "type": "string", + "description": "ALLRIS session ID (SILFDNR, e.g. '2665')." + } + }, + "required": ["session_id"] + } + }, + { + "name": "stadtrat_vorlage", + "description": "Inspect a specific council proposal (Vorlage) by VOLFDNR ID: extracts voting outcomes, responsible department, and verbatim resolution text.", + "inputSchema": { + "type": "object", + "properties": { + "vorlage_id": { + "type": "string", + "description": "ALLRIS proposal ID (VOLFDNR, e.g. '21362')." + }, + "convert_attachments": { + "type": "boolean", + "description": "Whether to convert linked PDF attachments to Markdown (default: false).", + "default": False + } + }, + "required": ["vorlage_id"] + } + }, + { + "name": "stadtrat_compile_topic", + "description": "Compile a multi-document synthesis of a topic across ALLRIS into the Obsidian knowledge-wiki.", + "inputSchema": { + "type": "object", + "properties": { + "topic": { + "type": "string", + "description": "Topic keyword (e.g. 'Protzenweiherbrücke', 'Kaufhof')." + }, + "vault": { + "type": "string", + "description": "Path to the Obsidian knowledge-wiki vault.", + "default": "/home/node/.openclaw/wiki/main" + } + }, + "required": ["topic"] + } + } +] + + +def run_mcp_server(): + client = AllrisClient() + + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + req = json.loads(line) + except json.JSONDecodeError: + continue + + method = req.get("method") + msg_id = req.get("id") + params = req.get("params", {}) + + if method == "initialize": + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": { + "protocolVersion": "2024-11-05", + "serverInfo": { + "name": "regensburg-stadtrat-mcp", + "version": "1.0.0" + }, + "capabilities": { + "tools": {} + } + } + } + sys.stdout.write(json.dumps(resp) + "\n") + sys.stdout.flush() + + elif method == "notifications/initialized": + pass + + elif method == "tools/list": + resp = { + "jsonrpc": "2.0", + "id": msg_id, + "result": { + "tools": 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 == "stadtrat_search": + data = client.search( + query=args.get("query", ""), + limit=int(args.get("limit", 10)) + ) + elif tool_name == "stadtrat_calendar": + data = client.get_calendar( + days=int(args.get("days", 30)), + from_date=args.get("from_date"), + to_date=args.get("to_date") + ) + elif tool_name == "stadtrat_session": + data = client.get_session( + silfdnr=str(args.get("session_id", "")) + ) + elif tool_name == "stadtrat_vorlage": + data = client.get_vorlage( + volfdnr=str(args.get("vorlage_id", "")), + convert_attachments=bool(args.get("convert_attachments", False)) + ) + elif tool_name == "stadtrat_compile_topic": + vault_path = args.get("vault", "/home/node/.openclaw/wiki/main") + data = client.compile_topic( + topic=args.get("topic", ""), + vault=vault_path + ) + 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)}\n{traceback.format_exc()}"}] + } + } + 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() + + +if __name__ == "__main__": + run_mcp_server() diff --git a/skills/shopping/scripts/universal_product_extractor.py b/skills/shopping/scripts/universal_product_extractor.py new file mode 100755 index 0000000..d1a54a8 --- /dev/null +++ b/skills/shopping/scripts/universal_product_extractor.py @@ -0,0 +1,160 @@ +#!/usr/bin/env python3 +# /// script +# dependencies = [ +# "beautifulsoup4>=4.12.0", +# "requests>=2.31.0", +# ] +# /// +""" +Universal E-Commerce Product Extractor (Tier 2 JSON-LD + OpenGraph + Microdata) +Extracts structured product data (title, brand, price, currency, availability, sku) +from arbitrary online stores in < 200ms using Schema.org metadata, bypassing heavy browser +overhead and bot detection. Falls back to Chromium CDP if JS rendering is strictly required. +""" + +import os +import sys +import json +import re +import urllib.request +import urllib.parse +from typing import Dict, Any, Optional, List + +DEFAULT_HEADERS = { + "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.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-US;q=0.8,en;q=0.7", +} + + +class UniversalProductExtractor: + def __init__(self, cdp_url: Optional[str] = None): + self.cdp_url = cdp_url or os.environ.get("OPENCLAW_CHROMIUM_CDP", "http://127.0.0.1:9222") + + def fetch_html(self, url: str, timeout: int = 15) -> str: + req = urllib.request.Request(url, headers=DEFAULT_HEADERS) + with urllib.request.urlopen(req, timeout=timeout) as response: + encoding = response.headers.get_content_charset() or "utf-8" + return response.read().decode(encoding, errors="replace") + + def extract_from_jsonld(self, html: str) -> List[Dict[str, Any]]: + """Extract all Schema.org Product objects from application/ld+json scripts.""" + products = [] + pattern = re.compile(r']*type=[\'"]application/ld\+json[\'"][^>]*>(.*?)', re.DOTALL | re.IGNORECASE) + for match in pattern.finditer(html): + raw_json = match.group(1).strip() + if not raw_json: + continue + try: + data = json.loads(raw_json) + except Exception: + continue + + items = data if isinstance(data, list) else [data] + for item in items: + if not isinstance(item, dict): + continue + # Support @graph structures + if "@graph" in item and isinstance(item["@graph"], list): + for sub in item["@graph"]: + if isinstance(sub, dict) and sub.get("@type") in ("Product", "IndividualProduct"): + products.append(self._normalize_schema_product(sub)) + elif item.get("@type") in ("Product", "IndividualProduct"): + products.append(self._normalize_schema_product(item)) + + return products + + def _normalize_schema_product(self, raw: dict) -> dict: + brand = raw.get("brand") + if isinstance(brand, dict): + brand = brand.get("name", "") + + offers = raw.get("offers", {}) + if isinstance(offers, list) and offers: + offers = offers[0] + + price = None + currency = "EUR" + availability = "Unknown" + + if isinstance(offers, dict): + price = offers.get("price") or offers.get("lowPrice") + currency = offers.get("priceCurrency", "EUR") + raw_avail = str(offers.get("availability", "")) + if "InStock" in raw_avail: + availability = "InStock" + elif "OutOfStock" in raw_avail: + availability = "OutOfStock" + elif "PreOrder" in raw_avail: + availability = "PreOrder" + + return { + "title": raw.get("name", "").strip(), + "brand": str(brand or "").strip(), + "price": float(price) if price and str(price).replace(".", "", 1).isdigit() else price, + "currency": currency, + "availability": availability, + "sku": str(raw.get("sku") or raw.get("gtin13") or raw.get("gtin") or "").strip(), + "image": raw.get("image") if isinstance(raw.get("image"), str) else (raw.get("image", [None])[0] if isinstance(raw.get("image"), list) else None), + "description": raw.get("description", "").strip()[:500] if raw.get("description") else None, + "extraction_tier": "json-ld" + } + + def extract_from_opengraph(self, html: str) -> Dict[str, Any]: + """Fallback extraction from OpenGraph / Twitter Cards / Meta tags.""" + title_m = re.search(r']*property=[\'"]og:title[\'"][^>]*content=[\'"](.*?)[\'"]', html, re.IGNORECASE) + price_m = re.search(r']*property=[\'"]product:price:amount[\'"][^>]*content=[\'"](.*?)[\'"]', html, re.IGNORECASE) or \ + re.search(r']*property=[\'"]og:price:amount[\'"][^>]*content=[\'"](.*?)[\'"]', html, re.IGNORECASE) + curr_m = re.search(r']*property=[\'"]product:price:currency[\'"][^>]*content=[\'"](.*?)[\'"]', html, re.IGNORECASE) or \ + re.search(r']*property=[\'"]og:price:currency[\'"][^>]*content=[\'"](.*?)[\'"]', html, re.IGNORECASE) + img_m = re.search(r']*property=[\'"]og:image[\'"][^>]*content=[\'"](.*?)[\'"]', html, re.IGNORECASE) + + title = title_m.group(1).strip() if title_m else "" + price = price_m.group(1).strip() if price_m else None + currency = curr_m.group(1).strip() if curr_m else "EUR" + image = img_m.group(1).strip() if img_m else None + + if title or price: + return { + "title": title, + "price": float(price) if price and price.replace(".", "", 1).isdigit() else price, + "currency": currency, + "image": image, + "extraction_tier": "opengraph" + } + return {} + + def extract(self, url: str) -> Dict[str, Any]: + """Main entry point: fetches URL, extracts JSON-LD, falls back to OpenGraph.""" + html = self.fetch_html(url) + jsonld_products = self.extract_from_jsonld(html) + if jsonld_products: + res = jsonld_products[0] + res["source_url"] = url + return res + + og_product = self.extract_from_opengraph(html) + if og_product: + og_product["source_url"] = url + return og_product + + return { + "error": "No structured product metadata found (JSON-LD or OpenGraph)", + "source_url": url, + "extraction_tier": "failed" + } + + +def main(): + if len(sys.argv) < 2: + print("Usage: universal_product_extractor.py ") + sys.exit(1) + + url = sys.argv[1] + extractor = UniversalProductExtractor() + result = extractor.extract(url) + print(json.dumps(result, indent=2, ensure_ascii=False)) + + +if __name__ == "__main__": + main()