#define PY_SSIZE_T_CLEAN

#include "Python.h"

#include "structmember.h"

#include "_iomodule.h"

_Py_IDENTIFIER(close);

_Py_IDENTIFIER(_dealloc_warn);

_Py_IDENTIFIER(decode);

_Py_IDENTIFIER(fileno);

_Py_IDENTIFIER(flush);

_Py_IDENTIFIER(getpreferredencoding);

_Py_IDENTIFIER(isatty);

_Py_IDENTIFIER(mode);

_Py_IDENTIFIER(name);

_Py_IDENTIFIER(raw);

_Py_IDENTIFIER(read);

_Py_IDENTIFIER(readable);

_Py_IDENTIFIER(replace);

_Py_IDENTIFIER(reset);

_Py_IDENTIFIER(seek);

_Py_IDENTIFIER(seekable);

_Py_IDENTIFIER(setstate);

_Py_IDENTIFIER(strict);

_Py_IDENTIFIER(tell);

_Py_IDENTIFIER(writable);

PyDoc_STRVAR(textiobase_doc,

"Base class for text I/O.\n"

"\n"

"This class provides a character and line based interface to stream\n"

"I/O. There is no readinto method because Python's character strings\n"

"are immutable. There is no public constructor.\n"

);

static PyObject *

_unsupported(const char *message)

{

_PyIO_State *state = IO_STATE();

if (state != NULL)

PyErr_SetString(state->unsupported_operation, message);

return NULL;

}

PyDoc_STRVAR(textiobase_detach_doc,

"Separate the underlying buffer from the TextIOBase and return it.\n"

"\n"

"After the underlying buffer has been detached, the TextIO is in an\n"

"unusable state.\n"

);

static PyObject *

textiobase_detach(PyObject *self, PyObject *Py_UNUSED(ignored))

{

return _unsupported("detach");

}

PyDoc_STRVAR(textiobase_read_doc,

"Read at most n characters from stream.\n"

"\n"

"Read from underlying buffer until we have n characters or we hit EOF.\n"

"If n is negative or omitted, read until EOF.\n"

);

static PyObject *

textiobase_read(PyObject *self, PyObject *args)

{

return _unsupported("read");

}

PyDoc_STRVAR(textiobase_readline_doc,

"Read until newline or EOF.\n"

"\n"

"Returns an empty string if EOF is hit immediately.\n"

);

static PyObject *

textiobase_readline(PyObject *self, PyObject *args)

{

return _unsupported("readline");

}

PyDoc_STRVAR(textiobase_write_doc,

"Write string to stream.\n"

"Returns the number of characters written (which is always equal to\n"

"the length of the string).\n"

);

static PyObject *

textiobase_write(PyObject *self, PyObject *args)

{

return _unsupported("write");

}

PyDoc_STRVAR(textiobase_encoding_doc,

"Encoding of the text stream.\n"

"\n"

"Subclasses should override.\n"

);

static PyObject *

textiobase_encoding_get(PyObject *self, void *context)

{

Py_RETURN_NONE;

}

PyDoc_STRVAR(textiobase_newlines_doc,

"Line endings translated so far.\n"

"\n"

"Only line endings translated during reading are considered.\n"

"\n"

"Subclasses should override.\n"

);

static PyObject *

textiobase_newlines_get(PyObject *self, void *context)

{

Py_RETURN_NONE;

}

PyDoc_STRVAR(textiobase_errors_doc,

"The error setting of the decoder or encoder.\n"

"\n"

"Subclasses should override.\n"

);

static PyObject *

textiobase_errors_get(PyObject *self, void *context)

{

Py_RETURN_NONE;

}

static PyMethodDef textiobase_methods[] = {

{"detach", textiobase_detach, METH_NOARGS, textiobase_detach_doc},

{"read", textiobase_read, METH_VARARGS, textiobase_read_doc},

{"readline", textiobase_readline, METH_VARARGS, textiobase_readline_doc},

{"write", textiobase_write, METH_VARARGS, textiobase_write_doc},

{NULL, NULL}

};

