Coverage for fastblocks / adapters / images / twicpics.py: 26%

1"""TwicPics adapter for FastBlocks with real-time image optimization.""" 

2 

3import asyncio 

4from contextlib import suppress 

5from typing import Any 

6from urllib.parse import quote 

7from uuid import UUID 

8 

9import httpx 

10from acb.depends import depends 

11from pydantic import SecretStr 

12 

13from ._base import ImagesBase, ImagesBaseSettings 

14 

15 

16class TwicPicsImagesSettings(ImagesBaseSettings): 

17 """Settings for TwicPics adapter.""" 

18 

19 # Required ACB 0.19.0+ metadata 

20 MODULE_ID: UUID = UUID("01937d86-6f4c-8e5d-9a7b-c6d7e8f9a0b1") # Static UUID7 

21 MODULE_STATUS: str = "stable" 

22 

23 # TwicPics configuration 

24 domain: str = "" # Your TwicPics domain (e.g., "demo.twic.pics") 

25 path_prefix: str = "" # Optional path prefix 

26 api_key: SecretStr = SecretStr("") # For upload operations 

27 

28 # Image optimization defaults 

29 default_quality: int = 85 

30 default_format: str = "auto" # auto, webp, avif, jpeg, png 

31 enable_placeholder: bool = True 

32 placeholder_quality: int = 10 

33 

34 # Performance settings 

35 enable_lazy_loading: bool = True 

36 enable_progressive: bool = True 

37 timeout: int = 30 

38 

39 

40class TwicPicsImages(ImagesBase): 

41 """TwicPics adapter with real-time image optimization.""" 

42 

43 # Required ACB 0.19.0+ metadata 

44 MODULE_ID: UUID = UUID("01937d86-6f4c-8e5d-9a7b-c6d7e8f9a0b1") # Static UUID7 

45 MODULE_STATUS: str = "stable" 

46 

47 def __init__(self) -> None: 

48 """Initialize TwicPics adapter.""" 

49 super().__init__() 

50 self.settings: TwicPicsImagesSettings | None = None 

51 self._client: httpx.AsyncClient | None = None 

52 

53 # Register with ACB dependency system 

54 with suppress(Exception): 

55 depends.set(self) 

56 

57 async def _get_client(self) -> httpx.AsyncClient: 

58 """Get or create HTTP client.""" 

59 if not self._client: 

60 if not self.settings: 

61 self.settings = TwicPicsImagesSettings() 

62 

63 headers = {} 

64 if self.settings.api_key.get_secret_value(): 

65 headers["Authorization"] = ( 

66 f"Bearer {self.settings.api_key.get_secret_value()}" 

67 ) 

68 

69 self._client = httpx.AsyncClient( 

70 timeout=self.settings.timeout, headers=headers 

71 ) 

72 return self._client 

73 

74 async def upload_image(self, file_data: bytes, filename: str) -> str: 

75 """Upload image to TwicPics (or return reference path).""" 

76 # Note: TwicPics typically works with existing images via URL references 

77 # For upload scenarios, you'd typically upload to your own storage 

78 # and then reference through TwicPics. This is a simplified implementation. 

79 

80 if not self.settings: 

81 self.settings = TwicPicsImagesSettings() 

82 

83 # For demo purposes, we'll create a reference based on filename 

84 # In real implementation, you'd upload to your storage backend 

85 # and return the path that TwicPics can access 

86 

87 # Clean filename for URL 

88 clean_filename = quote(filename, safe=".-_") 

89 

90 # Return the path that will be used with TwicPics 

91 if self.settings.path_prefix: 

92 return f"{self.settings.path_prefix}/{clean_filename}" 

93 return clean_filename 

94 

95 def _build_transform_parts(self, transformations: dict[str, Any]) -> list[str]: 

96 """Build transformation parameter list for TwicPics.""" 

97 transform_parts = [] 

98 

99 # Resize transformations 

100 if "width" in transformations: 

101 transform_parts.append(f"width={transformations['width']}") 

102 if "height" in transformations: 

103 transform_parts.append(f"height={transformations['height']}") 

104 

105 # Fit modes 

106 if "fit" in transformations: 

107 fit_mode = transformations["fit"] 

108 resize_map = {"crop": "fill", "contain": "contain", "cover": "cover"} 

109 resize_value = resize_map.get(fit_mode, fit_mode) 

110 transform_parts.append(f"resize={resize_value}") 

111 

112 # Quality and format 

113 transform_parts.append( 

114 f"quality={transformations.get('quality', self.settings.default_quality if self.settings else 80)}" 

115 ) 

