1from __future__ import annotations
2
3import os
4import platform
5import subprocess
6import sys
7from pathlib import Path
8from typing import Any
9
10import click
11import requests
12import tomlkit
13
14from plain.internal import internalcode
15from plain.packages import packages_registry
16from plain.runtime import APP_PATH, PLAIN_TEMP_PATH, settings
17
18
19@internalcode
20class Tailwind:
21 @property
22 def target_directory(self) -> str:
23 return str(PLAIN_TEMP_PATH)
24
25 @property
26 def standalone_path(self) -> str:
27 return os.path.join(self.target_directory, "tailwind")
28
29 @property
30 def version_lockfile_path(self) -> str:
31 return os.path.join(self.target_directory, "tailwind.version")
32
33 @property
34 def src_css_path(self) -> Path:
35 return settings.TAILWIND_SRC_PATH
36
37 @property
38 def dist_css_path(self) -> Path:
39 return settings.TAILWIND_DIST_PATH
40
41 def update_plain_sources(self) -> None:
42 paths = set()
43
44 # Add paths from installed packages
45 for package_config in packages_registry.get_package_configs():
46 abs_package_path = os.path.abspath(package_config.path)
47 abs_app_path = os.path.abspath(APP_PATH)
48 if os.path.commonpath([abs_app_path, abs_package_path]) != abs_app_path:
49 paths.add(os.path.relpath(abs_package_path, self.target_directory))
50
51 # Sort the paths so that the order is consistent
52 paths = sorted(paths)
53
54 plain_sources_path = os.path.join(self.target_directory, "tailwind.css")
55 with open(plain_sources_path, "w") as f:
56 for path in paths:
57 f.write(f'@source "{path}";\n')
58
59 def invoke(self, *args: Any, cwd: str | None = None) -> None:
60 result = subprocess.run([self.standalone_path] + list(args), cwd=cwd)
61 if result.returncode != 0:
62 sys.exit(result.returncode)
63
64 def is_installed(self) -> bool:
65 if not os.path.exists(self.target_directory):
66 os.mkdir(self.target_directory)
67 return os.path.exists(os.path.join(self.target_directory, "tailwind"))
68
69 def create_src_css(self) -> None:
70 os.makedirs(os.path.dirname(self.src_css_path), exist_ok=True)
71 with open(self.src_css_path, "w") as f:
72 f.write("""@import "tailwindcss";\n@import "./.plain/tailwind.css";\n""")
73
74 def needs_update(self) -> bool:
75 locked_version = self.get_installed_version()
76 if not locked_version:
77 return True
78
79 if locked_version != self.get_version_from_config():
80 return True
81
82 return False
83
84 def get_installed_version(self) -> str:
85 """Get the currently installed Tailwind version"""
86 if not os.path.exists(self.version_lockfile_path):
87 return ""
88
89 with open(self.version_lockfile_path) as f:
90 return f.read().strip()
91
92 def get_version_from_config(self) -> str:
93 pyproject_path = os.path.join(
94 os.path.dirname(self.target_directory), "pyproject.toml"
95 )
96
97 if not os.path.exists(pyproject_path):
98 return ""
99
100 with open(pyproject_path) as f:
101 config = tomlkit.load(f)
102 return (
103 config.get("tool", {})
104 .get("plain", {})
105 .get("tailwind", {})
106 .get("version", "")
107 )
108
109 def set_version_in_config(self, version: str) -> None:
110 pyproject_path = os.path.join(
111 os.path.dirname(self.target_directory), "pyproject.toml"
112 )
113
114 with open(pyproject_path) as f:
115 config = tomlkit.load(f)
116
117 config.setdefault("tool", {}).setdefault("plain", {}).setdefault(
118 "tailwind", {}
119 )["version"] = version
120
121 with open(pyproject_path, "w") as f:
122 tomlkit.dump(config, f)
123
124 def download(self, version: str = "") -> str:
125 if version:
126 if not version.startswith("v"):
127 version = f"v{version}"
128 url = f"https://github.com/tailwindlabs/tailwindcss/releases/download/{version}/tailwindcss-{self.detect_platform_slug()}"
129 else:
130 url = f"https://github.com/tailwindlabs/tailwindcss/releases/latest/download/tailwindcss-{self.detect_platform_slug()}"
131
132 # Optimized requests session with better connection pooling and headers
133 session = requests.Session()
134
135 # Better connection pooling
136 adapter = requests.adapters.HTTPAdapter( # type: ignore[attr-defined]
137 pool_connections=1, pool_maxsize=10, max_retries=3, pool_block=True
138 )
139 session.mount("https://", adapter)
140 session.mount("http://", adapter)
141
142 # Optimized headers for better performance
143 headers = {
144 "Accept-Encoding": "gzip, deflate, br",
145 "Connection": "keep-alive",
146 "User-Agent": "plain-tailwind/1.0",
147 }
148
149 with session.get(url, stream=True, headers=headers, timeout=300) as response:
150 response.raise_for_status()
151 total = int(response.headers.get("Content-Length", 0))
152
153 with open(self.standalone_path, "wb") as f:
154 with click.progressbar(
155 length=total,
156 label="Downloading Tailwind",
157 width=0,
158 ) as bar:
159 # Use 8MB chunks for maximum performance
160 for chunk in response.iter_content(
161 chunk_size=1024 * 1024, decode_unicode=False
162 ):
163 if chunk:
164 f.write(chunk)
165 bar.update(len(chunk))
166
167 os.chmod(self.standalone_path, 0o755)
168
169 if not version:
170 # Get the version from the redirect chain (latest -> vX.Y.Z)
171 version = response.history[1].url.split("/")[-2]
172
173 version = version.lstrip("v")
174
175 with open(self.version_lockfile_path, "w") as f:
176 f.write(version)
177
178 return version
179
180 def install(self, version: str = "") -> str:
181 installed_version = self.download(version)
182 self.set_version_in_config(installed_version)
183 return installed_version
184
185 @staticmethod
186 def detect_platform_slug() -> str:
187 uname = platform.uname()[0]
188
189 if uname == "Windows":
190 return "windows-x64.exe"
191
192 if uname == "Linux" and platform.uname()[4] == "aarch64":
193 return "linux-arm64"
194
195 if uname == "Linux":
196 return "linux-x64"
197
198 if uname == "Darwin" and platform.uname().machine == "arm64":
199 return "macos-arm64"
200
201 if uname == "Darwin":
202 return "macos-x64"
203
204 raise Exception("Unsupported platform for Tailwind standalone")