""" 노트 문서 링크 관련 API 엔드포인트 """ from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from sqlalchemy import and_, or_ from typing import List, Optional from pydantic import BaseModel import uuid from ...core.database import get_sync_db from ..dependencies import get_current_user from ...models.user import User from ...models.note_document import NoteDocument from ...models.document import Document from ...models.note_link import NoteLink router = APIRouter() # Pydantic 모델들 class NoteLinkCreate(BaseModel): target_note_id: Optional[str] = None target_document_id: Optional[str] = None selected_text: str start_offset: int end_offset: int link_text: Optional[str] = None description: Optional[str] = None target_text: Optional[str] = None target_start_offset: Optional[int] = None target_end_offset: Optional[int] = None link_type: Optional[str] = "note" class NoteLinkUpdate(BaseModel): target_note_id: Optional[str] = None target_document_id: Optional[str] = None link_text: Optional[str] = None description: Optional[str] = None target_text: Optional[str] = None target_start_offset: Optional[int] = None target_end_offset: Optional[int] = None link_type: Optional[str] = None class NoteLinkResponse(BaseModel): id: str source_note_id: Optional[str] = None source_document_id: Optional[str] = None target_note_id: Optional[str] = None target_document_id: Optional[str] = None target_content_type: Optional[str] = None # "document" or "note" selected_text: str start_offset: int end_offset: int link_text: Optional[str] = None description: Optional[str] = None target_text: Optional[str] = None target_start_offset: Optional[int] = None target_end_offset: Optional[int] = None link_type: str created_at: str updated_at: Optional[str] = None # 추가 정보 target_note_title: Optional[str] = None target_document_title: Optional[str] = None source_note_title: Optional[str] = None source_document_title: Optional[str] = None class Config: from_attributes = True @router.get("/note-documents/{note_id}/links", response_model=List[NoteLinkResponse]) def get_note_links( note_id: str, db: Session = Depends(get_sync_db), current_user: User = Depends(get_current_user) ): """노트에서 나가는 링크 목록 조회""" # 노트 존재 확인 note = db.query(NoteDocument).filter(NoteDocument.id == note_id).first() if not note: raise HTTPException(status_code=404, detail="Note not found") # 노트에서 나가는 링크들 조회 links = db.query(NoteLink).filter( NoteLink.source_note_id == note_id ).all() result = [] for link in links: link_data = { "id": str(link.id), "source_note_id": str(link.source_note_id) if link.source_note_id else None, "source_document_id": str(link.source_document_id) if link.source_document_id else None, "target_note_id": str(link.target_note_id) if link.target_note_id else None, "target_document_id": str(link.target_document_id) if link.target_document_id else None, "selected_text": link.selected_text, "start_offset": link.start_offset, "end_offset": link.end_offset, "link_text": link.link_text, "description": link.description, "target_text": link.target_text, "target_start_offset": link.target_start_offset, "target_end_offset": link.target_end_offset, "link_type": link.link_type, "created_at": link.created_at.isoformat() if link.created_at else None, "updated_at": link.updated_at.isoformat() if link.updated_at else None, } # 대상 제목 및 타입 추가 if link.target_note_id: target_note = db.query(NoteDocument).filter(NoteDocument.id == link.target_note_id).first() if target_note: link_data["target_note_title"] = target_note.title link_data["target_content_type"] = "note" elif link.target_document_id: target_doc = db.query(Document).filter(Document.id == link.target_document_id).first() if target_doc: link_data["target_document_title"] = target_doc.title link_data["target_content_type"] = "document" result.append(NoteLinkResponse(**link_data)) return result @router.get("/note-documents/{note_id}/backlinks", response_model=List[NoteLinkResponse]) def get_note_backlinks( note_id: str, db: Session = Depends(get_sync_db), current_user: User = Depends(get_current_user) ): """노트로 들어오는 백링크 목록 조회""" # 노트 존재 확인 note = db.query(NoteDocument).filter(NoteDocument.id == note_id).first() if not note: raise HTTPException(status_code=404, detail="Note not found") # 노트로 들어오는 백링크들 조회 backlinks = db.query(NoteLink).filter( NoteLink.target_note_id == note_id ).all() result = [] for link in backlinks: link_data = { "id": str(link.id), "source_note_id": str(link.source_note_id) if link.source_note_id else None, "source_document_id": str(link.source_document_id) if link.source_document_id else None, "target_note_id": str(link.target_note_id) if link.target_note_id else None, "target_document_id": str(link.target_document_id) if link.target_document_id else None, "selected_text": link.selected_text, "start_offset": link.start_offset, "end_offset": link.end_offset, "link_text": link.link_text, "description": link.description, "target_text": link.target_text, "target_start_offset": link.target_start_offset, "target_end_offset": link.target_end_offset, "link_type": link.link_type, "created_at": link.created_at.isoformat() if link.created_at else None, "updated_at": link.updated_at.isoformat() if link.updated_at else None, } # 출발지 제목 추가 if link.source_note_id: source_note = db.query(NoteDocument).filter(NoteDocument.id == link.source_note_id).first() if source_note: link_data["source_note_title"] = source_note.title elif link.source_document_id: source_doc = db.query(Document).filter(Document.id == link.source_document_id).first() if source_doc: link_data["source_document_title"] = source_doc.title result.append(NoteLinkResponse(**link_data)) return result @router.post("/note-documents/{note_id}/links", response_model=NoteLinkResponse) def create_note_link( note_id: str, link_data: NoteLinkCreate, db: Session = Depends(get_sync_db), current_user: User = Depends(get_current_user) ): """노트에서 다른 노트/문서로의 링크 생성""" # 출발지 노트 존재 확인 source_note = db.query(NoteDocument).filter(NoteDocument.id == note_id).first() if not source_note: raise HTTPException(status_code=404, detail="Source note not found") # 대상 확인 (노트 또는 문서 중 하나는 반드시 있어야 함) if not link_data.target_note_id and not link_data.target_document_id: raise HTTPException(status_code=400, detail="Either target_note_id or target_document_id is required") if link_data.target_note_id and link_data.target_document_id: raise HTTPException(status_code=400, detail="Cannot specify both target_note_id and target_document_id") # 대상 존재 확인 if link_data.target_note_id: target_note = db.query(NoteDocument).filter(NoteDocument.id == link_data.target_note_id).first() if not target_note: raise HTTPException(status_code=404, detail="Target note not found") if link_data.target_document_id: target_doc = db.query(Document).filter(Document.id == link_data.target_document_id).first() if not target_doc: raise HTTPException(status_code=404, detail="Target document not found") # 링크 생성 note_link = NoteLink( source_note_id=note_id, target_note_id=link_data.target_note_id, target_document_id=link_data.target_document_id, selected_text=link_data.selected_text, start_offset=link_data.start_offset, end_offset=link_data.end_offset, link_text=link_data.link_text, description=link_data.description, target_text=link_data.target_text, target_start_offset=link_data.target_start_offset, target_end_offset=link_data.target_end_offset, link_type=link_data.link_type or "note", created_by=current_user.id ) db.add(note_link) db.commit() db.refresh(note_link) # 응답 데이터 구성 response_data = { "id": str(note_link.id), "source_note_id": str(note_link.source_note_id) if note_link.source_note_id else None, "source_document_id": str(note_link.source_document_id) if note_link.source_document_id else None, "target_note_id": str(note_link.target_note_id) if note_link.target_note_id else None, "target_document_id": str(note_link.target_document_id) if note_link.target_document_id else None, "selected_text": note_link.selected_text, "start_offset": note_link.start_offset, "end_offset": note_link.end_offset, "link_text": note_link.link_text, "description": note_link.description, "target_text": note_link.target_text, "target_start_offset": note_link.target_start_offset, "target_end_offset": note_link.target_end_offset, "link_type": note_link.link_type, "created_at": note_link.created_at.isoformat() if note_link.created_at else None, "updated_at": note_link.updated_at.isoformat() if note_link.updated_at else None, } # 소스 및 타겟 타입 설정 response_data["source_content_type"] = "note" # 노트에서 출발하는 링크 if note_link.target_note_id: target_note = db.query(NoteDocument).filter(NoteDocument.id == note_link.target_note_id).first() if target_note: response_data["target_note_title"] = target_note.title response_data["target_content_type"] = "note" elif note_link.target_document_id: target_doc = db.query(Document).filter(Document.id == note_link.target_document_id).first() if target_doc: response_data["target_document_title"] = target_doc.title response_data["target_content_type"] = "document" return NoteLinkResponse(**response_data) @router.delete("/note-links/{link_id}") def delete_note_link( link_id: str, db: Session = Depends(get_sync_db), current_user: User = Depends(get_current_user) ): """노트 링크 삭제""" link = db.query(NoteLink).filter(NoteLink.id == link_id).first() if not link: raise HTTPException(status_code=404, detail="Link not found") # 권한 확인 (링크 생성자 또는 관리자만 삭제 가능) if link.created_by != current_user.id and not current_user.is_admin: raise HTTPException(status_code=403, detail="Permission denied") db.delete(link) db.commit() return {"message": "Link deleted successfully"}