Repository: streamlink/streamlink
Pull request #3993 (2021-09-05 04:51:23)

cli: add {time:format} var to --output / --title
This adds support for the `{time}` / `{time:format}` variable to `--title` and `--output`, which has been a much requested feature in the past. The answer has always been "use your command line shell", which is not that great and Streamlink should support something very basic like that.

The `:format` part is optional and based on the standard string format specifiers described in PEP3101:
https://www.python.org/dev/peps/pep-3101/#format-specifiers
The logic for this is however much simplified and doesn't support nesting, etc, because we don't need it here.

Format strings on variables where there's no formatting callback registered get ignored. Conversion flags like `!r`, `!s`, etc, get ignored as well.
https://www.python.org/dev/peps/pep-3101/#explicit-conversion-flag

----

- In order to re-format the variables' values, the Formatter class had to be rewritten, based on string.Formatter's parse() method. It doesn't inherit from string.Formatter because of the formatting simplification. This is the first commit, which also fixes/improves the tests.
- The second commit then adds support for the parsed `format_spec`. Instead of adding a second parameter to the Formatter's constructor with formatting functions, the mapping functions could've also received a parameter with the optional formatting string, but that would've prevented caching, so this is the better choice (I think).
- The third commit adds the `{time}` var to the CLI module, fixes the default string value of datetime.now() (for file name reasons), and adds documentation. Linking to the python docs ([`datetime.strftime()`](https://docs.python.org/3/library/datetime.html#strftime-and-strptime-format-codes)) via reST is unfortunately not possible in argparse arguments because of the `--help` output and man page content, which can't have reST.

gravyboat: Great addition @bastimeyer, thanks!

Patch:
diff --git a/src/streamlink_cli/argparser.py b/src/streamlink_cli/argparser.py
index cf95a7ea..87f31e91 100644
--- a/src/streamlink_cli/argparser.py
+++ b/src/streamlink_cli/argparser.py
@@ -569,6 +569,11 @@ def build_parser():
         {{url}}
             URL of the stream.
 
+        {{time}}
+            The current timestamp, which can optionally be formatted via {{time:format}}.
+            This format parameter string is passed to Python's datetime.strftime() method,
+            so all usual time directives are available. The default format is "%Y-%m-%d_%H-%M-%S".
+
         Examples:
 
             %(prog)s -p vlc --title "{{title}} -!- {{author}} -!- {{category}} \\$A" <url> [stream]
diff --git a/src/streamlink_cli/main.py b/src/streamlink_cli/main.py
index f6bb845b..77701975 100644
--- a/src/streamlink_cli/main.py
+++ b/src/streamlink_cli/main.py
@@ -1,5 +1,4 @@
 import argparse
-import datetime
 import errno
 import logging
 import os
@@ -32,7 +31,7 @@ from streamlink_cli.compat import DeprecatedPath, is_win32, stdout
 from streamlink_cli.console import ConsoleOutput, ConsoleUserInputRequester
 from streamlink_cli.constants import CONFIG_FILES, DEFAULT_STREAM_METADATA, LOG_DIR, PLUGIN_DIRS, STREAM_SYNONYMS
 from streamlink_cli.output import FileOutput, Output, PlayerOutput
-from streamlink_cli.utils import Formatter, HTTPServer, ignored, progress, stream_to_url
+from streamlink_cli.utils import Formatter, HTTPServer, datetime, ignored, progress, stream_to_url
 
 ACCEPTABLE_ERRNO = (errno.EPIPE, errno.EINVAL, errno.ECONNRESET)
 try:
@@ -51,6 +50,22 @@ streamlink: Streamlink = None
 log = logging.getLogger("streamlink.cli")
 
 
+def get_formatter(plugin: Plugin):
+    return Formatter(
+        {
+            "url": lambda: args.url,
+            "author": lambda: plugin.get_author(),
+            "category": lambda: plugin.get_category(),
+            "game": lambda: plugin.get_category(),
+            "title": lambda: plugin.get_title(),
+            "time": lambda: datetime.now()
+        },
+        {
+            "time": lambda dt, fmt: dt.strftime(fmt)
+        }
+    )
+
+
 def check_file_output(filename, force):
     """Checks if file already exists and ask the user if it should
     be overwritten if it does."""
@@ -427,13 +442,7 @@ def handle_stream(plugin, streams, stream_name):
                                   sorted(streams.keys())))
         file_output = args.output or args.stdout
 
-        formatter = Formatter({
-            "url": lambda: args.url,
-            "author": lambda: plugin.get_author(),
-            "category": lambda: plugin.get_category(),
-            "game": lambda: plugin.get_category(),
-            "title": lambda: plugin.get_title(),
-        })
+        formatter = get_formatter(plugin)
 
         for stream_name in [stream_name] + alt_streams:
             stream = streams[stream_name]
