← back to Japan Enrich

scrape_sangetsu_specs.py

291 lines

#!/usr/bin/env python3
"""
Sangetsu product-SPEC backfill — fills spec.width / spec.type (and material,
weight, backing, repeat, match) for the ~557 patterns whose original scrape
left spec.width AND spec.type both null. Those empty specs leave the japan.
viewer's Width facet and Fire-class facet blank.

Source: sangetsu-goodrich.co.th WooCommerce Store API (public; WP REST proper
is Kadence-locked, only /wc/store/v1 is open):
  GET /wc/store/v1/products?slug=<slug>   → parent; its "description" HTML holds
                                            the spec block.

The description HTML contains a spec block after either a "Specifications" or a
"Technical Specifications" header, as label:value pairs, e.g.:
  Material: Vinyl  Width1: 54 IN (1.37 M)  Type / Fire Rating: Type II / Class A
  Vertical Repeat: 3.0 in  Horizontal Repeat: 3.375 in
  Hang: Straight Match, Non-Reverse  Backing: Non-woven
A few residential rows use a freeform metric form ("... width 0.91 m") with no
label:value block — we recover width from that as a fallback.

SAFE / reversible: read-only public fetch + local atomic rewrite (temp+rename)
of the git-tracked staging/sangetsu-staging.jsonl. Only rows currently missing
BOTH spec.width and spec.type are touched; every other row is written back
byte-identical.

Usage: scrape_sangetsu_specs.py [--limit N]
"""
import gzip
import html
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request

ROOT = os.path.dirname(os.path.abspath(__file__))
STAGING = os.path.join(ROOT, "staging", "sangetsu-staging.jsonl")
API = "https://www.sangetsu-goodrich.co.th/wp-json/wc/store/v1"
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) japan-enrich/sangetsu-specs"

LIMIT = int(sys.argv[sys.argv.index("--limit") + 1]) if "--limit" in sys.argv else None


def slug_of(url):
    return (url or "").rstrip("/").rsplit("/", 1)[-1]


def get(path, tries=4):
    for i in range(tries):
        try:
            req = urllib.request.Request(
                API + path,
                headers={"User-Agent": UA, "Accept": "application/json",
                         "Accept-Encoding": "gzip"},
            )
            with urllib.request.urlopen(req, timeout=45) as r:
                data = r.read()
                if r.headers.get("Content-Encoding") == "gzip":
                    data = gzip.decompress(data)
                return json.loads(data.decode("utf-8"))
        except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError):
            if i == tries - 1:
                return None
            time.sleep(1.2 * (i + 1))
    return None


def clean_text(desc_html):
    """Strip tags + decode entities + collapse whitespace from description HTML.

    Entity decode is critical: widths are published with &#8243; (double-prime
    ″) and &#8221; etc., which the width regex must see decoded, not raw.
    """
    t = re.sub(r"<[^>]+>", " ", desc_html or "")
    t = html.unescape(t)          # &#8243;->″ &nbsp;->\xa0 &reg;->® &amp;->&
    t = t.replace("\xa0", " ").replace("®", "")
    t = re.sub(r"\s+", " ", t).strip()
    return t


# Labels we harvest from the spec block, in the order they appear.
# Each entry: (regex label alternation) -> canonical key. Multiple layouts exist
# on the vendor site, so this list is deliberately broad; width is resolved by a
# dedicated priority ladder below rather than a single label.
_LABELS = [
    (r"Material", "material"),
    (r"Width1|Width", "width_lbl"),
    (r"Type\s*/\s*Fire\s*Rating|Type\s*/\s*Fire|Type\s*&\s*Fire\s*Rating|Fire\s*Rating|Type", "type"),
    (r"Vertical\s*Repeat", "vrepeat"),
    (r"Horizontal\s*Repeat", "hrepeat"),
    (r"Hang(?:\s*/\s*Match)?|Match", "match"),
    (r"Backing", "backing"),
    (r"Weight", "weight"),
    (r"Brand", "brand"),
    (r"Origin", "origin"),
    (r"Bolt\s*Size", "bolt"),
    (r"Dimension\s*Per\s*Unit|Dimensions|Dimension", "dimension"),
    (r"Description", "descr"),
]

