feat: persistent noise pattern exclusion — reject NON_STORY_TEXT to block on next OCR run

smoke_signal_tab.py

@@ -57,6 +57,48 @@ for d in [SOURCE_DIR, MANIFEST_CSV.parent, PROFILES_DIR, OCR_RAW_DIR,
     d.mkdir(parents=True, exist_ok=True)
 
 GOLD_FILE    = GOLD_DIR / "gold_corrections.jsonl"
+NOISE_DIR    = SS_ROOT / "calibration" / "noise_patterns"
+
+
+def _noise_path(book_id: str) -> Path:
+    NOISE_DIR.mkdir(parents=True, exist_ok=True)
+    return NOISE_DIR / f"{book_id}_noise.json"
+
+
+def _load_noise_patterns(book_id: str) -> list:
+    path = _noise_path(book_id)
+    if not path.exists():
+        return []
+    try:
+        return json.loads(path.read_text())
+    except Exception:
+        return []
+
+
+def _save_noise_pattern(book_id: str, pattern: str) -> None:
+    pattern = pattern.strip()
+    if not pattern or len(pattern) < 3:
+        return
+    patterns = _load_noise_patterns(book_id)
+    if pattern not in patterns:
+        patterns.append(pattern)
+        _noise_path(book_id).write_text(json.dumps(patterns, indent=2))
+
+
+def _text_matches_noise(text: str, patterns: list) -> bool:
+    if not patterns or not text:
+        return False
+    text_tokens = set(text.lower().split())
+    if not text_tokens:
+        return False
+    for pattern in patterns:
+        pat_tokens = set(pattern.lower().split())
+        if not pat_tokens:
+            continue
+        overlap = len(text_tokens & pat_tokens) / len(text_tokens)
+        if overlap >= 0.6:
+            return True
+    return False
 DECISIONS_CSV = REVIEW_DIR / "review_decisions.csv"
 QUEUE_CSV    = REVIEW_DIR / "review_queue.csv"
 BANNER_DATA_URI_FILE = Path(__file__).resolve().parent / "assets" / "smoke_signal_banner_data_uri.txt"
@@ -2371,6 +2413,13 @@ def run_ocr(page_selection: str = "", page_exclusion: str = "", replace_book_que
                 method = "skipped-no-surya"
 
             raw_text = " ".join(r["text"] for r in regions)[:500]
+
+            # Skip pages matching known noise patterns for this book
+            _noise_pats = _load_noise_patterns(book_id)
+            if _noise_pats and _text_matches_noise(raw_text, _noise_pats):
+                log.append(log_line(f"  ⊘ {book_id} p{page_num}: matched noise pattern — skipped"))
+                continue
+
             corrected_text, punct_flags, punct_score = _apply_punctuation_corrections(
                 raw_text,
                 book_id,
@@ -2656,6 +2705,10 @@ def save_review_decision(idx: int, final_text: str, action: str, reviewer: str,
         }) + "\n")
 
     learned_pairs = 0
+    if action == "rejected" and str(reason).strip() == "NON_STORY_TEXT":
+        noise_text = (final_text.strip() or raw_text.strip())
+        if noise_text:
+            _save_noise_pattern(str(item.get("book_id", "")), noise_text)
     if action == "edited":
         learned_pairs = _record_punctuation_correction(
             raw_text_original,