""" Deep Analog — Web Application (deployment build) Privacy by design: * Images are processed entirely in memory and discarded when the response is sent. Nothing is written to disk, no database exists. * No filenames, image contents, or results are logged. * Access logging is disabled; no cookies, no analytics, no trackers. * Output images are re-encoded, which strips all EXIF/GPS metadata. Configuration (environment variables): STYLE_LUT_PATH path to StyleLUT checkpoint (default: weights/style_lut_slim.pth) ACCESS_CODE if set, every API call must carry this code (X-Access-Code header). Unset = gate disabled. MAX_SIDE max image dimension in px (default: 2048) PORT listen port (default: 7860, Hugging Face Spaces standard) """ import base64 import os import secrets from contextlib import asynccontextmanager import uvicorn from fastapi import FastAPI, UploadFile, File, Form, Header from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pipeline import DeepAnalogPipeline APP_DIR = os.path.dirname(os.path.abspath(__file__)) STYLE_LUT_PATH = os.environ.get( 'STYLE_LUT_PATH', os.path.join(APP_DIR, 'weights', 'style_lut_slim.pth')) ACCESS_CODE = os.environ.get('ACCESS_CODE', '') MAX_SIDE = int(os.environ.get('MAX_SIDE', '2048')) PORT = int(os.environ.get('PORT', '7860')) MAX_UPLOAD_BYTES = 80 * 1024 * 1024 # 80 MB per file (RAW files are large) pipeline: DeepAnalogPipeline = None @asynccontextmanager async def lifespan(app: FastAPI): global pipeline pipeline = DeepAnalogPipeline( style_lut_path=STYLE_LUT_PATH, device=os.environ.get('DEVICE', 'auto'), max_side=MAX_SIDE, ) print(f'[App] Ready on port {PORT} ' f'(access gate: {"ON" if ACCESS_CODE else "off"})') yield app = FastAPI(title='Deep Analog', lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None) app.mount('/static', StaticFiles(directory=os.path.join(APP_DIR, 'static')), name='static') # --------------------------------------------------------------------------- # Auth helper # --------------------------------------------------------------------------- def _check_access(code_header: str) -> bool: if not ACCESS_CODE: return True return secrets.compare_digest(code_header or '', ACCESS_CODE) def _denied(): return JSONResponse({'error': 'Invalid access code.'}, status_code=401) # --------------------------------------------------------------------------- # Routes # --------------------------------------------------------------------------- @app.get('/', response_class=HTMLResponse) async def index(): with open(os.path.join(APP_DIR, 'static', 'index.html'), 'r') as f: return f.read() @app.get('/api/debug/static') async def debug_static(): """Diagnostic: what static files does this container actually have?""" tree = {} static_root = os.path.join(APP_DIR, 'static') for root, _, files in os.walk(static_root): rel = os.path.relpath(root, static_root) jpgs = [f for f in files if f.endswith('.jpg')] if jpgs: sample = os.path.join(root, jpgs[0]) with open(sample, 'rb') as fh: head = fh.read(4) tree[rel] = {'jpg_count': len(jpgs), 'sample_is_real_jpeg': head[:3] == b'\xff\xd8\xff'} return tree @app.get('/api/config') async def config(): return { 'access_required': bool(ACCESS_CODE), 'max_side': MAX_SIDE, } @app.post('/api/auth') async def auth(x_access_code: str = Header(default='')): if _check_access(x_access_code): return {'ok': True} return _denied() @app.post('/api/process') async def process( reference: UploadFile = File(...), target: UploadFile = File(None), tone_strength: float = Form(0.7), grain_mult: float = Form(1.0), film_tone_strength: float = Form(0.7), halation_mult: float = Form(1.0), x_access_code: str = Header(default=''), ): if not _check_access(x_access_code): return _denied() ref_bytes = await reference.read() if len(ref_bytes) > MAX_UPLOAD_BYTES: return JSONResponse({'error': 'Reference file too large (max 80 MB). ' '“We’re gonna need a bigger boat” — or a smaller file.'}, status_code=413) tgt_bytes = None if target is not None: tgt_bytes = await target.read() if len(tgt_bytes) == 0: tgt_bytes = None elif len(tgt_bytes) > MAX_UPLOAD_BYTES: return JSONResponse({'error': 'Photo file too large (max 80 MB). ' '“We’re gonna need a bigger boat” — or a smaller file.'}, status_code=413) try: if tgt_bytes is None: # Mode 1: reference only → predicted LUT result = pipeline.process_reference_only( reference_bytes=ref_bytes, reference_filename=reference.filename or '', ) return JSONResponse({ 'mode': 'lut_only', 'cube_lut': result['cube_lut'], 'xmp': result['xmp'], 'costyle': result['costyle'], 'params': result['params'], 'chart_before': base64.b64encode(result['chart_before_jpg']).decode(), 'chart_after': base64.b64encode(result['chart_after_jpg']).decode(), }) # Mode 2: reference + target → LUT + rendered image result = pipeline.process( reference_bytes=ref_bytes, target_bytes=tgt_bytes, tone_strength=tone_strength, grain_mult=grain_mult, film_tone_strength=film_tone_strength, halation_mult=halation_mult, reference_filename=reference.filename or '', target_filename=target.filename or '', ) return JSONResponse({ 'mode': 'full', 'final': base64.b64encode(result['final_jpg']).decode(), 'graded': base64.b64encode(result['graded_jpg']).decode(), 'cube_lut': result['cube_lut'], 'xmp': result['xmp'], 'costyle': result['costyle'], 'params': result['params'], 'size': result['size'], }) except Exception as e: # Log the error type only — never image data or filenames print(f'[App] Processing error: {type(e).__name__}: {e}') return JSONResponse( {'error': 'Could not read this image format. ' '“Forget it, Jake. It’s Chinatown.” — try another file.'}, status_code=422) if __name__ == '__main__': # access_log=False: request paths/IPs are not logged (privacy by design) uvicorn.run(app, host='0.0.0.0', port=PORT, access_log=False)