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

1"""Internal shared utilities for chardet.""" 

2 

3from __future__ import annotations 

4 

5import codecs 

6from collections.abc import Callable 

7 

8#: Default maximum number of bytes to examine during detection. 

9DEFAULT_MAX_BYTES: int = 200_000 

10 

11#: Evidence cap: how much of the examination window the candidate-filtering, 

12#: validation, and probing stages consume before their answer is considered 

13#: converged (ADR-0006). Exhaustive checks (BOM, magic, UTF-8, ASCII, 

14#: binary, escape presence) take no cap. Must stay >= DEFAULT_MAX_BYTES so 

15#: every call using the default window is provably unaffected; a test 

16#: asserts the invariant. 

17EVIDENCE_CAP_BYTES: int = 256 * 1024 

18 

19#: Chunk size for whole-window validity decodes. Large enough that 

20#: per-chunk overhead vanishes, small enough that the transient decoded 

21#: ``str`` stays bounded regardless of input size. 

22_DECODE_CHUNK_SIZE: int = 1 << 20 

23 

24#: Default minimum confidence threshold for filtering results. 

25MINIMUM_THRESHOLD: float = 0.20 

26 

27#: Cache of incremental-decoder classes, keyed by encoding name. 

28#: 

29#: :func:`decodes_without_error` is called once per candidate encoding per 

30#: detection -- roughly 86 times for :attr:`~chardet.enums.EncodingEra.ALL` -- 

31#: and the factory lookup is a sixth of each call. The class is immutable and 

32#: shared; only the per-call *instance* carries decoder state, so caching the 

33#: class is thread-safe while caching an instance would not be. 

34#: 

35#: Only successful lookups are cached, which bounds the cache by the number of 

36#: installed codecs. A failed lookup returns ``False`` without storing 

37#: anything, so an unbounded stream of bogus names (a charset declaration in 

38#: markup is attacker-controlled) cannot grow it. 

39_INCREMENTAL_DECODERS: dict[str, Callable[..., codecs.IncrementalDecoder]] = {} 

40 

41 

42def _incremental_decoder( 

43 encoding: str, 

44) -> codecs.IncrementalDecoder | None: 

45 """Return a fresh incremental decoder for *encoding*, or ``None``. 

46 

47 The class lookup is cached; ``None`` means the codec does not exist. 

48 Every decode below catches ``UnicodeError`` rather than 

49 ``UnicodeDecodeError`` because the utf-16/utf-32 incremental decoders 

50 raise a bare ``UnicodeError`` when the stream does not start with a BOM. 

51 """ 

52 decoder_class = _INCREMENTAL_DECODERS.get(encoding) 

53 if decoder_class is None: 

54 try: 

55 decoder_class = codecs.getincrementaldecoder(encoding) 

56 except LookupError: 

57 return None 

58 _INCREMENTAL_DECODERS[encoding] = decoder_class 

59 return decoder_class() 

60 

61 

62def count_deleted(data: bytes, table: bytes) -> int: 

63 """Count how many bytes of *data* a deletion *table* would remove. 

64 

65 ``bytes.translate`` with a deletion table allocates an output buffer 

66 proportional to the input, so counting a whole large window through one 

67 call spikes memory by roughly the input size. Chunking keeps the 

68 transient bounded at one C call per chunk, and the count is identical: 

69 deletion is per-byte and carries no state across the split. 

70 

71 :param data: The raw byte data to scan. 

72 :param table: Byte values to delete. 

73 :returns: The number of bytes of *data* present in *table*. 

74 """ 

75 if len(data) <= _DECODE_CHUNK_SIZE: 

76 return len(data) - len(data.translate(None, table)) 

77 count = 0 

78 for pos in range(0, len(data), _DECODE_CHUNK_SIZE): 

79 chunk = data[pos : pos + _DECODE_CHUNK_SIZE] 

80 count += len(chunk) - len(chunk.translate(None, table)) 

81 return count 

82 

83 

84def decodes_without_error(data: bytes, encoding: str) -> bool: 

