NReed.Dev

Sitecore | Optimizely | .NET

Content Entry: Image Harvesting

A migration needed to happen manually and I was asked to help. Among the most frustrating things is getting all the assets out of a live site, so I created a Python script to do it for me and organize the downloaded images based on the url structure.

Python
pip install beautifulsoup
Python
#!/usr/bin/env python3
import argparse
import csv
import hashlib
import os
import re
import sys
from pathlib import Path
from urllib.parse import (
    urlparse, urljoin, unquote, urldefrag,
    urlunparse, parse_qsl, urlencode
)

import requests
from bs4 import BeautifulSoup
from PIL import Image

# ----------------------- helpers -----------------------

def safe_filename(name: str) -> str:
    name = name.strip().replace('\\', '/').split('/')[-1]
    name = re.sub(r'[<>:"/\\|?*\x00-\x1F]', '_', name)
    name = name[:200] or "file"
    return name

def filename_with_ext_from_url(img_url: str, fallback_seed: str) -> str:
    img_url, _ = urldefrag(img_url)
    parsed = urlparse(img_url)
    name = safe_filename(unquote(parsed.path))
    if not name or name.endswith('/'):
        name = ''
    _, ext = os.path.splitext(name)
    if not ext:
        m = re.search(r'\.(png|jpe?g|gif|webp|svg|bmp|tiff?)($|[\W_])', img_url, re.IGNORECASE)
        ext_guess = '.' + m.group(1).lower() if m else ''
        digest = hashlib.sha1(fallback_seed.encode('utf-8')).hexdigest()[:12]
        return f"image_{digest}{ext_guess}"
    return name

def ensure_unique(path: Path) -> Path:
    if not path.exists():
        return path
    stem, ext = os.path.splitext(path.name)
    i = 2
    while True:
        candidate = path.with_name(f"{stem}_{i}{ext}")
        if not candidate.exists():
            return candidate
        i += 1

def make_output_dir(page_url: str, base_out: Path) -> Path:
    p = urlparse(page_url)
    parts = [p.netloc] + [seg for seg in p.path.strip('/').split('/') if seg]
    out_dir = base_out.joinpath(*parts)
    out_dir.mkdir(parents=True, exist_ok=True)
    return out_dir

def guess_ext_from_content_type(ct: str) -> str:
    if not ct:
        return ""
    ct = ct.lower().split(";")[0].strip()
    mapping = {
        "image/jpeg": ".jpg", "image/jpg": ".jpg",
        "image/png": ".png", "image/gif": ".gif",
        "image/webp": ".webp", "image/svg+xml": ".svg",
        "image/bmp": ".bmp", "image/tiff": ".tif",
        "image/x-icon": ".ico", "image/vnd.microsoft.icon": ".ico",
    }
    return mapping.get(ct, "")

# -------------------- srcset parsing --------------------

_srcset_item_re = re.compile(r"""
    \s*
    (?P<url>\S+)
    (?:\s+
       (?P<descriptor>
           (?:
             (?P<w>\d+)w
           )|
           (?:
             (?P<x>\d+(?:\.\d+)?)x
           )
       )
    )?
    \s*
""", re.VERBOSE)

def parse_srcset(srcset: str):
    if not srcset:
        return []
    out = []
    for candidate in srcset.split(','):
        candidate = candidate.strip()
        if not candidate:
            continue
        m = _srcset_item_re.match(candidate)
        if not m:
            out.append({"url": candidate, "w": None, "x": None})
            continue
        url = m.group("url")
        w = m.group("w")
        x = m.group("x")
        out.append({"url": url, "w": int(w) if w else None, "x": float(x) if x else None})
    return out

def pick_largest(candidates):
    if not candidates:
        return None
    any_w = any(c["w"] is not None for c in candidates)
    if any_w:
        return max((c for c in candidates if c["w"] is not None), key=lambda c: c["w"])
    any_x = any(c["x"] is not None for c in candidates)
    if any_x:
        return max((c for c in candidates if c["x"] is not None), key=lambda c: c["x"])
    return candidates[0]

def attr_first(el, *names):
    for n in names:
        v = el.get(n)
        if v:
            return v
    return None

