Integrate punctuation corrector into OCR routing and review learning

smoke_signal/scripts/punct_corrector.py

@@ -0,0 +1,233 @@
+"""
+Smoke Signal — Punctuation Corrector
+Post-OCR punctuation correction and routing signal.
+"""
+
+from __future__ import annotations
+
+import difflib
+import json
+import os
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional
+
+SS_ROOT = Path(os.environ.get("SS_DATA_ROOT", os.environ.get("SS_ROOT", "/tmp/smoke_signal")))
+CALIBRATION_DIR = SS_ROOT / "calibration"
+CALIBRATION_DIR.mkdir(parents=True, exist_ok=True)
+
+CORRECTION_MAP_PATH = CALIBRATION_DIR / "punct_correction_map.json"
+FLAGS_LOG_PATH = CALIBRATION_DIR / "punct_flags_log.jsonl"
+
+TERMINAL_PUNCT = set(".!?…")
+INLINE_PUNCT = set(",;:'\"-—/()[]")
+SMART_QUOTES = {"“", "”", "‘", "’"}
+
+KNOWN_SUBSTITUTIONS: dict[str, str] = {
+    "l": "!",
+    "I": "!",
+    "|": "!",
+    "1": "!",
+    ",,": "\"",
+    "''": "\"",
+    "·": ",",
+    "ʼ": "'",
+}
+
+MIN_PUNCT_PER_10_WORDS = 0.8
+
+
+def _load_correction_map() -> dict[str, dict[str, int]]:
+    if CORRECTION_MAP_PATH.exists():
+        try:
+            return json.loads(CORRECTION_MAP_PATH.read_text(encoding="utf-8"))
+        except Exception:
+            return {}
+    return {}
+
+
+def _save_correction_map(cmap: dict[str, dict[str, int]]) -> None:
+    CORRECTION_MAP_PATH.write_text(json.dumps(cmap, indent=2, ensure_ascii=False), encoding="utf-8")
+
+
+def _top_substitution(cmap: dict[str, dict[str, int]], raw_char: str) -> Optional[str]:
+    candidates = cmap.get(raw_char, {})
+    if not candidates:
+        return None
+    best_char, best_count = max(candidates.items(), key=lambda x: x[1])
+    return best_char if int(best_count) >= 2 else None
+
+
+def _split_lines(text: str) -> list[str]:
+    return [line.strip() for line in text.splitlines() if line.strip()]
+
+
+def _word_count(text: str) -> int:
+    return len(re.findall(r"\b\w+\b", text))
+
+
+def _punct_count(text: str) -> int:
+    punct_set = TERMINAL_PUNCT | INLINE_PUNCT | SMART_QUOTES
+    return sum(1 for ch in text if ch in punct_set)
+
+
+def _ends_with_terminal(line: str) -> bool:
+    return bool(line) and line[-1] in TERMINAL_PUNCT
+
+
+def validate_punctuation(text: str) -> list[dict]:
+    flags: list[dict] = []
+    for idx, line in enumerate(_split_lines(text)):
+        words = _word_count(line)
+        if words >= 4 and not _ends_with_terminal(line):
+            flags.append(
+                {
+                    "rule": "missing_terminal",
+                    "line": line,
+                    "line_idx": idx,
+                    "detail": f"Line has {words} words but no terminal punctuation.",
+                }
+            )
+
+        if line and line[-1] in {"l", "|", "I"} and words >= 3:
+            flags.append(
+                {
+                    "rule": "probable_exclamation",
+                    "line": line,
+                    "line_idx": idx,
+                    "detail": f"Line ends with '{line[-1]}' which may be OCR misread of '!'.",
+                }
+            )
+
+        if re.search(r"[,;.]{2,}", line):
+            flags.append(
+                {
+                    "rule": "doubled_punct",
+                    "line": line,
+                    "line_idx": idx,
+                    "detail": "Consecutive punctuation marks detected.",
+                }
+            )
+
+        has_smart = any(ch in line for ch in SMART_QUOTES)
+        has_straight = any(ch in line for ch in {'"', "'"})
+        if has_smart and has_straight:
+            flags.append(
+                {
+                    "rule": "mixed_quotes",
+                    "line": line,
+                    "line_idx": idx,
+                    "detail": "Mixed smart and straight quotes in same line.",
+                }
+            )
+
+    return flags
+
+
+def punctuation_confidence_penalty(text: str, ocr_confidence: float) -> float:
+    words = _word_count(text)
+    if words < 5:
+        return float(ocr_confidence)
+
+    punct = _punct_count(text)
+    ratio = (punct / words) * 10.0
+    if ratio >= MIN_PUNCT_PER_10_WORDS:
+        return float(ocr_confidence)
+
+    shortfall = MIN_PUNCT_PER_10_WORDS - ratio
+    penalty = min(shortfall * 0.15, 0.4)
+    return round(max(float(ocr_confidence) - penalty, 0.05), 4)
+
+
+def _align_chars(raw: str, gold: str) -> list[tuple[str, str]]:
+    out: list[tuple[str, str]] = []
+    matcher = difflib.SequenceMatcher(None, raw, gold, autojunk=False)
+    punct_set = TERMINAL_PUNCT | INLINE_PUNCT | SMART_QUOTES
+    for tag, i1, i2, j1, j2 in matcher.get_opcodes():
+        if tag == "replace" and (i2 - i1) == 1 and (j2 - j1) == 1:
+            raw_char = raw[i1]
+            gold_char = gold[j1]
+            if gold_char in punct_set:
+                out.append((raw_char, gold_char))
+    return out
+
+
+def record_punctuation_correction(raw_text: str, gold_text: str, book_id: str = "") -> int:
+    if not raw_text or not gold_text or raw_text == gold_text:
+        return 0
+
+    pairs = _align_chars(raw_text, gold_text)
+    if not pairs:
+        return 0
+
+    cmap = _load_correction_map()
+    for raw_char, gold_char in pairs:
+        if raw_char not in cmap:
+            cmap[raw_char] = {}
+        cmap[raw_char][gold_char] = int(cmap[raw_char].get(gold_char, 0)) + 1
+    _save_correction_map(cmap)
+
+    entry = {
+        "ts": datetime.now(timezone.utc).isoformat(),
+        "book_id": book_id,
+        "pairs": pairs,
+    }
+    with FLAGS_LOG_PATH.open("a", encoding="utf-8") as fh:
+        fh.write(json.dumps(entry, ensure_ascii=False) + "\n")
+
+    return len(pairs)
+
+
+def apply_punctuation_corrections(raw_text: str, book_id: str = "") -> tuple[str, list[dict], float]:
+    _ = book_id
+    if not raw_text:
+        return raw_text, [], 1.0
+
+    cmap = _load_correction_map()
+
+    lines = raw_text.splitlines()
+    corrected_lines: list[str] = []
+    for line in lines:
+        stripped = line.rstrip()
+        if stripped and stripped[-1] in KNOWN_SUBSTITUTIONS:
+            candidate = KNOWN_SUBSTITUTIONS[stripped[-1]]
+            if candidate in TERMINAL_PUNCT:
+                stripped = stripped[:-1] + candidate
+        corrected_lines.append(stripped)
+    text = "\n".join(corrected_lines)
+
+    for raw_seq, gold_seq in KNOWN_SUBSTITUTIONS.items():
+        if len(raw_seq) > 1:
+            text = text.replace(raw_seq, gold_seq)
+
+    result_chars: list[str] = []
+    for ch in text:
+        learned = _top_substitution(cmap, ch)
+        result_chars.append(learned if learned else ch)
+    text = "".join(result_chars)
+
+    flags = validate_punctuation(text)
+    words = _word_count(text)
+    punct = _punct_count(text)
+    if words == 0:
+        punct_score = 1.0
+    else:
+        ratio = (punct / words) * 10.0
+        punct_score = round(min(ratio / MIN_PUNCT_PER_10_WORDS, 1.0), 4)
+
+    return text, flags, punct_score
+
+
+def correction_map_summary() -> dict:
+    cmap = _load_correction_map()
+    pairs: list[dict] = []
+    for raw_char, golds in cmap.items():
+        for gold_char, count in golds.items():
+            pairs.append({"raw": raw_char, "gold": gold_char, "count": int(count)})
+    pairs.sort(key=lambda x: x["count"], reverse=True)
+    return {
+        "total_pairs": len(pairs),
+        "top_substitutions": pairs[:20],
+        "map_path": str(CORRECTION_MAP_PATH),
+    }