static PyGetSetDef textiobase_getset[] = {

{"encoding", (getter)textiobase_encoding_get, NULL, textiobase_encoding_doc},

{"newlines", (getter)textiobase_newlines_get, NULL, textiobase_newlines_doc},

{"errors", (getter)textiobase_errors_get, NULL, textiobase_errors_doc},

{NULL}

};

PyTypeObject PyTextIOBase_Type = {

PyVarObject_HEAD_INIT(NULL, 0)

"_io._TextIOBase", /*tp_name*/

0, /*tp_basicsize*/

0, /*tp_itemsize*/

0, /*tp_dealloc*/

0, /*tp_print*/

0, /*tp_getattr*/

0, /*tp_setattr*/

0, /*tp_compare */

0, /*tp_repr*/

0, /*tp_as_number*/

0, /*tp_as_sequence*/

0, /*tp_as_mapping*/

0, /*tp_hash */

0, /*tp_call*/

0, /*tp_str*/

0, /*tp_getattro*/

0, /*tp_setattro*/

0, /*tp_as_buffer*/

Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE

| Py_TPFLAGS_HAVE_FINALIZE, /*tp_flags*/

textiobase_doc, /* tp_doc */

0, /* tp_traverse */

0, /* tp_clear */

0, /* tp_richcompare */

0, /* tp_weaklistoffset */

0, /* tp_iter */

0, /* tp_iternext */

textiobase_methods, /* tp_methods */

0, /* tp_members */

textiobase_getset, /* tp_getset */

&PyIOBase_Type, /* tp_base */

0, /* tp_dict */

0, /* tp_descr_get */

0, /* tp_descr_set */

0, /* tp_dictoffset */

0, /* tp_init */

0, /* tp_alloc */

0, /* tp_new */

0, /* tp_free */

0, /* tp_is_gc */

0, /* tp_bases */

0, /* tp_mro */

0, /* tp_cache */

0, /* tp_subclasses */

0, /* tp_weaklist */

0, /* tp_del */

0, /* tp_version_tag */

0, /* tp_finalize */

};

typedef struct {

PyObject_HEAD

PyObject *decoder;

PyObject *errors;

unsigned int pendingcr: 1;

unsigned int translate: 1;

unsigned int seennl: 3;

} nldecoder_object;

static int

_io_IncrementalNewlineDecoder___init___impl(nldecoder_object *self,

PyObject *decoder, int translate,

PyObject *errors)

{

self->decoder = decoder;

Py_INCREF(decoder);

if (errors == NULL) {

self->errors = _PyUnicode_FromId(&PyId_strict);

if (self->errors == NULL)

return -1;

}

else {

self->errors = errors;

}

Py_INCREF(self->errors);

self->translate = translate;

self->seennl = 0;

self->pendingcr = 0;

return 0;

}

static void

incrementalnewlinedecoder_dealloc(nldecoder_object *self)

{

Py_CLEAR(self->decoder);

Py_CLEAR(self->errors);

Py_TYPE(self)->tp_free((PyObject *)self);

}

static int

check_decoded(PyObject *decoded)

{

if (decoded == NULL)

return -1;

if (!PyUnicode_Check(decoded)) {

PyErr_Format(PyExc_TypeError,

"decoder should return a string result, not '%.200s'",

Py_TYPE(decoded)->tp_name);

Py_DECREF(decoded);

return -1;

}

if (PyUnicode_READY(decoded) < 0) {

Py_DECREF(decoded);

return -1;

}

return 0;

}

#define SEEN_CR 1

#define SEEN_LF 2

#define SEEN_CRLF 4

#define SEEN_ALL (SEEN_CR | SEEN_LF | SEEN_CRLF)

PyObject *

_PyIncrementalNewlineDecoder_decode(PyObject *myself,

PyObject *input, int final)

