Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Pipeline orchestrator — runs all detection stages in sequence.
3Note: ``from __future__ import annotations`` is intentionally omitted because
4this module is compiled with mypyc, which does not support PEP 563 string
5annotations.
6"""
8import warnings
10from chardet._utils import DEFAULT_MAX_BYTES, EVIDENCE_CAP_BYTES, decodes_without_error
11from chardet.enums import EncodingEra
12from chardet.pipeline import (
13 _NONE_RESULT,
14 DETERMINISTIC_CONFIDENCE,
15 HIGH_BYTES,
16 DetectionResult,
17 PipelineContext,
18)
19from chardet.pipeline.ascii import detect_ascii
20from chardet.pipeline.binary import is_binary
21from chardet.pipeline.bom import detect_bom
22from chardet.pipeline.escape import detect_escape_encoding
23from chardet.pipeline.language import fill_languages
24from chardet.pipeline.magic import detect_magic
25from chardet.pipeline.markup import detect_markup_charset
26from chardet.pipeline.postprocess import postprocess_results
27from chardet.pipeline.statistical import score_candidates
28from chardet.pipeline.structural import (
29 compute_lead_byte_diversity,
30 compute_multibyte_byte_coverage,
31 compute_structural_score,
32)
33from chardet.pipeline.utf8 import scan_utf8
34from chardet.pipeline.utf1632 import detect_utf1632_patterns
35from chardet.pipeline.validity import filter_by_validity
36from chardet.registry import EncodingInfo, get_candidates
38_BINARY_RESULT = DetectionResult(
39 encoding=None,
40 confidence=DETERMINISTIC_CONFIDENCE,
41 language=None,
42 mime_type="application/octet-stream",
43)
44# Threshold at which a CJK structural score is confident enough to trigger
45# combined structural+statistical ranking rather than purely statistical.
46_STRUCTURAL_CONFIDENCE_THRESHOLD = 0.85
48# Maximum bytes used for statistical bigram scoring. Bigram models
49# converge quickly — 16 KB is sufficient for discrimination across all
50# language models (single-byte and multi-byte alike) while avoiding
51# unnecessary work on large files. Experimentally verified: 0 real
52# accuracy losses across 835 test files at this threshold.
53_STAT_SCORE_MAX_BYTES = 16384
56def _make_fallback_or_none(
57 encoding: str,
58 allowed: frozenset[str],
59 param_name: str,
60 stacklevel: int = 6,
61) -> list[DetectionResult]:
62 """Return a low-confidence result for *encoding*, or ``encoding=None`` if filtered out.
64 The default ``stacklevel=6`` targets the public caller: detect() ->
65 DetectionRequest.run() -> run_pipeline() -> _run_pipeline_core() ->
66 _make_fallback_or_none(). A caller one frame deeper
67 (:func:`_hold_validity_past_cap`) passes 7.
68 """
69 if encoding not in allowed:
70 warnings.warn(
71 f"{param_name} {encoding!r} is excluded by "
72 f"include_encodings/exclude_encodings; returning encoding=None",
73 UserWarning,
74 stacklevel=stacklevel,
75 )
76 return [_NONE_RESULT]
77 return [DetectionResult(encoding=encoding, confidence=0.10, language=None)]
80def _hold_validity_past_cap(
81 data: bytes,
82 evidence: bytes,
83 results: list[DetectionResult],
84 allowed: frozenset[str],
85 no_match_encoding: str,
86) -> list[DetectionResult]:
87 """Keep the validity contract for the part of the window past the evidence cap.
89 Byte-validity filtering converges on the first ``EVIDENCE_CAP_BYTES``
90 of the window (ADR-0006), so on a longer window a candidate that
91 cannot decode the rest can still reach the top. What callers rely on
92 is that the answer decodes the window chardet examined: they pass
93 ``max_bytes`` precisely to say how much of the input the verdict is
94 about. So walk the corrected ranking and drop every entry ahead of
95 the first one that decodes the whole window, under the same tolerant
96 judgment the validity filter passes on the evidence slice. Those
97 entries are what validity would have removed had it seen the bytes;
98 the survivors keep their own ranks and confidences, since nothing
99 about the evidence they were scored on has changed.
101 The walk is bounded by the ranking's length, each step is one C-speed
102 decode that stops at the first bad byte, and it costs nothing at all
103 when the window fits inside the cap, which every default call does.
104 Entries below the first decodable one are not re-examined, so a
105 ``detect_all`` ranking on a long window is exact down to its winner
106 and unverified past it. When no listed entry decodes, the answer is
107 the no-match fallback, as it is when validity leaves nothing.
108 """
109 if len(data) <= len(evidence):
110 return results
111 for i, r in enumerate(results):
112 if r.encoding is not None and decodes_without_error(data, r.encoding):
113 return results if i == 0 else results[i:]
114 return _make_fallback_or_none(
115 no_match_encoding, allowed, "no_match_encoding", stacklevel=7
116 )
119# Minimum structural score (valid multi-byte sequences / lead bytes) required
120# to keep a CJK multi-byte candidate. Below this threshold the encoding is
121# eliminated as a false positive (e.g. Shift_JIS matching Latin data where
122# scattered high bytes look like lead bytes but rarely form valid pairs).
123_CJK_MIN_MB_RATIO = 0.05
124# Minimum number of non-ASCII bytes required for a CJK candidate to survive
125# gating. Very short inputs are validated by the other gates (structural
126# pair ratio, byte coverage) and by coverage-aware boosting in statistical
127# scoring — so we keep this threshold low to let even 1-character CJK
128# inputs compete.
129_CJK_MIN_NON_ASCII = 2
130# Minimum ratio of non-ASCII bytes that must participate in valid multi-byte
131# sequences for a CJK candidate to survive gating. Genuine CJK text has
132# nearly all non-ASCII bytes in valid pairs (coverage >= 0.95); Latin text
133# with scattered high bytes has many orphan bytes (coverage often < 0.5).
134# The lowest true-positive coverage in the test suite is ~0.39 (a CP932 HTML
135# file with many half-width katakana).
136_CJK_MIN_BYTE_COVERAGE = 0.35
137# Minimum number of distinct lead byte values for a CJK candidate to
138# survive gating. Genuine CJK text uses a wide range of lead bytes;
139# European false positives cluster in a narrow band. Only applied when
140# there are enough non-ASCII bytes to expect diversity (see
141# _CJK_DIVERSITY_MIN_NON_ASCII).
142_CJK_MIN_LEAD_DIVERSITY = 4
143# Minimum non-ASCII byte count before applying the lead diversity gate.
144# Very small files (e.g. 8 non-ASCII bytes) may genuinely have low
145# diversity even for real CJK text (e.g. repeated katakana).
146_CJK_DIVERSITY_MIN_NON_ASCII = 16
149def _gate_cjk_candidates(
150 data: bytes,
151 valid_candidates: tuple[EncodingInfo, ...],
152 ctx: PipelineContext,
153) -> tuple[EncodingInfo, ...]:
154 """Eliminate CJK multi-byte candidates that lack genuine multi-byte structure.
156 Four checks are applied in order to each multi-byte candidate:
158 1. **Structural pair ratio** (valid_pairs / lead_bytes) must be
159 >= ``_CJK_MIN_MB_RATIO``. Catches files with many orphan lead bytes.
161 2. **Minimum non-ASCII byte count**: the data must contain at least
162 ``_CJK_MIN_NON_ASCII`` bytes > 0x7F. Tiny files with 1-5 high bytes
163 can accidentally form perfect pairs and score 1.0 structurally.
165 3. **Byte coverage** (non-ASCII bytes in valid multi-byte sequences /
166 total non-ASCII bytes) must be >= ``_CJK_MIN_BYTE_COVERAGE``. Latin
167 text has many high bytes that are NOT consumed by multi-byte pairs;
168 genuine CJK text has nearly all high bytes accounted for.
170 4. **Lead byte diversity**: the number of distinct lead byte values in
171 valid pairs must be >= ``_CJK_MIN_LEAD_DIVERSITY``. Genuine CJK text
172 draws from a wide repertoire of lead bytes; European false positives
173 cluster in a narrow band (e.g. 0xC0-0xDF for accented Latin).
175 Returns the filtered candidate list. Structural scores are cached in
176 ``ctx.mb_scores`` for reuse in Stage 2b.
177 """
178 gated: list[EncodingInfo] = []
179 for enc in valid_candidates:
180 if enc.is_multibyte_cjk:
181 mb_score = compute_structural_score(data, enc, ctx)
182 ctx.mb_scores[enc.name] = mb_score
183 if mb_score < _CJK_MIN_MB_RATIO:
184 continue # No multi-byte structure -> eliminate
185 if ctx.non_ascii_count is None:
186 ctx.non_ascii_count = len(data) - len(data.translate(None, HIGH_BYTES))
187 if ctx.non_ascii_count < _CJK_MIN_NON_ASCII:
188 continue # Too few high bytes to trust the score
189 byte_coverage = compute_multibyte_byte_coverage(
190 data, enc, ctx, non_ascii_count=ctx.non_ascii_count
191 )
192 ctx.mb_coverage[enc.name] = byte_coverage
193 if byte_coverage < _CJK_MIN_BYTE_COVERAGE:
194 continue # Most high bytes are orphans -> not CJK
195 if ctx.non_ascii_count >= _CJK_DIVERSITY_MIN_NON_ASCII:
196 lead_diversity = compute_lead_byte_diversity(data, enc, ctx)
197 if lead_diversity < _CJK_MIN_LEAD_DIVERSITY:
198 continue # Too few distinct lead bytes -> not CJK
199 gated.append(enc)
200 return tuple(gated)
203def _score_structural_candidates(
204 data: bytes,
205 structural_scores: list[tuple[str, float]],
206 valid_candidates: tuple[EncodingInfo, ...],
207 ctx: PipelineContext,
208 *,
209 full_ranking: bool = False,
210) -> list[DetectionResult]:
211 """Score structurally-valid CJK candidates using statistical bigrams.
213 When multiple CJK encodings score equally high structurally, statistical
214 scoring differentiates them (e.g. euc-jp vs big5 for Japanese data).
215 Single-byte candidates are also scored and included so that the caller
216 can compare CJK vs single-byte confidence.
218 Multi-byte candidates with high byte coverage (>= 0.95) receive a
219 confidence boost proportional to coverage. When nearly all non-ASCII
220 bytes form valid multi-byte pairs, the structural evidence is strong
221 and should increase the candidate's ranking relative to single-byte
222 alternatives whose bigram models may score higher on small samples.
224 Note: boosted confidence values may exceed 1.0 and are used only for
225 relative ranking among candidates. ``run_pipeline`` clamps all
226 confidence values to [0.0, 1.0] before returning to callers.
227 """
228 enc_lookup: dict[str, EncodingInfo] = {
229 e.name: e for e in valid_candidates if e.is_multibyte_cjk
230 }
231 valid_mb = tuple(
232 enc_lookup[name] for name, _sc in structural_scores if name in enc_lookup
233 )
234 single_byte = tuple(e for e in valid_candidates if not e.is_multibyte_cjk)
235 results = list(
236 score_candidates(
237 data[:_STAT_SCORE_MAX_BYTES],
238 (*valid_mb, *single_byte),
239 full_ranking=full_ranking,
240 )
241 )
243 # Boost multi-byte candidates with high byte coverage.
244 boosted: list[DetectionResult] = []
245 for r in results:
246 coverage = ctx.mb_coverage.get(r.encoding, 0.0) if r.encoding else 0.0
247 if coverage >= 0.95:
248 boosted.append(
249 DetectionResult(
250 r.encoding, r.confidence * (1 + coverage), r.language, r.mime_type
251 )
252 )
253 else:
254 boosted.append(r)
255 boosted.sort(key=lambda x: x.confidence, reverse=True)
256 return boosted
259def _with_default_mime(result: DetectionResult) -> DetectionResult:
260 """Default ``mime_type`` to ``text/plain`` (text) or ``application/octet-stream`` (binary)."""
261 if result.mime_type is not None:
262 return result
263 mime = "text/plain" if result.encoding is not None else "application/octet-stream"
264 return DetectionResult(result.encoding, result.confidence, result.language, mime)
267def _run_pipeline_core( # noqa: PLR0913
268 data: bytes,
269 encoding_era: EncodingEra,
270 max_bytes: int = DEFAULT_MAX_BYTES,
271 *,
272 include_encodings: frozenset[str] | None = None,
273 exclude_encodings: frozenset[str] | None = None,
274 no_match_encoding: str = "cp1252",
275 empty_input_encoding: str = "utf-8",
276 full_ranking: bool = False,
277 input_truncated: bool = False,
278) -> list[DetectionResult]:
279 """Core pipeline logic. Returns list of results sorted by confidence."""
280 ctx = PipelineContext()
281 input_truncated = input_truncated or len(data) > max_bytes
282 data = data[:max_bytes]
284 # Build candidate set once — used for both early-exit gating and
285 # statistical scoring. The set incorporates encoding_era, include, and
286 # exclude filters so all pipeline stages are gated consistently.
287 candidates = get_candidates(encoding_era, include_encodings, exclude_encodings)
288 allowed: frozenset[str] = frozenset(enc.name for enc in candidates)
290 if not data:
291 return _make_fallback_or_none(
292 empty_input_encoding, allowed, "empty_input_encoding"
293 )
295 # Stage 1a: BOM detection (runs first — BOMs are definitive and
296 # UTF-16/32 data looks binary due to null bytes)
297 bom_result = detect_bom(data, input_truncated=input_truncated)
298 if bom_result is not None and bom_result.encoding in allowed:
299 return [bom_result]
301 # Stage 1a+: UTF-16/32 null-byte pattern detection (for files without
302 # BOMs — must run before binary detection since these encodings contain
303 # many null bytes that would trigger the binary check)
304 utf1632_result = detect_utf1632_patterns(data)
305 # The stage samples and decodes the first few KB only. The answer has
306 # to decode the whole window, so a lone surrogate or out-of-range unit
307 # past the sample sends the input on to the other stages instead.
308 if (
309 utf1632_result is not None
310 and utf1632_result.encoding is not None
311 and utf1632_result.encoding in allowed
312 and decodes_without_error(data, utf1632_result.encoding)
313 ):
314 return [utf1632_result]
316 # Escape-sequence encodings (ISO-2022, HZ-GB-2312, UTF-7): must run
317 # before binary detection (ESC is a control byte) and before ASCII
318 # detection (HZ-GB-2312 uses only printable ASCII plus tildes).
319 escape_result = detect_escape_encoding(data)
320 if (
321 escape_result is not None
322 and escape_result.encoding is not None
323 and escape_result.encoding in allowed
324 ):
325 return [escape_result]
327 # Magic number detection for known binary formats — runs before
328 # UTF-8/ASCII prechecks to avoid unnecessary analysis on binary data.
329 magic_result = detect_magic(data)
330 if magic_result is not None:
331 return [magic_result]
333 # Pre-check UTF-8 to prevent false binary classification. Valid UTF-8
334 # with multi-byte sequences can contain control bytes (e.g. ESC for ANSI
335 # codes) that would otherwise exceed the binary threshold. We compute
336 # the result now but return it at the normal pipeline position (after
337 # markup) so that explicit charset declarations still take precedence.
338 utf8_valid, utf8_precheck = scan_utf8(data)
340 # Pre-check ASCII to prevent false binary classification. ASCII text
341 # with null byte separators (e.g. find -print0 output) would exceed the
342 # binary threshold due to the null bytes. Like the UTF-8 precheck, we
343 # compute the result now but return it at the normal position (after
344 # markup) so explicit charset declarations still take precedence.
345 ascii_precheck = detect_ascii(data)
347 # Stage 0: Binary detection (skip when data is valid UTF-8 or ASCII)
348 # Binary detection (encoding=None) is NOT gated by filters.
349 if (
350 utf8_precheck is None
351 and ascii_precheck is None
352 and is_binary(data, max_bytes=max_bytes)
353 ):
354 return [_BINARY_RESULT]
356 # Stage 1b: Markup charset extraction. The stage owns the whole
357 # declaration decision: decode checks, superset promotion, the allowed
358 # gate, and the UTF-8 override (see its docstring).
359 markup_result = detect_markup_charset(data, allowed, utf8_precheck)
360 if markup_result is not None:
361 return [markup_result]
363 # Stage 1c: ASCII (use pre-computed result)
364 if ascii_precheck is not None and ascii_precheck.encoding in allowed:
365 return [ascii_precheck]
367 # Stage 1d: UTF-8 structural validation (use pre-computed result)
368 if utf8_precheck is not None and utf8_precheck.encoding in allowed:
369 return [utf8_precheck]
371 # The filtering, gating, and probing stages below converge on bounded
372 # evidence (ADR-0006). One slice, taken here, feeds them all: the
373 # structural analysis cache is keyed by encoding name only, so every
374 # consumer must see the same view of the data. For the decode-safety
375 # flip in postprocess, a window that outruns the cap counts as
376 # truncation — the slice's tail is a chardet-made cut, not the end of
377 # what the caller will decode.
378 evidence = data[:EVIDENCE_CAP_BYTES]
379 evidence_truncated = input_truncated or len(data) > len(evidence)
381 # Stage 2a: Byte validity filtering
382 valid_candidates = filter_by_validity(evidence, candidates)
384 # The exhaustive UTF-8 check saw the whole window; when it *rejected*
385 # the data but the window extends past the evidence cap, the slice
386 # alone may still look like valid UTF-8 to the statistical path. Honor
387 # the proof: chardet never calls data UTF-8 that is not valid UTF-8
388 # throughout the window. A missing precheck result is not a rejection
389 # — valid UTF-8 with no multi-byte evidence (pure ASCII, ASCII plus
390 # control bytes, ASCII plus a truncated tail) also returns None, and
391 # ruling utf-8 out for those would be wrong. (Within the cap the slice
392 # is the window, so validity already agrees and this never fires.)
393 if not utf8_valid and len(data) > len(evidence):
394 valid_candidates = tuple(e for e in valid_candidates if e.name != "utf-8")
396 if not valid_candidates:
397 return _make_fallback_or_none(no_match_encoding, allowed, "no_match_encoding")
399 # Gate: eliminate CJK multi-byte candidates that lack genuine
400 # multi-byte structure. Cache structural scores for Stage 2b.
401 valid_candidates = _gate_cjk_candidates(evidence, valid_candidates, ctx)
403 if not valid_candidates:
404 return _make_fallback_or_none(no_match_encoding, allowed, "no_match_encoding")
406 # Stage 2b: Structural probing for multi-byte encodings
407 # Reuse scores already computed during the CJK gate above.
408 structural_scores: list[tuple[str, float]] = []
409 for enc in valid_candidates:
410 if enc.is_multibyte_cjk:
411 score = ctx.mb_scores.get(enc.name)
412 if score is None: # pragma: no cover - gate always populates cache
413 score = compute_structural_score(evidence, enc, ctx)
414 if score > 0.0:
415 structural_scores.append((enc.name, score))
417 # If a multi-byte encoding scored very high, score all candidates
418 # (CJK + single-byte) statistically.
419 if structural_scores:
420 structural_scores.sort(key=lambda x: x[1], reverse=True)
421 _, best_score = structural_scores[0]
422 if best_score >= _STRUCTURAL_CONFIDENCE_THRESHOLD:
423 results = _score_structural_candidates(
424 evidence,
425 structural_scores,
426 valid_candidates,
427 ctx,
428 full_ranking=full_ranking,
429 )
430 if results:
431 results = postprocess_results(
432 evidence, results, input_truncated=evidence_truncated
433 )
434 return _hold_validity_past_cap(
435 data, evidence, results, allowed, no_match_encoding
436 )
438 # Stage 3: Statistical scoring for all remaining candidates.
439 # Bigram models converge quickly and don't benefit from scanning
440 # beyond 16 KB — cap the data to avoid unnecessary work on large files.
441 stat_data = evidence[:_STAT_SCORE_MAX_BYTES]
442 results = list(
443 score_candidates(stat_data, tuple(valid_candidates), full_ranking=full_ranking)
444 )
445 if not results:
446 return _make_fallback_or_none(no_match_encoding, allowed, "no_match_encoding")
448 # Rank corrections reason about the same evidence window the ranking
449 # came from, which also bounds their byte-presence scans.
450 results = postprocess_results(evidence, results, input_truncated=evidence_truncated)
451 return _hold_validity_past_cap(data, evidence, results, allowed, no_match_encoding)
454def run_pipeline( # noqa: PLR0913
455 data: bytes,
456 encoding_era: EncodingEra,
457 max_bytes: int = DEFAULT_MAX_BYTES,
458 *,
459 include_encodings: frozenset[str] | None = None,
460 exclude_encodings: frozenset[str] | None = None,
461 no_match_encoding: str = "cp1252",
462 empty_input_encoding: str = "utf-8",
463 full_ranking: bool = False,
464 input_truncated: bool = False,
465) -> list[DetectionResult]:
466 """Run the full detection pipeline.
468 :param data: The raw byte data to analyze.
469 :param encoding_era: Filter candidates to a specific era of encodings.
470 :param max_bytes: Maximum number of bytes to process.
471 :param include_encodings: If not ``None``, only return these encodings.
472 :param exclude_encodings: If not ``None``, never return these encodings.
473 :param no_match_encoding: Encoding returned when no candidate survives.
474 :param empty_input_encoding: Encoding returned for empty input.
475 :param full_ranking: When ``True``, statistical scoring evaluates every
476 candidate so the returned list is complete (``detect_all`` needs
477 this). When ``False``, candidates that provably cannot affect the
478 top of the ranking may be omitted from the tail.
479 :param input_truncated: Pass ``True`` when *data* is already a truncated
480 view of the caller's input (``UniversalDetector`` caps its buffer at
481 ``max_bytes``, which this function cannot see from ``len(data)``
482 alone). Truncation by the ``max_bytes`` slice here is detected
483 either way; the flag only ever widens it.
484 :returns: A list of :class:`DetectionResult` sorted by confidence descending.
485 """
486 results = _run_pipeline_core(
487 data,
488 encoding_era,
489 max_bytes,
490 input_truncated=input_truncated,
491 include_encodings=include_encodings,
492 exclude_encodings=exclude_encodings,
493 no_match_encoding=no_match_encoding,
494 empty_input_encoding=empty_input_encoding,
495 full_ranking=full_ranking,
496 )
497 # Same slice the core ran on: ``max_bytes`` is the caller's cap on what
498 # chardet may look at, and language fill is not exempt from it just
499 # because it applies a smaller cap of its own.
500 results = fill_languages(data[:max_bytes], results)
501 # The ANSI-art model is keyed under the "zxx" pseudo-language (ISO 639
502 # for "no linguistic content"). Kept internal so language fill does not
503 # overwrite it; callers see language=None.
504 results = [
505 DetectionResult(r.encoding, r.confidence, None, r.mime_type)
506 if r.language == "zxx"
507 else r
508 for r in results
509 ]
510 results = [_with_default_mime(r) for r in results]
511 if not results: # pragma: no cover
512 msg = "pipeline must always return at least one result"
513 raise RuntimeError(msg)
514 # Clamp confidence to [0.0, 1.0] at the public API boundary. Internal
515 # stages may boost confidence above 1.0 for ranking purposes (e.g.
516 # CJK byte-coverage boost), but callers expect a probability-like value.
517 return [
518 DetectionResult(r.encoding, min(r.confidence, 1.0), r.language, r.mime_type)
519 if r.confidence > 1.0
520 else r
521 for r in results
522 ]