Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/chardet/detector.py: 32%

1"""UniversalDetector — streaming encoding detection.""" 

2 

3from __future__ import annotations 

4 

5import warnings 

6from collections.abc import Iterable 

7from types import MappingProxyType 

8from typing import ClassVar 

9 

10from chardet import _utils 

11from chardet._utils import ( 

12 DEFAULT_MAX_BYTES, 

13 _resolve_prefer_superset, 

14 _validate_max_bytes, 

15) 

16from chardet.enums import EncodingEra, LanguageFilter 

17from chardet.output_names import ( 

18 PREFERRED_SUPERSET, 

19 apply_compat_names, 

20 apply_preferred_superset, 

21) 

22from chardet.pipeline import _NONE_RESULT, DetectionDict, DetectionResult 

23from chardet.pipeline.orchestrator import run_pipeline 

24from chardet.registry import _validate_encoding, normalize_encodings 

25 

26 

27class UniversalDetector: 

28 """Streaming character encoding detector. 

29 

30 Implements a feed/close pattern for incremental detection of character 

31 encoding from byte streams. Compatible with the chardet 6.x API. 

32 

33 All detection is performed by the same pipeline used by 

34 :func:`chardet.detect` and :func:`chardet.detect_all`, ensuring 

35 consistent results regardless of which API is used. 

36 

37 .. note:: 

38 

39 This class is **not** thread-safe. Each thread should create its own 

40 :class:`UniversalDetector` instance. 

41 """ 

42 

43 MINIMUM_THRESHOLD = _utils.MINIMUM_THRESHOLD 

44 # Exposed for backward compatibility with chardet 6.x callers that 

45 # reference UniversalDetector.LEGACY_MAP directly. 

46 LEGACY_MAP: ClassVar[MappingProxyType[str, str]] = MappingProxyType( 

47 PREFERRED_SUPERSET 

48 ) 

49 

50 def __init__( # noqa: PLR0913 

51 self, 

52 lang_filter: LanguageFilter = LanguageFilter.ALL, 

53 should_rename_legacy: bool = False, 

54 encoding_era: EncodingEra = EncodingEra.ALL, 

55 max_bytes: int = DEFAULT_MAX_BYTES, 

56 *, 

57 prefer_superset: bool = False, 

58 compat_names: bool = True, 

59 include_encodings: Iterable[str] | None = None, 

60 exclude_encodings: Iterable[str] | None = None, 

61 no_match_encoding: str = "cp1252", 

62 empty_input_encoding: str = "utf-8", 

63 ) -> None: 

64 """Initialize the detector. 

65 

66 :param lang_filter: Deprecated -- accepted for backward compatibility 

67 but has no effect. A warning is emitted when set to anything 

68 other than :attr:`LanguageFilter.ALL`. 

69 :param should_rename_legacy: Deprecated alias for *prefer_superset*. 

70 :param encoding_era: Restrict candidate encodings to the given era. 

71 :param max_bytes: Maximum number of bytes to buffer from 

72 :meth:`feed` calls before stopping accumulation. 

73 :param prefer_superset: If ``True``, remap subset encodings in the 

74 result to their decode-safe Windows/CP superset equivalents 

75 (e.g., ISO-8859-1 -> Windows-1252). If ``False`` (default), 

76 skip the renaming --- not a promise of the smallest matching 

77 encoding. The default will change to ``True`` in chardet 8.0. 

78 See :func:`chardet.detect` for details. 

79 :param compat_names: If ``True`` (default), return encoding names 

80 compatible with chardet 5.x/6.x. If ``False``, return raw Python 

81 codec names. 

82 :param include_encodings: If given, restrict detection to only these 

83 encodings (names or aliases). 

84 :param exclude_encodings: If given, remove these encodings from the 

85 candidate set. 

86 :param no_match_encoding: Encoding to return when no candidate 

87 survives the pipeline. Defaults to ``"cp1252"``. 

88 :param empty_input_encoding: Encoding to return for empty input. 

89 Defaults to ``"utf-8"``. 

90 """ 

91 if lang_filter != LanguageFilter.ALL: 

