Coverage for src / idx_api / routers / uploads.py: 21%

1"""Photo upload endpoints with automatic resize and optimization.""" 

2 

3import hashlib 

4import os 

5from datetime import datetime 

6from io import BytesIO 

7from pathlib import Path 

8 

9from fastapi import APIRouter, Depends, File, HTTPException, UploadFile 

10from PIL import Image 

11from pydantic import BaseModel 

12 

13from idx_api.auth import RequiredUser 

14from idx_api.config import settings 

15 

16router = APIRouter() 

17 

18# Upload configuration 

19UPLOAD_DIR = Path("/data/uploads") 

20ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp", ".gif"} 

21MAX_FILE_SIZE = 10 * 1024 * 1024 # 10MB 

22 

23# Image size configurations 

24SIZES = { 

25 "agent_photo": {"width": 800, "height": 800, "quality": 85}, 

26 "agent_thumb": {"width": 200, "height": 200, "quality": 80}, 

27 "brokerage_logo": {"width": 600, "height": 300, "quality": 90}, 

28 "brokerage_thumb": {"width": 200, "height": 100, "quality": 85}, 

29 "broker_contact_photo": {"width": 800, "height": 800, "quality": 85}, 

30 "broker_contact_thumb": {"width": 200, "height": 200, "quality": 80}, 

31} 

32 

33 

34class UploadResponse(BaseModel): 

35 """Upload response with URLs.""" 

36 

37 url: str 

38 thumbnail_url: str | None = None 

39 filename: str 

40 size: int 

41 width: int 

42 height: int 

43 

44 

45def ensure_upload_dir(): 

46 """Ensure upload directory exists.""" 

47 UPLOAD_DIR.mkdir(parents=True, exist_ok=True) 

48 for subdir in ["agents", "brokerages", "broker_contacts", "thumbs"]: 

49 (UPLOAD_DIR / subdir).mkdir(exist_ok=True) 

50 

51 

52def validate_image(file: UploadFile) -> None: 

53 """Validate uploaded image file.""" 

54 # Check file extension 

55 ext = Path(file.filename or "").suffix.lower() 

56 if ext not in ALLOWED_EXTENSIONS: 

57 raise HTTPException( 

58 status_code=400, 

59 detail=f"Invalid file type. Allowed: {', '.join(ALLOWED_EXTENSIONS)}", 

60 ) 

61 

62 # Check file size (this is approximate, actual check happens during read) 

63 if file.size and file.size > MAX_FILE_SIZE: 

64 raise HTTPException( 

65 status_code=400, 

66 detail=f"File too large. Maximum size: {MAX_FILE_SIZE // 1024 // 1024}MB", 

67 ) 

68 

69 

70def generate_filename(original_filename: str, content: bytes) -> str: 

71 """Generate unique filename using hash of content.""" 

72 ext = Path(original_filename).suffix.lower() 

73 

74 # Generate hash from content for deduplication 

75 content_hash = hashlib.sha256(content).hexdigest()[:16] 

76 timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S") 

77 

78 return f"{timestamp}_{content_hash}{ext}" 

79 

80 

81def optimize_image( 

82 image: Image.Image, 

83 max_width: int, 

84 max_height: int, 

85 quality: int = 85, 

86) -> BytesIO: 

87 """Resize and optimize image.""" 

88 # Convert RGBA to RGB if needed (for JPEG compatibility) 

89 if image.mode in ("RGBA", "LA", "P"): 

90 background = Image.new("RGB", image.size, (255, 255, 255)) 

91 if image.mode == "P": 

92 image = image.convert("RGBA") 

93 background.paste(image, mask=image.split()[-1] if image.mode in ("RGBA", "LA") else None) 

94 image = background 

95 

96 # Calculate resize dimensions maintaining aspect ratio 

97 img_width, img_height = image.size 

98 ratio = min(max_width / img_width, max_height / img_height) 

99 

100 if ratio < 1: # Only resize if image is larger 

101 new_width = int(img_width * ratio) 

102 new_height = int(img_height * ratio) 

103 image = image.resize((new_width, new_height), Image.Resampling.LANCZOS) 

104 

