diff --git a/app/api/users.py b/app/api/users.py index 043dd98..241e08f 100644 --- a/app/api/users.py +++ b/app/api/users.py @@ -1,28 +1,54 @@ -from fastapi import APIRouter, File, Form, UploadFile +from fastapi import APIRouter, File, Form, UploadFile, HTTPException from app.models.user import UserResponse +from app.services.user_service import ( + authenticate_user, + delete_user, + get_user, + register_user, +) + router = APIRouter() @router.post("/users/register", response_model=UserResponse) async def register(user_id: str = Form(...), image: UploadFile = File(...)): """회원 등록 엔드포인트""" - pass + image_data = await image.read() + registered_user = register_user(user_id, image_data) + + return UserResponse( + user_id=registered_user["user_id"], + registered_at=registered_user["registered_at"], + ) @router.post("/users/authenticate") async def authenticate(image: UploadFile = File(...)): """회원 인증 엔드포인트""" - pass + image_data = await image.read() + user_id = authenticate_user(image_data) + + if user_id is None: + raise HTTPException(status_code=401, detail="인증 실패: 사용자 없음") + + return {"user_id": user_id} @router.get("/users/{user_id}") def get_user_info(user_id: str): """회원 정보 조회 엔드포인트""" - pass + user_info = get_user(user_id) + if user_info is None: + raise HTTPException(status_code=404, detail="User not found") + return user_info @router.delete("/users/{user_id}") def delete_user_info(user_id: str): """회원 삭제 엔드포인트""" - pass + result = delete_user(user_id) + if not result: + raise + HTTPException(status_code=404, detail="사용자가 존재하지 않습니다.") + return {"message": "사용자가 성공적으로 삭제되었습니다."} diff --git a/app/face/face_db.py b/app/face/face_db.py index 380ae3a..f3e27b5 100644 --- a/app/face/face_db.py +++ b/app/face/face_db.py @@ -1,16 +1,27 @@ +import json +import os +from datetime import datetime DB_PATH = "face_db.json" def save_user(user_id, embedding): """사용자 등록""" - pass + db = load_db() + db[user_id] = {"embedding": embedding, "registered_at": str(datetime.now())} + with open(DB_PATH, "w") as f: + json.dump(db, f) def load_db(): """데이터베이스 로드""" - pass + if not os.path.exists(DB_PATH): + return {} + with open(DB_PATH, "r") as f: + return json.load(f) + def save_db(db): """데이터베이스 저장""" - pass + with open(DB_PATH, "w") as f: + json.dump(db, f) diff --git a/app/face/face_embedding.py b/app/face/face_embedding.py index 3acfb0c..05d770d 100644 --- a/app/face/face_embedding.py +++ b/app/face/face_embedding.py @@ -1,8 +1,13 @@ -def extract_embedding(image): - """이미지에서 임베딩 추출""" - pass +import numpy as np +from typing import Union +from deepface import DeepFace +def extract_embedding(img_path: Union[str, np.ndarray]) -> list: + """이미지에서 임베딩 추출""" + embedding = DeepFace.represent(img_path)[0]["embedding"] + return embedding -def verify_embedding(embedding1, embedding2): +def verify_embedding(embedding1: list, embedding2: list) -> bool: """두 임베딩이 같은 사람인지 검증""" - pass + is_same_person = DeepFace.verify(embedding1, embedding2)["verified"] + return is_same_person diff --git a/app/services/user_service.py b/app/services/user_service.py index 410b807..a0be8f2 100644 --- a/app/services/user_service.py +++ b/app/services/user_service.py @@ -1,18 +1,51 @@ +from datetime import datetime + +import cv2 +import numpy as np + +from app.face.face_db import load_db, save_db, save_user +from app.face.face_embedding import extract_embedding, verify_embedding + def register_user(user_id: str, image_bytes: bytes) -> dict: """이미지와 ID로 회원 등록""" - pass + image_np = np.frombuffer(image_bytes, dtype=np.uint8) + image = cv2.imdecode(image_np, cv2.IMREAD_COLOR) + embedding = extract_embedding(image) + save_user(user_id, embedding) + return {"user_id": user_id, "registered_at": str(datetime.now())} def authenticate_user(image_bytes: bytes): """이미지로 회원 인증""" - pass + image_np = np.frombuffer(image_bytes, dtype=np.uint8) + image = cv2.imdecode(image_np, cv2.IMREAD_COLOR) + db = load_db() + + embedding_to_check = extract_embedding(image) + + for user_id, data in db.items(): + if verify_embedding(data["embedding"], embedding_to_check): + return user_id + + return None def get_user(user_id): """user_id로 회원 정보 조회""" - pass + db = load_db() + user_data = db.get(user_id) + + if user_data: + return {"user_id": user_id, "registered_at": user_data["registered_at"]} + else: + return None def delete_user(user_id): """user_id로 회원 삭제""" - pass + db = load_db() + if user_id in db: + del db[user_id] + save_db(db) + return True + return False diff --git a/tests/conftest.py b/tests/conftest.py index e69de29..fcfb780 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -0,0 +1,22 @@ +import pytest +from fastapi.testclient import TestClient + +from app.face.face_db import save_user +from app.face.face_embedding import extract_embedding +from app.main import app + + +@pytest.fixture(scope="module") +def client(): + with TestClient(app) as c: + yield c + + +@pytest.fixture(scope="module") +def setup_user_db(): + image_paths = ["images/Aaron_Peirsol/Aaron_Peirsol_0001.jpg"] + user_ids = ["aaron_peirsol"] + for image_path, user_id in zip(image_paths, user_ids): + embedding = extract_embedding(image_path) + save_user(user_id, embedding) + yield \ No newline at end of file diff --git a/tests/test_api_users.py b/tests/test_api_users.py index e69de29..b9ca6ce 100644 --- a/tests/test_api_users.py +++ b/tests/test_api_users.py @@ -0,0 +1,51 @@ +def test_register_user_api(client): + image_path = "images/Aaron_Peirsol/Aaron_Peirsol_0001.jpg" + user_id = "aaron_peirsol" + + with open(image_path, "rb") as img_file: + response = client.post( + "/users/register", + data={"user_id": user_id}, + files={"image": ("user1.jpg", img_file, "image/jpeg")}, + ) + + assert response.status_code == 200 + data = response.json() + assert data["user_id"] == user_id + assert data["registered_at"] is not None + + +def test_authenticate_user_api(client, setup_user_db): + image_path = "images/Aaron_Peirsol/Aaron_Peirsol_0001.jpg" + + with open(image_path, "rb") as img_file: + response = client.post( + "/users/authenticate", + files={"image": ("user1.jpg", img_file, "image/jpeg")}, + ) + + assert response.status_code == 200 + data = response.json() + assert data["user_id"] == "aaron_peirsol" + + +def test_get_registered_user_api(client, setup_user_db): + user_id = "aaron_peirsol" + + response = client.get(f"/users/{user_id}") + + assert response.status_code == 200 + data = response.json() + assert data["user_id"] == user_id + assert "registered_at" in data + + +def test_delete_registered_user_api(client, setup_user_db): + user_id = "aaron_peirsol" + + response = client.delete(f"/users/{user_id}") + + assert response.status_code == 200 + + response = client.get(f"/users/{user_id}") + assert response.status_code == 404 \ No newline at end of file diff --git a/tests/test_face.py b/tests/test_face.py index e69de29..bcd5971 100644 --- a/tests/test_face.py +++ b/tests/test_face.py @@ -0,0 +1,32 @@ +import pytest + +from app.face.face_embedding import extract_embedding, verify_embedding + + +def test_extract_embedding(): + image_path = "images/Aaron_Peirsol/Aaron_Peirsol_0001.jpg" + + embedding = extract_embedding(image_path) + + assert embedding is not None + assert len(embedding) > 0 + + +def test_no_face_exception(): + image_path = "tests/images/no_face.jpg" + + with pytest.raises(ValueError): + extract_embedding(image_path) + + +def test_verify_embedding(): + image_path1 = "images/Aaron_Peirsol/Aaron_Peirsol_0001.jpg" + image_path2 = "images/Aaron_Peirsol/Aaron_Peirsol_0002.jpg" + image_path3 = "images/Olivia_Newton-John/Olivia_Newton-John_0001.jpg" + + embedding1 = extract_embedding(image_path1) + embedding2 = extract_embedding(image_path2) + embedding3 = extract_embedding(image_path3) + + assert verify_embedding(embedding1, embedding2) + assert not verify_embedding(embedding1, embedding3) \ No newline at end of file diff --git a/tests/test_service_users.py b/tests/test_service_users.py index e69de29..518955b 100644 --- a/tests/test_service_users.py +++ b/tests/test_service_users.py @@ -0,0 +1,66 @@ +from app.services.user_service import ( + authenticate_user, + delete_user, + get_user, + register_user, +) + + +def test_register_user(): + image_path = "images/Aaron_Peirsol/Aaron_Peirsol_0001.jpg" + user_id = "aaron_peirsol" + + with open(image_path, "rb") as f: + image_data = f.read() + result = register_user(user_id, image_data) + + assert result["user_id"] == "aaron_peirsol" + assert result["registered_at"] is not None + + +def test_authenticate_registered_user(setup_user_db): + image_path = "images/Aaron_Peirsol/Aaron_Peirsol_0001.jpg" + + with open(image_path, "rb") as f: + image_data = f.read() + user_id = authenticate_user(image_data) + + assert user_id == "aaron_peirsol" + + +def test_authenticate_unregistered_user(setup_user_db): + image_path = "images/Natasha_McElhone/Natasha_McElhone_0001.jpg" + with open(image_path, "rb") as f: + image_data = f.read() + + user_id = authenticate_user(image_data) + + assert user_id is None + + +def test_get_registered_user(setup_user_db): + user_id = "aaron_peirsol" + + user_info = get_user(user_id) + + assert user_info["user_id"] == user_id + assert "registered_at" in user_info + + +def test_delete_registered_user(setup_user_db): + user_id = "aaron_peirsol" + + result = delete_user(user_id) + + assert result is True + + user_info = get_user(user_id) + assert user_info is None + + +def test_get_unregistered_user(setup_user_db): + user_id = "userA" + + user_info = get_user(user_id) + + assert user_info is None \ No newline at end of file