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

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._request import DetectionRequest 

12from chardet._utils import DEFAULT_MAX_BYTES 

13from chardet.enums import EncodingEra, LanguageFilter 

14from chardet.output_names import PREFERRED_SUPERSET 

15from chardet.pipeline import _NONE_RESULT, DetectionDict 

16 

17 

18class UniversalDetector: 

19 """Streaming character encoding detector. 

20 

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

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

23 

24 All detection is performed by the same pipeline used by 

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

26 consistent results regardless of which API is used. 

27 

28 .. note:: 

29 

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

31 :class:`UniversalDetector` instance. 

32 """ 

33 

34 MINIMUM_THRESHOLD = _utils.MINIMUM_THRESHOLD 

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

36 # reference UniversalDetector.LEGACY_MAP directly. 

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

38 PREFERRED_SUPERSET 

39 ) 

40 

41 def __init__( # noqa: PLR0913 

42 self, 

43 lang_filter: LanguageFilter = LanguageFilter.ALL, 

44 should_rename_legacy: bool = False, 

45 encoding_era: EncodingEra = EncodingEra.ALL, 

46 max_bytes: int = DEFAULT_MAX_BYTES, 

47 *, 

48 prefer_superset: bool = False, 

49 compat_names: bool = True, 

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

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

52 no_match_encoding: str = "cp1252", 

53 empty_input_encoding: str = "utf-8", 

54 ) -> None: 

55 """Initialize the detector. 

56 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

72 codec names. 

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

74 encodings (names or aliases). 

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

76 candidate set. 

77 :param no_match_encoding: Encoding to return when no candidate 

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

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

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

81 """ 

82 if lang_filter != LanguageFilter.ALL: 

83 warnings.warn( 

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

85 "and will be ignored", 

86 DeprecationWarning, 

87 stacklevel=2, 

88 ) 

89 self._request = DetectionRequest.resolve( 

90 should_rename_legacy=should_rename_legacy, 

91 encoding_era=encoding_era, 

92 max_bytes=max_bytes, 

93 prefer_superset=prefer_superset, 

94 compat_names=compat_names, 

95 include_encodings=include_encodings, 

96 exclude_encodings=exclude_encodings, 

97 no_match_encoding=no_match_encoding, 

98 empty_input_encoding=empty_input_encoding, 

99 stacklevel=3, 

100 ) 

101 self._buffer = bytearray() 

102 self._input_truncated = False 

103 self._done = False 

104 self._closed = False 

105 self._result: DetectionDict | None = None 

106 

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

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

109 

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

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

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

113 

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

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

116 :meth:`reset`. 

117 """ 

118 if self._closed: 

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

120 raise ValueError(msg) 

121 if self._done: 

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

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

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

125 if byte_str: 

126 self._input_truncated = True 

127 return 

128 remaining = self._request.max_bytes - len(self._buffer) 

129 if remaining > 0: 

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

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

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

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

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

135 self._input_truncated = True 

136 if len(self._buffer) >= self._request.max_bytes: 

137 self._done = True 

138 

139 def close(self) -> DetectionDict: 

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

141 

142 Runs the full detection pipeline on the buffered data. 

143 

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

145 ``"language"``, and ``"mime_type"``. 

146 """ 

147 if not self._closed: 

148 self._closed = True 

149 data = bytes(self._buffer) 

150 results = self._request.run(data, input_truncated=self._input_truncated) 

151 # The buffer never exceeds max_bytes, so it is the whole window. 

152 self._result = self._request.render(results, data)[0] 

153 self._done = True 

154 return self.result 

155 

156 def reset(self) -> None: 

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

158 self._buffer = bytearray() 

159 self._input_truncated = False 

160 self._done = False 

161 self._closed = False 

162 self._result = None 

163 

164 @property 

165 def done(self) -> bool: 

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

167 return self._done 

168 

169 @property 

170 def result(self) -> DetectionDict: 

171 """The current best detection result.""" 

172 if self._result is not None: 

173 # A fresh copy per read: callers may mutate what they get. 

174 return self._result.copy() 

175 return _NONE_RESULT.to_dict()