Spaces:
Running
Running
File size: 5,881 Bytes
4b09d2d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """Structured data-quality warnings (spec 0001, design doc §12).
Every warning is an object - {code, severity, message, count?, affected_rows_sample?} -
never a bare string. Codes are UPPER_SNAKE and stable: downstream notebooks and the UI
key off them. Severity: "info" (status, not a problem) | "warning" (proceed with care).
Language detection is corpus-level only; short texts are exactly where detection is
unreliable, so uncertainty is reported instead of guessed away.
Deviation from spec 0001 (recorded there): langdetect instead of lingua - pure-Python,
~1 MB vs ~100 MB wheels; seeded for determinism. Upgrade path preserved by recording
detector + version in metadata.
"""
from __future__ import annotations
from dataclasses import dataclass
MIN_TOKENS_STABLE = 4 # texts below this are flagged TEXT_TOO_SHORT
DETECT_MIN_TOKENS = 5 # rows shorter than this are skipped for detection
DETECT_SAMPLE_MAX = 200 # rows sampled for corpus-level detection
DETECT_MIN_ROWS = 20 # fewer detectable rows -> LANGUAGE_UNCERTAIN
DETECT_CONFIDENCE = 0.70 # majority share below this -> LANGUAGE_UNCERTAIN
SAMPLE_ROWS_SHOWN = 5
def warning(code: str, severity: str, message: str, **extra) -> dict:
return {"code": code, "severity": severity, "message": message, **extra}
# ---------------------------------------------------------------- text QA
def short_text_warning(texts: list[str]) -> dict | None:
idx = [i for i, t in enumerate(texts) if len(t.split()) < MIN_TOKENS_STABLE]
if not idx:
return None
return warning(
"TEXT_TOO_SHORT",
"warning",
f"{len(idx)} text(s) contain fewer than {MIN_TOKENS_STABLE} words; "
"CCR scores may be unstable for very short texts.",
count=len(idx),
affected_rows_sample=idx[:SAMPLE_ROWS_SHOWN],
)
# ---------------------------------------------------------- language checks
@dataclass
class LanguageResult:
selected: str
detected: str | None
confidence: float | None
n_rows_sampled: int
detector: str
detector_version: str
def as_metadata(self) -> dict:
return {
"selected": self.selected,
"detected": self.detected,
"confidence": self.confidence,
"n_rows_sampled": self.n_rows_sampled,
"detector": self.detector,
"detector_version": self.detector_version,
}
def detect_corpus_language(texts: list[str], selected: str) -> tuple[LanguageResult, list[dict]]:
"""Corpus-level majority-vote detection on a sample of detectable rows."""
from langdetect import DetectorFactory, detect # lazy import
from langdetect.lang_detect_exception import LangDetectException
try:
from importlib.metadata import version as _v
detector_version = _v("langdetect")
except Exception:
detector_version = "unknown"
DetectorFactory.seed = 0 # determinism - same corpus, same result, every run
detectable = [t for t in texts if len(t.split()) >= DETECT_MIN_TOKENS][:DETECT_SAMPLE_MAX]
warnings: list[dict] = []
if len(detectable) < DETECT_MIN_ROWS:
result = LanguageResult(selected, None, None, len(detectable), "langdetect", detector_version)
warnings.append(
warning(
"LANGUAGE_UNCERTAIN",
"info",
f"Language could not be determined confidently ({len(detectable)} detectable "
f"row(s), need {DETECT_MIN_ROWS}); language checks were skipped.",
)
)
return result, warnings
votes: dict[str, int] = {}
for t in detectable:
try:
lang = detect(t) # one detection per row (detect() is the expensive call)
except LangDetectException:
continue
votes[lang] = votes.get(lang, 0) + 1
if not votes:
result = LanguageResult(selected, None, None, len(detectable), "langdetect", detector_version)
warnings.append(
warning("LANGUAGE_UNCERTAIN", "info",
"Language detection produced no result; language checks were skipped.")
)
return result, warnings
top_lang, top_count = max(votes.items(), key=lambda kv: kv[1])
confidence = round(top_count / sum(votes.values()), 3)
result = LanguageResult(selected, top_lang, confidence, len(detectable), "langdetect", detector_version)
if confidence < DETECT_CONFIDENCE:
warnings.append(
warning(
"LANGUAGE_UNCERTAIN",
"info",
f"Detected language is uncertain (top candidate '{top_lang}' at "
f"{confidence:.0%} of sampled rows); interpret language checks with care.",
)
)
elif top_lang != selected.lower():
warnings.append(
warning(
"LANGUAGE_MISMATCH",
"warning",
f"You selected '{selected}', but the corpus appears to be '{top_lang}' "
f"({confidence:.0%} of {len(detectable)} sampled rows).",
detected_language=top_lang,
selected_language=selected,
)
)
return result, warnings
def model_language_warning(selected: str, model_id: str, supported: frozenset[str],
language_set_name: str | None) -> dict | None:
if not supported or selected.lower() in supported:
return None
label = f"the '{language_set_name}' language set" if language_set_name else \
f"{sorted(supported)}"
return warning(
"MODEL_LANGUAGE_UNSUPPORTED",
"warning",
f"The selected model supports {label}, but you selected '{selected}'. "
"Switch to a multilingual model or proceed with caution.",
selected_language=selected,
model_id=model_id,
)
|