import os import cv2 import uuid import logging from fastapi import FastAPI, File, UploadFile, Form, HTTPException from deepface import DeepFace from fastapi.responses import FileResponse # Rasmni ko'rsatish uchun kerak # --- GLOBAL SOZLAMALAR --- logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') SOURCE_IMAGE_DIR = "images" PROCESSED_IMAGE_DIR = "/tmp/processed_images" LIVE_DETECTED_FACES_DIR = "/tmp/live_detected_faces" CHOSEN_MODEL = "ArcFace" CHOSEN_DETECTOR = "retinaface" # --- --- --- --- --- --- --- def run_preprocessing(): """ Manba papkasidagi rasmlardan yuzlarni aniqlab, qirqib oladi va ularni PROCESSED_IMAGE_DIR papkasiga saqlaydi. """ logging.info("--- Oldindan ishlov berish jarayoni boshlandi ---") os.makedirs(PROCESSED_IMAGE_DIR, exist_ok=True) os.makedirs(LIVE_DETECTED_FACES_DIR, exist_ok=True) logging.info(f"'{PROCESSED_IMAGE_DIR}' va '{LIVE_DETECTED_FACES_DIR}' papkalari tayyor.") source_files = [f for f in os.listdir(SOURCE_IMAGE_DIR) if f.lower().endswith(('.png', '.jpg', '.jpeg'))] existing_processed_files = os.listdir(PROCESSED_IMAGE_DIR) if not source_files: logging.warning(f"'{SOURCE_IMAGE_DIR}' papkasida ishlov berish uchun rasm topilmadi.") return logging.info(f"Jami {len(source_files)} ta rasm topildi. Tekshirish va qayta ishlash boshlandi...") processed_count = 0 skipped_count = 0 error_count = 0 for filename in source_files: if filename in existing_processed_files: skipped_count += 1 continue source_path = os.path.join(SOURCE_IMAGE_DIR, filename) target_path = os.path.join(PROCESSED_IMAGE_DIR, filename) try: face_objs = DeepFace.extract_faces( img_path=source_path, detector_backend=CHOSEN_DETECTOR, enforce_detection=True, align=True ) face_img = face_objs[0]['face'] if face_img.dtype == 'float64' or face_img.dtype == 'float32': face_img = (face_img * 255).astype('uint8') face_img_bgr = cv2.cvtColor(face_img, cv2.COLOR_RGB2BGR) success = cv2.imwrite(target_path, face_img_bgr) if not success: raise IOError(f"Faylni yozishda xatolik: {target_path}") logging.info(f"SUCCESS: '{filename}' rasmida yuz aniqlandi va saqlandi.") processed_count += 1 except Exception as e: logging.error(f"ERROR: '{filename}' rasmini qayta ishlashda xatolik: {str(e)}") error_count += 1 logging.info("--- Oldindan ishlov berish yakunlandi ---") logging.info(f"Muvaffaqiyatli: {processed_count}, O'tkazib yuborildi: {skipped_count}, Xatoliklar: {error_count}") # --- FastAPI ILOVASI --- app = FastAPI( title="Advanced Face Verification API", description="Talabalarning yuzini tekshirish va diagnostika qilish tizimi." ) @app.on_event("startup") def on_startup(): run_preprocessing() logging.info("Asosiy modellarni xotiraga yuklash...") try: DeepFace.build_model(CHOSEN_MODEL) logging.info(f"'{CHOSEN_MODEL}' tanish modeli muvaffaqiyatli yuklandi.") except Exception as e: logging.error(f"Modellarni yuklashda xatolik: {e}") logging.info("API so'rovlarni qabul qilishga tayyor.") @app.get("/") def read_root(): return {"message": "Face Verification API ishlamoqda."} @app.post("/verify/") async def verify_face( image: UploadFile = File(..., description="Kameradan olingan talaba rasmi"), pinfl: str = Form(..., description="Talabaning PINFL raqami") ): live_image_path = f"/tmp/{uuid.uuid4()}{os.path.splitext(image.filename)[1]}" detected_face_filename = None try: student_image_path = os.path.join(PROCESSED_IMAGE_DIR, f"{pinfl}.jpg") if not os.path.exists(student_image_path): raise HTTPException(status_code=404, detail=f"'{pinfl}' raqamli talaba uchun tizimda qayta ishlangan rasm topilmadi.") with open(live_image_path, "wb") as buffer: buffer.write(await image.read()) # Jonli rasmdagi yuzni ajratib saqlash (diagnostika uchun) try: live_face_objs = DeepFace.extract_faces(live_image_path, detector_backend=CHOSEN_DETECTOR, enforce_detection=True, align=True) live_face_img = live_face_objs[0]['face'] if live_face_img.dtype == 'float64' or live_face_img.dtype == 'float32': live_face_img = (live_face_img * 255).astype('uint8') live_face_bgr = cv2.cvtColor(live_face_img, cv2.COLOR_RGB2BGR) detected_face_filename = f"live_{pinfl}_{uuid.uuid4().hex[:8]}.jpg" detected_face_path = os.path.join(LIVE_DETECTED_FACES_DIR, detected_face_filename) cv2.imwrite(detected_face_path, live_face_bgr) except Exception as e: logging.warning(f"Jonli rasmdagi yuzni saqlashda xatolik: {e}") # Asosiy solishtirish result = DeepFace.verify( img1_path=student_image_path, img2_path=live_image_path, model_name=CHOSEN_MODEL, detector_backend=CHOSEN_DETECTOR, enforce_detection=True ) similarity = (1 - result['distance']) * 100 response_data = { "ok": bool(result["verified"]), "similarity": round(similarity, 2), "pinfl": pinfl, "view_detected_live_face_url": f"/view-live-face/{detected_face_filename}" if detected_face_filename else None } return response_data except HTTPException as e: raise e except Exception as e: logging.error(f"Verify endpoint'da kutilmagan xatolik: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Noma'lum ichki xatolik: {str(e)}") finally: if os.path.exists(live_image_path): os.remove(live_image_path) # --- DIAGNOSTIKA UCHUN ENDPOINTLAR (o'zgarishsiz qoldi) --- @app.get("/list-processed-files/") def list_processed_files(): if not os.path.exists(PROCESSED_IMAGE_DIR): return {"error": f"Papka topilmadi: {PROCESSED_IMAGE_DIR}"} files = os.listdir(PROCESSED_IMAGE_DIR) return {"directory": PROCESSED_IMAGE_DIR, "file_count": len(files), "files": files} @app.get("/view-image/{image_name}") def view_image(image_name: str): image_path = os.path.join(PROCESSED_IMAGE_DIR, image_name) if not os.path.exists(image_path): raise HTTPException(status_code=404, detail="Rasm topilmadi") return FileResponse(image_path) @app.get("/list-live-faces/") def list_live_faces(): if not os.path.exists(LIVE_DETECTED_FACES_DIR): return {"error": f"Papka topilmadi: {LIVE_DETECTED_FACES_DIR}"} files = os.listdir(LIVE_DETECTED_FACES_DIR) return {"directory": LIVE_DETECTED_FACES_DIR, "file_count": len(files), "files": files} @app.get("/view-live-face/{image_name}") def view_live_face(image_name: str): image_path = os.path.join(LIVE_DETECTED_FACES_DIR, image_name) if not os.path.exists(image_path): raise HTTPException(status_code=404, detail="Rasm topilmadi") return FileResponse(image_path)