mirror of
https://github.com/JonathanHerrewijnen/calibre-web.git
synced 2024-11-14 15:14:07 +00:00
bbf6d9b026
Bugfix for feeds
- removed categories related and up
- load new books now working
- category random now working
login page is free of non accessible elements
boolean custom column is vivible in UI
books with only with certain languages can be shown
book shelfs can be deleted from UI
Anonymous user view is more resticted
Added browse of series in sidebar
Dependencys in vendor folder are updated to newer versions (licencs files are now present)
Bugfix editing Authors names
Made upload on windows working
643 lines
20 KiB
Python
643 lines
20 KiB
Python
import re
|
|
import io
|
|
import os
|
|
import sys
|
|
import codecs
|
|
from weakref import WeakKeyDictionary
|
|
|
|
|
|
PY2 = sys.version_info[0] == 2
|
|
WIN = sys.platform.startswith('win')
|
|
DEFAULT_COLUMNS = 80
|
|
|
|
|
|
_ansi_re = re.compile('\033\[((?:\d|;)*)([a-zA-Z])')
|
|
|
|
|
|
def get_filesystem_encoding():
|
|
return sys.getfilesystemencoding() or sys.getdefaultencoding()
|
|
|
|
|
|
def _make_text_stream(stream, encoding, errors):
|
|
if encoding is None:
|
|
encoding = get_best_encoding(stream)
|
|
if errors is None:
|
|
errors = 'replace'
|
|
return _NonClosingTextIOWrapper(stream, encoding, errors,
|
|
line_buffering=True)
|
|
|
|
|
|
def is_ascii_encoding(encoding):
|
|
"""Checks if a given encoding is ascii."""
|
|
try:
|
|
return codecs.lookup(encoding).name == 'ascii'
|
|
except LookupError:
|
|
return False
|
|
|
|
|
|
def get_best_encoding(stream):
|
|
"""Returns the default stream encoding if not found."""
|
|
rv = getattr(stream, 'encoding', None) or sys.getdefaultencoding()
|
|
if is_ascii_encoding(rv):
|
|
return 'utf-8'
|
|
return rv
|
|
|
|
|
|
class _NonClosingTextIOWrapper(io.TextIOWrapper):
|
|
|
|
def __init__(self, stream, encoding, errors, **extra):
|
|
self._stream = stream = _FixupStream(stream)
|
|
io.TextIOWrapper.__init__(self, stream, encoding, errors, **extra)
|
|
|
|
# The io module is a place where the Python 3 text behavior
|
|
# was forced upon Python 2, so we need to unbreak
|
|
# it to look like Python 2.
|
|
if PY2:
|
|
def write(self, x):
|
|
if isinstance(x, str) or is_bytes(x):
|
|
try:
|
|
self.flush()
|
|
except Exception:
|
|
pass
|
|
return self.buffer.write(str(x))
|
|
return io.TextIOWrapper.write(self, x)
|
|
|
|
def writelines(self, lines):
|
|
for line in lines:
|
|
self.write(line)
|
|
|
|
def __del__(self):
|
|
try:
|
|
self.detach()
|
|
except Exception:
|
|
pass
|
|
|
|
def isatty(self):
|
|
# https://bitbucket.org/pypy/pypy/issue/1803
|
|
return self._stream.isatty()
|
|
|
|
|
|
class _FixupStream(object):
|
|
"""The new io interface needs more from streams than streams
|
|
traditionally implement. As such, this fix-up code is necessary in
|
|
some circumstances.
|
|
"""
|
|
|
|
def __init__(self, stream):
|
|
self._stream = stream
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self._stream, name)
|
|
|
|
def read1(self, size):
|
|
f = getattr(self._stream, 'read1', None)
|
|
if f is not None:
|
|
return f(size)
|
|
# We only dispatch to readline instead of read in Python 2 as we
|
|
# do not want cause problems with the different implementation
|
|
# of line buffering.
|
|
if PY2:
|
|
return self._stream.readline(size)
|
|
return self._stream.read(size)
|
|
|
|
def readable(self):
|
|
x = getattr(self._stream, 'readable', None)
|
|
if x is not None:
|
|
return x()
|
|
try:
|
|
self._stream.read(0)
|
|
except Exception:
|
|
return False
|
|
return True
|
|
|
|
def writable(self):
|
|
x = getattr(self._stream, 'writable', None)
|
|
if x is not None:
|
|
return x()
|
|
try:
|
|
self._stream.write('')
|
|
except Exception:
|
|
try:
|
|
self._stream.write(b'')
|
|
except Exception:
|
|
return False
|
|
return True
|
|
|
|
def seekable(self):
|
|
x = getattr(self._stream, 'seekable', None)
|
|
if x is not None:
|
|
return x()
|
|
try:
|
|
self._stream.seek(self._stream.tell())
|
|
except Exception:
|
|
return False
|
|
return True
|
|
|
|
|
|
if PY2:
|
|
text_type = unicode
|
|
bytes = str
|
|
raw_input = raw_input
|
|
string_types = (str, unicode)
|
|
iteritems = lambda x: x.iteritems()
|
|
range_type = xrange
|
|
|
|
def is_bytes(x):
|
|
return isinstance(x, (buffer, bytearray))
|
|
|
|
_identifier_re = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$')
|
|
|
|
# For Windows, we need to force stdout/stdin/stderr to binary if it's
|
|
# fetched for that. This obviously is not the most correct way to do
|
|
# it as it changes global state. Unfortunately, there does not seem to
|
|
# be a clear better way to do it as just reopening the file in binary
|
|
# mode does not change anything.
|
|
#
|
|
# An option would be to do what Python 3 does and to open the file as
|
|
# binary only, patch it back to the system, and then use a wrapper
|
|
# stream that converts newlines. It's not quite clear what's the
|
|
# correct option here.
|
|
#
|
|
# This code also lives in _winconsole for the fallback to the console
|
|
# emulation stream.
|
|
if WIN:
|
|
import msvcrt
|
|
def set_binary_mode(f):
|
|
try:
|
|
fileno = f.fileno()
|
|
except Exception:
|
|
pass
|
|
else:
|
|
msvcrt.setmode(fileno, os.O_BINARY)
|
|
return f
|
|
else:
|
|
set_binary_mode = lambda x: x
|
|
|
|
def isidentifier(x):
|
|
return _identifier_re.search(x) is not None
|
|
|
|
def get_binary_stdin():
|
|
return set_binary_mode(sys.stdin)
|
|
|
|
def get_binary_stdout():
|
|
return set_binary_mode(sys.stdout)
|
|
|
|
def get_binary_stderr():
|
|
return set_binary_mode(sys.stderr)
|
|
|
|
def get_text_stdin(encoding=None, errors=None):
|
|
rv = _get_windows_console_stream(sys.stdin, encoding, errors)
|
|
if rv is not None:
|
|
return rv
|
|
return _make_text_stream(sys.stdin, encoding, errors)
|
|
|
|
def get_text_stdout(encoding=None, errors=None):
|
|
rv = _get_windows_console_stream(sys.stdout, encoding, errors)
|
|
if rv is not None:
|
|
return rv
|
|
return _make_text_stream(sys.stdout, encoding, errors)
|
|
|
|
def get_text_stderr(encoding=None, errors=None):
|
|
rv = _get_windows_console_stream(sys.stderr, encoding, errors)
|
|
if rv is not None:
|
|
return rv
|
|
return _make_text_stream(sys.stderr, encoding, errors)
|
|
|
|
def filename_to_ui(value):
|
|
if isinstance(value, bytes):
|
|
value = value.decode(get_filesystem_encoding(), 'replace')
|
|
return value
|
|
else:
|
|
import io
|
|
text_type = str
|
|
raw_input = input
|
|
string_types = (str,)
|
|
range_type = range
|
|
isidentifier = lambda x: x.isidentifier()
|
|
iteritems = lambda x: iter(x.items())
|
|
|
|
def is_bytes(x):
|
|
return isinstance(x, (bytes, memoryview, bytearray))
|
|
|
|
def _is_binary_reader(stream, default=False):
|
|
try:
|
|
return isinstance(stream.read(0), bytes)
|
|
except Exception:
|
|
return default
|
|
# This happens in some cases where the stream was already
|
|
# closed. In this case, we assume the default.
|
|
|
|
def _is_binary_writer(stream, default=False):
|
|
try:
|
|
stream.write(b'')
|
|
except Exception:
|
|
try:
|
|
stream.write('')
|
|
return False
|
|
except Exception:
|
|
pass
|
|
return default
|
|
return True
|
|
|
|
def _find_binary_reader(stream):
|
|
# We need to figure out if the given stream is already binary.
|
|
# This can happen because the official docs recommend detaching
|
|
# the streams to get binary streams. Some code might do this, so
|
|
# we need to deal with this case explicitly.
|
|
if _is_binary_reader(stream, False):
|
|
return stream
|
|
|
|
buf = getattr(stream, 'buffer', None)
|
|
|
|
# Same situation here; this time we assume that the buffer is
|
|
# actually binary in case it's closed.
|
|
if buf is not None and _is_binary_reader(buf, True):
|
|
return buf
|
|
|
|
def _find_binary_writer(stream):
|
|
# We need to figure out if the given stream is already binary.
|
|
# This can happen because the official docs recommend detatching
|
|
# the streams to get binary streams. Some code might do this, so
|
|
# we need to deal with this case explicitly.
|
|
if _is_binary_writer(stream, False):
|
|
return stream
|
|
|
|
buf = getattr(stream, 'buffer', None)
|
|
|
|
# Same situation here; this time we assume that the buffer is
|
|
# actually binary in case it's closed.
|
|
if buf is not None and _is_binary_writer(buf, True):
|
|
return buf
|
|
|
|
def _stream_is_misconfigured(stream):
|
|
"""A stream is misconfigured if its encoding is ASCII."""
|
|
# If the stream does not have an encoding set, we assume it's set
|
|
# to ASCII. This appears to happen in certain unittest
|
|
# environments. It's not quite clear what the correct behavior is
|
|
# but this at least will force Click to recover somehow.
|
|
return is_ascii_encoding(getattr(stream, 'encoding', None) or 'ascii')
|
|
|
|
def _is_compatible_text_stream(stream, encoding, errors):
|
|
stream_encoding = getattr(stream, 'encoding', None)
|
|
stream_errors = getattr(stream, 'errors', None)
|
|
|
|
# Perfect match.
|
|
if stream_encoding == encoding and stream_errors == errors:
|
|
return True
|
|
|
|
# Otherwise, it's only a compatible stream if we did not ask for
|
|
# an encoding.
|
|
if encoding is None:
|
|
return stream_encoding is not None
|
|
|
|
return False
|
|
|
|
def _force_correct_text_reader(text_reader, encoding, errors):
|
|
if _is_binary_reader(text_reader, False):
|
|
binary_reader = text_reader
|
|
else:
|
|
# If there is no target encoding set, we need to verify that the
|
|
# reader is not actually misconfigured.
|
|
if encoding is None and not _stream_is_misconfigured(text_reader):
|
|
return text_reader
|
|
|
|
if _is_compatible_text_stream(text_reader, encoding, errors):
|
|
return text_reader
|
|
|
|
# If the reader has no encoding, we try to find the underlying
|
|
# binary reader for it. If that fails because the environment is
|
|
# misconfigured, we silently go with the same reader because this
|
|
# is too common to happen. In that case, mojibake is better than
|
|
# exceptions.
|
|
binary_reader = _find_binary_reader(text_reader)
|
|
if binary_reader is None:
|
|
return text_reader
|
|
|
|
# At this point, we default the errors to replace instead of strict
|
|
# because nobody handles those errors anyways and at this point
|
|
# we're so fundamentally fucked that nothing can repair it.
|
|
if errors is None:
|
|
errors = 'replace'
|
|
return _make_text_stream(binary_reader, encoding, errors)
|
|
|
|
def _force_correct_text_writer(text_writer, encoding, errors):
|
|
if _is_binary_writer(text_writer, False):
|
|
binary_writer = text_writer
|
|
else:
|
|
# If there is no target encoding set, we need to verify that the
|
|
# writer is not actually misconfigured.
|
|
if encoding is None and not _stream_is_misconfigured(text_writer):
|
|
return text_writer
|
|
|
|
if _is_compatible_text_stream(text_writer, encoding, errors):
|
|
return text_writer
|
|
|
|
# If the writer has no encoding, we try to find the underlying
|
|
# binary writer for it. If that fails because the environment is
|
|
# misconfigured, we silently go with the same writer because this
|
|
# is too common to happen. In that case, mojibake is better than
|
|
# exceptions.
|
|
binary_writer = _find_binary_writer(text_writer)
|
|
if binary_writer is None:
|
|
return text_writer
|
|
|
|
# At this point, we default the errors to replace instead of strict
|
|
# because nobody handles those errors anyways and at this point
|
|
# we're so fundamentally fucked that nothing can repair it.
|
|
if errors is None:
|
|
errors = 'replace'
|
|
return _make_text_stream(binary_writer, encoding, errors)
|
|
|
|
def get_binary_stdin():
|
|
reader = _find_binary_reader(sys.stdin)
|
|
if reader is None:
|
|
raise RuntimeError('Was not able to determine binary '
|
|
'stream for sys.stdin.')
|
|
return reader
|
|
|
|
def get_binary_stdout():
|
|
writer = _find_binary_writer(sys.stdout)
|
|
if writer is None:
|
|
raise RuntimeError('Was not able to determine binary '
|
|
'stream for sys.stdout.')
|
|
return writer
|
|
|
|
def get_binary_stderr():
|
|
writer = _find_binary_writer(sys.stderr)
|
|
if writer is None:
|
|
raise RuntimeError('Was not able to determine binary '
|
|
'stream for sys.stderr.')
|
|
return writer
|
|
|
|
def get_text_stdin(encoding=None, errors=None):
|
|
rv = _get_windows_console_stream(sys.stdin, encoding, errors)
|
|
if rv is not None:
|
|
return rv
|
|
return _force_correct_text_reader(sys.stdin, encoding, errors)
|
|
|
|
def get_text_stdout(encoding=None, errors=None):
|
|
rv = _get_windows_console_stream(sys.stdout, encoding, errors)
|
|
if rv is not None:
|
|
return rv
|
|
return _force_correct_text_writer(sys.stdout, encoding, errors)
|
|
|
|
def get_text_stderr(encoding=None, errors=None):
|
|
rv = _get_windows_console_stream(sys.stderr, encoding, errors)
|
|
if rv is not None:
|
|
return rv
|
|
return _force_correct_text_writer(sys.stderr, encoding, errors)
|
|
|
|
def filename_to_ui(value):
|
|
if isinstance(value, bytes):
|
|
value = value.decode(get_filesystem_encoding(), 'replace')
|
|
else:
|
|
value = value.encode('utf-8', 'surrogateescape') \
|
|
.decode('utf-8', 'replace')
|
|
return value
|
|
|
|
|
|
def get_streerror(e, default=None):
|
|
if hasattr(e, 'strerror'):
|
|
msg = e.strerror
|
|
else:
|
|
if default is not None:
|
|
msg = default
|
|
else:
|
|
msg = str(e)
|
|
if isinstance(msg, bytes):
|
|
msg = msg.decode('utf-8', 'replace')
|
|
return msg
|
|
|
|
|
|
def open_stream(filename, mode='r', encoding=None, errors='strict',
|
|
atomic=False):
|
|
# Standard streams first. These are simple because they don't need
|
|
# special handling for the atomic flag. It's entirely ignored.
|
|
if filename == '-':
|
|
if 'w' in mode:
|
|
if 'b' in mode:
|
|
return get_binary_stdout(), False
|
|
return get_text_stdout(encoding=encoding, errors=errors), False
|
|
if 'b' in mode:
|
|
return get_binary_stdin(), False
|
|
return get_text_stdin(encoding=encoding, errors=errors), False
|
|
|
|
# Non-atomic writes directly go out through the regular open functions.
|
|
if not atomic:
|
|
if encoding is None:
|
|
return open(filename, mode), True
|
|
return io.open(filename, mode, encoding=encoding, errors=errors), True
|
|
|
|
# Some usability stuff for atomic writes
|
|
if 'a' in mode:
|
|
raise ValueError(
|
|
'Appending to an existing file is not supported, because that '
|
|
'would involve an expensive `copy`-operation to a temporary '
|
|
'file. Open the file in normal `w`-mode and copy explicitly '
|
|
'if that\'s what you\'re after.'
|
|
)
|
|
if 'x' in mode:
|
|
raise ValueError('Use the `overwrite`-parameter instead.')
|
|
if 'w' not in mode:
|
|
raise ValueError('Atomic writes only make sense with `w`-mode.')
|
|
|
|
# Atomic writes are more complicated. They work by opening a file
|
|
# as a proxy in the same folder and then using the fdopen
|
|
# functionality to wrap it in a Python file. Then we wrap it in an
|
|
# atomic file that moves the file over on close.
|
|
import tempfile
|
|
fd, tmp_filename = tempfile.mkstemp(dir=os.path.dirname(filename),
|
|
prefix='.__atomic-write')
|
|
|
|
if encoding is not None:
|
|
f = io.open(fd, mode, encoding=encoding, errors=errors)
|
|
else:
|
|
f = os.fdopen(fd, mode)
|
|
|
|
return _AtomicFile(f, tmp_filename, filename), True
|
|
|
|
|
|
# Used in a destructor call, needs extra protection from interpreter cleanup.
|
|
if hasattr(os, 'replace'):
|
|
_replace = os.replace
|
|
_can_replace = True
|
|
else:
|
|
_replace = os.rename
|
|
_can_replace = not WIN
|
|
|
|
|
|
class _AtomicFile(object):
|
|
|
|
def __init__(self, f, tmp_filename, real_filename):
|
|
self._f = f
|
|
self._tmp_filename = tmp_filename
|
|
self._real_filename = real_filename
|
|
self.closed = False
|
|
|
|
@property
|
|
def name(self):
|
|
return self._real_filename
|
|
|
|
def close(self, delete=False):
|
|
if self.closed:
|
|
return
|
|
self._f.close()
|
|
if not _can_replace:
|
|
try:
|
|
os.remove(self._real_filename)
|
|
except OSError:
|
|
pass
|
|
_replace(self._tmp_filename, self._real_filename)
|
|
self.closed = True
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self._f, name)
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_value, tb):
|
|
self.close(delete=exc_type is not None)
|
|
|
|
def __repr__(self):
|
|
return repr(self._f)
|
|
|
|
|
|
auto_wrap_for_ansi = None
|
|
colorama = None
|
|
get_winterm_size = None
|
|
|
|
|
|
def strip_ansi(value):
|
|
return _ansi_re.sub('', value)
|
|
|
|
|
|
def should_strip_ansi(stream=None, color=None):
|
|
if color is None:
|
|
if stream is None:
|
|
stream = sys.stdin
|
|
return not isatty(stream)
|
|
return not color
|
|
|
|
|
|
# If we're on Windows, we provide transparent integration through
|
|
# colorama. This will make ANSI colors through the echo function
|
|
# work automatically.
|
|
if WIN:
|
|
# Windows has a smaller terminal
|
|
DEFAULT_COLUMNS = 79
|
|
|
|
from ._winconsole import _get_windows_console_stream
|
|
|
|
def _get_argv_encoding():
|
|
import locale
|
|
return locale.getpreferredencoding()
|
|
|
|
if PY2:
|
|
def raw_input(prompt=''):
|
|
sys.stderr.flush()
|
|
if prompt:
|
|
stdout = _default_text_stdout()
|
|
stdout.write(prompt)
|
|
stdin = _default_text_stdin()
|
|
return stdin.readline().rstrip('\r\n')
|
|
|
|
try:
|
|
import colorama
|
|
except ImportError:
|
|
pass
|
|
else:
|
|
_ansi_stream_wrappers = WeakKeyDictionary()
|
|
|
|
def auto_wrap_for_ansi(stream, color=None):
|
|
"""This function wraps a stream so that calls through colorama
|
|
are issued to the win32 console API to recolor on demand. It
|
|
also ensures to reset the colors if a write call is interrupted
|
|
to not destroy the console afterwards.
|
|
"""
|
|
try:
|
|
cached = _ansi_stream_wrappers.get(stream)
|
|
except Exception:
|
|
cached = None
|
|
if cached is not None:
|
|
return cached
|
|
strip = should_strip_ansi(stream, color)
|
|
ansi_wrapper = colorama.AnsiToWin32(stream, strip=strip)
|
|
rv = ansi_wrapper.stream
|
|
_write = rv.write
|
|
|
|
def _safe_write(s):
|
|
try:
|
|
return _write(s)
|
|
except:
|
|
ansi_wrapper.reset_all()
|
|
raise
|
|
|
|
rv.write = _safe_write
|
|
try:
|
|
_ansi_stream_wrappers[stream] = rv
|
|
except Exception:
|
|
pass
|
|
return rv
|
|
|
|
def get_winterm_size():
|
|
win = colorama.win32.GetConsoleScreenBufferInfo(
|
|
colorama.win32.STDOUT).srWindow
|
|
return win.Right - win.Left, win.Bottom - win.Top
|
|
else:
|
|
def _get_argv_encoding():
|
|
return getattr(sys.stdin, 'encoding', None) or get_filesystem_encoding()
|
|
|
|
_get_windows_console_stream = lambda *x: None
|
|
|
|
|
|
def term_len(x):
|
|
return len(strip_ansi(x))
|
|
|
|
|
|
def isatty(stream):
|
|
try:
|
|
return stream.isatty()
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _make_cached_stream_func(src_func, wrapper_func):
|
|
cache = WeakKeyDictionary()
|
|
def func():
|
|
stream = src_func()
|
|
try:
|
|
rv = cache.get(stream)
|
|
except Exception:
|
|
rv = None
|
|
if rv is not None:
|
|
return rv
|
|
rv = wrapper_func()
|
|
try:
|
|
cache[stream] = rv
|
|
except Exception:
|
|
pass
|
|
return rv
|
|
return func
|
|
|
|
|
|
_default_text_stdin = _make_cached_stream_func(
|
|
lambda: sys.stdin, get_text_stdin)
|
|
_default_text_stdout = _make_cached_stream_func(
|
|
lambda: sys.stdout, get_text_stdout)
|
|
_default_text_stderr = _make_cached_stream_func(
|
|
lambda: sys.stderr, get_text_stderr)
|
|
|
|
|
|
binary_streams = {
|
|
'stdin': get_binary_stdin,
|
|
'stdout': get_binary_stdout,
|
|
'stderr': get_binary_stderr,
|
|
}
|
|
|
|
text_streams = {
|
|
'stdin': get_text_stdin,
|
|
'stdout': get_text_stdout,
|
|
'stderr': get_text_stderr,
|
|
}
|