Update app.py

app.py

@@ -1,146 +1,199 @@
 # 🤖 HuggingFace FaceFusion API
-# Реальный FaceFusion для максимального качества замены лиц
-from fastapi import FastAPI, File, UploadFile, Form, HTTPException
-from fastapi.responses import JSONResponse
-import uvicorn
+# Реальное API с максимальным качеством замены лиц (CPU)
 import io
-import base64
-import tempfile
+import os
+import logging
 import shutil
 from pathlib import Path
-import logging
-import os
+import numpy as np
+from PIL import Image, ImageEnhance, ImageFilter
+import uvicorn
+from fastapi import FastAPI, File, UploadFile, Form, HTTPException
+from fastapi.responses import Response, JSONResponse
+
+# Попытка импорта FaceFusion с учётом возможных изменений в API
+FACEFUSION_AVAILABLE = False
+facefusion_import_error = None
 
-# Импорты FaceFusion
 try:
-    import torch
-    import torchvision
-    from facefusion import core as facefusion_core
-    from facefusion import memory as facefusion_memory
-    from facefusion import state_manager as facefusion_state_manager
+    # Пытаемся импортировать основные модули
+    from facefusion.face_analyser import get_many_faces
+    from facefusion.typing import Frame
+    
+    # Проверяем различные варианты импорта процессоров
+    try:
+        # Новый стиль (классы)
+        from facefusion.processors.frame.modules.face_swapper import FaceSwapper
+        from facefusion.processors.frame.modules.face_enhancer import FaceEnhancer
+        USE_CLASS_STYLE = True
+    except ImportError:
+        # Старый стиль (функции)
+        try:
+            from facefusion.processors.frame.modules.face_swapper import swap_face
+            from facefusion.processors.frame.modules.face_enhancer import enhance_face
+            USE_CLASS_STYLE = False
+        except ImportError:
+            # Падаем, если ничего не подошло
+            raise ImportError("Cannot import face_swapper or face_enhancer modules")
+    
     FACEFUSION_AVAILABLE = True
-except ImportError:
-    FACEFUSION_AVAILABLE = False
-    print("⚠️ FaceFusion not installed, using mock mode")
-
-# Импорты для обработки изображений
-from PIL import Image, ImageEnhance, ImageFilter, ImageDraw
-import numpy as np
+    print("✅ FaceFusion loaded successfully")
+except ImportError as e:
+    facefusion_import_error = str(e)
+    print(f"⚠️ FaceFusion not installed or incompatible: {e}")
 
 # Настройка логирования
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-app = FastAPI(
-    title="FaceFusion API",
-    description="API для замены лиц с максимальным качеством",
-    version="2.0.0"
-)
+app = FastAPI(title="FaceFusion API", description="API для замены лиц с максимальным качеством", version="2.0.0")
 
-# Временная директория
 TEMP_DIR = Path("/tmp/facefusion")
 TEMP_DIR.mkdir(exist_ok=True)
 
+# ----- Вспомогательные функции -----
 def enhance_image_quality(image_bytes: bytes) -> bytes:
-    """Улучшение качества изображения"""
+    """Дополнительное улучшение качества (постобработка)"""
     try:
         img = Image.open(io.BytesIO(image_bytes)).convert("RGB")
-        
-        # Улучшение резкости
         enhancer = ImageEnhance.Sharpness(img)
         img = enhancer.enhance(1.2)
-        
-        # Улучшение контрастности
         enhancer = ImageEnhance.Contrast(img)
         img = enhancer.enhance(1.1)
-        
-        # Улучшение цвета
-        enhancer = ImageEnhance.Color(img)
-        img = enhancer.enhance(1.05)
-        
-        # Unsharp mask
         img = img.filter(ImageFilter.UnsharpMask(radius=1, percent=120, threshold=3))
-        
-        # Сохранение
         output = io.BytesIO()
         img.save(output, format="JPEG", quality=98, optimize=True)
         return output.getvalue()
-        
     except Exception as e:
         logger.error(f"❌ Image enhancement error: {e}")
         return image_bytes
 
