#!/usr/bin/env python3 """Reproducible exact-phrase comparison for the Faurisson provenance note. This is deliberately a narrow test. It asks whether the 123 public body units in the 1988-text witness share long, contiguous normalized word sequences with five English-language witnesses to Faurisson texts that predate the report. It does not purport to identify an author, and it does not treat shared ideas, translations, editorial influence, or paraphrase as exact verbal copying. """ from __future__ import annotations import csv import hashlib import json import re import unicodedata from collections import defaultdict from datetime import datetime, timezone from pathlib import Path PROJECT = Path("/Volumes/Holocaust/projects/Leuchter_Line_by_Line") REPORT_UNITS = PROJECT / "data/paragraph_units_1988_witness.json" OUT_DIR = PROJECT / "data/provenance" OUT_JSON = OUT_DIR / "faurisson_exact_overlap.json" OUT_CSV = OUT_DIR / "faurisson_exact_overlap.csv" ARCHIVE = Path("/Volumes/Holocaust/projects/Journal_Archives/codoh") SOURCES = [ { "id": "letters_1978_1979", "label": "Three letters to Le Monde", "original_date": "1978-12-29 to 1979", "english_witness_date": "authorized English translation republished 2013", "path": ARCHIVE / "2013/faurissons-three-letters-to-le-monde-1978-1979/text.txt", "url": "https://codoh.com/library/document/faurissons-three-letters-to-le-monde-1978-1979/", "start_marker": "‘The Problem of the Gas Chambers’", "selection_note": ( "Comparison begins at the first translated letter, excluding the " "modern editorial introduction." ), }, { "id": "mechanics_1980", "label": "The Mechanics of Gassing", "original_date": "Spring 1980", "english_witness_date": "Spring 1980", "path": ARCHIVE / "2012/the-mechanics-of-gassing/text.txt", "url": "https://ihr.org/journal/v01p-23_faurisson", "selection_note": "Article body after the archive metadata header.", }, { "id": "physically_inconceivable_1980", "label": "The Gas Chambers of Auschwitz Appear to be Physically Inconceivable", "original_date": "1980 or earlier", "english_witness_date": "English republication witness archived 2012", "path": ARCHIVE / "2012/the-gas-chambers-of-auschwitz-appear-to-be/text.txt", "url": "https://codoh.com/library/document/the-gas-chambers-of-auschwitz-appear-to-be/", "selection_note": "Article body after the archive metadata header.", }, { "id": "truth_or_lie_1980", "label": "The Gas Chambers: Truth or Lie?", "original_date": "1980 interview", "english_witness_date": "English republication witness archived 2012", "path": ARCHIVE / "2012/the-gas-chambers-truth-or-lie/text.txt", "url": "https://codoh.com/library/document/the-gas-chambers-truth-or-lie/", "answer_blocks_only": True, "selection_note": ( "Only the answer blocks attributed to Faurisson are compared; " "interviewer questions are excluded." ), }, { "id": "paper_historian_1982", "label": "Response to a Paper Historian", "original_date": "1982; expanded December 1982", "english_witness_date": "English publication in 1986", "path": ARCHIVE / "2012/response-to-a-paper-historian/text.txt", "url": "https://www.ihr.org/journal/v07index.html", "selection_note": ( "English article witness after the archive metadata header, " "including its authorial introduction." ), }, ] MIN_REPORTED_WORDS = 4 COPYING_THRESHOLD_WORDS = 8 def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def strip_archive_header(text: str) -> str: divider = re.search(r"^[─—-]{20,}\s*$", text, flags=re.MULTILINE) return text[divider.end() :].lstrip() if divider else text def selected_source_text(source: dict) -> str: text = strip_archive_header(source["path"].read_text(encoding="utf-8")) marker = source.get("start_marker") if marker: position = text.find(marker) if position < 0: raise ValueError(f"Start marker not found for {source['id']}: {marker}") text = text[position:] if source.get("answer_blocks_only"): blocks = re.split(r"(?=^QUESTION\s+\d+:)", text, flags=re.MULTILINE) answers = [] for block in blocks: match = re.search( r"^ANSWER\s+\d+:\s*(.*?)(?=^QUESTION\s+\d+:|\Z)", block, flags=re.MULTILINE | re.DOTALL, ) if match: answers.append(match.group(1).strip()) if not answers: raise ValueError(f"No answer blocks found for {source['id']}") text = "\n\n".join(answers) return text def words(text: str) -> list[str]: normalized = unicodedata.normalize("NFKD", text).lower() normalized = normalized.replace("’", "'").replace("‘", "'") return re.findall(r"[a-z0-9]+(?:'[a-z0-9]+)?", normalized) def paragraphs(text: str) -> list[str]: return [ re.sub(r"\s+", " ", block).strip() for block in re.split(r"\n\s*\n", text) if block.strip() ] def ngram_index(source_paragraphs: list[dict], size: int) -> dict[tuple[str, ...], list[dict]]: index: dict[tuple[str, ...], list[dict]] = defaultdict(list) for paragraph in source_paragraphs: tokens = paragraph["tokens"] for start in range(0, len(tokens) - size + 1): index[tuple(tokens[start : start + size])].append( { "source_id": paragraph["source_id"], "source_label": paragraph["source_label"], "source_paragraph": paragraph["paragraph_number"], "source_token_start": start, } ) return index def maximal_matches(report_units: list[dict], source_paragraphs: list[dict]) -> list[dict]: indexes = { size: ngram_index(source_paragraphs, size) for size in range(MIN_REPORTED_WORDS, COPYING_THRESHOLD_WORDS + 1) } matches = {} for unit in report_units: tokens = words(unit["witness_text"]) for size in range(COPYING_THRESHOLD_WORDS, MIN_REPORTED_WORDS - 1, -1): index = indexes[size] for start in range(0, len(tokens) - size + 1): phrase_tokens = tuple(tokens[start : start + size]) hits = index.get(phrase_tokens, []) for hit in hits: key = (unit["id"], hit["source_id"], phrase_tokens) existing = matches.get(key) if existing and existing["word_count"] >= size: continue matches[key] = { "report_unit": unit["id"], "report_section": unit["section_number"], "report_kind": unit["kind"], "source_id": hit["source_id"], "source_label": hit["source_label"], "source_paragraph": hit["source_paragraph"], "word_count": size, "phrase": " ".join(phrase_tokens), } if any( match["report_unit"] == unit["id"] and match["word_count"] == size for match in matches.values() ): # Preserve the longest match(es) for each report unit. This # prevents shorter substrings of the same phrase swamping the # readable audit while retaining ties across sources. break return sorted( matches.values(), key=lambda item: ( -item["word_count"], item["report_unit"], item["source_id"], item["phrase"], ), ) def main() -> None: OUT_DIR.mkdir(parents=True, exist_ok=True) report_data = json.loads(REPORT_UNITS.read_text(encoding="utf-8")) report_units = [ unit for unit in report_data["units"] if unit.get("witness_status") == "aligned" ] source_paragraphs = [] source_register = [] for source in SOURCES: text = selected_source_text(source) source_register.append( { key: (str(value) if isinstance(value, Path) else value) for key, value in source.items() if key not in {"start_marker", "answer_blocks_only"} } | { "sha256": sha256(source["path"]), "selected_word_count": len(words(text)), "selected_paragraph_count": len(paragraphs(text)), } ) for number, paragraph in enumerate(paragraphs(text), start=1): tokenized = words(paragraph) if tokenized: source_paragraphs.append( { "source_id": source["id"], "source_label": source["label"], "paragraph_number": number, "tokens": tokenized, } ) matches = maximal_matches(report_units, source_paragraphs) threshold_hits = [ match for match in matches if match["word_count"] >= COPYING_THRESHOLD_WORDS ] longest = max((match["word_count"] for match in matches), default=0) result = { "schema_version": 1, "generated_utc": datetime.now(timezone.utc).isoformat(), "question": ( "Do the public body units in the 1988-text witness share long, " "contiguous normalized English word sequences with selected " "pre-report Faurisson witnesses?" ), "method": { "report_selection": "All 123 units with witness_status='aligned'.", "normalization": ( "Unicode NFKD, lowercase, alphanumeric words with internal " "apostrophes; punctuation and spacing ignored." ), "comparison": ( "Contiguous normalized word n-grams, tested from 8 words down " "to 4. The readable output retains the longest match per report " "unit, with source ties." ), "copying_threshold_words": COPYING_THRESHOLD_WORDS, "caution": ( "Absence of an 8-word match is not proof of independent " "authorship. Translation, paraphrase, oral transmission, " "editing, and shared technical vocabulary can erase or create " "short matches. This test cannot identify a ghostwriter." ), }, "report": { "path": str(REPORT_UNITS), "sha256": sha256(REPORT_UNITS), "unit_count": len(report_units), "normalized_word_count": sum( len(words(unit["witness_text"])) for unit in report_units ), }, "sources": source_register, "results": { "source_paragraph_count": len(source_paragraphs), "longest_match_words": longest, "matches_at_or_above_8_words": len(threshold_hits), "interpretation": ( "No sentence-length exact borrowing was found at the stated " "threshold." if not threshold_hits else "One or more threshold matches require manual review." ), "longest_matches": [ match for match in matches if match["word_count"] == longest ], "all_maximal_matches_4_to_8_words": matches, }, } OUT_JSON.write_text( json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) with OUT_CSV.open("w", newline="", encoding="utf-8") as handle: fields = [ "report_unit", "report_section", "report_kind", "source_id", "source_label", "source_paragraph", "word_count", "phrase", ] writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() writer.writerows(matches) print( json.dumps( { "report_units": len(report_units), "source_witnesses": len(source_register), "source_paragraphs": len(source_paragraphs), "longest_match_words": longest, "matches_at_or_above_8_words": len(threshold_hits), "json": str(OUT_JSON), "csv": str(OUT_CSV), }, indent=2, ) ) if __name__ == "__main__": main()