integrity_check.py

"""
Integrity / anomaly report for jacob_olie_sources.xlsx.

Reads the workbook and runs a battery of consistency checks; prints a summary
to stdout and writes per-anomaly row lists to ``integrity_report.txt``.

The checks are read-only — the workbook is opened with ``read_only=True`` and
never written back.
"""

import re
import sys
import collections
import unicodedata
from urllib.parse import urlparse, unquote

import openpyxl

XLSX_PATH = "jacob_olie_sources.xlsx"
REPORT_PATH = "integrity_report.txt"

UUID_RE = re.compile(
    r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", re.I
)

# Characters that should never appear inside an Excel cell. ASCII control
# chars (except TAB \t, LF \n, CR \r), Unicode line/paragraph separators,
# zero-width characters, and the byte-order mark.
NONPRINT_RE = re.compile(
    "["
    "\x00-\x08\x0B\x0C\x0E-\x1F\x7F"  # ASCII control (except \t \n \r)
    "

"                     # Unicode line / paragraph separator
    "​‌‍⁠"         # zero-width space / ZWNJ / ZWJ / WJ
    ""                           # BOM
    "]"
)


def char_label(ch):
    """Pretty-print a single character as ``U+XXXX (NAME)``."""
    name = unicodedata.name(ch, "")
    return f"U+{ord(ch):04X} ({name})" if name else f"U+{ord(ch):04X}"

METADATA_HEADERS = [
    "Titel (dc_title)",
    "Beschrijving (dc_description)",
    "Datering (dc_date)",
    "Documenttype (sk_documenttype)",
    "Vervaardiger (sk_vervaardiger)",
    "Collectie (dc_provenance)",
    "Geografische aanduiding (geografische_aanduiding)",
    "Gebouw (sk_gebouw)",
    "Inventarissen (dc_source)",
    "Afbeeldingsbestand (identifier)",
    "Rechthebbende (sr_rechthebbende)",
]


def cell(row, col):
    if col is None:
        return None
    v = row[col - 1].value
    return v


def stripped(v):
    if v is None:
        return ""
    return str(v).strip()