smoke_signal_tab.py

@@ -24,6 +24,7 @@ Self-improvement loop:
 import csv
 import concurrent.futures
 import hashlib
+import importlib.util
 import json
 import os
 import tempfile
@@ -58,6 +59,7 @@ GOLD_FILE    = GOLD_DIR / "gold_corrections.jsonl"
 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"
+PUNCT_CORRECTOR_PATH = Path(__file__).resolve().parent / "smoke_signal" / "scripts" / "punct_corrector.py"
 
 
 def _load_banner_image_css() -> str:
@@ -78,6 +80,77 @@ def _load_banner_image_css() -> str:
 
 SS_BANNER_IMAGE_CSS = _load_banner_image_css()
 
+
+_PUNCT_MODULE = None
+_PUNCT_MODULE_ERROR = None
+
+
+def _load_punct_module():
+    global _PUNCT_MODULE, _PUNCT_MODULE_ERROR
+    if _PUNCT_MODULE is not None:
+        return _PUNCT_MODULE
+    if _PUNCT_MODULE_ERROR is not None:
+        return None
+    if not PUNCT_CORRECTOR_PATH.exists():
+        _PUNCT_MODULE_ERROR = f"not found: {PUNCT_CORRECTOR_PATH}"
+        return None
+    try:
+        spec = importlib.util.spec_from_file_location("smoke_signal_punct_corrector", str(PUNCT_CORRECTOR_PATH))
+        if spec is None or spec.loader is None:
+            _PUNCT_MODULE_ERROR = "invalid import spec"
+            return None
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        _PUNCT_MODULE = module
+        return _PUNCT_MODULE
+    except Exception as e:
+        _PUNCT_MODULE_ERROR = str(e)
+        return None
+
+
+def _apply_punctuation_corrections(text: str, book_id: str) -> tuple[str, list, float]:
+    module = _load_punct_module()
+    if module is None:
+        return text, [], 1.0
+    try:
+        corrected, flags, score = module.apply_punctuation_corrections(text or "", book_id=book_id)
+        return corrected, flags, float(score)
+    except Exception:
+        return text, [], 1.0
+
+
+def _punctuation_confidence_penalty(text: str, confidence: float) -> float:
+    module = _load_punct_module()
+    if module is None:
+        return float(confidence)
+    try:
+        return float(module.punctuation_confidence_penalty(text or "", float(confidence)))
+    except Exception:
+        return float(confidence)
+
+
+def _record_punctuation_correction(raw_text: str, gold_text: str, book_id: str) -> int:
+    module = _load_punct_module()
+    if module is None:
+        return 0
+    try:
+        return int(module.record_punctuation_correction(raw_text or "", gold_text or "", book_id=book_id))
+    except Exception:
+        return 0
+
+
+def _punctuation_map_summary() -> dict:
+    module = _load_punct_module()
+    if module is None:
+        return {"total_pairs": 0, "top_substitutions": []}
+    try:
+        summary = module.correction_map_summary()
+        if not isinstance(summary, dict):
+            return {"total_pairs": 0, "top_substitutions": []}
+        return summary
+    except Exception:
+        return {"total_pairs": 0, "top_substitutions": []}
+
 # ── Confidence calibration state (in-memory, persisted to disk) ────────────────
 CALIBRATION_FILE = SS_ROOT / "manifest" / "confidence_calibration.json"
 