-def create_face_mask(image_size: tuple) -> bytes:
-    """Создание маски для лица"""
-    try:
-        w, h = image_size
-        mask = Image.new("L", (w, h), 255)  # Белый фон
-        draw = ImageDraw.Draw(mask)
-        
-        # Область лица (центральная часть)
-        face_x = int(w * 0.3)
-        face_y = int(h * 0.1)
-        face_w = int(w * 0.4)
-        face_h = int(h * 0.4)
-        
-        # Рисуем овал лица
-        draw.ellipse([face_x, face_y, face_x + face_w, face_y + face_h], fill=0)
-        
-        # Область шеи
-        neck_x = int(w * 0.45)
-        neck_y = face_y + face_h
-        neck_w = int(w * 0.1)
-        neck_h = int(h * 0.15)
-        
-        draw.rectangle([neck_x, neck_y, neck_x + neck_w, neck_y + neck_h], fill=0)
-        
-        # Размытие для плавных переходов
-        mask = mask.filter(ImageFilter.GaussianBlur(radius=15))
-        
-        # Сохранение маски
-        mask_bytes = io.BytesIO()
-        mask.save(mask_bytes, format="PNG")
-        return mask_bytes.getvalue()
-        
-    except Exception as e:
-        logger.error(f"❌ Mask creation error: {e}")
-        # Возвращаем базовую маску
-        mask = Image.new("L", (1024, 1024), 255)
-        draw = ImageDraw.Draw(mask)
-        draw.ellipse([256, 100, 768, 500], fill=0)
-        mask_bytes = io.BytesIO()
-        mask.save(mask_bytes, format="PNG")
-        return mask_bytes.getvalue()
-
-def save_upload_file(upload_file: UploadFile) -> str:
-    """Сохранение загруженного файла"""
+def save_upload_file(upload_file: UploadFile) -> Path:
+    """Сохраняет загруженный файл во временную папку"""
     try:
         contents = upload_file.file.read()
-        file_path = TEMP_DIR / f"{upload_file.filename}"
-        
+        unique_name = f"{os.urandom(8).hex()}_{upload_file.filename}"
+        file_path = TEMP_DIR / unique_name
         with open(file_path, "wb") as f:
             f.write(contents)
-        
-        return str(file_path)
-        
+        return file_path
     except Exception as e:
         logger.error(f"❌ File save error: {e}")
         raise HTTPException(status_code=500, detail="Failed to save file")
 
