Merge b525b65888 into b83ca24eb7

Authored by: seproDev

.github/workflows/build.yml

@@ -504,7 +504,8 @@ jobs:
       - windows32
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/download-artifact@v4
+      - name: Download artifacts
+        uses: actions/download-artifact@v4
         with:
           path: artifact
           pattern: build-bin-*

.github/workflows/release-master.yml

@@ -28,3 +28,20 @@ jobs:
       actions: write  # For cleaning up cache
       id-token: write  # mandatory for trusted publishing
     secrets: inherit
+
+  publish_pypi:
+    needs: [release]
+    if: vars.MASTER_PYPI_PROJECT != ''
+    runs-on: ubuntu-latest
+    permissions:
+      id-token: write  # mandatory for trusted publishing
+    steps:
+      - name: Download artifacts
+        uses: actions/download-artifact@v4
+        with:
+          path: dist
+          name: build-pypi
+      - name: Publish to PyPI
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          verbose: true

.github/workflows/release-nightly.yml

@@ -41,3 +41,20 @@ jobs:
       actions: write  # For cleaning up cache
       id-token: write  # mandatory for trusted publishing
     secrets: inherit
+
+  publish_pypi:
+    needs: [release]
+    if: vars.NIGHTLY_PYPI_PROJECT != ''
+    runs-on: ubuntu-latest
+    permissions:
+      id-token: write  # mandatory for trusted publishing
+    steps:
+      - name: Download artifacts
+        uses: actions/download-artifact@v4
+        with:
+          path: dist
+          name: build-pypi
+      - name: Publish to PyPI
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          verbose: true

.github/workflows/release.yml

@@ -2,10 +2,6 @@ name: Release
 on:
   workflow_call:
     inputs:
-      prerelease:
-        required: false
-        default: true
-        type: boolean
       source:
         required: false
         default: ''
@@ -18,6 +14,10 @@ on:
         required: false
         default: ''
         type: string
+      prerelease:
+        required: false
+        default: true
+        type: boolean
   workflow_dispatch:
     inputs:
       source:
@@ -278,11 +278,20 @@ jobs:
           make clean-cache
           python -m build --no-isolation .
 
