Improve Smoke Signal noise-pattern learning and matching

smoke_signal_tab.py

@@ -58,6 +58,18 @@ for d in [SOURCE_DIR, MANIFEST_CSV.parent, PROFILES_DIR, OCR_RAW_DIR,
 
 GOLD_FILE    = GOLD_DIR / "gold_corrections.jsonl"
 NOISE_DIR    = SS_ROOT / "calibration" / "noise_patterns"
+NOISE_GLOBAL_FILE = NOISE_DIR / "_global_noise.json"
+NOISE_MIN_CHARS = int(os.environ.get("SS_NOISE_MIN_CHARS", "3"))
+NOISE_MATCH_MIN_PATTERN_COVERAGE = float(os.environ.get("SS_NOISE_MATCH_MIN_PATTERN_COVERAGE", "0.65"))
+NOISE_MATCH_MIN_TEXT_COVERAGE = float(os.environ.get("SS_NOISE_MATCH_MIN_TEXT_COVERAGE", "0.08"))
+
+
+def _normalize_noise_text(text: str) -> str:
+    txt = (text or "").lower()
+    txt = re.sub(r"[\r\n\t]+", " ", txt)
+    txt = re.sub(r"[^a-z0-9\u4e00-\u9fff\s]+", " ", txt)
+    txt = re.sub(r"\s+", " ", txt).strip()
+    return txt
 
 
 def _noise_path(book_id: str) -> Path:
@@ -65,38 +77,90 @@ def _noise_path(book_id: str) -> Path:
     return NOISE_DIR / f"{book_id}_noise.json"
 
 
-def _load_noise_patterns(book_id: str) -> list:
-    path = _noise_path(book_id)
+def _load_noise_patterns_for_scope(path: Path) -> list[str]:
     if not path.exists():
         return []
     try:
-        return json.loads(path.read_text())
+        data = json.loads(path.read_text())
     except Exception:
         return []
+    if not isinstance(data, list):
+        return []
+    out: list[str] = []
+    for p in data:
+        raw = str(p or "").strip()
+        norm = _normalize_noise_text(raw)
+        if len(norm) < NOISE_MIN_CHARS:
+            continue
+        out.append(raw)
+    return out
+
+
+def _load_noise_patterns(book_id: str, include_global: bool = True) -> list[str]:
+    combined: list[str] = []
+    seen: set[str] = set()
+    paths = [_noise_path(book_id)]
+    if include_global:
+        paths.append(NOISE_GLOBAL_FILE)
+    for path in paths:
+        for pattern in _load_noise_patterns_for_scope(path):
+            norm = _normalize_noise_text(pattern)
+            if norm in seen:
+                continue
+            seen.add(norm)
+            combined.append(pattern)
+    return combined
 
 
-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 _save_noise_pattern(book_id: str, pattern: str, save_global: bool = True) -> bool:
+    pattern = str(pattern or "").strip()
+    norm = _normalize_noise_text(pattern)
+    if len(norm) < NOISE_MIN_CHARS:
+        return False
+    changed = False
+    targets = [_noise_path(book_id)]
+    if save_global:
+        targets.append(NOISE_GLOBAL_FILE)
+    for target in targets:
+        existing = _load_noise_patterns_for_scope(target)
+        existing_norm = {_normalize_noise_text(p) for p in existing}
+        if norm in existing_norm:
+            continue
+        existing.append(pattern)
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_text(json.dumps(existing, indent=2, ensure_ascii=False))
+        changed = True
+    return changed
 
 
 def _text_matches_noise(text: str, patterns: list) -> bool:
     if not patterns or not text:
         return False
-    text_tokens = set(text.lower().split())
+    normalized_text = _normalize_noise_text(text)
+    if not normalized_text:
+        return False
+    text_tokens = set(normalized_text.split())
     if not text_tokens:
         return False
     for pattern in patterns:
-        pat_tokens = set(pattern.lower().split())
+        pat_norm = _normalize_noise_text(str(pattern or ""))
+        if not pat_norm:
+            continue
+        # Best signal for recurring watermarks/headers
+        if pat_norm in normalized_text:
+            return True
+        pat_tokens = set(pat_norm.split())
         if not pat_tokens:
             continue
-        overlap = len(text_tokens & pat_tokens) / len(text_tokens)
-        if overlap >= 0.6:
+        overlap_count = len(text_tokens & pat_tokens)
+        if overlap_count == 0:
+            continue
+        pattern_coverage = overlap_count / len(pat_tokens)
+        text_coverage = overlap_count / len(text_tokens)
+        if (
+            pattern_coverage >= NOISE_MATCH_MIN_PATTERN_COVERAGE
+            and text_coverage >= NOISE_MATCH_MIN_TEXT_COVERAGE
+        ):
             return True
     return False
 DECISIONS_CSV = REVIEW_DIR / "review_decisions.csv"
@@ -2705,18 +2769,21 @@ def save_review_decision(idx: int, final_text: str, action: str, noise_text_inpu
         }) + "\n")
 
     learned_pairs = 0
+    learned_noise_added = 0
+    active_noise_patterns = 0
     if action == "rejected" and str(reason).strip() == "NON_STORY_TEXT":
         # Use pasted noise text if provided, otherwise fall back to raw OCR
         noise_source = str(noise_text_input or "").strip() or final_text.strip() or raw_text.strip()
         book_id_for_noise = str(item.get("book_id", ""))
-        saved_count = 0
         for line in noise_source.splitlines():
             line = line.strip()
             if line:
-                _save_noise_pattern(book_id_for_noise, line)
-                saved_count += 1
-        if saved_count == 0 and noise_source:
-            _save_noise_pattern(book_id_for_noise, noise_source)
+                if _save_noise_pattern(book_id_for_noise, line):
+                    learned_noise_added += 1
+        if learned_noise_added == 0 and noise_source:
+            if _save_noise_pattern(book_id_for_noise, noise_source):
+                learned_noise_added += 1
+        active_noise_patterns = len(_load_noise_patterns(book_id_for_noise))
     if action == "edited":
         learned_pairs = _record_punctuation_correction(
             raw_text_original,
@@ -2732,9 +2799,12 @@ def save_review_decision(idx: int, final_text: str, action: str, noise_text_inpu
 
     punct_pairs = int(_punctuation_map_summary().get("total_pairs", 0))
     learned_note = f" · +{learned_pairs} punct learns" if learned_pairs else ""
+    noise_note = ""
+    if action == "rejected" and str(reason).strip() == "NON_STORY_TEXT":
+        noise_note = f" · +{learned_noise_added} noise learns · active noise patterns: {active_noise_patterns}"
     feedback = (f"<div class='ss-training-badge'><div class='ss-pulse'></div>"
                 f"Gold set: {gold_count} examples · {corrections} corrections · punct map: {punct_pairs} pairs"
-                f"{learned_note} · auto-accept threshold: {default['auto_accept']:.0%}</div>")
+                f"{learned_note}{noise_note} · auto-accept threshold: {default['auto_accept']:.0%}</div>")
 
     return feedback, *get_review_item(0)