Codex_Extractor / src /codex_extractor.py
Pointf5ive's picture
Update src/codex_extractor.py
31b2eee verified
Raw
History Blame
55 kB
"""
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", {}))