File size: 3,909 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 |
from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Dict, Any, Optional
import json
import os
from backend.app.middleware.auth import (
get_current_user, get_current_user_optional, get_user_or_guest,
require_role, require_expert, User, GuestUser
)
router = APIRouter()
DOMAIN_SCHEMAS_PATH = "domain_schemas.json"
class DomainAxis(BaseModel):
name: str
description: str
keywords: List[str]
class DomainSchema(BaseModel):
domain_id: str
name: str
description: str
axes: List[DomainAxis]
class DomainListResponse(BaseModel):
domains: List[Dict[str, Any]]
def load_schemas() -> Dict:
"""ドメインスキーマファイルを読み込む"""
if os.path.exists(DOMAIN_SCHEMAS_PATH):
with open(DOMAIN_SCHEMAS_PATH, 'r', encoding='utf-8') as f:
return json.load(f)
return {"domains": {}}
def save_schemas(schemas: Dict):
"""ドメインスキーマファイルを保存"""
with open(DOMAIN_SCHEMAS_PATH, 'w', encoding='utf-8') as f:
json.dump(schemas, f, ensure_ascii=False, indent=2)
@router.get("/", response_model=DomainListResponse)
async def list_domains(
current_user: User = Depends(get_user_or_guest)
):
"""利用可能なドメイン一覧を取得（ゲストアクセス可）"""
schemas = load_schemas()
domains = []
for domain_id, schema in schemas.get("domains", {}).items():
domains.append({
"domain_id": domain_id,
"name": schema.get("name", domain_id),
"description": schema.get("description", ""),
"axis_count": len(schema.get("axes", []))
})
return {"domains": domains}
@router.get("/{domain_id}")
async def get_domain(
domain_id: str,
current_user: User = Depends(get_user_or_guest)
):
"""特定のドメインスキーマを取得（ゲストアクセス可）"""
schemas = load_schemas()
if domain_id not in schemas.get("domains", {}):
raise HTTPException(status_code=404, detail=f"Domain '{domain_id}' not found")
schema = schemas["domains"][domain_id]
return {
"domain_id": domain_id,
**schema
}
@router.put("/{domain_id}")
async def update_domain(
domain_id: str,
domain: DomainSchema,
current_user: User = Depends(require_role("editor"))
):
"""
ドメインスキーマを更新（または新規作成）。
editorまたはadminロールが必要。
"""
schemas = load_schemas()
if "domains" not in schemas:
schemas["domains"] = {}
schemas["domains"][domain_id] = {
"name": domain.name,
"description": domain.description,
"axes": [
{
"name": axis.name,
"description": axis.description,
"keywords": axis.keywords
}
for axis in domain.axes
]
}
save_schemas(schemas)
return {
"message": f"Domain '{domain_id}' updated successfully",
"domain_id": domain_id,
"updated_by": current_user.id
}
@router.delete("/{domain_id}")
async def delete_domain(
domain_id: str,
current_user: User = Depends(require_role("admin"))
):
"""ドメインを削除（adminのみ）"""
schemas = load_schemas()
if domain_id not in schemas.get("domains", {}):
raise HTTPException(status_code=404, detail=f"Domain '{domain_id}' not found")
# 基本ドメインは削除不可
protected_domains = ["medical", "legal", "economics"]
if domain_id in protected_domains:
raise HTTPException(status_code=403, detail=f"Cannot delete protected domain '{domain_id}'")
del schemas["domains"][domain_id]
save_schemas(schemas)
return {
"message": f"Domain '{domain_id}' deleted successfully",
"deleted_by": current_user.id
}
|