← back to Newmor Onboard
scripts/refresh-scrape-feed-first.py
218 lines
#!/usr/bin/env python3
"""
Newmor feed-first $0 refresh scraper (TK-10670 — REVERSIBLE CORE).
Fetches https://newmor.com/product-sitemap.xml -> product URLs, plain-fetches each
product page (Mozilla UA, follow redirects, skip 404s, ~1 req/sec), parses per-colorway
rows from the STRUCTURED <p class="color-title">Colours: XXX</p> shade list (NOT a blind
regex, so image-dimension noise like II-1024/GB-0321/MSET-150 is avoided by construction),
and writes ALL rows to a NEW staging table dw_unified.newmor_catalog_refresh_20260909.
HARD RAILS: staging table only. NEVER touches newmor_catalog. No Shopify. $0 plain fetch.
"""
import re, sys, json, time, html, urllib.request, urllib.error, subprocess, os
SITEMAP = "https://newmor.com/product-sitemap.xml"
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120 Safari/537.36"
STAGING = "newmor_catalog_refresh_20260909"
OUT = os.path.join(os.path.dirname(__file__), "..", "data", "newmor-refresh-20260909")
os.makedirs(OUT, exist_ok=True)
def fetch(url, timeout=30):
req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "text/html,application/xml"})
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
return r.getcode(), r.geturl(), r.read().decode("utf-8", "replace")
except urllib.error.HTTPError as e:
return e.code, url, ""
except Exception as e:
return -1, url, f"__ERR__ {e}"
def clean(s):
return re.sub(r"\s+", " ", html.unescape(re.sub(r"<[^>]+>", " ", s or ""))).strip()
def parse_product(url, h):
# pattern name: prefer <h1>, fall back to <title>
pat = ""
m = re.search(r"<h1[^>]*>(.*?)</h1>", h, re.S)
if m:
pat = clean(m.group(1))
if not pat:
m = re.search(r"<title>(.*?)</title>", h, re.S)
if m:
pat = clean(m.group(1)).split(" - Newmor")[0].split("|")[0].strip()
# collection: breadcrumb chain minus Home/Products/pattern (usually empty for Newmor)
collection = ""
crumbs = [clean(x) for x in re.findall(r'class="breadcrumb"[^>]*>(.*?)</a>', h, re.S)]
mid = [c for c in crumbs if c.lower() not in ("home", "products", "product", pat.lower())]
if mid:
collection = mid[-1]
# specs
width = length = repeat_v = match_type = material = ""
for m in re.finditer(
r'technical-container__title">\s*(.*?)\s*</div>\s*<p class="technical-container__link\s*([^"]*)">\s*(.*?)\s*</p>',
h, re.S):
label = clean(m.group(1)).lower()
val = clean(m.group(3))
if "roll siz" in label or "width" in label:
mm = re.findall(r"(\d{3,4})\s*MM", val, re.I)
if len(mm) >= 2:
width = f"{mm[1]}mm (trimmed, {mm[0]}mm printed)"
elif mm:
width = f"{mm[0]}mm"
lm = re.search(r"(\d{1,3})\s*m\b", val)
if lm:
length = f"{lm.group(1)}m"
elif "repeat" in label or m.group(2).strip() == "repeat":
repeat_v = val
mt = re.search(r"(straight match|half drop|drop match|free match|random match|offset match|no match)", val, re.I)
if mt:
match_type = mt.group(1).title()
elif "match" in label:
match_type = val
elif "material" in label or "composition" in label:
material = val
# spec-sheet PDF often carries the material family (e.g. Aurelia-Mylar-Spec.pdf)
if not material:
pm = re.search(r'href="[^"]*Spec[^"]*\.pdf"', h, re.I)
if pm:
fam = re.search(r"-(Mylar|Vinyl|Textile|Paper|Foil|Suede|Metallic|Cork|Grasscloth)-?", pm.group(0), re.I)
if fam:
material = fam.group(1).title()
# ---- STRUCTURED shade list: each <p class="color-title">Colours: XXX</p> ----
colorways = []
# split page on color-title <p> so we can grab the image that follows each one
parts = re.split(r'(<p class="color-title"[^>]*>.*?</p>)', h, flags=re.S)
idx = 0
while idx < len(parts):
seg = parts[idx]
cm = re.match(r'<p class="color-title"[^>]*>(.*?)</p>', seg, re.S)
if cm:
raw = clean(cm.group(1))
code = re.sub(r"^colou?rs?\s*:\s*", "", raw, flags=re.I).strip()
img = ""
tail = parts[idx + 1] if idx + 1 < len(parts) else ""
im = re.search(r'data-fblink="([^"]+)"', tail) or re.search(r'data-colour="([^"]+)"', tail)
if im:
img = im.group(1)
if code:
colorways.append((code, img))
idx += 1
# dedupe by code, preserve order
seen, cw = set(), []
for code, img in colorways:
k = code.lower()
if k not in seen:
seen.add(k)
cw.append((code, img))
# hero fallback image
hero = ""
hm = re.search(r'data-fblink="([^"]+)"', h) or re.search(r'<meta property="og:image" content="([^"]+)"', h)
if hm:
hero = hm.group(1)
rows = []
if cw:
for code, img in cw:
mfr = re.sub(r"\s+", "-", code.upper())
rows.append(dict(mfr_sku=mfr, pattern_name=pat, color_name=code, collection=collection,
width=width, length=length, repeat_v=repeat_v, match_type=match_type,
material=material, image_url=(img or hero), product_url=url, in_stock=True))
else:
# single-colorway / no structured shade list -> one pattern-level row (logged as caveat)
mfr = "NEWMOR-" + re.sub(r"[^A-Z0-9]+", "-", pat.upper()).strip("-")
rows.append(dict(mfr_sku=mfr, pattern_name=pat, color_name="", collection=collection,
width=width, length=length, repeat_v=repeat_v, match_type=match_type,
material=material, image_url=hero, product_url=url, in_stock=True))
return rows, pat
def main():
print("Fetching sitemap...", flush=True)
code, _, xml = fetch(SITEMAP)
urls = re.findall(r"<loc>([^<]+)</loc>", xml)
urls = [u for u in urls if "/product/" in u]
print(f" sitemap HTTP {code}, {len(urls)} product URLs", flush=True)
all_rows, dead, errors, no_colorway = [], [], [], []
for i, u in enumerate(urls, 1):
st, final, h = fetch(u)
if st == 404 or st == 410:
dead.append({"url": u, "status": st})
elif st == 200 and h and not h.startswith("__ERR__"):
try:
rows, pat = parse_product(final, h)
all_rows.extend(rows)
if rows and rows[0]["color_name"] == "":
no_colorway.append({"url": u, "pattern": pat})
except Exception as e:
errors.append({"url": u, "err": str(e)})
else:
errors.append({"url": u, "status": st, "detail": (h[:120] if h else "")})
if i % 25 == 0:
print(f" {i}/{len(urls)} rows={len(all_rows)} dead={len(dead)} err={len(errors)}", flush=True)
time.sleep(1.0) # polite ~1 req/sec
# write artifacts
json.dump(all_rows, open(os.path.join(OUT, "rows.json"), "w"), indent=0)
json.dump({"dead": dead, "errors": errors, "no_colorway_patterns": no_colorway,
"urls_total": len(urls)}, open(os.path.join(OUT, "scrape-log.json"), "w"), indent=2)
print(f"\nScrape done: {len(all_rows)} rows, dead={len(dead)}, errors={len(errors)}, "
f"no-colorway={len(no_colorway)}", flush=True)
# ---- load into STAGING table (staging only; never newmor_catalog) ----
load_staging(all_rows)
def sql_lit(v):
if v is None:
return "NULL"
if isinstance(v, bool):
return "true" if v else "false"
return "'" + str(v).replace("'", "''") + "'"
def load_staging(rows):
print(f"Loading {len(rows)} rows into {STAGING} (staging copy of newmor_catalog shape)...", flush=True)
ddl = f"""
DROP TABLE IF EXISTS {STAGING};
CREATE TABLE {STAGING} (
id serial PRIMARY KEY,
mfr_sku text, pattern_name text, color_name text, collection text,
product_type text DEFAULT 'wallcovering',
width text, length text, repeat_v text, match_type text, material text,
image_url text, product_url text, in_stock boolean DEFAULT true,
last_scraped timestamptz
);
"""
run_sql(ddl)
B = 200
for i in range(0, len(rows), B):
chunk = rows[i:i + B]
vals = []
for r in chunk:
vals.append("(" + ",".join([
sql_lit(r["mfr_sku"]), sql_lit(r["pattern_name"]), sql_lit(r["color_name"]),
sql_lit(r["collection"] or None), sql_lit(r["width"] or None), sql_lit(r["length"] or None),
sql_lit(r["repeat_v"] or None), sql_lit(r["match_type"] or None), sql_lit(r["material"] or None),
sql_lit(r["image_url"] or None), sql_lit(r["product_url"]), sql_lit(r["in_stock"]), "now()"
]) + ")")
ins = (f"INSERT INTO {STAGING} (mfr_sku,pattern_name,color_name,collection,width,length,"
f"repeat_v,match_type,material,image_url,product_url,in_stock,last_scraped) VALUES "
+ ",".join(vals) + ";")
run_sql(ins)
print("Staging load complete.", flush=True)
def run_sql(sql):
p = subprocess.run(["psql", "-h", "/tmp", "-d", "dw_unified", "-v", "ON_ERROR_STOP=1", "-c", sql],
capture_output=True, text=True)
if p.returncode != 0:
print("SQL ERROR:", p.stderr[:500], file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()