Clean up 'real_main` and fix oversight with tempfile fallbacks. (#57)

These changes are from the old revision of open-in-mpv that includes
win32 support. I figured it would be wise to get these changes in first
before implementing and testing on Windows 11.

The tempfile oversight is caused by using the context variant of it;
upon exiting the context the contents and directory are deleted so the
only solution is global variables. If you can think of a better method
im up to hearing it.

open_in_mpv/constants.py

@@ -1,4 +1,4 @@
-from typing import Sequence
+from typing import Final, Sequence
 import os
 import platform
 
@@ -8,42 +8,47 @@
            'MAC_HOSTS_DIRS', 'SYSTEM_HOSTS_DIRS', 'USER_CHROME_HOSTS_REG_PATH_WIN',
            'USER_HOSTS_DIRS')
 
-IS_MAC = bool(platform.mac_ver()[0])
-IS_WIN = bool(platform.win32_ver()[0])
-IS_LINUX = not IS_MAC and not IS_WIN
+IS_MAC: Final[bool] = bool(platform.mac_ver()[0])
+IS_WIN: Final[bool] = bool(platform.win32_ver()[0])
+IS_LINUX: Final[bool] = not IS_MAC and not IS_WIN
 
-JSON_FILENAME = 'sh.tat.open_in_mpv.json'
-HOME = os.environ.get('HOME', '')
+JSON_FILENAME: Final[str] = 'sh.tat.open_in_mpv.json'
+HOME: Final[str] = os.environ.get('HOME', '')
 
-USER_CHROME_HOSTS_REG_PATH_WIN = 'HKCU:\\Software\\Google\\Chrome\\NativeMessagingHosts'
+USER_CHROME_HOSTS_REG_PATH_WIN: Final[str] = 'HKCU:\\Software\\Google\\Chrome\\NativeMessagingHosts'
 
-MAC_HOSTS_DIRS = (f'{HOME}/Library/Application Support/Chromium/NativeMessagingHosts',
-                  f'{HOME}/Library/Application Support/Google/Chrome Beta/NativeMessagingHosts',
-                  f'{HOME}/Library/Application Support/Google/Chrome Canary/NativeMessagingHosts',
-                  f'{HOME}/Library/Application Support/Google/Chrome/NativeMessagingHosts')
+MAC_HOSTS_DIRS: Final[tuple[str, ...]] = (
+    f'{HOME}/Library/Application Support/Chromium/NativeMessagingHosts',
+    f'{HOME}/Library/Application Support/Google/Chrome Beta/NativeMessagingHosts',
+    f'{HOME}/Library/Application Support/Google/Chrome Canary/NativeMessagingHosts',
+    f'{HOME}/Library/Application Support/Google/Chrome/NativeMessagingHosts',
+    f'{HOME}/Library/Application Support/Mozilla/NativeMessagingHosts/')
 
-SYSTEM_HOSTS_DIRS = ('/etc/chromium/native-messaging-hosts',
-                     '/etc/opt/chrome/native-messaging-hosts',
-                     '/etc/opt/edge/native-messaging-hosts')
-USER_HOSTS_DIRS = (
+MACPORTS_BIN_PATH: Final[str] = '/opt/local/bin'
+
+SYSTEM_HOSTS_DIRS: Final[tuple[str, str, str]] = ('/etc/chromium/native-messaging-hosts',
+                                                  '/etc/opt/chrome/native-messaging-hosts',
+                                                  '/etc/opt/edge/native-messaging-hosts')
+USER_HOSTS_DIRS: Final[tuple[str, ...]] = (
     f'{xdg.BaseDirectory.xdg_config_home}/BraveSoftware/Brave-Browser/NativeMessagingHosts',
     f'{xdg.BaseDirectory.xdg_config_home}/chromium/NativeMessagingHosts',
     f'{xdg.BaseDirectory.xdg_config_home}/google-chrome-beta/NativeMessagingHosts',
     f'{xdg.BaseDirectory.xdg_config_home}/google-chrome-canary/NativeMessagingHosts',
-    f'{xdg.BaseDirectory.xdg_config_home}/google-chrome/NativeMessagingHosts')
+    f'{xdg.BaseDirectory.xdg_config_home}/google-chrome/NativeMessagingHosts',
+    f'{xdg.BaseDirectory.xdg_config_home}/.mozilla/native-messaging-hosts/')
 
