read the file and yield chunks of chunksize bytes defaults to file defaultchunksize return true if you can expect multiple chunks nb if a particular file representation is in memory subclasses should always return false there s no good reason to read from memory in chunks iterate over this filelike object by newlines line split after a r newline yield buffer continue with line line either split without a newline line continues after buffer or with rn newline line b n buffer handled clear it if this is the end of a n or rn line yield a filelike object that takes just raw content rather than an actual file return true if line a text or bytestring ends with r return line endswithr if isinstanceline str else br def endswithlfline return true if line a text or bytestring equals n return line n if isinstanceline str else bn read the file and yield chunks of chunk_size bytes defaults to file default_chunk_size return true if you can expect multiple chunks nb if a particular file representation is in memory subclasses should always return false there s no good reason to read from memory in chunks iterate over this file like object by newlines line split after a r newline yield buffer_ continue with line line either split without a newline line continues after buffer_ or with r n newline line b n buffer_ handled clear it if this is the end of a n or r n line yield a file like object that takes just raw content rather than an actual file clear the computed size return true if line a text or bytestring ends with r return true if line a text or bytestring ends with n return true if line a text or bytestring equals n
import os
from io import BytesIO, StringIO, UnsupportedOperation

from django.core.files.utils import FileProxyMixin
from django.utils.functional import cached_property


class File(FileProxyMixin):
    DEFAULT_CHUNK_SIZE = 64 * 2**10

    def __init__(self, file, name=None):
        self.file = file
        if name is None:
            name = getattr(file, "name", None)
        self.name = name
        if hasattr(file, "mode"):
            self.mode = file.mode

    def __str__(self):
        return self.name or ""

    def __repr__(self):
        return "<%s: %s>" % (self.__class__.__name__, self or "None")

    def __bool__(self):
        return bool(self.name)

    def __len__(self):
        return self.size

    @cached_property
    def size(self):
        if hasattr(self.file, "size"):
            return self.file.size
        if hasattr(self.file, "name"):
            try:
                return os.path.getsize(self.file.name)
            except (OSError, TypeError):
                pass
        if hasattr(self.file, "tell") and hasattr(self.file, "seek"):
            pos = self.file.tell()
            self.file.seek(0, os.SEEK_END)
            size = self.file.tell()
            self.file.seek(pos)
            return size
        raise AttributeError("Unable to determine the file's size.")

    def chunks(self, chunk_size=None):
        
        chunk_size = chunk_size or self.DEFAULT_CHUNK_SIZE
        try:
            self.seek(0)
        except (AttributeError, UnsupportedOperation):
            pass

        while True:
            data = self.read(chunk_size)
            if not data:
                break
            yield data

    def multiple_chunks(self, chunk_size=None):
        
        return self.size > (chunk_size or self.DEFAULT_CHUNK_SIZE)

    def __iter__(self):
        
        buffer_ = None
        for chunk in self.chunks():
            for line in chunk.splitlines(True):
                if buffer_:
                    if endswith_cr(buffer_) and not equals_lf(line):
                        
                        yield buffer_
                        
                    else:
                        
                        
                        
                        line = buffer_ + line
                    
                    buffer_ = None

                
                if endswith_lf(line):
                    yield line
                else:
                    buffer_ = line

        if buffer_ is not None:
            yield buffer_

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, tb):
        self.close()

    def open(self, mode=None, *args, **kwargs):
        if not self.closed:
            self.seek(0)
        elif self.name and os.path.exists(self.name):
            self.file = open(self.name, mode or self.mode, *args, **kwargs)
        else:
            raise ValueError("The file cannot be reopened.")
        return self

    def close(self):
        self.file.close()


class ContentFile(File):
    

    def __init__(self, content, name=None):
        stream_class = StringIO if isinstance(content, str) else BytesIO
        super().__init__(stream_class(content), name=name)
        self.size = len(content)

    def __str__(self):
        return "Raw content"

    def __bool__(self):
        return True

    def open(self, mode=None):
        self.seek(0)
        return self

    def close(self):
        pass

    def write(self, data):
        self.__dict__.pop("size", None)  
        return self.file.write(data)


def endswith_cr(line):
    
    return line.endswith("\r" if isinstance(line, str) else b"\r")


def endswith_lf(line):
    
    return line.endswith("\n" if isinstance(line, str) else b"\n")


def equals_lf(line):
    
    return line == ("\n" if isinstance(line, str) else b"\n")