Coverage for graphica/core/plugin_updates.py: 88.1%

1"""入れたプラグインに新しい版が出ていないかを調べる(公開プラグインの一覧と GitHub Releases への匿名の GET だけ)。 

2 

3一覧は本体リポジトリの docs/plugin_index.json(プラグインのフォルダ名・名前 → GitHub のリポジトリ)。 

4入れ替えは利用者が選んだものだけで、ここでは調べるのと、落とした zip を確かめるところまで。 

5""" 

6import json 

7import os 

8import urllib.request 

9import zipfile 

10from typing import Any, Callable, Iterable 

11 

12from graphica.core.plugin_manifest import ( 

13 PLUGIN_MANIFEST_FILENAME, PluginManifestError, is_compatible_api_version, load_plugin_manifest, 

14) 

15from graphica.core.update_check import GITHUB_REPO, _asset_info, is_newer_version 

16 

17PLUGIN_INDEX_URL = f"https://raw.githubusercontent.com/{GITHUB_REPO}/master/Graphica_project/docs/plugin_index.json" 

18_DEFAULT_TIMEOUT_SEC = 5 

19_HEADERS = {'Accept': 'application/vnd.github+json', 'User-Agent': 'Graphica-plugin-update-check'} 

20 

21 

22def _get_json(url: str, timeout: float) -> Any: 

23 request = urllib.request.Request(url, headers=_HEADERS) 

24 with urllib.request.urlopen(request, timeout=timeout) as response: 

25 return json.loads(response.read().decode('utf-8')) 

26 

27 

28def parse_plugin_index(data: Any) -> list[dict[str, str]]: 

29 """一覧の JSON から、folder と repo のそろった項目だけを返す(壊れた行は飛ばす)。 

30 

31 Args: 

32 data: {"plugins": [{"folder", "name", "repo"}, ...]}。 

33 

34 Returns: 

35 {"folder", "name", "repo"} のリスト。 

36 """ 

37 entries = data.get('plugins') if isinstance(data, dict) else None 

38 result = [] 

39 for entry in entries or []: 

40 if not isinstance(entry, dict): 

41 continue 

42 folder, repo = entry.get('folder'), entry.get('repo') 

43 if isinstance(folder, str) and folder and isinstance(repo, str) and repo.count('/') == 1: 

44 result.append({'folder': folder, 'name': str(entry.get('name') or folder), 'repo': repo}) 

45 return result 

46 

47 

48def fetch_plugin_index(timeout: float = _DEFAULT_TIMEOUT_SEC) -> list[dict[str, str]]: 

49 """公開プラグインの一覧。通信の失敗は例外のまま。""" 

50 return parse_plugin_index(_get_json(PLUGIN_INDEX_URL, timeout)) 

51 

52 

53def fetch_latest_plugin_release(repo: str, timeout: float = _DEFAULT_TIMEOUT_SEC) -> dict[str, Any]: 

54 """repo の最新リリース: {"version", "html_url", "body", "zip"(添付の zip の情報か None)}。""" 

55 data = _get_json(f"https://api.github.com/repos/{repo}/releases/latest", timeout) 

56 zips = [_asset_info(asset) for asset in data.get('assets') or [] 

57 if isinstance(asset, dict) and str(asset.get('name', '')).lower().endswith('.zip')] 

58 return { 

59 'version': str(data.get('tag_name') or '').lstrip('vV'), 

60 'html_url': data.get('html_url') or f"https://github.com/{repo}/releases", 

61 'body': data.get('body') or '', 

62 'zip': zips[0] if zips else None, 

63 } 

64 

65 

66def installed_user_plugins(plugins_dir: str) -> list[dict[str, str]]: 

67 """plugins_dir の直下のプラグイン({"folder", "name", "version"})。plugin.json が読めないものは飛ばす。""" 

68 result: list[dict[str, str]] = [] 

69 if not os.path.isdir(plugins_dir): 69 ↛ 70line 69 didn't jump to line 70 because the condition on line 69 was never true

70 return result 

71 for folder in sorted(os.listdir(plugins_dir)): 

72 path = os.path.join(plugins_dir, folder) 

73 if folder.startswith('.') or not os.path.isdir(path): 

74 continue 

75 manifest: dict[str, Any] | None 

76 try: 

77 manifest = load_plugin_manifest(path) 

78 except PluginManifestError: 

79 # API の版が合わなくなった古いものも、新しい版で直るかもしれないので一覧には入れる 

80 manifest = _read_manifest_loosely(path) 

81 if manifest is None: 

82 continue 

83 result.append({'folder': folder, 'name': str(manifest.get('name') or folder), 

84 'version': str(manifest.get('version') or '0')}) 