116 output_format = transformations.get( 

117 "format", self.settings.default_format if self.settings else "auto" 

118 ) 

119 if output_format != "auto": 

120 transform_parts.append(f"output={output_format}") 

121 

122 # Advanced effects 

123 for effect in ("blur", "brightness", "contrast", "saturation", "rotate"): 

124 if effect in transformations: 

125 transform_parts.append(f"{effect}={transformations[effect]}") 

126 

127 # Focus point 

128 if "focus" in transformations: 

129 focus = transformations["focus"] 

130 if isinstance(focus, dict) and "x" in focus and "y" in focus: 

131 transform_parts.append(f"focus={focus['x']}x{focus['y']}") 

132 elif isinstance(focus, str): 

133 transform_parts.append(f"focus={focus}") 

134 

135 # Progressive JPEG 

136 if ( 

137 self.settings 

138 and self.settings.enable_progressive 

139 and output_format 

140 in ( 

141 "jpeg", 

142 "jpg", 

143 "auto", 

144 ) 

145 ): 

146 transform_parts.append("progressive=true") 

147 

148 return transform_parts 

149 

150 async def get_image_url( 

151 self, image_id: str, transformations: dict[str, Any] | None = None 

152 ) -> str: 

153 """Generate TwicPics URL with real-time transformations.""" 

154 if not self.settings: 

155 self.settings = TwicPicsImagesSettings() 

156 

157 base_url = f"https://{self.settings.domain}/{image_id}" 

158 

159 # Apply transformations if provided 

160 if transformations: 

161 transform_parts = self._build_transform_parts(transformations) 

162 if transform_parts: 

163 transform_string = "/".join(transform_parts) 

164 return f"{base_url}?twic=v1/{transform_string}" 

165 

166 # Default optimizations 

167 default_transforms = [ 

168 f"quality={self.settings.default_quality}", 

169 f"output={self.settings.default_format}", 

170 ] 

171 if self.settings.enable_progressive: 

172 default_transforms.append("progressive=true") 

173 

174 return f"{base_url}?twic=v1/{'/'.join(default_transforms)}" 

175 

176 def get_img_tag(self, image_id: str, alt: str, **attributes: Any) -> str: 

177 """Generate img tag with TwicPics patterns and optimization.""" 

178 transformations = attributes.pop("transformations", {}) 

179 

180 # Generate optimized URL 

181 try: 

182 loop = asyncio.get_event_loop() 

183 url = loop.run_until_complete(self.get_image_url(image_id, transformations)) 

184 except RuntimeError: 

185 loop = asyncio.new_event_loop() 

186 asyncio.set_event_loop(loop) 

187 url = loop.run_until_complete(self.get_image_url(image_id, transformations)) 

188 

189 # Build base attributes 

190 img_attrs = {"src": url, "alt": alt} | attributes 

191 

192 # Add TwicPics-specific optimizations 

193 if self.settings and self.settings.enable_lazy_loading: 

194 img_attrs["loading"] = "lazy" 

195 

196 # Add placeholder for better UX 

197 if self.settings and self.settings.enable_placeholder: 

198 placeholder_transforms = { 

199 **transformations, 

200 "quality": self.settings.placeholder_quality, 

201 "width": 20, # Very small placeholder 

202 } 

203 with suppress(Exception): # Fallback to regular loading 

204 placeholder_url = loop.run_until_complete( 

205 self.get_image_url(image_id, placeholder_transforms) 

206 ) 

207 # For TwicPics, you might use data-src for lazy loading 

208 img_attrs["data-src"] = img_attrs["src"] 

209 img_attrs["src"] = placeholder_url 

210 img_attrs["class"] = ( 

211 f"{img_attrs.get('class', '')} twicpics-lazy".strip() 

212 ) 

213 

214 # Generate tag 

215 attr_string = " ".join( 

216 f'{k}="{v}"' for k, v in img_attrs.items() if v is not None 

217 ) 

218 return f"<img {attr_string}>" 

219 

220 def get_responsive_img_tag( 

221 self, 

222 image_id: str, 

223 alt: str, 

224 breakpoints: dict[str, dict[str, Any]] | None = None, 

225 **attributes: Any, 

226 ) -> str: 

227 """Generate responsive img tag with TwicPics breakpoint optimization.""" 

228 if not breakpoints: 

229 # Default responsive breakpoints 

