build_speed_csv.py

skills/production-speed-csv/scripts/build_speed_csv.py

1. #!/usr/bin/env python3
2. """
3. build_speed_csv.py: Generate a production-speed CSV from a Meridian Display Co. category URL.
5. Usage:
6. python build_speed_csv.py <category_url | category_html_file> [<output_csv_path>]
8. Example:
9. python build_speed_csv.py https://meridiandisplay.example/10x10-booths/
11. Offline mode: pass a saved category page instead of a URL. Each product page is
12. then read from <slug>.html in the same folder as the category file, with no
13. network access and no request spacing. The bundled test fixture works this way:
14. python build_speed_csv.py fixtures/category-10x10-booths.html out.csv
16. Output columns:
17. #, Product Name, Standard Turnaround Days, Express Turnaround Days,
18. Super Express Days, Production Days (slash-separated),
19. Production + Overnight Shipping (slash-separated)
20. """
22. import csv
23. import html
24. import os
25. import re
26. import sys
27. import time
28. import urllib.error
29. import urllib.parse
30. import urllib.request
32. # ---- Tunables ------------------------------------------------------------
34. # Separator used inside the two summary columns. Slash because Claude.ai's CSV
35. # previewer mishandles RFC 4180 quoted commas. Edit to "," if a downstream
36. # parser actually needs comma-separated values (file will still open correctly
37. # in Excel, Sheets, Numbers).
38. SEPARATOR = " / "
40. # Per-request spacing in seconds. 1.5s is the validated safe rate against
41. # meridiandisplay.example's WAF for sustained sequential requests.
42. REQUEST_SPACING_S = 1.5
44. USER_AGENT = (
45. "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
46. "AppleWebKit/537.36 (KHTML, like Gecko) "
47. "Chrome/120.0.0.0 Safari/537.36"
48. )
50. # ---- URL filtering -------------------------------------------------------
52. # Slugs that show up in nav menus, footers, etc. These are never products.
53. EXCLUDE_SLUGS = {
54. "shop-all",
55. "displays-menu",
56. "outdoor-menu",
57. "flags-menu",
58. "stands-menu",
59. "all-products",
60. "service-request",
61. "contact",
62. "partner-program",
63. "partner-benefits",
64. "resources",
65. "help",
66. "turnaround-guide",
67. "shipping",
68. "about",
69. "blog",
70. "faq",
71. "privacy",
72. "terms",
73. "returns",
74. }
76. # ---- Label variant patterns ---------------------------------------------
78. # Order matters: Super Express must be checked before Express (because "Super
79. # Express Turnaround" contains "Express"), and we want longest-match-first
80. # behavior.
81. TIER_PATTERNS = [
82. ("super_express", re.compile(r"\bsuper\s*express\b", re.I)),
83. ("express", re.compile(r"\bexpress\b", re.I)),
84. ("standard", re.compile(r"\bstandard\b", re.I)),
85. ]
87. # Number + "business day(s)" extraction. Tolerates inconsistent whitespace
88. # ("Turnaround10 business days") and singular/plural ("1 business day").
89. DAYS_RE = re.compile(r"(\d+)\s*business\s*day", re.I)
91. # Caveat capture: anything in parens after the day count except the price.
92. # E.g. "1 business day + transit, (only available for 8ft & 10ft widths)(+$845.00)"
93. # We want the size caveat, not the price.
94. CAVEAT_RE = re.compile(r"\(([^)]*(?:only|available|except|excludes|sizes?)[^)]*)\)", re.I)
96. # Storefront markup the parser keys on.
97. SPEED_SECTION_LABEL = "Turnaround Option" # matches "Turnaround Option" and "Turnaround Options"
98. OPTION_GROUP_OPEN = '<div class="md-option-group'
99. OPTION_LABEL_ATTR = "data-option-label"
102. # ---- Fetching ------------------------------------------------------------
104. def fetch(url: str, timeout: int = 30, retries: int = 2) -> str:
105. """GET a URL with a real browser UA. Returns decoded HTML text.
107. Retries on transient failures (timeouts, intermittent SSL hiccups, 5xx).
108. Does NOT retry on 4xx: those are real and retrying won't help.
109. """
110. last_err = None
111. for attempt in range(retries + 1):
112. try:
113. req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
114. with urllib.request.urlopen(req, timeout=timeout) as r:
115. return r.read().decode("utf-8", errors="ignore")
116. except urllib.error.HTTPError as e:
117. # 4xx: don't retry. 5xx: retry.
118. if 400 <= e.code < 500:
119. raise
120. last_err = e
121. except Exception as e:
122. last_err = e
123. if attempt < retries:
124. time.sleep(2.0 * (attempt + 1)) # 2s, then 4s
125. raise last_err # type: ignore[misc]
128. def read_local(path: str) -> str:
129. """Read a saved page from disk (offline mode)."""
130. with open(path, encoding="utf-8", errors="ignore") as f:
131. return f.read()
134. def local_product_path(local_dir: str, url: str) -> str:
135. """Map a product URL to its saved page: <local_dir>/<slug>.html."""
136. return os.path.join(local_dir, category_slug(url) + ".html")
139. # ---- Category-page parsing ----------------------------------------------
141. def extract_product_urls(category_html: str, category_url: str) -> list[str]:
142. """Pull product URLs from a category page in display order, deduped."""
143. pattern = re.compile(r'href="(https?://meridiandisplay\.example/([a-z0-9\-]+)/)"')
144. seen = set()
145. ordered = []
146. for m in pattern.finditer(category_html):
147. url, slug = m.group(1), m.group(2)
148. if url == category_url or url == category_url.rstrip("/") + "/":
149. continue
150. if slug in EXCLUDE_SLUGS:
151. continue
152. if slug.endswith("-menu") or slug.endswith("menu"):
153. continue
154. if url in seen:
155. continue
156. seen.add(url)
157. ordered.append(url)
158. return ordered
161. def canonical_url(page_html: str) -> str | None:
162. """The page's <link rel="canonical"> URL, if it has one (offline mode)."""
163. m = re.search(r'<link[^>]*rel="canonical"[^>]*href="([^"]+)"', page_html, re.I)
164. return m.group(1) if m else None
167. # ---- Product-page parsing -----------------------------------------------
169. def get_product_name(product_html: str) -> str:
170. """Pull the product H1, fall back to <title>."""
171. m = re.search(
172. r'<h1[^>]*class="[^"]*md-product__title[^"]*"[^>]*>(.*?)</h1>',
173. product_html, re.I | re.S,
174. )
175. if not m:
176. m = re.search(r"<h1[^>]*>(.*?)</h1>", product_html, re.I | re.S)
177. if m:
178. name = re.sub(r"<[^>]+>", "", m.group(1))
179. name = html.unescape(name)
180. name = re.sub(r"\s+", " ", name).strip()
181. if name:
182. return name
183. m = re.search(r"<title>(.*?)</title>", product_html, re.I | re.S)
184. if m:
185. t = html.unescape(m.group(1)).strip()
186. t = re.sub(r"\s*\|\s*Meridian Display.*$", "", t, flags=re.I)
187. return t
188. return "(unknown product)"
191. def get_speed_block(product_html: str) -> str | None:
192. """Isolate the option group containing the Turnaround Option(s) radios.
194. Handles both 'Turnaround Option' (singular, used on some pages)
195. and 'Turnaround Options' (plural).
196. """
197. m = re.search(SPEED_SECTION_LABEL, product_html)
198. if not m:
199. return None
200. idx = m.start()
201. # Walk backward to the enclosing option-group div.
202. start = product_html.rfind(OPTION_GROUP_OPEN, 0, idx)
203. if start == -1:
204. start = max(0, idx - 800)
205. # Forward to the next option-group div, which bounds the speed-option block.
206. rest = product_html[idx:]
207. next_fg = re.search(re.escape(OPTION_GROUP_OPEN), rest)
208. end = idx + next_fg.start() if next_fg else idx + 8000
209. return product_html[start:end]
212. def parse_options(block: str | None) -> list[dict]:
213. """Extract each speed option from the block.
215. Returns list of dicts: {tier, days, caveat, raw}.
216. tier is one of: 'standard', 'express', 'super_express', or 'unknown'.
217. """
218. if not block:
219. return []
220. options = []
221. # Pull every data-option-label attr; filter to those that look like a
222. # speed option (mention 'business day').
223. for raw_match in re.finditer(OPTION_LABEL_ATTR + r'="([^"]*?)"', block):
224. raw = raw_match.group(1)
225. raw_clean = re.sub(r"\s+", " ", html.unescape(raw)).strip()
226. if "business day" not in raw_clean.lower():
227. continue
228. days_m = DAYS_RE.search(raw_clean)
229. if not days_m:
230. continue
231. days = int(days_m.group(1))
233. tier = "unknown"
234. for tier_name, tier_re in TIER_PATTERNS:
235. if tier_re.search(raw_clean):
236. tier = tier_name
237. break
239. caveat_m = CAVEAT_RE.search(raw_clean)
240. caveat = caveat_m.group(1).strip() if caveat_m else None
242. options.append({
243. "tier": tier,
244. "days": days,
245. "caveat": caveat,
246. "raw": raw_clean,
247. })
248. return options
251. # ---- Output formatting --------------------------------------------------
253. def format_days_cell(days: int | None) -> str:
254. if days is None:
255. return ""
256. if days == 1:
257. return "1 business day"
258. return f"{days} business days"
261. def slash_join(values: list[int]) -> str:
262. return SEPARATOR.join(str(v) for v in values)
265. # ---- Main orchestration -------------------------------------------------
267. def category_slug(category_url: str) -> str:
268. path = urllib.parse.urlparse(category_url).path.strip("/")
269. return path.split("/")[-1] if path else "category"
272. def run(category_url: str, output_path: str | None = None) -> str:
273. # Offline mode: a saved category page on disk. URLs take the path below
274. # unchanged.
275. local_dir = None
276. if not category_url.startswith("http") and os.path.isfile(category_url):
277. local_file = category_url
278. local_dir = os.path.dirname(os.path.abspath(local_file))
279. print(f"Reading category page from file: {local_file}", file=sys.stderr)
280. cat_html = read_local(local_file)
281. category_url = canonical_url(cat_html) or ""
282. if output_path is None:
283. slug = category_slug(category_url) if category_url else \
284. os.path.splitext(os.path.basename(local_file))[0]
285. output_path = f"{slug}_production_speeds.csv"
286. else:
287. if not category_url.startswith("http"):
288. sys.exit(f"Bad URL: {category_url!r}")
289. if output_path is None:
290. output_path = f"{category_slug(category_url)}_production_speeds.csv"
292. print(f"Fetching category page: {category_url}", file=sys.stderr)
293. cat_html = fetch(category_url)
294. product_urls = extract_product_urls(cat_html, category_url)
295. print(f"Found {len(product_urls)} product URLs", file=sys.stderr)
297. rows = []
298. no_speed_section = []
299. unknown_tier_labels = []
300. caveats_dropped = []
302. for i, url in enumerate(product_urls, 1):
303. try:
304. if local_dir:
305. phtml = read_local(local_product_path(local_dir, url))
306. else:
307. time.sleep(REQUEST_SPACING_S)
308. phtml = fetch(url)
309. except Exception as e:
310. print(f" [{i:02d}] FETCH FAILED ({e}), skipping {url}", file=sys.stderr)
311. continue
313. name = get_product_name(phtml)
314. block = get_speed_block(phtml)
315. opts = parse_options(block)
317. if not opts:
318. no_speed_section.append((i, name, url))
319. print(f" [{i:02d}] {name} ⚠ no speed options found", file=sys.stderr)
320. continue
322. # Pick first occurrence per tier (radio options are unique by tier).
323. tiers = {}
324. for o in opts:
325. if o["tier"] == "unknown":
326. unknown_tier_labels.append((name, o["raw"]))
327. continue
328. tiers.setdefault(o["tier"], o)
329. if o["caveat"]:
330. caveats_dropped.append((name, o["tier"], o["caveat"]))
332. std = tiers.get("standard", {}).get("days")
333. exp = tiers.get("express", {}).get("days")
334. sup = tiers.get("super_express", {}).get("days")
336. tier_days = [d for d in (std, exp, sup) if d is not None]
337. rows.append({
338. "num": i,
339. "name": name,
340. "std": format_days_cell(std),
341. "exp": format_days_cell(exp),
342. "sup": format_days_cell(sup),
343. "values": slash_join(tier_days),
344. "values_p1": slash_join([d + 1 for d in tier_days]),
345. })
346. print(
347. f" [{i:02d}] {name} → "
348. f"std={std} exp={exp} sup={sup}",
349. file=sys.stderr,
350. )
352. with open(output_path, "w", newline="", encoding="utf-8") as f:
353. w = csv.writer(f, quoting=csv.QUOTE_MINIMAL)
354. w.writerow([
355. "#",
356. "Product Name",
357. "Standard Turnaround Days",
358. "Express Turnaround Days",
359. "Super Express Days",
360. "Production Days (slash-separated)",
361. "Production + Overnight Shipping (slash-separated)",
362. ])
363. for r in rows:
364. w.writerow([
365. r["num"], r["name"], r["std"], r["exp"], r["sup"],
366. r["values"], r["values_p1"],
367. ])
369. print("", file=sys.stderr)
370. print(f"Wrote {len(rows)} rows → {output_path}", file=sys.stderr)
371. if no_speed_section:
372. print(f"\n{len(no_speed_section)} products had no Turnaround Options section "
373. "(likely custom-quote-only):", file=sys.stderr)
374. for i, name, url in no_speed_section:
375. print(f" [{i:02d}] {name} ({url})", file=sys.stderr)
376. if caveats_dropped:
377. print(f"\n{len(caveats_dropped)} caveats stripped from cells "
378. "(tracked separately, not in CSV):", file=sys.stderr)
379. for name, tier, caveat in caveats_dropped:
380. print(f" {name} [{tier}] → {caveat}", file=sys.stderr)
381. if unknown_tier_labels:
382. print(f"\n{len(unknown_tier_labels)} option labels couldn't be matched "
383. "to a tier (add to TIER_PATTERNS):", file=sys.stderr)
384. for name, raw in unknown_tier_labels:
385. print(f" {name} → {raw!r}", file=sys.stderr)
387. return output_path
390. if __name__ == "__main__":
391. if len(sys.argv) < 2:
392. sys.exit(__doc__)
393. cat_url = sys.argv[1]
394. out = sys.argv[2] if len(sys.argv) > 2 else None
395. run(cat_url, out)