[minicurses] Fix when printing to file

Closes #1215

yt_dlp/YoutubeDL.py

@@ -514,6 +514,7 @@ def __init__(self, params=None, auto_init=True):
         self.cache = Cache(self)
 
         windows_enable_vt_mode()
+        # FIXME: This will break if we ever print color to stdout
         self.params['no_color'] = self.params.get('no_color') or not supports_terminal_sequences(self._err_file)
 
         if sys.version_info < (3, 6):
@@ -3298,6 +3299,9 @@ def python_implementation():
             KEYRING_AVAILABLE and 'keyring',
         )))) or 'none'
         self._write_string('[debug] Optional libraries: %s\n' % lib_str)
+        self._write_string('[debug] ANSI escape support: stdout = %s, stderr = %s\n' % (
+            supports_terminal_sequences(self._screen_file),
+            supports_terminal_sequences(self._err_file)))
 
         proxy_map = {}
         for handler in self._opener.handlers:

yt_dlp/downloader/common.py

@@ -3,7 +3,6 @@
 import copy
 import os
 import re
-import sys
 import time
 import random
 
@@ -247,9 +246,9 @@ def _prepare_multiline_status(self, lines=1):
         elif self.ydl.params.get('logger'):
             self._multiline = MultilineLogger(self.ydl.params['logger'], lines)
         elif self.params.get('progress_with_newline'):
-            self._multiline = BreaklineStatusPrinter(sys.stderr, lines)
+            self._multiline = BreaklineStatusPrinter(self.ydl._screen_file, lines)
         else:
-            self._multiline = MultilinePrinter(sys.stderr, lines, not self.params.get('quiet'))
+            self._multiline = MultilinePrinter(self.ydl._screen_file, lines, not self.params.get('quiet'))
 
     def _finish_multiline_status(self):
         self._multiline.end()

yt_dlp/minicurses.py

@@ -1,6 +1,6 @@
 import functools
 from threading import Lock
-from .utils import supports_terminal_sequences, TERMINAL_SEQUENCES
+from .utils import supports_terminal_sequences, TERMINAL_SEQUENCES, write_string
 
 
 class MultilinePrinterBase:
@@ -25,20 +25,26 @@ def _add_line_number(self, text, line):
             return f'{line + 1}: {text}'
         return text
 
+    def write(self, *text):
+        write_string(''.join(text), self.stream)
+
 
 class QuietMultilinePrinter(MultilinePrinterBase):
     pass
 
 
 class MultilineLogger(MultilinePrinterBase):
+    def write(self, *text):
+        self.stream.debug(''.join(text))
+
     def print_at_line(self, text, pos):
         # stream is the logger object, not an actual stream
-        self.stream.debug(self._add_line_number(text, pos))
+        self.write(self._add_line_number(text, pos))
 
 
 class BreaklineStatusPrinter(MultilinePrinterBase):
     def print_at_line(self, text, pos):
-        self.stream.write(self._add_line_number(text, pos) + '\n')
+        self.write(self._add_line_number(text, pos), '\n')
 
 
 class MultilinePrinter(MultilinePrinterBase):
@@ -58,50 +64,46 @@ def wrapper(self, *args, **kwargs):
 
     def _move_cursor(self, dest):
         current = min(self._lastline, self.maximum)
-        self.stream.write('\r')
+        yield '\r'
         distance = dest - current
         if distance < 0:
-            self.stream.write(TERMINAL_SEQUENCES['UP'] * -distance)
+            yield TERMINAL_SEQUENCES['UP'] * -distance
         elif distance > 0:
-            self.stream.write(TERMINAL_SEQUENCES['DOWN'] * distance)
+            yield TERMINAL_SEQUENCES['DOWN'] * distance
         self._lastline = dest
 
     @lock
     def print_at_line(self, text, pos):
         if self._HAVE_FULLCAP:
-            self._move_cursor(pos)
-            self.stream.write(TERMINAL_SEQUENCES['ERASE_LINE'])
-            self.stream.write(text)
-            return
+            self.write(*self._move_cursor(pos), TERMINAL_SEQUENCES['ERASE_LINE'], text)
 
         text = self._add_line_number(text, pos)
         textlen = len(text)
         if self._lastline == pos:
             # move cursor at the start of progress when writing to same line
-            self.stream.write('\r')
+            prefix = '\r'
             if self._lastlength > textlen:
                 text += ' ' * (self._lastlength - textlen)
             self._lastlength = textlen
         else:
             # otherwise, break the line
-            self.stream.write('\n')
+            prefix = '\n'
             self._lastlength = textlen
-        self.stream.write(text)
+        self.write(prefix, text)
         self._lastline = pos
 
     @lock
     def end(self):
         # move cursor to the end of the last line, and write line break
         # so that other to_screen calls can precede
-        if self._HAVE_FULLCAP:
-            self._move_cursor(self.maximum)
+        text = self._move_cursor(self.maximum) if self._HAVE_FULLCAP else []
         if self.preserve_output:
-            self.stream.write('\n')
+            self.write(*text, '\n')
             return
 
         if self._HAVE_FULLCAP:
-            self.stream.write(
-                TERMINAL_SEQUENCES['ERASE_LINE']
-                + f'{TERMINAL_SEQUENCES["UP"]}{TERMINAL_SEQUENCES["ERASE_LINE"]}' * self.maximum)
+            self.write(
+                *text, TERMINAL_SEQUENCES['ERASE_LINE'],
+                f'{TERMINAL_SEQUENCES["UP"]}{TERMINAL_SEQUENCES["ERASE_LINE"]}' * self.maximum)
         else:
-            self.stream.write(' ' * self._lastlength)
+            self.write(*text, ' ' * self._lastlength)

yt_dlp/utils.py

@@ -6458,7 +6458,7 @@ def jwt_encode_hs256(payload_data, key, headers={}):
 
 def supports_terminal_sequences(stream):
     if compat_os_name == 'nt':
-        if get_windows_version() < (10, ):
+        if get_windows_version() < (10, 0, 10586):
             return False
     elif not os.getenv('TERM'):
         return False