How to Compare Two XML Sitemaps Before and After a Site Migration With Python (Find Removed URLs and Missing Redirects)

#!/usr/bin/env python3
"""Compare two XML sitemaps (before and after a site migration)."""
import argparse
import csv
import gzip
import sys
import urllib.request
import xml.etree.ElementTree as ET
from urllib.parse import urlsplit, urlunsplit

NS = "{http://www.sitemaps.org/schemas/sitemap/0.9}"
UA = "sitemap-diff/1.0 (+https://example.com/contact)"


def read_source(src):
    """Return raw bytes from a URL or a local file path."""
    if src.startswith(("http://", "https://")):
        req = urllib.request.Request(src, headers={"User-Agent": UA})
        with urllib.request.urlopen(req, timeout=30) as resp:
            data = resp.read()
    else:
        with open(src, "rb") as fh:
            data = fh.read()
    if data[:2] == b"\x1f\x8b":  # gzip magic number
        data = gzip.decompress(data)
    return data


def collect_urls(src, seen=None, depth=0):
    """Return the set of page URLs in a sitemap, following sitemap indexes."""
    seen = seen if seen is not None else set()
    if src in seen or depth > 3:
        return set()
    seen.add(src)
    root = ET.fromstring(read_source(src))
    tag = root.tag.replace(NS, "")
    locs = [el.text.strip() for el in root.iter(NS + "loc") if el.text]
    if tag == "sitemapindex":
        urls = set()
        for child in locs:
            urls |= collect_urls(child, seen, depth + 1)
        return urls
    if tag == "urlset":
        return set(locs)
    raise ValueError(f"{src}: root element is <{tag}>, expected urlset or sitemapindex")


def normalise(url, old_host=None, new_host=None, ignore_slash=True):
    """Make URLs comparable: lowercase host, optional host swap, trailing slash."""
    parts = urlsplit(url)
    host = parts.netloc.lower()
    if old_host and host == old_host.lower():
        host = new_host.lower()
    path = parts.path or "/"
    if ignore_slash and len(path) > 1:
        path = path.rstrip("/")
    return urlunsplit((parts.scheme.lower(), host, path, parts.query, ""))


def load_redirects(path):
    """Read a two-column CSV of old_url,new_url."""
    with open(path, newline="", encoding="utf-8") as fh:
        return {row[0].strip(): row[1].strip() for row in csv.reader(fh) if len(row) >= 2}


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("before", help="old sitemap: URL or file")
    ap.add_argument("after", help="new sitemap: URL or file")
    ap.add_argument("--old-host", help="host used before the move, e.g. www.old.com")
    ap.add_argument("--new-host", help="host used after the move, e.g. www.new.com")
    ap.add_argument("--redirects", help="CSV of old_url,new_url to check removed URLs against")
    ap.add_argument("--out", default="sitemap-diff.csv", help="where to write the report")
    args = ap.parse_args()

    if bool(args.old_host) != bool(args.new_host):
        ap.error("--old-host and --new-host go together")

    before_raw = collect_urls(args.before)
    after_raw = collect_urls(args.after)

    before = {normalise(u, args.old_host, args.new_host): u for u in before_raw}
    after = {normalise(u): u for u in after_raw}

    kept = before.keys() & after.keys()
    removed = before.keys() - after.keys()
    added = after.keys() - before.keys()

    redirects = load_redirects(args.redirects) if args.redirects else {}
    after_norm = set(after.keys())

    rows = []
    for key in sorted(removed):
        old = before[key]
        target = redirects.get(old, "")
        if not redirects:
            note = ""
        elif not target:
            note = "no redirect mapped"
        elif normalise(target) not in after_norm:
            note = "redirect target not in new sitemap"
        else:
            note = "ok"
        rows.append(("removed", old, target, note))
    rows += [("added", after[k], "", "") for k in sorted(added)]
    rows += [("kept", after[k], "", "") for k in sorted(kept)]

    with open(args.out, "w", newline="", encoding="utf-8") as fh:
        writer = csv.writer(fh)
        writer.writerow(["status", "url", "redirect_to", "note"])
        writer.writerows(rows)

    print(f"before: {len(before)}  after: {len(after)}")
    print(f"kept: {len(kept)}  added: {len(added)}  removed: {len(removed)}")
    if redirects:
        problems = sum(1 for r in rows if r[0] == "removed" and r[3] != "ok")
        print(f"removed URLs without a valid redirect: {problems}")
    print(f"report: {args.out}")
    return 1 if redirects and problems else 0


if __name__ == "__main__":
    sys.exit(main())

python3 sitemap_diff.py old-sitemap.xml https://www.new-shop.com/sitemap_index.xml \
  --old-host www.old-shop.com --new-host www.new-shop.com \
  --redirects redirects.csv

before: 4  after: 5
kept: 2  added: 3  removed: 2
removed URLs without a valid redirect: 1
report: sitemap-diff.csv

status,url,redirect_to,note
removed,https://www.old-shop.com/about-us,https://www.new-shop.com/about,redirect target not in new sitemap
removed,https://www.old-shop.com/blog/launch-post,https://www.new-shop.com/blog/launch,ok
added,https://www.new-shop.com/blog/launch,,
added,https://www.new-shop.com/company,,
added,https://www.new-shop.com/integrations,,
kept,https://www.new-shop.com/,,
kept,https://www.new-shop.com/pricing,,

#!/usr/bin/env python3
"""For each removed URL in sitemap-diff.csv, request it and report where it lands."""
import csv
import sys
import urllib.error
import urllib.request


class NoFollow(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *args, **kwargs):
        return None  # stop at the first hop so we can see the status code


opener = urllib.request.build_opener(NoFollow)


def first_hop(url, method="HEAD"):
    req = urllib.request.Request(url, method=method, headers={"User-Agent": "sitemap-diff/1.0"})
    try:
        with opener.open(req, timeout=20) as resp:
            return resp.status, ""
    except urllib.error.HTTPError as err:  # 3xx lands here because we refuse to follow
        if method == "HEAD" and err.code in (405, 500, 501):
            return first_hop(url, "GET")  # some servers reject HEAD
        return err.code, err.headers.get("Location", "")
    except urllib.error.URLError as err:
        return 0, str(err.reason)


report = sys.argv[1] if len(sys.argv) > 1 else "sitemap-diff.csv"
with open(report, newline="", encoding="utf-8") as fh:
    for row in csv.DictReader(fh):
        if row["status"] != "removed":
            continue
        code, location = first_hop(row["url"])
        flag = "" if code in (301, 308) else "  <-- check"
        print(f"{code} {row['url']} -> {location}{flag}")

301 https://www.old-shop.com/blog/launch-post -> https://www.new-shop.com/blog/launch
404 https://www.old-shop.com/about-us ->   <-- check

# fails the job if any removed URL lacks a valid redirect
python3 sitemap_diff.py old-sitemap.xml staging-sitemap.xml \
  --old-host www.old-shop.com --new-host staging.new-shop.com \
  --redirects redirects.csv --out diff.csv