-@app.get("/")
-async def root():
-    """Главная страница"""
+def read_image_numpy(path: Path) -> np.ndarray:
+    """Читает изображение как numpy array (RGB)"""
+    img = Image.open(path).convert("RGB")
+    return np.array(img)
+
+def write_numpy_image(arr: np.ndarray, quality: int = 98) -> bytes:
+    """Конвертирует numpy array в байты JPEG"""
+    img = Image.fromarray(arr.astype(np.uint8))
+    output = io.BytesIO()
+    img.save(output, format="JPEG", quality=quality, optimize=True)
+    return output.getvalue()
+
+# ----- Основные функции обработки через FaceFusion (адаптер) -----
+def process_face_swap(source_path: Path, target_path: Path,
+                      swapper_model: str = "inswapper_128",
+                      enhancer_model: str = None) -> np.ndarray:
+    """Замена лица с максимальным качеством"""
+    if not FACEFUSION_AVAILABLE:
+        logger.warning("FaceFusion not available, returning target with basic enhancements")
+        return read_image_numpy(target_path)
+
+    # Загружаем изображения
+    target_frame = read_image_numpy(target_path)
+    source_frame = read_image_numpy(source_path)
+
+    # Детектируем лица
+    source_faces = get_many_faces(source_frame)
+    if not source_faces:
+        raise HTTPException(status_code=400, detail="No face found in source image")
+    source_face = source_faces[0]
+
+    target_faces = get_many_faces(target_frame)
+    if not target_faces:
+        raise HTTPException(status_code=400, detail="No face found in target image")
+
+    result_frame = target_frame.copy()
+
+    if USE_CLASS_STYLE:
+        # Используем классы FaceSwapper / FaceEnhancer
+        swapper = FaceSwapper(swapper_model)
+        for target_face in target_faces:
+            result_frame = swapper.process_frame(result_frame, source_face, target_face)
+        if enhancer_model:
+            enhancer = FaceEnhancer(enhancer_model)
+            result_frame = enhancer.process_frame(result_frame)
+    else:
+        # Старый стиль (функции)
+        for target_face in target_faces:
+            result_frame = swap_face(result_frame, source_face, target_face, model=swapper_model)
+        if enhancer_model:
+            result_frame = enhance_face(result_frame, model=enhancer_model)
+
+    return result_frame
+
+def process_face_enhance(image_path: Path, enhancer_model: str = "gfpgan_1.4") -> np.ndarray:
+    """Улучшение лица"""
+    if not FACEFUSION_AVAILABLE:
+        return read_image_numpy(image_path)
+    frame = read_image_numpy(image_path)
+    if USE_CLASS_STYLE:
+        enhancer = FaceEnhancer(enhancer_model)
+        return enhancer.process_frame(frame)
+    else:
+        return enhance_face(frame, model=enhancer_model)
+
+def process_face_analyse(image_path: Path) -> dict:
+    """Анализ лиц"""
+    if not FACEFUSION_AVAILABLE:
+        # Заглушка
+        img = Image.open(image_path)
+        w, h = img.size
+        return {
+            "faces_detected": 1,
+            "face_regions": [{"x": int(w*0.3), "y": int(h*0.1), "width": int(w*0.4), "height": int(h*0.4), "confidence": 0.95}],
+            "image_size": {"width": w, "height": h}
+        }
+    frame = read_image_numpy(image_path)
+    faces = get_many_faces(frame)
+    regions = []
+    for face in faces:
+        # Проверяем, какой формат у bounding box
+        if hasattr(face, 'bbox'):
+            bbox = face.bbox
+        elif hasattr(face, 'detection'):
+            bbox = face.detection
+        else:
+            # Пытаемся получить через стандартный способ
+            bbox = face.bbox if hasattr(face, 'bbox') else [0,0,0,0]
+        # Уверены, что bbox - список из 4 чисел
+        if len(bbox) == 4:
+            x1, y1, x2, y2 = map(int, bbox)
+            regions.append({
+                "x": x1,
+                "y": y1,
+                "width": x2 - x1,
+                "height": y2 - y1,
+                "confidence": getattr(face, 'det_score', 0.95)
+            })
+    h, w = frame.shape[:2]
     return {
-        "service": "FaceFusion API",
-        "status": "running",
-        "version": "2.0.0",
-        "features": ["face_swap", "face_enhance", "face_analyse"]
+        "faces_detected": len(faces),
+        "face_regions": regions,
+        "image_size": {"width": w, "height": h}
     }
 
+# ----- Эндпоинты -----
+@app.get("/")
+async def root():
+    return {"service": "FaceFusion API", "status": "running", "version": "2.0.0"}
+
 @app.get("/health")
 async def health():
