🎉 Initial commit: Document Server MVP
✨ Features implemented: - FastAPI backend with JWT authentication - PostgreSQL database with async SQLAlchemy - HTML document viewer with smart highlighting - Note system connected to highlights (1:1 relationship) - Bookmark system for quick navigation - Integrated search (documents + notes) - Tag system for document organization - Docker containerization with Nginx 🔧 Technical stack: - Backend: FastAPI + PostgreSQL + Redis - Frontend: Alpine.js + Tailwind CSS - Authentication: JWT tokens - File handling: HTML + PDF support - Search: Full-text search with relevance scoring 📋 Core functionality: - Text selection → Highlight creation - Highlight → Note attachment - Note management with search/filtering - Bookmark creation at scroll positions - Document upload with metadata - User management (admin creates accounts)
This commit is contained in:
Hyungi Ahn
354
backend/src/api/routes/search.py
Normal file
354
backend/src/api/routes/search.py
Normal file
@@ -0,0 +1,354 @@
|
||||
"""
|
||||
검색 API 라우터
|
||||
"""
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy import select, or_, and_, text
|
||||
from sqlalchemy.orm import joinedload, selectinload
|
||||
from typing import List, Optional, Dict, Any
|
||||
from datetime import datetime
|
||||
|
||||
from src.core.database import get_db
|
||||
from src.models.user import User
|
||||
from src.models.document import Document, Tag
|
||||
from src.models.highlight import Highlight
|
||||
from src.models.note import Note
|
||||
from src.api.dependencies import get_current_active_user
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class SearchResult(BaseModel):
|
||||
"""검색 결과"""
|
||||
type: str # "document", "note", "highlight"
|
||||
id: str
|
||||
title: str
|
||||
content: str
|
||||
document_id: str
|
||||
document_title: str
|
||||
created_at: datetime
|
||||
relevance_score: float = 0.0
|
||||
highlight_info: Optional[Dict[str, Any]] = None
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class SearchResponse(BaseModel):
|
||||
"""검색 응답"""
|
||||
query: str
|
||||
total_results: int
|
||||
results: List[SearchResult]
|
||||
facets: Dict[str, List[Dict[str, Any]]] = {}
|
||||
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/", response_model=SearchResponse)
|
||||
async def search_all(
|
||||
q: str = Query(..., description="검색어"),
|
||||
type_filter: Optional[str] = Query(None, description="검색 타입 필터: document, note, highlight"),
|
||||
document_id: Optional[str] = Query(None, description="특정 문서 내 검색"),
|
||||
tag: Optional[str] = Query(None, description="태그 필터"),
|
||||
skip: int = Query(0, ge=0),
|
||||
limit: int = Query(50, ge=1, le=100),
|
||||
current_user: User = Depends(get_current_active_user),
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""통합 검색 (문서 + 메모 + 하이라이트)"""
|
||||
results = []
|
||||
|
||||
# 1. 문서 검색
|
||||
if not type_filter or type_filter == "document":
|
||||
document_results = await search_documents(q, document_id, tag, current_user, db)
|
||||
results.extend(document_results)
|
||||
|
||||
# 2. 메모 검색
|
||||
if not type_filter or type_filter == "note":
|
||||
note_results = await search_notes(q, document_id, tag, current_user, db)
|
||||
results.extend(note_results)
|
||||
|
||||
# 3. 하이라이트 검색
|
||||
if not type_filter or type_filter == "highlight":
|
||||
highlight_results = await search_highlights(q, document_id, current_user, db)
|
||||
results.extend(highlight_results)
|
||||
|
||||
# 관련성 점수로 정렬
|
||||
results.sort(key=lambda x: x.relevance_score, reverse=True)
|
||||
|
||||
# 페이지네이션
|
||||
total_results = len(results)
|
||||
paginated_results = results[skip:skip + limit]
|
||||
|
||||
# 패싯 정보 생성
|
||||
facets = await generate_search_facets(results, current_user, db)
|
||||
|
||||
return SearchResponse(
|
||||
query=q,
|
||||
total_results=total_results,
|
||||
results=paginated_results,
|
||||
facets=facets
|
||||
)
|
||||
|
||||
|
||||
async def search_documents(
|
||||
query: str,
|
||||
document_id: Optional[str],
|
||||
tag: Optional[str],
|
||||
current_user: User,
|
||||
db: AsyncSession
|
||||
) -> List[SearchResult]:
|
||||
"""문서 검색"""
|
||||
query_obj = select(Document).options(
|
||||
selectinload(Document.uploader),
|
||||
selectinload(Document.tags)
|
||||
)
|
||||
|
||||
# 권한 필터링
|
||||
if not current_user.is_admin:
|
||||
query_obj = query_obj.where(
|
||||
or_(
|
||||
Document.is_public == True,
|
||||
Document.uploaded_by == current_user.id
|
||||
)
|
||||
)
|
||||
|
||||
# 특정 문서 필터
|
||||
if document_id:
|
||||
query_obj = query_obj.where(Document.id == document_id)
|
||||
|
||||
# 태그 필터
|
||||
if tag:
|
||||
query_obj = query_obj.join(Document.tags).where(Tag.name == tag)
|
||||
|
||||
# 텍스트 검색
|
||||
search_condition = or_(
|
||||
Document.title.ilike(f"%{query}%"),
|
||||
Document.description.ilike(f"%{query}%")
|
||||
)
|
||||
query_obj = query_obj.where(search_condition)
|
||||
|
||||
result = await db.execute(query_obj)
|
||||
documents = result.scalars().all()
|
||||
|
||||
search_results = []
|
||||
for doc in documents:
|
||||
# 관련성 점수 계산 (제목 매치가 더 높은 점수)
|
||||
score = 0.0
|
||||
if query.lower() in doc.title.lower():
|
||||
score += 2.0
|
||||
if doc.description and query.lower() in doc.description.lower():
|
||||
score += 1.0
|
||||
|
||||
search_results.append(SearchResult(
|
||||
type="document",
|
||||
id=str(doc.id),
|
||||
title=doc.title,
|
||||
content=doc.description or "",
|
||||
document_id=str(doc.id),
|
||||
document_title=doc.title,
|
||||
created_at=doc.created_at,
|
||||
relevance_score=score
|
||||
))
|
||||
|
||||
return search_results
|
||||
|
||||
|
||||
async def search_notes(
|
||||
query: str,
|
||||
document_id: Optional[str],
|
||||
tag: Optional[str],
|
||||
current_user: User,
|
||||
db: AsyncSession
|
||||
) -> List[SearchResult]:
|
||||
"""메모 검색"""
|
||||
query_obj = (
|
||||
select(Note)
|
||||
.options(
|
||||
joinedload(Note.highlight).joinedload(Highlight.document)
|
||||
)
|
||||
.join(Highlight)
|
||||
.where(Highlight.user_id == current_user.id)
|
||||
)
|
||||
|
||||
# 특정 문서 필터
|
||||
if document_id:
|
||||
query_obj = query_obj.where(Highlight.document_id == document_id)
|
||||
|
||||
# 태그 필터
|
||||
if tag:
|
||||
query_obj = query_obj.where(Note.tags.contains([tag]))
|
||||
|
||||
# 텍스트 검색 (메모 내용 + 하이라이트된 텍스트)
|
||||
search_condition = or_(
|
||||
Note.content.ilike(f"%{query}%"),
|
||||
Highlight.selected_text.ilike(f"%{query}%")
|
||||
)
|
||||
query_obj = query_obj.where(search_condition)
|
||||
|
||||
result = await db.execute(query_obj)
|
||||
notes = result.scalars().all()
|
||||
|
||||
search_results = []
|
||||
for note in notes:
|
||||
# 관련성 점수 계산
|
||||
score = 0.0
|
||||
if query.lower() in note.content.lower():
|
||||
score += 2.0
|
||||
if query.lower() in note.highlight.selected_text.lower():
|
||||
score += 1.5
|
||||
|
||||
search_results.append(SearchResult(
|
||||
type="note",
|
||||
id=str(note.id),
|
||||
title=f"메모: {note.highlight.selected_text[:50]}...",
|
||||
content=note.content,
|
||||
document_id=str(note.highlight.document.id),
|
||||
document_title=note.highlight.document.title,
|
||||
created_at=note.created_at,
|
||||
relevance_score=score,
|
||||
highlight_info={
|
||||
"highlight_id": str(note.highlight.id),
|
||||
"selected_text": note.highlight.selected_text,
|
||||
"start_offset": note.highlight.start_offset,
|
||||
"end_offset": note.highlight.end_offset
|
||||
}
|
||||
))
|
||||
|
||||
return search_results
|
||||
|
||||
|
||||
async def search_highlights(
|
||||
query: str,
|
||||
document_id: Optional[str],
|
||||
current_user: User,
|
||||
db: AsyncSession
|
||||
) -> List[SearchResult]:
|
||||
"""하이라이트 검색"""
|
||||
query_obj = (
|
||||
select(Highlight)
|
||||
.options(joinedload(Highlight.document))
|
||||
.where(Highlight.user_id == current_user.id)
|
||||
)
|
||||
|
||||
# 특정 문서 필터
|
||||
if document_id:
|
||||
query_obj = query_obj.where(Highlight.document_id == document_id)
|
||||
|
||||
# 텍스트 검색
|
||||
query_obj = query_obj.where(Highlight.selected_text.ilike(f"%{query}%"))
|
||||
|
||||
result = await db.execute(query_obj)
|
||||
highlights = result.scalars().all()
|
||||
|
||||
search_results = []
|
||||
for highlight in highlights:
|
||||
# 관련성 점수 계산
|
||||
score = 1.0 if query.lower() in highlight.selected_text.lower() else 0.5
|
||||
|
||||
search_results.append(SearchResult(
|
||||
type="highlight",
|
||||
id=str(highlight.id),
|
||||
title=f"하이라이트: {highlight.selected_text[:50]}...",
|
||||
content=highlight.selected_text,
|
||||
document_id=str(highlight.document.id),
|
||||
document_title=highlight.document.title,
|
||||
created_at=highlight.created_at,
|
||||
relevance_score=score,
|
||||
highlight_info={
|
||||
"highlight_id": str(highlight.id),
|
||||
"selected_text": highlight.selected_text,
|
||||
"start_offset": highlight.start_offset,
|
||||
"end_offset": highlight.end_offset,
|
||||
"highlight_color": highlight.highlight_color
|
||||
}
|
||||
))
|
||||
|
||||
return search_results
|
||||
|
||||
|
||||
async def generate_search_facets(
|
||||
results: List[SearchResult],
|
||||
current_user: User,
|
||||
db: AsyncSession
|
||||
) -> Dict[str, List[Dict[str, Any]]]:
|
||||
"""검색 결과 패싯 생성"""
|
||||
facets = {}
|
||||
|
||||
# 타입별 개수
|
||||
type_counts = {}
|
||||
for result in results:
|
||||
type_counts[result.type] = type_counts.get(result.type, 0) + 1
|
||||
|
||||
facets["types"] = [
|
||||
{"name": type_name, "count": count}
|
||||
for type_name, count in type_counts.items()
|
||||
]
|
||||
|
||||
# 문서별 개수
|
||||
document_counts = {}
|
||||
for result in results:
|
||||
doc_title = result.document_title
|
||||
document_counts[doc_title] = document_counts.get(doc_title, 0) + 1
|
||||
|
||||
facets["documents"] = [
|
||||
{"name": doc_title, "count": count}
|
||||
for doc_title, count in sorted(document_counts.items(), key=lambda x: x[1], reverse=True)[:10]
|
||||
]
|
||||
|
||||
return facets
|
||||
|
||||
|
||||
@router.get("/suggestions")
|
||||
async def get_search_suggestions(
|
||||
q: str = Query(..., min_length=2, description="검색어 (최소 2글자)"),
|
||||
current_user: User = Depends(get_current_active_user),
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""검색어 자동완성 제안"""
|
||||
suggestions = []
|
||||
|
||||
# 문서 제목에서 제안
|
||||
doc_result = await db.execute(
|
||||
select(Document.title)
|
||||
.where(
|
||||
and_(
|
||||
Document.title.ilike(f"%{q}%"),
|
||||
or_(
|
||||
Document.is_public == True,
|
||||
Document.uploaded_by == current_user.id
|
||||
) if not current_user.is_admin else text("true")
|
||||
)
|
||||
)
|
||||
.limit(5)
|
||||
)
|
||||
doc_titles = doc_result.scalars().all()
|
||||
suggestions.extend([{"text": title, "type": "document"} for title in doc_titles])
|
||||
|
||||
# 태그에서 제안
|
||||
tag_result = await db.execute(
|
||||
select(Tag.name)
|
||||
.where(Tag.name.ilike(f"%{q}%"))
|
||||
.limit(5)
|
||||
)
|
||||
tag_names = tag_result.scalars().all()
|
||||
suggestions.extend([{"text": name, "type": "tag"} for name in tag_names])
|
||||
|
||||
# 메모 태그에서 제안
|
||||
note_result = await db.execute(
|
||||
select(Note.tags)
|
||||
.join(Highlight)
|
||||
.where(Highlight.user_id == current_user.id)
|
||||
)
|
||||
notes = note_result.scalars().all()
|
||||
|
||||
note_tags = set()
|
||||
for note in notes:
|
||||
if note and isinstance(note, list):
|
||||
for tag in note:
|
||||
if q.lower() in tag.lower():
|
||||
note_tags.add(tag)
|
||||
|
||||
suggestions.extend([{"text": tag, "type": "note_tag"} for tag in list(note_tags)[:5]])
|
||||
|
||||
return {"suggestions": suggestions[:10]}
|
||||
Reference in New Issue
Block a user