models.py 12 KB

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