1import re
2
3import requests
4import tomlkit
5
6from plain.assets.finders import APP_ASSETS_DIR
7
8from .exceptions import (
9 UnknownContentTypeError,
10 VersionMismatchError,
11)
12
13VENDOR_DIR = APP_ASSETS_DIR / "vendor"
14
15
16def iter_next_version(version):
17 if len(version.split(".")) == 2:
18 major, minor = version.split(".")
19 yield f"{int(major) + 1}.0"
20 yield f"{major}.{int(minor) + 1}"
21 elif len(version.split(".")) == 3:
22 major, minor, patch = version.split(".")
23 yield f"{int(major) + 1}.0.0"
24 yield f"{major}.{int(minor) + 1}.0"
25 yield f"{major}.{minor}.{int(patch) + 1}"
26 else:
27 raise ValueError(f"Unable to iterate next version for {version}")
28
29
30class Dependency:
31 def __init__(self, name, **config):
32 self.name = name
33 self.url = config.get("url", "")
34 self.installed = config.get("installed", "")
35 self.filename = config.get("filename", "")
36 self.sourcemap = config.get("sourcemap", "")
37
38 @staticmethod
39 def parse_version_from_url(url):
40 if match := re.search(r"\d+\.\d+\.\d+", url):
41 return match.group(0)
42
43 if match := re.search(r"\d+\.\d+", url):
44 return match.group(0)
45
46 return ""
47
48 def __str__(self):
49 return f"{self.name} -> {self.url}"
50
51 def download(self, version):
52 # If the string contains a {version} placeholder, replace it
53 download_url = self.url.replace("{version}", version)
54
55 response = requests.get(download_url)
56 response.raise_for_status()
57
58 content_type = response.headers.get("content-type")
59 if content_type.lower() not in (
60 "application/javascript; charset=utf-8",
61 "text/javascript; charset=utf-8",
62 "application/json; charset=utf-8",
63 "text/css; charset=utf-8",
64 ):
65 raise UnknownContentTypeError(
66 f"Unknown content type for {self.name}: {content_type}"
67 )
68
69 # Good chance it will redirect to a more final URL (which we hope is versioned)
70 url = response.url
71 version = self.parse_version_from_url(url)
72
73 return version, response
74
75 def install(self):
76 if self.installed:
77 version, response = self.download(self.installed)
78 if version != self.installed:
79 raise VersionMismatchError(
80 f"Version mismatch for {self.name}: {self.installed} != {version}"
81 )
82 return self.vendor(response)
83 else:
84 return self.update()
85
86 def update(self):
87 def try_version(v):
88 try:
89 version, response = self.download(v)
90 return version, response
91 except requests.RequestException:
92 return "", None
93
94 if not self.installed:
95 # If we don't know the installed version yet,
96 # just use the url as given
97 version, response = self.download("")
98 else:
99 version, response = try_version("latest") # A lot of CDNs support this
100 if not version:
101 # Try the next few versions
102 for v in iter_next_version(self.installed):
103 version, response = try_version(v)
104 if version:
105 break
106
107 # TODO ideally this would keep going -- if we move to 2.0, and no 3.0, try 2.1, 2.2, etc.
108
109 if not version:
110 # Use the currently installed version if we found nothing else
111 version, response = self.download(self.installed)
112
113 vendored_path = self.vendor(response)
114 self.installed = version
115
116 if self.installed:
117 # If the exact version was in the string, replace it with {version} placeholder
118 self.url = self.url.replace(self.installed, "{version}")
119
120 self.save_config()
121 return vendored_path
122
123 def save_config(self):
124 with open("pyproject.toml") as f:
125 pyproject = tomlkit.load(f)
126
127 # Force [tool.plain.vendor.dependencies] to be a table
128 dependencies = tomlkit.table()
129 dependencies.update(
130 pyproject.get("tool", {})
131 .get("plain", {})
132 .get("vendor", {})
133 .get("dependencies", {})
134 )
135
136 # Force [tool.plain.vendor.dependencies.{name}] to be an inline table
137 # name = { url = "https://example.com", installed = "1.0.0" }
138 dependencies[self.name] = tomlkit.inline_table()
139 dependencies[self.name]["url"] = self.url
140 dependencies[self.name]["installed"] = self.installed
141 if self.filename:
142 dependencies[self.name]["filename"] = self.filename
143 if self.sourcemap:
144 dependencies[self.name]["sourcemap"] = self.sourcemap
145
146 # Have to give it the right structure in case they don't exist
147 if "tool" not in pyproject:
148 pyproject["tool"] = tomlkit.table()
149 if "plain" not in pyproject["tool"]:
150 pyproject["tool"]["plain"] = tomlkit.table()
151 if "vendor" not in pyproject["tool"]["plain"]:
152 pyproject["tool"]["plain"]["vendor"] = tomlkit.table()
153
154 pyproject["tool"]["plain"]["vendor"]["dependencies"] = dependencies
155
156 with open("pyproject.toml", "w") as f:
157 f.write(tomlkit.dumps(pyproject))
158
159 def vendor(self, response):
160 if not VENDOR_DIR.exists():
161 VENDOR_DIR.mkdir(parents=True)
162
163 if self.filename:
164 # Use a specific filename from config
165 filename = self.filename
166 else:
167 # Otherwise, use the filename from the URL
168 filename = response.url.split("/")[-1]
169
170 vendored_path = VENDOR_DIR / filename
171
172 with open(vendored_path, "wb") as f:
173 f.write(response.content)
174
175 # If a sourcemap is requested, download it as well
176 if self.sourcemap:
177 if isinstance(self.sourcemap, str):
178 # Use a specific filename from config
179 sourcemap_filename = self.sourcemap
180 else:
181 # Otherwise, append .map to the URL
182 sourcemap_filename = f"{filename}.map"
183
184 sourcemap_url = "/".join(
185 response.url.split("/")[:-1] + [sourcemap_filename]
186 )
187 sourcemap_response = requests.get(sourcemap_url)
188 sourcemap_response.raise_for_status()
189
190 sourcemap_path = VENDOR_DIR / sourcemap_filename
191
192 with open(sourcemap_path, "wb") as f:
193 f.write(sourcemap_response.content)
194
195 return vendored_path
196
197
198def get_deps():
199 with open("pyproject.toml") as f:
200 pyproject = tomlkit.load(f)
201
202 config = (
203 pyproject.get("tool", {})
204 .get("plain", {})
205 .get("vendor", {})
206 .get("dependencies", {})
207 )
208
209 deps = []
210
211 for name, data in config.items():
212 deps.append(Dependency(name, **data))
213
214 return deps