AliExpress Product Scraper: Parse Saved HTML with Python

python --version
python -m pip install requests==2.34.2 beautifulsoup4==4.15.0

"""Strict Product JSON-LD parser and optional, bounded AliExpress fetcher."""
import argparse
import csv
import json
import os
import re
import sys
import tempfile
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from urllib.parse import urljoin, urlsplit

import requests
from bs4 import BeautifulSoup

FIELDS = ['product_id', 'title', 'currency', 'price', 'availability', 'rating',
          'review_count', 'processed_at_utc', 'input_url', 'final_url', 'source_kind']
REDIRECTS = {301, 302, 303, 307, 308}


def product_id_from_url(url, expected_id=None):
    try:
        u = urlsplit(url)
        valid_host = u.hostname in {'aliexpress.com', 'www.aliexpress.com'}
        valid_port = u.port in (None, 443)
    except (ValueError, TypeError):
        raise ValueError('invalid product URL') from None
    match = re.fullmatch(r'/item/([0-9]+)\.html', u.path)
    if (u.scheme != 'https' or not valid_host or not valid_port or u.username
            or u.password or u.query or u.fragment or not match):
        raise ValueError('use an HTTPS aliexpress.com /item/ID.html URL without query or fragment')
    pid = match.group(1)
    if expected_id is not None and pid != expected_id:
        raise ValueError('redirect or Product identity mismatch')
    return pid


def has_type(obj, wanted):
    types = obj.get('@type', [])
    if isinstance(types, str):
        types = [types]
    return isinstance(types, list) and any(t in {wanted, 'https://schema.org/' + wanted,
                                                'http://schema.org/' + wanted} for t in types if isinstance(t, str))


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


def candidate_matches(item, requested_id):
    identities = []
    for key in ('url', '@id'):
        value = item.get(key)
        if not value:
            continue
        if not isinstance(value, str):
            return False
        # A fragment may identify the Product entity within its own page.
        try:
            identities.append(product_id_from_url(value.split('#', 1)[0]))
        except ValueError:
            return False
    # Never equate a seller SKU with the marketplace's product ID.
    return bool(identities) and all(pid == requested_id for pid in identities)


def numeric(value, label):
    if isinstance(value, bool):
        raise ValueError(label + ' must be numeric')
    try:
        number = Decimal(str(value))
    except (InvalidOperation, ValueError):
        raise ValueError(label + ' must be numeric') from None
    if not number.is_finite() or number < 0:
        raise ValueError(label + ' must be finite and nonnegative')
    return format(number, 'f')


def parse_product(html, input_url, final_url=None, source_kind='saved_html'):
    requested_id = product_id_from_url(input_url)
    final_url = final_url or input_url
    product_id_from_url(final_url, requested_id)
    soup = BeautifulSoup(html, 'html.parser')
    # These are sample stop markers, not a universal challenge detector.
    visible = soup.get_text(' ', strip=True).lower()
    if any(marker in visible for marker in ('slide to verify', 'verify you are human',
                                             '请拖动下方滑块完成验证')):
        raise ValueError('verification page returned')
    candidates = []
    for script in soup.find_all('script', attrs={'type': 'application/ld+json'}):
        try:
            data = json.loads(script.string or script.get_text(), parse_float=Decimal)
        except (json.JSONDecodeError, TypeError):
            continue
        candidates.extend(n for n in walk(data) if has_type(n, 'Product')
                          and candidate_matches(n, requested_id))
    if len(candidates) != 1:
        raise ValueError('requested Product identity is missing or ambiguous')
    product = candidates[0]
    title = product.get('name')
    if not isinstance(title, str) or not title.strip():
        raise ValueError('product title is missing')
    offer = product.get('offers')
    if isinstance(offer, list) and len(offer) == 1:
        offer = offer[0]
    if (not isinstance(offer, dict) or not has_type(offer, 'Offer')
            or has_type(offer, 'AggregateOffer') or 'lowPrice' in offer or 'highPrice' in offer):
        raise ValueError('a single Offer with one price is required')
    price = numeric(offer.get('price'), 'price')
    currency = str(offer.get('priceCurrency', '')).strip().upper()
    if not re.fullmatch('[A-Z]{3}', currency):
        raise ValueError('currency must contain three ASCII letters')
    availability = str(offer.get('availability', '')).rstrip('/').rsplit('/', 1)[-1]
    if availability not in {'InStock', 'OutOfStock'}:
        availability = 'Unknown'
    rating = product.get('aggregateRating', {})
    rating = rating if isinstance(rating, dict) else {}
    optional = {}
    for key, output in [('ratingValue', 'rating'), ('reviewCount', 'review_count')]:
        value = rating.get(key)
        optional[output] = ''
        if value is not None and value != '':
            try:
                parsed = numeric(value, key)
                if key != 'reviewCount' or Decimal(parsed) == Decimal(parsed).to_integral_value():
                    optional[output] = parsed
            except ValueError:
                pass
    return dict(product_id=requested_id, title=title.strip(), currency=currency,
                price=price, availability=availability, **optional,
                processed_at_utc=datetime.now(timezone.utc).isoformat(timespec='seconds'),
                input_url=input_url, final_url=final_url, source_kind=source_kind)