230 breakpoints = { 

231 "320w": {"width": 320}, 

232 "640w": {"width": 640}, 

233 "768w": {"width": 768}, 

234 "1024w": {"width": 1024}, 

235 "1280w": {"width": 1280}, 

236 "1536w": {"width": 1536}, 

237 } 

238 

239 base_transformations = attributes.pop("transformations", {}) 

240 

241 # Generate srcset 

242 srcset_parts = [] 

243 for descriptor, transforms in breakpoints.items(): 

244 combined_transforms = base_transformations | transforms 

245 try: 

246 loop = asyncio.get_event_loop() 

247 url = loop.run_until_complete( 

248 self.get_image_url(image_id, combined_transforms) 

249 ) 

250 srcset_parts.append(f"{url} {descriptor}") 

251 except Exception: 

252 continue 

253 

254 # Default src (largest size) 

255 default_transforms = {**base_transformations, "width": 1024} 

256 try: 

257 loop = asyncio.get_event_loop() 

258 default_src = loop.run_until_complete( 

259 self.get_image_url(image_id, default_transforms) 

260 ) 

261 except RuntimeError: 

262 loop = asyncio.new_event_loop() 

263 asyncio.set_event_loop(loop) 

264 default_src = loop.run_until_complete( 

265 self.get_image_url(image_id, default_transforms) 

266 ) 

267 

268 # Build responsive img tag 

269 img_attrs = { 

270 "src": default_src, 

271 "srcset": ", ".join(srcset_parts), 

272 "alt": alt, 

273 "sizes": attributes.pop("sizes", "(max-width: 768px) 100vw, 50vw"), 

274 } | attributes 

275 

276 if self.settings and self.settings.enable_lazy_loading: 

277 img_attrs["loading"] = "lazy" 

278 

279 attr_string = " ".join( 

280 f'{k}="{v}"' for k, v in img_attrs.items() if v is not None 

281 ) 

282 return f"<img {attr_string}>" 

283 

284 async def close(self) -> None: 

285 """Close HTTP client.""" 

286 if self._client: 

287 await self._client.aclose() 

288 self._client = None 

289 

290 

291# Template filter registration for FastBlocks 

292def register_twicpics_filters(env: Any) -> None: 

293 """Register TwicPics filters for Jinja2 templates.""" 

294 

295 @env.filter("twic_url") # type: ignore[misc] 

296 async def twic_url_filter(image_id: str, **transformations: Any) -> str: 

297 """Template filter for TwicPics URLs.""" 

298 images = await depends.get("images") 

299 if isinstance(images, TwicPicsImages): 

300 return await images.get_image_url(image_id, transformations) 

301 return f"#{image_id}" 

302 

303 @env.filter("twic_img") # type: ignore[misc] 

304 def twic_img_filter(image_id: str, alt: str = "", **attributes: Any) -> str: 

305 """Template filter for TwicPics img tags.""" 

306 images = depends.get_sync("images") 

307 if isinstance(images, TwicPicsImages): 

308 return images.get_img_tag(image_id, alt, **attributes) 

309 return f'<img src="#{image_id}" alt="{alt}">' 

310 

311 @env.global_("twicpics_responsive") # type: ignore[misc] 

312 def twicpics_responsive( 

313 image_id: str, 

314 alt: str, 

315 breakpoints: dict[str, dict[str, Any]] | None = None, 

316 **attributes: Any, 

317 ) -> str: 

318 """Generate responsive image with TwicPics optimization.""" 

319 images = depends.get_sync("images") 

320 if isinstance(images, TwicPicsImages): 

321 return images.get_responsive_img_tag( 

322 image_id, alt, breakpoints, **attributes 

323 ) 

324 return f'<img src="#{image_id}" alt="{alt}">' 

325 

326 @env.filter("twic_placeholder") # type: ignore[misc] 

327 async def twic_placeholder_filter( 

328 image_id: str, width: int = 20, quality: int = 10 

329 ) -> str: 

330 """Generate ultra-low quality placeholder URL.""" 

331 images = await depends.get("images") 

332 if isinstance(images, TwicPicsImages): 

333 return await images.get_image_url( 

334 image_id, {"width": width, "quality": quality} 

335 ) 

336 return f"#{image_id}" 

337 

338 

339ImagesSettings = TwicPicsImagesSettings 

340Images = TwicPicsImages 

341 

342depends.set(Images, "twicpics") 

343 

344# ACB 0.19.0+ compatibility 

345__all__ = [ 

346 "TwicPicsImages", 

347 "TwicPicsImagesSettings", 

348 "register_twicpics_filters", 

349 "Images", 

350 "ImagesSettings", 

351]