codec.py 3.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112
  1. from .core import encode, decode, alabel, ulabel, IDNAError
  2. import codecs
  3. import re
  4. from typing import Tuple, Optional
  5. _unicode_dots_re = re.compile('[\u002e\u3002\uff0e\uff61]')
  6. class Codec(codecs.Codec):
  7. def encode(self, data: str, errors: str = 'strict') -> Tuple[bytes, int]:
  8. if errors != 'strict':
  9. raise IDNAError('Unsupported error handling \"{}\"'.format(errors))
  10. if not data:
  11. return b"", 0
  12. return encode(data), len(data)
  13. def decode(self, data: bytes, errors: str = 'strict') -> Tuple[str, int]:
  14. if errors != 'strict':
  15. raise IDNAError('Unsupported error handling \"{}\"'.format(errors))
  16. if not data:
  17. return '', 0
  18. return decode(data), len(data)
  19. class IncrementalEncoder(codecs.BufferedIncrementalEncoder):
  20. def _buffer_encode(self, data: str, errors: str, final: bool) -> Tuple[str, int]: # type: ignore
  21. if errors != 'strict':
  22. raise IDNAError('Unsupported error handling \"{}\"'.format(errors))
  23. if not data:
  24. return "", 0
  25. labels = _unicode_dots_re.split(data)
  26. trailing_dot = ''
  27. if labels:
  28. if not labels[-1]:
  29. trailing_dot = '.'
  30. del labels[-1]
  31. elif not final:
  32. # Keep potentially unfinished label until the next call
  33. del labels[-1]
  34. if labels:
  35. trailing_dot = '.'
  36. result = []
  37. size = 0
  38. for label in labels:
  39. result.append(alabel(label))
  40. if size:
  41. size += 1
  42. size += len(label)
  43. # Join with U+002E
  44. result_str = '.'.join(result) + trailing_dot # type: ignore
  45. size += len(trailing_dot)
  46. return result_str, size
  47. class IncrementalDecoder(codecs.BufferedIncrementalDecoder):
  48. def _buffer_decode(self, data: str, errors: str, final: bool) -> Tuple[str, int]: # type: ignore
  49. if errors != 'strict':
  50. raise IDNAError('Unsupported error handling \"{}\"'.format(errors))
  51. if not data:
  52. return ('', 0)
  53. labels = _unicode_dots_re.split(data)
  54. trailing_dot = ''
  55. if labels:
  56. if not labels[-1]:
  57. trailing_dot = '.'
  58. del labels[-1]
  59. elif not final:
  60. # Keep potentially unfinished label until the next call
  61. del labels[-1]
  62. if labels:
  63. trailing_dot = '.'
  64. result = []
  65. size = 0
  66. for label in labels:
  67. result.append(ulabel(label))
  68. if size:
  69. size += 1
  70. size += len(label)
  71. result_str = '.'.join(result) + trailing_dot
  72. size += len(trailing_dot)
  73. return (result_str, size)
  74. class StreamWriter(Codec, codecs.StreamWriter):
  75. pass
  76. class StreamReader(Codec, codecs.StreamReader):
  77. pass
  78. def getregentry() -> codecs.CodecInfo:
  79. # Compatibility as a search_function for codecs.register()
  80. return codecs.CodecInfo(
  81. name='idna',
  82. encode=Codec().encode, # type: ignore
  83. decode=Codec().decode, # type: ignore
  84. incrementalencoder=IncrementalEncoder,
  85. incrementaldecoder=IncrementalDecoder,
  86. streamwriter=StreamWriter,
  87. streamreader=StreamReader,
  88. )