1from __future__ import annotations
2
3from encodings.aliases import aliases
4from re import sub
5from typing import Any, Iterator, List, Tuple
6
7from .constant import RE_POSSIBLE_ENCODING_INDICATION, TOO_BIG_SEQUENCE
8from .utils import iana_name, is_multi_byte_encoding, unicode_range
9
10
11class CharsetMatch:
12 def __init__(
13 self,
14 payload: bytes | bytearray,
15 guessed_encoding: str,
16 mean_mess_ratio: float,
17 has_sig_or_bom: bool,
18 languages: CoherenceMatches,
19 decoded_payload: str | None = None,
20 preemptive_declaration: str | None = None,
21 ):
22 self._payload: bytes | bytearray = payload
23
24 self._encoding: str = guessed_encoding
25 self._mean_mess_ratio: float = mean_mess_ratio
26 self._languages: CoherenceMatches = languages
27 self._has_sig_or_bom: bool = has_sig_or_bom
28 self._unicode_ranges: list[str] | None = None
29
30 self._leaves: list[CharsetMatch] = []
31 self._mean_coherence_ratio: float = 0.0
32
33 self._output_payload: bytes | None = None
34 self._output_encoding: str | None = None
35
36 self._string: str | None = decoded_payload
37
38 self._preemptive_declaration: str | None = preemptive_declaration
39
40 def __eq__(self, other: object) -> bool:
41 if not isinstance(other, CharsetMatch):
42 if isinstance(other, str):
43 # Use non-strict iana_name so an operand that is not a known
44 # encoding alias compares unequal instead of raising, keeping
45 # __eq__ total as the data model requires (mirrors the lookup
46 # in CharsetMatches.__getitem__).
47 return iana_name(other, False) == self.encoding
48 return False
49 return self.encoding == other.encoding and self.fingerprint == other.fingerprint
50
51 def __lt__(self, other: object) -> bool:
52 """
53 Implemented to make sorted available upon CharsetMatches items.
54 """
55 if not isinstance(other, CharsetMatch):
56 raise ValueError
57
58 chaos_difference: float = abs(self.chaos - other.chaos)
59 coherence_difference: float = abs(self.coherence - other.coherence)
60
61 # Below 0.5% difference --> Use Coherence
62 if chaos_difference < 0.005 and coherence_difference > 0.02:
63 return self.coherence > other.coherence
64 elif chaos_difference < 0.005 and coherence_difference <= 0.02:
65 # When having a difficult decision, use the result that decoded as many multi-byte as possible.
66 # preserve RAM usage!
67 if len(self._payload) >= TOO_BIG_SEQUENCE:
68 return self.chaos < other.chaos
69 return self.multi_byte_usage > other.multi_byte_usage
70
71 return self.chaos < other.chaos
72
73 @property
74 def multi_byte_usage(self) -> float:
75 # Empty payloads are valid (from_bytes(b"") returns a match); avoid /0.
76 raw_len = len(self.raw)
77 if raw_len == 0:
78 return 0.0
79 return 1.0 - (len(str(self)) / raw_len)
80
81 def __str__(self) -> str:
82 # Lazy Str Loading
83 if self._string is None:
84 self._string = str(self._payload, self._encoding, "strict")
85 # UTF-7 BOM is encoded in modified Base64 whose byte boundary
86 # can overlap with the next character, so raw-byte stripping
87 # is unreliable. Strip the decoded BOM character instead.
88 # GB18030 also retains its signature when lazily decoded.
89 if (
90 self._has_sig_or_bom
91 and self._encoding in {"utf_7", "gb18030"}
92 and self._string
93 and self._string[0] == "\ufeff"
94 ):
95 self._string = self._string[1:]
96 return self._string
97
98 def __repr__(self) -> str:
99 return f"<CharsetMatch '{self.encoding}' fp({self.fingerprint})>"
100
101 def add_submatch(self, other: CharsetMatch) -> None:
102 if not isinstance(other, CharsetMatch) or other == self:
103 raise ValueError(
104 "Unable to add instance <{}> as a submatch of a CharsetMatch".format(
105 other.__class__
106 )
107 )
108
109 other._string = None # Unload RAM usage; dirty trick.
110 self._leaves.append(other)
111
112 @property
113 def encoding(self) -> str:
114 return self._encoding
115
116 @property
117 def encoding_aliases(self) -> list[str]:
118 """
119 Encoding name are known by many name, using this could help when searching for IBM855 when it's listed as CP855.
120 """
121 also_known_as: list[str] = []
122 for u, p in aliases.items():
123 if self.encoding == u:
124 also_known_as.append(p)
125 elif self.encoding == p:
126 also_known_as.append(u)
127 return also_known_as
128
129 @property
130 def bom(self) -> bool:
131 return self._has_sig_or_bom
132
133 @property
134 def byte_order_mark(self) -> bool:
135 return self._has_sig_or_bom
136
137 @property
138 def languages(self) -> list[str]:
139 """
140 Return the complete list of possible languages found in decoded sequence.
141 Usually not really useful. Returned list may be empty even if 'language' property return something != 'Unknown'.
142 """
143 return [e[0] for e in self._languages]
144
145 @property
146 def language(self) -> str:
147 """
148 Most probable language found in decoded sequence. If none were detected or inferred, the property will return
149 "Unknown".
150 """
151 if not self._languages:
152 # Trying to infer the language based on the given encoding
153 # Its either English or we should not pronounce ourselves in certain cases.
154 if "ascii" in self.could_be_from_charset:
155 return "English"
156
157 # doing it there to avoid circular import
158 from charset_normalizer.cd import encoding_languages, mb_encoding_languages
159
160 languages = (
161 mb_encoding_languages(self.encoding)
162 if is_multi_byte_encoding(self.encoding)
163 else encoding_languages(self.encoding)
164 )
165
166 if len(languages) == 0 or "Latin Based" in languages:
167 return "Unknown"
168
169 return languages[0]
170
171 return self._languages[0][0]
172
173 @property
174 def chaos(self) -> float:
175 return self._mean_mess_ratio
176
177 @property
178 def coherence(self) -> float:
179 if not self._languages:
180 return 0.0
181 return self._languages[0][1]
182
183 @property
184 def percent_chaos(self) -> float:
185 return round(self.chaos * 100, ndigits=3)
186
187 @property
188 def percent_coherence(self) -> float:
189 return round(self.coherence * 100, ndigits=3)
190
191 @property
192 def raw(self) -> bytes | bytearray:
193 """
194 Original untouched bytes.
195 """
196 return self._payload
197
198 @property
199 def submatch(self) -> list[CharsetMatch]:
200 return self._leaves
201
202 @property
203 def has_submatch(self) -> bool:
204 return len(self._leaves) > 0
205
206 @property
207 def alphabets(self) -> list[str]:
208 if self._unicode_ranges is not None:
209 return self._unicode_ranges
210 # list detected ranges
211 detected_ranges: list[str | None] = [unicode_range(char) for char in str(self)]
212 # filter and sort
213 self._unicode_ranges = sorted(list({r for r in detected_ranges if r}))
214 return self._unicode_ranges
215
216 @property
217 def could_be_from_charset(self) -> list[str]:
218 """
219 The complete list of encoding that output the exact SAME str result and therefore could be the originating
220 encoding.
221 This list does include the encoding available in property 'encoding'.
222 """
223 return [self._encoding] + [m.encoding for m in self._leaves]
224
225 def output(self, encoding: str = "utf_8") -> bytes:
226 """
227 Method to get re-encoded bytes payload using given target encoding. Default to UTF-8.
228 Any errors will be simply ignored by the encoder NOT replaced.
229 """
230 if self._output_encoding is None or self._output_encoding != encoding:
231 decoded_string = str(self)
232 if self._preemptive_declaration is not None and (
233 self._preemptive_declaration.lower() not in ["utf-8", "utf8", "utf_8"]
234 or iana_name(encoding, False) not in ["utf_8", "utf_8_sig"]
235 ):
236 patched_header = sub(
237 RE_POSSIBLE_ENCODING_INDICATION,
238 lambda m: m.string[m.span()[0] : m.span()[1]].replace(
239 m.groups()[0],
240 iana_name(encoding).replace("_", "-"),
241 ),
242 decoded_string[:8192],
243 count=1,
244 )
245
246 decoded_string = patched_header + decoded_string[8192:]
247
248 self._output_payload = decoded_string.encode(encoding, "replace")
249 self._output_encoding = encoding
250
251 return self._output_payload # type: ignore
252
253 @property
254 def fingerprint(self) -> int:
255 """
256 Retrieve a hash fingerprint of the decoded payload, used for deduplication.
257 """
258 return hash(str(self))
259
260
261class CharsetMatches:
262 """
263 Container with every CharsetMatch items ordered by default from most probable to the less one.
264 Act like a list(iterable) but does not implements all related methods.
265 """
266
267 def __init__(self, results: list[CharsetMatch] | None = None):
268 self._results: list[CharsetMatch] = sorted(results) if results else []
269 self._is_sorted: bool = True
270
271 def _ensure_sorted(self) -> None:
272 if not self._is_sorted:
273 self._results.sort()
274 self._is_sorted = True
275
276 def __iter__(self) -> Iterator[CharsetMatch]:
277 self._ensure_sorted()
278 yield from self._results
279
280 def __getitem__(self, item: int | str) -> CharsetMatch:
281 """
282 Retrieve a single item either by its position or encoding name (alias may be used here).
283 Raise KeyError upon invalid index or encoding not present in results.
284 """
285 if isinstance(item, int):
286 self._ensure_sorted()
287 return self._results[item]
288 if isinstance(item, str):
289 item = iana_name(item, False)
290 for result in self._results:
291 if item in result.could_be_from_charset:
292 return result
293 raise KeyError
294
295 def __len__(self) -> int:
296 return len(self._results)
297
298 def __bool__(self) -> bool:
299 return len(self._results) > 0
300
301 def append(self, item: CharsetMatch) -> None:
302 """
303 Insert a single match. Will be inserted accordingly to preserve sort.
304 Can be inserted as a submatch.
305 """
306 if not isinstance(item, CharsetMatch):
307 raise ValueError(
308 "Cannot append instance '{}' to CharsetMatches".format(
309 str(item.__class__)
310 )
311 )
312 # We should disable the submatch factoring when the input file is too heavy (conserve RAM usage)
313 if len(item.raw) < TOO_BIG_SEQUENCE:
314 for match in self._results:
315 if match.fingerprint == item.fingerprint and match.chaos == item.chaos:
316 match.add_submatch(item)
317 return
318 self._results.append(item)
319 self._is_sorted = False
320
321 def best(self) -> CharsetMatch | None:
322 """
323 Simply return the first match. Strict equivalent to matches[0].
324 """
325 if not self._results:
326 return None
327 self._ensure_sorted()
328 return self._results[0]
329
330 def first(self) -> CharsetMatch | None:
331 """
332 Redundant method, call the method best(). Kept for BC reasons.
333 """
334 return self.best()
335
336
337CoherenceMatch = Tuple[str, float]
338CoherenceMatches = List[CoherenceMatch]
339
340
341class CliDetectionResult:
342 def __init__(
343 self,
344 path: str,
345 encoding: str | None,
346 encoding_aliases: list[str],
347 alternative_encodings: list[str],
348 language: str,
349 alphabets: list[str],
350 has_sig_or_bom: bool,
351 chaos: float,
352 coherence: float,
353 unicode_path: str | None,
354 is_preferred: bool,
355 ):
356 self.path: str = path
357 self.unicode_path: str | None = unicode_path
358 self.encoding: str | None = encoding
359 self.encoding_aliases: list[str] = encoding_aliases
360 self.alternative_encodings: list[str] = alternative_encodings
361 self.language: str = language
362 self.alphabets: list[str] = alphabets
363 self.has_sig_or_bom: bool = has_sig_or_bom
364 self.chaos: float = chaos
365 self.coherence: float = coherence
366 self.is_preferred: bool = is_preferred
367
368 @property
369 def __dict__(self) -> dict[str, Any]: # type: ignore
370 return {
371 "path": self.path,
372 "encoding": self.encoding,
373 "encoding_aliases": self.encoding_aliases,
374 "alternative_encodings": self.alternative_encodings,
375 "language": self.language,
376 "alphabets": self.alphabets,
377 "has_sig_or_bom": self.has_sig_or_bom,
378 "chaos": self.chaos,
379 "coherence": self.coherence,
380 "unicode_path": self.unicode_path,
381 "is_preferred": self.is_preferred,
382 }
383
384 def to_json(self) -> str:
385 from json import dumps
386
387 return dumps(self.__dict__, ensure_ascii=True, indent=4)