@@ -987,7 +996,7 @@ def setup_logger_and_console(stream=sys.stdout, filename=None, level="info", jso
     global console
 
     if filename == "-":
-        filename = LOG_DIR / datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S.log")
+        filename = LOG_DIR / f"{datetime.now()}.log"
     elif filename:
         filename = Path(filename).expanduser().resolve()
 
diff --git a/src/streamlink_cli/utils/__init__.py b/src/streamlink_cli/utils/__init__.py
index bacfcbcf..2d5fe067 100644
--- a/src/streamlink_cli/utils/__init__.py
+++ b/src/streamlink_cli/utils/__init__.py
@@ -1,5 +1,6 @@
 import json
 from contextlib import contextmanager
+from datetime import datetime as _datetime
 
 from streamlink_cli.utils.formatter import Formatter
 from streamlink_cli.utils.http_server import HTTPServer
@@ -9,6 +10,7 @@ from streamlink_cli.utils.stream import stream_to_url
 
 __all__ = [
     "Formatter", "HTTPServer", "JSONEncoder",
+    "datetime",
     "find_default_player", "ignored", "progress", "stream_to_url"
 ]
 
@@ -29,3 +31,9 @@ def ignored(*exceptions):
         yield
     except exceptions:
         pass
+
+
+# noinspection PyPep8Naming
+class datetime(_datetime):
+    def __str__(self):
+        return self.strftime("%Y-%m-%d_%H-%M-%S")
diff --git a/src/streamlink_cli/utils/formatter.py b/src/streamlink_cli/utils/formatter.py
index 8fe640d7..4b71ae4c 100644
--- a/src/streamlink_cli/utils/formatter.py
+++ b/src/streamlink_cli/utils/formatter.py
@@ -1,52 +1,63 @@
-from typing import Callable, Dict, Optional, Type, Union
+from string import Formatter as StringFormatter
+from typing import Any, Callable, Dict, Optional
 
 from streamlink_cli.utils.path import replace_chars
 
 
-class _UNKNOWN:
-    pass
+# we only need string.Formatter for calling its parse() method, which returns `_string.formatter_parser(string)`.
+_stringformatter = StringFormatter()
 
 
-class Formatter(dict):
-    def __init__(self, mapping: Dict[str, Callable]):
+def _identity(obj):
+    return obj
+
+
+class Formatter:
+    def __init__(self, mapping: Dict[str, Callable], formatting: Optional[Dict[str, Callable]] = None):
         super().__init__()
         self.mapping = mapping
-        self.cache = dict()
+        self.formatting = formatting or {}
+        self.cache = {}
 
-    def __missing__(self, key: str) -> Union[Type[_UNKNOWN], None, str]:
-        if key not in self.mapping:
-            return _UNKNOWN
+    def _get_value(self, field_name: str, format_spec: str, defaults: Dict[str, str]) -> Any:
+        if field_name not in self.mapping:
+            return defaults.get(field_name, f"{{{field_name}}}" if not format_spec else f"{{{field_name}:{format_spec}}}")
 
-        if key in self.cache:
-            return self.cache[key]
+        if field_name in self.cache:
+            value = self.cache[field_name]
+        else:
+            value = self.mapping[field_name]()
+            self.cache[field_name] = value
 
-        value = self.mapping[key]()
-        self.cache[key] = value
+        if value is None:
+            value = defaults.get(field_name, "")
+
+        if format_spec and field_name in self.formatting:
+            # noinspection PyBroadException
+            try:
+                return self.formatting[field_name](value, format_spec)
+            except Exception:
+                return f"{{{field_name}:{format_spec}}}"
 
         return value
 
-    def _format(self, string: str, wrapper: Callable[[str], str], defaults: Optional[Dict] = None) -> str:
-        return string.format_map(_Wrapper(self, wrapper, defaults))
+    def _format(self, string: str, mapper: Callable[[str], str], defaults: Dict[str, str]) -> str:
+        result = []
 
-    def filename(self, filename: str, charmap: Optional[str] = None) -> str:
-        return self._format(filename, lambda s: replace_chars(s, charmap))
+        for literal_text, field_name, format_spec, conversion in _stringformatter.parse(string):
+            if literal_text:
+                result.append(literal_text)
 
-    def title(self, title: str, defaults: Optional[Dict] = None) -> str:
-        return self._format(title, lambda s: s, defaults)
+            if field_name is None:
+                continue
 
+            value = self._get_value(field_name, format_spec, defaults)
+            result.append(mapper(str(value)))
 
-class _Wrapper(dict):
-    def __init__(self, formatter: Formatter, wrapper: Callable[[str], str], defaults: Optional[Dict[str, str]] = None):
-        super().__init__()
-        self.formatter = formatter
-        self.wrapper = wrapper
-        self.defaults = defaults or {}
-
-    def __missing__(self, key: str) -> str:
-        value = self.formatter[key]
-        if value is _UNKNOWN:
-            value = str(self.defaults.get(key, f"{{{key}}}"))
-        elif value is None:
-            value = str(self.defaults.get(key, ""))
-
-        return self.wrapper(value)
+        return "".join(result)
+
+    def filename(self, filename: str, charmap: Optional[str] = None) -> str:
+        return self._format(filename, lambda s: replace_chars(s, charmap), {})
+
+    def title(self, title: str, defaults: Optional[Dict[str, str]] = None) -> str:
+        return self._format(title, _identity, defaults or {})


Test patch:
diff --git a/tests/test_cli_utils_formatter.py b/tests/test_cli_utils_formatter.py
index 0a3d40a5..61fada67 100644
--- a/tests/test_cli_utils_formatter.py
+++ b/tests/test_cli_utils_formatter.py
@@ -1,19 +1,36 @@
 import unittest
+from datetime import datetime
 from unittest.mock import Mock, call, patch
 
+from freezegun import freeze_time
+
 from streamlink_cli.utils.formatter import Formatter
 
 
+@freeze_time(datetime(2000, 1, 2, 3, 4, 5, 6, None))
 class TestFormatter(unittest.TestCase):
+    class Obj:
+        def __str__(self):
+            return "obj"
+
     def setUp(self):
         self.prop = Mock(return_value="prop")
-        self.formatter = Formatter({
-            "prop": self.prop,
-            "empty": lambda: "",
-            "none": lambda: None
-        })
+        self.obj = self.Obj()
+        self.formatter = Formatter(
+            {
+                "prop": self.prop,
+                "obj": lambda: self.obj,
+                "time": datetime.now,
+                "empty": lambda: "",
+                "none": lambda: None
+            },
+            {
+                "time": lambda dt, fmt: dt.strftime(fmt)
+            }
+        )
 
     def test_unknown(self):
+        self.assertEqual(self.formatter.title("{}"), "{}")
         self.assertEqual(self.formatter.title("some {unknown} variable"), "some {unknown} variable")
         self.assertEqual(self.formatter.title("some {unknown} variable", {"unknown": "known"}), "some known variable")
         self.assertEqual(self.formatter.cache, dict())
@@ -23,7 +40,13 @@ class TestFormatter(unittest.TestCase):
         self.assertEqual(self.formatter.cache, dict(prop="prop", empty="", none=None))
         self.assertEqual(self.prop.call_count, 1)
 
-        self.assertEqual(self.formatter.title("text '{prop}' '{empty}' '{none}'"), "text 'prop' '' ''")
+        self.assertEqual(self.formatter.title("text '{prop}' '{obj}' '{empty}' '{none}'"), "text 'prop' 'obj' '' ''")
+        self.assertEqual(self.formatter.cache, dict(prop="prop", obj=self.obj, empty="", none=None))
+        self.assertEqual(self.prop.call_count, 1)
+
+        defaults = dict(prop="PROP", obj="OBJ", empty="EMPTY", none="NONE")
+        self.assertEqual(self.formatter.title("'{prop}' '{obj}' '{empty}' '{none}'", defaults), "'prop' 'obj' '' 'NONE'")
+        self.assertEqual(self.formatter.cache, dict(prop="prop", obj=self.obj, empty="", none=None))
         self.assertEqual(self.prop.call_count, 1)
 
     @patch("streamlink_cli.utils.formatter.replace_chars")
@@ -36,12 +59,26 @@ class TestFormatter(unittest.TestCase):
         self.assertEqual(mock_replace_chars.call_args_list, [
             call("prop", None), call("", None), call("", None)
         ])
-
         mock_replace_chars.reset_mock()
 
-        self.assertEqual(self.formatter.filename("text '{prop}' '{empty}' '{none}'", "foo"), "text 'PROP' '' ''")
-        self.assertEqual(self.formatter.cache, dict(prop="prop", empty="", none=None))
+        self.assertEqual(self.formatter.filename("text '{prop}' '{obj}' '{empty}' '{none}'", "foo"), "text 'PROP' 'OBJ' '' ''")
+        self.assertEqual(self.formatter.cache, dict(prop="prop", obj=self.obj, empty="", none=None))
         self.assertEqual(self.prop.call_count, 1)
         self.assertEqual(mock_replace_chars.call_args_list, [
-            call("prop", "foo"), call("", "foo"), call("", "foo")
+            call("prop", "foo"), call("obj", "foo"), call("", "foo"), call("", "foo")
         ])
+
+    def test_format_spec(self):
+        self.assertEqual(self.formatter.title("{time}"), "2000-01-02 03:04:05.000006")
+        self.assertEqual(self.formatter.cache, dict(time=datetime(2000, 1, 2, 3, 4, 5, 6, None)))
+        self.assertEqual(self.formatter.title("{time:%Y}"), "2000")
+        self.assertEqual(self.formatter.title("{time:%Y-%m-%d}"), "2000-01-02")
+        self.assertEqual(self.formatter.title("{time:%H:%M:%S}"), "03:04:05")
+        with patch("datetime.datetime.strftime", side_effect=ValueError):
+            self.assertEqual(self.formatter.title("{time:foo:bar}"), "{time:foo:bar}")
+        self.assertEqual(self.formatter.cache, dict(time=datetime(2000, 1, 2, 3, 4, 5, 6, None)))
+
+        self.assertEqual(self.formatter.title("{prop:foo}"), "prop")
+        self.assertEqual(self.formatter.title("{none:foo}"), "")
+        self.assertEqual(self.formatter.title("{unknown:format}"), "{unknown:format}")
+        self.assertEqual(self.formatter.title("{unknown:format}", {"unknown": "known"}), "known")