def fetch_product(url, session):
    requested_id = product_id_from_url(url)
    current = url
    for hop in range(6):
        with session.get(current, timeout=(10, 20), allow_redirects=False, stream=True) as response:
            if response.status_code in REDIRECTS:
                if hop == 5 or not response.headers.get('Location'):
                    raise ValueError('redirect limit exceeded or Location missing')
                target = urljoin(current, response.headers['Location'])
                product_id_from_url(target, requested_id)
                current = target
                continue
            if response.status_code == 429:
                raw = response.headers.get('Retry-After', 'not supplied')
                hint = ''.join(c for c in raw if c.isprintable())[:100]
                raise ValueError('HTTP 429; stop. Retry-After: ' + hint)
            if response.status_code != 200:
                raise ValueError('HTTP ' + str(response.status_code) + '; stop current request')
            chunks, size = [], 0
            for chunk in response.iter_content(65536):
                size += len(chunk)
                if size > 5_000_000:
                    raise ValueError('HTML exceeds 5 MB limit')
                chunks.append(chunk)
            html = b''.join(chunks).decode(response.encoding or 'utf-8', errors='replace')
            return parse_product(html, url, current, source_kind='online_response')
    raise ValueError('redirect limit exceeded')


def safe_cell(value):
    text = str(value)
    # Protect spreadsheet users from formula-like untrusted product titles.
    return "'" + text if text.lstrip().startswith(('=', '+', '-', '@')) else value


def write_csv(path, rows, fields=FIELDS):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    name = None
    try:
        with tempfile.NamedTemporaryFile('w', encoding='utf-8-sig', newline='',
                                         dir=path.parent, delete=False) as f:
            name = f.name
            writer = csv.DictWriter(f, fieldnames=fields)
            writer.writeheader()
            writer.writerows({key: safe_cell(row[key]) for key in fields} for row in rows)
        os.replace(name, path)
    finally:
        if name and os.path.exists(name):
            os.unlink(name)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('url', help='HTTPS product-detail URL, without tracking parameters')
    parser.add_argument('--html', type=Path, help='parse saved HTML without any network request')
    parser.add_argument('--output', type=Path, default=Path('product.csv'))
    parser.add_argument('--authorized', action='store_true', help='confirm permission for an online request')
    args = parser.parse_args()
    try:
        if args.html:
            row = parse_product(args.html.read_text(encoding='utf-8'), args.url)
        else:
            if not args.authorized:
                parser.error('online requests require --authorized; use --html for offline parsing')
            with requests.Session() as session:
                row = fetch_product(args.url, session)
        write_csv(args.output, [row])
        print('Wrote 1 record to ' + str(args.output))
    except requests.RequestException as exc:
        print('STOP: ' + type(exc).__name__ + '; check connection, proxy and credentials locally', file=sys.stderr)
        return 1
    except (ValueError, OSError) as exc:
        print('STOP: ' + str(exc), file=sys.stderr)
        return 1
    return 0


if __name__ == '__main__':
    sys.exit(main())

"""Create eight explicitly synthetic parser fixtures; no network requests."""
import json
from pathlib import Path