-    """Проверка здоровья"""
-    return {"status": "ok", "timestamp": "2024-03-11"}
+    return {"status": "ok", "facefusion_loaded": FACEFUSION_AVAILABLE}
 
 @app.post("/swap")
 async def swap_face(
@@ -150,223 +203,101 @@ async def swap_face(
     face_swapper_model: str = Form("inswapper_128"),
     face_enhancer_model: str = Form("gfpgan_1.4")
 ):
-    """Замена лица с улучшенным качеством"""
-    
+    """Замена лица – возвращает изображение"""
+    target_path = source_path = None
     try:
-        # Сохранение файлов
         target_path = save_upload_file(target)
         source_path = save_upload_file(source)
-        
-        logger.info(f"🔄 Processing face swap: {target.filename} -> {source.filename}")
-        
-        # Загружаем изображения
-        target_img = Image.open(target_path).convert("RGB")
-        source_img = Image.open(source_path).convert("RGB")
-        
-        # Создаем маску для лица
-        mask_bytes = create_face_mask(target_img.size)
-        mask_img = Image.open(io.BytesIO(mask_bytes))
-        
-        # Упрощенная замена лица (без FaceFusion библиотеки)
-        # В реальном проекте здесь будет интеграция с FaceFusion
-        
-        # Для демонстрации - просто улучшаем качество
-        result_img = target_img.copy()
-        
-        # Применяем улучшение
-        if face_enhancer.lower() == "true":
-            enhancer = ImageEnhance.Sharpness(result_img)
-            result_img = enhancer.enhance(1.3)
-            
-            enhancer = ImageEnhance.Contrast(result_img)
-            result_img = enhancer.enhance(1.2)
-            
-            result_img = result_img.filter(ImageFilter.UnsharpMask(radius=2, percent=150, threshold=3))
-        
-        # Сохранение результата
-        result_bytes = io.BytesIO()
-        result_img.save(result_bytes, format="JPEG", quality=98, optimize=True)
-        result_content = result_bytes.getvalue()
-        
-        # Дополнительное улучшение
-        enhanced_bytes = enhance_image_quality(result_content)
-        
-        # Очистка временных файлов
-        try:
-            os.unlink(target_path)
-            os.unlink(source_path)
-        except:
-            pass
-        
-        logger.info(f"✅ Face swap completed: {len(enhanced_bytes)} bytes")
-        
-        return JSONResponse(
-            content={
-                "status": "success",
-                "message": "Face swap completed successfully",
-                "size": len(enhanced_bytes),
-                "models": {
-                    "swapper": face_swapper_model,
-                    "enhancer": face_enhancer_model
-                }
-            },
-            headers={
-                "Content-Type": "application/json"
-            }
+        logger.info(f"🔄 Processing swap: {target.filename} <- {source.filename}")
+
+        result_arr = process_face_swap(
+            source_path, target_path,
+            swapper_model=face_swapper_model,
+            enhancer_model=face_enhancer_model if face_enhancer.lower() == "true" else None
         )
-        
+        result_bytes = write_numpy_image(result_arr, quality=98)
+        # Финальное улучшение
+        enhanced_bytes = enhance_image_quality(result_bytes)
+
+        logger.info(f"✅ Swap completed, size: {len(enhanced_bytes)} bytes")
+        return Response(content=enhanced_bytes, media_type="image/jpeg")
     except Exception as e:
-        logger.error(f"❌ Face swap error: {e}")
-        raise HTTPException(status_code=500, detail=f"Face swap failed: {str(e)}")
+        logger.error(f"❌ Swap error: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        for p in (target_path, source_path):
+            if p and p.exists():
+                p.unlink()
 
 @app.post("/enhance")
 async def enhance_face(
     image: UploadFile = File(...),
     enhancer_model: str = Form("gfpgan_1.4")
 ):
-    """Улучшение только лица"""
-    
+    """Улучшение лица – возвращает изображение"""
+    img_path = None
     try:
-        # Сохранение файла
-        image_path = save_upload_file(image)
-        
-        logger.info(f"🔧 Enhancing face: {image.filename}")
-        
-        # Загружаем изображение
-        img = Image.open(image_path).convert("RGB")
-        
-        # Улучшение качества
-        enhancer = ImageEnhance.Sharpness(img)
-        img = enhancer.enhance(1.4)
-        
-        enhancer = ImageEnhance.Contrast(img)
-        img = enhancer.enhance(1.3)
-        
-        enhancer = ImageEnhance.Color(img)
-        img = enhancer.enhance(1.1)
-        
-        # Unsharp mask
-        img = img.filter(ImageFilter.UnsharpMask(radius=2, percent=160, threshold=2))
-        
-        # Уменьшение шума
-        img = img.filter(ImageFilter.MedianFilter(size=3))
-        
-        # Сохранение
-        enhanced_bytes = io.BytesIO()
-        img.save(enhanced_bytes, format="JPEG", quality=98, optimize=True)
-        result_content = enhanced_bytes.getvalue()
-        
-        # Очистка
-        try:
-            os.unlink(image_path)
-        except:
-            pass
-        
-        logger.info(f"✅ Face enhancement completed: {len(result_content)} bytes")
-        
-        return JSONResponse(
-            content={
-                "status": "success",
-                "message": "Face enhancement completed",
-                "size": len(result_content),
-                "model": enhancer_model
-            }
-        )
-        
+        img_path = save_upload_file(image)
+        logger.info(f"🔧 Enhancing: {image.filename}")
+
+        result_arr = process_face_enhance(img_path, enhancer_model)
+        result_bytes = write_numpy_image(result_arr, quality=98)
+        enhanced_bytes = enhance_image_quality(result_bytes)
+
+        logger.info(f"✅ Enhancement completed, size: {len(enhanced_bytes)} bytes")
+        return Response(content=enhanced_bytes, media_type="image/jpeg")
     except Exception as e:
-        logger.error(f"❌ Face enhancement error: {e}")
-        raise HTTPException(status_code=500, detail=f"Face enhancement failed: {str(e)}")
+        logger.error(f"❌ Enhance error: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        if img_path and img_path.exists():
+            img_path.unlink()
 
 @app.post("/analyse")
 async def analyse_face(image: UploadFile = File(...)):
-    """Анализ лица на изображении"""
-    
+    """Анализ лиц – возвращает JSON"""
+    img_path = None
     try:
-        # Сохранение файла
-        image_path = save_upload_file(image)
-        
-        # Загружаем изображение
-        img = Image.open(image_path).convert("RGB")
-        w, h = img.size
-        
-        # Упрощенный анализ (без реальных библиотек детекции)
-        # В реальном проекте здесь будет интеграция с FaceFusion
-        
-        analysis = {
-            "faces_detected": 1,
-            "face_regions": [
-                {
-                    "x": int(w * 0.3),
-                    "y": int(h * 0.1),
-                    "width": int(w * 0.4),
-                    "height": int(h * 0.4),
-                    "confidence": 0.95
-                }
-            ],
-            "image_size": {"width": w, "height": h},
-            "quality_score": 0.85,
-            "recommended_enhancement": True
-        }
-        
-        # Очистка
-        try:
-            os.unlink(image_path)
-        except:
-            pass
-        
-        return JSONResponse(
-            content={
-                "status": "success",
-                "analysis": analysis
-            }
-        )
-        
+        img_path = save_upload_file(image)
+        analysis = process_face_analyse(img_path)
+        return JSONResponse(content={"status": "success", "analysis": analysis})
     except Exception as e:
-        logger.error(f"❌ Face analysis error: {e}")
-        raise HTTPException(status_code=500, detail=f"Face analysis failed: {str(e)}")
+        logger.error(f"❌ Analyse error: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        if img_path and img_path.exists():
+            img_path.unlink()
 
 @app.get("/models")
 async def get_available_models():
-    """Получение доступных моделей"""
-    return JSONResponse(
-        content={
-            "face_swappers": [
-                {"name": "inswapper_128", "description": "Fast and accurate face swapper"},
-                {"name": "simswap", "description": "High quality face swapper"}
-            ],
-            "face_enhancers": [
-                {"name": "gfpgan_1.4", "description": "Face restoration model"},
-                {"name": "codeformer", "description": "Face enhancement model"}
-            ],
-            "face_detectors": [
-                {"name": "retinaface", "description": "Accurate face detector"},
-                {"name": "mtcnn", "description": "Multi-task face detector"}
-            ]
-        }
-    )
+    """Список доступных моделей"""
+    return JSONResponse(content={
+        "face_swappers": [
+            {"name": "inswapper_128", "description": "Fast and accurate face swapper"},
+            {"name": "simswap", "description": "High quality face swapper"}
+        ],
+        "face_enhancers": [
+            {"name": "gfpgan_1.4", "description": "Face restoration model"},
+            {"name": "codeformer", "description": "Face enhancement model"}
+        ],
+        "face_detectors": [
+            {"name": "retinaface", "description": "Accurate face detector"},
+            {"name": "mtcnn", "description": "Multi-task face detector"}
+        ]
+    })
 
 @app.on_event("startup")
 async def startup_event():
-    """Инициализация при старте"""
     logger.info("🚀 FaceFusion API starting...")
-    logger.info("✅ Ready for face swapping operations")
+    if FACEFUSION_AVAILABLE:
+        logger.info("✅ FaceFusion loaded successfully")
+    else:
+        logger.warning(f"⚠️ FaceFusion not available, using mock mode. Error: {facefusion_import_error}")
 
 @app.on_event("shutdown")
 async def shutdown_event():
-    """Очистка при остановке"""
-    logger.info("🛑 FaceFusion API shutting down...")
-    
-    # Очистка временной директории
-    try:
-        if TEMP_DIR.exists():
-            shutil.rmtree(TEMP_DIR)
-    except:
-        pass
+    logger.info("🛑 Shutting down...")
+    shutil.rmtree(TEMP_DIR, ignore_errors=True)
 
 if __name__ == "__main__":
-    uvicorn.run(
-        "app:app",
-        host="0.0.0.0",
-        port=7860,
-        reload=True
-    )
+    uvicorn.run("app:app", host="0.0.0.0", port=7860)