Spaces:
Running
Running
| """ | |
| 🦊 Sisi PDF API - Universal HTML to PDF Converter | |
| Perfect support for Chinese (Simplified & Traditional), Emoji, SVG, and modern CSS | |
| Powered by Playwright + Chromium | |
| 🔒 Protected with API Key Authentication | |
| Hardening (v2.0.0): | |
| - API_KEY is required; the service refuses to start without it (no default key) | |
| - Shared Chromium browser launched once at startup; concurrency-limited (no per-request launch) | |
| - HTML payload size limit (default 2MB) | |
| - Content-Disposition filename sanitized (no header injection) | |
| - Renderer network requests to internal/private hosts blocked (SSRF guard) | |
| - CORS origins configurable via env (defaults to permissive to avoid breakage) | |
| """ | |
| import asyncio | |
| import base64 | |
| import ipaddress | |
| import logging | |
| import os | |
| import re | |
| import secrets | |
| from contextlib import asynccontextmanager | |
| from typing import Optional | |
| from urllib.parse import quote, urlparse | |
| from fastapi import Depends, FastAPI, Header, HTTPException | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import Response | |
| from playwright.async_api import async_playwright | |
| from pydantic import BaseModel | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| # ── Configuration (all from environment) ────────────────────────────────────── | |
| # 🔐 API key is mandatory. Refuse to start without it (no insecure default). | |
| API_KEY = os.environ.get("API_KEY") | |
| if not API_KEY: | |
| raise RuntimeError( | |
| "API_KEY environment variable is required. " | |
| "Set it as a Space secret before starting the service." | |
| ) | |
| # HTML payload cap (~2MB, aligned with the branded-report proxy guideline) | |
| MAX_HTML_BYTES = int(os.environ.get("MAX_HTML_BYTES", 2_000_000)) | |
| # Concurrent PDF renders sharing the single Chromium process | |
| PDF_MAX_CONCURRENCY = int(os.environ.get("PDF_MAX_CONCURRENCY", 4)) | |
| # When "strict", block ALL external http(s) sub-resources (only data:/blob:/about:) | |
| BLOCK_EXTERNAL_STRICT = os.environ.get("PDF_BLOCK_EXTERNAL", "").lower() == "strict" | |
| CHROMIUM_ARGS = [ | |
| "--no-sandbox", | |
| "--disable-setuid-sandbox", | |
| "--disable-dev-shm-usage", | |
| "--disable-gpu", | |
| ] | |
| ALWAYS_ALLOWED_SCHEMES = {"data", "blob", "about"} | |
| def verify_api_key(x_api_key: Optional[str] = Header(None)): | |
| """Verify API key from request header (constant-time comparison).""" | |
| if not x_api_key or not secrets.compare_digest(x_api_key, API_KEY): | |
| raise HTTPException( | |
| status_code=403, | |
| detail="🔒 Invalid or missing API key. Please contact SisiTheFox for access.", | |
| ) | |
| return x_api_key | |
| # ── Lifespan: launch the shared browser once ────────────────────────────────── | |
| async def lifespan(app: FastAPI): | |
| pw = await async_playwright().start() | |
| browser = await pw.chromium.launch(headless=True, args=CHROMIUM_ARGS) | |
| app.state.playwright = pw | |
| app.state.browser = browser | |
| app.state.semaphore = asyncio.Semaphore(PDF_MAX_CONCURRENCY) | |
| logger.info("🦊 Chromium launched (shared, max concurrency=%s)", PDF_MAX_CONCURRENCY) | |
| try: | |
| yield | |
| finally: | |
| await browser.close() | |
| await pw.stop() | |
| logger.info("🦊 Chromium closed") | |
| app = FastAPI( | |
| title="Sisi PDF API", | |
| description="Universal HTML to PDF converter with perfect Chinese, Emoji, and SVG support", | |
| version="2.0.0", | |
| lifespan=lifespan, | |
| ) | |
| # ── CORS (env-configurable; default permissive to avoid breaking callers) ────── | |
| _origins_env = os.environ.get("ALLOWED_ORIGINS", "*").strip() | |
| if _origins_env == "*": | |
| _allow_origins = ["*"] | |
| _allow_credentials = False # browsers forbid credentials with wildcard origin | |
| else: | |
| _allow_origins = [o.strip() for o in _origins_env.split(",") if o.strip()] | |
| _allow_credentials = True | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=_allow_origins, | |
| allow_credentials=_allow_credentials, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| class PDFRequest(BaseModel): | |
| """PDF generation request""" | |
| html_content: str # Full HTML content to convert | |
| filename: Optional[str] = "document.pdf" | |
| page_size: Optional[str] = "A4" # A4, Letter, A3, etc. | |
| margin_top: Optional[str] = "20mm" | |
| margin_bottom: Optional[str] = "20mm" | |
| margin_left: Optional[str] = "20mm" | |
| margin_right: Optional[str] = "20mm" | |
| print_background: Optional[bool] = True | |
| landscape: Optional[bool] = False | |
| prefer_css_page_size: Optional[bool] = False | |
| render_timeout_ms: Optional[int] = 30000 | |
| # ── Helpers ─────────────────────────────────────────────────────────────────── | |
| def sanitize_filename(name: Optional[str]) -> str: | |
| """Make a safe filename for Content-Disposition (prevents header injection).""" | |
| name = (name or "").strip() | |
| name = name.replace("\\", "/").split("/")[-1] # strip any path | |
| name = re.sub(r'[\r\n"]', "", name) # CR/LF/quotes | |
| name = "".join(ch for ch in name if ch >= " ") # drop control chars | |
| name = name.strip() | |
| if not name: | |
| name = "document.pdf" | |
| if not name.lower().endswith(".pdf"): | |
| name += ".pdf" | |
| return name[:200] | |
| def content_disposition(name: str) -> str: | |
| """Return an ASCII-safe header while preserving UTF-8 names for browsers.""" | |
| ascii_name = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip(".-") | |
| if not ascii_name: | |
| ascii_name = "document.pdf" | |
| elif not ascii_name.lower().endswith(".pdf"): | |
| ascii_name += ".pdf" | |
| encoded_name = quote(name, safe="") | |
| return f"attachment; filename=\"{ascii_name}\"; filename*=UTF-8''{encoded_name}" | |
| async def _host_is_internal(host: str) -> bool: | |
| """True if host resolves to a private/loopback/link-local/reserved address.""" | |
| try: | |
| loop = asyncio.get_running_loop() | |
| infos = await loop.getaddrinfo(host, None) | |
| except Exception: | |
| return True # cannot resolve -> treat as unsafe | |
| for info in infos: | |
| addr = info[4][0] | |
| try: | |
| ip = ipaddress.ip_address(addr) | |
| except ValueError: | |
| return True | |
| if ( | |
| ip.is_private or ip.is_loopback or ip.is_link_local | |
| or ip.is_reserved or ip.is_multicast or ip.is_unspecified | |
| ): | |
| return True | |
| return False | |
| async def _guard_route(route): | |
| """SSRF guard for sub-resource requests made while rendering the HTML.""" | |
| try: | |
| parsed = urlparse(route.request.url) | |
| scheme = parsed.scheme.lower() | |
| if scheme in ALWAYS_ALLOWED_SCHEMES: | |
| await route.continue_() | |
| return | |
| if scheme not in ("http", "https"): | |
| await route.abort() | |
| return | |
| if BLOCK_EXTERNAL_STRICT: | |
| await route.abort() | |
| return | |
| host = parsed.hostname | |
| if not host or await _host_is_internal(host): | |
| await route.abort() | |
| return | |
| await route.continue_() | |
| except Exception: | |
| # Never let the guard hang the render; fail closed. | |
| try: | |
| await route.abort() | |
| except Exception: | |
| pass | |
| async def prepare_page(page, request: PDFRequest): | |
| """Load HTML and wait for print assets before Chromium snapshots the page.""" | |
| timeout = max(5000, min(request.render_timeout_ms or 30000, 60000)) | |
| await page.emulate_media(media="print") | |
| await page.set_content(request.html_content, wait_until="networkidle", timeout=timeout) | |
| await page.evaluate("document.fonts.ready") | |
| await page.evaluate( | |
| """ | |
| async () => { | |
| await Promise.all([...document.images].map((image) => { | |
| if (image.complete) return Promise.resolve(); | |
| return new Promise((resolve) => { | |
| image.addEventListener('load', resolve, { once: true }); | |
| image.addEventListener('error', resolve, { once: true }); | |
| }); | |
| })); | |
| } | |
| """ | |
| ) | |
| async def render_pdf(request: PDFRequest) -> bytes: | |
| """Render a PDF using the shared browser, isolated per-request context.""" | |
| if len(request.html_content.encode("utf-8")) > MAX_HTML_BYTES: | |
| raise HTTPException( | |
| status_code=413, | |
| detail=f"HTML content too large (max {MAX_HTML_BYTES} bytes).", | |
| ) | |
| async with app.state.semaphore: | |
| context = await app.state.browser.new_context() | |
| try: | |
| page = await context.new_page() | |
| await page.route("**/*", _guard_route) | |
| await prepare_page(page, request) | |
| return await page.pdf( | |
| format=request.page_size, | |
| prefer_css_page_size=request.prefer_css_page_size, | |
| print_background=request.print_background, | |
| landscape=request.landscape, | |
| margin={ | |
| "top": request.margin_top, | |
| "bottom": request.margin_bottom, | |
| "left": request.margin_left, | |
| "right": request.margin_right, | |
| }, | |
| ) | |
| finally: | |
| await context.close() | |
| # ── Routes ──────────────────────────────────────────────────────────────────── | |
| def root(): | |
| """Health check""" | |
| return { | |
| "status": "online", | |
| "service": "Sisi PDF API", | |
| "version": "2.0.0", | |
| "features": [ | |
| "Perfect Chinese support (Simplified & Traditional)", | |
| "Colorful emoji rendering", | |
| "SVG graphics (100% accurate)", | |
| "Modern CSS (Flexbox, Grid, Gradients)", | |
| "Powered by Chromium", | |
| ], | |
| } | |
| def health_check(): | |
| """Health check endpoint for monitoring""" | |
| return {"status": "healthy", "service": "Sisi PDF API", "version": "2.0.0"} | |
| async def generate_pdf(request: PDFRequest, api_key: str = Depends(verify_api_key)): | |
| """ | |
| Generate PDF from HTML content. 🔒 Requires API Key authentication. | |
| Perfect support for Chinese, Emoji, SVG and modern CSS3. | |
| """ | |
| safe_name = sanitize_filename(request.filename) | |
| try: | |
| logger.info("📄 Generating PDF: %s", safe_name) | |
| pdf_bytes = await render_pdf(request) | |
| logger.info("✅ PDF generated: %d bytes", len(pdf_bytes)) | |
| return Response( | |
| content=pdf_bytes, | |
| media_type="application/pdf", | |
| headers={"Content-Disposition": content_disposition(safe_name)}, | |
| ) | |
| except HTTPException: | |
| raise | |
| except Exception as e: | |
| logger.error("❌ PDF generation error: %s", str(e)) | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def generate_pdf_base64(request: PDFRequest, api_key: str = Depends(verify_api_key)): | |
| """ | |
| Generate PDF and return as base64 string. 🔒 Requires API Key authentication. | |
| Useful for embedding in JSON responses. | |
| """ | |
| safe_name = sanitize_filename(request.filename) | |
| try: | |
| logger.info("📄 Generating PDF (base64): %s", safe_name) | |
| pdf_bytes = await render_pdf(request) | |
| pdf_base64 = base64.b64encode(pdf_bytes).decode("utf-8") | |
| logger.info("✅ PDF generated (base64): %d bytes", len(pdf_bytes)) | |
| return { | |
| "success": True, | |
| "filename": safe_name, | |
| "pdf_base64": pdf_base64, | |
| "size_bytes": len(pdf_bytes), | |
| } | |
| except HTTPException: | |
| raise | |
| except Exception as e: | |
| logger.error("❌ PDF generation error: %s", str(e)) | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run(app, host="0.0.0.0", port=7860) | |