""" ADH Image Compressor — Web API American Design Hub Reuses the exact compression / resize / convert logic proven in the desktop app (Tkinter/CustomTkinter version), exposed as a web API. Run locally: pip install -r requirements.txt uvicorn main:app --reload --port 8000 Endpoints: GET /api/health POST /api/process (multipart form: files[], quality, resize_mode, max_size, custom_w, custom_h, out_format) -> returns a ZIP of processed files, plus a stats header """ from fastapi import FastAPI, UploadFile, File, Form, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse from PIL import Image from pathlib import Path import io import zipfile import tempfile import shutil import os import json app = FastAPI(title="ADH Image Compressor API", version="1.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], # tighten to your domain(s) before production allow_methods=["*"], allow_headers=["*"], ) SUPPORTED_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".heic", ".heif", ".bmp", ".tiff", ".tif", ".gif", ".pdf", ".avif"} FORMAT_SAVE = { "JPEG": ("JPEG", ".jpg"), "PNG": ("PNG", ".png"), "WebP": ("WEBP", ".webp"), "AVIF": ("AVIF", ".avif"), } FREE_TIER_MAX_FILES = 50 # matches the pricing plan — enforced per-request here as an example MAX_UPLOAD_MB = 25 # ── Core processing (ported from the desktop app) ────────────────────────── def open_image(path: Path) -> Image.Image: ext = path.suffix.lower() if ext in (".heic", ".heif"): try: from pillow_heif import register_heif_opener register_heif_opener() except ImportError: raise RuntimeError("Server missing pillow-heif for HEIC support") return Image.open(path) def resize_crop(img: Image.Image, tw: int, th: int) -> Image.Image: sw, sh = img.size scale = max(tw / sw, th / sh) img = img.resize((int(sw * scale), int(sh * scale)), Image.LANCZOS) left = (img.width - tw) // 2 top = (img.height - th) // 2 return img.crop((left, top, left + tw, top + th)) def resize_fit(img: Image.Image, max_size: int) -> Image.Image: if img.width > max_size or img.height > max_size: img.thumbnail((max_size, max_size), Image.LANCZOS) return img def process_image(src: Path, dst_dir: Path, quality: int, resize_mode: str, max_size: int, cw: int, ch: int, fmt_name: str) -> Path: ext = src.suffix.lower() img = open_image(src) if img.mode not in ("RGB", "RGBA", "L"): img = img.convert("RGB") if resize_mode == "fit" and max_size > 0: img = resize_fit(img, max_size) elif resize_mode == "custom" and cw > 0 and ch > 0: img = resize_crop(img, cw, ch) # resize_mode == "none" -> leave as is if fmt_name == "Same as input": if ext in (".jpg", ".jpeg", ".bmp", ".tiff", ".tif", ".gif", ".heic", ".heif"): fmt, ext_out = "JPEG", ".jpg" elif ext == ".png": fmt, ext_out = "PNG", ".png" elif ext == ".webp": fmt, ext_out = "WEBP", ".webp" elif ext == ".avif": fmt, ext_out = "AVIF", ".avif" else: fmt, ext_out = "JPEG", ".jpg" else: fmt, ext_out = FORMAT_SAVE[fmt_name] if fmt == "JPEG" and img.mode == "RGBA": img = img.convert("RGB") out_path = dst_dir / (src.stem + ext_out) kw = {"format": fmt} if fmt in ("JPEG", "WEBP", "AVIF"): kw["quality"] = quality elif fmt == "PNG": kw["compress_level"] = max(1, min(9, int((100 - quality) / 11))) img.save(out_path, **kw) return out_path def process_pdf(src: Path, dst_dir: Path, quality: int, max_size: int, resize_mode: str, fmt_name: str) -> list[Path]: try: import fitz # PyMuPDF except ImportError: raise RuntimeError("Server missing pymupdf for PDF support") outputs = [] if fmt_name not in ("Same as input", "PDF"): fmt, ext_out = FORMAT_SAVE[fmt_name] doc = fitz.open(str(src)) for i, pg in enumerate(doc): pix = pg.get_pixmap(matrix=fitz.Matrix(1.5, 1.5)) img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples) if resize_mode == "fit" and max_size: img = resize_fit(img, max_size) p = dst_dir / f"{src.stem}_page{i+1}{ext_out}" img.save(p, format=fmt, quality=quality) outputs.append(p) doc.close() return outputs doc = fitz.open(str(src)) out_doc = fitz.open() for pg in doc: pix = pg.get_pixmap(matrix=fitz.Matrix(1.5, 1.5)) img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples) if resize_mode == "fit" and max_size: img = resize_fit(img, max_size) buf = io.BytesIO() img.save(buf, format="JPEG", quality=quality) buf.seek(0) img_pdf = fitz.open("pdf", fitz.open(stream=buf.read(), filetype="jpeg").convert_to_pdf()) out_doc.insert_pdf(img_pdf) out_path = dst_dir / f"{src.stem}.pdf" out_doc.save(str(out_path), garbage=4, deflate=True) doc.close() out_doc.close() outputs.append(out_path) return outputs # ── API routes ─────────────────────────────────────────────────────────────── @app.get("/api/health") def health(): return {"status": "ok", "service": "ADH Image Compressor API"} @app.post("/api/process") async def process( files: list[UploadFile] = File(...), quality: int = Form(85), resize_mode: str = Form("none"), # none | fit | custom max_size: int = Form(1500), custom_w: int = Form(1200), custom_h: int = Form(800), out_format: str = Form("Same as input"), tier: str = Form("free"), # free | pro | team — for future limit enforcement ): if tier == "free" and len(files) > FREE_TIER_MAX_FILES: raise HTTPException( status_code=402, detail=f"Free plan is limited to {FREE_TIER_MAX_FILES} files per month. Upgrade to Pro for unlimited batches." ) with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) in_dir = tmp_path / "in" out_dir = tmp_path / "out" in_dir.mkdir(); out_dir.mkdir() total_orig = 0 total_comp = 0 saved_paths: list[Path] = [] errors: list[str] = [] for uf in files: ext = Path(uf.filename).suffix.lower() if ext not in SUPPORTED_EXTS: errors.append(f"{uf.filename}: unsupported format") continue data = await uf.read() if len(data) > MAX_UPLOAD_MB * 1024 * 1024: errors.append(f"{uf.filename}: exceeds {MAX_UPLOAD_MB}MB limit") continue src_path = in_dir / uf.filename src_path.write_bytes(data) total_orig += len(data) try: if ext == ".pdf": outs = process_pdf(src_path, out_dir, quality, max_size, resize_mode, out_format) saved_paths.extend(outs) total_comp += sum(p.stat().st_size for p in outs) else: out_path = process_image(src_path, out_dir, quality, resize_mode, max_size, custom_w, custom_h, out_format) saved_paths.append(out_path) total_comp += out_path.stat().st_size except Exception as e: errors.append(f"{uf.filename}: {e}") if not saved_paths: raise HTTPException(status_code=422, detail={"message": "No files processed", "errors": errors}) # Build ZIP in memory zip_buf = io.BytesIO() with zipfile.ZipFile(zip_buf, "w", zipfile.ZIP_DEFLATED) as zf: for p in saved_paths: zf.write(p, arcname=p.name) zip_buf.seek(0) stats = { "files_processed": len(saved_paths), "errors": errors, "original_bytes": total_orig, "compressed_bytes": total_comp, "saved_bytes": total_orig - total_comp, "saved_pct": round((1 - total_comp / total_orig) * 100, 1) if total_orig else 0, } return StreamingResponse( zip_buf, media_type="application/zip", headers={ "Content-Disposition": "attachment; filename=adh-compressed.zip", "X-ADH-Stats": json.dumps(stats), "Access-Control-Expose-Headers": "X-ADH-Stats", }, ) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)