← back to Newmor Onboard
scripts/refresh-scrape-feed-first-v2.py
234 lines
#!/usr/bin/env python3
"""
Newmor feed-first $0 refresh scraper — V2 (TK-10670 2nd pass, REVERSIBLE CORE).
Improvement over v1: correct colorway-code truth.
- A colorway is a REAL VENDOR CODE iff its shade-list text contains a digit
(AUR 001, TEDLAR SKY1004, BLS11, Linear 15.1501). -> code_source='vendor_code',
mfr_sku = normalized code (spaces->dash, upper).
- A colorway with NO digit is genuinely CODE-LESS (Morris/Glyn Cottage/Clash of
the Tartans/Heron Sent/Twiggy/etc. present colorways by NAME only — verified:
no data-code, image filename is <Pattern>-<Color>.jpg, data-swatchid is a mere
index). -> code_source='synthesized_pattern_color', mfr_sku = <PATTERN-SLUG>-<COLOR>
(an obviously-synthesized composite key matching the existing catalog convention
e.g. WEEKEND-BLUE; NOT a fake vendor code).
family_prefix = leading alpha token of a real code (factual; used to split the
/product/healthcare/ collection into per-family groups). NULL for code-less.
Writes to a NEW staging table dw_unified.newmor_catalog_refresh_20260909_v2.
PRESERVES v1 table newmor_catalog_refresh_20260909 (does NOT drop it).
HARD RAILS: staging 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_v2"
OUT = os.path.join(os.path.dirname(__file__), "..", "data", "newmor-refresh-20260909-v2")
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 slug_upper(s):
return re.sub(r"[^A-Z0-9]+", "-", (s or "").upper()).strip("-")
def parse_product(url, h):
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 = ""
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]
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
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|Tedlar)-?", pm.group(0), re.I)
if fam:
material = fam.group(1).title()
# structured shade list
colorways = []
parts = re.split(r'(<p class="color-title"[^>]*>.*?</p>)', h, flags=re.S)
for idx in range(len(parts)):
cm = re.match(r'<p class="color-title"[^>]*>(.*?)</p>', parts[idx], re.S)
if cm:
raw = clean(cm.group(1))
code = re.sub(r"^colou?rs?\s*:\s*", "", raw, flags=re.I).strip()
tail = parts[idx + 1] if idx + 1 < len(parts) else ""
im = re.search(r'data-fblink="([^"]+)"', tail) or re.search(r'data-colour="([^"]+)"', tail)
img = im.group(1) if im else ""
if code:
colorways.append((code, img))
seen, cw = set(), []
for code, img in colorways:
k = code.lower()
if k not in seen:
seen.add(k); cw.append((code, img))
hero = ""
hm = re.search(r'data-fblink="([^"]+)"', h) or re.search(r'<meta property="og:image" content="([^"]+)"', h)
if hm:
hero = hm.group(1)
pat_slug = slug_upper(pat) or "NEWMOR"
rows = []
if cw:
for code, img in cw:
has_code = bool(re.search(r"\d", code))
if has_code:
mfr = re.sub(r"\s+", "-", code.upper())
color_name = code
code_source = "vendor_code"
fam = re.match(r"^([A-Za-z]+)", code)
family_prefix = fam.group(1).upper() if fam else ""
else:
color_name = code
mfr = f"{pat_slug}-{slug_upper(code)}"
code_source = "synthesized_pattern_color"
family_prefix = ""
rows.append(dict(mfr_sku=mfr, pattern_name=pat, color_name=color_name, 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,
code_source=code_source, family_prefix=family_prefix))
else:
rows.append(dict(mfr_sku=f"{pat_slug}", 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,
code_source="synthesized_pattern_only", family_prefix=""))
return rows, pat
def main():
print("Fetching sitemap...", flush=True)
code, _, xml = fetch(SITEMAP)
urls = [u for u in re.findall(r"<loc>([^<]+)</loc>", xml) 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 in (404, 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)
# dedupe by mfr_sku identity (keep first; prefer a non-healthcare page if a real code
# appears on both a dedicated page and the collection page)
by_key = {}
order = []
for r in all_rows:
k = r["mfr_sku"]
if k not in by_key:
by_key[k] = r; order.append(k)
else:
if "/product/healthcare/" in by_key[k]["product_url"] and "/product/healthcare/" not in r["product_url"]:
by_key[k] = r
deduped = [by_key[k] for k in order]
collapsed = len(all_rows) - len(deduped)
json.dump(deduped, 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), "rows_raw": len(all_rows), "rows_deduped": len(deduped),
"dupes_collapsed": collapsed}, open(os.path.join(OUT, "scrape-log.json"), "w"), indent=2)
print(f"\nScrape done: raw={len(all_rows)} deduped={len(deduped)} (collapsed {collapsed}), "
f"dead={len(dead)}, errors={len(errors)}, no-colorway={len(no_colorway)}", flush=True)
load_staging(deduped)
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}...", flush=True)
run_sql(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,
code_source text, family_prefix text, last_scraped timestamptz
);""")
B = 200
for i in range(0, len(rows), B):
vals = []
for r in rows[i:i + B]:
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"]),
sql_lit(r["code_source"]), sql_lit(r["family_prefix"] or None), "now()"]) + ")")
run_sql(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,code_source,family_prefix,"
f"last_scraped) VALUES " + ",".join(vals) + ";")
print("Staging v2 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()