-COMMON_HOST_DATA: dict[str, str | None] = {
+COMMON_HOST_DATA: Final[dict[str, str | None]] = {
     'description': 'Opens a URL in mpv (for use with extension).',
     'path': None,
     'type': 'stdio'
 }
-HOST_DATA: dict[str, str | None | Sequence[str]] = {
+HOST_DATA: Final[dict[str, str | None | Sequence[str]]] = {
     **COMMON_HOST_DATA,
     # cspell:disable-next-line
     'allowed_origins': ['chrome-extension://ggijpepdpiehgbiknmfpfbhcalffjlbj/'],
     'name': 'sh.tat.open_in_mpv',
 }
-HOST_DATA_FIREFOX: dict[str, str | None | Sequence[str]] = {
+HOST_DATA_FIREFOX: Final[dict[str, str | None | Sequence[str]]] = {
     **COMMON_HOST_DATA,
     'allowed_extensions': ['{43e6f3ef-84a0-55f4-b9dd-d879106a24a9}'],
     'name': 'sh.tat.open-in-mpv',

open_in_mpv/open_in_mpv.py

@@ -1,10 +1,9 @@
 # SPDX-License-Identifier: MIT
 from functools import lru_cache
-from os.path import dirname, exists, expanduser, isdir, join as path_join
-from typing import Any, Callable, Mapping, TextIO
+from os.path import dirname, exists, expanduser, expandvars, isdir, join as path_join
+from typing import Any, BinaryIO, Callable, Final, Mapping, TextIO, cast
 import json
 import os
-import platform
 import socket
 import struct
 import subprocess as sp
@@ -15,32 +14,85 @@
 import click
 import xdg.BaseDirectory
 
+from .constants import IS_MAC, IS_WIN, MACPORTS_BIN_PATH
+
+FALLBACKS: Final[dict[str, Any]] = {'log': None, 'socket': None}
+
 
 @lru_cache()
 def get_log_path() -> str:
-    if platform.mac_ver()[0]:
+    if IS_MAC:
         return expanduser('~/Library/Logs')
+    if IS_WIN:
+        return expandvars(r'%LOCALDATA%\open-in-mpv')  # cspell:disable-line
     try:
         return xdg.BaseDirectory.save_state_path('open-in-mpv')
     except KeyError:
-        with tempfile.TemporaryDirectory(prefix='open-in-mpv') as dir_:
-            return dir_
+        FALLBACKS['log'] = tempfile.TemporaryDirectory(prefix='open-in-mpv')  # pylint: disable=R1732
+        return str(FALLBACKS['log'].name)
 
 
 @lru_cache()
 def get_socket_path() -> str:
-    if platform.mac_ver()[0]:
+    if IS_MAC:
         return expanduser('~/Library/Caches/open-in-mpv.sock')
+    if IS_WIN:
+        return expandvars(r'\\.\pipe\open-in-mpv')
     try:
         return path_join(xdg.BaseDirectory.get_runtime_dir(), 'open-in-mpv.sock')
     except KeyError:
-        with tempfile.NamedTemporaryFile(prefix='open-in-mpv', suffix='.sock') as socket_fp:
-            return socket_fp.name
+        FALLBACKS['socket'] = tempfile.NamedTemporaryFile(prefix='open-in-mpv', suffix='.sock')  # pylint: disable=R1732
+        return str(FALLBACKS['socket'].name)
 
 
 LOG_PATH = get_log_path()
 MPV_SOCKET = get_socket_path()
-VERSION = 'v0.0.6'
+VERSION = 'v0.1.7'
+
+
+def environment(data_resp: dict[str, Any], debugging: bool) -> dict[str, Any]:
+    env: dict[str, Any] = os.environ.copy()
+    if isdir(MACPORTS_BIN_PATH):
+        logger.info('Detected MacPorts. Setting PATH.')
+        data_resp['macports'] = True
+        old_path = os.environ.get('PATH')
+        env['PATH'] = MACPORTS_BIN_PATH if not old_path else ':'.join((MACPORTS_BIN_PATH, old_path))
+    if debugging:
+        logger.debug('Environment:')
+        for k, value in env.items():
+            logger.debug(f'  {k}={value}')
+    return env
+
+
+def response(data: dict[str, Any]) -> None:
+    resp = json.dumps(data).encode()
+    size = struct.pack('@i', len(resp))
+    stdout_buffer: BinaryIO = sys.stdout.buffer
+    stdout_buffer.write(size)
+    stdout_buffer.write(resp)
+
+
+def request(buffer: BinaryIO) -> dict[str, Any]:
+    req_len = struct.unpack('@i', buffer.read(4))[0]
+    message = json.loads(buffer.read(req_len).decode())
+    logger.debug('Message contents (%d): %s', req_len, message)
+    return {
+        'init': 'init' in message,
+        'url': message.get('url', None),
+        'debug': message.get('debug', False),
+        'single': message.get('single')
+    }
+
+
+def remove_socket() -> bool:
+    if FALLBACKS['socket']:
+        FALLBACKS['socket'].close()
+        return True
+    try:
+        os.remove(MPV_SOCKET)
+    except OSError:
+        return False
+    return True
 
 
 def spawn(func: Callable[[], Any]) -> None:
@@ -91,7 +143,8 @@ def callback() -> None:
                       env=new_env,
                       stderr=log,
                       stdout=log)
-        os.remove(MPV_SOCKET)
+        if not remove_socket():
+            logger.error('Failed to remove socket file.')
 
     return callback
 
@@ -116,10 +169,8 @@ def callback() -> None:
             sock.send(json.dumps(dict(command=['loadfile', url])).encode(errors='strict') + b'\n')
         except socket.error:
             logger.exception('Connection refused')
-            try:
-                os.remove(MPV_SOCKET)
-            except OSError:
-                pass
+            if not remove_socket():
+                logger.error('Failed to remove socket file')
             spawn_init(url, log, new_env, debug)
 
     return callback
@@ -133,53 +184,31 @@ def real_main(log: TextIO) -> int:
     message. Then the message is expected to be proceed.
     """
     os.makedirs(dirname(MPV_SOCKET), exist_ok=True)
-    stdin_buffer = sys.stdin.buffer
-    req_len = struct.unpack('@i', stdin_buffer.read(4))[0]
-    message = json.loads(stdin_buffer.read(req_len).decode())
-    logger.debug(f'Message contents ({req_len}): {message}')
-    if 'init' in message:
-        resp = json.dumps(dict(version=VERSION, logPath=log.name, socketPath=MPV_SOCKET)).encode()
-        size = struct.pack('@i', len(resp))
-        stdout_buffer = sys.stdout.buffer
-        stdout_buffer.write(size)
-        stdout_buffer.write(resp)
+    message: dict[str, Any] = request(sys.stdin.buffer)
+    if message['init']:
+        response(dict(version=VERSION, logPath=log.name, socketPath=MPV_SOCKET))
         log.close()
         return 0
-    try:
-        url: str = message['url']
-    except KeyError:
+    if (url := message.get('url', None) is None):
         logger.exception('No URL was given')
         print(json.dumps(dict(message='Missing URL!')))
         return 1
     if (is_debug := message.get('debug', False)):
         logger.info('Debug mode enabled.')
     single: bool = message.get('single', True)
     # MacPorts
-    new_env = os.environ.copy()
     data_resp: dict[str, Any] = dict(version=VERSION, log_path=log.name, message='About to spawn')
-    if isdir('/opt/local/bin'):
-        logger.info('Detected MacPorts. Setting PATH.')
-        data_resp['macports'] = True
-        old_path = os.environ.get('PATH')
-        new_env['PATH'] = '/opt/local/bin' if not old_path else ':'.join(
-            ('/opt/local/bin', old_path))
-    data_resp['env'] = new_env
-    if is_debug:
-        logger.debug('Environment:')
-        for k, value in new_env.items():
-            logger.debug(f'  {k}={value}')
+    data_resp['env'] = environment(data_resp, is_debug)
     logger.debug('About to spawn')
-    resp = json.dumps(data_resp).encode()
-    size = struct.pack('@i', len(resp))
-    stdout_buffer = sys.stdout.buffer
-    stdout_buffer.write(size)
-    stdout_buffer.write(resp)
+    response(data_resp)
     if exists(MPV_SOCKET) and single:
-        spawn(get_callback(url, log, new_env, is_debug))
+        spawn(get_callback(cast(str, url), log, data_resp['env'], is_debug))
     else:
-        spawn_init(url, log, new_env, is_debug)
+        spawn_init(cast(str, url), log, data_resp['env'], is_debug)
     logger.debug('mpv should open soon')
     logger.debug('Exiting with status 0')
+    if FALLBACKS['log']:
+        FALLBACKS['log'].cleanup()
     return 0