← back to Newmor Onboard
scripts/spec-pdf-refresh.py
470 lines
#!/usr/bin/env python3
"""
Newmor commercial-spec refresh (TK-11424) — the "Norman scraper" data pipeline.
Recaptures fire_rating / finish / application / match_type / repeat / coverage,
which are EMPTY in newmor_catalog, by:
1. Crawling every /product/ page on the live sitemap for its technical-container
block(s) (HTML text: Roll Sizing / Pattern Repeat) AND any linked PDF spec sheet
(Specification / Technical Download).
2. Downloading + de-garbling each unique PDF. Newmor's spec-sheet PDFs are exported
from Photoshop: pdftotext extracts real text but (a) silently drops the "ti"
ligature glyph everywhere it occurs, and (b) inserts arbitrary whitespace mid-word
with no reliable word-boundary information. Fix: strip ALL whitespace to get a
legible (unspaced) character stream, then match target vocabulary with "ti"
treated as optional wherever it appears in the search keyword.
3. Extracting a small SHORT, NORMALIZED value per field (not raw mangled prose) —
recognized fire-code citations (Euroclass/ASTM E84/BS 476/EN 13501/NFPA 701/IMO),
CCC Type I/II/III (read from the PDF FILENAME first — reliable, vendor-authored,
never garbled — falling back to body text), named match types, and a repeat value
+ unit.
4. Writing findings to a staging table, snapshotting current newmor_catalog spec
columns (reversibility restore-map), then fill-only-never-clobber UPDATEing
newmor_catalog (never overwrites an existing non-empty value).
HARD RAILS: $0 local (no paid API, no vision/OCR). Writes ONLY to the Mac2-canonical
newmor_catalog staging table (dw_unified). NEVER touches Shopify / live. The live
metafield backfill is a separate, Steve-gated step (see artifacts/METAFIELD-MAP.md).
Usage:
python3 spec-pdf-refresh.py --scrape # crawl + extract, write JSON artifacts only
python3 spec-pdf-refresh.py --stage # load extraction JSON into a staging table
python3 spec-pdf-refresh.py --apply # fill-only UPDATE newmor_catalog (snapshots first)
python3 spec-pdf-refresh.py --all # scrape + stage + apply
"""
import argparse
import hashlib
import html
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.request
HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.join(HERE, "..")
OUT = os.path.join(ROOT, "data", "spec-refresh-20260912")
PDF_CACHE = os.path.join(ROOT, "data", "spec-pdfs")
os.makedirs(OUT, exist_ok=True)
os.makedirs(PDF_CACHE, exist_ok=True)
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_TABLE = "newmor_catalog_specrefresh_20260912"
RESTORE_MAP = os.path.join(ROOT, "artifacts", "TK-11424-restore-map.json")
# ---------------------------------------------------------------- fetch helpers
def fetch(url, timeout=30):
req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "*/*"})
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
return r.getcode(), r.geturl(), r.read()
except urllib.error.HTTPError as e:
return e.code, url, b""
except Exception as e:
return -1, url, f"__ERR__ {e}".encode()
def clean(s):
return re.sub(r"\s+", " ", html.unescape(re.sub(r"<[^>]+>", " ", s or ""))).strip()
# ---------------------------------------------------------------- HTML parse
TECH_RE = re.compile(
r'technical-container__title">\s*(.*?)\s*</div>\s*'
r'<(a|p)([^>]*)class="technical-container__link[^"]*"([^>]*)>\s*(.*?)\s*</\2>', re.DOTALL)
def parse_page(url, h):
pat = ""
m = re.search(r"<h1[^>]*>(.*?)</h1>", h, re.DOTALL)
if m:
pat = clean(m.group(1))
if not pat:
m = re.search(r"<title>(.*?)</title>", h, re.DOTALL)
if m:
pat = clean(m.group(1)).split(" - Newmor")[0].split("|")[0].strip()
tech_items = []
for m in TECH_RE.finditer(h):
title, tag, attrs1, attrs2, content = m.groups()
href = None
hm = re.search(r'href="([^"]*)"', (attrs1 or "") + (attrs2 or ""))
if hm:
href = hm.group(1)
tech_items.append({"title": clean(title), "tag": tag, "href": href, "text": clean(content)})
pdf_urls = sorted({t["href"] for t in tech_items if t["href"] and t["href"].lower().endswith(".pdf")})
# HTML-only signals (Roll Sizing / Pattern Repeat plain-text items)
html_width = html_length = html_repeat = html_match = ""
for t in tech_items:
title_l = t["title"].lower()
if "roll siz" in title_l or "width" in title_l:
mm = re.findall(r"(\d{3,4})\s*mm", t["text"], re.IGNORECASE)
cm = re.findall(r"(\d{2,4})\s*cm", t["text"], re.IGNORECASE)
if cm:
html_width = f"{cm[0]}cm"
elif mm:
html_width = f"{mm[0]}mm"
lm = re.search(r"(\d{1,3}(?:\.\d+)?)\s*m\b", t["text"])
if lm:
html_length = f"{lm.group(1)}m"
if "repeat" in title_l or (t["tag"] == "p" and "repeat" in (t.get("text") or "").lower()):
html_repeat = html_repeat or t["text"]
mt = re.search(r"(straight match|half drop|drop match|free match|random match|offset match|no match|reverse hang)",
t["text"], re.IGNORECASE)
if mt:
html_match = mt.group(1).title()
return {
"url": url, "pattern": pat, "tech_items": tech_items, "pdf_urls": pdf_urls,
"html_width": html_width, "html_length": html_length,
"html_repeat": html_repeat, "html_match": html_match,
}
# ---------------------------------------------------------------- PDF de-garble + extraction
def fuzzy_ti(word):
"""Match `word` allowing the 'ti' ligature to be silently dropped anywhere it occurs
(confirmed Photoshop-PDF export defect: 'Specification'->'Specificaon', 'Rating'->'Rang').
`word` may contain spaces for readability; the blob it searches has NONE (all
whitespace is stripped in pdf_to_blob), so spaces/hyphens in `word` are dropped first."""
word = re.sub(r"[\s-]+", "", word)
parts = word.split("ti")
return "(?:ti)?".join(re.escape(p) for p in parts)
def pdf_to_blob(pdf_path):
p = subprocess.run(["pdftotext", "-raw", pdf_path, "-"], capture_output=True, text=True)
raw = p.stdout or ""
return re.sub(r"\s+", "", raw)
FIRE_CODES = [
(re.compile(r"euroclass\s*([a-f])", re.IGNORECASE), lambda m: f"Euroclass {m.group(1).upper()}"),
(re.compile(r"astm\s*e-?84\D{0,10}class\s*([a-c])", re.IGNORECASE), lambda m: f"ASTM E84 Class {m.group(1).upper()}"),
(re.compile(r"class\s*([a-c])\D{0,20}astm\s*e-?84", re.IGNORECASE), lambda m: f"ASTM E84 Class {m.group(1).upper()}"),
(re.compile(r"astm\s*e-?84", re.IGNORECASE), lambda m: "ASTM E84"),
(re.compile(r"bs\s*476[\s\-]*part\s*(\d+)", re.IGNORECASE), lambda m: f"BS 476 Part {m.group(1)}"),
(re.compile(r"bs\s*en\s*476", re.IGNORECASE), lambda m: "BS EN 476"),
(re.compile(r"bs\s*476", re.IGNORECASE), lambda m: "BS 476"),
(re.compile(r"en\s*13501[\s\-]*1", re.IGNORECASE), lambda m: "EN 13501-1"),
(re.compile(r"nfpa\s*701", re.IGNORECASE), lambda m: "NFPA 701"),
(re.compile(r"\bimo\b", re.IGNORECASE), lambda m: "IMO"),
]
MATCH_TERMS = ["Straight Match", "Half Drop", "Drop Match", "Free Match", "Random Match",
"Offset Match", "No Match", "Reverse Hang"]
def extract_window(blob, label_words, window=260):
for label in label_words:
pat = re.compile(fuzzy_ti(label), re.IGNORECASE)
m = pat.search(blob)
if m:
return blob[m.end(): m.end() + window]
return ""
def extract_fire_rating(blob):
window = extract_window(blob, ["Fire Rating", "Fire Classification", "Fire Performance",
"Flame Spread", "Flammability"], window=300)
hay = window or blob # fall back to whole blob if no explicit label found
found = []
for rx, fmt in FIRE_CODES:
for m in rx.finditer(hay):
v = fmt(m)
if v not in found:
found.append(v)
# de-dupe "ASTM E84" if a more specific "ASTM E84 Class X" already captured
if any(v.startswith("ASTM E84 Class") for v in found) and "ASTM E84" in found:
found.remove("ASTM E84")
return "; ".join(found)
def extract_finish(blob):
# CONSERVATIVE BY DESIGN: verified against a 19-PDF sample (3 targeted + 16 random) that
# Newmor's spec sheets do NOT publish a distinct sheen/surface-finish field anywhere —
# only an explicit "Finish:" label (if one ever appears) is trusted; a bare keyword
# search over the whitespace-stripped blob throws false positives (e.g. "Satin" matched
# inside "Contains an active" once spaces were stripped). Case-SENSITIVE on the label.
window = extract_window(blob, ["Finish"], window=40)
if not window:
return ""
for term in ["Satin", "Matte", "Gloss", "Semi-Gloss", "Textured", "Smooth", "Embossed"]:
if re.match(fuzzy_ti(term), window):
return term
return ""
def extract_application_from_filename(pdf_url):
m = re.search(r"type[-\s]?(i{1,3})\b", pdf_url, re.IGNORECASE)
if m:
n = len(m.group(1))
return f"Type {'I'*n}"
return ""
def extract_application_from_blob(blob):
m = re.search(r"type(iii|ii|i)\b", blob, re.IGNORECASE)
if m:
roman = m.group(1).upper()
return f"Type {roman}"
return ""
def extract_match(blob):
found = []
for term in MATCH_TERMS:
if re.search(fuzzy_ti(term), blob): # case-sensitive: these are always Title Case labels
found.append(term)
if "Half Drop" in found and "Drop Match" in found:
found.remove("Drop Match") # "Half Drop" already implies a drop match
return "; ".join(found)
def extract_repeat(blob):
m = re.search(r"\(?(\d+(?:\.\d+)?)\s*(mm|cm|in)\s*repeat", blob, re.IGNORECASE)
if not m:
m = re.search(r"repeat\D{0,10}(\d+(?:\.\d+)?)\s*(mm|cm|in)", blob, re.IGNORECASE)
if m:
return f"{m.group(1)}{m.group(2).lower()}"
return ""
def extract_roll_size(blob):
m = re.search(r"(\d{2,4})\s*cm\s*x\s*(\d{1,3}(?:\.\d+)?)\s*m\b", blob, re.IGNORECASE)
if m:
return m.group(1), m.group(2)
return None, None
def compute_coverage(width_cm, length_m):
try:
w = float(width_cm) / 100.0
l = float(length_m)
sqm = w * l
sqyd = sqm * 1.19599
return f"{sqm:.1f} sqm ({sqyd:.1f} sq yd) per roll"
except Exception:
return ""
def pdf_cache_path(pdf_url):
h = hashlib.sha1(pdf_url.encode()).hexdigest()[:16]
base = re.sub(r"[^A-Za-z0-9_.-]", "-", os.path.basename(pdf_url))[:60]
return os.path.join(PDF_CACHE, f"{h}-{base}")
def get_pdf_blob(pdf_url, cache):
if pdf_url in cache:
return cache[pdf_url]
path = pdf_cache_path(pdf_url)
if not os.path.exists(path):
code, final, data = fetch(pdf_url, timeout=45)
if code != 200 or not data or data.startswith(b"__ERR__"):
cache[pdf_url] = {"blob": "", "status": code}
return cache[pdf_url]
with open(path, "wb") as f:
f.write(data)
time.sleep(0.6)
blob = pdf_to_blob(path)
cache[pdf_url] = {"blob": blob, "status": 200}
return cache[pdf_url]
# ---------------------------------------------------------------- crawl
def crawl():
print("Fetching sitemap...", flush=True)
code, _, xml_b = fetch(SITEMAP)
xml = xml_b.decode("utf-8", "replace")
urls = sorted({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)
pages, dead, errors = [], [], []
for i, u in enumerate(urls, 1):
code, final, data = fetch(u)
if code in (404, 410):
dead.append({"url": u, "status": code})
elif code == 200 and data and not data.startswith(b"__ERR__"):
h = data.decode("utf-8", "replace")
try:
pages.append(parse_page(final, h))
except Exception as e:
errors.append({"url": u, "err": str(e)})
else:
errors.append({"url": u, "status": code})
if i % 25 == 0:
print(f" {i}/{len(urls)} pages={len(pages)} dead={len(dead)} err={len(errors)}", flush=True)
time.sleep(0.8)
print(f"Pages parsed: {len(pages)}. Dead: {len(dead)}. Errors: {len(errors)}.", flush=True)
pdf_urls = sorted({p for pg in pages for p in pg["pdf_urls"]})
print(f"Unique PDF spec sheets: {len(pdf_urls)}", flush=True)
pdf_cache = {}
pdf_findings = {}
for i, pdf_url in enumerate(pdf_urls, 1):
entry = get_pdf_blob(pdf_url, pdf_cache)
blob = entry["blob"]
if not blob:
pdf_findings[pdf_url] = {"status": entry["status"], "empty": True}
continue
w, l = extract_roll_size(blob)
pdf_findings[pdf_url] = {
"status": 200,
"fire_rating": extract_fire_rating(blob),
"finish": extract_finish(blob),
"application": extract_application_from_filename(pdf_url) or extract_application_from_blob(blob),
"match_type": extract_match(blob),
"repeat": extract_repeat(blob),
"roll_width_cm": w, "roll_length_m": l,
"coverage": compute_coverage(w, l) if (w and l) else "",
"blob_len": len(blob),
}
if i % 20 == 0:
print(f" pdf {i}/{len(pdf_urls)}", flush=True)
results = []
for pg in pages:
merged = {"fire_rating": "", "finish": "", "application": "", "match_type": "",
"repeat": "", "coverage": "", "source_pdfs": pg["pdf_urls"]}
for pdf_url in pg["pdf_urls"]:
f = pdf_findings.get(pdf_url, {})
for k in ("fire_rating", "finish", "application", "match_type", "repeat", "coverage"):
v = f.get(k) or ""
if v and not merged[k]:
merged[k] = v
# HTML fallback for match_type/repeat if PDF gave nothing
if not merged["match_type"] and pg["html_match"]:
merged["match_type"] = pg["html_match"]
if not merged["repeat"] and pg["html_repeat"]:
rm = re.search(r"(\d+(?:\.\d+)?)\s*(mm|cm|in)", pg["html_repeat"], re.IGNORECASE)
if rm:
merged["repeat"] = f"{rm.group(1)}{rm.group(2).lower()}"
results.append({
"url": pg["url"], "pattern": pg["pattern"],
"html_width": pg["html_width"], "html_length": pg["html_length"],
**merged,
})
json.dump(results, open(os.path.join(OUT, "extraction.json"), "w"), indent=2)
json.dump(pdf_findings, open(os.path.join(OUT, "pdf-findings.json"), "w"), indent=2)
json.dump({"dead": dead, "errors": errors, "urls_total": len(urls),
"pages_parsed": len(pages), "unique_pdfs": len(pdf_urls)},
open(os.path.join(OUT, "scrape-log.json"), "w"), indent=2)
cov = {k: sum(1 for r in results if r[k]) for k in
("fire_rating", "finish", "application", "match_type", "repeat", "coverage")}
print(f"\nCoverage out of {len(results)} product pages: {cov}", flush=True)
return results
# ---------------------------------------------------------------- staging + apply
def run_sql(sql, capture=False):
args = ["psql", "-h", "/tmp", "-d", "dw_unified", "-v", "ON_ERROR_STOP=1"]
if capture:
args.append("-tA")
args += ["-c", sql]
p = subprocess.run(args, capture_output=True, text=True)
if p.returncode != 0:
print("SQL ERROR:", p.stderr[:800], file=sys.stderr); sys.exit(1)
return p.stdout
def sql_lit(v):
if v is None or v == "":
return "NULL"
return "'" + str(v).replace("'", "''") + "'"
def stage(results):
print(f"Loading {len(results)} rows into {STAGING_TABLE}...", flush=True)
run_sql(f"""
DROP TABLE IF EXISTS {STAGING_TABLE};
CREATE TABLE {STAGING_TABLE} (
id serial PRIMARY KEY, product_url text, pattern text,
fire_rating text, finish text, application text, match_type text,
repeat_v text, coverage text, source_pdfs text, scraped_at timestamptz DEFAULT now()
);""")
B = 100
for i in range(0, len(results), B):
vals = []
for r in results[i:i + B]:
vals.append("(" + ",".join([
sql_lit(r["url"]), sql_lit(r["pattern"]), sql_lit(r["fire_rating"]),
sql_lit(r["finish"]), sql_lit(r["application"]), sql_lit(r["match_type"]),
sql_lit(r["repeat"]), sql_lit(r["coverage"]),
sql_lit(json.dumps(r.get("source_pdfs") or []))]) + ")")
run_sql(f"INSERT INTO {STAGING_TABLE} (product_url,pattern,fire_rating,finish,application,"
f"match_type,repeat_v,coverage,source_pdfs) VALUES " + ",".join(vals) + ";")
print("Staging load complete:", STAGING_TABLE, flush=True)
FIELD_MAP = {"fire_rating": "fire_rating", "finish": "finish", "application": "application",
"match_type": "match_type", "repeat": "repeat_v", "coverage": "coverage"}
def snapshot_and_apply(results):
# snapshot current values for every row whose product_url is in scope (reversibility)
urls = sorted({r["url"] for r in results if any(r[k] for k in FIELD_MAP)})
if not urls:
print("Nothing to apply — no new values extracted.", flush=True)
return
print(f"Snapshotting current newmor_catalog spec columns for {len(urls)} product_urls...", flush=True)
url_array = "ARRAY[" + ",".join(sql_lit(u) for u in urls) + "]"
snap_raw = run_sql(
f"select id, product_url, fire_rating, finish, application, match_type, repeat_v, coverage "
f"from newmor_catalog where product_url = ANY({url_array});", capture=True)
snapshot = []
for line in snap_raw.splitlines():
parts = [p.strip() for p in line.split("|")]
if len(parts) < 7 or not parts[0]:
continue
snapshot.append({"id": parts[0], "product_url": parts[1], "fire_rating": parts[2],
"finish": parts[3], "application": parts[4], "match_type": parts[5],
"repeat_v": parts[6], "coverage": parts[7] if len(parts) > 7 else ""})
os.makedirs(os.path.dirname(RESTORE_MAP), exist_ok=True)
json.dump({"taken_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"table": "newmor_catalog", "row_count": len(snapshot), "rows": snapshot},
open(RESTORE_MAP, "w"), indent=2)
print(f"Restore-map written: {RESTORE_MAP} ({len(snapshot)} rows)", flush=True)
before = {}
for col in set(FIELD_MAP.values()):
n = run_sql(f"select count(*) from newmor_catalog where product_url = ANY({url_array}) "
f"and {col} is not null and {col} <> '';", capture=True).strip()
before[col] = n
# one bulk fill-only UPDATE per column, driven by a VALUES list of (url, new_value)
for src_field, col in FIELD_MAP.items():
rows = [(r["url"], r[src_field]) for r in results if r[src_field]]
if not rows:
continue
values_sql = ",".join(f"({sql_lit(u)},{sql_lit(v)})" for u, v in rows)
sql = (f"UPDATE newmor_catalog AS c SET {col} = v.val, updated_at = now() "
f"FROM (VALUES {values_sql}) AS v(url, val) "
f"WHERE c.product_url = v.url AND (c.{col} IS NULL OR c.{col} = '');")
run_sql(sql)
after = {}
for col in set(FIELD_MAP.values()):
n = run_sql(f"select count(*) from newmor_catalog where product_url = ANY({url_array}) "
f"and {col} is not null and {col} <> '';", capture=True).strip()
after[col] = n
print("Fill-only apply complete. Non-empty counts scoped to the", len(urls), "touched product_urls:")
for col in sorted(before):
print(f" {col}: {before[col].strip()} -> {after[col].strip()}")
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--scrape", action="store_true")
ap.add_argument("--stage", action="store_true")
ap.add_argument("--apply", action="store_true")
ap.add_argument("--all", action="store_true")
args = ap.parse_args()
results = None
if args.scrape or args.all:
results = crawl()
if args.stage or args.all:
if results is None:
results = json.load(open(os.path.join(OUT, "extraction.json")))
stage(results)
if args.apply or args.all:
if results is None:
results = json.load(open(os.path.join(OUT, "extraction.json")))
snapshot_and_apply(results)
if not (args.scrape or args.stage or args.apply or args.all):
ap.print_help()
if __name__ == "__main__":
main()