85 return result 

86 

87 

88def _read_manifest_loosely(path: str) -> dict[str, Any] | None: 

89 try: 

90 with open(os.path.join(path, PLUGIN_MANIFEST_FILENAME), encoding='utf-8') as f: 

91 manifest = json.load(f) 

92 except (OSError, ValueError): 

93 return None 

94 return manifest if isinstance(manifest, dict) else None 

95 

96 

97def find_plugin_updates(installed: Iterable[dict[str, str]], index: list[dict[str, str]], 

98 fetch_release: Callable[[str], dict[str, Any]]) -> list[dict[str, Any]]: 

99 """入れたプラグインのうち、一覧にあって新しい版の zip が出ているもの。 

100 

101 Args: 

102 installed: installed_user_plugins() の結果。 

103 index: fetch_plugin_index() の結果。フォルダ名で、無ければ名前で突き合わせる。 

104 fetch_release: リポジトリ名 → fetch_latest_plugin_release() の結果。1件の失敗は飛ばす。 

105 

106 Returns: 

107 {"folder", "name", "current", "latest", "repo", "html_url", "body", "zip"} のリスト。 

108 """ 

109 by_folder = {entry['folder']: entry for entry in index} 

110 by_name = {entry['name']: entry for entry in index} 

111 updates = [] 

112 for plugin in installed: 

113 entry = by_folder.get(plugin['folder']) or by_name.get(plugin['name']) 

114 if entry is None: 

115 continue 

116 try: 

117 release = fetch_release(entry['repo']) 

118 except (OSError, ValueError, KeyError, TypeError, AttributeError): 

119 # 1つのリポジトリの失敗(消された・通信・想定外の応答)でほかを止めない 

120 continue 

121 if release.get('zip') is None or not is_newer_version(release['version'], plugin['version']): 

122 continue 

123 updates.append({ 

124 'folder': plugin['folder'], 'name': plugin['name'], 'current': plugin['version'], 

125 'latest': release['version'], 'repo': entry['repo'], 'html_url': release['html_url'], 

126 'body': release['body'], 'zip': release['zip'], 

127 }) 

128 return updates 

129 

130 

131def check_plugin_updates(plugins_dir: str, timeout: float = _DEFAULT_TIMEOUT_SEC, 

132 **_ignored: Any) -> list[dict[str, Any]]: 

133 """TaskRunner から呼ぶ入口。入れたものが無ければ通信しない。""" 

134 installed = installed_user_plugins(plugins_dir) 

135 if not installed: 135 ↛ 137line 135 didn't jump to line 137 because the condition on line 135 was always true

136 return [] 

137 index = fetch_plugin_index(timeout) 

138 return find_plugin_updates(installed, index, lambda repo: fetch_latest_plugin_release(repo, timeout)) 

139 

140 

141def zip_manifest(zip_path: str) -> dict[str, Any]: 

142 """zip の中のプラグインの plugin.json(トップのフォルダ直下)。見つからなければ PluginManifestError。""" 

143 try: 

144 with zipfile.ZipFile(zip_path) as zf: 

145 names = [n for n in zf.namelist() 

146 if n.endswith('/' + PLUGIN_MANIFEST_FILENAME) and n.count('/') == 1] 

147 if not names: 

148 raise PluginManifestError(f"zip に {PLUGIN_MANIFEST_FILENAME} が見つかりません。") 

149 manifest = json.loads(zf.read(names[0]).decode('utf-8')) 

150 except (zipfile.BadZipFile, ValueError, KeyError) as e: 

151 raise PluginManifestError(f"zip の {PLUGIN_MANIFEST_FILENAME} を読めません: {e}") from e 

152 if not isinstance(manifest, dict): 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true

153 raise PluginManifestError(f"{PLUGIN_MANIFEST_FILENAME} はJSONオブジェクトである必要があります。") 

154 return manifest 

155 

156 

157def check_update_zip(zip_path: str, folder: str) -> dict[str, Any]: 

158 """入れ替える前の確認: 同じフォルダ名のプラグインで、この Graphica の API で読めること。""" 

159 with zipfile.ZipFile(zip_path) as zf: 

160 tops = {name.split('/', 1)[0] for name in zf.namelist() if name.strip('/')} 

161 if tops != {folder}: 

162 raise PluginManifestError(f"zip の中身が「{folder}」のプラグインではありません。") 

163 manifest = zip_manifest(zip_path) 

164 if not is_compatible_api_version(manifest.get('api_version')): 

165 raise PluginManifestError( 

166 f"新しい版はこの Graphica では使えません(プラグイン API {manifest.get('api_version')})。" 

167 "Graphica を更新してから入れてください。") 

168 return manifest