Compare historical domain snapshots reproducibly

Python example: Research

Requires Python 3.9 or later. Uses the standard library only, processes local files, and makes no network requests. Save the script beside your downloaded inputs, or provide full paths.

Download research.py

Run the example

python3 research.py before.txt.gz after.txt.gz

Results are printed to the terminal. Redirect standard output to a file if you want to save the report. For the SQLite example, the database is saved at the path you specify.

Complete script

import argparse
import csv
import gzip
import sys


def domains(path):
    with gzip.open(path, "rt", encoding="utf-8") as source:
        for line in source:
            domain = line.strip().lower().rstrip(".")
            if domain:
                yield domain

import hashlib
import json
import sqlite3
import tempfile
from pathlib import Path

parser = argparse.ArgumentParser(description="Compare two domain snapshots")
parser.add_argument("before")
parser.add_argument("after")
args = parser.parse_args()

def checksum(path):
    digest = hashlib.sha256()
    with open(path, "rb") as source:
        for chunk in iter(lambda: source.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()

with tempfile.TemporaryDirectory() as temp:
    connection = sqlite3.connect(str(Path(temp) / "comparison.sqlite"))
    try:
        with connection:
            for table, path in [("before_names", args.before), ("after_names", args.after)]:
                connection.execute(f"CREATE TABLE {table} (domain TEXT PRIMARY KEY)")
                connection.executemany(f"INSERT OR IGNORE INTO {table} VALUES (?)",
                                       ((name,) for name in domains(path)))
        added = connection.execute("SELECT COUNT(*) FROM (SELECT domain FROM after_names "
                                   "EXCEPT SELECT domain FROM before_names)").fetchone()[0]
        removed = connection.execute("SELECT COUNT(*) FROM (SELECT domain FROM before_names "
                                     "EXCEPT SELECT domain FROM after_names)").fetchone()[0]
    finally:
        connection.close()
print(json.dumps({"before_file": args.before, "after_file": args.after,
                  "before_sha256": checksum(args.before), "after_sha256": checksum(args.after),
                  "observed_additions": added, "observed_removals": removed}, indent=2))

To automate input downloads, follow the API documentation for tokens, supported endpoints, and historical dates. Keep API tokens out of shared scripts.