mirror of
https://github.com/clinton-hall/nzbToMedia.git
synced 2025-08-20 21:33:13 -07:00
Updates vendored subliminal to 2.1.0
Updates rarfile to 3.1 Updates stevedore to 3.5.0 Updates appdirs to 1.4.4 Updates click to 8.1.3 Updates decorator to 5.1.1 Updates dogpile.cache to 1.1.8 Updates pbr to 5.11.0 Updates pysrt to 1.1.2 Updates pytz to 2022.6 Adds importlib-metadata version 3.1.1 Adds typing-extensions version 4.1.1 Adds zipp version 3.11.0
This commit is contained in:
Labrys of Knossos
parent
d8da02cb69
commit
f05b09f349
694 changed files with 16621 additions and 11056 deletions
|
|
@ -1,62 +1,56 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
click._termui_impl
|
||||
~~~~~~~~~~~~~~~~~~
|
||||
|
||||
This module contains implementations for the termui module. To keep the
|
||||
import time of Click down, some infrequently used functionality is
|
||||
placed in this module and only imported as needed.
|
||||
|
||||
:copyright: © 2014 by the Pallets team.
|
||||
:license: BSD, see LICENSE.rst for more details.
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
import math
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import math
|
||||
import contextlib
|
||||
from ._compat import _default_text_stdout, range_type, PY2, isatty, \
|
||||
open_stream, strip_ansi, term_len, get_best_encoding, WIN, int_types, \
|
||||
CYGWIN
|
||||
from .utils import echo
|
||||
import typing as t
|
||||
from gettext import gettext as _
|
||||
|
||||
from ._compat import _default_text_stdout
|
||||
from ._compat import CYGWIN
|
||||
from ._compat import get_best_encoding
|
||||
from ._compat import isatty
|
||||
from ._compat import open_stream
|
||||
from ._compat import strip_ansi
|
||||
from ._compat import term_len
|
||||
from ._compat import WIN
|
||||
from .exceptions import ClickException
|
||||
from .utils import echo
|
||||
|
||||
V = t.TypeVar("V")
|
||||
|
||||
if os.name == 'nt':
|
||||
BEFORE_BAR = '\r'
|
||||
AFTER_BAR = '\n'
|
||||
if os.name == "nt":
|
||||
BEFORE_BAR = "\r"
|
||||
AFTER_BAR = "\n"
|
||||
else:
|
||||
BEFORE_BAR = '\r\033[?25l'
|
||||
AFTER_BAR = '\033[?25h\n'
|
||||
BEFORE_BAR = "\r\033[?25l"
|
||||
AFTER_BAR = "\033[?25h\n"
|
||||
|
||||
|
||||
def _length_hint(obj):
|
||||
"""Returns the length hint of an object."""
|
||||
try:
|
||||
return len(obj)
|
||||
except (AttributeError, TypeError):
|
||||
try:
|
||||
get_hint = type(obj).__length_hint__
|
||||
except AttributeError:
|
||||
return None
|
||||
try:
|
||||
hint = get_hint(obj)
|
||||
except TypeError:
|
||||
return None
|
||||
if hint is NotImplemented or \
|
||||
not isinstance(hint, int_types) or \
|
||||
hint < 0:
|
||||
return None
|
||||
return hint
|
||||
|
||||
|
||||
class ProgressBar(object):
|
||||
|
||||
def __init__(self, iterable, length=None, fill_char='#', empty_char=' ',
|
||||
bar_template='%(bar)s', info_sep=' ', show_eta=True,
|
||||
show_percent=None, show_pos=False, item_show_func=None,
|
||||
label=None, file=None, color=None, width=30):
|
||||
class ProgressBar(t.Generic[V]):
|
||||
def __init__(
|
||||
self,
|
||||
iterable: t.Optional[t.Iterable[V]],
|
||||
length: t.Optional[int] = None,
|
||||
fill_char: str = "#",
|
||||
empty_char: str = " ",
|
||||
bar_template: str = "%(bar)s",
|
||||
info_sep: str = " ",
|
||||
show_eta: bool = True,
|
||||
show_percent: t.Optional[bool] = None,
|
||||
show_pos: bool = False,
|
||||
item_show_func: t.Optional[t.Callable[[t.Optional[V]], t.Optional[str]]] = None,
|
||||
label: t.Optional[str] = None,
|
||||
file: t.Optional[t.TextIO] = None,
|
||||
color: t.Optional[bool] = None,
|
||||
update_min_steps: int = 1,
|
||||
width: int = 30,
|
||||
) -> None:
|
||||
self.fill_char = fill_char
|
||||
self.empty_char = empty_char
|
||||
self.bar_template = bar_template
|
||||
|
|
@ -65,77 +59,87 @@ class ProgressBar(object):
|
|||
self.show_percent = show_percent
|
||||
self.show_pos = show_pos
|
||||
self.item_show_func = item_show_func
|
||||
self.label = label or ''
|
||||
self.label = label or ""
|
||||
if file is None:
|
||||
file = _default_text_stdout()
|
||||
self.file = file
|
||||
self.color = color
|
||||
self.update_min_steps = update_min_steps
|
||||
self._completed_intervals = 0
|
||||
self.width = width
|
||||
self.autowidth = width == 0
|
||||
|
||||
if length is None:
|
||||
length = _length_hint(iterable)
|
||||
from operator import length_hint
|
||||
|
||||
length = length_hint(iterable, -1)
|
||||
|
||||
if length == -1:
|
||||
length = None
|
||||
if iterable is None:
|
||||
if length is None:
|
||||
raise TypeError('iterable or length is required')
|
||||
iterable = range_type(length)
|
||||
raise TypeError("iterable or length is required")
|
||||
iterable = t.cast(t.Iterable[V], range(length))
|
||||
self.iter = iter(iterable)
|
||||
self.length = length
|
||||
self.length_known = length is not None
|
||||
self.pos = 0
|
||||
self.avg = []
|
||||
self.avg: t.List[float] = []
|
||||
self.start = self.last_eta = time.time()
|
||||
self.eta_known = False
|
||||
self.finished = False
|
||||
self.max_width = None
|
||||
self.max_width: t.Optional[int] = None
|
||||
self.entered = False
|
||||
self.current_item = None
|
||||
self.current_item: t.Optional[V] = None
|
||||
self.is_hidden = not isatty(self.file)
|
||||
self._last_line = None
|
||||
self.short_limit = 0.5
|
||||
self._last_line: t.Optional[str] = None
|
||||
|
||||
def __enter__(self):
|
||||
def __enter__(self) -> "ProgressBar":
|
||||
self.entered = True
|
||||
self.render_progress()
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_value, tb):
|
||||
def __exit__(self, exc_type, exc_value, tb): # type: ignore
|
||||
self.render_finish()
|
||||
|
||||
def __iter__(self):
|
||||
def __iter__(self) -> t.Iterator[V]:
|
||||
if not self.entered:
|
||||
raise RuntimeError('You need to use progress bars in a with block.')
|
||||
raise RuntimeError("You need to use progress bars in a with block.")
|
||||
self.render_progress()
|
||||
return self.generator()
|
||||
|
||||
def is_fast(self):
|
||||
return time.time() - self.start <= self.short_limit
|
||||
def __next__(self) -> V:
|
||||
# Iteration is defined in terms of a generator function,
|
||||
# returned by iter(self); use that to define next(). This works
|
||||
# because `self.iter` is an iterable consumed by that generator,
|
||||
# so it is re-entry safe. Calling `next(self.generator())`
|
||||
# twice works and does "what you want".
|
||||
return next(iter(self))
|
||||
|
||||
def render_finish(self):
|
||||
if self.is_hidden or self.is_fast():
|
||||
def render_finish(self) -> None:
|
||||
if self.is_hidden:
|
||||
return
|
||||
self.file.write(AFTER_BAR)
|
||||
self.file.flush()
|
||||
|
||||
@property
|
||||
def pct(self):
|
||||
def pct(self) -> float:
|
||||
if self.finished:
|
||||
return 1.0
|
||||
return min(self.pos / (float(self.length) or 1), 1.0)
|
||||
return min(self.pos / (float(self.length or 1) or 1), 1.0)
|
||||
|
||||
@property
|
||||
def time_per_iteration(self):
|
||||
def time_per_iteration(self) -> float:
|
||||
if not self.avg:
|
||||
return 0.0
|
||||
return sum(self.avg) / float(len(self.avg))
|
||||
|
||||
@property
|
||||
def eta(self):
|
||||
if self.length_known and not self.finished:
|
||||
def eta(self) -> float:
|
||||
if self.length is not None and not self.finished:
|
||||
return self.time_per_iteration * (self.length - self.pos)
|
||||
return 0.0
|
||||
|
||||
def format_eta(self):
|
||||
def format_eta(self) -> str:
|
||||
if self.eta_known:
|
||||
t = int(self.eta)
|
||||
seconds = t % 60
|
||||
|
|
@ -145,41 +149,44 @@ class ProgressBar(object):
|
|||
hours = t % 24
|
||||
t //= 24
|
||||
if t > 0:
|
||||
days = t
|
||||
return '%dd %02d:%02d:%02d' % (days, hours, minutes, seconds)
|
||||
return f"{t}d {hours:02}:{minutes:02}:{seconds:02}"
|
||||
else:
|
||||
return '%02d:%02d:%02d' % (hours, minutes, seconds)
|
||||
return ''
|
||||
return f"{hours:02}:{minutes:02}:{seconds:02}"
|
||||
return ""
|
||||
|
||||
def format_pos(self):
|
||||
def format_pos(self) -> str:
|
||||
pos = str(self.pos)
|
||||
if self.length_known:
|
||||
pos += '/%s' % self.length
|
||||
if self.length is not None:
|
||||
pos += f"/{self.length}"
|
||||
return pos
|
||||
|
||||
def format_pct(self):
|
||||
return ('% 4d%%' % int(self.pct * 100))[1:]
|
||||
def format_pct(self) -> str:
|
||||
return f"{int(self.pct * 100): 4}%"[1:]
|
||||
|
||||
def format_bar(self):
|
||||
if self.length_known:
|
||||
def format_bar(self) -> str:
|
||||
if self.length is not None:
|
||||
bar_length = int(self.pct * self.width)
|
||||
bar = self.fill_char * bar_length
|
||||
bar += self.empty_char * (self.width - bar_length)
|
||||
elif self.finished:
|
||||
bar = self.fill_char * self.width
|
||||
else:
|
||||
bar = list(self.empty_char * (self.width or 1))
|
||||
chars = list(self.empty_char * (self.width or 1))
|
||||
if self.time_per_iteration != 0:
|
||||
bar[int((math.cos(self.pos * self.time_per_iteration)
|
||||
/ 2.0 + 0.5) * self.width)] = self.fill_char
|
||||
bar = ''.join(bar)
|
||||
chars[
|
||||
int(
|
||||
(math.cos(self.pos * self.time_per_iteration) / 2.0 + 0.5)
|
||||
* self.width
|
||||
)
|
||||
] = self.fill_char
|
||||
bar = "".join(chars)
|
||||
return bar
|
||||
|
||||
def format_progress_line(self):
|
||||
def format_progress_line(self) -> str:
|
||||
show_percent = self.show_percent
|
||||
|
||||
info_bits = []
|
||||
if self.length_known and show_percent is None:
|
||||
if self.length is not None and show_percent is None:
|
||||
show_percent = not self.show_pos
|
||||
|
||||
if self.show_pos:
|
||||
|
|
@ -193,16 +200,25 @@ class ProgressBar(object):
|
|||
if item_info is not None:
|
||||
info_bits.append(item_info)
|
||||
|
||||
return (self.bar_template % {
|
||||
'label': self.label,
|
||||
'bar': self.format_bar(),
|
||||
'info': self.info_sep.join(info_bits)
|
||||
}).rstrip()
|
||||
return (
|
||||
self.bar_template
|
||||
% {
|
||||
"label": self.label,
|
||||
"bar": self.format_bar(),
|
||||
"info": self.info_sep.join(info_bits),
|
||||
}
|
||||
).rstrip()
|
||||
|
||||
def render_progress(self):
|
||||
from .termui import get_terminal_size
|
||||
def render_progress(self) -> None:
|
||||
import shutil
|
||||
|
||||
if self.is_hidden:
|
||||
# Only output the label as it changes if the output is not a
|
||||
# TTY. Use file=stderr if you expect to be piping stdout.
|
||||
if self._last_line != self.label:
|
||||
self._last_line = self.label
|
||||
echo(self.label, file=self.file, color=self.color)
|
||||
|
||||
return
|
||||
|
||||
buf = []
|
||||
|
|
@ -211,10 +227,10 @@ class ProgressBar(object):
|
|||
old_width = self.width
|
||||
self.width = 0
|
||||
clutter_length = term_len(self.format_progress_line())
|
||||
new_width = max(0, get_terminal_size()[0] - clutter_length)
|
||||
new_width = max(0, shutil.get_terminal_size().columns - clutter_length)
|
||||
if new_width < old_width:
|
||||
buf.append(BEFORE_BAR)
|
||||
buf.append(' ' * self.max_width)
|
||||
buf.append(" " * self.max_width) # type: ignore
|
||||
self.max_width = new_width
|
||||
self.width = new_width
|
||||
|
||||
|
|
@ -229,18 +245,18 @@ class ProgressBar(object):
|
|||
self.max_width = line_len
|
||||
|
||||
buf.append(line)
|
||||
buf.append(' ' * (clear_width - line_len))
|
||||
line = ''.join(buf)
|
||||
buf.append(" " * (clear_width - line_len))
|
||||
line = "".join(buf)
|
||||
# Render the line only if it changed.
|
||||
|
||||
if line != self._last_line and not self.is_fast():
|
||||
if line != self._last_line:
|
||||
self._last_line = line
|
||||
echo(line, file=self.file, color=self.color, nl=False)
|
||||
self.file.flush()
|
||||
|
||||
def make_step(self, n_steps):
|
||||
def make_step(self, n_steps: int) -> None:
|
||||
self.pos += n_steps
|
||||
if self.length_known and self.pos >= self.length:
|
||||
if self.length is not None and self.pos >= self.length:
|
||||
self.finished = True
|
||||
|
||||
if (time.time() - self.last_eta) < 1.0:
|
||||
|
|
@ -258,97 +274,134 @@ class ProgressBar(object):
|
|||
|
||||
self.avg = self.avg[-6:] + [step]
|
||||
|
||||
self.eta_known = self.length_known
|
||||
self.eta_known = self.length is not None
|
||||
|
||||
def update(self, n_steps):
|
||||
self.make_step(n_steps)
|
||||
self.render_progress()
|
||||
def update(self, n_steps: int, current_item: t.Optional[V] = None) -> None:
|
||||
"""Update the progress bar by advancing a specified number of
|
||||
steps, and optionally set the ``current_item`` for this new
|
||||
position.
|
||||
|
||||
def finish(self):
|
||||
self.eta_known = 0
|
||||
:param n_steps: Number of steps to advance.
|
||||
:param current_item: Optional item to set as ``current_item``
|
||||
for the updated position.
|
||||
|
||||
.. versionchanged:: 8.0
|
||||
Added the ``current_item`` optional parameter.
|
||||
|
||||
.. versionchanged:: 8.0
|
||||
Only render when the number of steps meets the
|
||||
``update_min_steps`` threshold.
|
||||
"""
|
||||
if current_item is not None:
|
||||
self.current_item = current_item
|
||||
|
||||
self._completed_intervals += n_steps
|
||||
|
||||
if self._completed_intervals >= self.update_min_steps:
|
||||
self.make_step(self._completed_intervals)
|
||||
self.render_progress()
|
||||
self._completed_intervals = 0
|
||||
|
||||
def finish(self) -> None:
|
||||
self.eta_known = False
|
||||
self.current_item = None
|
||||
self.finished = True
|
||||
|
||||
def generator(self):
|
||||
"""
|
||||
Returns a generator which yields the items added to the bar during
|
||||
construction, and updates the progress bar *after* the yielded block
|
||||
returns.
|
||||
def generator(self) -> t.Iterator[V]:
|
||||
"""Return a generator which yields the items added to the bar
|
||||
during construction, and updates the progress bar *after* the
|
||||
yielded block returns.
|
||||
"""
|
||||
# WARNING: the iterator interface for `ProgressBar` relies on
|
||||
# this and only works because this is a simple generator which
|
||||
# doesn't create or manage additional state. If this function
|
||||
# changes, the impact should be evaluated both against
|
||||
# `iter(bar)` and `next(bar)`. `next()` in particular may call
|
||||
# `self.generator()` repeatedly, and this must remain safe in
|
||||
# order for that interface to work.
|
||||
if not self.entered:
|
||||
raise RuntimeError('You need to use progress bars in a with block.')
|
||||
raise RuntimeError("You need to use progress bars in a with block.")
|
||||
|
||||
if self.is_hidden:
|
||||
for rv in self.iter:
|
||||
yield rv
|
||||
yield from self.iter
|
||||
else:
|
||||
for rv in self.iter:
|
||||
self.current_item = rv
|
||||
|
||||
# This allows show_item_func to be updated before the
|
||||
# item is processed. Only trigger at the beginning of
|
||||
# the update interval.
|
||||
if self._completed_intervals == 0:
|
||||
self.render_progress()
|
||||
|
||||
yield rv
|
||||
self.update(1)
|
||||
|
||||
self.finish()
|
||||
self.render_progress()
|
||||
|
||||
|
||||
def pager(generator, color=None):
|
||||
def pager(generator: t.Iterable[str], color: t.Optional[bool] = None) -> None:
|
||||
"""Decide what method to use for paging through text."""
|
||||
stdout = _default_text_stdout()
|
||||
if not isatty(sys.stdin) or not isatty(stdout):
|
||||
return _nullpager(stdout, generator, color)
|
||||
pager_cmd = (os.environ.get('PAGER', None) or '').strip()
|
||||
pager_cmd = (os.environ.get("PAGER", None) or "").strip()
|
||||
if pager_cmd:
|
||||
if WIN:
|
||||
return _tempfilepager(generator, pager_cmd, color)
|
||||
return _pipepager(generator, pager_cmd, color)
|
||||
if os.environ.get('TERM') in ('dumb', 'emacs'):
|
||||
if os.environ.get("TERM") in ("dumb", "emacs"):
|
||||
return _nullpager(stdout, generator, color)
|
||||
if WIN or sys.platform.startswith('os2'):
|
||||
return _tempfilepager(generator, 'more <', color)
|
||||
if hasattr(os, 'system') and os.system('(less) 2>/dev/null') == 0:
|
||||
return _pipepager(generator, 'less', color)
|
||||
if WIN or sys.platform.startswith("os2"):
|
||||
return _tempfilepager(generator, "more <", color)
|
||||
if hasattr(os, "system") and os.system("(less) 2>/dev/null") == 0:
|
||||
return _pipepager(generator, "less", color)
|
||||
|
||||
import tempfile
|
||||
|
||||
fd, filename = tempfile.mkstemp()
|
||||
os.close(fd)
|
||||
try:
|
||||
if hasattr(os, 'system') and os.system('more "%s"' % filename) == 0:
|
||||
return _pipepager(generator, 'more', color)
|
||||
if hasattr(os, "system") and os.system(f'more "{filename}"') == 0:
|
||||
return _pipepager(generator, "more", color)
|
||||
return _nullpager(stdout, generator, color)
|
||||
finally:
|
||||
os.unlink(filename)
|
||||
|
||||
|
||||
def _pipepager(generator, cmd, color):
|
||||
def _pipepager(generator: t.Iterable[str], cmd: str, color: t.Optional[bool]) -> None:
|
||||
"""Page through text by feeding it to another program. Invoking a
|
||||
pager through this might support colors.
|
||||
"""
|
||||
import subprocess
|
||||
|
||||
env = dict(os.environ)
|
||||
|
||||
# If we're piping to less we might support colors under the
|
||||
# condition that
|
||||
cmd_detail = cmd.rsplit('/', 1)[-1].split()
|
||||
if color is None and cmd_detail[0] == 'less':
|
||||
less_flags = os.environ.get('LESS', '') + ' '.join(cmd_detail[1:])
|
||||
cmd_detail = cmd.rsplit("/", 1)[-1].split()
|
||||
if color is None and cmd_detail[0] == "less":
|
||||
less_flags = f"{os.environ.get('LESS', '')}{' '.join(cmd_detail[1:])}"
|
||||
if not less_flags:
|
||||
env['LESS'] = '-R'
|
||||
env["LESS"] = "-R"
|
||||
color = True
|
||||
elif 'r' in less_flags or 'R' in less_flags:
|
||||
elif "r" in less_flags or "R" in less_flags:
|
||||
color = True
|
||||
|
||||
c = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE,
|
||||
env=env)
|
||||
encoding = get_best_encoding(c.stdin)
|
||||
c = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, env=env)
|
||||
stdin = t.cast(t.BinaryIO, c.stdin)
|
||||
encoding = get_best_encoding(stdin)
|
||||
try:
|
||||
for text in generator:
|
||||
if not color:
|
||||
text = strip_ansi(text)
|
||||
|
||||
c.stdin.write(text.encode(encoding, 'replace'))
|
||||
except (IOError, KeyboardInterrupt):
|
||||
stdin.write(text.encode(encoding, "replace"))
|
||||
except (OSError, KeyboardInterrupt):
|
||||
pass
|
||||
else:
|
||||
c.stdin.close()
|
||||
stdin.close()
|
||||
|
||||
# Less doesn't respect ^C, but catches it for its own UI purposes (aborting
|
||||
# search or other commands inside less).
|
||||
|
|
@ -367,24 +420,30 @@ def _pipepager(generator, cmd, color):
|
|||
break
|
||||
|
||||
|
||||
def _tempfilepager(generator, cmd, color):
|
||||
def _tempfilepager(
|
||||
generator: t.Iterable[str], cmd: str, color: t.Optional[bool]
|
||||
) -> None:
|
||||
"""Page through text by invoking a program on a temporary file."""
|
||||
import tempfile
|
||||
filename = tempfile.mktemp()
|
||||
|
||||
fd, filename = tempfile.mkstemp()
|
||||
# TODO: This never terminates if the passed generator never terminates.
|
||||
text = "".join(generator)
|
||||
if not color:
|
||||
text = strip_ansi(text)
|
||||
encoding = get_best_encoding(sys.stdout)
|
||||
with open_stream(filename, 'wb')[0] as f:
|
||||
with open_stream(filename, "wb")[0] as f:
|
||||
f.write(text.encode(encoding))
|
||||
try:
|
||||
os.system(cmd + ' "' + filename + '"')
|
||||
os.system(f'{cmd} "{filename}"')
|
||||
finally:
|
||||
os.close(fd)
|
||||
os.unlink(filename)
|
||||
|
||||
|
||||
def _nullpager(stream, generator, color):
|
||||
def _nullpager(
|
||||
stream: t.TextIO, generator: t.Iterable[str], color: t.Optional[bool]
|
||||
) -> None:
|
||||
"""Simply print unformatted text. This is the ultimate fallback."""
|
||||
for text in generator:
|
||||
if not color:
|
||||
|
|
@ -392,159 +451,184 @@ def _nullpager(stream, generator, color):
|
|||
stream.write(text)
|
||||
|
||||
|
||||
class Editor(object):
|
||||
|
||||
def __init__(self, editor=None, env=None, require_save=True,
|
||||
extension='.txt'):
|
||||
class Editor:
|
||||
def __init__(
|
||||
self,
|
||||
editor: t.Optional[str] = None,
|
||||
env: t.Optional[t.Mapping[str, str]] = None,
|
||||
require_save: bool = True,
|
||||
extension: str = ".txt",
|
||||
) -> None:
|
||||
self.editor = editor
|
||||
self.env = env
|
||||
self.require_save = require_save
|
||||
self.extension = extension
|
||||
|
||||
def get_editor(self):
|
||||
def get_editor(self) -> str:
|
||||
if self.editor is not None:
|
||||
return self.editor
|
||||
for key in 'VISUAL', 'EDITOR':
|
||||
for key in "VISUAL", "EDITOR":
|
||||
rv = os.environ.get(key)
|
||||
if rv:
|
||||
return rv
|
||||
if WIN:
|
||||
return 'notepad'
|
||||
for editor in 'vim', 'nano':
|
||||
if os.system('which %s >/dev/null 2>&1' % editor) == 0:
|
||||
return "notepad"
|
||||
for editor in "sensible-editor", "vim", "nano":
|
||||
if os.system(f"which {editor} >/dev/null 2>&1") == 0:
|
||||
return editor
|
||||
return 'vi'
|
||||
return "vi"
|
||||
|
||||
def edit_file(self, filename):
|
||||
def edit_file(self, filename: str) -> None:
|
||||
import subprocess
|
||||
|
||||
editor = self.get_editor()
|
||||
environ: t.Optional[t.Dict[str, str]] = None
|
||||
|
||||
if self.env:
|
||||
environ = os.environ.copy()
|
||||
environ.update(self.env)
|
||||
else:
|
||||
environ = None
|
||||
|
||||
try:
|
||||
c = subprocess.Popen('%s "%s"' % (editor, filename),
|
||||
env=environ, shell=True)
|
||||
c = subprocess.Popen(f'{editor} "{filename}"', env=environ, shell=True)
|
||||
exit_code = c.wait()
|
||||
if exit_code != 0:
|
||||
raise ClickException('%s: Editing failed!' % editor)
|
||||
raise ClickException(
|
||||
_("{editor}: Editing failed").format(editor=editor)
|
||||
)
|
||||
except OSError as e:
|
||||
raise ClickException('%s: Editing failed: %s' % (editor, e))
|
||||
raise ClickException(
|
||||
_("{editor}: Editing failed: {e}").format(editor=editor, e=e)
|
||||
) from e
|
||||
|
||||
def edit(self, text):
|
||||
def edit(self, text: t.Optional[t.AnyStr]) -> t.Optional[t.AnyStr]:
|
||||
import tempfile
|
||||
|
||||
text = text or ''
|
||||
if text and not text.endswith('\n'):
|
||||
text += '\n'
|
||||
if not text:
|
||||
data = b""
|
||||
elif isinstance(text, (bytes, bytearray)):
|
||||
data = text
|
||||
else:
|
||||
if text and not text.endswith("\n"):
|
||||
text += "\n"
|
||||
|
||||
fd, name = tempfile.mkstemp(prefix='editor-', suffix=self.extension)
|
||||
try:
|
||||
if WIN:
|
||||
encoding = 'utf-8-sig'
|
||||
text = text.replace('\n', '\r\n')
|
||||
data = text.replace("\n", "\r\n").encode("utf-8-sig")
|
||||
else:
|
||||
encoding = 'utf-8'
|
||||
text = text.encode(encoding)
|
||||
data = text.encode("utf-8")
|
||||
|
||||
f = os.fdopen(fd, 'wb')
|
||||
f.write(text)
|
||||
f.close()
|
||||
fd, name = tempfile.mkstemp(prefix="editor-", suffix=self.extension)
|
||||
f: t.BinaryIO
|
||||
|
||||
try:
|
||||
with os.fdopen(fd, "wb") as f:
|
||||
f.write(data)
|
||||
|
||||
# If the filesystem resolution is 1 second, like Mac OS
|
||||
# 10.12 Extended, or 2 seconds, like FAT32, and the editor
|
||||
# closes very fast, require_save can fail. Set the modified
|
||||
# time to be 2 seconds in the past to work around this.
|
||||
os.utime(name, (os.path.getatime(name), os.path.getmtime(name) - 2))
|
||||
# Depending on the resolution, the exact value might not be
|
||||
# recorded, so get the new recorded value.
|
||||
timestamp = os.path.getmtime(name)
|
||||
|
||||
self.edit_file(name)
|
||||
|
||||
if self.require_save \
|
||||
and os.path.getmtime(name) == timestamp:
|
||||
if self.require_save and os.path.getmtime(name) == timestamp:
|
||||
return None
|
||||
|
||||
f = open(name, 'rb')
|
||||
try:
|
||||
with open(name, "rb") as f:
|
||||
rv = f.read()
|
||||
finally:
|
||||
f.close()
|
||||
return rv.decode('utf-8-sig').replace('\r\n', '\n')
|
||||
|
||||
if isinstance(text, (bytes, bytearray)):
|
||||
return rv
|
||||
|
||||
return rv.decode("utf-8-sig").replace("\r\n", "\n") # type: ignore
|
||||
finally:
|
||||
os.unlink(name)
|
||||
|
||||
|
||||
def open_url(url, wait=False, locate=False):
|
||||
def open_url(url: str, wait: bool = False, locate: bool = False) -> int:
|
||||
import subprocess
|
||||
|
||||
def _unquote_file(url):
|
||||
try:
|
||||
import urllib
|
||||
except ImportError:
|
||||
import urllib
|
||||
if url.startswith('file://'):
|
||||
url = urllib.unquote(url[7:])
|
||||
def _unquote_file(url: str) -> str:
|
||||
from urllib.parse import unquote
|
||||
|
||||
if url.startswith("file://"):
|
||||
url = unquote(url[7:])
|
||||
|
||||
return url
|
||||
|
||||
if sys.platform == 'darwin':
|
||||
args = ['open']
|
||||
if sys.platform == "darwin":
|
||||
args = ["open"]
|
||||
if wait:
|
||||
args.append('-W')
|
||||
args.append("-W")
|
||||
if locate:
|
||||
args.append('-R')
|
||||
args.append("-R")
|
||||
args.append(_unquote_file(url))
|
||||
null = open('/dev/null', 'w')
|
||||
null = open("/dev/null", "w")
|
||||
try:
|
||||
return subprocess.Popen(args, stderr=null).wait()
|
||||
finally:
|
||||
null.close()
|
||||
elif WIN:
|
||||
if locate:
|
||||
url = _unquote_file(url)
|
||||
args = 'explorer /select,"%s"' % _unquote_file(
|
||||
url.replace('"', ''))
|
||||
url = _unquote_file(url.replace('"', ""))
|
||||
args = f'explorer /select,"{url}"'
|
||||
else:
|
||||
args = 'start %s "" "%s"' % (
|
||||
wait and '/WAIT' or '', url.replace('"', ''))
|
||||
url = url.replace('"', "")
|
||||
wait_str = "/WAIT" if wait else ""
|
||||
args = f'start {wait_str} "" "{url}"'
|
||||
return os.system(args)
|
||||
elif CYGWIN:
|
||||
if locate:
|
||||
url = _unquote_file(url)
|
||||
args = 'cygstart "%s"' % (os.path.dirname(url).replace('"', ''))
|
||||
url = os.path.dirname(_unquote_file(url).replace('"', ""))
|
||||
args = f'cygstart "{url}"'
|
||||
else:
|
||||
args = 'cygstart %s "%s"' % (
|
||||
wait and '-w' or '', url.replace('"', ''))
|
||||
url = url.replace('"', "")
|
||||
wait_str = "-w" if wait else ""
|
||||
args = f'cygstart {wait_str} "{url}"'
|
||||
return os.system(args)
|
||||
|
||||
try:
|
||||
if locate:
|
||||
url = os.path.dirname(_unquote_file(url)) or '.'
|
||||
url = os.path.dirname(_unquote_file(url)) or "."
|
||||
else:
|
||||
url = _unquote_file(url)
|
||||
c = subprocess.Popen(['xdg-open', url])
|
||||
c = subprocess.Popen(["xdg-open", url])
|
||||
if wait:
|
||||
return c.wait()
|
||||
return 0
|
||||
except OSError:
|
||||
if url.startswith(('http://', 'https://')) and not locate and not wait:
|
||||
if url.startswith(("http://", "https://")) and not locate and not wait:
|
||||
import webbrowser
|
||||
|
||||
webbrowser.open(url)
|
||||
return 0
|
||||
return 1
|
||||
|
||||
|
||||
def _translate_ch_to_exc(ch):
|
||||
if ch == u'\x03':
|
||||
def _translate_ch_to_exc(ch: str) -> t.Optional[BaseException]:
|
||||
if ch == "\x03":
|
||||
raise KeyboardInterrupt()
|
||||
if ch == u'\x04' and not WIN: # Unix-like, Ctrl+D
|
||||
|
||||
if ch == "\x04" and not WIN: # Unix-like, Ctrl+D
|
||||
raise EOFError()
|
||||
if ch == u'\x1a' and WIN: # Windows, Ctrl+Z
|
||||
|
||||
if ch == "\x1a" and WIN: # Windows, Ctrl+Z
|
||||
raise EOFError()
|
||||
|
||||
return None
|
||||
|
||||
|
||||
if WIN:
|
||||
import msvcrt
|
||||
|
||||
@contextlib.contextmanager
|
||||
def raw_terminal():
|
||||
yield
|
||||
def raw_terminal() -> t.Iterator[int]:
|
||||
yield -1
|
||||
|
||||
def getchar(echo):
|
||||
def getchar(echo: bool) -> str:
|
||||
# The function `getch` will return a bytes object corresponding to
|
||||
# the pressed character. Since Windows 10 build 1803, it will also
|
||||
# return \x00 when called a second time after pressing a regular key.
|
||||
|
|
@ -574,48 +658,60 @@ if WIN:
|
|||
#
|
||||
# Anyway, Click doesn't claim to do this Right(tm), and using `getwch`
|
||||
# is doing the right thing in more situations than with `getch`.
|
||||
func: t.Callable[[], str]
|
||||
|
||||
if echo:
|
||||
func = msvcrt.getwche
|
||||
func = msvcrt.getwche # type: ignore
|
||||
else:
|
||||
func = msvcrt.getwch
|
||||
func = msvcrt.getwch # type: ignore
|
||||
|
||||
rv = func()
|
||||
if rv in (u'\x00', u'\xe0'):
|
||||
|
||||
if rv in ("\x00", "\xe0"):
|
||||
# \x00 and \xe0 are control characters that indicate special key,
|
||||
# see above.
|
||||
rv += func()
|
||||
|
||||
_translate_ch_to_exc(rv)
|
||||
return rv
|
||||
|
||||
else:
|
||||
import tty
|
||||
import termios
|
||||
|
||||
@contextlib.contextmanager
|
||||
def raw_terminal():
|
||||
def raw_terminal() -> t.Iterator[int]:
|
||||
f: t.Optional[t.TextIO]
|
||||
fd: int
|
||||
|
||||
if not isatty(sys.stdin):
|
||||
f = open('/dev/tty')
|
||||
f = open("/dev/tty")
|
||||
fd = f.fileno()
|
||||
else:
|
||||
fd = sys.stdin.fileno()
|
||||
f = None
|
||||
|
||||
try:
|
||||
old_settings = termios.tcgetattr(fd)
|
||||
|
||||
try:
|
||||
tty.setraw(fd)
|
||||
yield fd
|
||||
finally:
|
||||
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
||||
sys.stdout.flush()
|
||||
|
||||
if f is not None:
|
||||
f.close()
|
||||
except termios.error:
|
||||
pass
|
||||
|
||||
def getchar(echo):
|
||||
def getchar(echo: bool) -> str:
|
||||
with raw_terminal() as fd:
|
||||
ch = os.read(fd, 32)
|
||||
ch = ch.decode(get_best_encoding(sys.stdin), 'replace')
|
||||
ch = os.read(fd, 32).decode(get_best_encoding(sys.stdin), "replace")
|
||||
|
||||
if echo and isatty(sys.stdout):
|
||||
sys.stdout.write(ch)
|
||||
|
||||
_translate_ch_to_exc(ch)
|
||||
return ch
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue