feat: initial file

Dockerfile

@@ -0,0 +1,39 @@
+FROM python:3.9
+
+# 캐시 디렉토리 환경변수 설정
+ENV TRANSFORMERS_CACHE=/tmp/huggingface_cache
+ENV HF_HOME=/tmp/huggingface_cache
+ENV PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+# 시스템 패키지 설치
+RUN apt-get update && apt-get install -y \
+    build-essential \
+    libgl1-mesa-glx \
+    libglib2.0-0 \
+    && rm -rf /var/lib/apt/lists/*
+
+# 캐시 디렉토리 생성 및 권한 설정
+RUN mkdir -p $TRANSFORMERS_CACHE && chmod -R 777 $TRANSFORMERS_CACHE
+
+# 임시 디렉토리 권한 설정
+RUN mkdir -p /tmp/uploads && chmod 777 /tmp/uploads
+ENV TMPDIR=/tmp/uploads
+
+# 요구사항 파일 복사 및 설치
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# 애플리케이션 파일 복사
+COPY . .
+
+# 환경 변수 설정
+ENV PYTHONPATH=/app
+
+# kiwipiepy 초기화 파일 다운로드 - 사전 다운로드 문제 해결
+RUN python -c "from kiwipiepy import Kiwi; Kiwi()"
+
+# 애플리케이션 실행 (포트 7861로 변경)
+EXPOSE 7861
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7861"]

api/routes/__init__.py

@@ -0,0 +1,6 @@
+"""
+API 라우트 패키지
+"""
+from .matching_routers import router as matching_router
+
+__all__ = ['matching_router']

api/routes/matching_routers.py

@@ -0,0 +1,79 @@
+"""
+습득물 매칭 관련 API 라우트
+"""
+import os
+import sys
+import logging
+from typing import List, Dict, Any, Optional
+from fastapi import APIRouter, Depends, HTTPException, Query, Body
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field, validator
+import base64
+from io import BytesIO
+from PIL import Image
+
+# 로깅 설정
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+# 라우터 생성
+router = APIRouter(
+    prefix="/api/matching",
+    tags=["matching"],
+    responses={404: {"description": "Not found"}},
+)
+
+# Pydantic 모델 정의
+class LostItemPost(BaseModel):
+    """사용자가 분실한 물품 게시글 모델"""
+    category: str = Field(..., description="분실물 카테고리 (예: 지갑, 가방, 전자기기)")
+    item_name: str = Field(..., description="물품명 (예: 검은색 가죽 지갑)")
+    color: Optional[str] = Field(None, description="물품 색상")
+    content: str = Field(..., description="게시글 내용")
+    location: Optional[str] = Field(None, description="분실 장소")
+    image_url: Optional[str] = Field(None, description="이미지 URL (있는 경우)")
+    lost_items: Optional[List[Dict[str, Any]]] = Field(None, description="비교할 습득물 데이터 (API 테스트용)")
+    
+    class Config:
+        schema_extra = {
+            "example": {
+                "category": "지갑",
+                "item_name": "검은색 가죽 지갑",
+                "color": "검정색",
+                "content": "지난주 토요일 강남역 근처에서 검정색 가죽 지갑을 잃어버렸습니다. 현금과 카드가 들어있어요.",
+                "location": "강남역",
+                "image_url": None
+            }
+        }
+
+class ImageMatchingRequest(BaseModel):
+    """이미지 기반 매칭 요청 모델"""
+    category: Optional[str] = Field(None, description="분실물 카테고리")
+    item_name: Optional[str] = Field(None, description="물품명")
+    color: Optional[str] = Field(None, description="색상")
+    content: Optional[str] = Field(None, description="내용")
+    image_base64: Optional[str] = Field(None, description="Base64 인코딩된 이미지")
+    lost_items: Optional[List[Dict[str, Any]]] = Field(None, description="비교할 습득물 데이터 (API 테스트용)")
+    
+    class Config:
+        schema_extra = {
+            "example": {
+                "category": "지갑",
+                "item_name": "검은색 가죽 지갑",
+                "color": "검정색",
+                "content": "지난주 토요일 강남역 근처에서 검정색 가죽 지갑을 잃어버렸습니다.",
+                "image_base64": "[base64 encoded image string]"
+            }
+        }
+
+class MatchingResult(BaseModel):
+    """매칭 결과 모델"""
+    total_matches: int = Field(..., description="매칭된 항목 수")
+    similarity_threshold: float = Field(..., description="유사도 임계값")
+    matches: List[Dict[str, Any]] = Field(..., description="매칭된 항목 목록")
+
+class MatchingResponse(BaseModel):
+    """API 응답 모델"""
+    success: bool = Field(..., description="요청 성공 여부")
+    message: str = Field(..., description="응답 메시지")
+    result: Optional[MatchingResult] = Field(None, description="매칭 결과")

main.py

@@ -0,0 +1,221 @@
+"""
+FastAPI 애플리케이션 메인 모듈
+"""
+import os
+import sys
+import logging
+import tempfile
+from fastapi import FastAPI, Request, HTTPException, Query, Body
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from typing import List, Dict, Any, Optional, Union
+import json
+import base64
+from io import BytesIO
+from PIL import Image
+
+# 캐시 디렉토리 설정
+os.environ['TRANSFORMERS_CACHE'] = '/tmp/huggingface_cache'
+os.environ['HF_HOME'] = '/tmp/huggingface_cache'
+
+# 디렉토리 생성
+os.makedirs('/tmp/huggingface_cache', exist_ok=True)
+os.makedirs('/tmp/uploads', exist_ok=True)
+
+# 로깅 설정
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+# 필요한 모듈 임포트
+from models.clip_model import KoreanCLIPModel
+from utils.similarity import calculate_similarity, find_similar_items
+from api.routes.matching_routers import LostItemPost, ImageMatchingRequest, MatchingResult, MatchingResponse
+
+# 모델 초기화 (싱글톤으로 로드)
+clip_model = None
+
+def get_clip_model():
+    """
+    한국어 CLIP 모델 인스턴스를 반환 (싱글톤 패턴)
+    """
+    global clip_model
+    if clip_model is None:
+        try:
+            clip_model = KoreanCLIPModel()
+            return clip_model
+        except Exception as e:
+            logger.error(f"CLIP 모델 초기화 실패: {str(e)}")
+            # 실패 시 None 반환 (텍스트 기반 매칭만 가능)
+            return None
+    return clip_model
+
+# FastAPI 애플리케이션 생성
+app = FastAPI(
+    title="습득물 유사도 검색 API",
+    description="한국어 CLIP 모델을 사용하여 사용자 게시글과 습득물 간의 유사도를 계산하는 API",
+    version="1.0.0"
+)
+
+# CORS 미들웨어 설정
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 전역 예외 처리
+@app.exception_handler(Exception)
+async def global_exception_handler(request: Request, exc: Exception):
+    """
+    전역 예외 처리기
+    """
+    logger.error(f"요청 처리 중 예외 발생: {str(exc)}")
+    return JSONResponse(
+        status_code=500,
+        content={"success": False, "message": f"서버 오류가 발생했습니다: {str(exc)}"}
+    )
+
+# API 엔드포인트 정의
+@app.post("/api/matching/find-similar", response_model=MatchingResponse)
+async def find_similar_items_api(
+    request: Union[LostItemPost, ImageMatchingRequest],
+    threshold: float = Query(0.7, description="유사도 임계값 (0.0 ~ 1.0)"),
+    limit: int = Query(10, description="반환할 최대 항목 수")
+):
+    """
+    사용자 게시글과 유사한 습득물을 찾는 API 엔드포인트
+    
+    Args:
+        request: 사용자의 분실물 게시글 또는 이미지 매칭 요청
+        threshold: 유사도 임계값
+        limit: 반환할 최대 항목 수
+        
+    Returns:
+        MatchingResponse: 매칭 결과가 포함된 응답
+    """
+    try:
+        logger.info(f"유사 습득물 검색 요청: threshold={threshold}, limit={limit}")
+        
+        # 요청 데이터 변환
+        user_post = {}
+        if isinstance(request, LostItemPost):
+            user_post = request.dict()
+        else:
+            user_post = request.dict()
+            # Base64 이미지가 있으면 이미지 처리 로직 추가
+            if user_post.get("image_base64"):
+                try:
+                    # Base64 이미지 디코딩
+                    base64_str = user_post["image_base64"]
+                    
+                    # Base64 문자열에서 헤더 제거 (있을 경우)
+                    if "," in base64_str:
+                        base64_str = base64_str.split(",")[1]
+                    
+                    image_data = base64.b64decode(base64_str)
+                    image = Image.open(BytesIO(image_data))
+                    
+                    # 이미지 사용 (CLIP 모델에 전달)
+                    user_post["image"] = image
+                    
+                    logger.info("Base64 이미지 처리 완료")
+                except Exception as e:
+                    logger.error(f"Base64 이미지 처리 실패: {str(e)}")
+        
+        # 여기서 DB 대신 요청에서 전달된 습득물 데이터를 사용합니다.
+        lost_items = []
+        
+        # 요청에 습득물 데이터가 있으면 사용
+        if hasattr(request, 'lost_items') and request.lost_items:
+            lost_items = request.lost_items
+        
+        if not lost_items:
+            return MatchingResponse(
+                success=False,
+                message="습득물 데이터가 없습니다. 요청에 습득물 데이터를 포함해주세요.",
+                result=None
+            )
+        
+        # CLIP 모델 로드
+        clip_model_instance = get_clip_model()
+        
+        # 유사한 항목 찾기
+        similar_items = find_similar_items(user_post, lost_items, threshold, clip_model_instance)
+        
+        # 결과 제한
+        similar_items = similar_items[:limit]
+        
+        # 응답 구성
+        result = MatchingResult(
+            total_matches=len(similar_items),
+            similarity_threshold=threshold,
+            matches=[
+                {
+                    "item": item["item"],
+                    "similarity": round(item["similarity"], 4),
+                    "details": {
+                        "text_similarity": round(item["details"]["text_similarity"], 4),
+                        "image_similarity": round(item["details"]["image_similarity"], 4) if item["details"]["image_similarity"] else None,
+                        "category_similarity": round(item["details"]["details"]["category"], 4),
+                        "item_name_similarity": round(item["details"]["details"]["item_name"], 4),
+                        "color_similarity": round(item["details"]["details"]["color"], 4),
+                        "content_similarity": round(item["details"]["details"]["content"], 4)
+                    }
+                }
+                for item in similar_items
+            ]
+        )
+        
+        return MatchingResponse(
+            success=True,
+            message=f"{len(similar_items)}개의 유사한 습득물을 찾았습니다.",
+            result=result
+        )
+    
+    except Exception as e:
+        logger.error(f"API 호출 중 오류 발생: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"요청 처리 중 오류가 발생했습니다: {str(e)}")
+
+@app.get("/api/matching/test")
+async def test_endpoint():
+    """
+    API 테스트용 엔드포인트
+    
+    Returns:
+        dict: 테스트 응답
+    """
+    return {"message": "API가 정상적으로 작동 중입니다."}
+
+@app.get("/api/status")
+async def status():
+    """
+    API 상태 엔드포인트
+    
+    Returns:
+        dict: API 상태 정보
+    """
+    # CLIP 모델 로드 시도
+    model = get_clip_model()
+    
+    return {
+        "status": "ok",
+        "models_loaded": model is not None,
+        "version": "1.0.0"
+    }
+
+# 루트 엔드포인트
+@app.get("/")
+async def root():
+    """
+    루트 엔드포인트 - API 정보 제공
+    """
+    return {
+        "app_name": "습득물 유사도 검색 API",
+        "version": "1.0.0",
+        "description": "한국어 CLIP 모델을 사용하여 사용자 게시글과 습득물 간의 유사도를 계산합니다.",
+        "api_endpoint": "/api/matching/find-similar",
+        "test_endpoint": "/api/matching/test",
+        "status_endpoint": "/api/status"
+    }

models/__init__.py

@@ -0,0 +1,6 @@
+"""
+모델 관련 모듈 패키지
+"""
+from .clip_model import KoreanCLIPModel
+
+__all__ = ['KoreanCLIPModel']

models/clip_model.py

@@ -0,0 +1,178 @@
+"""
+한국어 CLIP 모델 구현
+이 모듈은 HuggingFace의 한국어 CLIP 모델을 사용하여 텍스트와 이미지의 임베딩을 생성
+"""
+import os
+import sys
+import logging
+import torch
+from transformers import CLIPProcessor, CLIPModel
+from PIL import Image
+import requests
+from io import BytesIO
+import numpy as np
+
+# 캐시 디렉토리 설정
+os.environ['TRANSFORMERS_CACHE'] = '/tmp/huggingface_cache'
+os.environ['HF_HOME'] = '/tmp/huggingface_cache'
+
+# 디렉토리 생성
+os.makedirs('/tmp/huggingface_cache', exist_ok=True)
+os.makedirs('/tmp/uploads', exist_ok=True)
+
+# 로깅 설정
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+# 모델 설정 - 환경 변수에서 가져오거나 기본값 사용
+CLIP_MODEL_NAME = os.getenv('CLIP_MODEL_NAME', 'Bingsu/clip-vit-large-patch14-ko')
+DEVICE = "cuda" if torch.cuda.is_available() and os.getenv('USE_GPU', 'True').lower() == 'true' else "cpu"
+
+class KoreanCLIPModel:
+    """
+    한국어 CLIP 모델 클래스
+    텍스트와 이미지를 임베딩하고 유사도를 계산하는 기능 제공
+    """
+    
+    def __init__(self, model_name=CLIP_MODEL_NAME, device=DEVICE):
+        """
+        CLIP 모델 초기화
+        
+        Args:
+            model_name (str): 사용할 CLIP 모델 이름 또는 경로
+            device (str): 사용할 장치 ('cuda' 또는 'cpu')
+        """
+        self.device = device
+        self.model_name = model_name
+        
+        logger.info(f"CLIP 모델 '{model_name}' 로드 중 (device: {device})...")
+        
+        try:
+            # 캐시 디렉토리 설정
+            os.environ["TRANSFORMERS_CACHE"] = "/tmp/transformers_cache"
+            os.makedirs("/tmp/transformers_cache", exist_ok=True)
+            
+            self.model = CLIPModel.from_pretrained(model_name).to(device)
+            self.processor = CLIPProcessor.from_pretrained(model_name)
+            logger.info("CLIP 모델 로드 완료")
+        except Exception as e:
+            logger.error(f"CLIP 모델 로드 실패: {str(e)}")
+            raise
+            
+    def encode_text(self, text):
+        """
+        텍스트를 임베딩 벡터로 변환
+        
+        Args:
+            text (str or list): 인코딩할 텍스트 또는 텍스트 리스트
+            
+        Returns:
+            numpy.ndarray: 임베딩 벡터
+        """
+        if isinstance(text, str):
+            text = [text]
+            
+        try:
+            with torch.no_grad():
+                # 텍스트 인코딩
+                inputs = self.processor(text=text, return_tensors="pt", padding=True, truncation=True).to(self.device)
+                text_features = self.model.get_text_features(**inputs)
+                
+                # 텍스트 특성 정규화
+                text_embeddings = text_features / text_features.norm(dim=1, keepdim=True)
+                
+            return text_embeddings.cpu().numpy()
+        except Exception as e:
+            logger.error(f"텍스트 인코딩 중 오류 발생: {str(e)}")
+            return np.zeros((len(text), self.model.text_embed_dim))
+            
+    def encode_image(self, image_source):
+        """
+        이미지를 임베딩 벡터로 변환
+        
+        Args:
+            image_source: 인코딩할 이미지 (PIL Image, URL 또는 이미지 경로)
+            
+        Returns:
+            numpy.ndarray: 임베딩 벡터
+        """
+        try:
+            # 이미지 로드 (URL, 파일 경로, PIL 이미지 객체 또는 Base64)
+            if isinstance(image_source, str):
+                if image_source.startswith('http'):
+                    # URL에서 이미지 로드
+                    response = requests.get(image_source)
+                    image = Image.open(BytesIO(response.content)).convert('RGB')
+                else:
+                    # 로컬 파일에서 이미지 로드
+                    image = Image.open(image_source).convert('RGB')
+            else:
+                # 이미 PIL 이미지 객체인 경우
+                image = image_source.convert('RGB')
+                
+            with torch.no_grad():
+                # 이미지 인코딩
+                inputs = self.processor(images=image, return_tensors="pt").to(self.device)
+                image_features = self.model.get_image_features(**inputs)
+                
+                # 이미지 특성 정규화
+                image_embeddings = image_features / image_features.norm(dim=1, keepdim=True)
+                
+            return image_embeddings.cpu().numpy()
+        except Exception as e:
+            logger.error(f"이미지 인코딩 중 오류 발생: {str(e)}")
+            return np.zeros((1, self.model.vision_embed_dim))
+    
+    def calculate_similarity(self, text_embedding, image_embedding=None):
+        """
+        텍스트와 이미지 임베딩 간의 유사도 계산
+        
+        Args:
+            text_embedding (numpy.ndarray): 텍스트 임베딩
+            image_embedding (numpy.ndarray, optional): 이미지 임베딩 (없으면 텍스트만 비교)
+            
+        Returns:
+            float: 유사도 점수 (0~1 사이)
+        """
+        if image_embedding is None:
+            # 텍스트-텍스트 유사도 계산 (코사인 유사도)
+            similarity = np.dot(text_embedding, text_embedding.T)[0, 0]
+        else:
+            # 텍스트-이미지 유사도 계산 (코사인 유사도)
+            similarity = np.dot(text_embedding, image_embedding.T)[0, 0]
+            
+        # 유사도를 0~1 범위로 정규화
+        similarity = (similarity + 1) / 2
+        return float(similarity)
+        
+    def encode_batch_texts(self, texts):
+        """
+        여러 텍스트를 한 번에 임베딩
+        
+        Args:
+            texts (list): 텍스트 목록
+            
+        Returns:
+            numpy.ndarray: 임베딩 벡터 배열
+        """
+        # 배치 처리를 위한 코드
+        # 실제 구현에서는 메모리 크기에 따라 적절한 배치 크기 조정 필요
+        return self.encode_text(texts)
+
+# 모듈 테스트용 코드
+if __name__ == "__main__":
+    # 모델 초기화
+    clip_model = KoreanCLIPModel()
+    
+    # 샘플 텍스트 인코딩
+    sample_text = "검은색 지갑을 잃어버렸습니다. 현금과 카드가 들어있어요."
+    text_embedding = clip_model.encode_text(sample_text)
+    
+    print(f"텍스트 임베딩 shape: {text_embedding.shape}")
+    
+    # 유사도 계산 (텍스트만)
+    sample_text2 = "검은색 지갑을 찾았습니다. 안에 현금과 카드가 있습니다."
+    text_embedding2 = clip_model.encode_text(sample_text2)
+    
+    similarity = clip_model.calculate_similarity(text_embedding, text_embedding2)
+    print(f"텍스트 간 유사도: {similarity:.4f}")

requirements.txt

@@ -0,0 +1,13 @@
+fastapi
+uvicorn
+transformers==4.26.0
+torch
+Pillow
+pydantic
+python-multipart
+httpx
+python-dotenv
+aiofiles
+kiwipiepy==0.20.4
+numpy
+requests

utils/__init__.py

@@ -0,0 +1,6 @@
+"""
+유틸리티 관련 모듈 패키지
+"""
+from .similarity import calculate_similarity, find_similar_items
+
+__all__ = ['calculate_similarity', 'find_similar_items']

utils/similarity.py

@@ -0,0 +1,355 @@
+"""
+유사도 계산 및 관련 유틸리티 함수
+Kiwi 형태소 분석기를 사용하여 한국어 텍스트 분석 개선
+"""
+import os
+import sys
+import logging
+import numpy as np
+import re
+from collections import Counter
+from kiwipiepy import Kiwi
+
+# 로깅 설정
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+# Kiwi 형태소 분석기 초기화
+kiwi = Kiwi()
+
+# 설정 값 (환경변수 또는 기본값)
+SIMILARITY_THRESHOLD = float(os.getenv('SIMILARITY_THRESHOLD', '0.7'))
+TEXT_WEIGHT = float(os.getenv('TEXT_WEIGHT', '0.7'))
+IMAGE_WEIGHT = float(os.getenv('IMAGE_WEIGHT', '0.3'))
+CATEGORY_WEIGHT = float(os.getenv('CATEGORY_WEIGHT', '0.5'))
+ITEM_NAME_WEIGHT = float(os.getenv('ITEM_NAME_WEIGHT', '0.3'))
+COLOR_WEIGHT = float(os.getenv('COLOR_WEIGHT', '0.1'))
+CONTENT_WEIGHT = float(os.getenv('CONTENT_WEIGHT', '0.1'))
+
+def preprocess_text(text):
+    """
+    텍스트 전처리 함수
+    
+    Args:
+        text (str): 전처리할 텍스트
+        
+    Returns:
+        str: 전처리된 텍스트
+    """
+    if not text:
+        return ""
+        
+    # 소문자 변환 (영어의 경우)
+    text = text.lower()
+    
+    # 불필요한 공백 제거
+    text = re.sub(r'\s+', ' ', text).strip()
+    
+    # 특수 문자 제거 (단, 한글, 영문, 숫자는 유지)
+    text = re.sub(r'[^\w\s가-힣ㄱ-ㅎㅏ-ㅣ]', ' ', text)
+    
+    return text
+
+def extract_keywords(text):
+    """
+    Kiwi 형태소 분석기를 사용하여 텍스트에서 중요 키워드 추출
+    
+    Args:
+        text (str): 키워드를 추출할 텍스트
+        
+    Returns:
+        list: 키워드 리스트 (주로 명사와 형용사)
+    """
+    if not text:
+        return []
+    
+    # 텍스트 전처리
+    processed_text = preprocess_text(text)
+    
+    try:
+        # Kiwi 형태소 분석 수행
+        result = kiwi.analyze(processed_text)
+        
+        # 중요 키워드 추출 (명사, 형용사 등)
+        keywords = []
+        for token in result[0][0]:
+            # NNG: 일반명사, NNP: 고유명사, VA: 형용사, VV: 동사, SL: 외국어(영어 등)
+            if token.tag in ['NNG', 'NNP', 'VA', 'SL']:
+                # 한 글자 명사는 중요도 낮을 수 있어 필터링 (선택적)
+                if len(token.form) > 1 or token.tag in ['SL']:
+                    keywords.append(token.form)
+        
+        logger.debug(f"키워드 추출 결과: {keywords}")
+        return keywords
+    
+    except Exception as e:
+        logger.warning(f"형태소 분석 오류: {str(e)}, 기본 분리 방식으로 대체")
+        # 오류 발생 시 기본 방식으로 대체
+        words = processed_text.split()
+        return words
+
+def calculate_text_similarity(text1, text2, weights=None):
+    """
+    두 텍스트 간의 유사도 계산 (Kiwi 형태소 분석 활용)
+    
+    Args:
+        text1 (str): 첫 번째 텍스트
+        text2 (str): 두 번째 텍스트
+        weights (dict, optional): 각 부분에 대한 가중치
+        
+    Returns:
+        float: 유사도 점수 (0~1 사이)
+    """
+    if not text1 or not text2:
+        return 0.0
+    
+    # 기본 가중치 설정
+    if weights is None:
+        weights = {
+            'common_words': 0.7,  # 공통 단어 비율의 가중치 증가
+            'length_ratio': 0.15,
+            'word_order': 0.15
+        }
+    
+    # 텍스트에서 키워드 추출 (Kiwi 형태소 분석기 사용)
+    keywords1 = extract_keywords(text1)
+    keywords2 = extract_keywords(text2)
+    
+    if not keywords1 or not keywords2:
+        return 0.0
+    
+    # 1. 공통 단어 비율 계산
+    common_words = set(keywords1) & set(keywords2)
+    common_ratio = len(common_words) / max(1, min(len(set(keywords1)), len(set(keywords2))))
+    
+    # 2. 텍스트 길이 유사도
+    length_ratio = min(len(keywords1), len(keywords2)) / max(1, max(len(keywords1), len(keywords2)))
+    
+    # 3. 단어 순서 유사도 (선택적)
+    word_order_sim = 0.0
+    if common_words:
+        # 공통 단어의 위치 차이 기반 유사도
+        positions1 = {word: i for i, word in enumerate(keywords1) if word in common_words}
+        positions2 = {word: i for i, word in enumerate(keywords2) if word in common_words}
+        
+        if positions1 and positions2:
+            pos_diff_sum = sum(abs(positions1[word] - positions2[word]) for word in common_words if word in positions1 and word in positions2)
+            max_diff = len(keywords1) + len(keywords2)
+            word_order_sim = 1.0 - (pos_diff_sum / max(1, max_diff))
+    
+    # 가중치 적용하여 최종 유사도 계산
+    similarity = (
+        weights['common_words'] * common_ratio + 
+        weights['length_ratio'] * length_ratio + 
+        weights['word_order'] * word_order_sim
+    )
+    
+    return min(1.0, max(0.0, similarity))
+
+def calculate_category_similarity(category1, category2):
+    """
+    두 카테고리 간의 유사도 계산 (기타 카테고리 고려)
+    
+    Args:
+        category1 (str): 첫 번째 카테고리
+        category2 (str): 두 번째 카테고리
+        
+    Returns:
+        float: 유사도 점수 (0~1 사이)
+    """
+    if not category1 or not category2:
+        return 0.0
+    
+    # 카테고리 전처리
+    cat1 = preprocess_text(str(category1))
+    cat2 = preprocess_text(str(category2))
+    
+    # 정확히 일치하는 경우
+    if cat1 == cat2:
+        return 1.0
+    
+    # Kiwi로 키워드 추출
+    keywords1 = set(extract_keywords(cat1))
+    keywords2 = set(extract_keywords(cat2))
+    
+    # '기타' 카테고리 처리
+    if '기타' in cat1 or '기타' in cat2:
+        # 키워드 추출 및 교집합 비교
+        if not keywords1 or not keywords2:
+            return 0.3  # 기타 카테고리는 기본 유사도 부여
+        
+        # 교집합 단어가 있으면 높은 유사도
+        common_words = keywords1 & keywords2
+        if common_words:
+            return 0.7
+        
+        return 0.3  # 기타 카테고리지만 공통 키워드 없음
+    
+    # 일반 카테고리 유사도
+    return calculate_text_similarity(cat1, cat2)
+
+def calculate_similarity(user_post, lost_item, clip_model=None):
+    """
+    사용자 게시글과 습득물 항목 간의 종합 유사도 계산
+    
+    Args:
+        user_post (dict): 사용자 게시글 정보
+        lost_item (dict): 습득물 데이터
+        clip_model (KoreanCLIPModel, optional): CLIP 모델 인스턴스
+        
+    Returns:
+        float: 유사도 점수 (0~1 사이)
+        dict: 세부 유사도 정보
+    """
+    # 텍스트 유사도 계산
+    text_similarities = {}
+    
+    # 1. 카테고리 유사도
+    category_sim = 0.0
+    if 'category' in user_post and 'category' in lost_item:
+        category_sim = calculate_category_similarity(user_post['category'], lost_item['category'])
+    text_similarities['category'] = category_sim
+    
+    # 2. 물품명 유사도
+    item_name_sim = 0.0
+    user_item_name = user_post.get('item_name', '')
+    lost_item_name = lost_item.get('item_name', '')
+    if user_item_name and lost_item_name:
+        item_name_sim = calculate_text_similarity(user_item_name, lost_item_name)
+    text_similarities['item_name'] = item_name_sim
+    
+    # 3. 색상 유사도
+    color_sim = 0.0
+    user_color = user_post.get('color', '')
+    lost_color = lost_item.get('color', '')
+    if user_color and lost_color:
+        color_sim = calculate_text_similarity(user_color, lost_color)
+    text_similarities['color'] = color_sim
+    
+    # 4. 내용 유사도
+    content_sim = 0.0
+    user_content = user_post.get('content', '')
+    lost_content = lost_item.get('content', '')
+    if user_content and lost_content:
+        content_sim = calculate_text_similarity(user_content, lost_content)
+    text_similarities['content'] = content_sim
+    
+    # 텍스트 종합 유사도 계산 (가중치 적용)
+    text_similarity = (
+        CATEGORY_WEIGHT * category_sim +
+        ITEM_NAME_WEIGHT * item_name_sim +
+        COLOR_WEIGHT * color_sim +
+        CONTENT_WEIGHT * content_sim
+    )
+    
+    # CLIP 모델을 사용한 이미지-텍스트 유사도 계산
+    image_similarity = 0.0
+    has_image = False
+    
+    if clip_model is not None:
+        # 사용자 게시글과 습득물에 모두 이미지가 있는 경우
+        user_image = user_post.get('image', None) or user_post.get('image_url', None)
+        lost_image = lost_item.get('image', None) or lost_item.get('image_url', None)
+        
+        if user_image and lost_image:
+            try:
+                # CLIP 모델을 사용한 유사도 계산
+                user_text_embedding = clip_model.encode_text(user_post.get('content', ''))
+                user_image_embedding = clip_model.encode_image(user_image)
+                
+                item_text_embedding = clip_model.encode_text(lost_item.get('content', ''))
+                item_image_embedding = clip_model.encode_image(lost_image)
+                
+                # 텍스트-이미지 교차 유사도 계산
+                text_to_image_sim = clip_model.calculate_similarity(user_text_embedding, item_image_embedding)
+                image_to_text_sim = clip_model.calculate_similarity(item_text_embedding, user_image_embedding)
+                image_to_image_sim = clip_model.calculate_similarity(user_image_embedding, item_image_embedding)
+                
+                image_similarity = (text_to_image_sim + image_to_text_sim + image_to_image_sim) / 3
+                has_image = True
+            except Exception as e:
+                logger.warning(f"이미지 유사도 계산 중 오류 발생: {str(e)}")
+    
+    # 최종 유사도 계산 (텍스트와 이미지 가중치 적용)
+    if has_image:
+        final_similarity = TEXT_WEIGHT * text_similarity + IMAGE_WEIGHT * image_similarity
+    else:
+        final_similarity = text_similarity
+    
+    # 세부 유사도 정보
+    similarity_details = {
+        'text_similarity': text_similarity,
+        'image_similarity': image_similarity if has_image else None,
+        'final_similarity': final_similarity,
+        'details': text_similarities
+    }
+    
+    return final_similarity, similarity_details
+
+def find_similar_items(user_post, lost_items, threshold=SIMILARITY_THRESHOLD, clip_model=None):
+    """
+    사용자 게시글과 유사한 습득물 목록 찾기
+    
+    Args:
+        user_post (dict): 사용자 게시글 정보
+        lost_items (list): 습득물 데이터 목록
+        threshold (float): 유사도 임계값 (기본값: config에서 설정)
+        clip_model (KoreanCLIPModel, optional): CLIP 모델 인스턴스
+        
+    Returns:
+        list: 유사도가 임계값 이상인 습득물 목록 (유사도 높은 순)
+    """
+    similar_items = []
+    
+    logger.info(f"사용자 게시글과 {len(lost_items)}개 습득물 비교 중...")
+    
+    for item in lost_items:
+        similarity, details = calculate_similarity(user_post, item, clip_model)
+        
+        if similarity >= threshold:
+            similar_items.append({
+                'item': item,
+                'similarity': similarity,
+                'details': details
+            })
+    
+    # 유사도 높은 순으로 정렬
+    similar_items.sort(key=lambda x: x['similarity'], reverse=True)
+    
+    logger.info(f"유사도 {threshold} 이상인 습득물 {len(similar_items)}개 발견")
+    
+    return similar_items
+
+# 모듈 테스트용 코드
+if __name__ == "__main__":
+    # 텍스트 유사도 테스트
+    text1 = "검은색 가죽 지갑을 잃어버렸습니다."
+    text2 = "검정 가죽 지갑을 찾았습니다."
+    text3 = "노트북을 분실했습니다."
+    
+    # 키워드 추출 테스트
+    print("[ 키워드 추출 테스트 ]")
+    print(f"텍스트 1: '{text1}'")
+    print(f"추출된 키워드: {extract_keywords(text1)}")
+    print(f"텍스트 2: '{text2}'")
+    print(f"추출된 키워드: {extract_keywords(text2)}")
+    
+    # 유사도 테스트
+    sim12 = calculate_text_similarity(text1, text2)
+    sim13 = calculate_text_similarity(text1, text3)
+    
+    print("\n[ 유사도 테스트 ]")
+    print(f"텍스트 1-2 유사도: {sim12:.4f}")
+    print(f"텍스트 1-3 유사도: {sim13:.4f}")
+    
+    # 카테고리 유사도 테스트
+    cat1 = "지갑"
+    cat2 = "가방/지갑"
+    cat3 = "기타"
+    
+    cat_sim12 = calculate_category_similarity(cat1, cat2)
+    cat_sim13 = calculate_category_similarity(cat1, cat3)
+    
+    print("\n[ 카테고리 유사도 테스트 ]")
+    print(f"카테고리 1-2 유사도: {cat_sim12:.4f}")
+    print(f"카테고리 1-3 유사도: {cat_sim13:.4f}")