백엔드와+실시간 검색어 기능(9)

실시간 검색어 기능과 사용자가 엔진에 직접 접근하는 것을 막기위해서 중간에 백엔드를 생성한다 여기서는 python으로 구축된 Fast Api를 이용해서 구축했다 백엔드가 생겨서 중간에 데이터 저장이나 필터링등 추가적인 기능을 넣을 수 있다

FAST API

pip install fastapi uvicorn requests fast api를 실행하기 위해 패키지를 설치한다

app = FastAPI() #fastapi 시작 코드
# 이 아래에 함수들 작성

# 실시간 검색어를 조회하는 API 엔드포인트
@app.get("/api")
def get_search_terms():


# 학식을 조회하는 API 엔드포인트
@app.get("/apis")
def get_menus():
  
@app.post("/api")
def post_search(request: SearchRequest):
   

fast api는 위 코드들처럼 각각 url과 post, get, delete, update를 사용할 수 있다

 

백엔드 코드

기능별 백엔드 코드

검색어를 받고 검색어를 토큰화

# 검색어를 Nori 분석기를 통해 토큰화하는 함수
def tokenize_query_with_nori(query: str) -> List[str]: 
    analyze_request_body = {
        "analyzer": "nori_analyzer",  # Nori 분석기 사용
        "text": query
    }
    
    response = requests.post(
        f"{ES_HOST}/{ES_INDEX}/_analyze",
        json=analyze_request_body,
        auth=(ES_USER, ES_PASSWORD)
    )
    
    if response.status_code == 200:
        tokens = [token['token'] for token in response.json().get('tokens', [])]
        return tokens
    else:
        raise HTTPException(status_code=response.status_code, detail="Nori 분석기를 사용한 토큰화 실패")

문자열로 받은 검색어를 토큰화 시켜서 문자열 리스트로 반환 한다

 

토큰화 된 검색어 DB저장

# 검색어와 검색 시간을 MariaDB에 저장하는 함수
def store_search_terms_in_db(tokens: List[str]):
    current_time = datetime.now()
    db_session = SessionLocal()
    
    try:
        for token in tokens:
            query = text("INSERT INTO search_tokens (token, search_time) VALUES (:token, :search_time)")
            db_session.execute(query, {"token": token, "search_time": current_time})
        
        db_session.commit()  # 데이터베이스에 변경 사항 커밋
    except Exception as e:
        db_session.rollback()  # 오류 발생 시 롤백
        raise HTTPException(status_code=500, detail="데이터 저장 중 오류가 발생했습니다.")
    finally:
        db_session.close()  # 세션 종료

토큰화된 검색어를 시간과 함께 db에 저장한다 이 검색어와 시간은 실시간 검색어를 위해서 사용된다

 

엘라스틱 서치 엔진에 검색 요청

# Elasticsearch에 검색 요청을 보내는 함수
def search_elasticsearch(tokens: List[str]):
    query_string = " ".join(tokens)
    es_query = {
        "query": {
            "multi_match": {
                "query": query_string,
                "fields": ["title^2", "content"],
                "type": "best_fields"
            }
        }
    }
    
    response = requests.get(
        f"{ES_HOST}/{ES_INDEX}/_search",
        json=es_query,
        auth=(ES_USER, ES_PASSWORD)
    )
    
    if response.status_code == 200:
        return response.json()
    else:
        raise HTTPException(status_code=response.status_code, detail="Elasticsearch 검색에 실패했습니다.")

토큰화된 검색어를 띄어쓰기로 연결하고 그 문자을 쿼리로 보내 title과 content에서 검색한다

multi_match 를 이용해서 title에는 좀더 가중치를 추고 type: best fields를 이용해서 가장 정확한 결과를 가져오도록 설정한다

 

검색 api

# 검색 API 엔드포인트
@app.post("/search")
def search(request: SearchRequest):
    
    # 검색어 토큰화 (Nori 분석기 사용)
    tokens = tokenize_query_with_nori(request.query)
    
    # 검색어 토큰 저장 (MariaDB)
    store_search_terms_in_db(tokens)
    
    # Elasticsearch로 검색 요청
    search_results = search_elasticsearch(tokens)
    
    return {
        "tokens": tokens,
        "results": search_results
    }

위 함수들을 연속적으로 수행하고 토큰을 결과로 받고 또한 검색 결과도 받는다

 

실시간 검색어 함수

# 최근 24시간 이내에 가장 많이 검색된 토큰 상위 5개를 조회하는 함수
def get_top_search_terms_from_db(limit: int = 5) -> List[str]:  # 기본값을 5로 설정
    past_24_hours = datetime.now() - timedelta(hours=24)
    
    with engine.connect() as connection:
        query = text("""
            SELECT token, COUNT(*) as count 
            FROM search_tokens 
            WHERE search_time >= :past_24_hours
            GROUP BY token 
            ORDER BY count DESC 
            LIMIT :limit
        """)
        result = connection.execute(query, {"past_24_hours": past_24_hours, "limit": limit})
        
        result_dict = result.mappings().all()
        return [{"token": row["token"], "count": row["count"]} for row in result_dict]

# 실시간 검색어를 조회하는 API 엔드포인트
@app.get("/search-terms")
def get_search_terms():
    # 가장 많이 검색된 검색어 상위 5개 반환 (최근 24시간 기준)
    top_search_terms = get_top_search_terms_from_db(limit=5)
    return {
        "realtime_search_terms": top_search_terms
    }

결과를 5개만 받기위해 5로 설정하고 기본값도 5로 두었다 sql문을 보면 토큰으로 기준 그룹으로 묶고 시간이 24시간 이전인것의 개수를 새고 내림차순으로 정렬한 다음 5개를 출력한다 아래는 이 함수를 호출할 수 있는 api를 정의

 

학식 메뉴 받기

# restaurant DB의 menus 테이블에서 모든 데이터를 가져오는 함수
def get_menus_from_db():
    db_session = SessionLocalRestaurant()
    
    try:
        query = text("SELECT * FROM menus")
        result = db_session.execute(query).fetchall()  # 모든 결과 행을 가져옴
        menus = [dict(row._mapping) for row in result]  # 각 행을 딕셔너리로 변환
        return menus
    except Exception as e:
        raise HTTPException(status_code=500, detail="메뉴 데이터를 가져오는 중 오류가 발생했습니다.")
    finally:
        db_session.close()
        
# 학식을 조회하는 API 엔드포인트
@app.get("/menus")
def get_menus():
    store_search_terms_in_db(["학식"])
    menus = get_menus_from_db()
    return menus

db의 학식 메뉴를 저장하는 테이블은 이전 데이터들을 모두 삭제하고 이번주 데이터만 존재하기 때문에 모든 정보를 가져온다 또한 학식을 검색했기 때문에 db에 학식 검색을 따로 추가했다 검색중 학식 / 일반 검색어 구분은 프론트엔드에서 진행