Spaces:
Sleeping
Sleeping
| """ | |
| 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 shutil import which | |
| from typing import Any | |
| # ── OPTIONAL IMPORTS WITH GRACEFUL FALLBACK ────────────────────────────────── | |
| try: | |
| import pdfplumber | |
| PDF_AVAILABLE = True | |
| except ImportError: | |
| PDF_AVAILABLE = False | |
| try: | |
| import pypdfium2 as pdfium | |
| PDFIUM_AVAILABLE = True | |
| except Exception: | |
| PDFIUM_AVAILABLE = False | |
| try: | |
| import pytesseract | |
| PYTESSERACT_AVAILABLE = True | |
| except Exception: | |
| PYTESSERACT_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 _word_count(text: str) -> int: | |
| return len(SIMPLE_TOKENISE_PATTERN.findall((text or "").lower())) | |
| def _ocr_runtime_ready() -> bool: | |
| return PDFIUM_AVAILABLE and PYTESSERACT_AVAILABLE and which("tesseract") is not None | |
| def extract_text_from_pdf_ocr( | |
| pdf_path: str | Path, | |
| start_page: int | None = None, | |
| end_page: int | None = None, | |
| ) -> tuple[str, str]: | |
| """ | |
| OCR fallback for scanned/image-only PDFs. | |
| Returns (story_text, raw_text) using the same page-filter logic as native extraction. | |
| """ | |
| if not PDFIUM_AVAILABLE: | |
| raise RuntimeError("OCR fallback unavailable: pypdfium2 is not installed.") | |
| if not PYTESSERACT_AVAILABLE: | |
| raise RuntimeError("OCR fallback unavailable: pytesseract is not installed.") | |
| if which("tesseract") is None: | |
| raise RuntimeError("OCR fallback unavailable: tesseract binary is not installed.") | |
| render_scale = float(os.getenv("OCR_RENDER_SCALE", "2.0")) | |
| ocr_lang = os.getenv("OCR_LANG", "eng") | |
| ocr_config = os.getenv("OCR_CONFIG", "--oem 1 --psm 6") | |
| max_pages = int(os.getenv("OCR_MAX_PAGES", "300")) | |
| raw_parts: list[str] = [] | |
| story_parts: list[str] = [] | |
| pdf = pdfium.PdfDocument(str(pdf_path)) | |
| total_pages = min(len(pdf), max_pages) | |
| 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 in range(total_pages): | |
| page = pdf[page_idx] | |
| bitmap = page.render(scale=render_scale) | |
| pil_image = bitmap.to_pil() | |
| page_text = pytesseract.image_to_string(pil_image, lang=ocr_lang, config=ocr_config) or "" | |
| raw_parts.append(page_text) | |
| 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 | |
| if total_pages > 1 and page_idx == 0: | |
| continue | |
| if is_front_matter_page(page_text): | |
| continue | |
| story_parts.append(page_text) | |
| return "\n".join(story_parts), "\n".join(raw_parts) | |
| 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 total_pages > 1 and 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) | |
| # If native text-layer extraction looks healthy, use it. | |
| if _word_count(story_text) >= 20: | |
| return story_text, raw_text | |
| # Otherwise, try OCR fallback for scanned/image-only pages. | |
| try: | |
| ocr_story_text, ocr_raw_text = extract_text_from_pdf_ocr( | |
| pdf_path, | |
| start_page=start_page, | |
| end_page=end_page, | |
| ) | |
| if _word_count(ocr_story_text) >= max(20, _word_count(story_text)): | |
| return ocr_story_text, ocr_raw_text | |
| except Exception: | |
| # Fall through: process_upload will raise a clear error if text is still insufficient. | |
| pass | |
| 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: | |
| if str(file_path).lower().endswith(".pdf") and not _ocr_runtime_ready(): | |
| return ( | |
| "ERROR: No usable text extracted from file and OCR runtime is unavailable. " | |
| "Install OCR dependencies (`pypdfium2`, `pytesseract`) and system package " | |
| "`tesseract-ocr` in the Space build.", | |
| {}, | |
| ) | |
| return ( | |
| "ERROR: No usable text extracted from file. " | |
| "OCR fallback could not recover enough text. " | |
| "Try a cleaner scan, higher resolution pages, or 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", {})) | |