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 if (
89 self._has_sig_or_bom
90 and self._encoding == "utf_7"
91 and self._string
92 and self._string[0] == "\ufeff"
93 ):
94 self._string = self._string[1:]
95 return self._string
96
97 def __repr__(self) -> str:
98 return f"<CharsetMatch '{self.encoding}' fp({self.fingerprint})>"
99
100 def add_submatch(self, other: CharsetMatch) -> None:
101 if not isinstance(other, CharsetMatch) or other == self:
102 raise ValueError(
103 "Unable to add instance <{}> as a submatch of a CharsetMatch".format(
104 other.__class__
105 )
106 )
107
108 other._string = None # Unload RAM usage; dirty trick.
109 self._leaves.append(other)
110
111 @property
112 def encoding(self) -> str:
113 return self._encoding
114
115 @property
116 def encoding_aliases(self) -> list[str]:
117 """
118 Encoding name are known by many name, using this could help when searching for IBM855 when it's listed as CP855.
119 """
120 also_known_as: list[str] = []
121 for u, p in aliases.items():
122 if self.encoding == u:
123 also_known_as.append(p)
124 elif self.encoding == p:
125 also_known_as.append(u)
126 return also_known_as
127
128 @property
129 def bom(self) -> bool:
130 return self._has_sig_or_bom
131
132 @property
133 def byte_order_mark(self) -> bool:
134 return self._has_sig_or_bom
135
136 @property
137 def languages(self) -> list[str]:
138 """
139 Return the complete list of possible languages found in decoded sequence.
140 Usually not really useful. Returned list may be empty even if 'language' property return something != 'Unknown'.
141 """
142 return [e[0] for e in self._languages]
143
144 @property
145 def language(self) -> str:
146 """
147 Most probable language found in decoded sequence. If none were detected or inferred, the property will return
148 "Unknown".
149 """
150 if not self._languages:
151 # Trying to infer the language based on the given encoding
152 # Its either English or we should not pronounce ourselves in certain cases.
153 if "ascii" in self.could_be_from_charset:
154 return "English"
155
156 # doing it there to avoid circular import
157 from charset_normalizer.cd import encoding_languages, mb_encoding_languages
158
159 languages = (
160 mb_encoding_languages(self.encoding)
161 if is_multi_byte_encoding(self.encoding)
162 else encoding_languages(self.encoding)
163 )
164
165 if len(languages) == 0 or "Latin Based" in languages:
166 return "Unknown"
167
168 return languages[0]
169
170 return self._languages[0][0]
171
172 @property
173 def chaos(self) -> float:
174 return self._mean_mess_ratio
175
176 @property
177 def coherence(self) -> float:
178 if not self._languages:
179 return 0.0
180 return self._languages[0][1]
181
182 @property
183 def percent_chaos(self) -> float:
184 return round(self.chaos * 100, ndigits=3)
185
186 @property
187 def percent_coherence(self) -> float:
188 return round(self.coherence * 100, ndigits=3)
189
190 @property
191 def raw(self) -> bytes | bytearray:
192 """
193 Original untouched bytes.
194 """
195 return self._payload
196
197 @property
198 def submatch(self) -> list[CharsetMatch]:
199 return self._leaves
200
201 @property
202 def has_submatch(self) -> bool:
203 return len(self._leaves) > 0
204
205 @property
206 def alphabets(self) -> list[str]:
207 if self._unicode_ranges is not None:
208 return self._unicode_ranges
209 # list detected ranges
210 detected_ranges: list[str | None] = [unicode_range(char) for char in str(self)]
211 # filter and sort
212 self._unicode_ranges = sorted(list({r for r in detected_ranges if r}))
213 return self._unicode_ranges
214
215 @property
216 def could_be_from_charset(self) -> list[str]:
217 """
218 The complete list of encoding that output the exact SAME str result and therefore could be the originating
219 encoding.
220 This list does include the encoding available in property 'encoding'.
221 """
222 return [self._encoding] + [m.encoding for m in self._leaves]
223
224 def output(self, encoding: str = "utf_8") -> bytes:
225 """
226 Method to get re-encoded bytes payload using given target encoding. Default to UTF-8.
227 Any errors will be simply ignored by the encoder NOT replaced.
228 """
229 if self._output_encoding is None or self._output_encoding != encoding:
230 self._output_encoding = encoding
231 decoded_string = str(self)
232 if (
233 self._preemptive_declaration is not None
234 and self._preemptive_declaration.lower()
235 not in ["utf-8", "utf8", "utf_8"]
236 ):
237 patched_header = sub(
238 RE_POSSIBLE_ENCODING_INDICATION,
239 lambda m: m.string[m.span()[0] : m.span()[1]].replace(
240 m.groups()[0],
241 iana_name(self._output_encoding).replace("_", "-"), # type: ignore[arg-type]
242 ),
243 decoded_string[:8192],
244 count=1,
245 )
246
247 decoded_string = patched_header + decoded_string[8192:]
248
249 self._output_payload = decoded_string.encode(encoding, "replace")
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)