def main():
    wb = openpyxl.load_workbook(XLSX_PATH, read_only=True)
    ws = wb.active

    headers = {}
    for c in range(1, ws.max_column + 1):
        v = ws.cell(row=1, column=c).value
        if v is not None:
            headers[str(v).strip()] = c

    out_lines = []

    def out(s=""):
        out_lines.append(s)

    out(f"Integrity report for {XLSX_PATH}")
    out(f"Sheet: {ws.title}    rows: {ws.max_row}    cols: {ws.max_column}")
    out("=" * 78)

    required = [
        "Filename", "File URL (Commons)", "Source URL",
        "Archief Amsterdam URL", "Archief Amsterdam Detail URL",
        "Beta Archief Amsterdam Detail URL", "URL replacement succes",
    ]
    missing_headers = [h for h in required if h not in headers]
    if missing_headers:
        out(f"!!! Required headers missing: {missing_headers}")
    out()

    fname_c = headers.get("Filename")
    file_url_c = headers.get("File URL (Commons)")
    source_c = headers.get("Source URL")
    archief_url_c = headers.get("Archief Amsterdam URL")
    detail_c = headers.get("Archief Amsterdam Detail URL")
    beta_c = headers.get("Beta Archief Amsterdam Detail URL")
    result_c = headers.get("URL replacement succes")
    inv_c = headers.get("Inventarissen (dc_source)")
    metadata_cols = [(h, headers[h]) for h in METADATA_HEADERS if h in headers]

    # Collect every row's key fields once
    rows_data = []
    for row in ws.iter_rows(min_row=2):
        per_field = {h: stripped(cell(row, c)) for h, c in metadata_cols}
        rows_data.append({
            "row_idx": row[0].row,
            "filename": stripped(cell(row, fname_c)),
            "file_url": stripped(cell(row, file_url_c)),
            "source_url": stripped(cell(row, source_c)),
            "archief_url": stripped(cell(row, archief_url_c)),
            "detail_url": stripped(cell(row, detail_c)),
            "beta_url": stripped(cell(row, beta_c)),
            "inventarissen": stripped(cell(row, inv_c)) if inv_c else "",
            "result": stripped(cell(row, result_c)).upper(),
            "per_field": per_field,
            "metadata_filled": sum(1 for v in per_field.values() if v),
            "metadata_with_error": [
                h for h, v in per_field.items() if v.startswith("ERROR")
            ],
        })

    total = len(rows_data)
    out(f"Data rows: {total}")
    out()

    # 1. Outcome distribution
    outcomes = collections.Counter(r["result"] or "(empty)" for r in rows_data)
    out("--- Outcome distribution (URL replacement succes) ---")
    for k, v in outcomes.most_common():
        out(f"  {k:12s} {v:5d}")
    out()

    # 2. Anomaly: TRUE rows must have a Beta URL
    no_beta_but_true = [r for r in rows_data if r["result"] == "TRUE" and not r["beta_url"]]
    out(f"--- Anomaly: TRUE rows with no Beta URL: {len(no_beta_but_true)} ---")
    for r in no_beta_but_true[:20]:
        out(f"  row {r['row_idx']}: {r['filename']!r}")
    if len(no_beta_but_true) > 20:
        out(f"  ... and {len(no_beta_but_true) - 20} more")
    out()

    # 3. FALSE rows breakdown (informational, not an anomaly)
    # The "no_beta_url" sub-bucket is the normal outcome for rows that have
    # no replacement target on the Beeldbank — not a failure of the script.
    # Other sub-buckets (no_filename, no_file_url, no_source_url, other)
    # represent genuine processing failures worth investigating.
    false_rows = [r for r in rows_data if r["result"] == "FALSE"]
    out(f"--- FALSE rows: {len(false_rows)} ---")
    breakdown = collections.Counter()
    for r in false_rows:
        if not r["beta_url"]:
            breakdown["no_beta_url"] += 1
        elif not r["filename"]:
            breakdown["no_filename"] += 1
        elif not r["file_url"]:
            breakdown["no_file_url"] += 1
        elif not r["source_url"]:
            breakdown["no_source_url"] += 1
        else:
            breakdown["other"] += 1
    for k, v in breakdown.most_common():
        out(f"  {k:20s} {v:5d}")
    if breakdown.get("other"):
        out("  (sample 'other' rows)")
        for r in [x for x in false_rows if x["beta_url"] and x["filename"] and x["file_url"] and x["source_url"]][:10]:
            out(f"    row {r['row_idx']}: {r['filename']}")
    out()

    # 5. Anomaly: empty result column
    empty_result = [r for r in rows_data if not r["result"]]
    out(f"--- Anomaly: rows with EMPTY result cell: {len(empty_result)} ---")
    for r in empty_result[:10]:
        out(f"  row {r['row_idx']}: {r['filename']!r}")
    if len(empty_result) > 10:
        out(f"  ... and {len(empty_result) - 10} more")
    out()

    # 6. UUID consistency between Detail URL columns
    uuid_mismatch = []
    for r in rows_data:
        d_match = UUID_RE.search(r["detail_url"]) if r["detail_url"] else None
        b_match = UUID_RE.search(r["beta_url"]) if r["beta_url"] else None
        d_uuid = d_match.group(0).lower() if d_match else None
        b_uuid = b_match.group(0).lower() if b_match else None
        if d_uuid and b_uuid and d_uuid != b_uuid:
            uuid_mismatch.append((r, d_uuid, b_uuid))
        elif (r["detail_url"] and not d_uuid) or (r["beta_url"] and not b_uuid):
            uuid_mismatch.append((r, d_uuid, b_uuid))
    out(f"--- Anomaly: UUID mismatch / unparseable between Detail URL columns: {len(uuid_mismatch)} ---")
    for r, d_uuid, b_uuid in uuid_mismatch[:10]:
        out(f"  row {r['row_idx']}: {r['filename']!r}  detail={d_uuid}  beta={b_uuid}")
    if len(uuid_mismatch) > 10:
        out(f"  ... and {len(uuid_mismatch) - 10} more")
    out()

    # 7. Duplicate filenames
    fname_counts = collections.Counter(r["filename"] for r in rows_data if r["filename"])
    dup_fnames = [(n, c) for n, c in fname_counts.items() if c > 1]
    out(f"--- Anomaly: duplicate Filename values: {len(dup_fnames)} ---")
    for n, c in sorted(dup_fnames, key=lambda x: -x[1])[:10]:
        out(f"  x{c}: {n!r}")
    out()

    # 7b. Duplicate Source URLs (informational — multiple Commons files
    #     can legitimately share a Beeldbank source URL, but a high count
    #     for any single URL is worth a spot-check).
    src_counts = collections.Counter(
        r["source_url"] for r in rows_data
        if r["source_url"] and r["source_url"].startswith(("http://", "https://"))
    )
    dup_sources = [(u, c) for u, c in src_counts.items() if c > 1]
    out(f"--- Duplicate Source URLs (same Beeldbank URL on multiple rows): {len(dup_sources)} ---")
    for u, c in sorted(dup_sources, key=lambda x: -x[1])[:10]:
        out(f"  x{c}: {u}")
    if len(dup_sources) > 10:
        out(f"  ... and {len(dup_sources) - 10} more")
    out()

    # 8. Duplicate UUIDs (same Beta URL pointing from multiple rows)
    beta_counts = collections.Counter(r["beta_url"] for r in rows_data if r["beta_url"])
    dup_betas = [(u, c) for u, c in beta_counts.items() if c > 1]
    out(f"--- Anomaly: duplicate Beta URLs (same Memorix record from multiple rows): {len(dup_betas)} ---")
    for u, c in sorted(dup_betas, key=lambda x: -x[1])[:10]:
        out(f"  x{c}: {u}")
    out()

    # 9. Metadata cells containing ERROR strings
    rows_with_metadata_errors = [r for r in rows_data if r["metadata_with_error"]]
    out(f"--- Anomaly: rows with 'ERROR ...' in a metadata cell: {len(rows_with_metadata_errors)} ---")
    for r in rows_with_metadata_errors[:20]:
        out(f"  row {r['row_idx']}: {r['filename']!r}  fields={r['metadata_with_error']}")
    if len(rows_with_metadata_errors) > 20:
        out(f"  ... and {len(rows_with_metadata_errors) - 20} more")
    out()

    # 10. TRUE rows with no metadata at all (potential missed fixes)
    true_no_metadata = [r for r in rows_data if r["result"] == "TRUE" and r["metadata_filled"] == 0]
    out(f"--- Anomaly: TRUE rows with ZERO metadata fields populated: {len(true_no_metadata)} ---")
    for r in true_no_metadata[:20]:
        out(f"  row {r['row_idx']}: {r['filename']!r}")
    if len(true_no_metadata) > 20:
        out(f"  ... and {len(true_no_metadata) - 20} more")
    out()

    # 10b. TRUE rows missing one or more required metadata fields. These
    #      fields should be present on every Beeldbank record we successfully
    #      resolved; absence usually means add_metadata.py skipped the row
    #      and a backfill never landed.
    REQUIRED_FOR_TRUE = [
        "Titel (dc_title)",
        "Datering (dc_date)",
        "Documenttype (sk_documenttype)",
        "Vervaardiger (sk_vervaardiger)",
        "Collectie (dc_provenance)",
        "Afbeeldingsbestand (identifier)",
    ]
    rows_missing_required = []
    for r in rows_data:
        if r["result"] != "TRUE":
            continue
        missing = [
            h.split(" (")[0]
            for h in REQUIRED_FOR_TRUE
            if h in r["per_field"] and not r["per_field"][h]
        ]
        if missing:
            rows_missing_required.append((r, missing))
    out(
        f"--- Anomaly: TRUE rows missing one or more required metadata "
        f"fields: {len(rows_missing_required)} ---"
    )
    for r, missing in rows_missing_required[:20]:
        out(f"  row {r['row_idx']}: {r['filename']!r}  missing={missing}")
    if len(rows_missing_required) > 20:
        out(f"  ... and {len(rows_missing_required) - 20} more")
    out()

    # 11. Metadata coverage histogram
    cov = collections.Counter(r["metadata_filled"] for r in rows_data)
    total_md = len(METADATA_HEADERS)
    out(f"--- Metadata coverage (out of {total_md} fields) ---")
    for n in range(0, total_md + 1):
        out(f"  {n:2d} fields: {cov.get(n, 0):5d}")
    out()

    # 11b. Per-field empty-cell count — complements the histogram by showing
    #      *which* fields drive the gaps. Sorted from most-empty to most-filled.
    out(f"--- Per-field empty cells (out of {len(rows_data)} data rows) ---")
    per_field_empty = []
    for h in METADATA_HEADERS:
        if h in headers:
            n = sum(1 for r in rows_data if not r["per_field"].get(h))
            per_field_empty.append((h, n))
    for h, n in sorted(per_field_empty, key=lambda kv: -kv[1]):
        out(f"  {h:55s} {n:5d}")
    out()

    # 12. Source URL anomalies
    source_anomalies = collections.Counter()
    sample_anomalies = {}
    for r in rows_data:
        s = r["source_url"]
        if not s:
            source_anomalies["empty"] += 1
        elif s == "PAGE NOT FOUND":
            source_anomalies["PAGE NOT FOUND"] += 1
            sample_anomalies.setdefault("PAGE NOT FOUND", r["row_idx"])
        elif s.startswith("ERROR"):
            source_anomalies["ERROR string"] += 1
            sample_anomalies.setdefault("ERROR string", r["row_idx"])
        elif not (s.startswith("http://") or s.startswith("https://")):
            source_anomalies["non-URL value"] += 1
            sample_anomalies.setdefault("non-URL value", r["row_idx"])
    out("--- Source URL value distribution (anomalies only; URLs not counted) ---")
    for k, v in source_anomalies.most_common():
        sample = f"  (e.g. row {sample_anomalies[k]})" if k in sample_anomalies else ""
        out(f"  {k:20s} {v:5d}{sample}")
    out()

    # 13. File URL (Commons) consistency with Filename
    inconsistent_file_urls = []
    for r in rows_data:
        if not r["filename"] or not r["file_url"]:
            continue
        # The Commons fullurl is the filename with spaces replaced by underscores
        # and the "File:" prefix preserved. Compare suffix only (host can differ).
        try:
            host = urlparse(r["file_url"]).hostname or ""
            if "wikimedia.org" not in host and "wikipedia.org" not in host:
                inconsistent_file_urls.append((r, "non-wikimedia host"))
                continue
        except Exception:
            inconsistent_file_urls.append((r, "unparseable"))
            continue
        # Decode the path tail and compare (loose match — spaces vs underscores).
        path_tail = unquote(r["file_url"].rsplit("/", 1)[-1])
        normalized_filename = r["filename"].replace(" ", "_")
        if normalized_filename != path_tail:
            inconsistent_file_urls.append((r, "filename / file-URL mismatch"))
    out(f"--- Anomaly: File URL (Commons) inconsistent with Filename: {len(inconsistent_file_urls)} ---")
    for r, reason in inconsistent_file_urls[:10]:
        out(f"  row {r['row_idx']}: {reason}: {r['filename']!r} -> {r['file_url']!r}")
    if len(inconsistent_file_urls) > 10:
        out(f"  ... and {len(inconsistent_file_urls) - 10} more")
    out()

    # 14. Whitespace-padded cells
    padded = []
    for row in ws.iter_rows(min_row=2):
        for c in row:
            v = c.value
            if isinstance(v, str) and v != v.strip() and v.strip():
                padded.append((c.row, ws.cell(row=1, column=c.column).value))
                if len(padded) >= 30:
                    break
        if len(padded) >= 30:
            break
    out(f"--- Anomaly: cells with leading/trailing whitespace (first 30): {len(padded)} ---")
    for row_idx, hname in padded[:30]:
        out(f"  row {row_idx} col {hname!r}")
    out()

    # 14b. Cells containing control / non-printable characters. Walks every
    #      string cell once and looks for any char in NONPRINT_RE — ASCII
    #      control chars (except \t \n \r), Unicode line/paragraph
    #      separators, zero-width chars, BOM. Flags up to 30 cells; prints
    #      the codepoint(s) found in each so the offender is identifiable.
    nonprint_hits = []
    for row in ws.iter_rows(min_row=2):
        for c in row:
            v = c.value
            if not isinstance(v, str):
                continue
            chars_found = NONPRINT_RE.findall(v)
            if chars_found:
                # Deduplicate while preserving order so the report stays compact.
                seen = []
                for ch in chars_found:
                    if ch not in seen:
                        seen.append(ch)
                hname = ws.cell(row=1, column=c.column).value
                nonprint_hits.append((c.row, hname, seen))
                if len(nonprint_hits) >= 30:
                    break
        if len(nonprint_hits) >= 30:
            break
    out(
        "--- Anomaly: cells with control / non-printable characters "
        f"(first 30): {len(nonprint_hits)} ---"
    )
    for row_idx, hname, chars in nonprint_hits[:30]:
        labels = ", ".join(char_label(ch) for ch in chars)
        out(f"  row {row_idx} col {hname!r}: {labels}")
    out()

    # 15. Filename column issues (rows with no filename)
    no_filename = [r for r in rows_data if not r["filename"]]
    out(f"--- Anomaly: rows with EMPTY Filename: {len(no_filename)} ---")
    for r in no_filename[:10]:
        out(f"  row {r['row_idx']}")
    out()

    # 16. URL format validation per column. For each URL column we have a
    #     known shape (host + path pattern); cells that don't match are
    #     flagged. Empty cells are skipped.
    UUID_PAT = r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
    url_rules = [
        ("Beta Archief Amsterdam Detail URL", "beta_url",
         "beta.archief.amsterdam",
         re.compile(rf"^/detail/{UUID_PAT}$", re.I)),
        ("Archief Amsterdam Detail URL", "detail_url",
         "archief.amsterdam",
         re.compile(rf"^/beeldbank/detail/{UUID_PAT}$", re.I)),
        ("File URL (Commons)", "file_url",
         "commons.wikimedia.org",
         re.compile(r"^/wiki/File:")),
        ("Inventarissen (dc_source)", "inventarissen",
         "archief.amsterdam",
         re.compile(r"^/archief/\d+(/\S+)?$")),
    ]
    out("--- URL format validation per column ---")
    for header, key, expected_host, path_pattern in url_rules:
        if header not in headers:
            continue
        bad = []
        for r in rows_data:
            v = r[key]
            if not v:
                continue
            try:
                p = urlparse(v)
            except ValueError:
                bad.append((r["row_idx"], v, "unparseable"))
                continue
            host = (p.hostname or "").lower()
            if host != expected_host:
                bad.append((r["row_idx"], v, f"host={host or '(none)'}"))
                continue
            if not path_pattern.match(p.path):
                bad.append((r["row_idx"], v, f"path={p.path!r}"))
        out(f"  {header:42s} bad: {len(bad):3d} (expected host={expected_host})")
        for row_idx, v, reason in bad[:5]:
            short = v if len(v) <= 80 else v[:77] + "..."
            out(f"    row {row_idx}: {reason}  -> {short}")
        if len(bad) > 5:
            out(f"    ... and {len(bad) - 5} more")
    out()

    out("=" * 78)
    out("End of report.")

    text = "\n".join(out_lines) + "\n"
    with open(REPORT_PATH, "w", encoding="utf-8") as f:
        f.write(text)
    print(text)
    print(f"\nReport also written to {REPORT_PATH}")


if __name__ == "__main__":
    main()