+      - name: Upload artifacts
+        if: github.event_name != 'workflow_dispatch'
+        uses: actions/upload-artifact@v4
+        with:
+          name: build-pypi
+          path: |
+            dist/*
+          compression-level: 0
+
       - name: Publish to PyPI
+        if: github.event_name == 'workflow_dispatch'
         uses: pypa/gh-action-pypi-publish@release/v1
         with:
           verbose: true
-          attestations: false  # Currently doesn't work w/ reusable workflows (breaks nightly)
 
   publish:
     needs: [prepare, build]

devscripts/generate_aes_testdata.py

@@ -11,13 +11,12 @@ import codecs
 import subprocess
 
 from yt_dlp.aes import aes_encrypt, key_expansion
-from yt_dlp.utils import intlist_to_bytes
 
 secret_msg = b'Secret message goes here'
 
 
 def hex_str(int_list):
-    return codecs.encode(intlist_to_bytes(int_list), 'hex')
+    return codecs.encode(bytes(int_list), 'hex')
 
 
 def openssl_encode(algo, key, iv):

pyproject.toml

@@ -52,7 +52,7 @@ default = [
     "pycryptodomex",
     "requests>=2.32.2,<3",
     "urllib3>=1.26.17,<3",
-    "websockets>=13.0",
+    "websockets>=13.0,<14",
 ]
 curl-cffi = [
     "curl-cffi==0.5.10; os_name=='nt' and implementation_name=='cpython'",
@@ -304,15 +304,13 @@ banned-from = [
 ]
 
 [tool.ruff.lint.flake8-tidy-imports.banned-api]
-"yt_dlp.compat.compat_str".msg = "Use `str` instead."
+"yt_dlp.compat.compat_os_name".msg = "Use `os.name` instead."
-"yt_dlp.compat.compat_b64decode".msg = "Use `base64.b64decode` instead."
+"yt_dlp.compat.compat_realpath".msg = "Use `os.path.realpath` instead."
-"yt_dlp.compat.compat_urlparse".msg = "Use `urllib.parse` instead."
-"yt_dlp.compat.compat_parse_qs".msg = "Use `urllib.parse.parse_qs` instead."
-"yt_dlp.compat.compat_urllib_parse_unquote".msg = "Use `urllib.parse.unquote` instead."
-"yt_dlp.compat.compat_urllib_parse_urlencode".msg = "Use `urllib.parse.urlencode` instead."
-"yt_dlp.compat.compat_urllib_parse_urlparse".msg = "Use `urllib.parse.urlparse` instead."
 "yt_dlp.compat.compat_shlex_quote".msg = "Use `yt_dlp.utils.shell_quote` instead."
-"yt_dlp.utils.error_to_compat_str".msg = "Use `str` instead."
+"yt_dlp.compat.functools".msg = "Use `functools` instead."
+"yt_dlp.utils.bytes_to_intlist".msg = "Use `list` instead."
+"yt_dlp.utils.compiled_regex_type".msg = "Use `re.Pattern` instead."
+"yt_dlp.utils.intlist_to_bytes".msg = "Use `bytes` instead."
 
 [tool.autopep8]
 max_line_length = 120

test/helper.py

@@ -9,7 +9,6 @@ import types
 
 import yt_dlp.extractor
 from yt_dlp import YoutubeDL
-from yt_dlp.compat import compat_os_name
 from yt_dlp.utils import preferredencoding, try_call, write_string, find_available_port
 
 if 'pytest' in sys.modules:
@@ -49,7 +48,7 @@ def report_warning(message, *args, **kwargs):
     Print the message to stderr, it will be prefixed with 'WARNING:'
     If stderr is a tty file the 'WARNING:' will be colored
     """
-    if sys.stderr.isatty() and compat_os_name != 'nt':
+    if sys.stderr.isatty() and os.name != 'nt':
         _msg_header = '\033[0;33mWARNING:\033[0m'
     else:
         _msg_header = 'WARNING:'

test/test_YoutubeDL.py

@@ -15,7 +15,6 @@ import json
 
 from test.helper import FakeYDL, assertRegexpMatches, try_rm
 from yt_dlp import YoutubeDL
-from yt_dlp.compat import compat_os_name
 from yt_dlp.extractor import YoutubeIE
 from yt_dlp.extractor.common import InfoExtractor
 from yt_dlp.postprocessor.common import PostProcessor
@@ -839,8 +838,8 @@ class TestYoutubeDL(unittest.TestCase):
         test('%(filesize)#D', '1Ki')
         test('%(height)5.2D', ' 1.08k')
         test('%(title4)#S', 'foo_bar_test')
-        test('%(title4).10S', ('foo ＂bar＂ ', 'foo ＂bar＂' + ('#' if compat_os_name == 'nt' else ' ')))
+        test('%(title4).10S', ('foo ＂bar＂ ', 'foo ＂bar＂' + ('#' if os.name == 'nt' else ' ')))
-        if compat_os_name == 'nt':
+        if os.name == 'nt':
             test('%(title4)q', ('"foo ""bar"" test"', None))
             test('%(formats.:.id)#q', ('"id 1" "id 2" "id 3"', None))
             test('%(formats.0.id)#q', ('"id 1"', None))
@@ -903,9 +902,9 @@ class TestYoutubeDL(unittest.TestCase):
 
         # Environment variable expansion for prepare_filename
         os.environ['__yt_dlp_var'] = 'expanded'
-        envvar = '%__yt_dlp_var%' if compat_os_name == 'nt' else '$__yt_dlp_var'
+        envvar = '%__yt_dlp_var%' if os.name == 'nt' else '$__yt_dlp_var'
         test(envvar, (envvar, 'expanded'))
-        if compat_os_name == 'nt':
+        if os.name == 'nt':
             test('%s%', ('%s%', '%s%'))
             os.environ['s'] = 'expanded'
             test('%s%', ('%s%', 'expanded'))  # %s% should be expanded before escaping %s

test/test_aes.py

@@ -27,7 +27,6 @@ from yt_dlp.aes import (
     pad_block,
 )
 from yt_dlp.dependencies import Cryptodome
-from yt_dlp.utils import bytes_to_intlist, intlist_to_bytes
 
 # the encrypted data can be generate with 'devscripts/generate_aes_testdata.py'
 
@@ -40,33 +39,33 @@ class TestAES(unittest.TestCase):
     def test_encrypt(self):
         msg = b'message'
         key = list(range(16))
-        encrypted = aes_encrypt(bytes_to_intlist(msg), key)
+        encrypted = aes_encrypt(list(msg), key)
-        decrypted = intlist_to_bytes(aes_decrypt(encrypted, key))
+        decrypted = bytes(aes_decrypt(encrypted, key))
         self.assertEqual(decrypted, msg)
 
     def test_cbc_decrypt(self):
         data = b'\x97\x92+\xe5\x0b\xc3\x18\x91ky9m&\xb3\xb5@\xe6\x27\xc2\x96.\xc8u\x88\xab9-[\x9e|\xf1\xcd'
-        decrypted = intlist_to_bytes(aes_cbc_decrypt(bytes_to_intlist(data), self.key, self.iv))
+        decrypted = bytes(aes_cbc_decrypt(list(data), self.key, self.iv))
         self.assertEqual(decrypted.rstrip(b'\x08'), self.secret_msg)
         if Cryptodome.AES:
-            decrypted = aes_cbc_decrypt_bytes(data, intlist_to_bytes(self.key), intlist_to_bytes(self.iv))
+            decrypted = aes_cbc_decrypt_bytes(data, bytes(self.key), bytes(self.iv))
             self.assertEqual(decrypted.rstrip(b'\x08'), self.secret_msg)
 
     def test_cbc_encrypt(self):
-        data = bytes_to_intlist(self.secret_msg)
+        data = list(self.secret_msg)
-        encrypted = intlist_to_bytes(aes_cbc_encrypt(data, self.key, self.iv))
+        encrypted = bytes(aes_cbc_encrypt(data, self.key, self.iv))
         self.assertEqual(
             encrypted,
             b'\x97\x92+\xe5\x0b\xc3\x18\x91ky9m&\xb3\xb5@\xe6\'\xc2\x96.\xc8u\x88\xab9-[\x9e|\xf1\xcd')
 
     def test_ctr_decrypt(self):
-        data = bytes_to_intlist(b'\x03\xc7\xdd\xd4\x8e\xb3\xbc\x1a*O\xdc1\x12+8Aio\xd1z\xb5#\xaf\x08')
+        data = list(b'\x03\xc7\xdd\xd4\x8e\xb3\xbc\x1a*O\xdc1\x12+8Aio\xd1z\xb5#\xaf\x08')
-        decrypted = intlist_to_bytes(aes_ctr_decrypt(data, self.key, self.iv))
+        decrypted = bytes(aes_ctr_decrypt(data, self.key, self.iv))
         self.assertEqual(decrypted.rstrip(b'\x08'), self.secret_msg)
 
     def test_ctr_encrypt(self):
-        data = bytes_to_intlist(self.secret_msg)
+        data = list(self.secret_msg)
-        encrypted = intlist_to_bytes(aes_ctr_encrypt(data, self.key, self.iv))
+        encrypted = bytes(aes_ctr_encrypt(data, self.key, self.iv))
         self.assertEqual(
             encrypted,
             b'\x03\xc7\xdd\xd4\x8e\xb3\xbc\x1a*O\xdc1\x12+8Aio\xd1z\xb5#\xaf\x08')
@@ -75,19 +74,19 @@ class TestAES(unittest.TestCase):
         data = b'\x159Y\xcf5eud\x90\x9c\x85&]\x14\x1d\x0f.\x08\xb4T\xe4/\x17\xbd'
         authentication_tag = b'\xe8&I\x80rI\x07\x9d}YWuU@:e'
 
-        decrypted = intlist_to_bytes(aes_gcm_decrypt_and_verify(
+        decrypted = bytes(aes_gcm_decrypt_and_verify(
-            bytes_to_intlist(data), self.key, bytes_to_intlist(authentication_tag), self.iv[:12]))
+            list(data), self.key, list(authentication_tag), self.iv[:12]))
         self.assertEqual(decrypted.rstrip(b'\x08'), self.secret_msg)
         if Cryptodome.AES:
             decrypted = aes_gcm_decrypt_and_verify_bytes(
-                data, intlist_to_bytes(self.key), authentication_tag, intlist_to_bytes(self.iv[:12]))
+                data, bytes(self.key), authentication_tag, bytes(self.iv[:12]))
             self.assertEqual(decrypted.rstrip(b'\x08'), self.secret_msg)
 
     def test_gcm_aligned_decrypt(self):
         data = b'\x159Y\xcf5eud\x90\x9c\x85&]\x14\x1d\x0f'
         authentication_tag = b'\x08\xb1\x9d!&\x98\xd0\xeaRq\x90\xe6;\xb5]\xd8'
 
-        decrypted = intlist_to_bytes(aes_gcm_decrypt_and_verify(
+        decrypted = bytes(aes_gcm_decrypt_and_verify(
             list(data), self.key, list(authentication_tag), self.iv[:12]))
         self.assertEqual(decrypted.rstrip(b'\x08'), self.secret_msg[:16])
         if Cryptodome.AES:
@@ -96,38 +95,38 @@ class TestAES(unittest.TestCase):
             self.assertEqual(decrypted.rstrip(b'\x08'), self.secret_msg[:16])
 
     def test_decrypt_text(self):
-        password = intlist_to_bytes(self.key).decode()
+        password = bytes(self.key).decode()
         encrypted = base64.b64encode(
-            intlist_to_bytes(self.iv[:8])
+            bytes(self.iv[:8])
             + b'\x17\x15\x93\xab\x8d\x80V\xcdV\xe0\t\xcdo\xc2\xa5\xd8ksM\r\xe27N\xae',
         ).decode()
         decrypted = (aes_decrypt_text(encrypted, password, 16))
         self.assertEqual(decrypted, self.secret_msg)
 
-        password = intlist_to_bytes(self.key).decode()
+        password = bytes(self.key).decode()
         encrypted = base64.b64encode(
-            intlist_to_bytes(self.iv[:8])
+            bytes(self.iv[:8])
             + b'\x0b\xe6\xa4\xd9z\x0e\xb8\xb9\xd0\xd4i_\x85\x1d\x99\x98_\xe5\x80\xe7.\xbf\xa5\x83',
         ).decode()
         decrypted = (aes_decrypt_text(encrypted, password, 32))
         self.assertEqual(decrypted, self.secret_msg)
 
     def test_ecb_encrypt(self):
-        data = bytes_to_intlist(self.secret_msg)
+        data = list(self.secret_msg)
-        encrypted = intlist_to_bytes(aes_ecb_encrypt(data, self.key))
+        encrypted = bytes(aes_ecb_encrypt(data, self.key))
         self.assertEqual(
             encrypted,
             b'\xaa\x86]\x81\x97>\x02\x92\x9d\x1bR[[L/u\xd3&\xd1(h\xde{\x81\x94\xba\x02\xae\xbd\xa6\xd0:')
 
     def test_ecb_decrypt(self):
-        data = bytes_to_intlist(b'\xaa\x86]\x81\x97>\x02\x92\x9d\x1bR[[L/u\xd3&\xd1(h\xde{\x81\x94\xba\x02\xae\xbd\xa6\xd0:')
+        data = list(b'\xaa\x86]\x81\x97>\x02\x92\x9d\x1bR[[L/u\xd3&\xd1(h\xde{\x81\x94\xba\x02\xae\xbd\xa6\xd0:')
-        decrypted = intlist_to_bytes(aes_ecb_decrypt(data, self.key, self.iv))
+        decrypted = bytes(aes_ecb_decrypt(data, self.key, self.iv))
         self.assertEqual(decrypted.rstrip(b'\x08'), self.secret_msg)
 
     def test_key_expansion(self):
         key = '4f6bdaa39e2f8cb07f5e722d9edef314'
 
-        self.assertEqual(key_expansion(bytes_to_intlist(bytearray.fromhex(key))), [
+        self.assertEqual(key_expansion(list(bytearray.fromhex(key))), [
             0x4F, 0x6B, 0xDA, 0xA3, 0x9E, 0x2F, 0x8C, 0xB0, 0x7F, 0x5E, 0x72, 0x2D, 0x9E, 0xDE, 0xF3, 0x14,
             0x53, 0x66, 0x20, 0xA8, 0xCD, 0x49, 0xAC, 0x18, 0xB2, 0x17, 0xDE, 0x35, 0x2C, 0xC9, 0x2D, 0x21,
             0x8C, 0xBE, 0xDD, 0xD9, 0x41, 0xF7, 0x71, 0xC1, 0xF3, 0xE0, 0xAF, 0xF4, 0xDF, 0x29, 0x82, 0xD5,

test/test_compat.py

@@ -12,12 +12,7 @@ import struct
 
 from yt_dlp import compat
 from yt_dlp.compat import urllib  # isort: split
-from yt_dlp.compat import (
+from yt_dlp.compat import compat_etree_fromstring, compat_expanduser
-    compat_etree_fromstring,
-    compat_expanduser,
-    compat_urllib_parse_unquote,  # noqa: TID251
-    compat_urllib_parse_urlencode,  # noqa: TID251
-)
 from yt_dlp.compat.urllib.request import getproxies
 
 
@@ -43,39 +38,6 @@ class TestCompat(unittest.TestCase):
         finally:
             os.environ['HOME'] = old_home or ''
 
-    def test_compat_urllib_parse_unquote(self):
-        self.assertEqual(compat_urllib_parse_unquote('abc%20def'), 'abc def')
-        self.assertEqual(compat_urllib_parse_unquote('%7e/abc+def'), '~/abc+def')
-        self.assertEqual(compat_urllib_parse_unquote(''), '')
-        self.assertEqual(compat_urllib_parse_unquote('%'), '%')
-        self.assertEqual(compat_urllib_parse_unquote('%%'), '%%')
-        self.assertEqual(compat_urllib_parse_unquote('%%%'), '%%%')
-        self.assertEqual(compat_urllib_parse_unquote('%2F'), '/')
-        self.assertEqual(compat_urllib_parse_unquote('%2f'), '/')
-        self.assertEqual(compat_urllib_parse_unquote('%E6%B4%A5%E6%B3%A2'), '津波')
-        self.assertEqual(
-            compat_urllib_parse_unquote('''<meta property="og:description" content="%E2%96%81%E2%96%82%E2%96%83%E2%96%84%25%E2%96%85%E2%96%86%E2%96%87%E2%96%88" />
-%<a href="https://ar.wikipedia.org/wiki/%D8%AA%D8%B3%D9%88%D9%86%D8%A7%D9%85%D9%8A">%a'''),
-            '''<meta property="og:description" content="▁▂▃▄%▅▆▇█" />
-%<a href="https://ar.wikipedia.org/wiki/تسونامي">%a''')
-        self.assertEqual(
-            compat_urllib_parse_unquote('''%28%5E%E2%97%A3_%E2%97%A2%5E%29%E3%81%A3%EF%B8%BB%E3%83%87%E2%95%90%E4%B8%80    %E2%87%80    %E2%87%80    %E2%87%80    %E2%87%80    %E2%87%80    %E2%86%B6%I%Break%25Things%'''),
-            '''(^◣_◢^)っ︻デ═一    ⇀    ⇀    ⇀    ⇀    ⇀    ↶%I%Break%Things%''')
-
-    def test_compat_urllib_parse_unquote_plus(self):
-        self.assertEqual(urllib.parse.unquote_plus('abc%20def'), 'abc def')
-        self.assertEqual(urllib.parse.unquote_plus('%7e/abc+def'), '~/abc def')
-
-    def test_compat_urllib_parse_urlencode(self):
-        self.assertEqual(compat_urllib_parse_urlencode({'abc': 'def'}), 'abc=def')
-        self.assertEqual(compat_urllib_parse_urlencode({'abc': b'def'}), 'abc=def')
-        self.assertEqual(compat_urllib_parse_urlencode({b'abc': 'def'}), 'abc=def')
-        self.assertEqual(compat_urllib_parse_urlencode({b'abc': b'def'}), 'abc=def')
-        self.assertEqual(compat_urllib_parse_urlencode([('abc', 'def')]), 'abc=def')
-        self.assertEqual(compat_urllib_parse_urlencode([('abc', b'def')]), 'abc=def')
-        self.assertEqual(compat_urllib_parse_urlencode([(b'abc', 'def')]), 'abc=def')
-        self.assertEqual(compat_urllib_parse_urlencode([(b'abc', b'def')]), 'abc=def')
-
     def test_compat_etree_fromstring(self):
         xml = '''
             <root foo="bar" spam="中文">

test/test_downloader_http.py

@@ -15,7 +15,6 @@ import threading
 from test.helper import http_server_port, try_rm
 from yt_dlp import YoutubeDL
 from yt_dlp.downloader.http import HttpFD
-from yt_dlp.utils import encodeFilename
 from yt_dlp.utils._utils import _YDLLogger as FakeLogger
 
 TEST_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -82,12 +81,12 @@ class TestHttpFD(unittest.TestCase):
         ydl = YoutubeDL(params)
         downloader = HttpFD(ydl, params)
         filename = 'testfile.mp4'
-        try_rm(encodeFilename(filename))
+        try_rm(filename)
         self.assertTrue(downloader.real_download(filename, {
             'url': f'http://127.0.0.1:{self.port}/{ep}',
         }), ep)
-        self.assertEqual(os.path.getsize(encodeFilename(filename)), TEST_SIZE, ep)
+        self.assertEqual(os.path.getsize(filename), TEST_SIZE, ep)
-        try_rm(encodeFilename(filename))
+        try_rm(filename)
 
     def download_all(self, params):
         for ep in ('regular', 'no-content-length', 'no-range', 'no-range-no-content-length'):

test/test_utils.py

@@ -21,7 +21,6 @@ import xml.etree.ElementTree
 from yt_dlp.compat import (
     compat_etree_fromstring,
     compat_HTMLParseError,
-    compat_os_name,
 )
 from yt_dlp.utils import (
     Config,
@@ -49,7 +48,6 @@ from yt_dlp.utils import (
     dfxp2srt,
     encode_base_n,
     encode_compat_str,
-    encodeFilename,
     expand_path,
     extract_attributes,
     extract_basic_auth,
@@ -69,7 +67,6 @@ from yt_dlp.utils import (
     get_elements_html_by_class,
     get_elements_text_and_html_by_attribute,
     int_or_none,
-    intlist_to_bytes,
     iri_to_uri,
     is_html,
     join_nonempty,
@@ -567,10 +564,10 @@ class TestUtil(unittest.TestCase):
         self.assertEqual(res_data, {'a': 'b', 'c': 'd'})
 
     def test_shell_quote(self):
-        args = ['ffmpeg', '-i', encodeFilename('ñ€ß\'.mp4')]
+        args = ['ffmpeg', '-i', 'ñ€ß\'.mp4']
         self.assertEqual(
             shell_quote(args),
-            """ffmpeg -i 'ñ€ß'"'"'.mp4'""" if compat_os_name != 'nt' else '''ffmpeg -i "ñ€ß'.mp4"''')
+            """ffmpeg -i 'ñ€ß'"'"'.mp4'""" if os.name != 'nt' else '''ffmpeg -i "ñ€ß'.mp4"''')
 
     def test_float_or_none(self):
         self.assertEqual(float_or_none('42.42'), 42.42)
@@ -1310,15 +1307,10 @@ class TestUtil(unittest.TestCase):
         self.assertEqual(clean_html('a:\n   "b"'), 'a: "b"')
         self.assertEqual(clean_html('a<br>\xa0b'), 'a\nb')
 
-    def test_intlist_to_bytes(self):
-        self.assertEqual(
-            intlist_to_bytes([0, 1, 127, 128, 255]),
-            b'\x00\x01\x7f\x80\xff')
-
     def test_args_to_str(self):
         self.assertEqual(
             args_to_str(['foo', 'ba/r', '-baz', '2 be', '']),
-            'foo ba/r -baz \'2 be\' \'\'' if compat_os_name != 'nt' else 'foo ba/r -baz "2 be" ""',
+            'foo ba/r -baz \'2 be\' \'\'' if os.name != 'nt' else 'foo ba/r -baz "2 be" ""',
         )
 
     def test_parse_filesize(self):
@@ -2118,7 +2110,7 @@ Line 1
         assert extract_basic_auth('http://user:@foo.bar') == ('http://foo.bar', 'Basic dXNlcjo=')
         assert extract_basic_auth('http://user:pass@foo.bar') == ('http://foo.bar', 'Basic dXNlcjpwYXNz')
 
-    @unittest.skipUnless(compat_os_name == 'nt', 'Only relevant on Windows')
+    @unittest.skipUnless(os.name == 'nt', 'Only relevant on Windows')
     def test_windows_escaping(self):
         tests = [
             'test"&',

yt_dlp/YoutubeDL.py

@@ -26,7 +26,7 @@ import unicodedata
 
 from .cache import Cache
 from .compat import urllib  # isort: split
-from .compat import compat_os_name, urllib_req_to_req
+from .compat import urllib_req_to_req
 from .cookies import CookieLoadError, LenientSimpleCookie, load_cookies
 from .downloader import FFmpegFD, get_suitable_downloader, shorten_protocol_name
 from .downloader.rtmp import rtmpdump_version
@@ -109,7 +109,6 @@ from .utils import (
     determine_ext,
     determine_protocol,
     encode_compat_str,
-    encodeFilename,
     escapeHTML,
     expand_path,
     extract_basic_auth,
@@ -167,7 +166,7 @@ from .utils.networking import (
 )
 from .version import CHANNEL, ORIGIN, RELEASE_GIT_HEAD, VARIANT, __version__
 
-if compat_os_name == 'nt':
+if os.name == 'nt':
     import ctypes
 
 
@@ -643,7 +642,7 @@ class YoutubeDL:
             out=stdout,
             error=sys.stderr,
             screen=sys.stderr if self.params.get('quiet') else stdout,
-            console=None if compat_os_name == 'nt' else next(
+            console=None if os.name == 'nt' else next(
                 filter(supports_terminal_sequences, (sys.stderr, sys.stdout)), None),
         )
 
@@ -952,7 +951,7 @@ class YoutubeDL:
             self._write_string(f'{self._bidi_workaround(message)}\n', self._out_files.error, only_once=only_once)
 
     def _send_console_code(self, code):
-        if compat_os_name == 'nt' or not self._out_files.console:
+        if os.name == 'nt' or not self._out_files.console:
             return
         self._write_string(code, self._out_files.console)
 
@@ -960,7 +959,7 @@ class YoutubeDL:
         if not self.params.get('consoletitle', False):
             return
         message = remove_terminal_sequences(message)
-        if compat_os_name == 'nt':
+        if os.name == 'nt':
             if ctypes.windll.kernel32.GetConsoleWindow():
                 # c_wchar_p() might not be necessary if `message` is
                 # already of type unicode()
@@ -3255,9 +3254,9 @@ class YoutubeDL:
 
         if full_filename is None:
             return
-        if not self._ensure_dir_exists(encodeFilename(full_filename)):
+        if not self._ensure_dir_exists(full_filename):
             return
-        if not self._ensure_dir_exists(encodeFilename(temp_filename)):
+        if not self._ensure_dir_exists(temp_filename):
             return
 
         if self._write_description('video', info_dict,
@@ -3289,16 +3288,16 @@ class YoutubeDL:
         if self.params.get('writeannotations', False):
             annofn = self.prepare_filename(info_dict, 'annotation')
         if annofn:
-            if not self._ensure_dir_exists(encodeFilename(annofn)):
+            if not self._ensure_dir_exists(annofn):
                 return
-            if not self.params.get('overwrites', True) and os.path.exists(encodeFilename(annofn)):
+            if not self.params.get('overwrites', True) and os.path.exists(annofn):
                 self.to_screen('[info] Video annotations are already present')
             elif not info_dict.get('annotations'):
                 self.report_warning('There are no annotations to write.')
             else:
                 try:
                     self.to_screen('[info] Writing video annotations to: ' + annofn)
-                    with open(encodeFilename(annofn), 'w', encoding='utf-8') as annofile:
+                    with open(annofn, 'w', encoding='utf-8') as annofile:
                         annofile.write(info_dict['annotations'])
                 except (KeyError, TypeError):
                     self.report_warning('There are no annotations to write.')
@@ -3314,14 +3313,14 @@ class YoutubeDL:
                     f'Cannot write internet shortcut file because the actual URL of "{info_dict["webpage_url"]}" is unknown')
                 return True
             linkfn = replace_extension(self.prepare_filename(info_dict, 'link'), link_type, info_dict.get('ext'))
-            if not self._ensure_dir_exists(encodeFilename(linkfn)):
+            if not self._ensure_dir_exists(linkfn):
                 return False
-            if self.params.get('overwrites', True) and os.path.exists(encodeFilename(linkfn)):
+            if self.params.get('overwrites', True) and os.path.exists(linkfn):
                 self.to_screen(f'[info] Internet shortcut (.{link_type}) is already present')
                 return True
             try:
                 self.to_screen(f'[info] Writing internet shortcut (.{link_type}) to: {linkfn}')
-                with open(encodeFilename(to_high_limit_path(linkfn)), 'w', encoding='utf-8',
+                with open(to_high_limit_path(linkfn), 'w', encoding='utf-8',
                           newline='\r\n' if link_type == 'url' else '\n') as linkfile:
                     template_vars = {'url': url}
                     if link_type == 'desktop':
@@ -3352,7 +3351,7 @@ class YoutubeDL:
 
         if self.params.get('skip_download'):
             info_dict['filepath'] = temp_filename
-            info_dict['__finaldir'] = os.path.dirname(os.path.abspath(encodeFilename(full_filename)))
+            info_dict['__finaldir'] = os.path.dirname(os.path.abspath(full_filename))
             info_dict['__files_to_move'] = files_to_move
             replace_info_dict(self.run_pp(MoveFilesAfterDownloadPP(self, False), info_dict))
             info_dict['__write_download_archive'] = self.params.get('force_write_download_archive')
@@ -3482,7 +3481,7 @@ class YoutubeDL:
                         self.report_file_already_downloaded(dl_filename)
 
                 dl_filename = dl_filename or temp_filename
-                info_dict['__finaldir'] = os.path.dirname(os.path.abspath(encodeFilename(full_filename)))
+                info_dict['__finaldir'] = os.path.dirname(os.path.abspath(full_filename))
 
             except network_exceptions as err:
                 self.report_error(f'unable to download video data: {err}')
@@ -4297,7 +4296,7 @@ class YoutubeDL:
         else:
             try:
                 self.to_screen(f'[info] Writing {label} description to: {descfn}')
-                with open(encodeFilename(descfn), 'w', encoding='utf-8') as descfile:
+                with open(descfn, 'w', encoding='utf-8') as descfile:
                     descfile.write(ie_result['description'])
             except OSError:
                 self.report_error(f'Cannot write {label} description file {descfn}')
@@ -4397,7 +4396,7 @@ class YoutubeDL:
                 try:
                     uf = self.urlopen(Request(t['url'], headers=t.get('http_headers', {})))
                     self.to_screen(f'[info] Writing {thumb_display_id} to: {thumb_filename}')
-                    with open(encodeFilename(thumb_filename), 'wb') as thumbf:
+                    with open(thumb_filename, 'wb') as thumbf:
                         shutil.copyfileobj(uf, thumbf)
                     ret.append((thumb_filename, thumb_filename_final))
                     t['filepath'] = thumb_filename

yt_dlp/__init__.py

@@ -14,7 +14,6 @@ import os
 import re
 import traceback
 
-from .compat import compat_os_name
 from .cookies import SUPPORTED_BROWSERS, SUPPORTED_KEYRINGS, CookieLoadError
 from .downloader.external import get_external_downloader
 from .extractor import list_extractor_classes
@@ -44,7 +43,6 @@ from .utils import (
     GeoUtils,
     PlaylistEntries,
     SameFileError,
-    decodeOption,
     download_range_func,
     expand_path,
     float_or_none,
@@ -883,8 +881,8 @@ def parse_options(argv=None):
         'listsubtitles': opts.listsubtitles,
         'subtitlesformat': opts.subtitlesformat,
         'subtitleslangs': opts.subtitleslangs,
-        'matchtitle': decodeOption(opts.matchtitle),
+        'matchtitle': opts.matchtitle,
-        'rejecttitle': decodeOption(opts.rejecttitle),
+        'rejecttitle': opts.rejecttitle,
         'max_downloads': opts.max_downloads,
         'prefer_free_formats': opts.prefer_free_formats,
         'trim_file_name': opts.trim_file_name,
@@ -1053,7 +1051,7 @@ def _real_main(argv=None):
             ydl.warn_if_short_id(args)
 
             # Show a useful error message and wait for keypress if not launched from shell on Windows
-            if not args and compat_os_name == 'nt' and getattr(sys, 'frozen', False):
+            if not args and os.name == 'nt' and getattr(sys, 'frozen', False):
                 import ctypes.wintypes
                 import msvcrt
 

yt_dlp/aes.py

@@ -3,7 +3,6 @@ from math import ceil
 
 from .compat import compat_ord
 from .dependencies import Cryptodome
-from .utils import bytes_to_intlist, intlist_to_bytes
 
 if Cryptodome.AES:
     def aes_cbc_decrypt_bytes(data, key, iv):
@@ -17,15 +16,15 @@ if Cryptodome.AES:
 else:
     def aes_cbc_decrypt_bytes(data, key, iv):
         """ Decrypt bytes with AES-CBC using native implementation since pycryptodome is unavailable """
-        return intlist_to_bytes(aes_cbc_decrypt(*map(bytes_to_intlist, (data, key, iv))))
+        return bytes(aes_cbc_decrypt(*map(list, (data, key, iv))))
 
     def aes_gcm_decrypt_and_verify_bytes(data, key, tag, nonce):
         """ Decrypt bytes with AES-GCM using native implementation since pycryptodome is unavailable """
-        return intlist_to_bytes(aes_gcm_decrypt_and_verify(*map(bytes_to_intlist, (data, key, tag, nonce))))
+        return bytes(aes_gcm_decrypt_and_verify(*map(list, (data, key, tag, nonce))))
 
 
 def aes_cbc_encrypt_bytes(data, key, iv, **kwargs):
-    return intlist_to_bytes(aes_cbc_encrypt(*map(bytes_to_intlist, (data, key, iv)), **kwargs))
+    return bytes(aes_cbc_encrypt(*map(list, (data, key, iv)), **kwargs))
 
 
 BLOCK_SIZE_BYTES = 16
@@ -221,7 +220,7 @@ def aes_gcm_decrypt_and_verify(data, key, tag, nonce):
         j0 = [*nonce, 0, 0, 0, 1]
     else:
         fill = (BLOCK_SIZE_BYTES - (len(nonce) % BLOCK_SIZE_BYTES)) % BLOCK_SIZE_BYTES + 8
-        ghash_in = nonce + [0] * fill + bytes_to_intlist((8 * len(nonce)).to_bytes(8, 'big'))
+        ghash_in = nonce + [0] * fill + list((8 * len(nonce)).to_bytes(8, 'big'))
         j0 = ghash(hash_subkey, ghash_in)
 
     # TODO: add nonce support to aes_ctr_decrypt
@@ -235,7 +234,7 @@ def aes_gcm_decrypt_and_verify(data, key, tag, nonce):
         hash_subkey,
         data
         + [0] * pad_len                                  # pad
-        + bytes_to_intlist((0 * 8).to_bytes(8, 'big')           # length of associated data
+        + list((0 * 8).to_bytes(8, 'big')                # length of associated data
                + ((len(data) * 8).to_bytes(8, 'big'))),  # length of data
     )
 
@@ -300,8 +299,8 @@ def aes_decrypt_text(data, password, key_size_bytes):
     """
     NONCE_LENGTH_BYTES = 8
 
-    data = bytes_to_intlist(base64.b64decode(data))
+    data = list(base64.b64decode(data))
-    password = bytes_to_intlist(password.encode())
+    password = list(password.encode())
 
     key = password[:key_size_bytes] + [0] * (key_size_bytes - len(password))
     key = aes_encrypt(key[:BLOCK_SIZE_BYTES], key_expansion(key)) * (key_size_bytes // BLOCK_SIZE_BYTES)
@@ -310,7 +309,7 @@ def aes_decrypt_text(data, password, key_size_bytes):
     cipher = data[NONCE_LENGTH_BYTES:]
 
     decrypted_data = aes_ctr_decrypt(cipher, key, nonce + [0] * (BLOCK_SIZE_BYTES - NONCE_LENGTH_BYTES))
-    return intlist_to_bytes(decrypted_data)
+    return bytes(decrypted_data)
 
 
 RCON = (0x8d, 0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80, 0x1b, 0x36)

yt_dlp/compat/__init__.py

@@ -1,5 +1,4 @@
 import os
-import sys
 import xml.etree.ElementTree as etree
 
 from .compat_utils import passthrough_module
@@ -24,33 +23,14 @@ def compat_etree_fromstring(text):
     return etree.XML(text, parser=etree.XMLParser(target=_TreeBuilder()))
 
 
-compat_os_name = os._name if os.name == 'java' else os.name
-
-
-def compat_shlex_quote(s):
-    from ..utils import shell_quote
-    return shell_quote(s)
-
-
 def compat_ord(c):
     return c if isinstance(c, int) else ord(c)
 
 
-if compat_os_name == 'nt' and sys.version_info < (3, 8):
-    # os.path.realpath on Windows does not follow symbolic links
-    # prior to Python 3.8 (see https://bugs.python.org/issue9949)
-    def compat_realpath(path):
-        while os.path.islink(path):
-            path = os.path.abspath(os.readlink(path))
-        return os.path.realpath(path)
-else:
-    compat_realpath = os.path.realpath
-
-
 # Python 3.8+ does not honor %HOME% on windows, but this breaks compatibility with youtube-dl
 # See https://github.com/yt-dlp/yt-dlp/issues/792
 # https://docs.python.org/3/library/os.path.html#os.path.expanduser
-if compat_os_name in ('nt', 'ce'):
+if os.name in ('nt', 'ce'):
     def compat_expanduser(path):
         HOME = os.environ.get('HOME')
         if not HOME:

yt_dlp/compat/_deprecated.py

@@ -8,16 +8,14 @@ passthrough_module(__name__, '.._legacy', callback=lambda attr: warnings.warn(
     DeprecationWarning(f'{__name__}.{attr} is deprecated'), stacklevel=6))
 del passthrough_module
 
-import base64
+import functools  # noqa: F401
-import urllib.error
+import os
-import urllib.parse
 
-compat_str = str
 
-compat_b64decode = base64.b64decode
+compat_os_name = os.name
+compat_realpath = os.path.realpath
 
-compat_urlparse = urllib.parse
+
-compat_parse_qs = urllib.parse.parse_qs
+def compat_shlex_quote(s):
-compat_urllib_parse_unquote = urllib.parse.unquote
+    from ..utils import shell_quote
-compat_urllib_parse_urlencode = urllib.parse.urlencode
+    return shell_quote(s)
-compat_urllib_parse_urlparse = urllib.parse.urlparse

yt_dlp/compat/_legacy.py

@@ -30,7 +30,7 @@ from asyncio import run as compat_asyncio_run  # noqa: F401
 from re import Pattern as compat_Pattern  # noqa: F401
 from re import match as compat_Match  # noqa: F401
 
-from . import compat_expanduser, compat_HTMLParseError, compat_realpath
+from . import compat_expanduser, compat_HTMLParseError
 from .compat_utils import passthrough_module
 from ..dependencies import brotli as compat_brotli  # noqa: F401
 from ..dependencies import websockets as compat_websockets  # noqa: F401
@@ -78,7 +78,7 @@ compat_kwargs = lambda kwargs: kwargs
 compat_map = map
 compat_numeric_types = (int, float, complex)
 compat_os_path_expanduser = compat_expanduser
-compat_os_path_realpath = compat_realpath
+compat_os_path_realpath = os.path.realpath
 compat_print = print
 compat_shlex_split = shlex.split
 compat_socket_create_connection = socket.create_connection
@@ -104,5 +104,12 @@ compat_xml_parse_error = compat_xml_etree_ElementTree_ParseError = etree.ParseEr
 compat_xpath = lambda xpath: xpath
 compat_zip = zip
 workaround_optparse_bug9161 = lambda: None
+compat_str = str
+compat_b64decode = base64.b64decode
+compat_urlparse = urllib.parse
+compat_parse_qs = urllib.parse.parse_qs
+compat_urllib_parse_unquote = urllib.parse.unquote
+compat_urllib_parse_urlencode = urllib.parse.urlencode
+compat_urllib_parse_urlparse = urllib.parse.urlparse
 
 legacy = []

yt_dlp/compat/functools.py

@@ -1,7 +0,0 @@
-# flake8: noqa: F405
-from functools import *  # noqa: F403
-
-from .compat_utils import passthrough_module
-
-passthrough_module(__name__, 'functools')
-del passthrough_module

yt_dlp/compat/urllib/request.py

@@ -7,9 +7,9 @@ passthrough_module(__name__, 'urllib.request')
 del passthrough_module
 
 
-from .. import compat_os_name
+import os
 
-if compat_os_name == 'nt':
+if os.name == 'nt':
     # On older Python versions, proxies are extracted from Windows registry erroneously. [1]
     # If the https proxy in the registry does not have a scheme, urllib will incorrectly add https:// to it. [2]
     # It is unlikely that the user has actually set it to be https, so we should be fine to safely downgrade
@@ -37,4 +37,4 @@ if compat_os_name == 'nt':
     def getproxies():
         return getproxies_environment() or getproxies_registry_patched()
 
-del compat_os_name
+del os

yt_dlp/cookies.py

@@ -25,7 +25,6 @@ from .aes import (
     aes_gcm_decrypt_and_verify_bytes,
     unpad_pkcs7,
 )
-from .compat import compat_os_name
 from .dependencies import (
     _SECRETSTORAGE_UNAVAILABLE_REASON,
     secretstorage,
@@ -343,7 +342,7 @@ def _extract_chrome_cookies(browser_name, profile, keyring, logger):
             logger.debug(f'cookie version breakdown: {counts}')
             return jar
         except PermissionError as error:
-            if compat_os_name == 'nt' and error.errno == 13:
+            if os.name == 'nt' and error.errno == 13:
                 message = 'Could not copy Chrome cookie database. See  https://github.com/yt-dlp/yt-dlp/issues/7271  for more info'
                 logger.error(message)
                 raise DownloadError(message)  # force exit

yt_dlp/dependencies/Cryptodome.py

@@ -24,7 +24,7 @@ try:
         from Crypto.Cipher import AES, PKCS1_OAEP, Blowfish, PKCS1_v1_5  # noqa: F401
         from Crypto.Hash import CMAC, SHA1  # noqa: F401
         from Crypto.PublicKey import RSA  # noqa: F401
-except ImportError:
+except (ImportError, OSError):
     __version__ = f'broken {__version__}'.strip()
 
 

yt_dlp/downloader/common.py

@@ -20,9 +20,7 @@ from ..utils import (
     Namespace,
     RetryManager,
     classproperty,
-    decodeArgument,
     deprecation_warning,
-    encodeFilename,
     format_bytes,
     join_nonempty,
     parse_bytes,
@@ -219,7 +217,7 @@ class FileDownloader:
     def temp_name(self, filename):
         """Returns a temporary filename for the given filename."""
         if self.params.get('nopart', False) or filename == '-' or \
-                (os.path.exists(encodeFilename(filename)) and not os.path.isfile(encodeFilename(filename))):
+                (os.path.exists(filename) and not os.path.isfile(filename)):
             return filename
         return filename + '.part'
 
@@ -273,7 +271,7 @@ class FileDownloader:
         """Try to set the last-modified time of the given file."""
         if last_modified_hdr is None:
             return
-        if not os.path.isfile(encodeFilename(filename)):
+        if not os.path.isfile(filename):
             return
         timestr = last_modified_hdr
         if timestr is None:
@@ -432,13 +430,13 @@ class FileDownloader:
         """
         nooverwrites_and_exists = (
             not self.params.get('overwrites', True)
-            and os.path.exists(encodeFilename(filename))
+            and os.path.exists(filename)
         )
 
         if not hasattr(filename, 'write'):
             continuedl_and_exists = (
                 self.params.get('continuedl', True)
-                and os.path.isfile(encodeFilename(filename))
+                and os.path.isfile(filename)
                 and not self.params.get('nopart', False)
             )
 
@@ -448,7 +446,7 @@ class FileDownloader:
                 self._hook_progress({
                     'filename': filename,
                     'status': 'finished',
-                    'total_bytes': os.path.getsize(encodeFilename(filename)),
+                    'total_bytes': os.path.getsize(filename),
                 }, info_dict)
                 self._finish_multiline_status()
                 return True, False
@@ -489,7 +487,7 @@ class FileDownloader:
         if not self.params.get('verbose', False):
             return
 
-        str_args = [decodeArgument(a) for a in args]
+        str_args = list(args)
 
         if exe is None:
             exe = os.path.basename(str_args[0])

yt_dlp/downloader/external.py

@@ -23,7 +23,6 @@ from ..utils import (
     cli_valueless_option,
     determine_ext,
     encodeArgument,
-    encodeFilename,
     find_available_port,
     remove_end,
     traverse_obj,
@@ -67,7 +66,7 @@ class ExternalFD(FragmentFD):
                 'elapsed': time.time() - started,
             }
             if filename != '-':
-                fsize = os.path.getsize(encodeFilename(tmpfilename))
+                fsize = os.path.getsize(tmpfilename)
                 self.try_rename(tmpfilename, filename)
                 status.update({
                     'downloaded_bytes': fsize,
@@ -184,9 +183,9 @@ class ExternalFD(FragmentFD):
             dest.write(decrypt_fragment(fragment, src.read()))
             src.close()
             if not self.params.get('keep_fragments', False):
-                self.try_remove(encodeFilename(fragment_filename))
+                self.try_remove(fragment_filename)
         dest.close()
-        self.try_remove(encodeFilename(f'{tmpfilename}.frag.urls'))
+        self.try_remove(f'{tmpfilename}.frag.urls')
         return 0
 
     def _call_process(self, cmd, info_dict):
@@ -620,7 +619,7 @@ class FFmpegFD(ExternalFD):
         args += self._configuration_args(('_o1', '_o', ''))
 
         args = [encodeArgument(opt) for opt in args]
-        args.append(encodeFilename(ffpp._ffmpeg_filename_argument(tmpfilename), True))
+        args.append(ffpp._ffmpeg_filename_argument(tmpfilename))
         self._debug_cmd(args)
 
         piped = any(fmt['url'] in ('-', 'pipe:') for fmt in selected_formats)

yt_dlp/downloader/fragment.py

@@ -9,10 +9,9 @@ import time
 from .common import FileDownloader
 from .http import HttpFD
 from ..aes import aes_cbc_decrypt_bytes, unpad_pkcs7
-from ..compat import compat_os_name
 from ..networking import Request
 from ..networking.exceptions import HTTPError, IncompleteRead
-from ..utils import DownloadError, RetryManager, encodeFilename, traverse_obj
+from ..utils import DownloadError, RetryManager, traverse_obj
 from ..utils.networking import HTTPHeaderDict
 from ..utils.progress import ProgressCalculator
 
@@ -152,7 +151,7 @@ class FragmentFD(FileDownloader):
             if self.__do_ytdl_file(ctx):
                 self._write_ytdl_file(ctx)
             if not self.params.get('keep_fragments', False):
-                self.try_remove(encodeFilename(ctx['fragment_filename_sanitized']))
+                self.try_remove(ctx['fragment_filename_sanitized'])
             del ctx['fragment_filename_sanitized']
 
     def _prepare_frag_download(self, ctx):
@@ -188,7 +187,7 @@ class FragmentFD(FileDownloader):
         })
 
         if self.__do_ytdl_file(ctx):
-            ytdl_file_exists = os.path.isfile(encodeFilename(self.ytdl_filename(ctx['filename'])))
+            ytdl_file_exists = os.path.isfile(self.ytdl_filename(ctx['filename']))
             continuedl = self.params.get('continuedl', True)
             if continuedl and ytdl_file_exists:
                 self._read_ytdl_file(ctx)
@@ -390,7 +389,7 @@ class FragmentFD(FileDownloader):
             def __exit__(self, exc_type, exc_val, exc_tb):
                 pass
 
-        if compat_os_name == 'nt':
+        if os.name == 'nt':
             def future_result(future):
                 while True:
                     try:

yt_dlp/downloader/http.py

@@ -15,7 +15,6 @@ from ..utils import (
     ThrottledDownload,
     XAttrMetadataError,
     XAttrUnavailableError,
-    encodeFilename,
     int_or_none,
     parse_http_range,
     try_call,
@@ -58,9 +57,8 @@ class HttpFD(FileDownloader):
 
         if self.params.get('continuedl', True):
             # Establish possible resume length
-            if os.path.isfile(encodeFilename(ctx.tmpfilename)):
+            if os.path.isfile(ctx.tmpfilename):
-                ctx.resume_len = os.path.getsize(
+                ctx.resume_len = os.path.getsize(ctx.tmpfilename)
-                    encodeFilename(ctx.tmpfilename))
 
         ctx.is_resume = ctx.resume_len > 0
 
@@ -241,7 +239,7 @@ class HttpFD(FileDownloader):
                     ctx.resume_len = byte_counter
                 else:
                     try:
-                        ctx.resume_len = os.path.getsize(encodeFilename(ctx.tmpfilename))
+                        ctx.resume_len = os.path.getsize(ctx.tmpfilename)
                     except FileNotFoundError:
                         ctx.resume_len = 0
                 raise RetryDownload(e)

yt_dlp/downloader/rtmp.py

@@ -8,7 +8,6 @@ from ..utils import (
     Popen,
     check_executable,
     encodeArgument,
-    encodeFilename,
     get_exe_version,
 )
 
@@ -179,7 +178,7 @@ class RtmpFD(FileDownloader):
             return False
 
         while retval in (RD_INCOMPLETE, RD_FAILED) and not test and not live:
-            prevsize = os.path.getsize(encodeFilename(tmpfilename))
+            prevsize = os.path.getsize(tmpfilename)
             self.to_screen(f'[rtmpdump] Downloaded {prevsize} bytes')
             time.sleep(5.0)  # This seems to be needed
             args = [*basic_args, '--resume']
@@ -187,7 +186,7 @@ class RtmpFD(FileDownloader):
                 args += ['--skip', '1']
             args = [encodeArgument(a) for a in args]
             retval = run_rtmpdump(args)
-            cursize = os.path.getsize(encodeFilename(tmpfilename))
+            cursize = os.path.getsize(tmpfilename)
             if prevsize == cursize and retval == RD_FAILED:
                 break
             # Some rtmp streams seem abort after ~ 99.8%. Don't complain for those
@@ -196,7 +195,7 @@ class RtmpFD(FileDownloader):
                 retval = RD_SUCCESS
                 break
         if retval == RD_SUCCESS or (test and retval == RD_INCOMPLETE):
-            fsize = os.path.getsize(encodeFilename(tmpfilename))
+            fsize = os.path.getsize(tmpfilename)
             self.to_screen(f'[rtmpdump] Downloaded {fsize} bytes')
             self.try_rename(tmpfilename, filename)
             self._hook_progress({

yt_dlp/downloader/rtsp.py

@@ -2,7 +2,7 @@ import os
 import subprocess
 
 from .common import FileDownloader
-from ..utils import check_executable, encodeFilename
+from ..utils import check_executable
 
 
 class RtspFD(FileDownloader):
@@ -26,7 +26,7 @@ class RtspFD(FileDownloader):
 
         retval = subprocess.call(args)
         if retval == 0:
-            fsize = os.path.getsize(encodeFilename(tmpfilename))
+            fsize = os.path.getsize(tmpfilename)
             self.to_screen(f'\r[{args[0]}] {fsize} bytes')
             self.try_rename(tmpfilename, filename)
             self._hook_progress({

yt_dlp/extractor/abematv.py

@@ -18,10 +18,8 @@ from ..networking.exceptions import TransportError
 from ..utils import (
     ExtractorError,
     OnDemandPagedList,
-    bytes_to_intlist,
     decode_base_n,
     int_or_none,
-    intlist_to_bytes,
     time_seconds,
     traverse_obj,
     update_url_query,
@@ -72,15 +70,15 @@ class AbemaLicenseRH(RequestHandler):
             })
 
         res = decode_base_n(license_response['k'], table=self._STRTABLE)
-        encvideokey = bytes_to_intlist(struct.pack('>QQ', res >> 64, res & 0xffffffffffffffff))
+        encvideokey = list(struct.pack('>QQ', res >> 64, res & 0xffffffffffffffff))
 
         h = hmac.new(
             binascii.unhexlify(self._HKEY),
             (license_response['cid'] + self.ie._DEVICE_ID).encode(),
             digestmod=hashlib.sha256)
-        enckey = bytes_to_intlist(h.digest())
+        enckey = list(h.digest())
 
-        return intlist_to_bytes(aes_ecb_decrypt(encvideokey, enckey))
+        return bytes(aes_ecb_decrypt(encvideokey, enckey))
 
 
 class AbemaTVBaseIE(InfoExtractor):

yt_dlp/extractor/adn.py

@@ -11,11 +11,9 @@ from ..networking.exceptions import HTTPError
 from ..utils import (
     ExtractorError,
     ass_subtitles_timecode,
-    bytes_to_intlist,
     bytes_to_long,
     float_or_none,
     int_or_none,
-    intlist_to_bytes,
     join_nonempty,
     long_to_bytes,
     parse_iso8601,
@@ -198,7 +196,7 @@ Format: Marked,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text'''
 
         links_url = try_get(options, lambda x: x['video']['url']) or (video_base_url + 'link')
         self._K = ''.join(random.choices('0123456789abcdef', k=16))
-        message = bytes_to_intlist(json.dumps({
+        message = list(json.dumps({
             'k': self._K,
             't': token,
         }))
@@ -207,7 +205,7 @@ Format: Marked,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text'''
         # a different random padding
         links_data = None
         for _ in range(3):
-            padded_message = intlist_to_bytes(pkcs1pad(message, 128))
+            padded_message = bytes(pkcs1pad(message, 128))
             n, e = self._RSA_KEY
             encrypted_message = long_to_bytes(pow(bytes_to_long(padded_message), e, n))
             authorization = base64.b64encode(encrypted_message).decode()

yt_dlp/extractor/anvato.py

@@ -8,10 +8,8 @@ import time
 from .common import InfoExtractor
 from ..aes import aes_encrypt
 from ..utils import (
-    bytes_to_intlist,
     determine_ext,
     int_or_none,
-    intlist_to_bytes,
     join_nonempty,
     smuggle_url,
     strip_jsonp,
@@ -234,8 +232,8 @@ class AnvatoIE(InfoExtractor):
         server_time = self._server_time(access_key, video_id)
         input_data = f'{server_time}~{md5_text(video_data_url)}~{md5_text(server_time)}'
 
-        auth_secret = intlist_to_bytes(aes_encrypt(
+        auth_secret = bytes(aes_encrypt(
-            bytes_to_intlist(input_data[:64]), bytes_to_intlist(self._AUTH_KEY)))
+            list(input_data[:64]), list(self._AUTH_KEY)))
         query = {
             'X-Anvato-Adst-Auth': base64.b64encode(auth_secret).decode('ascii'),
             'rtyp': 'fp',

yt_dlp/extractor/common.py

@@ -25,7 +25,6 @@ import xml.etree.ElementTree
 from ..compat import (
     compat_etree_fromstring,
     compat_expanduser,
-    compat_os_name,
     urllib_req_to_req,
 )
 from ..cookies import LenientSimpleCookie
@@ -1028,7 +1027,7 @@ class InfoExtractor:
         filename = sanitize_filename(f'{basen}.dump', restricted=True)
         # Working around MAX_PATH limitation on Windows (see
         # http://msdn.microsoft.com/en-us/library/windows/desktop/aa365247(v=vs.85).aspx)
-        if compat_os_name == 'nt':
+        if os.name == 'nt':
             absfilepath = os.path.abspath(filename)
             if len(absfilepath) > 259:
                 filename = fR'\\?\{absfilepath}'

yt_dlp/extractor/shemaroome.py

@@ -4,8 +4,6 @@ from .common import InfoExtractor
 from ..aes import aes_cbc_decrypt, unpad_pkcs7
 from ..utils import (
     ExtractorError,
-    bytes_to_intlist,
-    intlist_to_bytes,
     unified_strdate,
 )
 
@@ -68,10 +66,10 @@ class ShemarooMeIE(InfoExtractor):
         data_json = self._download_json('https://www.shemaroome.com/users/user_all_lists', video_id, data=data.encode())
         if not data_json.get('status'):
             raise ExtractorError('Premium videos cannot be downloaded yet.', expected=True)
-        url_data = bytes_to_intlist(base64.b64decode(data_json['new_play_url']))
+        url_data = list(base64.b64decode(data_json['new_play_url']))
-        key = bytes_to_intlist(base64.b64decode(data_json['key']))
+        key = list(base64.b64decode(data_json['key']))
         iv = [0] * 16
-        m3u8_url = unpad_pkcs7(intlist_to_bytes(aes_cbc_decrypt(url_data, key, iv))).decode('ascii')
+        m3u8_url = unpad_pkcs7(bytes(aes_cbc_decrypt(url_data, key, iv))).decode('ascii')
         headers = {'stream_key': data_json['stream_key']}
         formats, m3u8_subs = self._extract_m3u8_formats_and_subtitles(m3u8_url, video_id, fatal=False, headers=headers)
         for fmt in formats:

yt_dlp/postprocessor/common.py

@@ -9,7 +9,6 @@ from ..utils import (
     RetryManager,
     _configuration_args,
     deprecation_warning,
-    encodeFilename,
 )
 
 
@@ -151,7 +150,7 @@ class PostProcessor(metaclass=PostProcessorMetaClass):
 
     def try_utime(self, path, atime, mtime, errnote='Cannot update utime of file'):
         try:
-            os.utime(encodeFilename(path), (atime, mtime))
+            os.utime(path, (atime, mtime))
         except Exception:
             self.report_warning(errnote)
 

yt_dlp/postprocessor/embedthumbnail.py

@@ -12,7 +12,6 @@ from ..utils import (
     PostProcessingError,
     check_executable,
     encodeArgument,
-    encodeFilename,
     prepend_extension,
     shell_quote,
 )
@@ -68,7 +67,7 @@ class EmbedThumbnailPP(FFmpegPostProcessor):
             self.to_screen('There are no thumbnails on disk')
             return [], info
         thumbnail_filename = info['thumbnails'][idx]['filepath']
-        if not os.path.exists(encodeFilename(thumbnail_filename)):
+        if not os.path.exists(thumbnail_filename):
             self.report_warning('Skipping embedding the thumbnail because the file is missing.')
             return [], info
 
@@ -85,7 +84,7 @@ class EmbedThumbnailPP(FFmpegPostProcessor):
             thumbnail_filename = convertor.convert_thumbnail(thumbnail_filename, 'png')
             thumbnail_ext = 'png'
 
-        mtime = os.stat(encodeFilename(filename)).st_mtime
+        mtime = os.stat(filename).st_mtime
 
         success = True
         if info['ext'] == 'mp3':
@@ -154,12 +153,12 @@ class EmbedThumbnailPP(FFmpegPostProcessor):
                 else:
                     if not prefer_atomicparsley:
                         self.to_screen('mutagen was not found. Falling back to AtomicParsley')
-                    cmd = [encodeFilename(atomicparsley, True),
+                    cmd = [atomicparsley,
-                           encodeFilename(filename, True),
+                           filename,
                            encodeArgument('--artwork'),
-                           encodeFilename(thumbnail_filename, True),
+                           thumbnail_filename,
                            encodeArgument('-o'),
-                           encodeFilename(temp_filename, True)]
+                           temp_filename]
                     cmd += [encodeArgument(o) for o in self._configuration_args('AtomicParsley')]
 
                     self._report_run('atomicparsley', filename)

yt_dlp/postprocessor/ffmpeg.py

@@ -21,7 +21,6 @@ from ..utils import (
     determine_ext,
     dfxp2srt,
     encodeArgument,
-    encodeFilename,
     filter_dict,
     float_or_none,
     is_outdated_version,
@@ -243,13 +242,13 @@ class FFmpegPostProcessor(PostProcessor):
         try:
             if self.probe_available:
                 cmd = [
-                    encodeFilename(self.probe_executable, True),
+                    self.probe_executable,
                     encodeArgument('-show_streams')]
             else:
                 cmd = [
-                    encodeFilename(self.executable, True),
+                    self.executable,
                     encodeArgument('-i')]
-            cmd.append(encodeFilename(self._ffmpeg_filename_argument(path), True))
+            cmd.append(self._ffmpeg_filename_argument(path))
             self.write_debug(f'{self.basename} command line: {shell_quote(cmd)}')
             stdout, stderr, returncode = Popen.run(
                 cmd, text=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
@@ -282,7 +281,7 @@ class FFmpegPostProcessor(PostProcessor):
         self.check_version()
 
         cmd = [
-            encodeFilename(self.probe_executable, True),
+            self.probe_executable,
             encodeArgument('-hide_banner'),
             encodeArgument('-show_format'),
             encodeArgument('-show_streams'),
@@ -335,9 +334,9 @@ class FFmpegPostProcessor(PostProcessor):
         self.check_version()
 
         oldest_mtime = min(
-            os.stat(encodeFilename(path)).st_mtime for path, _ in input_path_opts if path)
+            os.stat(path).st_mtime for path, _ in input_path_opts if path)
 
-        cmd = [encodeFilename(self.executable, True), encodeArgument('-y')]
+        cmd = [self.executable, encodeArgument('-y')]
         # avconv does not have repeat option
         if self.basename == 'ffmpeg':
             cmd += [encodeArgument('-loglevel'), encodeArgument('repeat+info')]
@@ -353,7 +352,7 @@ class FFmpegPostProcessor(PostProcessor):
                 args.append('-i')
             return (
                 [encodeArgument(arg) for arg in args]
-                + [encodeFilename(self._ffmpeg_filename_argument(file), True)])
+                + [self._ffmpeg_filename_argument(file)])
 
         for arg_type, path_opts in (('i', input_path_opts), ('o', output_path_opts)):
             cmd += itertools.chain.from_iterable(
@@ -522,8 +521,8 @@ class FFmpegExtractAudioPP(FFmpegPostProcessor):
                 return [], information
             orig_path = prepend_extension(path, 'orig')
             temp_path = prepend_extension(path, 'temp')
-        if (self._nopostoverwrites and os.path.exists(encodeFilename(new_path))
+        if (self._nopostoverwrites and os.path.exists(new_path)
-                and os.path.exists(encodeFilename(orig_path))):
+                and os.path.exists(orig_path)):
             self.to_screen(f'Post-process file {new_path} exists, skipping')
             return [], information
 
@@ -838,7 +837,7 @@ class FFmpegMergerPP(FFmpegPostProcessor):
                 args.extend(['-map', f'{i}:v:0'])
         self.to_screen(f'Merging formats into "{filename}"')
         self.run_ffmpeg_multiple_files(info['__files_to_merge'], temp_filename, args)
-        os.rename(encodeFilename(temp_filename), encodeFilename(filename))
+        os.rename(temp_filename, filename)
         return info['__files_to_merge'], info
 
     def can_merge(self):
@@ -1039,7 +1038,7 @@ class FFmpegSplitChaptersPP(FFmpegPostProcessor):
 
     def _ffmpeg_args_for_chapter(self, number, chapter, info):
         destination = self._prepare_filename(number, chapter, info)
-        if not self._downloader._ensure_dir_exists(encodeFilename(destination)):
+        if not self._downloader._ensure_dir_exists(destination):
             return
 
         chapter['filepath'] = destination

yt_dlp/postprocessor/movefilesafterdownload.py

@@ -4,8 +4,6 @@ from .common import PostProcessor
 from ..compat import shutil
 from ..utils import (
     PostProcessingError,
-    decodeFilename,
-    encodeFilename,
     make_dir,
 )
 
@@ -21,25 +19,25 @@ class MoveFilesAfterDownloadPP(PostProcessor):
         return 'MoveFiles'
 
     def run(self, info):
-        dl_path, dl_name = os.path.split(encodeFilename(info['filepath']))
+        dl_path, dl_name = os.path.split(info['filepath'])
         finaldir = info.get('__finaldir', dl_path)
         finalpath = os.path.join(finaldir, dl_name)
         if self._downloaded:
-            info['__files_to_move'][info['filepath']] = decodeFilename(finalpath)
+            info['__files_to_move'][info['filepath']] = finalpath
 
-        make_newfilename = lambda old: decodeFilename(os.path.join(finaldir, os.path.basename(encodeFilename(old))))
+        make_newfilename = lambda old: os.path.join(finaldir, os.path.basename(old))
         for oldfile, newfile in info['__files_to_move'].items():
             if not newfile:
                 newfile = make_newfilename(oldfile)
-            if os.path.abspath(encodeFilename(oldfile)) == os.path.abspath(encodeFilename(newfile)):
+            if os.path.abspath(oldfile) == os.path.abspath(newfile):
                 continue
-            if not os.path.exists(encodeFilename(oldfile)):
+            if not os.path.exists(oldfile):
                 self.report_warning(f'File "{oldfile}" cannot be found')
                 continue
-            if os.path.exists(encodeFilename(newfile)):
+            if os.path.exists(newfile):
                 if self.get_param('overwrites', True):
                     self.report_warning(f'Replacing existing file "{newfile}"')
-                    os.remove(encodeFilename(newfile))
+                    os.remove(newfile)
                 else:
                     self.report_warning(
                         f'Cannot move file "{oldfile}" out of temporary directory since "{newfile}" already exists. ')

yt_dlp/postprocessor/sponskrub.py

@@ -9,7 +9,6 @@ from ..utils import (
     check_executable,
     cli_option,
     encodeArgument,
-    encodeFilename,
     prepend_extension,
     shell_quote,
     str_or_none,
@@ -52,7 +51,7 @@ class SponSkrubPP(PostProcessor):
             return [], information
 
         filename = information['filepath']
-        if not os.path.exists(encodeFilename(filename)):  # no download
+        if not os.path.exists(filename):  # no download
             return [], information
 
         if information['extractor_key'].lower() != 'youtube':
@@ -71,8 +70,8 @@ class SponSkrubPP(PostProcessor):
                 self.report_warning('If sponskrub is run multiple times, unintended parts of the video could be cut out.')
 
         temp_filename = prepend_extension(filename, self._temp_ext)
-        if os.path.exists(encodeFilename(temp_filename)):
+        if os.path.exists(temp_filename):
-            os.remove(encodeFilename(temp_filename))
+            os.remove(temp_filename)
 
         cmd = [self.path]
         if not self.cutout:

yt_dlp/postprocessor/xattrpp.py

@@ -1,7 +1,6 @@
 import os
 
 from .common import PostProcessor
-from ..compat import compat_os_name
 from ..utils import (
     PostProcessingError,
     XAttrMetadataError,
@@ -57,7 +56,7 @@ class XAttrMetadataPP(PostProcessor):
                 elif e.reason == 'VALUE_TOO_LONG':
                     self.report_warning(f'Unable to write extended attribute "{xattrname}" due to too long values.')
                 else:
-                    tip = ('You need to use NTFS' if compat_os_name == 'nt'
+                    tip = ('You need to use NTFS' if os.name == 'nt'
                            else 'You may have to enable them in your "/etc/fstab"')
                     raise PostProcessingError(f'This filesystem doesn\'t support extended attributes. {tip}')
 

yt_dlp/update.py

@@ -13,7 +13,6 @@ import sys
 from dataclasses import dataclass
 from zipimport import zipimporter
 
-from .compat import compat_realpath
 from .networking import Request
 from .networking.exceptions import HTTPError, network_exceptions
 from .utils import (
@@ -523,7 +522,7 @@ class Updater:
     @functools.cached_property
     def filename(self):
         """Filename of the executable"""
-        return compat_realpath(_get_variant_and_executable_path()[1])
+        return os.path.realpath(_get_variant_and_executable_path()[1])
 
     @functools.cached_property
     def cmd(self):

yt_dlp/utils/_deprecated.py

@@ -9,31 +9,23 @@ passthrough_module(__name__, '.._legacy', callback=lambda attr: warnings.warn(
 del passthrough_module
 
 
-from ._utils import preferredencoding
+import re
+import struct
 
 
-def encodeFilename(s, for_subprocess=False):
+def bytes_to_intlist(bs):
-    assert isinstance(s, str)
+    if not bs:
-    return s
+        return []
+    if isinstance(bs[0], int):  # Python 3
+        return list(bs)
+    else:
+        return [ord(c) for c in bs]
 
 
-def decodeFilename(b, for_subprocess=False):
+def intlist_to_bytes(xs):
-    return b
+    if not xs:
+        return b''
+    return struct.pack('%dB' % len(xs), *xs)
 
 
-def decodeArgument(b):
+compiled_regex_type = type(re.compile(''))
-    return b
-
-
-def decodeOption(optval):
-    if optval is None:
-        return optval
-    if isinstance(optval, bytes):
-        optval = optval.decode(preferredencoding())
-
-    assert isinstance(optval, str)
-    return optval
-
-
-def error_to_compat_str(err):
-    return str(err)

yt_dlp/utils/_legacy.py

@@ -313,3 +313,30 @@ def make_HTTPS_handler(params, **kwargs):
 
 def process_communicate_or_kill(p, *args, **kwargs):
     return Popen.communicate_or_kill(p, *args, **kwargs)
+
+
+def encodeFilename(s, for_subprocess=False):
+    assert isinstance(s, str)
+    return s
+
+
+def decodeFilename(b, for_subprocess=False):
+    return b
+
+
+def decodeArgument(b):
+    return b
+
+
+def decodeOption(optval):
+    if optval is None:
+        return optval
+    if isinstance(optval, bytes):
+        optval = optval.decode(preferredencoding())
+
+    assert isinstance(optval, str)
+    return optval
+
+
+def error_to_compat_str(err):
+    return str(err)

yt_dlp/utils/_utils.py

@@ -49,15 +49,11 @@ from ..compat import (
     compat_etree_fromstring,
     compat_expanduser,
     compat_HTMLParseError,
-    compat_os_name,
 )
 from ..dependencies import xattr
 
 __name__ = __name__.rsplit('.', 1)[0]  # noqa: A001: Pretend to be the parent module
 
-# This is not clearly defined otherwise
-compiled_regex_type = type(re.compile(''))
-
 
 class NO_DEFAULT:
     pass
@@ -874,7 +870,7 @@ class Popen(subprocess.Popen):
             kwargs.setdefault('encoding', 'utf-8')
             kwargs.setdefault('errors', 'replace')
 
-        if shell and compat_os_name == 'nt' and kwargs.get('executable') is None:
+        if shell and os.name == 'nt' and kwargs.get('executable') is None:
             if not isinstance(args, str):
                 args = shell_quote(args, shell=True)
             shell = False
@@ -1457,7 +1453,7 @@ def system_identifier():
 @functools.cache
 def get_windows_version():
     """ Get Windows version. returns () if it's not running on Windows """
-    if compat_os_name == 'nt':
+    if os.name == 'nt':
         return version_tuple(platform.win32_ver()[1])
     else:
         return ()
@@ -1470,7 +1466,7 @@ def write_string(s, out=None, encoding=None):
     if not out:
         return
 
-    if compat_os_name == 'nt' and supports_terminal_sequences(out):
+    if os.name == 'nt' and supports_terminal_sequences(out):
         s = re.sub(r'([\r\n]+)', r' \1', s)
 
     enc, buffer = None, out
@@ -1503,21 +1499,6 @@ def deprecation_warning(msg, *, printer=None, stacklevel=0, **kwargs):
 deprecation_warning._cache = set()
 
 
-def bytes_to_intlist(bs):
-    if not bs:
-        return []
-    if isinstance(bs[0], int):  # Python 3
-        return list(bs)
-    else:
-        return [ord(c) for c in bs]
-
-
-def intlist_to_bytes(xs):
-    if not xs:
-        return b''
-    return struct.pack('%dB' % len(xs), *xs)
-
-
 class LockingUnsupportedError(OSError):
     msg = 'File locking is not supported'
 
@@ -1701,7 +1682,7 @@ _CMD_QUOTE_TRANS = str.maketrans({
 def shell_quote(args, *, shell=False):
     args = list(variadic(args))
 
-    if compat_os_name != 'nt':
+    if os.name != 'nt':
         return shlex.join(args)
 
     trans = _CMD_QUOTE_TRANS if shell else _WINDOWS_QUOTE_TRANS
@@ -4516,7 +4497,7 @@ def urshift(val, n):
 def write_xattr(path, key, value):
     # Windows: Write xattrs to NTFS Alternate Data Streams:
     # http://en.wikipedia.org/wiki/NTFS#Alternate_data_streams_.28ADS.29
-    if compat_os_name == 'nt':
+    if os.name == 'nt':
         assert ':' not in key
         assert os.path.exists(path)
 
@@ -4778,12 +4759,12 @@ def jwt_decode_hs256(jwt):
     return json.loads(base64.urlsafe_b64decode(f'{payload_b64}==='))
 
 
-WINDOWS_VT_MODE = False if compat_os_name == 'nt' else None
+WINDOWS_VT_MODE = False if os.name == 'nt' else None
 
 
 @functools.cache
 def supports_terminal_sequences(stream):
-    if compat_os_name == 'nt':
+    if os.name == 'nt':
         if not WINDOWS_VT_MODE:
             return False
     elif not os.getenv('TERM'):
@@ -4878,7 +4859,7 @@ def parse_http_range(range):
 
 def read_stdin(what):
     if what:
-        eof = 'Ctrl+Z' if compat_os_name == 'nt' else 'Ctrl+D'
+        eof = 'Ctrl+Z' if os.name == 'nt' else 'Ctrl+D'
         write_string(f'Reading {what} from STDIN - EOF ({eof}) to end:\n')
     return sys.stdin