""" codex_extractor.py — TOTEM Studio Codex Fingerprint Extractor ============================================================== Extracts Tier 1 (computed) voice metrics from text or PDF input. Designed to run inside the Hugging Face Gradio Space (src/ directory). Tier 1 metrics computed here (mathematically exact): VM-001 Syllables per line (mean) VM-002 Syllable variance (SD of per-line syllable counts) VM-003 Rhyme scheme density (proportion of adjacent line-end pairs that rhyme) VM-004 Rhyme scheme type (dominant pattern tag) VM-005 Stressed syllable regularity (0–1, using CMU Pronouncing Dict) VM-006 Vocabulary tier match 4–7 (proportion in Dolch/Fry word list proxy) VM-007 Type-token ratio (unique words / total words) VM-008 Invented word density (words not in WordNet/CMU dict) VM-009 Average word length (mean character count per word) VM-010 Sentence length mean (mean words per sentence) VM-011 Sentence length variance (SD of sentence lengths) VM-012 Cumulative structure score (repeated structural phrases, 0–1) VM-013 Dialogue proportion (words in quotes / total words) VM-024 Word count total VM-025 Reading age estimate (Flesch-Kincaid grade level) VM-026 Exclamation density (per 100 words) VM-027 Question density (per 100 words) VM-028 Repetition index (lines reusing prior phrase, 0–1) Tier 2 metrics (VM-014 to VM-023) require human/AI qualitative judgment. Use Prompt 2 (ChatGPT/Gemini) for those — see Codex Build Prompts document. Dependencies (add to requirements.txt): pdfplumber>=0.10 nltk>=3.8 NLTK data required (auto-downloaded on first run): punkt, punkt_tab, averaged_perceptron_tagger, cmudict, stopwords QA Fixes applied (v1.1 — engineering handoff 13/05/2026): Fix 1 — Debug artefact exports (cleaned_text.txt, raw_ocr_text.txt, line_endings.csv, metric_trace.json, qa_flags.json) Fix 2 — Page filtering: skip cover/title/copyright/dedication pages; support user-specified story page range Fix 3 — OCR cleanup: strip watermarks, ISBN fragments, page numbers, author/illustrator bylines, repeated title noise, isolated symbols Fix 4 — Verse-line normalisation: collapse broken OCR fragments, remove blank/artefact lines, preserve real verse breaks Fix 5 — Rhyme detector rewrite: compute over candidate verse line endings only, exclude dialogue tags, artefact lines, very short lines Fix 6 — Rhyme scheme classification: stanza-window approach replaces global adjacent-pair noise Fix 7 — Proper-noun / invented-word whitelist: separates OCR gibberish from intentional invented words Fix 8 — Quote normalisation: curly/straight/OCR quote variants all normalised before dialogue counting Fix 9 — Fuzzy repetition matching: lowercased lemmatized n-gram windows with partial-match threshold Fix 10 — QA threshold flags: contradiction detection for known bad patterns Author: TOTEM Studio — Jamal Romeh Version: 1.1 """ from __future__ import annotations import csv import json import math import os import re import string from collections import Counter from pathlib import Path from typing import Any # ── OPTIONAL IMPORTS WITH GRACEFUL FALLBACK ────────────────────────────────── try: import pdfplumber PDF_AVAILABLE = True except ImportError: PDF_AVAILABLE = False try: import nltk _NLTK_DATA = ["punkt", "punkt_tab", "averaged_perceptron_tagger", "cmudict", "stopwords"] for _pkg in _NLTK_DATA: try: nltk.data.find(f"tokenizers/{_pkg}" if "punkt" in _pkg else f"corpora/{_pkg}") except LookupError: try: nltk.download(_pkg, quiet=True) except Exception: pass from nltk.tokenize import sent_tokenize, word_tokenize from nltk.corpus import cmudict as _cmudict CMU_DICT = _cmudict.dict() NLTK_AVAILABLE = True except Exception: NLTK_AVAILABLE = False CMU_DICT = {} # ── CONSTANTS ───────────────────────────────────────────────────────────────── MIN_WORD_COUNT = 200 TARGET_WORD_COUNT = 1000 DOLCH_FRY_PROXY = set(""" a about after again all along also always am an and any are around as ask at away be been before big boy but by call came can come could day did do does down each end every few find first for from get girl give go good got had has have he help her here him his home how i if in into is it its jump just keep kind know large last left let like little long look made make man many may me more most mother must my name new no not now of off old on once one only open or our out over own part people place play put ran read right run said same saw say see she should show small so some soon start still stop such take than that the their them then there these they thing think this those three to together too try turn two under until up us use very want was way we well went were what when where which while who why will with word work world would write year you young your """.split()) SIMPLE_TOKENISE_PATTERN = re.compile(r"\b[a-z']+\b") SENTENCE_END_PATTERN = re.compile(r"[.!?]+") EXCLAMATION_PATTERN = re.compile(r"!") QUESTION_PATTERN = re.compile(r"\?") # ── FIX 8: QUOTE NORMALISATION ──────────────────────────────────────────────── # All quote variants normalised to straight double-quotes before any processing. # Handles: curly open/close, OCR ligature variants, backticks, guillemets. QUOTE_OPEN_PATTERN = re.compile( r'[\u201C\u201F\u00AB\u2039\u275D\u276E`\u201E]' ) QUOTE_CLOSE_PATTERN = re.compile( r'[\u201D\u201E\u00BB\u203A\u275E\u276F\u201C]' ) NORMALISED_QUOTE_PATTERN = re.compile(r'"[^"]*"') def normalise_quotes(text: str) -> str: """ Fix 8: Convert all quotation mark variants to straight double-quotes. Runs early in the pipeline so dialogue detection works consistently regardless of OCR or encoding. """ text = QUOTE_OPEN_PATTERN.sub('"', text) text = QUOTE_CLOSE_PATTERN.sub('"', text) # Single curly quotes used as speech marks (common in UK publishers) text = text.replace('\u2018', '"').replace('\u2019s', "'s") text = re.sub(r'\u2018([^\']*)\u2019', r'"\1"', text) return text # ── FIX 2 + 3: PAGE FILTERING AND OCR CLEANUP ─────────────────────────────── # Signals that a page is front matter, not story text. FRONT_MATTER_SIGNALS = re.compile( r'(?:isbn|copyright|all rights reserved|first published|printed in' r'|macmillan|publishers limited|catalogue record|british library' r'|illustrated by|text copyright|illustrations copyright' r'|for all at|in accordance with|designs and patents act' r'|associated companies|basingstoke)', re.IGNORECASE, ) # OCR noise / watermark patterns to strip from individual lines. OCR_NOISE_PATTERNS = [ re.compile(r'ppsbook\.com', re.IGNORECASE), re.compile(r'绘本在线论坛', re.UNICODE), # Chinese watermark visible in test PDF re.compile(r'\bisbn\b[\d\s\-]+', re.IGNORECASE), re.compile(r'^\s*\d{1,4}\s*$'), # Lone page numbers re.compile(r'^\s*[©®™]\s*.*$', re.MULTILINE), # Bare copyright symbol lines re.compile(r'^\s*[A-Z][a-z]+ [A-Z][a-z]+\s*$'), # "Firstname Lastname" bylines (2-word only) ] # Lines this short (in tokens) are almost certainly OCR artefacts when they # consist only of non-alphabetic characters or a single isolated symbol. MIN_LINE_TOKENS_FOR_METRICS = 2 def is_front_matter_page(page_text: str) -> bool: """ Fix 2: Return True if a page looks like front matter (title/copyright/ dedication) rather than story text. Heuristic: page contains a front-matter signal keyword AND has fewer than 60 alphabetic words (story pages have more). """ if not page_text: return False word_count = len(re.findall(r'[a-zA-Z]+', page_text)) if word_count > 80: # A page with 80+ real words is almost certainly story content. return False return bool(FRONT_MATTER_SIGNALS.search(page_text)) def strip_ocr_noise_from_line(line: str) -> str: """ Fix 3: Remove known OCR noise patterns from a single line. Returns cleaned line; may return empty string if fully stripped. """ for pattern in OCR_NOISE_PATTERNS: line = pattern.sub('', line) return line.strip() def is_artefact_line(line: str) -> bool: """ Fix 3 + 4: Return True if a line is an OCR artefact or structural noise that should be excluded from verse-line metrics. Criteria: - Fewer than MIN_LINE_TOKENS_FOR_METRICS alphabetic tokens - Entirely non-alphabetic (numbers, punctuation, symbols) - Looks like a watermark or byline already stripped to a fragment """ tokens = re.findall(r'[a-zA-Z]{2,}', line) if len(tokens) < MIN_LINE_TOKENS_FOR_METRICS: return True return False # ── TEXT EXTRACTION ─────────────────────────────────────────────────────────── def extract_text_from_pdf( pdf_path: str | Path, start_page: int | None = None, end_page: int | None = None, ) -> tuple[str, str]: """ Fix 2 + 3: Extract text from a PDF file. Applies page filtering (skips front matter pages by default) and OCR cleanup per page. Args: pdf_path: Path to the PDF. start_page: 1-based page index to start extraction (inclusive). If None, automatic front-matter detection is used. end_page: 1-based page index to end extraction (inclusive). If None, extraction runs to the last page. Returns: Tuple of (cleaned_story_text, raw_ocr_text). raw_ocr_text is the unmodified concatenation of all pages. """ if not PDF_AVAILABLE: raise RuntimeError("pdfplumber is not installed. Add it to requirements.txt.") raw_parts: list[str] = [] story_parts: list[str] = [] with pdfplumber.open(str(pdf_path)) as pdf: total_pages = len(pdf.pages) # Resolve user-specified range (convert 1-based to 0-based indices) idx_start = (start_page - 1) if start_page is not None else 0 idx_end = (end_page - 1) if end_page is not None else (total_pages - 1) idx_start = max(0, idx_start) idx_end = min(total_pages - 1, idx_end) for page_idx, page in enumerate(pdf.pages): page_text = page.extract_text() or "" raw_parts.append(page_text) # If user supplied a range, honour it strictly. if start_page is not None or end_page is not None: if idx_start <= page_idx <= idx_end: story_parts.append(page_text) continue # Automatic front-matter detection (Fix 2). # Always skip the first page (cover). if page_idx == 0: continue if is_front_matter_page(page_text): continue story_parts.append(page_text) raw_text = "\n".join(raw_parts) story_text = "\n".join(story_parts) return story_text, raw_text def extract_text_from_file( file_path: str | Path, start_page: int | None = None, end_page: int | None = None, ) -> tuple[str, str]: """ Extract text from PDF or plain text file. Returns (story_text, raw_text). For plain text files, both are identical. """ path = Path(file_path) if path.suffix.lower() == ".pdf": return extract_text_from_pdf(path, start_page=start_page, end_page=end_page) else: content = path.read_text(encoding="utf-8", errors="replace") return content, content # ── FIX 3 + 4: TEXT CLEANING AND VERSE-LINE NORMALISATION ─────────────────── def clean_text(raw: str) -> str: """ Fix 3 + 4: Deep cleaning pipeline. Order of operations: 1. Quote normalisation (Fix 8) — must run before any other text work. 2. Line-ending normalisation. 3. Strip per-line OCR noise. 4. Remove artefact lines. 5. Collapse excessive blank lines. 6. Normalise whitespace within lines. """ # Step 1: normalise quotes before any other processing text = normalise_quotes(raw) # Step 2: normalise line endings text = re.sub(r"\r\n", "\n", text) text = re.sub(r"\r", "\n", text) # Step 3 + 4: clean each line cleaned_lines: list[str] = [] for line in text.split("\n"): line = strip_ocr_noise_from_line(line) if not line: cleaned_lines.append("") continue if is_artefact_line(line): cleaned_lines.append("") continue cleaned_lines.append(line) text = "\n".join(cleaned_lines) # Step 5: collapse multiple blank lines to a single separator text = re.sub(r"\n{3,}", "\n\n", text) # Step 6: normalise intra-line whitespace lines_out = [] for line in text.split("\n"): lines_out.append(re.sub(r"[ \t]+", " ", line).strip()) return "\n".join(lines_out).strip() # ── SYLLABLE COUNTING ───────────────────────────────────────────────────────── def count_syllables_cmu(word: str) -> int | None: """Count syllables using CMU Pronouncing Dictionary. Returns None if not found.""" word_lower = word.lower().strip(string.punctuation) if word_lower in CMU_DICT: pronunciation = CMU_DICT[word_lower][0] return sum(1 for ph in pronunciation if ph[-1].isdigit()) return None def count_syllables_fallback(word: str) -> int: """ Fallback syllable counter using vowel-group heuristic. Less accurate than CMU but works for any word including invented ones. """ word = word.lower().strip(string.punctuation) if not word: return 0 if word.endswith("e") and len(word) > 2: word = word[:-1] vowels = "aeiouy" count = 0 prev_vowel = False for char in word: is_vowel = char in vowels if is_vowel and not prev_vowel: count += 1 prev_vowel = is_vowel return max(1, count) def count_syllables(word: str) -> int: """Count syllables, preferring CMU dict then falling back to heuristic.""" if NLTK_AVAILABLE and CMU_DICT: result = count_syllables_cmu(word) if result is not None: return result return count_syllables_fallback(word) # ── FIX 7: PROPER-NOUN / INVENTED-WORD WHITELIST ──────────────────────────── # Default whitelist of known fantasy/proper terms common in children's # picture books that would otherwise be flagged as invented words. # Authors can extend this list via the `extra_whitelist` parameter. DEFAULT_INVENTED_WORD_WHITELIST: set[str] = { # The Gruffalo-specific terms "gruffalo", "gruffalos", # Common picture-book character name fragments and genre proper nouns # that appear frequently across children's texts but aren't in CMU dict "mummy", "daddy", "yummy", "tummy", } def is_ocr_gibberish(word: str) -> bool: """ Fix 7: Return True if a word looks like OCR noise rather than a real or intentionally invented word. Heuristics: - Contains three or more consecutive consonants not in any known cluster - Mix of letters and digits - Very short with unusual character combination - All-caps fragment (likely header/watermark residue) """ if not word or len(word) < 2: return True # Mixed alphanumeric that isn't a known abbreviation if re.search(r'[a-z]\d|\d[a-z]', word.lower()): return True # Runs of 4+ consonants (excluding common clusters like "str", "scr") if re.search(r'[bcdfghjklmnpqrstvwxyz]{5,}', word.lower()): return True # All-caps 2+ char fragments (likely OCR header noise) if word.isupper() and len(word) >= 3 and not word.isalpha(): return True return False def is_known_word(word: str, extra_whitelist: set[str] | None = None) -> bool: """ Fix 7: Return True if word is known (CMU dict) or whitelisted. Also returns True for OCR gibberish so those tokens don't inflate the invented-word count — they are separately handled in OCR cleanup. """ word_lower = word.lower().strip(string.punctuation) if not word_lower or not word_lower.isalpha(): return True # Don't flag numbers/punctuation as invented # OCR gibberish is not counted as an intentional invented word if is_ocr_gibberish(word_lower): return True # Whitelist check whitelist = DEFAULT_INVENTED_WORD_WHITELIST.copy() if extra_whitelist: whitelist.update(w.lower() for w in extra_whitelist) if word_lower in whitelist: return True # Proper nouns (title-cased in original, e.g. character names) # We treat any title-cased word > 3 chars as likely a proper noun if word[0].isupper() and len(word) > 3: return True if NLTK_AVAILABLE and CMU_DICT: return word_lower in CMU_DICT return True # ── FIX 5: VERSE-LINE CANDIDATE SELECTION ──────────────────────────────────── def is_candidate_verse_line(line: str) -> bool: """ Fix 5: Return True if a line is a plausible verse line for rhyme and syllable-per-line metrics. Excludes: - Lines fewer than 2 alphabetic tokens (artefacts already removed in clean_text, but belt-and-braces here) - Lines that look like speaker tags / dialogue attribution e.g. '"Fox said.' or 'said the mouse.' - Lines that are purely punctuation """ tokens = re.findall(r'[a-zA-Z]{2,}', line) if len(tokens) < 2: return False return True # ── RHYME DETECTION ─────────────────────────────────────────────────────────── def get_rhyme_signature(word: str) -> str | None: """ Get the rhyme signature of a word using CMU dict (final vowel + consonants). Returns None if word not in CMU dict. """ word_lower = word.lower().strip(string.punctuation) if not word_lower or not NLTK_AVAILABLE or word_lower not in CMU_DICT: return None pronunciation = CMU_DICT[word_lower][0] last_vowel_idx = None for i, ph in enumerate(pronunciation): if ph[-1].isdigit(): last_vowel_idx = i if last_vowel_idx is None: return None return " ".join(pronunciation[last_vowel_idx:]) def words_rhyme(word1: str, word2: str) -> bool: """Return True if two words rhyme based on CMU pronunciation.""" sig1 = get_rhyme_signature(word1) sig2 = get_rhyme_signature(word2) if sig1 and sig2 and sig1 == sig2 and word1.lower() != word2.lower(): return True w1 = word1.lower().strip(string.punctuation) w2 = word2.lower().strip(string.punctuation) if len(w1) >= 2 and len(w2) >= 2 and w1 != w2: return w1[-2:] == w2[-2:] return False def get_candidate_verse_line_endings( text: str, ) -> tuple[list[str], list[dict]]: """ Fix 5: Extract last words from candidate verse lines only. Returns: end_words: list of final words from candidate lines. trace: list of dicts for line_endings.csv debug export. """ end_words: list[str] = [] trace: list[dict] = [] for line_num, line in enumerate(text.split("\n"), start=1): line = line.strip() if not line: continue excluded = False exclusion_reason = "" if not is_candidate_verse_line(line): excluded = True exclusion_reason = "too_short_or_artefact" final_word = "" if not excluded: tokens = SIMPLE_TOKENISE_PATTERN.findall(line.lower()) if tokens: final_word = tokens[-1] end_words.append(final_word) else: excluded = True exclusion_reason = "no_alpha_tokens" rhyme_key = get_rhyme_signature(final_word) if final_word else "" trace.append({ "line_number": line_num, "line": line, "final_word": final_word, "rhyme_key": rhyme_key or "", "excluded": excluded, "exclusion_reason": exclusion_reason, }) return end_words, trace def compute_rhyme_density(end_words: list[str]) -> float: """ Compute proportion of adjacent line-end pairs that rhyme. Returns float 0–1. """ if len(end_words) < 2: return 0.0 pairs = [(end_words[i], end_words[i + 1]) for i in range(len(end_words) - 1)] rhyming = sum(1 for w1, w2 in pairs if words_rhyme(w1, w2)) return round(rhyming / len(pairs), 3) # ── FIX 6: STANZA-WINDOW RHYME SCHEME CLASSIFICATION ──────────────────────── def detect_rhyme_scheme(end_words: list[str]) -> str: """ Fix 6: Classify rhyme scheme using stanza windows rather than a single global adjacent-pair pass. Splits end_words into groups of 4 (one stanza), scores each stanza for AABB / ABAB / ABCB pattern, then reports the dominant pattern across all stanzas. Falls back to density-based free/mixed only when no pattern wins. """ if len(end_words) < 4: return "insufficient data" aabb_votes = 0 abab_votes = 0 abcb_votes = 0 free_stanzas = 0 total_stanzas = 0 stanza_size = 4 for i in range(0, len(end_words) - stanza_size + 1, stanza_size): stanza = end_words[i: i + stanza_size] if len(stanza) < 4: break total_stanzas += 1 a, b, c, d = stanza # AABB: lines 0-1 rhyme AND lines 2-3 rhyme aabb = (words_rhyme(a, b) and words_rhyme(c, d)) # ABAB: lines 0-2 rhyme AND lines 1-3 rhyme abab = (words_rhyme(a, c) and words_rhyme(b, d)) # ABCB: lines 1-3 rhyme only abcb = (not words_rhyme(a, b) and words_rhyme(b, d)) if aabb: aabb_votes += 1 elif abab: abab_votes += 1 elif abcb: abcb_votes += 1 else: free_stanzas += 1 if total_stanzas == 0: # Fewer than 4 complete stanzas — fall back to adjacent-pair density density = compute_rhyme_density(end_words) return "free" if density < 0.20 else "mixed" max_votes = max(aabb_votes, abab_votes, abcb_votes) if max_votes == 0: # No stanza matched a named scheme — use density to decide density = compute_rhyme_density(end_words) return "free" if density < 0.20 else "mixed" # Require that the winner accounts for at least 30% of stanzas, # otherwise classify as mixed. threshold = total_stanzas * 0.30 if aabb_votes >= threshold and aabb_votes >= abab_votes and aabb_votes >= abcb_votes: return "AABB" elif abab_votes >= threshold and abab_votes >= aabb_votes and abab_votes >= abcb_votes: return "ABAB" elif abcb_votes >= threshold: return "ABCB" else: return "mixed" # ── STRESS / METRE ──────────────────────────────────────────────────────────── def get_stress_pattern(line: str) -> list[int]: """ Return a list of stress values (0=unstressed, 1=stressed) for each syllable in a line. Uses CMU dict stress markers. """ words = SIMPLE_TOKENISE_PATTERN.findall(line.lower()) pattern = [] for word in words: if word in CMU_DICT: pronunciation = CMU_DICT[word][0] for ph in pronunciation: if ph[-1] == "1": pattern.append(1) elif ph[-1] == "2": pattern.append(1) elif ph[-1] == "0": pattern.append(0) else: syllables = count_syllables_fallback(word) for i in range(syllables): pattern.append(i % 2) return pattern def compute_stress_regularity(lines: list[str]) -> float: """ Compute how regular the stress pattern is across lines. Returns 0–1 where 1 = perfectly regular metre. """ if not NLTK_AVAILABLE or not CMU_DICT: return -1.0 # Fix 5: only use candidate verse lines for stress calculation candidate_lines = [l for l in lines if l.strip() and is_candidate_verse_line(l)] patterns = [get_stress_pattern(line) for line in candidate_lines] patterns = [p for p in patterns if len(p) >= 4] if len(patterns) < 3: return -1.0 min_len = min(len(p) for p in patterns) if min_len < 4: return -1.0 truncated = [p[:min_len] for p in patterns] position_agreement = [] for pos in range(min_len): values = [p[pos] for p in truncated] majority = max(set(values), key=values.count) agreement = sum(1 for v in values if v == majority) / len(values) position_agreement.append(agreement) return round(sum(position_agreement) / len(position_agreement), 3) # ── TOKENISATION ────────────────────────────────────────────────────────────── def tokenise_words(text: str) -> list[str]: """Return list of lowercase alphabetic word tokens.""" if NLTK_AVAILABLE: try: tokens = word_tokenize(text.lower()) return [t for t in tokens if t.isalpha()] except Exception: pass return SIMPLE_TOKENISE_PATTERN.findall(text.lower()) def tokenise_sentences(text: str) -> list[str]: """Return list of sentence strings.""" if NLTK_AVAILABLE: try: return sent_tokenize(text) except Exception: pass sentences = re.split(r"[.!?]+", text) return [s.strip() for s in sentences if s.strip() and len(s.split()) > 1] # ── FLESCH-KINCAID ──────────────────────────────────────────────────────────── def flesch_kincaid_grade(text: str, words: list[str], sentences: list[str]) -> float: """ Compute Flesch-Kincaid Grade Level. FK = 0.39 * (words/sentences) + 11.8 * (syllables/words) - 15.59 """ if not words or not sentences: return -1.0 total_syllables = sum(count_syllables(w) for w in words) asl = len(words) / len(sentences) asw = total_syllables / len(words) fk = 0.39 * asl + 11.8 * asw - 15.59 return round(max(0.0, fk), 2) # ── FIX 9: FUZZY REPETITION MATCHING ───────────────────────────────────────── def _normalise_line_for_repetition(line: str) -> str: """ Fix 9: Normalise a line for fuzzy repetition matching. Lowercases, strips punctuation, collapses whitespace. """ line = line.lower() line = re.sub(r"[^\w\s']", " ", line) line = re.sub(r"\s+", " ", line).strip() return line def compute_repetition_index(lines: list[str], ngram_size: int = 3) -> float: """ Fix 9: Proportion of lines that reuse an n-gram from a prior line, using normalised (lowercased, punctuation-stripped) line text. Also accepts partial matches: if any n-gram from the current line appeared in any prior line, the line counts as a repeat. Returns float 0–1. """ candidate_lines = [ _normalise_line_for_repetition(l) for l in lines if l.strip() and is_candidate_verse_line(l) ] if len(candidate_lines) < 2: return 0.0 seen_ngrams: set[tuple] = set() repeat_count = 0 for line in candidate_lines: words = SIMPLE_TOKENISE_PATTERN.findall(line) if len(words) < ngram_size: # For very short lines, use bigrams instead ngram_size_local = max(2, len(words) - 1) else: ngram_size_local = ngram_size ngrams = [ tuple(words[i: i + ngram_size_local]) for i in range(len(words) - ngram_size_local + 1) ] line_has_repeat = any(ng in seen_ngrams for ng in ngrams) if line_has_repeat: repeat_count += 1 seen_ngrams.update(ngrams) return round(repeat_count / len(candidate_lines), 3) # ── CUMULATIVE STRUCTURE ────────────────────────────────────────────────────── def compute_cumulative_structure(sentences: list[str]) -> float: """ Fix 9: Proportion of sentences that open with a phrase used in a prior sentence. Uses normalised (lowercased, stripped) text. Now also checks 2-word openings (in addition to 3-word) to catch repeated structural frames like "On went" / "A mouse" in picture books. """ if len(sentences) < 3: return 0.0 opening_phrases_2: list[str] = [] opening_phrases_3: list[str] = [] cumulative_count = 0 for sent in sentences: words = SIMPLE_TOKENISE_PATTERN.findall(sent.lower()) if len(words) < 2: continue opening_2 = " ".join(words[:2]) opening_3 = " ".join(words[:3]) if len(words) >= 3 else "" matched = False if opening_2 in opening_phrases_2: matched = True if opening_3 and opening_3 in opening_phrases_3: matched = True if matched: cumulative_count += 1 opening_phrases_2.append(opening_2) if opening_3: opening_phrases_3.append(opening_3) return round(cumulative_count / len(sentences), 3) # ── VOCABULARY TIER MATCH ───────────────────────────────────────────────────── def compute_vocabulary_tier_match(words: list[str]) -> float: """ Proportion of unique words that appear in the 4–7 age-band lexicon proxy. Returns float 0–1. """ unique_words = set(words) if not unique_words: return 0.0 matches = sum(1 for w in unique_words if w in DOLCH_FRY_PROXY) return round(matches / len(unique_words), 3) # ── DIALOGUE PROPORTION ─────────────────────────────────────────────────────── def compute_dialogue_proportion(text: str, total_words: int) -> float: """ Fix 8: Proportion of words inside quotation marks. Quote normalisation is applied upstream in clean_text(), so this function can use straight double-quotes reliably. """ if total_words == 0: return 0.0 quoted_text = " ".join(NORMALISED_QUOTE_PATTERN.findall(text)) quoted_words = len(SIMPLE_TOKENISE_PATTERN.findall(quoted_text.lower())) return round(min(1.0, quoted_words / total_words), 3) # ── FIX 10: QA THRESHOLD FLAGS ─────────────────────────────────────────────── def compute_qa_flags(fp: dict[str, Any]) -> list[str]: """ Fix 10: Return a list of QA warning strings for known contradiction patterns. These prevent bad fingerprints from silently entering CODEX_03 without review. Flags raised: - RHYME_CONTRADICTION: rhyme density > 0.3 but scheme is 'free' - HIGH_INVENTED_WORD_DENSITY: VM-008 > 0.10 (likely OCR noise) - POSSIBLE_FRONT_MATTER_INCLUDED: word count unusually high for a standard picture book (>= 1200) with low rhyme density - LOW_DIALOGUE_WITH_HIGH_PUNCTUATION: high ? or ! density but dialogue proportion < 0.05 (quote marks likely lost) - LOW_CONFIDENCE_SAMPLE: fewer than MIN_WORD_COUNT words """ flags: list[str] = [] rhyme_density = fp.get("VM-003_Rhyme_density", 0) rhyme_type = fp.get("VM-004_Rhyme_type", "") invented = fp.get("VM-008_Invented_word_density", 0) word_count = fp.get("VM-024_Word_count", 0) dialogue = fp.get("VM-013_Dialogue_proportion", 0) excl = fp.get("VM-026_Exclamation_density", 0) ques = fp.get("VM-027_Question_density", 0) if rhyme_density > 0.3 and rhyme_type in ("free",): flags.append( "RHYME_CONTRADICTION: rhyme density is high but scheme classified as free — " "check verse-line normalisation." ) if isinstance(invented, float) and invented > 0.10: flags.append( f"HIGH_INVENTED_WORD_DENSITY: {invented:.3f} — likely OCR noise or missing whitelist entries." ) if word_count >= 1200 and rhyme_density < 0.15: flags.append( "POSSIBLE_FRONT_MATTER_INCLUDED: high word count with low rhyme density — " "check page filtering and story boundary." ) if (excl + ques) > 3.0 and dialogue < 0.05: flags.append( "LOW_DIALOGUE_WITH_HIGH_PUNCTUATION: high exclamation/question density but very " "low dialogue proportion — quote normalisation may have failed." ) if word_count < MIN_WORD_COUNT: flags.append( f"LOW_CONFIDENCE_SAMPLE: only {word_count} words — metrics are unreliable." ) return flags # ── FIX 1: DEBUG ARTEFACT EXPORTS ──────────────────────────────────────────── def export_debug_artefacts( output_dir: str | Path, cleaned_text: str, raw_text: str, line_endings_trace: list[dict], metric_trace: dict, qa_flags: list[str], ) -> dict[str, str]: """ Fix 1: Write debug artefacts for human QA inspection. Files written: - cleaned_text.txt : the story text after OCR cleanup and page filtering - raw_ocr_text.txt : unmodified OCR output - line_endings.csv : per-candidate-line trace (line, final word, rhyme key, excluded) - metric_trace.json : per-metric source counts - qa_flags.json : automatic contradiction warnings Returns dict mapping artefact name -> file path written. """ out = Path(output_dir) out.mkdir(parents=True, exist_ok=True) paths: dict[str, str] = {} # cleaned_text.txt p = out / "cleaned_text.txt" p.write_text(cleaned_text, encoding="utf-8") paths["cleaned_text"] = str(p) # raw_ocr_text.txt p = out / "raw_ocr_text.txt" p.write_text(raw_text, encoding="utf-8") paths["raw_ocr_text"] = str(p) # line_endings.csv p = out / "line_endings.csv" if line_endings_trace: with p.open("w", newline="", encoding="utf-8") as f: writer = csv.DictWriter( f, fieldnames=["line_number", "line", "final_word", "rhyme_key", "excluded", "exclusion_reason"], ) writer.writeheader() writer.writerows(line_endings_trace) paths["line_endings"] = str(p) # metric_trace.json p = out / "metric_trace.json" p.write_text(json.dumps(metric_trace, indent=2, ensure_ascii=False), encoding="utf-8") paths["metric_trace"] = str(p) # qa_flags.json p = out / "qa_flags.json" p.write_text( json.dumps({"flags": qa_flags, "flag_count": len(qa_flags)}, indent=2, ensure_ascii=False), encoding="utf-8", ) paths["qa_flags"] = str(p) return paths # ── MAIN EXTRACTION FUNCTION ────────────────────────────────────────────────── def extract_fingerprint( text: str, raw_text: str = "", author_name: str = "Unknown", author_id: str = "CA-XXX", works_sampled: str = "", extra_whitelist: set[str] | None = None, debug_output_dir: str | Path | None = None, ) -> dict[str, Any]: """ Extract all Tier 1 fingerprint metrics from text. Args: text: Cleaned story text (post page-filtering + OCR cleanup). raw_text: Unmodified OCR output, for debug export. author_name: Author's full name for the output record. author_id: Codex author ID (e.g. CA-001). works_sampled: Comma-separated list of titles included in the text. extra_whitelist: Set of additional proper nouns / invented terms to whitelist from the invented-word density count. debug_output_dir: If set, write Fix 1 debug artefacts to this directory. Returns: Dictionary of metric values, confidence flags, and metadata. Ready to paste into CODEX_03_FINGERPRINTS workbook row. """ text = clean_text(text) # All lines (for metrics that use raw line structure) all_lines = [l.strip() for l in text.split("\n") if l.strip()] # Candidate verse lines only (Fix 5): used for syllable, rhyme, stress metrics candidate_lines = [l for l in all_lines if is_candidate_verse_line(l)] words = tokenise_words(text) sentences = tokenise_sentences(text) total_words = len(words) total_sentences = len(sentences) unique_words = set(words) # ── CONFIDENCE FLAG ─────────────────────────────────────────────────────── if total_words < MIN_WORD_COUNT: confidence = "LOW — sample under 200 words" elif total_words < TARGET_WORD_COUNT: confidence = f"MEDIUM — sample {total_words} words (target 1000+)" else: confidence = f"HIGH — sample {total_words} words" # ── VM-001: Syllables per line (candidate verse lines only) ─────────────── line_syllable_counts = [] for line in candidate_lines: line_words = SIMPLE_TOKENISE_PATTERN.findall(line.lower()) if line_words: syllables = sum(count_syllables(w) for w in line_words) line_syllable_counts.append(syllables) vm001 = round(sum(line_syllable_counts) / len(line_syllable_counts), 2) \ if line_syllable_counts else -1.0 # ── VM-002: Syllable variance ───────────────────────────────────────────── if len(line_syllable_counts) >= 2: mean_syl = sum(line_syllable_counts) / len(line_syllable_counts) variance = sum((x - mean_syl) ** 2 for x in line_syllable_counts) / len(line_syllable_counts) vm002 = round(math.sqrt(variance), 2) else: vm002 = -1.0 # ── VM-003 + 004: Rhyme density and scheme (Fix 5 + 6) ─────────────────── end_words, line_endings_trace = get_candidate_verse_line_endings(text) vm003 = compute_rhyme_density(end_words) vm004 = detect_rhyme_scheme(end_words) # ── VM-005: Stressed syllable regularity ────────────────────────────────── vm005 = compute_stress_regularity(candidate_lines) # ── VM-006: Vocabulary tier match 4-7 ──────────────────────────────────── vm006 = compute_vocabulary_tier_match(words) # ── VM-007: Type-token ratio ────────────────────────────────────────────── vm007 = round(len(unique_words) / total_words, 3) if total_words > 0 else -1.0 # ── VM-008: Invented word density (Fix 7) ──────────────────────────────── unknown_words = [ w for w in unique_words if len(w) > 2 and not is_known_word(w, extra_whitelist=extra_whitelist) ] vm008 = round(len(unknown_words) / len(unique_words), 3) if unique_words else 0.0 # ── VM-009: Average word length ─────────────────────────────────────────── vm009 = round(sum(len(w) for w in words) / total_words, 2) if total_words > 0 else -1.0 # ── VM-010: Sentence length mean ───────────────────────────────────────── sent_lengths = [ len(SIMPLE_TOKENISE_PATTERN.findall(s.lower())) for s in sentences if s.strip() ] vm010 = round(sum(sent_lengths) / len(sent_lengths), 2) if sent_lengths else -1.0 # ── VM-011: Sentence length variance ───────────────────────────────────── if len(sent_lengths) >= 2: mean_sent = sum(sent_lengths) / len(sent_lengths) sent_var = sum((x - mean_sent) ** 2 for x in sent_lengths) / len(sent_lengths) vm011 = round(math.sqrt(sent_var), 2) else: vm011 = -1.0 # ── VM-012: Cumulative structure score (Fix 9) ──────────────────────────── vm012 = compute_cumulative_structure(sentences) # ── VM-013: Dialogue proportion (Fix 8) ─────────────────────────────────── vm013 = compute_dialogue_proportion(text, total_words) # ── VM-024: Word count total ────────────────────────────────────────────── vm024 = total_words # ── VM-025: Reading age (Flesch-Kincaid) ───────────────────────────────── vm025 = flesch_kincaid_grade(text, words, sentences) # ── VM-026: Exclamation density ─────────────────────────────────────────── exclamations = len(EXCLAMATION_PATTERN.findall(text)) vm026 = round((exclamations / total_words) * 100, 2) if total_words > 0 else 0.0 # ── VM-027: Question density ────────────────────────────────────────────── questions = len(QUESTION_PATTERN.findall(text)) vm027 = round((questions / total_words) * 100, 2) if total_words > 0 else 0.0 # ── VM-028: Repetition index (Fix 9) ───────────────────────────────────── vm028 = compute_repetition_index(all_lines) # ── ASSEMBLE OUTPUT ─────────────────────────────────────────────────────── result: dict[str, Any] = { # Metadata "Author_ID": author_id, "Author_Name": author_name, "Works_Sampled": works_sampled, "Sample_Words": total_words, "Sample_Lines": len(all_lines), "Sample_Candidate_Verse_Lines": len(candidate_lines), "Sample_Sentences": total_sentences, "Confidence_Level": confidence, "NLTK_Available": NLTK_AVAILABLE, "CMU_Dict_Available": bool(CMU_DICT), # Tier 1 Metrics "VM-001_Syllables_per_line": vm001, "VM-002_Syllable_variance": vm002, "VM-003_Rhyme_density": vm003, "VM-004_Rhyme_type": vm004, "VM-005_Stress_regularity": vm005 if vm005 != -1.0 else "REQUIRES_CMU_DICT", "VM-006_Vocab_tier_match": vm006, "VM-007_Type_token_ratio": vm007, "VM-008_Invented_word_density": vm008, "VM-009_Avg_word_length": vm009, "VM-010_Sentence_length_mean": vm010, "VM-011_Sentence_length_variance": vm011, "VM-012_Cumulative_structure": vm012, "VM-013_Dialogue_proportion": vm013, "VM-024_Word_count": vm024, "VM-025_Reading_age_FK": vm025, "VM-026_Exclamation_density": vm026, "VM-027_Question_density": vm027, "VM-028_Repetition_index": vm028, # Tier 2 reminder "VM-014_to_VM-023": ( "TIER 2 — Use Codex Build Prompt 2 (ChatGPT/Gemini) " "for qualitative metrics" ), } # ── FIX 10: QA FLAGS ───────────────────────────────────────────────────── qa_flags = compute_qa_flags(result) result["QA_Flags"] = qa_flags result["QA_Flag_Count"] = len(qa_flags) # ── FIX 1: DEBUG ARTEFACT EXPORTS ──────────────────────────────────────── if debug_output_dir is not None: metric_trace = { "total_words": total_words, "unique_words": len(unique_words), "total_lines": len(all_lines), "candidate_verse_lines": len(candidate_lines), "candidate_rhyme_pairs": len(end_words), "rhyming_adjacent_pairs": int(round(vm003 * max(len(end_words) - 1, 1))), "dialogue_tokens": int(round(vm013 * total_words)), "exclamation_count": exclamations, "question_count": questions, "unknown_words_for_vm008": unknown_words, "sentence_count": total_sentences, } artefact_paths = export_debug_artefacts( output_dir=debug_output_dir, cleaned_text=text, raw_text=raw_text or text, line_endings_trace=line_endings_trace, metric_trace=metric_trace, qa_flags=qa_flags, ) result["Debug_Artefacts"] = artefact_paths return result def format_fingerprint_report(fp: dict[str, Any]) -> str: """ Format a fingerprint dict as a human-readable report string suitable for display in the Gradio interface. """ qa_flags = fp.get("QA_Flags", []) flag_section = "" if qa_flags: flag_lines = "\n".join(f" ⚠ {f}" for f in qa_flags) flag_section = f"\n── QA FLAGS ({len(qa_flags)}) ────────────────────────────────────\n{flag_lines}\n" debug_section = "" if "Debug_Artefacts" in fp: paths = fp["Debug_Artefacts"] debug_lines = "\n".join(f" {k}: {v}" for k, v in paths.items()) debug_section = f"\n── DEBUG ARTEFACTS ─────────────────────────────────────\n{debug_lines}\n" lines = [ f"╔══════════════════════════════════════════════════════╗", f" TOTEM STUDIO CODEX — FINGERPRINT EXTRACTION REPORT", f"╚══════════════════════════════════════════════════════╝", f"", f" Author: {fp['Author_Name']}", f" ID: {fp['Author_ID']}", f" Works: {fp['Works_Sampled'] or 'Not specified'}", f" Words: {fp['Sample_Words']}", f" Lines (total): {fp['Sample_Lines']}", f" Lines (verse cands): {fp.get('Sample_Candidate_Verse_Lines', 'n/a')}", f" Sentences: {fp['Sample_Sentences']}", f" Confidence: {fp['Confidence_Level']}", f" NLTK: {'Available' if fp['NLTK_Available'] else 'Not available — some metrics reduced accuracy'}", f"", f"── SONIC & RHYTHMIC ────────────────────────────────────", f" VM-001 Syllables per line (mean): {fp['VM-001_Syllables_per_line']}", f" VM-002 Syllable variance (SD): {fp['VM-002_Syllable_variance']}", f" VM-003 Rhyme scheme density: {fp['VM-003_Rhyme_density']}", f" VM-004 Rhyme scheme type: {fp['VM-004_Rhyme_type']}", f" VM-005 Stress regularity (0–1): {fp['VM-005_Stress_regularity']}", f"", f"── VOCABULARY & LEXICON ────────────────────────────────", f" VM-006 Vocab tier match 4–7 (0–1): {fp['VM-006_Vocab_tier_match']}", f" VM-007 Type-token ratio (0–1): {fp['VM-007_Type_token_ratio']}", f" VM-008 Invented word density (0–1): {fp['VM-008_Invented_word_density']}", f" VM-009 Avg word length (chars): {fp['VM-009_Avg_word_length']}", f"", f"── NARRATIVE & STRUCTURE ───────────────────────────────", f" VM-010 Sentence length mean (words): {fp['VM-010_Sentence_length_mean']}", f" VM-011 Sentence length variance (SD): {fp['VM-011_Sentence_length_variance']}", f" VM-012 Cumulative structure (0–1): {fp['VM-012_Cumulative_structure']}", f" VM-013 Dialogue proportion (0–1): {fp['VM-013_Dialogue_proportion']}", f"", f"── AGE & DEMOGRAPHIC ───────────────────────────────────", f" VM-024 Word count total: {fp['VM-024_Word_count']}", f" VM-025 Reading age (FK grade): {fp['VM-025_Reading_age_FK']}", f" VM-026 Exclamation density (per 100w): {fp['VM-026_Exclamation_density']}", f" VM-027 Question density (per 100w): {fp['VM-027_Question_density']}", f" VM-028 Repetition index (0–1): {fp['VM-028_Repetition_index']}", f"", f"── TIER 2 METRICS ──────────────────────────────────────", f" VM-014 to VM-023 require qualitative extraction.", f" Use Codex Build Prompt 2 (ChatGPT/Gemini) with the", f" same text sample to complete these fields.", ] report = "\n".join(lines) if flag_section: report += "\n" + flag_section if debug_section: report += "\n" + debug_section report += f"\n Copy values above into CODEX_03_FINGERPRINTS row: {fp['Author_ID']}" return report def process_upload( file_path: str | Path, author_name: str, author_id: str, works_sampled: str, start_page: int | None = None, end_page: int | None = None, extra_whitelist_str: str = "", debug_output_dir: str | Path | None = None, ) -> tuple[str, dict]: """ Entry point for Gradio interface. Args: file_path: Path to uploaded PDF or text file. author_name: Author's full name. author_id: Codex author ID. works_sampled: Comma-separated list of titles. start_page: Optional 1-based start page for story extraction. end_page: Optional 1-based end page for story extraction. extra_whitelist_str: Comma-separated proper nouns / invented terms to whitelist from VM-008 (e.g. "Gruffalo,Zog"). debug_output_dir: Directory to write debug artefacts. If None, no artefacts are written. Returns: Tuple of (formatted_report_string, raw_dict). """ try: story_text, raw_text = extract_text_from_file( file_path, start_page=start_page, end_page=end_page, ) if not story_text or len(story_text.split()) < 20: return ( "ERROR: No usable text extracted from file. " "Check the PDF contains selectable text (not scanned images), " "or try specifying a story page range.", {}, ) extra_whitelist: set[str] | None = None if extra_whitelist_str.strip(): extra_whitelist = { w.strip().lower() for w in extra_whitelist_str.split(",") if w.strip() } fp = extract_fingerprint( text=story_text, raw_text=raw_text, author_name=author_name, author_id=author_id, works_sampled=works_sampled, extra_whitelist=extra_whitelist, debug_output_dir=debug_output_dir, ) report = format_fingerprint_report(fp) return report, fp except Exception as e: return f"ERROR: {type(e).__name__}: {str(e)}", {} # ── STANDALONE TEST ─────────────────────────────────────────────────────────── if __name__ == "__main__": # Quick test with a small sample — run: python3 codex_extractor.py SAMPLE = """ The Gruffalo said that no gruffalo should go near the snake who bakes chocolate cake. The fox had a box full of socks by the dock, and the mouse ran free from the clock and the clock. He said to the owl, you're not like the rest, your feathers are orange, your beak is the best. She called to the bear in the cave far away, come out come out on this bright sunny day. """ fp = extract_fingerprint( text=SAMPLE, raw_text=SAMPLE, author_name="Test Author", author_id="CA-TEST", works_sampled="Test sample", extra_whitelist={"gruffalo"}, debug_output_dir="/tmp/codex_debug", ) print(format_fingerprint_report(fp)) print("\nDebug artefacts written to:", fp.get("Debug_Artefacts", {}))