{

PyObject *output;

Py_ssize_t output_len;

nldecoder_object *self = (nldecoder_object *) myself;

if (self->decoder == NULL) {

PyErr_SetString(PyExc_ValueError,

"IncrementalNewlineDecoder.__init__ not called");

return NULL;

}

/* decode input (with the eventual \r from a previous pass) */

if (self->decoder != Py_None) {

output = PyObject_CallMethodObjArgs(self->decoder,

_PyIO_str_decode, input, final ? Py_True : Py_False, NULL);

}

else {

output = input;

Py_INCREF(output);

}

if (check_decoded(output) < 0)

return NULL;

output_len = PyUnicode_GET_LENGTH(output);

if (self->pendingcr && (final || output_len > 0)) {

/* Prefix output with CR */

int kind;

PyObject *modified;

char *out;

modified = PyUnicode_New(output_len + 1,

PyUnicode_MAX_CHAR_VALUE(output));

if (modified == NULL)

goto error;

kind = PyUnicode_KIND(modified);

out = PyUnicode_DATA(modified);

PyUnicode_WRITE(kind, PyUnicode_DATA(modified), 0, '\r');

memcpy(out + kind, PyUnicode_DATA(output), kind * output_len);

Py_DECREF(output);

output = modified; /* output remains ready */

self->pendingcr = 0;

output_len++;

}

/* retain last \r even when not translating data:

if (!final) {

if (output_len > 0

&& PyUnicode_READ_CHAR(output, output_len - 1) == '\r')

{

PyObject *modified = PyUnicode_Substring(output, 0, output_len -1);

if (modified == NULL)

goto error;

Py_DECREF(output);

output = modified;

self->pendingcr = 1;

}

}

/* Record which newlines are read and do newline translation if desired,

{

void *in_str;

Py_ssize_t len;

int seennl = self->seennl;

int only_lf = 0;

int kind;

in_str = PyUnicode_DATA(output);

len = PyUnicode_GET_LENGTH(output);

kind = PyUnicode_KIND(output);

if (len == 0)

return output;

/* If, up to now, newlines are consistently \n, do a quick check

if (seennl == SEEN_LF || seennl == 0) {

only_lf = (memchr(in_str, '\r', kind * len) == NULL);

}

if (only_lf) {

/* If not already seen, quick scan for a possible "\n" character.

if (seennl == 0 &&

memchr(in_str, '\n', kind * len) != NULL) {

if (kind == PyUnicode_1BYTE_KIND)

seennl |= SEEN_LF;

else {

Py_ssize_t i = 0;

for (;;) {

Py_UCS4 c;

/* Fast loop for non-control characters */

while (PyUnicode_READ(kind, in_str, i) > '\n')

i++;

c = PyUnicode_READ(kind, in_str, i++);

if (c == '\n') {

seennl |= SEEN_LF;

break;

}

if (i >= len)

break;

}

}

}

/* Finished: we have scanned for newlines, and none of them

}

else if (!self->translate) {

Py_ssize_t i = 0;

/* We have already seen all newline types, no need to scan again */

if (seennl == SEEN_ALL)

goto endscan;

for (;;) {

Py_UCS4 c;

/* Fast loop for non-control characters */

while (PyUnicode_READ(kind, in_str, i) > '\r')

i++;

c = PyUnicode_READ(kind, in_str, i++);

if (c == '\n')

seennl |= SEEN_LF;

else if (c == '\r') {

if (PyUnicode_READ(kind, in_str, i) == '\n') {

seennl |= SEEN_CRLF;

i++;

}

else

seennl |= SEEN_CR;

}

if (i >= len)

break;

if (seennl == SEEN_ALL)

break;

}

endscan:

;

}

else {

void *translated;

int kind = PyUnicode_KIND(output);

void *in_str = PyUnicode_DATA(output);

Py_ssize_t in, out;

/* XXX: Previous in-place translation here is disabled as

/* We could try to optimize this so that we only do a copy

translated = PyMem_Malloc(kind * len);

if (translated == NULL) {

PyErr_NoMemory();

goto error;

}

in = out = 0;

for (;;) {

Py_UCS4 c;

/* Fast loop for non-control characters */

while ((c = PyUnicode_READ(kind, in_str, in++)) > '\r')

PyUnicode_WRITE(kind, translated, out++, c);

if (c == '\n') {

PyUnicode_WRITE(kind, translated, out++, c);

seennl |= SEEN_LF;

continue;

}

if (c == '\r') {

if (PyUnicode_READ(kind, in_str, in) == '\n') {

in++;

seennl |= SEEN_CRLF;

}

else

seennl |= SEEN_CR;

PyUnicode_WRITE(kind, translated, out++, '\n');

continue;

}

if (in > len)

break;

PyUnicode_WRITE(kind, translated, out++, c);

}

