From: 小龙 <xl@szjxzxh.cn>
Date: Mon, 20 Jul 2026 03:45:56 +0000 (+0800)
Subject: Socket端口绑定替代文件锁 v3.3
X-Git-Url: https://szjxzxh.cn/gitweb/lucky-day.html?a=commitdiff_plain;h=93716d484d3b6761c3a66ea54bd083d0d8bf16aa;p=%E5%BE%AE%E4%BF%A1%E7%9B%91%E6%8E%A7

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

diff --git a/wechat_auto_login.py b/wechat_auto_login.py
index 92c00a5..8bd13a9 100644
--- 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__":