Reimplement duplicated run detection as a file lock

.gitignore

@@ -13,5 +13,6 @@ __pycache__
 /cache
 /*.jar
 log.txt
+/lock.file
 settings.json
 /lang/English.json

.vscode/launch.json

@@ -16,7 +16,7 @@
             "type": "python",
             "request": "launch",
             "program": "main.py",
-            "args": ["--no-run-check", "-vv"],
+            "args": ["-vv"],
             "console": "integratedTerminal",
             "justMyCode": false
         },
@@ -25,7 +25,7 @@
             "type": "python",
             "request": "launch",
             "program": "main.py",
-            "args": ["--no-run-check", "-vv", "--tray"],
+            "args": ["-vv", "--tray"],
             "console": "integratedTerminal",
             "justMyCode": false
         },
@@ -34,7 +34,7 @@
             "type": "python",
             "request": "launch",
             "program": "main.py",
-            "args": ["--no-run-check", "-vvv"],
+            "args": ["-vvv"],
             "console": "integratedTerminal",
             "justMyCode": false
         },
@@ -43,7 +43,7 @@
             "type": "python",
             "request": "launch",
             "program": "main.py",
-            "args": ["--no-run-check", "-vvv", "--debug-ws"],
+            "args": ["-vvv", "--debug-ws"],
             "console": "integratedTerminal",
             "justMyCode": false
         },

constants.py

@@ -64,6 +64,7 @@ LANG_PATH = _resource_path("lang")
 # Other Paths
 LOG_PATH = Path(WORKING_DIR, "log.txt")
 CACHE_PATH = Path(WORKING_DIR, "cache")
+LOCK_PATH = Path(WORKING_DIR, "lock.file")
 CACHE_DB = Path(CACHE_PATH, "mapping.json")
 COOKIES_PATH = Path(WORKING_DIR, "cookies.jar")
 SETTINGS_PATH = Path(WORKING_DIR, "settings.json")

main.py

@@ -18,9 +18,6 @@ if __name__ == "__main__":
     from tkinter import messagebox
     from typing import IO, NoReturn
 
-    if sys.platform == "win32":
-        import win32gui
-
     if sys.platform == "linux" and sys.version_info >= (3, 10):
         import truststore
         truststore.inject_into_ssl()
@@ -30,8 +27,8 @@ if __name__ == "__main__":
     from settings import Settings
     from version import __version__
     from exceptions import CaptchaRequired
-    from utils import resource_path, set_root_icon
-    from constants import CALL, SELF_PATH, FILE_FORMATTER, LOG_PATH, WINDOW_TITLE
+    from utils import lock_file, resource_path, set_root_icon
+    from constants import CALL, SELF_PATH, FILE_FORMATTER, LOG_PATH, LOCK_PATH
 
     warnings.simplefilter("default", ResourceWarning)
 
@@ -107,9 +104,6 @@ if __name__ == "__main__":
     parser.add_argument("--tray", action="store_true")
     parser.add_argument("--log", action="store_true")
     # undocumented debug args
-    parser.add_argument(
-        "--no-run-check", dest="no_run_check", action="store_true", help=argparse.SUPPRESS
-    )
     parser.add_argument(
         "--debug-ws", dest="_debug_ws", action="store_true", help=argparse.SUPPRESS
     )
@@ -131,17 +125,8 @@ if __name__ == "__main__":
     # get rid of unneeded objects
     del root, parser
 
-    # check if we're not already running
-    if sys.platform == "win32":
-        try:
-            exists = win32gui.FindWindow(None, WINDOW_TITLE)
-        except AttributeError:
-            # we're not on Windows - continue
-            exists = False
-        if exists and not settings.no_run_check:
-            # already running - exit
-            sys.exit(3)
-
+    # client run
+    async def main():
         # set language
         try:
             _.set_language(settings.language)
@@ -163,8 +148,6 @@ if __name__ == "__main__":
         logging.getLogger("TwitchDrops.gql").setLevel(settings.debug_gql)
         logging.getLogger("TwitchDrops.websocket").setLevel(settings.debug_ws)
 
-    # client run
-    async def main():
         exit_status = 0
         client = Twitch(settings)
         loop = asyncio.get_running_loop()
@@ -200,4 +183,13 @@ if __name__ == "__main__":
         client.gui.close_window()
         sys.exit(exit_status)
 
+    try:
+        # use lock_file to check if we're not already running
+        success, file = lock_file(LOCK_PATH)
+        if not success:
+            # already running - exit
+            sys.exit(3)
+
         asyncio.run(main())
+    finally:
+        file.close()

utils.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import io
 import os
 import sys
 import json
@@ -77,6 +78,29 @@ def format_traceback(exc: BaseException, **kwargs: Any) -> str:
     return ''.join(traceback.format_exception(type(exc), exc, **kwargs))
 
 
+def lock_file(path: Path) -> tuple[bool, io.TextIOWrapper]:
+    file = path.open('w', encoding="utf8")
+    file.write('ツ')
+    file.flush()
+    if sys.platform == "win32":
+        import msvcrt
+        try:
+            # we need to lock at least one byte for this to work
+            msvcrt.locking(file.fileno(), msvcrt.LK_NBLCK, max(path.stat().st_size, 1))
+        except Exception:
+            return False, file
+        return True, file
+    if sys.platform == "linux":
+        import fcntl
+        try:
+            fcntl.lockf(file, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except Exception:
+            return False, file
+        return True, file
+    # for unsupported systems, just always return True
+    return True, file
+
+
 def json_minify(data: JsonType | list[JsonType]) -> str:
     """
     Returns minified JSON for payload usage.