← back to Carnegie Reprice
carnegie_phase1_scrape.py
189 lines
#!/usr/bin/env python3
"""
TK-10671 Carnegie re-onboard PHASE 1 — complete carnegie_catalog (staging only).
$0 plain-fetch of carnegiefabrics.com (Magento). NO Shopify/Kamatera writes.
NO pricing writes. Fills spec columns + per-color swatch image; color names
legitimately don't exist on Carnegie (numeric colorways) so color_name is left
as-is (placeholder) and NOT fabricated.
"""
import re, sys, json, time, html as H, urllib.request, urllib.error
import subprocess
DSN = "host=/tmp dbname=dw_unified"
OUT_SQL = "/tmp/carnegie_p1_updates.sql"
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124 Safari/537.36"
# Carnegie product-description label -> our column
SPEC_MAP = {
"Width": "width",
"Contents": "content", "Content": "content",
"Durability": "durability_wyzenbeek",
"Repeat": "_repeat", # split into repeat_h / repeat_v
"Finish/es (as stocked)": "finish", "Available Finish(es)": "_finish_alt",
"Backing": "backing", "Backing (as stocked)": "_backing_alt",
"Cleaning Code": "cleaning_code",
"Flammability": "flammability",
"Manufactured In": "origin",
}
def fetch(url, tries=3):
for i in range(tries):
try:
req = urllib.request.Request(url, headers={"User-Agent": UA})
with urllib.request.urlopen(req, timeout=40) as r:
return r.read().decode("utf-8", "replace")
except Exception as e:
if i == tries - 1:
return None
time.sleep(2 + i * 2)
return None
def clean(s):
s = re.sub(r"<[^>]+>", " ", H.unescape(s))
return re.sub(r"\s+", " ", s).strip()
def parse_specs(h):
"""Return dict label->value (first occurrence wins) from product-description-row."""
out = {}
for k, v in re.findall(
r'<span class="product-description-title">(.*?)</span>\s*'
r'<span class="product-description-text">(.*?)</span>', h, re.S):
k = clean(k).rstrip(":")
v = clean(v)
if k and k not in out:
out[k] = v
return out
def split_repeat(v):
"""Carnegie repeat, e.g. '15.25\" (39 cm) Length x 7.75\" (20 cm) Width'.
Length = vertical repeat, Width = horizontal repeat. Returns (repeat_h, repeat_v)."""
if not v:
return None, None
vv = re.search(r'(.+?)\s*Length', v, re.I) # vertical (down the roll)
hh = re.search(r'x\s*(.+?)\s*Width', v, re.I) # horizontal (across)
if vv or hh:
return (hh.group(1).strip() if hh else None,
vv.group(1).strip() if vv else None)
# H:/V: style fallback
h2 = re.search(r'H[:\s]*([^V]+?)(?:\s*V[:\s]|$)', v, re.I)
v2 = re.search(r'V[:\s]*(.+)$', v, re.I)
if h2 or v2:
return (h2.group(1).strip() if h2 else None,
v2.group(1).strip() if v2 else None)
return v, v # single value applies both ways
def parse_colors(h):
"""color_number(str) -> swatch image url, from Magento jsonConfig + jsonSwatchConfig."""
# option-id -> color label(number)
mc = re.search(r'"878":\{"id":"878","code":"color_number","label":"[^"]*","options":(\[.*?\]),"position"', h)
if not mc:
return {}
try:
opts = json.loads(mc.group(1))
except Exception:
return {}
label_by_oid = {o["id"]: o["label"] for o in opts}
# locate 878 block inside jsonSwatchConfig
mi = h.find("jsonSwatchConfig")
if mi < 0:
return {}
seg = h[mi:mi + 80000]
b = seg.find('"878":{')
if b < 0:
return {}
sub = seg[b + 7:]
depth = 0; end = len(sub)
for i, ch in enumerate(sub):
if ch == "{":
depth += 1
elif ch == "}":
if depth == 0:
end = i; break
depth -= 1
block = sub[:end]
out = {}
for oid, img in re.findall(r'"(\d+)":\{[^{}]*?"value":"([^"]+)"[^{}]*?\}', block):
num = label_by_oid.get(oid)
if num:
out[str(num).strip()] = img.replace("\\/", "/")
return out
def q(v):
"""SQL literal or NULL."""
if v is None or v == "":
return "NULL"
return "'" + str(v).replace("'", "''") + "'"
def get_urls():
r = subprocess.run(
["psql", DSN, "-tAc",
"SELECT DISTINCT product_url FROM carnegie_catalog WHERE product_url IS NOT NULL ORDER BY product_url"],
capture_output=True, text=True, check=True)
return [u for u in r.stdout.splitlines() if u.strip()]
def main():
urls = get_urls()
total = len(urls)
print(f"[carnegie phase1] {total} distinct product URLs", flush=True)
ok_pages = 0; fail_pages = 0; spec_pages = 0; img_stmts = 0
with open(OUT_SQL, "w") as sql:
sql.write("BEGIN;\n")
for idx, url in enumerate(urls, 1):
h = fetch(url)
if not h:
fail_pages += 1
print(f" [{idx}/{total}] FETCH-FAIL {url}", flush=True)
continue
specs = parse_specs(h)
colors = parse_colors(h)
rh, rv = split_repeat(specs.get("Repeat"))
vals = {
"width": specs.get("Width"),
"content": specs.get("Contents") or specs.get("Content"),
"durability_wyzenbeek": specs.get("Durability"),
"repeat_h": rh,
"repeat_v": rv,
"finish": specs.get("Finish/es (as stocked)") or specs.get("Available Finish(es)"),
"backing": specs.get("Backing") or specs.get("Backing (as stocked)"),
"cleaning_code": specs.get("Cleaning Code"),
"flammability": specs.get("Flammability"),
"origin": specs.get("Manufactured In"),
}
specs_json = json.dumps(specs) if specs else None
if any(vals.values()) or specs_json:
spec_pages += 1
sql.write(
"UPDATE carnegie_catalog SET "
f"width={q(vals['width'])}, content={q(vals['content'])}, "
f"durability_wyzenbeek={q(vals['durability_wyzenbeek'])}, "
f"repeat_h={q(vals['repeat_h'])}, repeat_v={q(vals['repeat_v'])}, "
f"finish={q(vals['finish'])}, backing={q(vals['backing'])}, "
f"cleaning_code={q(vals['cleaning_code'])}, flammability={q(vals['flammability'])}, "
f"origin={q(vals['origin'])}, "
f"specs={q(specs_json)}::jsonb, updated_at=now() "
f"WHERE product_url={q(url)};\n")
for num, img in colors.items():
sql.write(
f"UPDATE carnegie_catalog SET swatch_image_url={q(img)}, updated_at=now() "
f"WHERE product_url={q(url)} AND color_number={q(str(num))};\n")
img_stmts += 1
ok_pages += 1
if idx % 25 == 0:
print(f" [{idx}/{total}] pages_ok={ok_pages} spec_pages={spec_pages} img_stmts={img_stmts}", flush=True)
time.sleep(0.4)
sql.write("COMMIT;\n")
print(f"[SCRAPE DONE] pages ok={ok_pages} fail={fail_pages} spec_pages={spec_pages} img_stmts={img_stmts}", flush=True)
print(f"[APPLY] running {OUT_SQL} via psql ...", flush=True)
r = subprocess.run(["psql", DSN, "-v", "ON_ERROR_STOP=1", "-f", OUT_SQL],
capture_output=True, text=True)
print(r.stdout[-1500:]);
if r.returncode != 0:
print("PSQL ERROR:\n", r.stderr[-2000:], flush=True)
sys.exit(1)
print("[APPLIED OK]", flush=True)
if __name__ == "__main__":
main()