@@ -1166,6 +1239,12 @@ def run_ocr() -> tuple:
             )
         )
 
+    punct_mod = _load_punct_module()
+    if punct_mod is None:
+        log.append(log_line(f"⚠ Punctuation corrector unavailable ({_PUNCT_MODULE_ERROR or 'unknown'})"))
+    else:
+        log.append(log_line("✓ Punctuation corrector active (rule-check + confidence penalty + learning map)"))
+
     for _, row in eligible.iterrows():
         book_id = row["book_id"]
         profile_path = PROFILES_DIR / f"{book_id}_page_profile.json"
@@ -1371,33 +1450,45 @@ def run_ocr() -> tuple:
                 conf = 0.0
                 method = "skipped-no-surya"
 
+            raw_text = " ".join(r["text"] for r in regions)[:500]
+            corrected_text, punct_flags, punct_score = _apply_punctuation_corrections(raw_text, book_id)
+            conf_adjusted = _punctuation_confidence_penalty(corrected_text, conf)
+
             if method in ("tesseract-noise-filtered", "tesseract-lowconf-filtered") and not regions:
                 conf_class = "auto-accept"
-            elif conf >= default_cal["auto_accept"]:
+            elif conf_adjusted >= default_cal["auto_accept"]:
                 conf_class = "auto-accept"
