feat(skills): add regensburg_ris_mcp.py and universal_product_extractor.py
This commit is contained in:
Moritz Graf
+160
@@ -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'<script[^>]*type=[\'"]application/ld\+json[\'"][^>]*>(.*?)</script>', 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'<meta[^>]*property=[\'"]og:title[\'"][^>]*content=[\'"](.*?)[\'"]', html, re.IGNORECASE)
|
||||
price_m = re.search(r'<meta[^>]*property=[\'"]product:price:amount[\'"][^>]*content=[\'"](.*?)[\'"]', html, re.IGNORECASE) or \
|
||||
re.search(r'<meta[^>]*property=[\'"]og:price:amount[\'"][^>]*content=[\'"](.*?)[\'"]', html, re.IGNORECASE)
|
||||
curr_m = re.search(r'<meta[^>]*property=[\'"]product:price:currency[\'"][^>]*content=[\'"](.*?)[\'"]', html, re.IGNORECASE) or \
|
||||
re.search(r'<meta[^>]*property=[\'"]og:price:currency[\'"][^>]*content=[\'"](.*?)[\'"]', html, re.IGNORECASE)
|
||||
img_m = re.search(r'<meta[^>]*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 <product_url>")
|
||||
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()
|
||||
Reference in New Issue
Block a user