Py_DECREF(output);

output = PyUnicode_FromKindAndData(kind, translated, out);

PyMem_Free(translated);

if (!output)

return NULL;

}

self->seennl |= seennl;

}

return output;

error:

Py_DECREF(output);

return NULL;

}

static PyObject *

_io_IncrementalNewlineDecoder_decode_impl(nldecoder_object *self,

PyObject *input, int final)

{

return _PyIncrementalNewlineDecoder_decode((PyObject *) self, input, final);

}

static PyObject *

_io_IncrementalNewlineDecoder_getstate_impl(nldecoder_object *self)

{

PyObject *buffer;

unsigned long long flag;

if (self->decoder != Py_None) {

PyObject *state = PyObject_CallMethodObjArgs(self->decoder,

_PyIO_str_getstate, NULL);

if (state == NULL)

return NULL;

if (!PyTuple_Check(state)) {

PyErr_SetString(PyExc_TypeError,

"illegal decoder state");

Py_DECREF(state);

return NULL;

}

if (!PyArg_ParseTuple(state, "OK;illegal decoder state",

&buffer, &flag))

{

Py_DECREF(state);

return NULL;

}

Py_INCREF(buffer);

Py_DECREF(state);

}

else {

buffer = PyBytes_FromString("");

flag = 0;

}

flag <<= 1;

if (self->pendingcr)

flag |= 1;

return Py_BuildValue("NK", buffer, flag);

}

static PyObject *

_io_IncrementalNewlineDecoder_setstate(nldecoder_object *self,

PyObject *state)

{

PyObject *buffer;

unsigned long long flag;

if (!PyTuple_Check(state)) {

PyErr_SetString(PyExc_TypeError, "state argument must be a tuple");

return NULL;

}

if (!PyArg_ParseTuple(state, "OK;setstate(): illegal state argument",

&buffer, &flag))

{

return NULL;

}

self->pendingcr = (int) (flag & 1);

flag >>= 1;

if (self->decoder != Py_None)

return _PyObject_CallMethodId(self->decoder,

&PyId_setstate, "((OK))", buffer, flag);

else

Py_RETURN_NONE;

}

static PyObject *

_io_IncrementalNewlineDecoder_reset_impl(nldecoder_object *self)

{

self->seennl = 0;

self->pendingcr = 0;

if (self->decoder != Py_None)

return PyObject_CallMethodObjArgs(self->decoder, _PyIO_str_reset, NULL);

else

Py_RETURN_NONE;

}

static PyObject *

incrementalnewlinedecoder_newlines_get(nldecoder_object *self, void *context)

{

switch (self->seennl) {

case SEEN_CR:

return PyUnicode_FromString("\r");

case SEEN_LF:

return PyUnicode_FromString("\n");

case SEEN_CRLF:

return PyUnicode_FromString("\r\n");

case SEEN_CR | SEEN_LF:

return Py_BuildValue("ss", "\r", "\n");

case SEEN_CR | SEEN_CRLF:

return Py_BuildValue("ss", "\r", "\r\n");

case SEEN_LF | SEEN_CRLF:

return Py_BuildValue("ss", "\n", "\r\n");

case SEEN_CR | SEEN_LF | SEEN_CRLF:

return Py_BuildValue("sss", "\r", "\n", "\r\n");

default:

Py_RETURN_NONE;

}

}

typedef PyObject *

(*encodefunc_t)(PyObject *, PyObject *);

typedef struct

{

PyObject_HEAD

int ok; /* initialized? */

int detached;

Py_ssize_t chunk_size;

PyObject *buffer;

PyObject *encoding;

PyObject *encoder;

PyObject *decoder;

PyObject *readnl;

PyObject *errors;

const char *writenl; /* ASCII-encoded; NULL stands for \n */

char line_buffering;

char write_through;

char readuniversal;

char readtranslate;

char writetranslate;

char seekable;

char has_read1;

char telling;

char finalizing;

/* Specialized encoding func (see below) */

encodefunc_t encodefunc;

/* Whether or not it's the start of the stream */

char encoding_start_of_stream;

/* Reads and writes are internally buffered in order to speed things up.

PyObject *decoded_chars; /* buffer for text returned from decoder */

Py_ssize_t decoded_chars_used; /* offset into _decoded_chars for read() */

PyObject *pending_bytes; /* list of bytes objects waiting to be

Py_ssize_t pending_bytes_count;

/* snapshot is either NULL, or a tuple (dec_flags, next_input) where

PyObject *snapshot;

/* Bytes-to-characters ratio for the current chunk. Serves as input for

double b2cratio;

/* Cache raw object if it's a FileIO object */

PyObject *raw;

PyObject *weakreflist;

PyObject *dict;

} textio;