-            elif conf >= default_cal["review"]:
+            elif conf_adjusted >= default_cal["review"]:
                 conf_class = "review-required"
-                review_pages.append(page_num)
-            elif conf >= default_cal["quarantine"]:
+            elif conf_adjusted >= default_cal["quarantine"]:
                 conf_class = "low-confidence"
-                review_pages.append(page_num)
             else:
                 conf_class = "quarantine"
+
+            # If punctuation rules detect likely text-quality issues, force at least review-required.
+            if punct_flags and conf_class == "auto-accept":
+                conf_class = "review-required"
+
+            if conf_class in ("review-required", "low-confidence"):
+                review_pages.append(page_num)
+            elif conf_class == "quarantine":
                 quarantine_pages.append(page_num)
 
             ocr_pages.append({
                 "page_number": page_num,
                 "route": route,
                 "regions": regions,
-                "page_confidence": conf,
+                "page_confidence_raw": conf,
+                "page_confidence": conf_adjusted,
                 "confidence_class": conf_class,
                 "extraction_method": method,
+                "punctuation_score": punct_score,
+                "punctuation_flags": punct_flags,
                 "render_path": page_data.get("render_path"),
                 "ocred_at": datetime.utcnow().isoformat() + "Z",
             })
 
             if conf_class in ("review-required", "low-confidence", "quarantine"):
