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