# Union of every label token, used to find where the NEXT field starts so a
# value doesn't greedily swallow the following label.
_ALL_LABEL_ALT = "|".join(alt for alt, _ in _LABELS)
_NEXT = re.compile(r"\s*(?:%s)\s*:" % _ALL_LABEL_ALT, re.I)


def _value_after(text, label_alt):
    """Return the value string following 'label:' up to the next known label."""
    m = re.search(r"(?:%s)\s*:\s*" % label_alt, text, re.I)
    if not m:
        return None
    rest = text[m.end():]
    nm = _NEXT.search(rest)
    val = rest[: nm.start()] if nm else rest
    val = val.strip(" .,–-")
    return val or None


# A width token: a number + a length unit (in / inches / IN / cm / m / mm / ″ / ")
# — deliberately EXCLUDES "yd/yards" so "54″ x 30 yards" yields the 54″ width,
# not the 30-yard bolt length. An optional bracketed metric equivalent
# ("(1.37 M)") is captured whole.
_WIDTH_TOK = re.compile(
    r"(\d+(?:\.\d+)?\s*(?:inch(?:es)?|in\.?|IN|cm|mm|m|″|\")(?![a-z])"
    r"(?:\s*\([^)]{0,24}\))?)",
    re.I,
)


def _first_width(s):
    """Extract the first plausible width token from a string, or None."""
    if not s:
        return None
    m = _WIDTH_TOK.search(s)
    if not m:
        return None
    return m.group(1).strip(" .,")


def parse_spec(desc_html):
    """Parse a spec dict from a product description. Returns {} if no block."""
    text = clean_text(desc_html)
    # Isolate the region after a Specifications / Technical Specifications header
    hm = re.search(r"(Technical\s+Specifications|Specifications)", text, re.I)
    region = text[hm.end():] if hm else text

    raw = {}
    for alt, key in _LABELS:
        v = _value_after(region, alt)
        if v:
            raw[key] = v

    out = {}
    for k in ("material", "type", "vrepeat", "hrepeat", "match", "weight"):
        if raw.get(k):
            out[k] = raw[k]

    # Backing: reject when the field is actually a full material sentence
    # ("Vinyl Fabric Back", "non-woven backing") that duplicates Material.
    b = raw.get("backing")
    if b and len(b) <= 40 and not re.search(r"\bfabric back\b", b, re.I):
        out["backing"] = b

    # ---- width priority ladder --------------------------------------------
    # 1) explicit Width/Width1 label
    # 2) "... width 0.52 m" freeform (Length L x width W)
    # 3) Dimension / Dimension Per Unit / Dimensions field
    # 4) Description field ("... on non-woven backing Dimensions 91cm")
    width = _first_width(raw.get("width_lbl"))
    if not width:
        fm = re.search(r"\bwidth\s+([\d.]+\s*(?:m|cm|mm|in\.?|inch(?:es)?|IN)\b)", region, re.I)
        if fm:
            width = fm.group(1).strip()
    if not width:
        width = _first_width(raw.get("dimension"))
    if not width:
        # standalone "WIDTH: 54 INCHES" all-caps variant already covered by
        # width_lbl, but Description-embedded dimensions are not:
        width = _first_width(raw.get("descr"))
    if width:
        out["width"] = width

    return out


GRASS_KW = re.compile(r"grasscloth|grass\s*cloth|sisal|jute|hemp|paperweave|natural\s*fiber", re.I)
METAL_KW = re.compile(r"metallic|foil|mylar|gilt|gold\s*leaf|silver\s*leaf", re.I)