-                raw_text = " ".join(r["text"] for r in regions)[:500]
                 region_id = f"{book_id}_p{page_num:04d}"
                 queue_rows.append({
                     "book_id": book_id,
@@ -1406,9 +1497,13 @@ def run_ocr() -> tuple:
                     "region_id": region_id,
                     "region_class": "narration",
                     "crop_path": page_data.get("render_path", ""),
-                    "raw_ocr": raw_text,
-                    "confidence": conf,
+                    "raw_ocr": corrected_text,
+                    "raw_ocr_original": raw_text,
+                    "confidence_raw": conf,
+                    "confidence": conf_adjusted,
                     "confidence_class": conf_class,
+                    "punct_score": punct_score,
+                    "punct_flags_count": len(punct_flags),
                     "status": "quarantine" if conf_class == "quarantine" else "pending",
                     "reviewer": "",
                     "correction": "",
@@ -1445,7 +1540,7 @@ def run_ocr() -> tuple:
         qdf = pd.DataFrame(queue_rows)
         if QUEUE_CSV.exists():
             existing = pd.read_csv(QUEUE_CSV)
-            qdf = pd.concat([existing, qdf], ignore_index=True).drop_duplicates(subset=["region_id"])
+            qdf = pd.concat([existing, qdf], ignore_index=True).drop_duplicates(subset=["region_id"], keep="last")
         qdf.to_csv(QUEUE_CSV, index=False)
 
     save_manifest_df(df)
@@ -1464,8 +1559,13 @@ def _ocr_status_html() -> str:
     cal = load_calibration()
     default = cal.get("_default", DEFAULT_CALIBRATION["_default"])
     corrections = sum(v.get("corrections",0) for v in cal.values() if isinstance(v, dict))
+    punct_pairs = int(_punctuation_map_summary().get("total_pairs", 0))
 
-    training_badge = f'<div class="ss-training-badge"><div class="ss-pulse"></div>{corrections} corrections logged · thresholds auto={default["auto_accept"]:.0%} review={default["review"]:.0%}</div>'
+    training_badge = (
+        f'<div class="ss-training-badge"><div class="ss-pulse"></div>'
+        f'{corrections} corrections logged · punct map={punct_pairs} pairs · '
+        f'thresholds auto={default["auto_accept"]:.0%} review={default["review"]:.0%}</div>'
+    )
 
     return f"""
     <div class="ss-metrics">
@@ -1544,7 +1644,8 @@ def save_review_decision(idx: int, final_text: str, action: str, reviewer: str,
     idx  = idx % len(pending)
     item = pending.iloc[idx]
 
-    raw_text    = item.get("raw_ocr","")
+    raw_text = item.get("raw_ocr", "")
+    raw_text_original = item.get("raw_ocr_original", raw_text)
     was_correct = (final_text.strip() == raw_text.strip())
     region_class = item.get("region_class","narration")
 
@@ -1559,6 +1660,7 @@ def save_review_decision(idx: int, final_text: str, action: str, reviewer: str,
         "status":      action,
         "final_text":  final_text,
         "raw_text":    raw_text,
+        "raw_text_original": raw_text_original,
         "reason_code": reason,
         "reviewer":    reviewer or "reviewer",
         "was_correct": was_correct,
@@ -1574,22 +1676,39 @@ def save_review_decision(idx: int, final_text: str, action: str, reviewer: str,
     d_df.to_csv(DECISIONS_CSV, index=False)
 
     # Append to gold training set
+    try:
+        punct_score = float(item.get("punct_score", 1.0))
+    except Exception:
+        punct_score = 1.0
+    try:
+        punct_flags_count = int(float(item.get("punct_flags_count", 0)))
+    except Exception:
+        punct_flags_count = 0
+
     with open(GOLD_FILE, "a", encoding="utf-8") as f:
         f.write(json.dumps({
             **decision,
             "region_class":  region_class,
             "confidence":    float(item.get("confidence", 0)),
             "conf_class":    item.get("confidence_class",""),
+            "punct_score":   punct_score,
+            "punct_flags_count": punct_flags_count,
         }) + "\n")
 
+    learned_pairs = 0
+    if action == "edited":
+        learned_pairs = _record_punctuation_correction(raw_text_original, final_text, item.get("book_id", ""))
+
     cal         = load_calibration()
     default     = cal.get("_default", DEFAULT_CALIBRATION["_default"])
     corrections = sum(v.get("corrections",0) for v in cal.values() if isinstance(v,dict))
     gold_count  = sum(1 for _ in open(GOLD_FILE)) if GOLD_FILE.exists() else 0
 
+    punct_pairs = int(_punctuation_map_summary().get("total_pairs", 0))
+    learned_note = f" · +{learned_pairs} punct learns" if learned_pairs else ""
     feedback = (f"<div class='ss-training-badge'><div class='ss-pulse'></div>"
-                f"Gold set: {gold_count} examples · {corrections} corrections · "
-                f"auto-accept threshold: {default['auto_accept']:.0%}</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>")
 
     return feedback, *get_review_item(0)