static void

textiowrapper_set_decoded_chars(textio *self, PyObject *chars);

static PyObject *

ascii_encode(textio *self, PyObject *text)

{

return _PyUnicode_AsASCIIString(text, PyUnicode_AsUTF8(self->errors));

}

static PyObject *

utf16be_encode(textio *self, PyObject *text)

{

return _PyUnicode_EncodeUTF16(text,

PyUnicode_AsUTF8(self->errors), 1);

}

static PyObject *

utf16le_encode(textio *self, PyObject *text)

{

return _PyUnicode_EncodeUTF16(text,

PyUnicode_AsUTF8(self->errors), -1);

}

static PyObject *

utf16_encode(textio *self, PyObject *text)

{

if (!self->encoding_start_of_stream) {

/* Skip the BOM and use native byte ordering */

#if PY_BIG_ENDIAN

return utf16be_encode(self, text);

return utf16le_encode(self, text);

}

return _PyUnicode_EncodeUTF16(text,

PyUnicode_AsUTF8(self->errors), 0);

}

static PyObject *

utf32be_encode(textio *self, PyObject *text)

{

return _PyUnicode_EncodeUTF32(text,

PyUnicode_AsUTF8(self->errors), 1);

}

static PyObject *

utf32le_encode(textio *self, PyObject *text)

{

return _PyUnicode_EncodeUTF32(text,

PyUnicode_AsUTF8(self->errors), -1);

}

static PyObject *

utf32_encode(textio *self, PyObject *text)

{

if (!self->encoding_start_of_stream) {

/* Skip the BOM and use native byte ordering */

#if PY_BIG_ENDIAN

return utf32be_encode(self, text);

return utf32le_encode(self, text);

}

return _PyUnicode_EncodeUTF32(text,

PyUnicode_AsUTF8(self->errors), 0);

}

static PyObject *

utf8_encode(textio *self, PyObject *text)

{

return _PyUnicode_AsUTF8String(text, PyUnicode_AsUTF8(self->errors));

}

static PyObject *

latin1_encode(textio *self, PyObject *text)

{

return _PyUnicode_AsLatin1String(text, PyUnicode_AsUTF8(self->errors));

}

typedef struct {

const char *name;

encodefunc_t encodefunc;

} encodefuncentry;

static const encodefuncentry encodefuncs[] = {

{"ascii", (encodefunc_t) ascii_encode},

{"iso8859-1", (encodefunc_t) latin1_encode},

{"utf-8", (encodefunc_t) utf8_encode},

{"utf-16-be", (encodefunc_t) utf16be_encode},

{"utf-16-le", (encodefunc_t) utf16le_encode},

{"utf-16", (encodefunc_t) utf16_encode},

{"utf-32-be", (encodefunc_t) utf32be_encode},

{"utf-32-le", (encodefunc_t) utf32le_encode},

{"utf-32", (encodefunc_t) utf32_encode},

{NULL, NULL}

};

static int

validate_newline(const char *newline)

{

if (newline && newline[0] != '\0'

&& !(newline[0] == '\n' && newline[1] == '\0')

&& !(newline[0] == '\r' && newline[1] == '\0')

&& !(newline[0] == '\r' && newline[1] == '\n' && newline[2] == '\0')) {

PyErr_Format(PyExc_ValueError,

"illegal newline value: %s", newline);

return -1;

}

return 0;

}

static int

set_newline(textio *self, const char *newline)

