Coverage for graphica/gui/single_instance.py: 74.4%

1"""起動中の Graphica にファイルを渡す(ファイルをダブルクリックしたとき、新しいウィンドウではなく新しいタブで開くため)。 

2 

3最初に起動したものがローカルサーバーを開き、あとから起動したものはそこへパスを送って終わる。起動画面より前に使うので、 

4ここでは PySide6 だけを読む。 

5""" 

6import getpass 

7import json 

8import logging 

9import os 

10import re 

11 

12from PySide6.QtCore import QEvent, QObject, Signal 

13from PySide6.QtNetwork import QLocalServer, QLocalSocket 

14 

15logger = logging.getLogger(__name__) 

16 

17# 起動中のアプリがほかの受け渡しを処理している間は待たされることがある 

18_CONNECT_TIMEOUT_MS = 2000 

19_WRITE_TIMEOUT_MS = 1000 

20 

21 

22def server_name(): 

23 """利用者ごとに別の名前(同じ PC の別の利用者のアプリには渡さない)。""" 

24 try: 

25 user = getpass.getuser() 

26 except (OSError, KeyError, ImportError): # 利用者名が取れない環境でも起動は止めない 

27 user = "user" 

28 return "Graphica-" + re.sub(r"[^A-Za-z0-9_.-]", "_", user) 

29 

30 

31def files_to_open(argv): 

32 """起動の引数のうち、開くファイル(オプション以外で、実在するもの)。""" 

33 return [os.path.abspath(arg) for arg in argv[1:] if not arg.startswith("-") and os.path.isfile(arg)] 

34 

35 

36def instance_running(name=None): 

37 """ほかの Graphica が受け渡しの窓口を開いているか。""" 

38 socket = QLocalSocket() 

39 socket.connectToServer(name or server_name()) 

40 running = socket.waitForConnected(_CONNECT_TIMEOUT_MS) 

41 if running: 

42 socket.disconnectFromServer() 

43 return running 

44 

45 

46def send_to_running_instance(paths, name=None): 

47 """起動中のアプリがあればパスを渡して True。無ければ False(自分が起動を続ける)。""" 

48 socket = QLocalSocket() 

49 socket.connectToServer(name or server_name()) 

50 if not socket.waitForConnected(_CONNECT_TIMEOUT_MS): 50 ↛ 52line 50 didn't jump to line 52 because the condition on line 50 was always true

51 return False 

52 socket.write((json.dumps({"open": list(paths)}) + "\n").encode("utf-8")) 

53 socket.flush() 

54 # Windows のパイプは書き込みが後から終わる。切断が済むまで待たないと、すぐ終わるこのプロセスと一緒に消える 

55 socket.disconnectFromServer() 

56 if socket.state() != QLocalSocket.LocalSocketState.UnconnectedState: 

57 socket.waitForDisconnected(_WRITE_TIMEOUT_MS) 

58 return socket.state() == QLocalSocket.LocalSocketState.UnconnectedState and socket.bytesToWrite() == 0 

59 

60 

61class InstanceServer(QObject): 

62 """あとから起動したアプリが送ってきたパスを open_requested で知らせる。""" 

63 

64 open_requested = Signal(list) 

65 

66 def __init__(self, name=None, parent=None): 

67 super().__init__(parent) 

68 self._name = name or server_name() 

69 self._server = QLocalServer(self) 

70 self._server.newConnection.connect(self._on_new_connection) 

71 self._buffers = {} 

72 

73 def listen(self): 

74 """開けなければ False。前回の異常終了で残った名前(Unix のソケットファイル)は消して開き直す。""" 

75 if self._server.listen(self._name): 75 ↛ 77line 75 didn't jump to line 77 because the condition on line 75 was always true

76 return True 

77 QLocalServer.removeServer(self._name) 

78 if self._server.listen(self._name): 

79 return True 

80 logger.warning("起動中のアプリへのファイルの受け渡しを開けませんでした: %s", self._server.errorString()) 

81 return False 

82 

83 def close(self): 

84 self._server.close() 

85 

86 def _on_new_connection(self): 

87 # 接続を掴むラムダは使わない(接続が消えたあとに呼ばれると落ちる)。送り手は sender() で引く 

88 while self._server.hasPendingConnections(): 

89 socket = self._server.nextPendingConnection() 

90 self._buffers[socket] = b"" 

91 socket.readyRead.connect(self._on_ready_read) 

92 socket.disconnected.connect(self._on_disconnected) 

93 

94 def _on_ready_read(self): 

95 socket = self.sender() 

96 if socket is None: 96 ↛ 97line 96 didn't jump to line 97 because the condition on line 96 was never true

97 return 

98 self._buffers[socket] = self._buffers.get(socket, b"") + bytes(socket.readAll()) 

99 self._emit_complete_lines(socket) 

100 

101 def _on_disconnected(self): 

102 socket = self.sender() 

103 if socket is None: 103 ↛ 104line 103 didn't jump to line 104 because the condition on line 103 was never true

104 return 

105 self._buffers[socket] = self._buffers.get(socket, b"") + bytes(socket.readAll()) 

106 self._emit_complete_lines(socket, final=True) 

107 self._buffers.pop(socket, None) 

108 socket.deleteLater() 

109 

110 def _emit_complete_lines(self, socket, final=False): 

111 data = self._buffers.get(socket, b"") 

112 *lines, rest = data.split(b"\n") 

113 if final and rest: 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true

114 lines.append(rest) 

115 rest = b"" 

116 self._buffers[socket] = rest 

117 for line in lines: 

118 if not line.strip(): 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true

119 continue 

120 try: 

121 paths = json.loads(line.decode("utf-8")).get("open", []) 

122 except (ValueError, AttributeError): 

123 logger.warning("起動中のアプリへの受け渡しの中身が読めませんでした: %r", line[:200]) 

124 continue 

125 paths = [p for p in paths if isinstance(p, str)] 

126 if paths: 126 ↛ 117line 126 didn't jump to line 117 because the condition on line 126 was always true

127 self.open_requested.emit(paths) 

128 

129 

130class FileOpenEventFilter(QObject): 

131 """macOS で Finder から開いたファイル(QFileOpenEvent。起動中でも届く)を open_requested で知らせる。""" 

132 

133 open_requested = Signal(list) 

134 

135 def eventFilter(self, watched, event): 

136 if event.type() == QEvent.Type.FileOpen: 

137 path = event.file() 

138 if path: 138 ↛ 140line 138 didn't jump to line 140 because the condition on line 138 was always true

139 self.open_requested.emit([path]) 

140 return True 

141 return super().eventFilter(watched, event)