def resolve(base_url: str, url: str) -> str:
    return urljoin(base_url, url)

def best_image_from_picture(img_tag, base_url: str):
    picture = img_tag.find_parent("picture")
    if not picture:
        return None
    all_cands = []
    for source in picture.find_all("source"):
        srcset = attr_first(source, "srcset", "data-srcset", "data-lazy-srcset")
        if not srcset:
            continue
        parsed = parse_srcset(srcset)
        for c in parsed:
            if c["url"]:
                c["url"] = resolve(base_url, c["url"])
        all_cands.extend(parsed)
    if not all_cands:
        return None
    return pick_largest(all_cands)

def best_image_url_for_img(img_tag, base_url: str):
    best_from_picture = best_image_from_picture(img_tag, base_url)
    if best_from_picture and best_from_picture.get("url"):
        return best_from_picture["url"]
    srcset = attr_first(img_tag, "srcset", "data-srcset", "data-lazy-srcset")
    if srcset:
        parsed = parse_srcset(srcset)
        for c in parsed:
            if c["url"]:
                c["url"] = resolve(base_url, c["url"])
        chosen = pick_largest(parsed)
        if chosen and chosen.get("url"):
            return chosen["url"]
    src_like = attr_first(img_tag, "src", "data-src", "data-lazy-src")
    if src_like:
        if src_like.lower().startswith("data:"):
            return None
        return resolve(base_url, src_like)
    return None

# -------------------- page collection --------------------

def collect_best_img_urls(page_url: str, html: str):
    soup = BeautifulSoup(html, "html.parser")
    urls = []
    for img in soup.find_all("img"):
        best = best_image_url_for_img(img, page_url)
        if best:
            urls.append(best)
    # Deduplicate while preserving order
    seen = set()
    deduped = []
    for u in urls:
        if u not in seen:
            deduped.append(u)
            seen.add(u)
    return deduped

# -------------------- webp/avif -> jpg rewrite --------------------

def force_jpg_url(u: str) -> str:
    """
    Rewrite common webp patterns to jpg:
    - Path .webp/.avif -> .jpg
    - fm=webp, format=webp, output=webp, ext=webp -> ...=jpg
    - type=image/webp -> image/jpeg
    - auto=webp/format -> drop those hints
    """
    try:
        p = urlparse(u)

        # swap .webp/.avif extension in the path
        path = re.sub(r'\.(webp|avif)(?=$|[?&#])', '.jpg', p.path, flags=re.IGNORECASE)

        # rewrite query params
        pairs = parse_qsl(p.query, keep_blank_values=True)
        new_pairs = []
        for k, v in pairs:
            kl = k.lower()
            vl = (v or "").lower()
            if kl in ("fm", "format", "f", "output", "ext"):
                if "webp" in vl or "avif" in vl:
                    v = "jpg"
            elif kl == "type" and "webp" in vl:
                v = "image/jpeg"
            elif kl == "auto":
                tokens = [t for t in re.split(r'[,\s]+', vl) if t]
                tokens = [t for t in tokens if t not in ("webp", "format")]
                if tokens:
                    v = ",".join(tokens)
                else:
                    continue  # drop param entirely
            new_pairs.append((k, v))

        query = urlencode(new_pairs, doseq=True)
        return urlunparse(p._replace(path=path, query=query))
    except Exception:
        return u

# -------------------- downloading (+ size & rename) --------------------

def get_image_size(path: Path):
    try:
        with Image.open(path) as im:
            im.load()
            return im.size  # (width, height)
    except Exception:
        return None

def rename_with_dimensions(path: Path, width: int, height: int) -> Path:
    stem, ext = os.path.splitext(path.name)
    # Avoid doubling if already has __WxH
    if re.search(r"__\d+x\d+$", stem):
        new_name = f"{stem}{ext}"
    else:
        new_name = f"{stem}__{width}x{height}{ext}"
    new_path = ensure_unique(path.with_name(new_name))
    if new_path != path:
        try:
            path.rename(new_path)
            return new_path
        except Exception:
            pass
    return path

