| 315 | |
| 316 | class IncrementalDecoder(codecs.BufferedIncrementalDecoder): |
| 317 | def _buffer_decode(self, input, errors, final): |
| 318 | if errors != 'strict': |
| 319 | raise UnicodeError(f"Unsupported error handling: {errors}") |
| 320 | |
| 321 | if not input: |
| 322 | return ("", 0) |
| 323 | |
| 324 | # IDNA allows decoding to operate on Unicode strings, too. |
| 325 | if isinstance(input, str): |
| 326 | labels = dots.split(input) |
| 327 | else: |
| 328 | # Must be ASCII string |
| 329 | try: |
| 330 | input = str(input, "ascii") |
| 331 | except (UnicodeEncodeError, UnicodeDecodeError) as exc: |
| 332 | raise UnicodeDecodeError("idna", input, |
| 333 | exc.start, exc.end, exc.reason) |
| 334 | labels = input.split(".") |
| 335 | |
| 336 | trailing_dot = '' |
| 337 | if labels: |
| 338 | if not labels[-1]: |
| 339 | trailing_dot = '.' |
| 340 | del labels[-1] |
| 341 | elif not final: |
| 342 | # Keep potentially unfinished label until the next call |
| 343 | del labels[-1] |
| 344 | if labels: |
| 345 | trailing_dot = '.' |
| 346 | |
| 347 | result = [] |
| 348 | size = 0 |
| 349 | for label in labels: |
| 350 | try: |
| 351 | u_label = ToUnicode(label) |
| 352 | except (UnicodeEncodeError, UnicodeDecodeError) as exc: |
| 353 | raise UnicodeDecodeError( |
| 354 | "idna", |
| 355 | input.encode("ascii", errors="backslashreplace"), |
| 356 | size + exc.start, |
| 357 | size + exc.end, |
| 358 | exc.reason, |
| 359 | ) |
| 360 | else: |
| 361 | result.append(u_label) |
| 362 | if size: |
| 363 | size += 1 |
| 364 | size += len(label) |
| 365 | |
| 366 | result = ".".join(result) + trailing_dot |
| 367 | size += len(trailing_dot) |
| 368 | return (result, size) |
| 369 | |
| 370 | class StreamWriter(Codec,codecs.StreamWriter): |
| 371 | pass |