105 # Save optimized image 

106 output = BytesIO() 

107 image.save(output, format="JPEG", quality=quality, optimize=True) 

108 output.seek(0) 

109 

110 return output 

111 

112 

113@router.post("/upload/agent-photo", response_model=UploadResponse) 

114async def upload_agent_photo( 

115 user: RequiredUser, 

116 file: UploadFile = File(...), 

117): 

118 """ 

119 Upload and optimize agent photo. 

120 

121 Generates two versions: 

122 - Full size: 800x800 max 

123 - Thumbnail: 200x200 max 

124 """ 

125 ensure_upload_dir() 

126 validate_image(file) 

127 

128 try: 

129 # Read file content 

130 content = await file.read() 

131 if len(content) > MAX_FILE_SIZE: 

132 raise HTTPException(status_code=400, detail="File too large") 

133 

134 # Open and validate image 

135 try: 

136 image = Image.open(BytesIO(content)) 

137 original_width, original_height = image.size 

138 except Exception as e: 

139 raise HTTPException(status_code=400, detail=f"Invalid image file: {str(e)}") 

140 

141 # Generate filename 

142 filename = generate_filename(file.filename or "photo.jpg", content) 

143 

144 # Process full size 

145 full_config = SIZES["agent_photo"] 

146 full_img = optimize_image( 

147 image.copy(), 

148 full_config["width"], 

149 full_config["height"], 

150 full_config["quality"], 

151 ) 

152 

153 # Save full size 

154 full_path = UPLOAD_DIR / "agents" / filename 

155 with open(full_path, "wb") as f: 

156 f.write(full_img.read()) 

157 

158 # Process thumbnail 

159 thumb_config = SIZES["agent_thumb"] 

160 thumb_img = optimize_image( 

161 image.copy(), 

162 thumb_config["width"], 

163 thumb_config["height"], 

164 thumb_config["quality"], 

165 ) 

166 

167 # Save thumbnail 

168 thumb_filename = f"thumb_{filename}" 

169 thumb_path = UPLOAD_DIR / "thumbs" / thumb_filename 

170 with open(thumb_path, "wb") as f: 

171 f.write(thumb_img.read()) 

172 

173 # Get final dimensions 

174 final_img = Image.open(full_path) 

175 final_width, final_height = final_img.size 

176 final_size = full_path.stat().st_size 

177 

178 # Return URLs relative to uploads endpoint 

179 return UploadResponse( 

180 url=f"/uploads/agents/{filename}", 

181 thumbnail_url=f"/uploads/thumbs/{thumb_filename}", 

182 filename=filename, 

183 size=final_size, 

184 width=final_width, 

185 height=final_height, 

186 ) 

187 

188 except HTTPException: 

189 raise 

190 except Exception as e: 

191 raise HTTPException(status_code=500, detail=f"Upload failed: {str(e)}") 

192 

193 

194@router.post("/upload/brokerage-logo", response_model=UploadResponse) 

195async def upload_brokerage_logo( 

196 user: RequiredUser, 

197 file: UploadFile = File(...), 

198): 

199 """ 

200 Upload and optimize brokerage logo. 

201 

202 Generates two versions: 

203 - Full size: 600x300 max 

204 - Thumbnail: 200x100 max 

205 """ 

206 ensure_upload_dir() 

207 validate_image(file) 

208 

209 try: 

210 # Read file content 

211 content = await file.read() 

212 if len(content) > MAX_FILE_SIZE: 

213 raise HTTPException(status_code=400, detail="File too large") 

214 

215 # Open and validate image 

216 try: 

217 image = Image.open(BytesIO(content)) 

218 original_width, original_height = image.size 

219 except Exception as e: 

220 raise HTTPException(status_code=400, detail=f"Invalid image file: {str(e)}") 

221 

222 # Generate filename 

223 filename = generate_filename(file.filename or "logo.jpg", content) 

224 

225 # Process full size 

226 full_config = SIZES["brokerage_logo"] 

227 full_img = optimize_image( 

228 image.copy(), 

229 full_config["width"], 

230 full_config["height"], 

231 full_config["quality"], 

232 ) 

233 

