296 lines
12 KiB
Python
296 lines
12 KiB
Python
"""
|
||
Karakterler — sınıf (class) + skill tanımları ("chars" izni).
|
||
Sınıf: ne yapar, rolü. Skill: max level, ne yaptığı, bağlı sınıf.
|
||
WAF nedeniyle sadece GET/POST.
|
||
GET /chars/classes → sınıf listesi
|
||
POST /chars/classes → yeni sınıf
|
||
POST /chars/classes/{id} → güncelle
|
||
POST /chars/classes/{id}/delete→ sil
|
||
GET /chars/skills → skill listesi (sınıf adıyla)
|
||
POST /chars/skills → yeni skill
|
||
POST /chars/skills/{id} → güncelle
|
||
POST /chars/skills/{id}/delete → sil
|
||
"""
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from pydantic import BaseModel, Field
|
||
from sqlalchemy import text
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from deps import require_perm, require_panel, get_user_perms, has_perm
|
||
from models import User, get_db
|
||
|
||
router = APIRouter(prefix="/chars", tags=["chars"])
|
||
|
||
|
||
def _author(user: User) -> str:
|
||
return (user.display_name or "").strip() or user.email
|
||
|
||
|
||
async def _assert_owner_or_admin(db: AsyncSession, user: User, table: str, row_id: int):
|
||
perms = await get_user_perms(user, db)
|
||
if has_perm(perms, "*"):
|
||
return
|
||
owner = (await db.execute(text(
|
||
f"SELECT author_id FROM {table} WHERE id=:id"
|
||
), {"id": row_id})).scalar_one_or_none()
|
||
if owner is None:
|
||
raise HTTPException(404, "Kayıt yok")
|
||
if owner != user.id:
|
||
raise HTTPException(403, "Sadece kendi kaydını düzenleyebilir/silebilirsin")
|
||
|
||
|
||
# ---------------- Sınıflar ----------------
|
||
|
||
class ClassIn(BaseModel):
|
||
name: str = Field(min_length=1, max_length=80)
|
||
summary: str | None = Field(default=None, max_length=300)
|
||
description: str | None = Field(default=None, max_length=20000)
|
||
notes: str | None = Field(default=None, max_length=20000)
|
||
sort_order: int = 0
|
||
|
||
|
||
class ClassOut(BaseModel):
|
||
id: int
|
||
name: str
|
||
summary: str | None
|
||
description: str | None
|
||
notes: str | None
|
||
sort_order: int
|
||
author: str | None
|
||
author_id: int | None
|
||
updated_at: str
|
||
|
||
|
||
def _crow(r) -> ClassOut:
|
||
d = dict(r)
|
||
d["updated_at"] = d["updated_at"].isoformat()
|
||
return ClassOut(**d)
|
||
|
||
|
||
@router.get("/classes", response_model=list[ClassOut])
|
||
async def list_classes(user: User = Depends(require_panel), db: AsyncSession = Depends(get_db)):
|
||
rows = (await db.execute(text(
|
||
"SELECT id,name,summary,description,notes,sort_order,author,author_id,updated_at "
|
||
"FROM game_classes ORDER BY sort_order,name"
|
||
))).mappings().all()
|
||
return [_crow(r) for r in rows]
|
||
|
||
|
||
@router.post("/classes", response_model=ClassOut)
|
||
async def add_class(c: ClassIn, user: User = Depends(require_perm("chars")), db: AsyncSession = Depends(get_db)):
|
||
row = (await db.execute(text(
|
||
"INSERT INTO game_classes(name,summary,description,notes,sort_order,author,author_id) "
|
||
"VALUES(:name,:summary,:description,:notes,:so,:author,:aid) "
|
||
"RETURNING id,name,summary,description,notes,sort_order,author,author_id,updated_at"
|
||
), {"name": c.name, "summary": c.summary, "description": c.description, "notes": c.notes,
|
||
"so": c.sort_order, "author": _author(user), "aid": user.id})).mappings().one()
|
||
await db.commit()
|
||
return _crow(row)
|
||
|
||
|
||
@router.post("/classes/{class_id}", response_model=ClassOut)
|
||
async def update_class(class_id: int, c: ClassIn, user: User = Depends(require_perm("chars")),
|
||
db: AsyncSession = Depends(get_db)):
|
||
await _assert_owner_or_admin(db, user, "game_classes", class_id)
|
||
row = (await db.execute(text(
|
||
"UPDATE game_classes SET name=:name,summary=:summary,description=:description,notes=:notes,"
|
||
"sort_order=:so,updated_at=now() WHERE id=:id "
|
||
"RETURNING id,name,summary,description,notes,sort_order,author,author_id,updated_at"
|
||
), {"id": class_id, "name": c.name, "summary": c.summary, "description": c.description,
|
||
"notes": c.notes, "so": c.sort_order})).mappings().one_or_none()
|
||
if not row:
|
||
raise HTTPException(404, "Sınıf yok")
|
||
await db.commit()
|
||
return _crow(row)
|
||
|
||
|
||
@router.post("/classes/{class_id}/delete")
|
||
async def delete_class(class_id: int, user: User = Depends(require_perm("chars")),
|
||
db: AsyncSession = Depends(get_db)):
|
||
await _assert_owner_or_admin(db, user, "game_classes", class_id)
|
||
res = await db.execute(text("DELETE FROM game_classes WHERE id=:id"), {"id": class_id})
|
||
await db.commit()
|
||
if res.rowcount == 0:
|
||
raise HTTPException(404, "Sınıf yok")
|
||
return {"deleted": class_id}
|
||
|
||
|
||
# ---------------- Skiller ----------------
|
||
|
||
class SkillIn(BaseModel):
|
||
name: str = Field(min_length=1, max_length=80)
|
||
max_level: int | None = Field(default=None, ge=0, le=1000)
|
||
effect: str | None = Field(default=None, max_length=20000)
|
||
description: str | None = Field(default=None, max_length=20000)
|
||
class_id: int | None = None
|
||
notes: str | None = Field(default=None, max_length=20000)
|
||
sort_order: int = 0
|
||
|
||
|
||
class SkillOut(BaseModel):
|
||
id: int
|
||
name: str
|
||
max_level: int | None
|
||
effect: str | None
|
||
description: str | None
|
||
class_id: int | None
|
||
class_name: str | None
|
||
notes: str | None
|
||
sort_order: int
|
||
author: str | None
|
||
author_id: int | None
|
||
updated_at: str
|
||
|
||
|
||
def _srow(r) -> SkillOut:
|
||
d = dict(r)
|
||
d["updated_at"] = d["updated_at"].isoformat()
|
||
return SkillOut(**d)
|
||
|
||
|
||
async def _check_class(db: AsyncSession, class_id: int | None) -> None:
|
||
if class_id is None:
|
||
return
|
||
ok = (await db.execute(text("SELECT 1 FROM game_classes WHERE id=:id"), {"id": class_id})).scalar()
|
||
if not ok:
|
||
raise HTTPException(400, "Sınıf bulunamadı")
|
||
|
||
|
||
@router.get("/skills", response_model=list[SkillOut])
|
||
async def list_skills(user: User = Depends(require_panel), db: AsyncSession = Depends(get_db)):
|
||
rows = (await db.execute(text(
|
||
"SELECT s.id,s.name,s.max_level,s.effect,s.description,s.class_id,c.name AS class_name,"
|
||
"s.notes,s.sort_order,s.author,s.author_id,s.updated_at "
|
||
"FROM game_skills s LEFT JOIN game_classes c ON c.id=s.class_id "
|
||
"ORDER BY s.sort_order,s.name"
|
||
))).mappings().all()
|
||
return [_srow(r) for r in rows]
|
||
|
||
|
||
@router.post("/skills", response_model=SkillOut)
|
||
async def add_skill(s: SkillIn, user: User = Depends(require_perm("chars")), db: AsyncSession = Depends(get_db)):
|
||
await _check_class(db, s.class_id)
|
||
sid = (await db.execute(text(
|
||
"INSERT INTO game_skills(name,max_level,effect,description,class_id,notes,sort_order,author,author_id) "
|
||
"VALUES(:name,:mx,:effect,:description,:cid,:notes,:so,:author,:aid) RETURNING id"
|
||
), {"name": s.name, "mx": s.max_level, "effect": s.effect, "description": s.description,
|
||
"cid": s.class_id, "notes": s.notes, "so": s.sort_order,
|
||
"author": _author(user), "aid": user.id})).scalar_one()
|
||
await db.commit()
|
||
return await _get_skill(db, sid)
|
||
|
||
|
||
@router.post("/skills/{skill_id}", response_model=SkillOut)
|
||
async def update_skill(skill_id: int, s: SkillIn, user: User = Depends(require_perm("chars")),
|
||
db: AsyncSession = Depends(get_db)):
|
||
await _assert_owner_or_admin(db, user, "game_skills", skill_id)
|
||
await _check_class(db, s.class_id)
|
||
res = await db.execute(text(
|
||
"UPDATE game_skills SET name=:name,max_level=:mx,effect=:effect,description=:description,"
|
||
"class_id=:cid,notes=:notes,sort_order=:so,updated_at=now() WHERE id=:id"
|
||
), {"id": skill_id, "name": s.name, "mx": s.max_level, "effect": s.effect,
|
||
"description": s.description, "cid": s.class_id, "notes": s.notes, "so": s.sort_order})
|
||
if res.rowcount == 0:
|
||
raise HTTPException(404, "Skill yok")
|
||
await db.commit()
|
||
return await _get_skill(db, skill_id)
|
||
|
||
|
||
@router.post("/skills/{skill_id}/delete")
|
||
async def delete_skill(skill_id: int, user: User = Depends(require_perm("chars")),
|
||
db: AsyncSession = Depends(get_db)):
|
||
await _assert_owner_or_admin(db, user, "game_skills", skill_id)
|
||
res = await db.execute(text("DELETE FROM game_skills WHERE id=:id"), {"id": skill_id})
|
||
await db.commit()
|
||
if res.rowcount == 0:
|
||
raise HTTPException(404, "Skill yok")
|
||
return {"deleted": skill_id}
|
||
|
||
|
||
async def _get_skill(db: AsyncSession, skill_id: int) -> SkillOut:
|
||
row = (await db.execute(text(
|
||
"SELECT s.id,s.name,s.max_level,s.effect,s.description,s.class_id,c.name AS class_name,"
|
||
"s.notes,s.sort_order,s.author,s.author_id,s.updated_at "
|
||
"FROM game_skills s LEFT JOIN game_classes c ON c.id=s.class_id WHERE s.id=:id"
|
||
), {"id": skill_id})).mappings().one()
|
||
return _srow(row)
|
||
|
||
|
||
# ---------------- Irklar ----------------
|
||
|
||
class RaceIn(BaseModel):
|
||
name: str = Field(min_length=1, max_length=80)
|
||
summary: str | None = Field(default=None, max_length=300)
|
||
description: str | None = Field(default=None, max_length=20000)
|
||
traits: str | None = Field(default=None, max_length=20000)
|
||
notes: str | None = Field(default=None, max_length=20000)
|
||
sort_order: int = 0
|
||
|
||
|
||
class RaceOut(BaseModel):
|
||
id: int
|
||
name: str
|
||
summary: str | None
|
||
description: str | None
|
||
traits: str | None
|
||
notes: str | None
|
||
sort_order: int
|
||
author: str | None
|
||
author_id: int | None
|
||
updated_at: str
|
||
|
||
|
||
def _rrow(r) -> RaceOut:
|
||
d = dict(r)
|
||
d["updated_at"] = d["updated_at"].isoformat()
|
||
return RaceOut(**d)
|
||
|
||
|
||
_RACE_COLS = "id,name,summary,description,traits,notes,sort_order,author,author_id,updated_at"
|
||
|
||
|
||
@router.get("/races", response_model=list[RaceOut])
|
||
async def list_races(user: User = Depends(require_panel), db: AsyncSession = Depends(get_db)):
|
||
rows = (await db.execute(text(
|
||
f"SELECT {_RACE_COLS} FROM game_races ORDER BY sort_order,name"
|
||
))).mappings().all()
|
||
return [_rrow(r) for r in rows]
|
||
|
||
|
||
@router.post("/races", response_model=RaceOut)
|
||
async def add_race(r: RaceIn, user: User = Depends(require_perm("chars")), db: AsyncSession = Depends(get_db)):
|
||
row = (await db.execute(text(
|
||
"INSERT INTO game_races(name,summary,description,traits,notes,sort_order,author,author_id) "
|
||
"VALUES(:name,:summary,:description,:traits,:notes,:so,:author,:aid) "
|
||
f"RETURNING {_RACE_COLS}"
|
||
), {"name": r.name, "summary": r.summary, "description": r.description, "traits": r.traits,
|
||
"notes": r.notes, "so": r.sort_order, "author": _author(user), "aid": user.id})).mappings().one()
|
||
await db.commit()
|
||
return _rrow(row)
|
||
|
||
|
||
@router.post("/races/{race_id}", response_model=RaceOut)
|
||
async def update_race(race_id: int, r: RaceIn, user: User = Depends(require_perm("chars")),
|
||
db: AsyncSession = Depends(get_db)):
|
||
await _assert_owner_or_admin(db, user, "game_races", race_id)
|
||
row = (await db.execute(text(
|
||
"UPDATE game_races SET name=:name,summary=:summary,description=:description,traits=:traits,"
|
||
"notes=:notes,sort_order=:so,updated_at=now() WHERE id=:id "
|
||
f"RETURNING {_RACE_COLS}"
|
||
), {"id": race_id, "name": r.name, "summary": r.summary, "description": r.description,
|
||
"traits": r.traits, "notes": r.notes, "so": r.sort_order})).mappings().one_or_none()
|
||
if not row:
|
||
raise HTTPException(404, "Irk yok")
|
||
await db.commit()
|
||
return _rrow(row)
|
||
|
||
|
||
@router.post("/races/{race_id}/delete")
|
||
async def delete_race(race_id: int, user: User = Depends(require_perm("chars")),
|
||
db: AsyncSession = Depends(get_db)):
|
||
await _assert_owner_or_admin(db, user, "game_races", race_id)
|
||
res = await db.execute(text("DELETE FROM game_races WHERE id=:id"), {"id": race_id})
|
||
await db.commit()
|
||
if res.rowcount == 0:
|
||
raise HTTPException(404, "Irk yok")
|
||
return {"deleted": race_id}
|