Add files via upload

Author: YuukaNekoUwU

pyfoxfile.py

@@ -25,6 +25,7 @@
 import time
 import stat
 import zlib
+import mmap
 import base64
 import shutil
 import socket
@@ -97,6 +98,14 @@
 except NameError:  # Py3
     unicode = str
 
+if PY2:
+    # In Py2, 'str' is bytes; define a 'bytes' alias for clarity
+    bytes_type = str
+    text_type = unicode  # noqa: F821 (Py2-only)
+else:
+    bytes_type = bytes
+    text_type = str
+
 def to_text(s, encoding="utf-8", errors="ignore"):
     if s is None:
         return u""
@@ -163,6 +172,12 @@ def _wrap(stream):
 except NameError:
     FileNotFoundError = IOError
 
+try:
+    UnsupportedOperation = io.UnsupportedOperation  # Py3
+except AttributeError:
+    class UnsupportedOperation(IOError):  # Py2 fallback
+        pass
+
 # RAR file support
 rarfile_support = False
 try:
@@ -5838,6 +5853,310 @@ def UncompressBytesAltFP(fp, formatspecs=__file_format_multi_dict__, filestart=0
     filefp.seek(0, 0)
     return filefp
 
+def _extract_base_fp(obj):
+    """Return deepest file-like with a working fileno(), or None."""
+    seen = set()
+    cur = obj
+    while cur and id(cur) not in seen:
+        seen.add(id(cur))
+        fileno = getattr(cur, "fileno", None)
+        if callable(fileno):
+            try:
+                fileno()  # probe
+                return cur
+            except Exception:
+                pass
+        # Walk common wrappers (gzip/bz2/lzma/TextIOWrapper/Buffered* etc.)
+        for attr in ("fileobj", "fp", "_fp", "buffer", "raw"):
+            nxt = getattr(cur, attr, None)
+            if nxt is not None and id(nxt) not in seen:
+                cur = nxt
+                break
+        else:
+            cur = None
+    return None
+
+
+class FileLikeAdapter(object):
+    """
+    Py2/3-compatible file-like adapter that can wrap:
+      - BytesIO / memory buffers
+      - real files
+      - compressed streams (gzip/bz2/lzma/...)
+      - an (fp, mmap) pair for uncompressed random-access speed
+
+    Bytes-only API. Honors mode ("rb", "wb", "r+b", etc.).
+    """
+
+    def __init__(self, fp_like, mode="rb", mm=None, name=None):
+        self._fp = fp_like               # underlying stream (BytesIO/file/gzip/...)
+        self._mm = mm                    # optional memory map for uncompressed files
+        self._pos = 0                    # mapping position when using _mm
+        self._mode = mode
+        self.name = name if name is not None else getattr(fp_like, "name", None)
+        self._closed = False
+
+        # permissions (simple flags from mode)
+        self._readable = ("r" in mode) or ("+" in mode)
+        self._writable = ("w" in mode) or ("a" in mode) or ("x" in mode) or ("+" in mode)
+
+        # Accept write_through attr; ignore (compat shim)
+        self.write_through = False
+
+    # ---- capability flags ----
+    def readable(self):
+        return bool(self._readable)
+
+    def writable(self):
+        return bool(self._writable)
+
+    def seekable(self):
+        if self._mm is not None:
+            return True
+        s = getattr(self._fp, "seekable", None)
+        if callable(s):
+            try:
+                return bool(s())
+            except Exception:
+                return hasattr(self._fp, "seek")
+        return hasattr(self._fp, "seek")
+
+    @property
+    def closed(self):
+        base_closed = getattr(self._fp, "closed", None)
+        return bool(base_closed) or self._closed
+
+    # ---- position ----
+    def tell(self):
+        if self._mm is not None:
+            return self._pos
+        return self._fp.tell()
+
+    def seek(self, offset, whence=io.SEEK_SET):
+        if self._mm is None:
+            return self._fp.seek(offset, whence)
+
+        if whence == io.SEEK_SET:
+            new = offset
+        elif whence == io.SEEK_CUR:
+            new = self._pos + offset
+        elif whence == io.SEEK_END:
+            new = len(self._mm) + offset
+        else:
+            raise ValueError("bad whence")
+
+        if not (0 <= new <= len(self._mm)):
+            raise ValueError("seek out of range")
+        self._pos = new
+        return self._pos
+
+    # ---- reads ----
+    def read(self, n=-1):
+        if not self._readable:
+            raise UnsupportedOperation("not readable")
+
+        if self._mm is None:
+            return self._fp.read(n)
+
+        if n is None or n < 0:
+            n = len(self._mm) - self._pos
+        end = min(self._pos + n, len(self._mm))
+        if end <= self._pos:
+            return b"" if not PY2 else bytes_type()
+        # In Py2, bytes(self._mm[slice]) returns str (bytes); fine.
+        out = bytes(self._mm[self._pos:end])
+        self._pos = end
+        return out
+
+    def readinto(self, b):
+        if not self._readable:
+            raise UnsupportedOperation("not readable")
+
+        if self._mm is None:
+            readinto = getattr(self._fp, "readinto", None)
+            if callable(readinto):
+                return readinto(b)
+            # Emulate readinto if missing (common on Py2 wrappers)
+            data = self._fp.read(len(b))
+            if not data:
+                return 0
+            mv = memoryview(b)
+            n = min(len(mv), len(data))
+            mv[:n] = data[:n]
+            return n
+
+        mv = memoryview(b)
+        remaining = len(self._mm) - self._pos
+        n = min(len(mv), remaining)
+        if n <= 0:
+            return 0
+        mv[:n] = self._mm[self._pos:self._pos + n]
+        self._pos += n
+        return n
+
+    def readline(self, limit=-1):
+        if not self._readable:
+            raise UnsupportedOperation("not readable")
+
+        if self._mm is None:
+            return self._fp.readline(limit)
+
+        if limit is not None and limit >= 0:
+            end_limit = min(self._pos + limit, len(self._mm))
+        else:
+            end_limit = len(self._mm)
+
+        nl = self._mm.find(b"\n", self._pos, end_limit)
+        if nl == -1:
+            end = end_limit
+        else:
+            end = nl + 1
+        out = bytes(self._mm[self._pos:end])
+        self._pos = end
+        return out
+
+    def readlines(self, hint=-1):
+        lines, total = [], 0
+        while True:
+            line = self.readline()
+            if not line:
+                break
+            lines.append(line)
+            total += len(line)
+            if hint >= 0 and total >= hint:
+                break
+        return lines
+
+    # Iteration (Py2/3)
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        line = self.readline()
+        if not line:
+            raise StopIteration
+        return line
+
+    # Py2 alias
+    if PY2:
+        next = __next__
+
+    # ---- writes ----
+    def write(self, b):
+        if not self._writable:
+            raise UnsupportedOperation("not writable")
+
+        if not isinstance(b, bytes_type):
+            # for safety, only bytes; caller handles text encoding externally
+            raise TypeError("write() requires bytes")
+
+        if self._mm is None:
+            return self._fp.write(b)
+
+        mv = memoryview(b)
+        end = self._pos + len(mv)
+        if end > len(self._mm):
+            raise IOError("write past mapped size; pre-size or resize()")
+        self._mm[self._pos:end] = mv
+        self._pos = end
+        return len(mv)
+
+    def writelines(self, lines):
+        for line in lines:
+            self.write(line)
+
+    # ---- durability & size ----
+    def flush(self):
+        # 1) flush mapping first
+        if self._mm is not None:
+            try:
+                self._mm.flush()
+            except Exception:
+                pass
+        # 2) flush Python/stdio buffers
+        try:
+            self._fp.flush()
+        except Exception:
+            pass
+        # 3) fsync real file if any (skips BytesIO and many compressed)
+        base = _extract_base_fp(self._fp)
+        if base is not None:
+            try:
+                os.fsync(base.fileno())
+            except Exception:
+                pass
+
+    def truncate(self, size=None):
+        if self._mm is not None:
+            base = _extract_base_fp(self._fp)
+            if base is None:
+                raise UnsupportedOperation("truncate unsupported for mmapped non-file")
+            if size is None:
+                size = self.tell()
+            # Safe approach across OSes: close map, truncate file, re-map
+            was_pos = self._pos
+            try:
+                self._mm.close()
+            except Exception:
+                pass
+            base.truncate(size)
+            access = mmap.ACCESS_WRITE if self._writable else mmap.ACCESS_READ
+            self._mm = mmap.mmap(base.fileno(), size, access=access)
+            self._pos = min(was_pos, size)
+            return size
+
+        trunc = getattr(self._fp, "truncate", None)
+        if not callable(trunc):
+            raise UnsupportedOperation("truncate unsupported by underlying object")
+        return trunc(size)
+
+    # ---- fd/tty ----
+    def fileno(self):
+        f = getattr(self._fp, "fileno", None)
+        if callable(f):
+            return f()
+        raise UnsupportedOperation("no fileno()")
+
+    def isatty(self):
+        f = getattr(self._fp, "isatty", None)
+        try:
+            return bool(f()) if callable(f) else False
+        except Exception:
+            return False
+
+    # ---- close & ctx mgr ----
+    def close(self):
+        if self._closed:
+            return
+        try:
+            if self._writable:
+                self.flush()
+        finally:
+            if self._mm is not None:
+                try:
+                    self._mm.close()
+                except Exception:
+                    pass
+                self._mm = None
+            try:
+                self._fp.close()
+            except Exception:
+                pass
+            self._closed = True
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        self.close()
+
+    # Accept write_through sets (compat with your current code)
+    def __setattr__(self, name, value):
+        if name == "write_through":
+            object.__setattr__(self, name, value)
+            return
+        object.__setattr__(self, name, value)
+
 
 def CompressOpenFileAlt(fp, compression="auto", compressionlevel=None, compressionuselist=compressionlistalt, formatspecs=__file_format_dict__):
     if(not hasattr(fp, "read")):