Socket端口绑定替代文件锁 v3.3

diff --git a/wechat_auto_login.py b/wechat_auto_login.py
index 92c00a513f73849ebb34c0d80ae104d7410bd0bd..8bd13a961abf9906690616590cf2c16dbc27484b 100644 (file)
--- a/wechat_auto_login.py
+++ b/wechat_auto_login.py
@@ -18,7 +18,7 @@ try:
     import random
     import logging
     import configparser
-    import atexit
+    import socket
     import tempfile
     import base64
 
@@ -56,36 +56,35 @@ try:
         log.critical("导入第三方库失败: %s", e)
         sys.exit(1)
 
-    # ── single instance lock ─────────────────────────────────────────────────
-    _lock_file = os.path.join(tempfile.gettempdir(), "wechat_auto_login.lock")
-    _lock_fd = None
+    class SingleInstance:
+        """基于Socket端口绑定的单实例保护。进程死OS自动释放端口，零残留。"""
+        def __init__(self, app_name="wechat_auto_login", port=52001):
+            self._sock = None
+            self._port = port
+            self.app_name = app_name
 
-
-    def _acquire_lock():
-        global _lock_fd
-        try:
-            _lock_fd = os.open(_lock_file, os.O_CREAT | os.O_EXCL | os.O_RDWR)
-            return True
-        except FileExistsError:
-            return False
-
-
-    def _release_lock():
-        global _lock_fd
-        if _lock_fd is not None:
-            os.close(_lock_fd)
+        def try_lock(self):
             try:
-                os.unlink(_lock_file)
+                self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+                self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                self._sock.bind(("127.0.0.1", self._port))
+                self._sock.listen(1)
+                return True
             except OSError:
-                pass
+                return False
+
+        def release(self):
+            if self._sock:
+                try:
+                    self._sock.close()
+                except OSError:
+                    pass
+                self._sock = None
 
 
     # ── 模块初始化（用 try 捕获所有意外错误） ───────────────────────────────
     try:
-        if not _acquire_lock():
-            log.error("已有实例在运行，退出")
-            sys.exit(1)
-        atexit.register(_release_lock)
+
 
         # ── config ─────────────────────────────────────────────────────────────────
         config = configparser.ConfigParser()
@@ -850,8 +849,15 @@ try:
 
 
     def main():
+        si = SingleInstance()
+        if not si.try_lock():
+            log.error("已有实例在运行")
+            sys.exit(1)
         app = MonitorApp()
-        app.root.mainloop()
+        try:
+            app.root.mainloop()
+        finally:
+            si.release()
 
 
     if __name__ == "__main__":