234 # Save full size 

235 full_path = UPLOAD_DIR / "brokerages" / filename 

236 with open(full_path, "wb") as f: 

237 f.write(full_img.read()) 

238 

239 # Process thumbnail 

240 thumb_config = SIZES["brokerage_thumb"] 

241 thumb_img = optimize_image( 

242 image.copy(), 

243 thumb_config["width"], 

244 thumb_config["height"], 

245 thumb_config["quality"], 

246 ) 

247 

248 # Save thumbnail 

249 thumb_filename = f"thumb_{filename}" 

250 thumb_path = UPLOAD_DIR / "thumbs" / thumb_filename 

251 with open(thumb_path, "wb") as f: 

252 f.write(thumb_img.read()) 

253 

254 # Get final dimensions 

255 final_img = Image.open(full_path) 

256 final_width, final_height = final_img.size 

257 final_size = full_path.stat().st_size 

258 

259 # Return URLs relative to uploads endpoint 

260 return UploadResponse( 

261 url=f"/uploads/brokerages/{filename}", 

262 thumbnail_url=f"/uploads/thumbs/{thumb_filename}", 

263 filename=filename, 

264 size=final_size, 

265 width=final_width, 

266 height=final_height, 

267 ) 

268 

269 except HTTPException: 

270 raise 

271 except Exception as e: 

272 raise HTTPException(status_code=500, detail=f"Upload failed: {str(e)}") 

273 

274 

275@router.post("/upload/broker-contact-photo", response_model=UploadResponse) 

276async def upload_broker_contact_photo( 

277 user: RequiredUser, 

278 file: UploadFile = File(...), 

279): 

280 """ 

281 Upload and optimize broker contact photo. 

282 

283 Generates two versions: 

284 - Full size: 800x800 max 

285 - Thumbnail: 200x200 max 

286 """ 

287 ensure_upload_dir() 

288 validate_image(file) 

289 

290 try: 

291 # Read file content 

292 content = await file.read() 

293 if len(content) > MAX_FILE_SIZE: 

294 raise HTTPException(status_code=400, detail="File too large") 

295 

296 # Open and validate image 

297 try: 

298 image = Image.open(BytesIO(content)) 

299 original_width, original_height = image.size 

300 except Exception as e: 

301 raise HTTPException(status_code=400, detail=f"Invalid image file: {str(e)}") 

302 

303 # Generate filename 

304 filename = generate_filename(file.filename or "photo.jpg", content) 

305 

306 # Process full size 

307 full_config = SIZES["broker_contact_photo"] 

308 full_img = optimize_image( 

309 image.copy(), 

310 full_config["width"], 

311 full_config["height"], 

312 full_config["quality"], 

313 ) 

314 

315 # Save full size 

316 full_path = UPLOAD_DIR / "broker_contacts" / filename 

317 with open(full_path, "wb") as f: 

318 f.write(full_img.read()) 

319 

320 # Process thumbnail 

321 thumb_config = SIZES["broker_contact_thumb"] 

322 thumb_img = optimize_image( 

323 image.copy(), 

324 thumb_config["width"], 

325 thumb_config["height"], 

326 thumb_config["quality"], 

327 ) 

328 

329 # Save thumbnail 

330 thumb_filename = f"thumb_{filename}" 

331 thumb_path = UPLOAD_DIR / "thumbs" / thumb_filename 

332 with open(thumb_path, "wb") as f: 

333 f.write(thumb_img.read()) 

334 

335 # Get final dimensions 

336 final_img = Image.open(full_path) 

337 final_width, final_height = final_img.size 

338 final_size = full_path.stat().st_size 

339 

340 # Return URLs relative to uploads endpoint 

341 return UploadResponse( 

342 url=f"/uploads/broker_contacts/{filename}", 

343 thumbnail_url=f"/uploads/thumbs/{thumb_filename}", 

344 filename=filename, 

345 size=final_size, 

346 width=final_width, 

347 height=final_height, 

348 ) 

349 

350 except HTTPException: 

351 raise 

352 except Exception as e: 

353 raise HTTPException(status_code=500, detail=f"Upload failed: {str(e)}")