Skip to content

Commit 80aa5c5

Browse files
Zaid Salemclaude
authored andcommitted
feat(wb-alerts): Phase 3b — noise-robust checkpoint name recovery
3a's live capture revealed most of the verified 67% "coverage gap" is real WHITELISTED checkpoints lost to social-media noise, not unknown ones. find_checkpoint gains two ADDITIVE recovery steps — they fire ONLY when exact/alias/substring all miss, and resolve ONLY to an exact whitelist name (no open-ended fuzzy, no wrong-checkpoint risk), so existing matches are unchanged: - de-elongation: collapse stretched letters (المربععه→المربعه, بحححرررري→بحري) - distinctive whole-token match: a known >=4-char single-token name present as a complete token in the de-elongated text (عطارة البرج فتحت الان → عطاره), recovering a real checkpoint buried in status/prefix noise. The >=4 + whole-token gate avoids short-substring false positives. Validated on the REAL captured candidates against the production 234-entry KB: 3/3 real checkpoints (Atara, Al-Murabba'a, Turmus Ayya) recovered, 3/3 garbage fragments still rejected. The status-required parser gate further guards against bare mentions. Suite 169 passed, FP audit 96/96. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent dac0b4c commit 80aa5c5

2 files changed

Lines changed: 106 additions & 0 deletions

File tree

services/westbank-alerts/app/checkpoint_knowledge_base.py

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,10 +10,17 @@
1010

1111
import json
1212
import logging
13+
import re
1314
from pathlib import Path
1415
from typing import Optional
1516
from .checkpoint_parser import _normalise
1617

18+
19+
def _collapse_elongation(s: str) -> str:
20+
"""Collapse runs of 2+ identical characters to one — social-media letter
21+
stretching (المربععه → المربعه, بحححرررري → بحري)."""
22+
return re.sub(r"(.)\1+", r"\1", s)
23+
1724
log = logging.getLogger("checkpoint_knowledge_base")
1825

1926

@@ -159,6 +166,35 @@ def find_checkpoint(self, name_ar: str) -> Optional[str]:
159166
# Incoming is substring of known — always accept
160167
return canonical_key
161168

169+
# ── Phase 3b: noise-robust recovery. ADDITIVE — only reached when exact /
170+
# alias / substring all missed, and resolves ONLY to an exact whitelist
171+
# name (no open-ended fuzzy), so existing matches are unchanged and there
172+
# is no wrong-checkpoint risk.
173+
174+
# 3.5 De-elongation retry: collapse stretched letters and re-match
175+
# (المربععه بحري → المربعه ...).
176+
deelong = _collapse_elongation(name_norm)
177+
if deelong != name_norm:
178+
if deelong in self.by_name_norm:
179+
return self.by_name_norm[deelong]
180+
if deelong in self.aliases:
181+
return self.aliases[deelong]
182+
for norm_known, canonical_key in self.all_names:
183+
if len(norm_known) >= 3 and norm_known in deelong:
184+
extra = len(deelong.split()) - len(norm_known.split())
185+
if extra <= len(norm_known.split()):
186+
return canonical_key
187+
188+
# 4. Distinctive whole-token match: a known single-token name of >= 4 chars
189+
# appearing as a complete token in the (de-elongated) text recovers a real
190+
# whitelisted checkpoint surrounded by status/prefix noise
191+
# (عطارة البرج فتحت الان → عطاره). The >= 4 chars + whole-token gate avoids
192+
# short-substring false positives ("تل" inside "مقاتلو").
193+
tokens = set(deelong.split())
194+
for norm_known, canonical_key in self.all_names:
195+
if len(norm_known) >= 4 and " " not in norm_known and norm_known in tokens:
196+
return canonical_key
197+
162198
return None
163199

164200
def get_checkpoint(self, canonical_key: str) -> Optional[dict]:
Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
"""Phase 3b: recover real WHITELISTED checkpoints buried in social-media noise.
2+
3+
The live candidate capture (3a) revealed that much of the 67% "coverage gap" is
4+
not unknown checkpoints — it's known ones the strict matcher dropped because the
5+
name carried noise: letter elongation (المربععه), trailing status words
6+
(عطارة البرج فتحت الان), entrance prefixes (على مدخل ترمسعيا).
7+
8+
These are purely-additive recovery steps in find_checkpoint: they fire ONLY when
9+
exact/alias/substring all miss, and only resolve to an EXACT whitelist name (no
10+
open-ended fuzzy), so existing matches are unchanged and there is no
11+
wrong-checkpoint risk. Garbage and too-short tokens must still miss.
12+
"""
13+
import os
14+
os.environ.setdefault("API_SECRET_KEY", "test-secret-key-0123456789abcdef")
15+
16+
from app.checkpoint_knowledge_base import CheckpointKnowledgeBase
17+
from app.checkpoint_parser import _normalise
18+
19+
20+
def _kb():
21+
kb = CheckpointKnowledgeBase()
22+
rows = [
23+
("عطاره", "عطارة", "Atara"),
24+
("المربعه", "المربعة", "Al-Murabbaa"),
25+
("ترمسعيا", "ترمسعيا", "Turmus Ayya"),
26+
("بيتا", "بيتا", "Beita"), # 4-char name, for the distinctive-token guard
27+
]
28+
for ck, name_ar, name_en in rows:
29+
cp = {"canonical_key": ck, "name_ar": name_ar, "name_en": name_en,
30+
"region": "x", "checkpoint_type": "checkpoint", "latitude": 32.0, "longitude": 35.0}
31+
kb.by_canonical_key[ck] = cp
32+
n = _normalise(name_ar)
33+
kb.by_name_norm[n] = ck
34+
kb.all_names.append((n, ck))
35+
kb.all_names.sort(key=lambda x: -len(x[0]))
36+
return kb
37+
38+
39+
def test_clean_names_still_match_exactly():
40+
kb = _kb()
41+
assert kb.find_checkpoint("عطارة") == "عطاره"
42+
assert kb.find_checkpoint("ترمسعيا") == "ترمسعيا"
43+
44+
45+
def test_recovers_letter_elongation():
46+
# المربععه (doubled ع) + garbled suffix → Al-Murabba'a.
47+
assert _kb().find_checkpoint("المربععه بحححرررري") == "المربعه"
48+
49+
50+
def test_recovers_trailing_status_words():
51+
# عطارة + "البرج فتحت الان" (the tower opened now) → Atara.
52+
assert _kb().find_checkpoint("عطارة البرج فتحت الان") == "عطاره"
53+
54+
55+
def test_recovers_entrance_prefix():
56+
# "على مدخل" (at the entrance of) + ترمسعيا → Turmus Ayya.
57+
assert _kb().find_checkpoint("على مدخل ترمسعيا") == "ترمسعيا"
58+
59+
60+
def test_garbage_fragments_still_miss():
61+
kb = _kb()
62+
for junk in ["صار في", "وفي", "اندماج", "الوضع هادئ"]:
63+
assert kb.find_checkpoint(junk) is None, junk
64+
65+
66+
def test_short_token_not_distinctive_enough():
67+
# A <4-char token must not match as a distinctive whole-token (avoids the
68+
# "تل inside مقاتلو" class of false positive). "في" is 2 chars.
69+
kb = _kb()
70+
assert kb.find_checkpoint("شيء ما في مكان") is None

0 commit comments

Comments
 (0)