File size: 5,428 Bytes
594ed40 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """
Knowledge Base API
知識タイル一覧表示と検証マーク機構
"""
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import FileResponse, StreamingResponse
from typing import List, Optional
from datetime import datetime
import os
import json
import io
from sqlalchemy.orm import Session
from backend.app.middleware.auth import get_current_user_optional, get_current_user, User
from backend.app.config import settings
from backend.app.database.session import get_db
from backend.app.services.knowledge_service import KnowledgeService, get_knowledge_service
from backend.app.schemas.knowledge import KnowledgeTile, KnowledgeListResponse, KnowledgeDetailResponse, EditRequest, VerificationMark
router = APIRouter()
@router.get("/", response_model=KnowledgeListResponse)
async def list_knowledge_tiles(
domain_id: Optional[str] = Query(None, description="ドメインでフィルタ"),
verification_type: Optional[str] = Query(None, description="検証タイプでフィルタ"),
search: Optional[str] = Query(None, description="検索クエリ"),
page: int = Query(1, ge=1, description="ページ番号"),
page_size: int = Query(20, ge=1, le=100, description="ページサイズ"),
db: Session = Depends(get_db),
service: KnowledgeService = Depends(get_knowledge_service)
):
tiles_orm, total_count = service.list_tiles(
db=db, page=page, page_size=page_size,
domain_id=domain_id, verification_type=verification_type, search=search
)
tiles_pydantic = [KnowledgeTile.from_orm(t) for t in tiles_orm]
return KnowledgeListResponse(
tiles=tiles_pydantic,
total_count=total_count,
page=page,
page_size=page_size,
has_more=(page * page_size) < total_count
)
@router.get("/{tile_id}", response_model=KnowledgeDetailResponse)
async def get_knowledge_tile(
tile_id: str,
db: Session = Depends(get_db),
service: KnowledgeService = Depends(get_knowledge_service)
):
tile_orm = service.get_tile(db, tile_id=tile_id)
if not tile_orm:
raise HTTPException(status_code=404, detail="Knowledge tile not found")
tile_pydantic = KnowledgeTile.from_orm(tile_orm)
return KnowledgeDetailResponse(
tile=tile_pydantic,
full_content=tile_orm.content,
# TODO: Implement sources, related_tiles, and edit_history from DB
sources=[],
related_tiles=[],
edit_history=[]
)
@router.put("/{tile_id}", response_model=KnowledgeTile)
async def update_knowledge_tile(
tile_id: str,
request: EditRequest,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
service: KnowledgeService = Depends(get_knowledge_service)
):
updated_tile_orm = service.update_tile(
db=db, tile_id=tile_id, content=request.content, user=current_user
)
if not updated_tile_orm:
raise HTTPException(status_code=404, detail="Knowledge tile not found")
return KnowledgeTile.from_orm(updated_tile_orm)
@router.get("/coordinates")
async def get_coordinates_for_3d_visualization(
domain_id: Optional[str] = Query(None, description="特定ドメインのみ取得"),
db: Session = Depends(get_db),
service: KnowledgeService = Depends(get_knowledge_service)
):
"""
3D可視化用の座標データを取得
座標を持つタイルのみを返し、必要最小限の情報のみを含める
"""
# Fetch all tiles (large page size to get all)
tiles_orm, _ = service.list_tiles(db=db, page_size=10000, domain_id=domain_id)
# Filter tiles that have coordinates and extract minimal data
coordinates_data = []
for tile in tiles_orm:
if tile.coordinates: # Only include tiles with coordinates
coordinates_data.append({
"tile_id": tile.id,
"topic": tile.topic,
"domain_id": tile.domain_id,
"coordinates": tile.coordinates, # [x, y, z, c, g, v]
"confidence_score": tile.confidence_score,
"verification_type": tile.verification_type
})
return {
"tiles": coordinates_data,
"count": len(coordinates_data),
"domain_id": domain_id or "all"
}
@router.get("/export/json")
async def export_db_json(
domain_id: Optional[str] = Query(None, description="特定ドメインのみエクスポート"),
db: Session = Depends(get_db),
service: KnowledgeService = Depends(get_knowledge_service)
):
# Fetch all tiles for export
tiles_orm, _ = service.list_tiles(db=db, page_size=10000, domain_id=domain_id) # A large page size to get all
tiles_pydantic = [KnowledgeTile.from_orm(t).dict() for t in tiles_orm]
export_data = {
"metadata": {
"export_date": datetime.now().isoformat(),
"source": "NullAI Knowledge Base",
"domain_filter": domain_id or "all",
"tile_count": len(tiles_pydantic)
},
"tiles": tiles_pydantic
}
json_str = json.dumps(export_data, indent=2, ensure_ascii=False, default=str)
return StreamingResponse(
io.BytesIO(json_str.encode('utf-8')),
media_type="application/json",
headers={
"Content-Disposition": f"attachment; filename=null_ai_knowledge_{datetime.now().strftime('%Y%m%d')}.json"
}
)
|