backend/chapters.py (8958 bytes)
1 """Chapter matching that works on any script. 2 3 The alignment backend decides which text chapter belongs to which audio chapter 4 by character-level edit similarity (`rapidfuzz.fuzz.ratio`) against a fixed 5 threshold of 40. That is calibrated for Japanese and does not transfer. Measured 6 noise floor between *unrelated* chapters of the same book: 7 8 Japanese 23-26 <- threshold 40 discriminates well 9 Russian 39 <- borderline 10 Spanish 44 11 English 45-47 <- unrelated chapters clear the threshold 12 13 With twenty-six letters, any two prose texts are ~45% similar by chance, so for 14 Latin scripts the gate is *below the noise* and accepts anything. Cleaning does 15 not fix it (a generic punctuation strip moved English only 45.3 -> 42.4); the 16 cause is alphabet size, not typography. 17 18 This module scores on character n-gram overlap instead, and calibrates the 19 threshold against the book it is actually looking at. Measured on the same data: 20 21 metric ru floor en floor ja floor true runner-up ratio 22 fuzz.ratio 39.2 45.1 26.1 86.3 40.8 2.1x 23 word jaccard 10.4 17.7 1.9 63.2 14.8 4.3x 24 4-gram jaccard 4.8 10.6 2.5 53.3 5.6 9.5x 25 26 n-grams rather than words because plenty of languages do not put spaces between 27 them; a word-level metric collapses on Japanese, where a "word" is a whole run of 28 characters. 29 """ 30 31 from __future__ import annotations 32 33 import random 34 import statistics as st 35 import unicodedata 36 from dataclasses import dataclass 37 38 import regex 39 40 NGRAM = 4 41 42 # Everything that is not a letter or a number: punctuation, spacing, marks. 43 # Dropped so typography cannot influence the comparison. 44 _NOISE = regex.compile(r"[^\p{L}\p{N}]+") 45 46 47 def normalize(text: str) -> str: 48 return _NOISE.sub("", unicodedata.normalize("NFKD", text.casefold())) 49 50 51 def fingerprint(text: str, n: int = NGRAM) -> frozenset[str]: 52 """The set of character n-grams in `text`, after normalisation. 53 54 Computed once per chapter and reused: comparing two fingerprints is a set 55 intersection, which is far cheaper than an edit distance over the same text. 56 """ 57 s = normalize(text) 58 if len(s) < n: 59 return frozenset() 60 return frozenset(s[i : i + n] for i in range(len(s) - n + 1)) 61 62 63 def jaccard(a: frozenset[str], b: frozenset[str]) -> float: 64 if not a or not b: 65 return 0.0 66 return 100.0 * len(a & b) / len(a | b) 67 68 69 def containment(probe: frozenset[str], reference: frozenset[str]) -> float: 70 """How much of `probe` appears in `reference`, as a percentage. 71 72 Used when a short transcript sample is compared against a whole chapter: 73 Jaccard would punish the length difference through the union term, since a 74 three-minute sample cannot cover a thirty-minute chapter. Containment asks 75 the question we actually mean - "is this passage in that chapter" - and is 76 unaffected by how much longer the chapter is. 77 """ 78 if not probe or not reference: 79 return 0.0 80 return 100.0 * len(probe & reference) / len(probe) 81 82 83 @dataclass(frozen=True) 84 class Calibration: 85 """The similarity this book produces by chance. 86 87 Sampled from the book itself, so it adapts to script, vocabulary and 88 register instead of relying on a constant that only suits one language. 89 """ 90 91 floor: float 92 high: float # worst case seen among unrelated pairs 93 samples: int 94 95 # A real match must clear the floor by this factor... 96 FLOOR_MULTIPLE = 1.6 97 # ...and beat whatever came second by this much. This is the test a fixed 98 # threshold cannot do, and the one that catches a wrong book whose language 99 # alone makes every chapter score alike. 100 RUNNER_UP_MULTIPLE = 1.5 101 102 @property 103 def accept_at(self) -> float: 104 """Minimum score worth taking seriously for this book.""" 105 return max(self.floor * self.FLOOR_MULTIPLE, self.high * 1.15, 1.0) 106 107 108 def calibrate( 109 chapter_texts: list[str], 110 pairs: int = 150, 111 seed: int = 0, 112 probe_chars: int = 2000, 113 ) -> Calibration: 114 """Estimate the chance-similarity floor from unrelated pairs of this book. 115 116 Calibrated with the *same* measure and the *same shape* of comparison that 117 real matching uses: a short probe against a whole chapter. Measuring the 118 floor with Jaccard while matching with containment would put the threshold 119 on a different scale from the scores it gates. 120 """ 121 usable = [t for t in chapter_texts if len(normalize(t)) > probe_chars // 2] 122 if len(usable) < 3: 123 # Too little to calibrate against; fall back to something conservative. 124 return Calibration(floor=20.0, high=40.0, samples=0) 125 126 references = [fingerprint(t) for t in usable] 127 # A probe is a slice the size of a transcript sample, so the floor reflects 128 # what an unrelated passage of this length scores against a full chapter. 129 probes = [fingerprint(normalize(t)[:probe_chars]) for t in usable] 130 131 rng = random.Random(seed) 132 scores = [] 133 for _ in range(pairs): 134 a, b = rng.sample(range(len(usable)), 2) 135 scores.append(containment(probes[a], references[b])) 136 137 scores.sort() 138 return Calibration( 139 floor=st.median(scores), 140 high=scores[min(len(scores) - 1, int(0.95 * len(scores)))], 141 samples=len(scores), 142 ) 143 144 145 @dataclass(frozen=True) 146 class Match: 147 text_index: int | None 148 score: float 149 runner_up: float 150 accepted: bool 151 calibration: Calibration 152 153 @property 154 def confidence(self) -> float: 155 """How far clear of the runner-up this match is. 1.0 means a tie.""" 156 if self.runner_up <= 0: 157 return float("inf") if self.score > 0 else 0.0 158 return self.score / self.runner_up 159 160 @property 161 def margin_over_noise(self) -> float: 162 if self.calibration.floor <= 0: 163 return float("inf") if self.score > 0 else 0.0 164 return self.score / self.calibration.floor 165 166 167 def best_match( 168 probe: frozenset[str], 169 chapters: list[frozenset[str]], 170 calibration: Calibration, 171 exclude: set[int] | None = None, 172 ) -> Match: 173 """Find the chapter a sample of audio came from. 174 175 Accepting requires clearing the book's own noise floor *and* beating the 176 runner-up. The second test is what a fixed threshold cannot do: if two 177 chapters score alike, the winner is arbitrary, however high the number. 178 """ 179 skip = exclude or set() 180 ranked = sorted( 181 ( 182 (containment(probe, ch), i) 183 for i, ch in enumerate(chapters) 184 if i not in skip and ch 185 ), 186 reverse=True, 187 ) 188 if not ranked: 189 return Match(None, 0.0, 0.0, False, calibration) 190 191 score, index = ranked[0] 192 runner_up = ranked[1][0] if len(ranked) > 1 else 0.0 193 194 accepted = score >= calibration.accept_at and ( 195 runner_up <= 0 or score >= runner_up * Calibration.RUNNER_UP_MULTIPLE 196 ) 197 return Match(index if accepted else None, score, runner_up, accepted, calibration) 198 199 200 def assign_monotonic(matrix: list[list[float]], accept_at: float) -> list[int | None]: 201 """Pair audio chapters to text chapters in order, maximising total score. 202 203 The backend's own matcher walks audio chapters in order and *consumes* text 204 chapters as it goes, with no backtracking: an early chapter - usually 205 chapter 0, where publisher announcements live and which is therefore the 206 least representative chapter in the book - can permanently take the text 207 that belonged to another. 208 209 This solves the whole assignment at once, and requires it to be 210 order-preserving, which is true of every book: chapter k cannot come from 211 text that precedes chapter k-1's. Skips are allowed on both sides, for front 212 matter and for chapters nobody narrated. 213 """ 214 n_audio, n_text = len(matrix), (len(matrix[0]) if matrix else 0) 215 if not n_audio or not n_text: 216 return [None] * n_audio 217 218 NEG = float("-inf") 219 # best[i][j] = best total score pairing the first i audio with first j text 220 best = [[0.0] * (n_text + 1) for _ in range(n_audio + 1)] 221 back = [[0] * (n_text + 1) for _ in range(n_audio + 1)] # 0 skip-a,1 skip-t,2 pair 222 223 for i in range(1, n_audio + 1): 224 for j in range(1, n_text + 1): 225 skip_audio = best[i - 1][j] 226 skip_text = best[i][j - 1] 227 score = matrix[i - 1][j - 1] 228 pair = best[i - 1][j - 1] + (score if score >= accept_at else NEG) 229 230 chosen = max(skip_audio, skip_text, pair) 231 best[i][j] = chosen 232 back[i][j] = 2 if chosen == pair else (0 if chosen == skip_audio else 1) 233 234 out: list[int | None] = [None] * n_audio 235 i, j = n_audio, n_text 236 while i > 0 and j > 0: 237 move = back[i][j] 238 if move == 2: 239 out[i - 1] = j - 1 240 i, j = i - 1, j - 1 241 elif move == 0: 242 i -= 1 243 else: 244 j -= 1 245 return out