{

PyObject *old = self->readnl;

if (newline == NULL) {

self->readnl = NULL;

}

else {

self->readnl = PyUnicode_FromString(newline);

if (self->readnl == NULL) {

self->readnl = old;

return -1;

}

}

self->readuniversal = (newline == NULL || newline[0] == '\0');

self->readtranslate = (newline == NULL);

self->writetranslate = (newline == NULL || newline[0] != '\0');

if (!self->readuniversal && self->readnl != NULL) {

// validate_newline() accepts only ASCII newlines.

assert(PyUnicode_KIND(self->readnl) == PyUnicode_1BYTE_KIND);

self->writenl = (const char *)PyUnicode_1BYTE_DATA(self->readnl);

if (strcmp(self->writenl, "\n") == 0) {

self->writenl = NULL;

}

}

else {

#ifdef MS_WINDOWS

self->writenl = "\r\n";

self->writenl = NULL;

}

Py_XDECREF(old);

return 0;

}

static int

_textiowrapper_set_decoder(textio *self, PyObject *codec_info,

const char *errors)

{

PyObject *res;

int r;

res = _PyObject_CallMethodId(self->buffer, &PyId_readable, NULL);

if (res == NULL)

return -1;

r = PyObject_IsTrue(res);

Py_DECREF(res);

if (r == -1)

return -1;

if (r != 1)

return 0;

Py_CLEAR(self->decoder);

self->decoder = _PyCodecInfo_GetIncrementalDecoder(codec_info, errors);

if (self->decoder == NULL)

return -1;

if (self->readuniversal) {

PyObject *incrementalDecoder = PyObject_CallFunction(

(PyObject *)&PyIncrementalNewlineDecoder_Type,

"Oi", self->decoder, (int)self->readtranslate);

if (incrementalDecoder == NULL)

return -1;

Py_CLEAR(self->decoder);

self->decoder = incrementalDecoder;

}

return 0;

}

static PyObject*

_textiowrapper_decode(PyObject *decoder, PyObject *bytes, int eof)

{

PyObject *chars;

if (Py_TYPE(decoder) == &PyIncrementalNewlineDecoder_Type)

chars = _PyIncrementalNewlineDecoder_decode(decoder, bytes, eof);

else

chars = PyObject_CallMethodObjArgs(decoder, _PyIO_str_decode, bytes,

eof ? Py_True : Py_False, NULL);

if (check_decoded(chars) < 0)

// check_decoded already decreases refcount

return NULL;

return chars;

}

static int

_textiowrapper_set_encoder(textio *self, PyObject *codec_info,

const char *errors)

{

PyObject *res;

int r;

res = _PyObject_CallMethodId(self->buffer, &PyId_writable, NULL);

if (res == NULL)

return -1;

r = PyObject_IsTrue(res);

Py_DECREF(res);

if (r == -1)

return -1;

if (r != 1)

return 0;

Py_CLEAR(self->encoder);

self->encodefunc = NULL;

self->encoder = _PyCodecInfo_GetIncrementalEncoder(codec_info, errors);

if (self->encoder == NULL)

return -1;

/* Get the normalized named of the codec */

if (_PyObject_LookupAttrId(codec_info, &PyId_name, &res) < 0) {

return -1;

}

if (res != NULL && PyUnicode_Check(res)) {

const encodefuncentry *e = encodefuncs;

while (e->name != NULL) {

if (_PyUnicode_EqualToASCIIString(res, e->name)) {

self->encodefunc = e->encodefunc;

break;

}

e++;

}

}

Py_XDECREF(res);

return 0;

}

static int

_textiowrapper_fix_encoder_state(textio *self)

{

if (!self->seekable || !self->encoder) {

return 0;

}

self->encoding_start_of_stream = 1;

PyObject *cookieObj = PyObject_CallMethodObjArgs(

self->buffer, _PyIO_str_tell, NULL);

if (cookieObj == NULL) {

return -1;

}

int cmp = PyObject_RichCompareBool(cookieObj, _PyLong_Zero, Py_EQ);

Py_DECREF(cookieObj);

if (cmp < 0) {

return -1;

}

if (cmp == 0) {

self->encoding_start_of_stream = 0;

PyObject *res = PyObject_CallMethodObjArgs(

self->encoder, _PyIO_str_setstate, _PyLong_Zero, NULL);

if (res == NULL) {

return -1;

}

Py_DECREF(res);

}

return 0;

}