92 warnings.warn( 

93 "lang_filter is not implemented in this version of chardet " 

94 "and will be ignored", 

95 DeprecationWarning, 

96 stacklevel=2, 

97 ) 

98 prefer_superset = _resolve_prefer_superset( 

99 should_rename_legacy, prefer_superset 

100 ) 

101 self._prefer_superset = prefer_superset 

102 self._compat_names = compat_names 

103 _validate_max_bytes(max_bytes) 

104 self._encoding_era = encoding_era 

105 self._max_bytes = max_bytes 

106 self._include_encodings = normalize_encodings( 

107 include_encodings, "include_encodings" 

108 ) 

109 self._exclude_encodings = normalize_encodings( 

110 exclude_encodings, "exclude_encodings" 

111 ) 

112 self._no_match_encoding = _validate_encoding( 

113 no_match_encoding, "no_match_encoding" 

114 ) 

115 self._empty_input_encoding = _validate_encoding( 

116 empty_input_encoding, "empty_input_encoding" 

117 ) 

118 self._buffer = bytearray() 

119 self._input_truncated = False 

120 self._done = False 

121 self._closed = False 

122 self._result: DetectionResult | None = None 

123 

124 def feed(self, byte_str: bytes | bytearray) -> None: 

125 """Feed a chunk of bytes to the detector. 

126 

127 Data is accumulated in an internal buffer. Once *max_bytes* have 

128 been buffered, :attr:`done` is set to ``True`` and further data is 

129 ignored until :meth:`reset` is called. 

130 

131 :param byte_str: The next chunk of bytes to examine. 

132 :raises ValueError: If called after :meth:`close` without a 

133 :meth:`reset`. 

134 """ 

135 if self._closed: 

136 msg = "feed() called after close() without reset()" 

137 raise ValueError(msg) 

138 if self._done: 

139 # The buffer already holds max_bytes exactly, which the pipeline 

140 # cannot tell apart from an input that was max_bytes long; only 

141 # this flag says these bytes were cut off it. 

142 if byte_str: 

143 self._input_truncated = True 

144 return 

145 remaining = self._max_bytes - len(self._buffer) 

146 if remaining > 0: 

147 self._buffer.extend(byte_str[:remaining]) 

148 if len(byte_str) > max(remaining, 0): 

149 # Bytes beyond the cap were dropped: the buffer is a truncated 

150 # view of the caller's stream, which the pipeline cannot infer 

151 # from the buffer length alone (it equals max_bytes exactly). 

152 self._input_truncated = True 

153 if len(self._buffer) >= self._max_bytes: 

154 self._done = True 

155 

156 def close(self) -> DetectionDict: 

157 """Finalize detection and return the best result. 

158 

159 Runs the full detection pipeline on the buffered data. 

160 

161 :returns: A dictionary with keys ``"encoding"``, ``"confidence"``, 

162 and ``"language"``. 

163 """ 

164 if not self._closed: 

165 self._closed = True 

166 data = bytes(self._buffer) 

167 results = run_pipeline( 

168 data, 

169 self._encoding_era, 

170 max_bytes=self._max_bytes, 

171 input_truncated=self._input_truncated, 

172 include_encodings=self._include_encodings, 

173 exclude_encodings=self._exclude_encodings, 

174 no_match_encoding=self._no_match_encoding, 

175 empty_input_encoding=self._empty_input_encoding, 

176 ) 

177 self._result = results[0] 

178 self._done = True 

179 return self.result 

180 

181 def reset(self) -> None: 

182 """Reset the detector to its initial state for reuse.""" 

183 self._buffer = bytearray() 

184 self._input_truncated = False 

185 self._done = False 

186 self._closed = False 

187 self._result = None 

188 

189 @property 

190 def done(self) -> bool: 

191 """Whether detection is complete and no more data is needed.""" 

192 return self._done 

193 

194 @property 

195 def result(self) -> DetectionDict: 

196 """The current best detection result.""" 

197 if self._result is not None: 

198 d = self._result.to_dict() 

199 if self._prefer_superset: 

200 apply_preferred_superset(d, bytes(self._buffer)) 

201 if self._compat_names: 

202 apply_compat_names(d) 

203 return d 

204 return _NONE_RESULT.to_dict()