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