def download_image(session: requests.Session, img_url: str, dest_path: Path):
    try:
        r = session.get(img_url, stream=True, timeout=30)
        r.raise_for_status()
    except Exception as e:
        print(f"  ! Failed: {img_url} -> {e}", file=sys.stderr)
        return False, None, None, None

    if dest_path.suffix == "":
        ext = guess_ext_from_content_type(r.headers.get("Content-Type", ""))
        if ext:
            dest_path = dest_path.with_suffix(ext)
            dest_path = ensure_unique(dest_path)

    with open(dest_path, "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            if chunk:
                f.write(chunk)

    size = get_image_size(dest_path)
    if size:
        w, h = size
        final_path = rename_with_dimensions(dest_path, w, h)
        return True, final_path, w, h
    else:
        # Could not read size (e.g., SVG/AVIF), return as-is
        return True, dest_path, None, None

# ------------------------ main -------------------------

def main():
    ap = argparse.ArgumentParser(
        description="Download the LARGEST image variants (via srcset), rewrite webpโ†’jpg where possible, rename files with __WIDTHxHEIGHT, and write a CSV manifest."
    )
    ap.add_argument("url", help="Page URL (e.g., https://example.com/foo/bar)")
    ap.add_argument("-o", "--out", default="downloaded_images", help="Base output directory (default: downloaded_images)")
    ap.add_argument("--manifest", default="images_manifest.csv", help="Manifest CSV filename (default: images_manifest.csv)")
    args = ap.parse_args()

    page_url = args.url
    out_base = Path(args.out)

    headers = {
        "User-Agent": "Mozilla/5.0 (compatible; ImageScraper/1.3; +https://example.org)",
        "Accept": "image/png,image/jpeg,image/gif;q=0.9,image/*;q=0.8,*/*;q=0.5",
        "Referer": page_url,
    }

    with requests.Session() as s:
        s.headers.update(headers)
        try:
            resp = s.get(page_url, timeout=45)
            resp.raise_for_status()
        except Exception as e:
            print(f"Failed to fetch page: {e}", file=sys.stderr)
            sys.exit(1)

        out_dir = make_output_dir(page_url, out_base)
        print(f"Saving images to: {out_dir}")

        img_urls = collect_best_img_urls(page_url, resp.text)
        if not img_urls:
            print("No images found.")
            return

        print(f"Found {len(img_urls)} image(s). Downloading the largest variants...")
        count_ok = 0
        manifest_rows = []

        for i, original_url in enumerate(img_urls, start=1):
            # try JPG-friendly rewrite first
            rewritten_url = force_jpg_url(original_url)

            fname = filename_with_ext_from_url(rewritten_url, fallback_seed=rewritten_url)
            dest  = ensure_unique(out_dir / fname)
            ok, saved_path, w, h = download_image(s, rewritten_url, dest)

            # fallback to original URL if rewrite fails
            if not ok and rewritten_url != original_url:
                fname = filename_with_ext_from_url(original_url, fallback_seed=original_url)
                dest  = ensure_unique(out_dir / fname)
                ok, saved_path, w, h = download_image(s, original_url, dest)

            if ok:
                count_ok += 1
                dims = f"{w}x{h}" if (w and h) else "unknown"
                print(f"  [{i}/{len(img_urls)}] Saved: {saved_path.name} ({dims})")
                manifest_rows.append({
                    "source_url": original_url,
                    "saved_filename": saved_path.name,
                    "width": w or "",
                    "height": h or ""
                })
            else:
                print(f"  [{i}/{len(img_urls)}] Skipped: {original_url}")

        # Write manifest CSV in the same output directory
        manifest_path = out_dir / args.manifest
        with open(manifest_path, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=["source_url", "saved_filename", "width", "height"])
            writer.writeheader()
            writer.writerows(manifest_rows)

        print(f"Done. {count_ok}/{len(img_urls)} image(s) saved in {out_dir}")
        print(f"Manifest written to: {manifest_path}")

if __name__ == "__main__":
    Image.MAX_IMAGE_PIXELS = None  # allow very large images
    main()

Leave a Reply

Your email address will not be published. Required fields are marked *