Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/chardet/_request.py: 70%
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"""One detection request: resolved options, the pipeline run, and rendering.
3:func:`chardet.detect`, :func:`chardet.detect_all`,
4:class:`chardet.UniversalDetector` and the ``chardetect`` CLI all ask the
5same question with the same options. :class:`DetectionRequest` owns that
6sequence once: validate and resolve the public keyword arguments
7(:meth:`DetectionRequest.resolve`), run the pipeline with them
8(:meth:`DetectionRequest.run`), and turn the ranked results into the dicts
9callers see (:meth:`DetectionRequest.render`). The public functions are
10thin adapters over it.
11"""
13from __future__ import annotations
15import dataclasses
16import warnings
17from collections.abc import Iterable
19from chardet._utils import DEFAULT_MAX_BYTES, MINIMUM_THRESHOLD
20from chardet.enums import EncodingEra
21from chardet.output_names import apply_compat_names, apply_preferred_superset
22from chardet.pipeline import DetectionDict, DetectionResult
23from chardet.pipeline.orchestrator import run_pipeline
24from chardet.registry import _validate_encoding, normalize_encodings
26#: Default chunk_size value (deprecated, kept for backward-compat signatures).
27_DEFAULT_CHUNK_SIZE: int = 65_536
30def _warn_deprecated_chunk_size(chunk_size: int, stacklevel: int = 3) -> None:
31 """Emit a deprecation warning if *chunk_size* differs from the default."""
32 if chunk_size != _DEFAULT_CHUNK_SIZE:
33 warnings.warn(
34 "chunk_size is not used in this version of chardet and will be ignored",
35 DeprecationWarning,
36 stacklevel=stacklevel,
37 )
40def _validate_max_bytes(max_bytes: int) -> None:
41 """Raise ValueError if *max_bytes* is not a positive integer."""
42 if isinstance(max_bytes, bool) or not isinstance(max_bytes, int) or max_bytes < 1:
43 msg = "max_bytes must be a positive integer"
44 raise ValueError(msg)
47def _resolve_prefer_superset(
48 *, should_rename_legacy: bool, prefer_superset: bool, stacklevel: int = 3
49) -> bool:
50 """Resolve the deprecated *should_rename_legacy* into *prefer_superset*."""
51 if should_rename_legacy:
52 warnings.warn(
53 "should_rename_legacy is deprecated, use prefer_superset instead",
54 DeprecationWarning,
55 stacklevel=stacklevel,
56 )
57 return True
58 return prefer_superset
61@dataclasses.dataclass(frozen=True, slots=True)
62class DetectionRequest:
63 """The resolved options of one detection, and the work they drive.
65 Build one with :meth:`resolve`, which validates the public keyword
66 arguments; the fields hold canonical values only.
67 """
69 encoding_era: EncodingEra = EncodingEra.ALL
70 max_bytes: int = DEFAULT_MAX_BYTES
71 prefer_superset: bool = False
72 compat_names: bool = True
73 include_encodings: frozenset[str] | None = None
74 exclude_encodings: frozenset[str] | None = None
75 no_match_encoding: str = "cp1252"
76 empty_input_encoding: str = "utf-8"
78 @classmethod
79 def resolve( # noqa: PLR0913
80 cls,
81 *,
82 should_rename_legacy: bool = False,
83 encoding_era: EncodingEra = EncodingEra.ALL,
84 chunk_size: int = _DEFAULT_CHUNK_SIZE,
85 max_bytes: int = DEFAULT_MAX_BYTES,
86 prefer_superset: bool = False,
87 compat_names: bool = True,
88 include_encodings: Iterable[str] | None = None,
89 exclude_encodings: Iterable[str] | None = None,
90 no_match_encoding: str = "cp1252",
91 empty_input_encoding: str = "utf-8",
92 stacklevel: int = 2,
93 ) -> DetectionRequest:
94 """Validate the public options and resolve them into a request.
96 Takes the keyword arguments exactly as :func:`chardet.detect`
97 receives them, deprecated ones included, and owns their validation
98 and deprecation warnings.
100 :param stacklevel: Where the deprecation warnings point, counted as
101 :func:`warnings.warn` counts it from the caller of this method:
102 ``2`` is that caller, ``3`` the caller's caller (what a public
103 function passes so the warning names the user's line).
104 :raises ValueError: If *max_bytes* is not a positive integer or an
105 encoding name is unknown.
106 """
107 # Each helper adds its own frame, hence the +1.
108 _warn_deprecated_chunk_size(chunk_size, stacklevel=stacklevel + 1)
109 _validate_max_bytes(max_bytes)
110 prefer_superset = _resolve_prefer_superset(
111 should_rename_legacy=should_rename_legacy,
112 prefer_superset=prefer_superset,
113 stacklevel=stacklevel + 1,
114 )
115 return cls(
116 encoding_era=encoding_era,
117 max_bytes=max_bytes,
118 prefer_superset=prefer_superset,
119 compat_names=compat_names,
120 include_encodings=normalize_encodings(
121 include_encodings, "include_encodings"
122 ),
123 exclude_encodings=normalize_encodings(
124 exclude_encodings, "exclude_encodings"
125 ),
126 no_match_encoding=_validate_encoding(
127 no_match_encoding, "no_match_encoding"
128 ),
129 empty_input_encoding=_validate_encoding(
130 empty_input_encoding, "empty_input_encoding"
131 ),
132 )
134 def run(
135 self,
136 data: bytes,
137 *,
138 full_ranking: bool = False,
139 input_truncated: bool = False,
140 ) -> list[DetectionResult]:
141 """Run the detection pipeline on *data* with this request's options.
143 :param data: The caller's bytes; the pipeline examines at most
144 :attr:`max_bytes` of them.
145 :param full_ranking: Rank every candidate, as ``detect_all`` needs.
146 :param input_truncated: *data* is already a cut-off view of the
147 caller's input (``UniversalDetector`` caps its buffer).
148 :returns: The ranked results, best first.
149 """
150 return run_pipeline(
151 data,
152 self.encoding_era,
153 max_bytes=self.max_bytes,
154 include_encodings=self.include_encodings,
155 exclude_encodings=self.exclude_encodings,
156 no_match_encoding=self.no_match_encoding,
157 empty_input_encoding=self.empty_input_encoding,
158 full_ranking=full_ranking,
159 input_truncated=input_truncated,
160 )
162 def render(
163 self,
164 results: list[DetectionResult],
165 window: bytes,
166 *,
167 all_results: bool = False,
168 ignore_threshold: bool = False,
169 ) -> list[DetectionDict]:
170 """Turn ranked results into the dicts callers see.
172 Applies the ``prefer_superset`` remap and then the display names,
173 each checked against *window* so the reported name decodes it.
174 That costs one whole-window decode for a remapped superset and one
175 for a display name in
176 :data:`~chardet.output_names.NARROWER_DISPLAY_NAMES`, none
177 otherwise.
179 :param results: Ranked results from :meth:`run`.
180 :param window: The bytes the verdict is about (``data[:max_bytes]``).
181 :param all_results: Render the whole ranking the way
182 :func:`chardet.detect_all` reports it: drop results at or below
183 :data:`~chardet.MINIMUM_THRESHOLD` unless that leaves nothing,
184 sort by confidence, and list each name once. When ``False``,
185 render only the top result.
186 :param ignore_threshold: With *all_results*, keep the results at or
187 below the threshold too.
188 :returns: The rendered dicts, best first.
189 """
190 if not all_results:
191 return [self._render_one(results[0].to_dict(), window)]
192 dicts = [r.to_dict() for r in results]
193 if not ignore_threshold:
194 filtered = [d for d in dicts if d["confidence"] > MINIMUM_THRESHOLD]
195 if filtered:
196 dicts = filtered
197 return self._rank_unique([self._render_one(d, window) for d in dicts])
199 def _render_one(self, d: DetectionDict, window: bytes) -> DetectionDict:
200 """Apply the output remaps to one result dict, in place."""
201 if self.prefer_superset:
202 apply_preferred_superset(
203 d,
204 window,
205 include_encodings=self.include_encodings,
206 exclude_encodings=self.exclude_encodings,
207 )
208 if self.compat_names:
209 apply_compat_names(d, window)
210 return d
212 @staticmethod
213 def _rank_unique(dicts: list[DetectionDict]) -> list[DetectionDict]:
214 """Sort by confidence and keep the first dict for each name."""
215 dicts.sort(key=lambda d: d["confidence"], reverse=True)
216 # The superset remap can land two candidates on one name (ISO-8859-7
217 # and Windows-1253 both become Windows-1253). Keep the first, which
218 # carries the higher confidence; the sort is stable, so ties keep the
219 # pipeline's order.
220 seen: set[str | None] = set()
221 unique: list[DetectionDict] = []
222 for d in dicts:
223 if d["encoding"] not in seen:
224 seen.add(d["encoding"])
225 unique.append(d)
226 return unique