Coverage for src / charset_normalizer / cd.py: 99%
212 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-11-01 13:35 +0000
« prev ^ index » next coverage.py v7.13.5, created at 2026-11-01 13:35 +0000
1from __future__ import annotations
3import importlib
4from codecs import IncrementalDecoder
5from functools import lru_cache
7from .constant import (
8 FREQUENCIES,
9 KO_NAMES,
10 LANGUAGE_SUPPORTED_COUNT,
11 TOO_SMALL_SEQUENCE,
12 ZH_NAMES,
13 _FREQUENCIES_SET,
14 _FREQUENCIES_RANK,
15)
16from .md import _ASCII_CHAR_INFO, _char_info, is_suspiciously_successive_range
17from .models import CoherenceMatches
18from .utils import (
19 is_multi_byte_encoding,
20 is_unicode_range_secondary,
21)
24def encoding_unicode_range(iana_name: str) -> list[str]:
25 """
26 Return associated unicode ranges in a single byte code page.
27 """
28 if is_multi_byte_encoding(iana_name):
29 raise OSError( # Defensive:
30 "Function not supported on multi-byte code page"
31 )
33 decoder = importlib.import_module(f"encodings.{iana_name}").IncrementalDecoder
35 p: IncrementalDecoder = decoder(errors="ignore")
36 seen_ranges: dict[str, int] = {}
37 character_count: int = 0
39 for i in range(0x40, 0xFF):
40 chunk: str = p.decode(bytes([i]))
42 if chunk:
43 chunk_codepoint = ord(chunk)
44 character_range: str | None = (
45 _ASCII_CHAR_INFO[chunk_codepoint].range
46 if chunk_codepoint < 128
47 else _char_info(chunk).range
48 )
50 if character_range is None:
51 continue
53 if not is_unicode_range_secondary(character_range):
54 if character_range not in seen_ranges:
55 seen_ranges[character_range] = 0
56 seen_ranges[character_range] += 1
57 character_count += 1
59 return sorted(
60 [
61 character_range
62 for character_range in seen_ranges
63 if seen_ranges[character_range] / character_count >= 0.15
64 ]
65 )
68def unicode_range_languages(primary_range: str) -> list[str]:
69 """
70 Return inferred languages used with a unicode range.
71 """
72 languages: list[str] = []
74 for language, characters in FREQUENCIES.items():
75 for character in characters:
76 codepoint = ord(character)
77 info = (
78 _ASCII_CHAR_INFO[codepoint]
79 if codepoint < 128
80 else _char_info(character)
81 )
82 if info.range == primary_range:
83 languages.append(language)
84 break
86 return languages
89@lru_cache()
90def encoding_languages(iana_name: str) -> list[str]:
91 """
92 Single-byte encoding language association. Some code page are heavily linked to particular language(s).
93 This function does the correspondence.
94 """
95 try:
96 unicode_ranges: list[str] = encoding_unicode_range(iana_name)
97 except ImportError: # Defensive: encoding unavailable on this build.
98 return []
100 primary_range: str | None = None
102 for specified_range in unicode_ranges:
103 if "Latin" not in specified_range:
104 primary_range = specified_range
105 break
107 if primary_range is None:
108 return ["Latin Based"]
110 return unicode_range_languages(primary_range)
113@lru_cache()
114def mb_encoding_languages(iana_name: str) -> list[str]:
115 """
116 Multi-byte encoding language association. Some code page are heavily linked to particular language(s).
117 This function does the correspondence.
118 """
119 if (
120 iana_name.startswith("shift_")
121 or iana_name.startswith("iso2022_jp")
122 or iana_name.startswith("euc_j")
123 or iana_name == "cp932"
124 ):
125 return ["Japanese"]
126 if iana_name.startswith("gb") or iana_name in ZH_NAMES:
127 return ["Chinese"]
128 if iana_name.startswith("iso2022_kr") or iana_name in KO_NAMES:
129 return ["Korean"]
131 return []
134@lru_cache(maxsize=LANGUAGE_SUPPORTED_COUNT)
135def get_target_features(language: str) -> tuple[bool, bool]:
136 """
137 Determine main aspects from a supported language if it contains accents and if is pure Latin.
138 """
139 target_have_accents: bool = False
140 target_pure_latin: bool = True
142 for character in FREQUENCIES[language]:
143 codepoint = ord(character)
144 info = _ASCII_CHAR_INFO[codepoint] if codepoint < 128 else _char_info(character)
145 if not target_have_accents and info.accentuated:
146 target_have_accents = True
147 if target_pure_latin and not info.latin:
148 target_pure_latin = False
150 return target_have_accents, target_pure_latin
153def alphabet_languages(
154 characters: list[str], ignore_non_latin: bool = False
155) -> list[str]:
156 """
157 Return associated languages associated to given characters.
158 """
159 languages: list[tuple[str, float]] = []
161 characters_set: frozenset[str] = frozenset(characters)
162 source_have_accents = False
163 for character in characters:
164 codepoint = ord(character)
165 info = _ASCII_CHAR_INFO[codepoint] if codepoint < 128 else _char_info(character)
166 if info.accentuated:
167 source_have_accents = True
168 break
170 for language, language_characters in FREQUENCIES.items():
171 target_have_accents, target_pure_latin = get_target_features(language)
173 if ignore_non_latin and not target_pure_latin:
174 continue
176 if not target_have_accents and source_have_accents:
177 continue
179 character_count: int = len(language_characters)
181 character_match_count: int = len(_FREQUENCIES_SET[language] & characters_set)
183 ratio: float = character_match_count / character_count
185 if ratio >= 0.2:
186 languages.append((language, ratio))
188 languages = sorted(languages, key=lambda x: x[1], reverse=True)
190 return [compatible_language[0] for compatible_language in languages]
193def characters_popularity_compare(
194 language: str, ordered_characters: list[str]
195) -> float:
196 """
197 Determine if a ordered characters list (by occurrence from most appearance to rarest) match a particular language.
198 The result is a ratio between 0. (absolutely no correspondence) and 1. (near perfect fit).
199 Beware that is function is not strict on the match in order to ease the detection. (Meaning close match is 1.)
200 """
201 if language not in FREQUENCIES:
202 raise ValueError(f"{language} not available") # Defensive:
204 character_approved_count: int = 0
205 lang_rank: dict[str, int] = _FREQUENCIES_RANK[language]
207 ordered_characters_count: int = len(ordered_characters)
208 target_language_characters_count: int = len(FREQUENCIES[language])
210 large_alphabet: bool = target_language_characters_count > 26
211 large_alphabet_threshold: float = target_language_characters_count / 3
213 expected_projection_ratio: float = (
214 target_language_characters_count / ordered_characters_count
215 )
217 # Single pass: characters present in the language vocabulary, as
218 # (language rank, popularity rank) pairs. The scoring below only ever
219 # needs ranks, never the characters themselves.
220 common_lr: list[int] = []
221 common_orr: list[int] = []
222 for popularity_rank, character in enumerate(ordered_characters):
223 language_rank = lang_rank.get(character)
224 if language_rank is not None:
225 common_lr.append(language_rank)
226 common_orr.append(popularity_rank)
228 for character_rank_in_language, character_rank in zip(common_lr, common_orr):
229 character_rank_projection: int = int(character_rank * expected_projection_ratio)
231 if (
232 not large_alphabet
233 and abs(character_rank_projection - character_rank_in_language) > 4
234 ):
235 continue
237 if (
238 large_alphabet
239 and abs(character_rank_projection - character_rank_in_language)
240 < large_alphabet_threshold
241 ):
242 character_approved_count += 1
243 continue
245 if character_rank_in_language == 0:
246 # before_match_count is structurally 0 here (no pair can have a
247 # smaller language rank): the historic "before <= 4" acceptance
248 # always holds. (The symmetric "after_len == 0" case is
249 # impossible: language ranks are strictly below the language
250 # character count, hence after_len >= 1.)
251 character_approved_count += 1
252 continue
254 after_len: int = target_language_characters_count - character_rank_in_language
256 # Count how many characters appear "before" in both orderings, and
257 # how many appear "at or after" in both orderings. Both counts grow
258 # monotonically and the approval thresholds
259 # (before / rank >= 0.4 or after / after_len >= 0.4) are known
260 # upfront, expressed below as exact integer comparisons: exit as
261 # soon as one is crossed.
262 before_match_count: int = 0
263 after_match_count: int = 0
265 for lr_i, orr_i in zip(common_lr, common_orr):
266 if lr_i < character_rank_in_language:
267 if orr_i < character_rank:
268 before_match_count += 1
269 if 5 * before_match_count >= 2 * character_rank_in_language:
270 character_approved_count += 1
271 break
272 else:
273 if orr_i >= character_rank:
274 after_match_count += 1
275 if 5 * after_match_count >= 2 * after_len:
276 character_approved_count += 1
277 break
279 return character_approved_count / len(ordered_characters)
282def alpha_unicode_split(decoded_sequence: str) -> list[str]:
283 """
284 Given a decoded text sequence, return a list of str. Unicode range / alphabet separation.
285 Ex. a text containing English/Latin with a bit a Hebrew will return two items in the resulting list;
286 One containing the latin letters and the other hebrew.
287 """
288 layers: dict[str, list[str]] = {}
290 # Fast path: track single-layer key to skip dict iteration for single-script text.
291 single_layer_key: str | None = None
292 multi_layer: bool = False
294 # Cache the last character_range and its resolved layer to avoid repeated
295 # is_suspiciously_successive_range calls for consecutive same-range chars.
296 prev_character_range: str | None = None
297 prev_layer_target: str | None = None
299 for character in decoded_sequence:
300 # Reuse the per-codepoint CharInfo cache: info.alpha and info.range
301 # are computed with the very same str.isalpha() / unicode_range()
302 # calls this loop historically made per character occurrence.
303 codepoint: int = ord(character)
304 if codepoint < 128:
305 info = _ASCII_CHAR_INFO[codepoint]
306 else:
307 info = _char_info(character)
309 if not info.alpha:
310 continue
312 character_range: str | None = info.range
314 if character_range is None:
315 continue
317 # Fast path: same range as previous character → reuse cached layer target.
318 if character_range == prev_character_range:
319 if prev_layer_target is not None:
320 layers[prev_layer_target].append(character)
321 continue
323 layer_target_range: str | None = None
325 if multi_layer:
326 for discovered_range in layers:
327 if not is_suspiciously_successive_range(
328 discovered_range, character_range
329 ):
330 layer_target_range = discovered_range
331 break
332 elif single_layer_key is not None:
333 if not is_suspiciously_successive_range(single_layer_key, character_range):
334 layer_target_range = single_layer_key
336 if layer_target_range is None:
337 layer_target_range = character_range
339 if layer_target_range not in layers:
340 layers[layer_target_range] = []
341 if single_layer_key is None:
342 single_layer_key = layer_target_range
343 else:
344 multi_layer = True
346 layers[layer_target_range].append(character)
348 # Cache for next iteration
349 prev_character_range = character_range
350 prev_layer_target = layer_target_range
352 return ["".join(chars).lower() for chars in layers.values()]
355def merge_coherence_ratios(results: list[CoherenceMatches]) -> CoherenceMatches:
356 """
357 This function merge results previously given by the function coherence_ratio.
358 The return type is the same as coherence_ratio.
359 """
360 per_language_ratios: dict[str, list[float]] = {}
361 for result in results:
362 for sub_result in result:
363 language, ratio = sub_result
364 if language not in per_language_ratios:
365 per_language_ratios[language] = [ratio]
366 continue
367 per_language_ratios[language].append(ratio)
369 merge = [
370 (
371 language,
372 round(
373 sum(per_language_ratios[language]) / len(per_language_ratios[language]),
374 4,
375 ),
376 )
377 for language in per_language_ratios
378 ]
380 return sorted(merge, key=lambda x: x[1], reverse=True)
383def filter_alt_coherence_matches(results: CoherenceMatches) -> CoherenceMatches:
384 """
385 We shall NOT return "English—" in CoherenceMatches because it is an alternative
386 of "English". This function only keeps the best match and remove the em-dash in it.
387 """
388 index_results: dict[str, list[float]] = dict()
390 for result in results:
391 language, ratio = result
392 no_em_name: str = language.replace("—", "")
394 if no_em_name not in index_results:
395 index_results[no_em_name] = []
397 index_results[no_em_name].append(ratio)
399 return [(language, max(ratios)) for language, ratios in index_results.items()]
402def coherence_ratio(
403 decoded_sequence: str, threshold: float = 0.1, lg_inclusion: str | None = None
404) -> CoherenceMatches:
405 """
406 Detect ANY language that can be identified in given sequence. The sequence will be analysed by layers.
407 A layer = Character extraction by alphabets/ranges.
408 """
410 results: list[tuple[str, float]] = []
411 ignore_non_latin: bool = False
413 sufficient_match_count: int = 0
415 lg_inclusion_list = lg_inclusion.split(",") if lg_inclusion is not None else []
416 if "Latin Based" in lg_inclusion_list:
417 ignore_non_latin = True
418 lg_inclusion_list.remove("Latin Based")
420 for layer in alpha_unicode_split(decoded_sequence):
421 # Native counting + stable sort reproduce Counter.most_common()
422 # ordering exactly (ties keep first-appearance order) without the
423 # interpreted Counter machinery in the compiled hot path.
424 char_counts: dict[str, int] = {}
425 for layer_character in layer:
426 char_counts[layer_character] = char_counts.get(layer_character, 0) + 1
428 character_count: int = len(layer)
430 if character_count <= TOO_SMALL_SEQUENCE:
431 continue
433 popular_character_ordered: list[str] = [
434 item[0]
435 for item in sorted(
436 char_counts.items(), key=lambda item: item[1], reverse=True
437 )
438 ]
440 for language in lg_inclusion_list or alphabet_languages(
441 popular_character_ordered, ignore_non_latin
442 ):
443 ratio: float = characters_popularity_compare(
444 language, popular_character_ordered
445 )
447 if ratio < threshold:
448 continue
449 elif ratio >= 0.8:
450 sufficient_match_count += 1
452 results.append((language, round(ratio, 4)))
454 if sufficient_match_count >= 3:
455 break
457 return sorted(
458 filter_alt_coherence_matches(results), key=lambda x: x[1], reverse=True
459 )