Best Buy商品数据提取：用Python解析JSON-LD并导出CSV

python --version

import argparse
import csv
import json
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from html.parser import HTMLParser
from pathlib import Path


class JsonLdParser(HTMLParser):
    def __init__(self):
        super().__init__()
        self.inside = False
        self.buffer = []
        self.blocks = []

    def handle_starttag(self, tag, attrs):
        if tag == 'script' and dict(attrs).get('type', '').lower() == 'application/ld+json':
            self.inside = True
            self.buffer = []

    def handle_data(self, data):
        if self.inside:
            self.buffer.append(data)

    def handle_endtag(self, tag):
        if tag == 'script' and self.inside:
            self.blocks.append(''.join(self.buffer))
            self.inside = False


def walk(value):
    if isinstance(value, dict):
        yield value
        for child in value.values():
            yield from walk(child)
    elif isinstance(value, list):
        for child in value:
            yield from walk(child)


def has_type(obj, kind):
    types = obj.get('@type', [])
    return kind == types if isinstance(types, str) else isinstance(types, list) and kind in types


def text(value):
    return '' if value is None else str(value)


def named(value):
    return text(value.get('name')) if isinstance(value, dict) else text(value)


def money(value):
    if value is None or value == '' or isinstance(value, bool):
        raise ValueError('Missing or invalid price')
    try:
        amount = Decimal(str(value))
    except InvalidOperation as error:
        raise ValueError('Invalid price') from error
    if not amount.is_finite() or amount < 0:
        raise ValueError('Invalid price')
    return format(amount, 'f')


def parse_rows(source, expected_sku, observed_at, region):
    observed = datetime.fromisoformat(observed_at.replace('Z', '+00:00'))
    if observed.tzinfo is None:
        raise ValueError('observed-at must include a timezone')
    if not expected_sku.strip() or not region.strip():
        raise ValueError('sku and region must not be empty')
    source = Path(source)
    raw = source.read_text(encoding='utf-8-sig')
    if raw.lstrip().startswith(('{', '[')):
        blocks = [raw]
    else:
        parser = JsonLdParser()
        parser.feed(raw)
        blocks = parser.blocks
    products = [obj for block in blocks for obj in walk(json.loads(block))
                if has_type(obj, 'Product')]
    if not products:
        raise ValueError('No Product JSON-LD found')
    matches = [p for p in products if text(p.get('sku')) == expected_sku]
    if not matches:
        raise ValueError(f'SKU mismatch: expected {expected_sku}')
    rows = []
    for product in matches:
        rating = product.get('aggregateRating') or {}
        if not isinstance(rating, dict):
            raise ValueError('aggregateRating must be an object')
        offers = product.get('offers', [])
        offers = offers if isinstance(offers, list) else [offers]
        for offer in offers:
            if not isinstance(offer, dict) or not has_type(offer, 'Offer'):
                continue
            row = {
                'sku': expected_sku,
                'name': text(product.get('name')),
                'model': named(product.get('model')),
                'brand': named(product.get('brand')),
                'color': text(product.get('color')),
                'seller': named(offer.get('seller')),
                'condition': text(offer.get('itemCondition')),
                'price': money(offer.get('price')),
                'currency': text(offer.get('priceCurrency')),
                'availability': text(offer.get('availability')),
                'rating': text(rating.get('ratingValue')),
                'review_count': text(rating.get('reviewCount')),
                'input_file': source.name,
                'observed_at': observed.isoformat(),
                'region': region,
            }
            for key in ('seller', 'currency'):
                if not row[key].strip():
                    raise ValueError(f'Missing required field: {key}')
            rows.append(row)
    if not rows:
        raise ValueError('No individual Offer found for target SKU')
    return rows


def safe_cell(value):
    value = text(value)
    candidate = value.lstrip(' \t\r\n')
    return "'" + value if candidate[:1] in {'=', '+', '-', '@'} or value[:1] in {'\t', '\r', '\n'} else value


def export_csv(rows, output):
    if not rows:
        raise ValueError('No rows to export')
    exported_at = datetime.now(timezone.utc).isoformat()
    with Path(output).open('x', newline='', encoding='utf-8-sig') as handle:
        writer = csv.DictWriter(handle, fieldnames=[*rows[0], 'exported_at'])
        writer.writeheader()
        for row in rows:
            writer.writerow({key: safe_cell(value) for key, value in
                             {**row, 'exported_at': exported_at}.items()})


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('input')
    parser.add_argument('--sku', required=True)
    parser.add_argument('--observed-at', required=True)
    parser.add_argument('--region', required=True)
    parser.add_argument('--output', default='bestbuy.csv')
    args = parser.parse_args()
    try:
        rows = parse_rows(args.input, args.sku, args.observed_at, args.region)
        export_csv(rows, args.output)
    except (ValueError, OSError) as error:
        parser.exit(1, f'ERROR: {error}\n')
    print(f'PASS: {len(rows)} Offers exported to {args.output}')


if __name__ == '__main__':
    main()

{
  "@context": "https://schema.org",
  "@type": "Product",
  "sku": "DEMO-SKU-001",
  "name": "Demo headphones",
  "brand": {"@type": "Brand", "name": "Demo Brand"},
  "aggregateRating": {
    "@type": "AggregateRating",
    "ratingValue": "4.5",
    "reviewCount": 12
  },
  "offers": [
    {
      "@type": "Offer",
      "seller": {"name": "Demo Seller"},
      "itemCondition": "https://schema.org/NewCondition",
      "price": "199.99",
      "priceCurrency": "USD"
    },
    {
      "@type": "Offer",
      "seller": {"name": "Demo Refurbished Seller"},
      "itemCondition": "https://schema.org/RefurbishedCondition",
      "price": "149.99",
      "priceCurrency": "USD"
    }
  ]
}

python bestbuy_parser.py product.json --sku DEMO-SKU-001 --observed-at 2026-09-24T00:00:00+00:00 --region DEMO --output bestbuy-demo.csv

PASS: 2 Offers exported to bestbuy-demo.csv

python -m pip install "requests==2.32.5"

$env:ROLA_PROXY_USER = Read-Host "请输入代理用户名"
$proxySecret = Read-Host "请输入代理密码" -AsSecureString
$env:ROLA_PROXY_PASSWORD = (New-Object System.Net.NetworkCredential('', $proxySecret)).Password
$env:ROLA_PROXY_HOST = Read-Host "请输入代理地址（不含协议和端口）"
$env:ROLA_PROXY_PORT = Read-Host "请输入代理端口"

import os
from urllib.parse import quote
import requests

user = quote(os.environ["ROLA_PROXY_USER"], safe="")
password = quote(os.environ["ROLA_PROXY_PASSWORD"], safe="")
host = os.environ["ROLA_PROXY_HOST"]
port = os.environ["ROLA_PROXY_PORT"]
proxy = f"http://{user}:{password}@{host}:{port}"

with requests.Session() as session:
    session.trust_env = False
    session.proxies.update({"http": proxy, "https": proxy})
    try:
        response = session.get("https://ipinfo.io/json", timeout=20)
        response.raise_for_status()
        data = response.json()
        if not data.get("ip"):
            raise ValueError("Diagnostic response has no IP address")
        print({"ip": data["ip"], "country": data.get("country")})
    except requests.exceptions.HTTPError:
        print("Stopped:", response.status_code,
              "Retry-After:", response.headers.get("Retry-After"))
        raise
    except (requests.exceptions.RequestException, ValueError):
        print("Proxy diagnostic failed; check connection and response.")
        raise