519 lines
20 KiB
Python
519 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Scrape products + category tree from http://e-shop.kompress.sk (PrestaShop 1.5/1.6).
|
|
|
|
Outputs (next to this script):
|
|
categories.json - full category tree (flat list with parent_id + path)
|
|
products.json - one entry per product, with category memberships
|
|
images/<id>/ - original (highest quality) product images
|
|
"""
|
|
|
|
import html
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from html.parser import HTMLParser
|
|
|
|
BASE = "http://e-shop.kompress.sk"
|
|
OUT = os.path.dirname(os.path.abspath(__file__))
|
|
IMG_DIR = os.path.join(OUT, "images")
|
|
UA = "Mozilla/5.0 (X11; Linux x86_64) data-migration-scraper"
|
|
DELAY = 0.3
|
|
|
|
|
|
# ---------------------------------------------------------------- fetching
|
|
|
|
def fetch(url, binary=False, retries=3):
|
|
for attempt in range(retries):
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": UA})
|
|
with urllib.request.urlopen(req, timeout=45) as r:
|
|
data = r.read()
|
|
return data if binary else data.decode("utf-8", "replace")
|
|
except urllib.error.HTTPError as e:
|
|
if e.code == 404:
|
|
return None
|
|
if attempt == retries - 1:
|
|
print(" !! %s -> HTTP %s" % (url, e.code), file=sys.stderr)
|
|
return None
|
|
except Exception as e: # noqa: BLE001 - network flakiness
|
|
if attempt == retries - 1:
|
|
print(" !! %s -> %s" % (url, e), file=sys.stderr)
|
|
return None
|
|
time.sleep(1 + attempt)
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------- helpers
|
|
|
|
TAG_RE = re.compile(r"<[^>]+>")
|
|
WS_RE = re.compile(r"[ \t\r\f\v]+")
|
|
|
|
|
|
def unesc(s):
|
|
return html.unescape(s or "").strip()
|
|
|
|
|
|
def strip_tags(h):
|
|
"""HTML -> readable plain text, keeping block/list structure as newlines."""
|
|
if not h:
|
|
return ""
|
|
t = re.sub(r"(?i)<\s*br\s*/?>", "\n", h)
|
|
t = re.sub(r"(?i)</\s*(p|div|li|tr|h[1-6]|ul|ol)\s*>", "\n", t)
|
|
t = re.sub(r"(?i)<\s*li[^>]*>", "• ", t)
|
|
t = TAG_RE.sub("", t)
|
|
t = html.unescape(t)
|
|
t = WS_RE.sub(" ", t)
|
|
t = re.sub(r" *\n *", "\n", t)
|
|
t = re.sub(r"\n{3,}", "\n\n", t)
|
|
return t.strip()
|
|
|
|
|
|
def find(pattern, text, group=1, flags=re.S):
|
|
m = re.search(pattern, text, flags)
|
|
return m.group(group) if m else None
|
|
|
|
|
|
def clean_content(html):
|
|
"""Strip the old shop's wrapper markup, keeping the authored rich text.
|
|
|
|
A description is genuinely rich text — `<p>`, `<ul>`, `<li>`, `<strong>` are
|
|
content and are kept verbatim. What is not content is the layout scaffolding
|
|
that accumulated in these fields over the years: `<div>`s someone pasted in
|
|
(`<div id="product_images" class="hlavni_obrazek">`) and stray `id`
|
|
attributes left over from the PrestaShop theme (`tabs-1`, `idTab-yotpo`).
|
|
Those ids would be injected into every page that renders the description and
|
|
can collide with the host page's own ids; the `<div>`s carry no formatting
|
|
here. Only the tags are dropped, never their contents.
|
|
"""
|
|
if not html:
|
|
return ""
|
|
h = re.sub(r"(?i)</?div\b[^>]*>", "", html)
|
|
h = re.sub(r'(?i)\s+id\s*=\s*"[^"]*"', "", h)
|
|
return h.strip()
|
|
|
|
|
|
def extract_balanced(page, open_pattern, tag="div"):
|
|
"""Inner HTML of the element opened by `open_pattern`, nesting-aware.
|
|
|
|
The description fields contain nested `<div>`s (the shop's own content has
|
|
leftovers like `<div id="product_images">` pasted into them). A non-greedy
|
|
`(.*?)</div>` stops at the *first* closing tag, which truncates the field
|
|
mid-element and yields markup with an unclosed `<div>` — that then breaks the
|
|
layout of whatever page renders it. Counting depth is the only correct way to
|
|
find the matching close.
|
|
"""
|
|
m = re.search(open_pattern, page, re.S | re.I)
|
|
if not m:
|
|
return None
|
|
start, depth = m.end(), 1
|
|
token = re.compile(r"<(/?)%s\b[^>]*>" % tag, re.S | re.I)
|
|
pos = start
|
|
while True:
|
|
t = token.search(page, pos)
|
|
if not t:
|
|
# Source itself is unbalanced; return what we have rather than lose it.
|
|
return page[start:]
|
|
depth += -1 if t.group(1) else 1
|
|
if depth == 0:
|
|
return page[start:t.start()]
|
|
pos = t.end()
|
|
|
|
|
|
def cat_id_from_url(url):
|
|
m = re.match(r"^%s/(\d+)-" % re.escape(BASE), url)
|
|
return int(m.group(1)) if m else None
|
|
|
|
|
|
def product_id_from_url(url):
|
|
m = re.search(r"/(\d+)-[^/]*\.html", url)
|
|
return int(m.group(1)) if m else None
|
|
|
|
|
|
# ---------------------------------------------------------------- categories
|
|
|
|
class TreeParser(HTMLParser):
|
|
"""Parses the nested <ul class="tree"> category block into (id, name, title, url, parent)."""
|
|
|
|
def __init__(self):
|
|
super().__init__(convert_charrefs=False)
|
|
self.in_tree = 0
|
|
self.depth = 0
|
|
self.stack = [] # category id per <ul> level
|
|
self.cur = None
|
|
self.rows = []
|
|
|
|
def handle_starttag(self, tag, attrs):
|
|
a = dict(attrs)
|
|
if tag == "ul":
|
|
if self.in_tree:
|
|
self.depth += 1
|
|
elif "tree" in (a.get("class") or ""):
|
|
self.in_tree = 1
|
|
self.depth = 1
|
|
elif tag == "a" and self.in_tree and a.get("href", "").startswith(BASE):
|
|
cid = cat_id_from_url(a["href"])
|
|
if cid is None:
|
|
return
|
|
parent = self.stack[self.depth - 2] if self.depth >= 2 else None
|
|
self.cur = {
|
|
"id": cid,
|
|
"name": "",
|
|
"title": unesc(a.get("title", "")),
|
|
"url": a["href"],
|
|
"parent_id": parent,
|
|
}
|
|
while len(self.stack) < self.depth:
|
|
self.stack.append(None)
|
|
self.stack[self.depth - 1] = cid
|
|
self.rows.append(self.cur)
|
|
|
|
def handle_endtag(self, tag):
|
|
if tag == "ul" and self.in_tree:
|
|
self.depth -= 1
|
|
if self.depth <= 0:
|
|
self.in_tree = 0
|
|
elif tag == "a":
|
|
self.cur = None
|
|
|
|
def handle_data(self, data):
|
|
if self.cur is not None:
|
|
self.cur["name"] += data
|
|
|
|
def handle_entityref(self, name):
|
|
if self.cur is not None:
|
|
self.cur["name"] += "&%s;" % name
|
|
|
|
def handle_charref(self, name):
|
|
if self.cur is not None:
|
|
self.cur["name"] += "&#%s;" % name
|
|
|
|
|
|
def scrape_categories():
|
|
home = fetch(BASE + "/")
|
|
p = TreeParser()
|
|
p.feed(home)
|
|
cats = {}
|
|
for row in p.rows:
|
|
row["name"] = unesc(row["name"])
|
|
cats[row["id"]] = row
|
|
|
|
# enrich each category from its own page: h1, description, product count
|
|
def enrich(c):
|
|
page = fetch(c["url"])
|
|
time.sleep(DELAY)
|
|
if not page:
|
|
return
|
|
h1 = find(r'<h1[^>]*>(.*?)</h1>', page)
|
|
if h1:
|
|
c["name"] = strip_tags(h1)
|
|
desc = find(r'<div[^>]*class="[^"]*cat_desc[^"]*"[^>]*>\s*(.*?)\s*</div>\s*</div>', page)
|
|
if desc is not None and desc.startswith("<div>") is False:
|
|
desc = find(r'<div[^>]*class="[^"]*cat_desc[^"]*"[^>]*>\s*(.*?)\s*</div>', page)
|
|
if desc:
|
|
desc = re.sub(r"^<div>\s*|\s*</div>$", "", desc.strip())
|
|
c["description_html"] = (desc or "").strip()
|
|
c["description"] = strip_tags(desc)
|
|
|
|
# category thumbnail (original upload, if any)
|
|
img = find(r'<img[^>]+src="[^"]*/c/%d-\w+_default/[^"]*"' % c["id"], page, 0)
|
|
c["image"] = None
|
|
if img or ('/c/%d-' % c["id"]) in page:
|
|
data = fetch("%s/img/c/%d.jpg" % (BASE, c["id"]), binary=True)
|
|
if data:
|
|
os.makedirs(os.path.join(IMG_DIR, "categories"), exist_ok=True)
|
|
rel = "images/categories/%d.jpg" % c["id"]
|
|
with open(os.path.join(OUT, rel), "wb") as f:
|
|
f.write(data)
|
|
c["image"] = {"url": "%s/img/c/%d.jpg" % (BASE, c["id"]), "file": rel}
|
|
|
|
c["product_urls"] = list_category_products(c["url"], page)
|
|
c["product_ids"] = [product_id_from_url(u) for u in c["product_urls"]]
|
|
|
|
with ThreadPoolExecutor(max_workers=4) as ex:
|
|
list(ex.map(enrich, cats.values()))
|
|
|
|
# path / depth
|
|
def path_of(cid):
|
|
parts, seen = [], set()
|
|
while cid and cid in cats and cid not in seen:
|
|
seen.add(cid)
|
|
parts.append(cats[cid]["name"])
|
|
cid = cats[cid]["parent_id"]
|
|
return list(reversed(parts))
|
|
|
|
for c in cats.values():
|
|
c["path"] = path_of(c["id"])
|
|
c["depth"] = len(c["path"]) - 1
|
|
return cats
|
|
|
|
|
|
def list_category_products(url, first_page=None):
|
|
"""All product URLs in a category, following pagination."""
|
|
urls, seen_pages, page_no = [], set(), 1
|
|
page = first_page if first_page is not None else fetch(url)
|
|
while page:
|
|
for m in re.finditer(r'<a[^>]+class="product_name"[^>]+href="([^"]+)"', page):
|
|
u = unesc(m.group(1))
|
|
if u not in urls:
|
|
urls.append(u)
|
|
if not urls: # theme variation: any product link inside the list
|
|
block = find(r'<ul id="product_list".*?</ul>', page, 0)
|
|
for m in re.finditer(r'href="(%s/[^"]+\.html)"' % re.escape(BASE), block or ""):
|
|
u = unesc(m.group(1))
|
|
if u not in urls:
|
|
urls.append(u)
|
|
page_no += 1
|
|
nxt = "%s?p=%d" % (url, page_no)
|
|
if page_no > 40 or ('p=%d' % page_no) not in page or nxt in seen_pages:
|
|
break
|
|
seen_pages.add(nxt)
|
|
time.sleep(DELAY)
|
|
page = fetch(nxt)
|
|
return urls
|
|
|
|
|
|
# ---------------------------------------------------------------- products
|
|
|
|
def original_image_url(image_id):
|
|
"""PrestaShop stores the untouched upload at /img/p/<digits split by />/<id>.jpg."""
|
|
s = str(image_id)
|
|
return "%s/img/p/%s/%s.jpg" % (BASE, "/".join(s), s)
|
|
|
|
|
|
ADD_COMBINATION_RE = re.compile(
|
|
r"addCombination\(\s*(\d+)\s*,\s*new Array\(([^)]*)\)\s*,\s*(-?\d+)\s*,"
|
|
r"\s*(-?[\d.]+)\s*,\s*(-?[\d.]+)\s*,\s*(-?\d+)\s*,\s*'([^']*)'\s*,"
|
|
r"\s*(-?[\d.]+)\s*,\s*(\d+)\s*,")
|
|
|
|
|
|
def parse_variants(page, base_price_excl, tax_rate):
|
|
"""Attribute groups + combinations (PrestaShop's addCombination() calls).
|
|
|
|
addCombination(id_product_attribute, [id_attribute...], quantity, price_impact,
|
|
ecotax, id_image, reference, unit_price, minimal_quantity, ...)
|
|
where price_impact is tax-EXCLUDED, per themes/default/js/product.js.
|
|
"""
|
|
# id_attribute -> group slug, from the attributesCombinations JS array
|
|
attr_group = {}
|
|
for blk in re.findall(r"tabInfos\['id_attribute'\].*?attributesCombinations\.push", page, re.S):
|
|
aid = find(r"tabInfos\['id_attribute'\] = '(\d+)'", blk)
|
|
if aid:
|
|
attr_group[int(aid)] = find(r"tabInfos\['group'\] = '([^']*)'", blk)
|
|
|
|
# human labels from the #attributes widgets (<select> on most products,
|
|
# radio buttons on a few)
|
|
groups, labels = [], {}
|
|
attrs_html = find(r'<div id="attributes">(.*?)\n\s*</div>\s*<p id="product_reference"', page) or \
|
|
find(r'<div id="attributes">(.*)</div>', page) or ""
|
|
for fs in re.findall(r'<fieldset class="attribute_fieldset">(.*?)</fieldset>', attrs_html, re.S):
|
|
gname = strip_tags(find(r'<label[^>]*class="attribute_label"[^>]*>(.*?)</label>', fs) or "")
|
|
gname = gname.rstrip(": ").strip()
|
|
gid = find(r'name="group_(\d+)"', fs)
|
|
opts = []
|
|
for m in re.finditer(r'<option value="(\d+)"([^>]*)>(.*?)</option>', fs, re.S):
|
|
aid, rest, txt = int(m.group(1)), m.group(2), strip_tags(m.group(3))
|
|
labels[aid] = txt
|
|
opts.append({"id": aid, "label": txt, "default": "selected" in rest})
|
|
for m in re.finditer(
|
|
r'<input type="radio"[^>]*value="(\d+)"([^>]*)>\s*<span>(.*?)</span>', fs, re.S):
|
|
aid, rest, txt = int(m.group(1)), m.group(2), strip_tags(m.group(3))
|
|
labels[aid] = txt
|
|
opts.append({"id": aid, "label": txt, "default": "checked" in rest})
|
|
groups.append({"id": int(gid) if gid else None, "name": gname,
|
|
"slug": attr_group.get(opts[0]["id"]) if opts else None,
|
|
"attributes": opts})
|
|
|
|
variants = []
|
|
for m in ADD_COMBINATION_RE.finditer(page):
|
|
ids = [int(x) for x in re.findall(r"'(\d+)'", m.group(2))]
|
|
impact = float(m.group(4))
|
|
variants.append({
|
|
"id": int(m.group(1)), # id_product_attribute
|
|
"attribute_ids": ids,
|
|
"label": " / ".join(labels.get(i, str(i)) for i in ids),
|
|
"attributes": [{"group": attr_group.get(i), "label": labels.get(i, "")} for i in ids],
|
|
"reference": m.group(7),
|
|
"quantity_available": int(m.group(3)),
|
|
"price_impact_tax_excluded": impact,
|
|
"price_tax_excluded": round(base_price_excl + impact, 6) if base_price_excl is not None else None,
|
|
"price": round((base_price_excl + impact) * (1 + tax_rate / 100.0), 2)
|
|
if base_price_excl is not None else None,
|
|
"minimal_quantity": int(m.group(9)),
|
|
"image_id": int(m.group(6)) if int(m.group(6)) > 0 else None,
|
|
})
|
|
default_ids = {a["id"] for g in groups for a in g["attributes"] if a["default"]}
|
|
for v in variants:
|
|
v["is_default"] = bool(default_ids) and set(v["attribute_ids"]) == default_ids
|
|
return groups, variants
|
|
|
|
|
|
def parse_product(url, page):
|
|
pid = product_id_from_url(url) or int(find(r"var id_product = '(\d+)'", page) or 0)
|
|
|
|
name = strip_tags(find(r'<div id="pb-left-column">\s*<h1[^>]*>(.*?)</h1>', page)
|
|
or find(r'<h1[^>]*>(.*?)</h1>', page) or "")
|
|
|
|
short_html = clean_content(extract_balanced(page, r'<div id="short_description_content"[^>]*>'))
|
|
desc_html = clean_content(extract_balanced(page, r'<div id="idTab1"[^>]*>'))
|
|
|
|
# breadcrumb -> category trail (last node is the product name, not a link)
|
|
crumb = find(r'<div class="breadcrumb">(.*?)</div>', page) or ""
|
|
trail = []
|
|
for m in re.finditer(r'<a[^>]+href="(%s/(\d+)-[^"]*)"[^>]*>(.*?)</a>' % re.escape(BASE), crumb):
|
|
trail.append({"id": int(m.group(2)), "name": strip_tags(m.group(3)), "url": m.group(1)})
|
|
|
|
# images: thumbnails list, else the single main image
|
|
img_ids = [int(i) for i in re.findall(r'<li id="thumbnail_(\d+)"', page)]
|
|
if not img_ids:
|
|
main = find(r'id="bigpic"[^>]*src="[^"]*?/(\d+)-large_default/', page) or \
|
|
find(r'src="[^"]*?/(\d+)-large_default/', page)
|
|
if main:
|
|
img_ids = [int(main)]
|
|
default_img = find(r"var idDefaultImage = (\d+);", page)
|
|
|
|
qty = find(r"var quantityAvailable = (-?\d+);", page)
|
|
price_excl = find(r"var productPriceTaxExcluded = ([\d.]+)", page)
|
|
tax = float(find(r"var taxRate = ([\d.]+);", page) or 0)
|
|
groups, variants = parse_variants(
|
|
page, float(price_excl) if price_excl else None, tax)
|
|
|
|
return {
|
|
"id": pid,
|
|
"name": name,
|
|
"url": url,
|
|
"reference": unesc(find(r'<p id="product_reference"[^>]*>.*?<span[^>]*>(.*?)</span>', page) or
|
|
find(r"var productReference = '([^']*)'", page) or ""),
|
|
"price": float(find(r"var productPrice = '([\d.]+)'", page) or 0),
|
|
"price_display": strip_tags(find(r'id="our_price_display"[^>]*>(.*?)</span>', page) or ""),
|
|
"price_tax_excluded": round(float(price_excl), 6) if price_excl else None,
|
|
"price_without_reduction": float(find(r"var productPriceWithoutReduction = '([\d.]+)'", page) or 0),
|
|
"tax_rate": tax,
|
|
"currency": "EUR",
|
|
"quantity_available": int(qty) if qty is not None else None,
|
|
"available_for_order": find(r"var productAvailableForOrder = '(\d)'", page) == "1",
|
|
"has_attributes": find(r"var productHasAttributes = (\w+);", page) == "true",
|
|
"attribute_groups": groups,
|
|
"variants": variants,
|
|
"short_description_html": (short_html or "").strip(),
|
|
"short_description": strip_tags(short_html),
|
|
"description_html": (desc_html or "").strip(),
|
|
"description": strip_tags(desc_html),
|
|
"breadcrumb": trail,
|
|
"category_path": [c["name"] for c in trail],
|
|
"default_image_id": int(default_img) if default_img else None,
|
|
"images": [{"id": i,
|
|
"url": original_image_url(i),
|
|
"page_url": url,
|
|
"file": "images/%d/%d.jpg" % (pid, i)} for i in img_ids],
|
|
}
|
|
|
|
|
|
def download_images(prod):
|
|
d = os.path.join(IMG_DIR, str(prod["id"]))
|
|
os.makedirs(d, exist_ok=True)
|
|
for img in prod["images"]:
|
|
path = os.path.join(OUT, img["file"])
|
|
if os.path.exists(path) and os.path.getsize(path) > 0:
|
|
img["bytes"] = os.path.getsize(path)
|
|
continue
|
|
data = fetch(img["url"], binary=True)
|
|
if data is None: # fall back to the largest generated thumbnail
|
|
alt = "%s/%d-thickbox_default/x.jpg" % (BASE, img["id"])
|
|
data = fetch(alt, binary=True)
|
|
if data:
|
|
img["url"] = alt
|
|
if data:
|
|
with open(path, "wb") as f:
|
|
f.write(data)
|
|
img["bytes"] = len(data)
|
|
else:
|
|
img["bytes"] = 0
|
|
time.sleep(DELAY)
|
|
|
|
|
|
# ---------------------------------------------------------------- main
|
|
|
|
def main():
|
|
os.makedirs(IMG_DIR, exist_ok=True)
|
|
|
|
print("Scraping category tree ...")
|
|
cats = scrape_categories()
|
|
print(" %d categories" % len(cats))
|
|
|
|
# product URLs: category listings + sitemap (catches anything unlisted)
|
|
urls = []
|
|
for c in cats.values():
|
|
for u in c["product_urls"]:
|
|
if u not in urls:
|
|
urls.append(u)
|
|
sm = fetch(BASE + "/sitemap.xml") or ""
|
|
for m in re.finditer(r'(%s/[^\]\s]+\.html)' % re.escape(BASE), sm):
|
|
if m.group(1) not in urls:
|
|
urls.append(m.group(1))
|
|
print(" %d product URLs" % len(urls))
|
|
|
|
products, failed = [], []
|
|
|
|
def work(u):
|
|
page = fetch(u)
|
|
time.sleep(DELAY)
|
|
if not page or 'id="product_page_product_id"' not in page:
|
|
failed.append(u)
|
|
return None
|
|
p = parse_product(u, page)
|
|
download_images(p)
|
|
return p
|
|
|
|
with ThreadPoolExecutor(max_workers=4) as ex:
|
|
for i, p in enumerate(ex.map(work, urls), 1):
|
|
if p:
|
|
products.append(p)
|
|
if i % 20 == 0:
|
|
print(" %d/%d" % (i, len(urls)))
|
|
|
|
# a product can be served under an old and a new slug; keep the first
|
|
# (category-listing) URL, which is the canonical one
|
|
seen_ids, unique = set(), []
|
|
for p in products:
|
|
if p["id"] in seen_ids:
|
|
continue
|
|
seen_ids.add(p["id"])
|
|
unique.append(p)
|
|
products = sorted(unique, key=lambda p: p["id"])
|
|
|
|
# attach every category a product is listed in (a product can be in several)
|
|
by_id = {p["id"]: p for p in products}
|
|
for p in products:
|
|
p["categories"] = []
|
|
for c in sorted(cats.values(), key=lambda c: c["id"]):
|
|
for pid in c["product_ids"]:
|
|
if pid in by_id:
|
|
by_id[pid]["categories"].append(
|
|
{"id": c["id"], "name": c["name"], "path": c["path"]})
|
|
|
|
cat_list = sorted(cats.values(), key=lambda c: (c["depth"], c["id"]))
|
|
for c in cat_list:
|
|
c["product_count"] = len(c["product_ids"])
|
|
c.pop("product_urls", None)
|
|
|
|
with open(os.path.join(OUT, "categories.json"), "w", encoding="utf-8") as f:
|
|
json.dump(cat_list, f, ensure_ascii=False, indent=2)
|
|
with open(os.path.join(OUT, "products.json"), "w", encoding="utf-8") as f:
|
|
json.dump(products, f, ensure_ascii=False, indent=2)
|
|
|
|
imgs = sum(len(p["images"]) for p in products)
|
|
print("\nDone: %d products, %d categories, %d images" % (len(products), len(cat_list), imgs))
|
|
if failed:
|
|
print("Failed URLs (%d):" % len(failed))
|
|
for u in failed:
|
|
print(" " + u)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|