EXAMPLES = [
    ('USB-C 7-in-1 Hub', 'USD', '24.90', 'InStock', '4.6', 328),
    ('Portable Laptop Stand', 'USD', '19.50', 'InStock', '4.5', 217),
    ('Compact Cable Organizer', 'USD', '8.25', 'InStock', None, None),
    ('Travel Packing Cubes', 'EUR', '17.30', 'InStock', '4.2', 84),
    ('Foldable Travel Bottle', 'EUR', '11.80', 'OutOfStock', None, None),
    ('Desk Cable Clips', 'EUR', '5.40', 'InStock', '4.1', 63),
    ('Adjustable Desk Lamp', 'JPY', '2480', 'InStock', '4.7', 105),
    ('Canvas Pencil Case', 'JPY', '780', 'OutOfStock', None, None),
]


def build():
    folder = Path(__file__).parent / 'saved_html'
    folder.mkdir(exist_ok=True)
    manifest = []
    for i, (title, currency, price, stock, rating, count) in enumerate(EXAMPLES, 100000001):
        url = f'https://www.aliexpress.com/item/{i}.html'
        product = {'@context': 'https://schema.org', '@type': 'Product', 'url': url,
                   'name': title, 'offers': {'@type': 'Offer', 'price': price,
                   'priceCurrency': currency, 'availability': 'https://schema.org/' + stock}}
        if rating is not None:
            product['aggregateRating'] = {'@type': 'AggregateRating', 'ratingValue': rating, 'reviewCount': count}
        html = '<!doctype html><html lang="en"><meta charset="utf-8"><title>Synthetic fixture</title><body><h1>Synthetic parser fixture; not a marketplace capture</h1><script type="application/ld+json">' + json.dumps(product) + '</script></body></html>'
        (folder / f'{i}.html').write_text(html, encoding='utf-8')
        manifest.append({'file': f'{i}.html', 'url': url, 'source_kind': 'synthetic_fixture'})
    (folder / 'manifest.json').write_text(json.dumps(manifest, indent=2), encoding='utf-8')
    print('Created 8 synthetic HTML fixtures; no marketplace requests were made.')


if __name__ == '__main__':
    build()

"""Parse the delivered synthetic HTML and export CSV summaries."""
import json
from collections import Counter
from pathlib import Path
from article_ali_scraper import parse_product, write_csv


def main():
    root = Path(__file__).parent
    folder = root / 'saved_html'
    manifest = json.loads((folder / 'manifest.json').read_text(encoding='utf-8'))
    rows = [parse_product((folder / entry['file']).read_text(encoding='utf-8'),
                          entry['url'], source_kind=entry['source_kind']) for entry in manifest]
    write_csv(root / 'product_fields.csv', rows)
    summaries = []
    for currency in sorted({r['currency'] for r in rows}):
        counts = Counter(r['availability'] for r in rows if r['currency'] == currency)
        summaries.append(dict(currency=currency, in_stock=counts['InStock'],
                              out_of_stock=counts['OutOfStock'], unknown=counts['Unknown'], total=sum(counts.values())))
    write_csv(root / 'currency_availability_summary.csv', summaries,
              ['currency', 'in_stock', 'out_of_stock', 'unknown', 'total'])
    (root / 'product_fields.json').write_text(json.dumps(rows, indent=2), encoding='utf-8')
    for row in rows:
        print(f"{row['product_id']} | {row['currency']} {row['price']} | {row['availability']}")
    print('Wrote 8 synthetic records to product_fields.csv')
    for item in summaries:
        print(item)


if __name__ == '__main__':
    main()

python build_examples.py
python run_local_pipeline.py

python article_ali_scraper.py "https://www.aliexpress.com/item/PRODUCT_ID.html" --html saved-product.html --output product.csv

from pathlib import Path
from article_ali_scraper import parse_product

html = Path("saved_html/100000001.html").read_text(encoding="utf-8")
record = parse_product(
    html,
    "https://www.aliexpress.com/item/100000001.html",
    source_kind="synthetic_fixture",
)
print(record["product_id"], record["currency"], record["price"])

=COUNTIFS('Product Fields'!$C$5:$C$12,$A5,'Product Fields'!$E$5:$E$12,"InStock")

=COUNTBLANK('Product Fields'!F5:F12)

python article_ali_scraper.py "https://www.aliexpress.com/item/PRODUCT_ID.html" --authorized --output online-product.csv

export HTTPS_PROXY="http://USERNAME:PASSWORD@HOST:PORT"

$env:HTTPS_PROXY = "http://USERNAME:PASSWORD@HOST:PORT"