193 lines
7.4 KiB
Python
193 lines
7.4 KiB
Python
from typing import List, Any, Coroutine, Optional
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from pydantic import BaseModel
|
|
from starlette.exceptions import HTTPException
|
|
from starlette.requests import Request
|
|
|
|
from api.models.AssetDTO import AssetsResponse, AssetResponse
|
|
from api.models.GenerationRequest import GenerationRequest, GenerationResponse
|
|
from models.Asset import Asset
|
|
from models.Character import Character
|
|
from api.models.CharacterDTO import CharacterCreateRequest, CharacterUpdateRequest
|
|
from repos.dao import DAO
|
|
from api.dependency import get_dao
|
|
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
from api.endpoints.auth import get_current_user
|
|
from api.dependency import get_project_id
|
|
|
|
router = APIRouter(prefix="/api/characters", tags=["Characters"], dependencies=[Depends(get_current_user)])
|
|
|
|
|
|
@router.get("/", response_model=List[Character])
|
|
async def get_characters(
|
|
request: Request,
|
|
dao: DAO = Depends(get_dao),
|
|
current_user: dict = Depends(get_current_user),
|
|
project_id: Optional[str] = Depends(get_project_id),
|
|
limit: int = 100,
|
|
offset: int = 0
|
|
) -> List[Character]:
|
|
logger.info(f"get_characters called. Limit: {limit}, Offset: {offset}")
|
|
|
|
user_id_filter = str(current_user["_id"])
|
|
if project_id:
|
|
project = await dao.projects.get_project(project_id)
|
|
if not project or str(current_user["_id"]) not in project.members:
|
|
raise HTTPException(status_code=403, detail="Project access denied")
|
|
user_id_filter = None
|
|
|
|
characters = await dao.chars.get_all_characters(
|
|
created_by=user_id_filter,
|
|
project_id=project_id,
|
|
limit=limit,
|
|
offset=offset
|
|
)
|
|
return characters
|
|
|
|
|
|
@router.get("/{character_id}/assets", response_model=AssetsResponse)
|
|
async def get_character_assets(character_id: str, dao: DAO = Depends(get_dao), limit: int = 10,
|
|
offset: int = 0, current_user: dict = Depends(get_current_user)) -> AssetsResponse:
|
|
logger.info(f"get_character_assets called. CharacterID: {character_id}, Limit: {limit}, Offset: {offset}")
|
|
character = await dao.chars.get_character(character_id)
|
|
if character is None:
|
|
raise HTTPException(status_code=404, detail="Character not found")
|
|
|
|
# Access Check
|
|
is_creator = character.created_by == str(current_user["_id"])
|
|
is_project_member = False
|
|
if character.project_id and character.project_id in current_user.get("project_ids", []):
|
|
is_project_member = True
|
|
|
|
if not is_creator and not is_project_member:
|
|
raise HTTPException(status_code=403, detail="Access denied")
|
|
|
|
assets = await dao.assets.get_assets_by_char_id(character_id, limit, offset)
|
|
# Filter assets by user ownership as well?
|
|
# Usually if you own character, you see its assets.
|
|
# But assets also have specific created_by.
|
|
# Let's assume if you own character you can see its assets.
|
|
|
|
total_count = await dao.assets.get_asset_count(character_id)
|
|
|
|
asset_responses = [AssetResponse.model_validate(a.model_dump()) for a in assets]
|
|
return AssetsResponse(assets=asset_responses, total_count=total_count)
|
|
|
|
|
|
@router.get("/{character_id}", response_model=Character)
|
|
async def get_character_by_id(character_id: str, request: Request, dao: DAO = Depends(get_dao), current_user: dict = Depends(get_current_user)) -> Character:
|
|
logger.debug(f"get_character_by_id called. ID: {character_id}")
|
|
character = await dao.chars.get_character(character_id)
|
|
|
|
if not character:
|
|
raise HTTPException(status_code=404, detail="Character not found")
|
|
|
|
if character:
|
|
is_creator = character.created_by == str(current_user["_id"])
|
|
is_project_member = False
|
|
if character.project_id and character.project_id in current_user.get("project_ids", []):
|
|
is_project_member = True
|
|
|
|
if not is_creator and not is_project_member:
|
|
raise HTTPException(status_code=403, detail="Access denied")
|
|
|
|
return character
|
|
|
|
|
|
@router.post("/", response_model=Character)
|
|
async def create_character(
|
|
char_req: CharacterCreateRequest,
|
|
project_id: Optional[str] = Depends(get_project_id),
|
|
dao: DAO = Depends(get_dao),
|
|
current_user: dict = Depends(get_current_user)
|
|
) -> Character:
|
|
logger.info("create_character called")
|
|
char_req.project_id = project_id
|
|
char_data = char_req.model_dump()
|
|
char_data["created_by"] = str(current_user["_id"])
|
|
if "id" not in char_data:
|
|
char_data["id"] = None
|
|
|
|
if project_id:
|
|
project = await dao.projects.get_project(project_id)
|
|
if not project or str(current_user["_id"]) not in project.members:
|
|
raise HTTPException(status_code=403, detail="Project access denied")
|
|
|
|
new_char = Character(**char_data)
|
|
new_char.avatar_asset_id = new_char.avatar_image.split("/")[-1]
|
|
created_char = await dao.chars.add_character(new_char)
|
|
return created_char
|
|
|
|
|
|
@router.put("/{character_id}", response_model=Character)
|
|
async def update_character(
|
|
character_id: str,
|
|
char_update: CharacterUpdateRequest,
|
|
dao: DAO = Depends(get_dao),
|
|
current_user: dict = Depends(get_current_user)
|
|
) -> Character:
|
|
logger.info(f"update_character called. ID: {character_id}")
|
|
|
|
existing_char = await dao.chars.get_character(character_id)
|
|
if not existing_char:
|
|
raise HTTPException(status_code=404, detail="Character not found")
|
|
|
|
is_creator = existing_char.created_by == str(current_user["_id"])
|
|
is_project_member = False
|
|
if existing_char.project_id and existing_char.project_id in current_user.get("project_ids", []):
|
|
is_project_member = True
|
|
|
|
if not is_creator and not is_project_member:
|
|
raise HTTPException(status_code=403, detail="Access denied")
|
|
|
|
update_data = char_update.model_dump(exclude_unset=True)
|
|
|
|
if "project_id" in update_data and update_data["project_id"]:
|
|
new_project_id = update_data["project_id"]
|
|
project = await dao.projects.get_project(new_project_id)
|
|
if not project or str(current_user["_id"]) not in project.members:
|
|
raise HTTPException(status_code=403, detail="Target project access denied")
|
|
|
|
updated_char_data = existing_char.model_dump()
|
|
updated_char_data.update(update_data)
|
|
|
|
updated_char = Character(**updated_char_data)
|
|
|
|
success = await dao.chars.update_char(character_id, updated_char)
|
|
if not success:
|
|
raise HTTPException(status_code=500, detail="Failed to update character")
|
|
|
|
return updated_char
|
|
|
|
|
|
@router.delete("/{character_id}", status_code=204)
|
|
async def delete_character(
|
|
character_id: str,
|
|
dao: DAO = Depends(get_dao),
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
logger.info(f"delete_character called. ID: {character_id}")
|
|
|
|
existing_char = await dao.chars.get_character(character_id)
|
|
if not existing_char:
|
|
raise HTTPException(status_code=404, detail="Character not found")
|
|
|
|
is_creator = existing_char.created_by == str(current_user["_id"])
|
|
is_project_member = False
|
|
if existing_char.project_id and existing_char.project_id in current_user.get("project_ids", []):
|
|
is_project_member = True
|
|
|
|
if not is_creator and not is_project_member:
|
|
raise HTTPException(status_code=403, detail="Access denied")
|
|
|
|
success = await dao.chars.delete_character(character_id)
|
|
if not success:
|
|
raise HTTPException(status_code=500, detail="Failed to delete character")
|
|
|
|
return
|