85 """Return ``True`` if *data* decodes cleanly under *encoding*. 

86 

87 Equivalent to ``data.decode(encoding, errors="strict")`` except that an 

88 incomplete multi-byte sequence at the *end* of *data* is accepted instead of 

89 raising. 

90 

91 Detection input is nearly always a prefix of a larger whole. Callers pass 

92 the first N bytes of a file, and chardet slices further on its own. 

93 For a two-byte encoding any of those cuts lands mid-character roughly half 

94 the time. 

95 

96 A one-shot strict decode cannot tell a truncated tail from corrupt data: it 

97 raises either way, so the candidate is discarded and every CJK encoding can 

98 disappear from the candidate set over a single dangling lead byte. An 

99 incremental decoder with ``final=False`` defers the partial tail instead, 

100 while still raising on genuine corruption anywhere before it. 

101 

102 Large inputs are fed in chunks and the decoded text is discarded, so a 

103 whole-window check never allocates a matching ``str``. Chunking is 

104 transparent: the decoder carries its state across calls, so a sequence 

105 straddling a chunk edge decodes exactly as it would in one call. 

106 

107 :param data: The raw byte data to test. 

108 :param encoding: Name of the codec to test *data* against. 

109 :returns: ``True`` if *data* decodes without error, ``False`` otherwise. 

110 """ 

111 decoder = _incremental_decoder(encoding) 

112 if decoder is None: 

113 return False 

114 try: 

115 if len(data) <= _DECODE_CHUNK_SIZE: 

116 decoder.decode(data, final=False) 

117 else: 

118 for pos in range(0, len(data), _DECODE_CHUNK_SIZE): 

119 decoder.decode(data[pos : pos + _DECODE_CHUNK_SIZE], final=False) 

120 except UnicodeError: 

121 return False 

122 return True 

123 

124 

125def decodes_completely(data: bytes, encoding: str) -> bool: 

126 """Return ``True`` if *data* decodes under *encoding* with nothing deferred. 

127 

128 The strict sibling of :func:`decodes_without_error`: ``final=True`` makes 

129 an incomplete multi-byte sequence at the end of *data* an error rather 

130 than a deferred tail. This is the question that matters when *data* is 

131 the caller's entire input --- ``data.decode(encoding)`` will make exactly 

132 this judgment. 

133 

134 :param data: The raw byte data to test. 

135 :param encoding: Name of the codec to test *data* against. 

136 :returns: ``True`` if *data* decodes completely, ``False`` otherwise. 

137 """ 

138 decoder = _incremental_decoder(encoding) 

139 if decoder is None: 

140 return False 

141 try: 

142 decoder.decode(data, final=True) 

143 except UnicodeError: 

144 return False 

145 return True 

146 

147 

148def dangling_tail_with_ascii_prefix(data: bytes, encoding: str) -> bool: 

149 """Return ``True`` if *data* is ASCII text plus an incomplete tail. 

150 

151 One decode pass answers both halves of the decode-safety question: 

152 the tolerant (``final=False``) decode yields the text before any 

153 deferred tail, and flushing the decoder afterwards raises exactly when 

154 a deferred tail existed. True means the candidate decoded real ASCII 

155 characters and then hit an incomplete multi-byte sequence at the end 

156 --- its only *non-ASCII* evidence is the undecodable tail itself. 

157 

158 Deliberately False when the tolerant decode yields nothing at all 

159 (the entire input is one dangling sequence): that candidate has zero 

160 decoded evidence, not ASCII evidence, and a clipped multi-byte 

161 fragment is better served by the ranking's own judgment. 

162 

163 :param data: The raw byte data to test. 

164 :param encoding: Name of the codec to decode *data* with. 

165 :returns: ``True`` if *data* decodes to non-empty pure ASCII with an 

166 incomplete sequence deferred at the end. 

167 """ 

168 decoder = _incremental_decoder(encoding) 

169 if decoder is None: 

170 return False 

171 try: 

172 text = decoder.decode(data, final=False) 

173 except UnicodeError: 

174 return False 

175 if not text or not text.isascii(): 

176 return False 

177 try: 

178 decoder.decode(b"", final=True) 

179 except UnicodeError: 

180 return True 

181 return False