|
jpayne@68
|
1 import codecs
|
|
jpayne@68
|
2 import re
|
|
jpayne@68
|
3 from typing import Any, Optional, Tuple
|
|
jpayne@68
|
4
|
|
jpayne@68
|
5 from .core import IDNAError, alabel, decode, encode, ulabel
|
|
jpayne@68
|
6
|
|
jpayne@68
|
7 _unicode_dots_re = re.compile("[\u002e\u3002\uff0e\uff61]")
|
|
jpayne@68
|
8
|
|
jpayne@68
|
9
|
|
jpayne@68
|
10 class Codec(codecs.Codec):
|
|
jpayne@68
|
11 def encode(self, data: str, errors: str = "strict") -> Tuple[bytes, int]:
|
|
jpayne@68
|
12 if errors != "strict":
|
|
jpayne@68
|
13 raise IDNAError('Unsupported error handling "{}"'.format(errors))
|
|
jpayne@68
|
14
|
|
jpayne@68
|
15 if not data:
|
|
jpayne@68
|
16 return b"", 0
|
|
jpayne@68
|
17
|
|
jpayne@68
|
18 return encode(data), len(data)
|
|
jpayne@68
|
19
|
|
jpayne@68
|
20 def decode(self, data: bytes, errors: str = "strict") -> Tuple[str, int]:
|
|
jpayne@68
|
21 if errors != "strict":
|
|
jpayne@68
|
22 raise IDNAError('Unsupported error handling "{}"'.format(errors))
|
|
jpayne@68
|
23
|
|
jpayne@68
|
24 if not data:
|
|
jpayne@68
|
25 return "", 0
|
|
jpayne@68
|
26
|
|
jpayne@68
|
27 return decode(data), len(data)
|
|
jpayne@68
|
28
|
|
jpayne@68
|
29
|
|
jpayne@68
|
30 class IncrementalEncoder(codecs.BufferedIncrementalEncoder):
|
|
jpayne@68
|
31 def _buffer_encode(self, data: str, errors: str, final: bool) -> Tuple[bytes, int]:
|
|
jpayne@68
|
32 if errors != "strict":
|
|
jpayne@68
|
33 raise IDNAError('Unsupported error handling "{}"'.format(errors))
|
|
jpayne@68
|
34
|
|
jpayne@68
|
35 if not data:
|
|
jpayne@68
|
36 return b"", 0
|
|
jpayne@68
|
37
|
|
jpayne@68
|
38 labels = _unicode_dots_re.split(data)
|
|
jpayne@68
|
39 trailing_dot = b""
|
|
jpayne@68
|
40 if labels:
|
|
jpayne@68
|
41 if not labels[-1]:
|
|
jpayne@68
|
42 trailing_dot = b"."
|
|
jpayne@68
|
43 del labels[-1]
|
|
jpayne@68
|
44 elif not final:
|
|
jpayne@68
|
45 # Keep potentially unfinished label until the next call
|
|
jpayne@68
|
46 del labels[-1]
|
|
jpayne@68
|
47 if labels:
|
|
jpayne@68
|
48 trailing_dot = b"."
|
|
jpayne@68
|
49
|
|
jpayne@68
|
50 result = []
|
|
jpayne@68
|
51 size = 0
|
|
jpayne@68
|
52 for label in labels:
|
|
jpayne@68
|
53 result.append(alabel(label))
|
|
jpayne@68
|
54 if size:
|
|
jpayne@68
|
55 size += 1
|
|
jpayne@68
|
56 size += len(label)
|
|
jpayne@68
|
57
|
|
jpayne@68
|
58 # Join with U+002E
|
|
jpayne@68
|
59 result_bytes = b".".join(result) + trailing_dot
|
|
jpayne@68
|
60 size += len(trailing_dot)
|
|
jpayne@68
|
61 return result_bytes, size
|
|
jpayne@68
|
62
|
|
jpayne@68
|
63
|
|
jpayne@68
|
64 class IncrementalDecoder(codecs.BufferedIncrementalDecoder):
|
|
jpayne@68
|
65 def _buffer_decode(self, data: Any, errors: str, final: bool) -> Tuple[str, int]:
|
|
jpayne@68
|
66 if errors != "strict":
|
|
jpayne@68
|
67 raise IDNAError('Unsupported error handling "{}"'.format(errors))
|
|
jpayne@68
|
68
|
|
jpayne@68
|
69 if not data:
|
|
jpayne@68
|
70 return ("", 0)
|
|
jpayne@68
|
71
|
|
jpayne@68
|
72 if not isinstance(data, str):
|
|
jpayne@68
|
73 data = str(data, "ascii")
|
|
jpayne@68
|
74
|
|
jpayne@68
|
75 labels = _unicode_dots_re.split(data)
|
|
jpayne@68
|
76 trailing_dot = ""
|
|
jpayne@68
|
77 if labels:
|
|
jpayne@68
|
78 if not labels[-1]:
|
|
jpayne@68
|
79 trailing_dot = "."
|
|
jpayne@68
|
80 del labels[-1]
|
|
jpayne@68
|
81 elif not final:
|
|
jpayne@68
|
82 # Keep potentially unfinished label until the next call
|
|
jpayne@68
|
83 del labels[-1]
|
|
jpayne@68
|
84 if labels:
|
|
jpayne@68
|
85 trailing_dot = "."
|
|
jpayne@68
|
86
|
|
jpayne@68
|
87 result = []
|
|
jpayne@68
|
88 size = 0
|
|
jpayne@68
|
89 for label in labels:
|
|
jpayne@68
|
90 result.append(ulabel(label))
|
|
jpayne@68
|
91 if size:
|
|
jpayne@68
|
92 size += 1
|
|
jpayne@68
|
93 size += len(label)
|
|
jpayne@68
|
94
|
|
jpayne@68
|
95 result_str = ".".join(result) + trailing_dot
|
|
jpayne@68
|
96 size += len(trailing_dot)
|
|
jpayne@68
|
97 return (result_str, size)
|
|
jpayne@68
|
98
|
|
jpayne@68
|
99
|
|
jpayne@68
|
100 class StreamWriter(Codec, codecs.StreamWriter):
|
|
jpayne@68
|
101 pass
|
|
jpayne@68
|
102
|
|
jpayne@68
|
103
|
|
jpayne@68
|
104 class StreamReader(Codec, codecs.StreamReader):
|
|
jpayne@68
|
105 pass
|
|
jpayne@68
|
106
|
|
jpayne@68
|
107
|
|
jpayne@68
|
108 def search_function(name: str) -> Optional[codecs.CodecInfo]:
|
|
jpayne@68
|
109 if name != "idna2008":
|
|
jpayne@68
|
110 return None
|
|
jpayne@68
|
111 return codecs.CodecInfo(
|
|
jpayne@68
|
112 name=name,
|
|
jpayne@68
|
113 encode=Codec().encode,
|
|
jpayne@68
|
114 decode=Codec().decode,
|
|
jpayne@68
|
115 incrementalencoder=IncrementalEncoder,
|
|
jpayne@68
|
116 incrementaldecoder=IncrementalDecoder,
|
|
jpayne@68
|
117 streamwriter=StreamWriter,
|
|
jpayne@68
|
118 streamreader=StreamReader,
|
|
jpayne@68
|
119 )
|
|
jpayne@68
|
120
|
|
jpayne@68
|
121
|
|
jpayne@68
|
122 codecs.register(search_function)
|
