> Multi-Language Product Catalog Extraction Pipeline (Python)
// Created at: 24-09-2026
[Project Overview]
A data extraction pipeline for an e-commerce brand with three regional stores (Italian, English, Spanish). It discovers the catalog on its own, extracts nine content fields per product, and produces a single, formatted Excel workbook. If it's interrupted, it picks up where it stopped. Tech stack: Python, Playwright (async), BeautifulSoup, pandas, openpyxl.
[_Case Study]
Discover the catalog from the navigation, not from a hardcoded list
Each store's header menu is read and filtered: same domain only, no cart, login or blog pages, no query strings, and only category-depth paths. The same code works on all three locales.
async def discover_category_urls(page, domain: str) -> list[str]:
"""Collect catalog categories straight from the header navigation."""
categories = set()
for selector in MENU_SELECTORS:
hrefs = await page.eval_on_selector_all(selector, "els => els.map(e => e.href)")
for href in hrefs:
if not href or looks_skippable(href): # cart, login, blog, "?" ...
continue
clean = href.split("#")[0].split("?")[0].rstrip("/")
if same_domain(clean, domain) and is_category_depth(clean):
categories.add(clean)
return sorted(categories)
def is_category_depth(url: str) -> bool:
path = urlparse(url).path.strip("/")
return bool(path) and len(path.split("/")) <= 2Collect every product by paging until nothing new appears
The collector walks each category page by page and stops when a page adds no new products, which covers both an empty page and a store that keeps serving the last page.
async def collect_product_links(page, category_url: str, domain: str) -> list[str]:
"""Walk ?page=N until a page yields no new products."""
links, page_no = set(), 1
while True:
response = await page.goto(f"{category_url}?page={page_no}",
wait_until="domcontentloaded")
if not response or response.status != 200:
break
cards = await page.query_selector_all(CARD_SELECTOR)
if not cards:
break
before = len(links)
for card in cards:
a = await card.query_selector(LINK_SELECTOR)
if a and (href := await a.get_attribute("href")):
links.add(urljoin(domain, href).split("#")[0])
if len(links) == before: # same page served again / past the last page
break
page_no += 1
return sorted(links)Extract content in a language-agnostic way
Product pages hold their information in accordion sections whose titles are translated. Instead of matching text, the extractor anchors on each section's icon, so the same selectors work in Italian, English and Spanish.
SECTION_ICONS = {
"full_description": "i.ph-clipboard-text",
"how_to_use": "i.ph-presentation-chart",
"who_its_for": "i.ph-users",
"ingredients": "i.ph-leaf",
"green_packaging": "i.ph-recycle",
}
async def accordion_text(page, icon: str) -> str:
"""Find the section by its icon, not by its (translated) title."""
body = await page.query_selector(f"div.accordion-item:has({icon}) div.accordion-body")
return (await body.inner_text()).strip() if body else ""
async def read_sections(page) -> dict:
return {field: await accordion_text(page, icon) for field, icon in SECTION_ICONS.items()}Be resumable by design
Every product is appended to a checkpoint file and forced to disk right away. On restart, the pipeline skips what's already done, so a crash or a lost connection never means starting over.
def append_row(csv_path: str, row: dict, columns: list[str]):
new_file = not os.path.exists(csv_path)
with open(csv_path, "a", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=columns)
if new_file:
writer.writeheader()
writer.writerow(row)
f.flush()
os.fsync(f.fileno()) # rows already written survive a crash
def load_done_urls(csv_path: str) -> set[str]:
if not os.path.exists(csv_path):
return set()
return set(pd.read_csv(csv_path, usecols=["url"])["url"].dropna().astype(str))
# main loop: a restart resumes where it stopped
remaining = [u for u in product_urls if u not in load_done_urls(partial_csv)]Deliver a workbook people can actually use
The output is one Excel file with a tab per language: duplicates removed, readable column names, a styled header row, frozen panes, wrapped text and sensible column widths. It opens ready to read, with no manual cleanup.
with pd.ExcelWriter(OUTPUT_PATH, engine="openpyxl") as writer:
for locale in LOCALES:
df = pd.read_csv(partial_csv(locale))[COLUMNS].drop_duplicates(subset=["url"])
df.rename(columns=DISPLAY_NAMES).to_excel(writer, sheet_name=locale, index=False)
ws = writer.sheets[locale]
ws.freeze_panes = "A2"
for cell in ws[1]: # header row
cell.fill = PatternFill("solid", start_color="1F497D")
cell.font = Font(bold=True, color="FFFFFF")
cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
for row in ws.iter_rows(min_row=2): # body: wrapped, top-aligned
for cell in row:
cell.alignment = Alignment(wrap_text=True, vertical="top")
for idx, width in enumerate(COLUMN_WIDTHS, start=1):
ws.column_dimensions[get_column_letter(idx)].width = width