product-recognition-demo/recognization/data/external-eval/kewpie/_fetch.py

"""Fetch helper for the kewpie external-proxy collection.

Kept in the owned directory so every download is reproducible from the
manifest's source_page. Not part of the product code.
"""
import gzip
import json
import re
import ssl
import sys
import urllib.parse
import urllib.request

try:
    import certifi

    _CTX = ssl.create_default_context(cafile=certifi.where())
except ImportError:  # pragma: no cover - certifi ships with this repo's venv
    _CTX = ssl.create_default_context()

UA = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
)


def get(url, referer=None, lang="ja,en-US;q=0.9,en;q=0.8"):
    req = urllib.request.Request(url)
    req.add_header("User-Agent", UA)
    req.add_header("Accept", "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8")
    req.add_header("Accept-Language", lang)
    req.add_header("Accept-Encoding", "gzip, deflate")
    if referer:
        req.add_header("Referer", referer)
    with urllib.request.urlopen(req, timeout=40, context=_CTX) as r:
        raw = r.read()
        if r.headers.get("Content-Encoding") == "gzip":
            raw = gzip.decompress(raw)
        return raw, r.headers.get("Content-Type", "")


def text(url, **kw):
    raw, ctype = get(url, **kw)
    for enc in ("utf-8", "shift_jis", "euc-jp"):
        try:
            return raw.decode(enc)
        except UnicodeDecodeError:
            continue
    return raw.decode("utf-8", "replace")


def download(url, path, referer=None):
    raw, ctype = get(url, referer=referer)
    with open(path, "wb") as f:
        f.write(raw)
    return len(raw), ctype


def amazon_search(query, limit=20):
    """Return [(asin, title)] from an amazon.co.jp search."""
    h = text("https://www.amazon.co.jp/s?k=" + urllib.parse.quote(query))
    out = []
    for block in re.split(r'(?=<div[^>]*data-asin="[A-Z0-9]{10}")', h):
        m = re.match(r'<div[^>]*data-asin="([A-Z0-9]{10})"', block)
        if not m:
            continue
        asin = m.group(1)
        t = re.search(r'<h2[^>]*aria-label="([^"]+)"', block) or re.search(r"<h2[^>]*>(.*?)</h2>", block, re.S)
        title = re.sub(r"<[^>]+>", "", t.group(1)).strip() if t else ""
        if title and (asin, title) not in out:
            out.append((asin, title))
        if len(out) >= limit:
            break
    return out

def _json_after(h, marker, opener="[", closer="]"):
    """Extract the JSON literal that follows `marker` in an Amazon page."""
    i = h.find(marker)
    if i < 0:
        return None
    i = h.find(opener, i)
    if i < 0:
        return None
    depth, j = 0, i
    while j < len(h):
        if h[j] == opener:
            depth += 1
        elif h[j] == closer:
            depth -= 1
            if depth == 0:
                break
        j += 1
    try:
        return json.loads(h[i : j + 1])
    except json.JSONDecodeError:
        return None


def amazon_gallery(asin):
    """Return [(hiRes, variant)] for every gallery image on an amazon.co.jp page."""
    h = text("https://www.amazon.co.jp/dp/" + asin)
    m = re.search(r"'colorImages':\s*\{\s*'initial':\s*A\.\$\.parseJSON\('(\[.*?\])'\)", h, re.S)
    if not m:
        return []
    blob = m.group(1).encode().decode("unicode_escape")
    try:
        data = json.loads(blob)
    except json.JSONDecodeError:
        return []
    return [(d.get("hiRes"), d.get("variant")) for d in data]


def amazon_images(asin):
    """Return [(hiRes, variant)] for every gallery image on an amazon.co.jp page."""
    return amazon_gallery(asin)


if __name__ == "__main__":
    cmd = sys.argv[1]
    if cmd == "asearch":
        for asin, title in amazon_search(sys.argv[2]):
            print(asin, "|", title)
    elif cmd == "aimg":
        for url, variant in amazon_gallery(sys.argv[2]):
            print(url, "|", variant)
    elif cmd == "get":
        print(text(sys.argv[2])[: int(sys.argv[3]) if len(sys.argv) > 3 else 3000])