def apply_spec(row, parsed):
    """Merge parsed values into row['spec']; return True if width or type set."""
    spec = row.get("spec") or {}
    changed = False

    if parsed.get("width") and not spec.get("width"):
        spec["width"] = parsed["width"]
        changed = True
    if parsed.get("type") and not spec.get("type"):
        spec["type"] = parsed["type"]
        changed = True
    # NA repeats -> skip; only meaningful values
    if parsed.get("weight") and not spec.get("weight"):
        spec["weight"] = parsed["weight"]
    if parsed.get("backing") and not spec.get("backing"):
        spec["backing"] = parsed["backing"]
    if parsed.get("material"):
        spec["material"] = parsed["material"]
    if parsed.get("match"):
        spec["match"] = parsed["match"]

    vr = parsed.get("vrepeat")
    hr = parsed.get("hrepeat")
    def _real(x):
        return x and not re.fullmatch(r"n/?a", x.strip(), re.I)
    if _real(vr) or _real(hr):
        parts = [p for p in (vr, hr) if _real(p)]
        spec["repeat"] = " x ".join(parts)

    # material-keyword flags (only flip to True when obvious; never clear)
    mat = (parsed.get("material") or "") + " " + (parsed.get("backing") or "")
    if GRASS_KW.search(mat):
        spec["is_grasscloth"] = True
    if METAL_KW.search(mat):
        spec["is_metallic"] = True

    row["spec"] = spec
    return changed


def main():
    rows = [json.loads(l) for l in open(STAGING) if l.strip()]

    def sp(r):
        return r.get("spec") or {}

    targets = [r for r in rows if not sp(r).get("width") and not sp(r).get("type")]
    todo = targets[:LIMIT] if LIMIT else targets
    print(f"sangetsu specs: {len(rows)} rows, {len(targets)} missing width+type, doing {len(todo)}",
          flush=True)

    got_width = got_type = parsed_any = no_block = fetch_fail = 0
    for i, r in enumerate(todo, 1):
        slug = slug_of(r.get("source_url"))
        plist = get(f"/products?slug={slug}")
        if not (isinstance(plist, list) and plist):
            fetch_fail += 1
            print(f"  [{i}/{len(todo)}] {(r.get('pattern') or '?'):26} — FETCH FAIL ({slug})", flush=True)
            time.sleep(0.25)
            continue
        parsed = parse_spec(plist[0].get("description") or "")
        if parsed:
            changed = apply_spec(r, parsed)
            r["spec_backfilled"] = True
            if r["spec"].get("width"):
                got_width += 1
            if r["spec"].get("type"):
                got_type += 1
            if changed:
                parsed_any += 1
            bits = []
            if r["spec"].get("width"):
                bits.append("w=" + r["spec"]["width"])
            if r["spec"].get("type"):
                bits.append("type=" + r["spec"]["type"])
            print(f"  [{i}/{len(todo)}] {(r.get('pattern') or '?'):26} {'  '.join(bits) or '(spec fields, no w/type)'}",
                  flush=True)
        else:
            no_block += 1
            print(f"  [{i}/{len(todo)}] {(r.get('pattern') or '?'):26} — no spec block", flush=True)
        time.sleep(0.25)  # polite

    tmp = STAGING + ".tmp"
    with open(tmp, "w") as f:
        for r in rows:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    os.replace(tmp, STAGING)

    total_width = sum(1 for r in rows if sp(r).get("width"))
    total_type = sum(1 for r in rows if sp(r).get("type"))
    still = sum(1 for r in rows if not sp(r).get("width") and not sp(r).get("type"))
    print(f"\ndone. this pass: +{got_width} width, +{got_type} type "
          f"({parsed_any} rows changed), {no_block} no-block, {fetch_fail} fetch-fail")
    print(f"catalog now: {total_width} patterns with width, {total_type} with type, "
          f"{still} still missing both  → {STAGING}", flush=True)


if __name__ == "__main__":
    main()