378 lines
16 KiB
Python
378 lines
16 KiB
Python
import csv
|
|
import io
|
|
import os
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, UploadFile, File, Form, status
|
|
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
from pydantic import BaseModel
|
|
|
|
from ..database import get_db
|
|
from ..models import User, Serie, Set, Card
|
|
from ..auth import require_admin, hash_password
|
|
|
|
router = APIRouter()
|
|
templates = Jinja2Templates(directory="app/templates")
|
|
|
|
UPLOAD_DIR = Path("app/static/images")
|
|
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
# ─── Pydantic schemas ───────────────────────────────────────────────
|
|
|
|
class SerieCreate(BaseModel):
|
|
name: str
|
|
description: str | None = None
|
|
|
|
class SetCreate(BaseModel):
|
|
code: str
|
|
name: str
|
|
release_date: str | None = None
|
|
description: str | None = None
|
|
|
|
class CardCreate(BaseModel):
|
|
number: str
|
|
name: str
|
|
image: str | None = None # URL
|
|
|
|
class NewUserForm(BaseModel):
|
|
username: str
|
|
password: str
|
|
role: str = "user"
|
|
|
|
|
|
# ─── Admin HTML pages ───────────────────────────────────────────────
|
|
|
|
@router.get("/admin", response_class=HTMLResponse)
|
|
async def admin_dashboard(request: Request, user: User = Depends(require_admin)):
|
|
return templates.TemplateResponse("admin/dashboard.html", {"request": request})
|
|
|
|
|
|
@router.get("/admin/series", response_class=HTMLResponse)
|
|
async def admin_series(request: Request, db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(require_admin)):
|
|
result = await db.execute(select(Serie).options(selectinload(Serie.sets)).order_by(Serie.name))
|
|
series = result.scalars().all()
|
|
return templates.TemplateResponse("admin/series.html", {"request": request, "series": series})
|
|
|
|
|
|
@router.get("/admin/sets", response_class=HTMLResponse)
|
|
async def admin_sets_page(request: Request, db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(require_admin)):
|
|
result = await db.execute(
|
|
select(Set).options(selectinload(Set.serie), selectinload(Set.cards)).order_by(Set.code)
|
|
)
|
|
sets = result.scalars().all()
|
|
series_result = await db.execute(select(Serie).order_by(Serie.name))
|
|
series = series_result.scalars().all()
|
|
return templates.TemplateResponse("admin/sets.html", {"request": request, "sets": sets, "series": series})
|
|
|
|
|
|
@router.get("/admin/cards/{set_code}", response_class=HTMLResponse)
|
|
async def admin_cards_page(request: Request, set_code: str, db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(require_admin)):
|
|
result = await db.execute(
|
|
select(Set).where(Set.code == set_code.upper()).options(
|
|
selectinload(Set.cards), selectinload(Set.serie)
|
|
)
|
|
)
|
|
s = result.scalar_one_or_none()
|
|
if not s:
|
|
return templates.TemplateResponse("404.html", {"request": request}, status_code=404)
|
|
return templates.TemplateResponse("admin/cards.html", {"request": request, "set": s})
|
|
|
|
# ─── Users (admin) ─────────────────────────────────────────────────
|
|
|
|
@router.post("/api/admin/users")
|
|
async def create_user(data: NewUserForm, db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_admin)):
|
|
exists = await db.execute(select(User).where(User.username == data.username))
|
|
if exists.scalar_one_or_none():
|
|
raise HTTPException(status_code=400, detail="Bruger findes allerede")
|
|
user = User(username=data.username, password_hash=hash_password(data.password), role=data.role)
|
|
db.add(user)
|
|
await db.commit()
|
|
return {"ok": True, "username": user.username, "role": user.role}
|
|
|
|
@router.get("/api/admin/users")
|
|
async def list_users(db: AsyncSession = Depends(get_db), _: User = Depends(require_admin)):
|
|
result = await db.execute(select(User).order_by(User.username))
|
|
return [{"id": u.id, "username": u.username, "role": u.role, "created_at": str(u.created_at)}
|
|
for u in result.scalars().all()]
|
|
|
|
|
|
# ─── API: Series ────────────────────────────────────────────────────
|
|
|
|
@router.post("/api/admin/series")
|
|
async def create_serie(data: SerieCreate, db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_admin)):
|
|
result = await db.execute(select(Serie).where(Serie.name == data.name))
|
|
if result.scalar_one_or_none():
|
|
raise HTTPException(status_code=400, detail="Serie findes allerede")
|
|
serie = Serie(name=data.name, description=data.description)
|
|
db.add(serie)
|
|
await db.commit()
|
|
return {"ok": True, "id": serie.id, "name": serie.name}
|
|
|
|
@router.get("/api/admin/series")
|
|
async def list_series(db: AsyncSession = Depends(get_db), _: User = Depends(require_admin)):
|
|
result = await db.execute(select(Serie).options(selectinload(Serie.sets)).order_by(Serie.name))
|
|
return [{"id": s.id, "name": s.name, "description": s.description or "",
|
|
"set_count": len(s.sets)} for s in result.scalars().all()]
|
|
|
|
@router.delete("/api/admin/series/{serie_id}")
|
|
async def delete_serie(serie_id: int, db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_admin)):
|
|
result = await db.execute(select(Serie).where(Serie.id == serie_id))
|
|
s = result.scalar_one_or_none()
|
|
if not s:
|
|
raise HTTPException(status_code=404, detail="Serie ikke fundet")
|
|
await db.delete(s)
|
|
await db.commit()
|
|
return {"ok": True}
|
|
|
|
@router.post("/api/admin/series/import")
|
|
async def import_series(file: UploadFile = File(...), db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_admin)):
|
|
content = await file.read()
|
|
text = content.decode("utf-8-sig")
|
|
reader = csv.DictReader(io.StringIO(text))
|
|
created = 0
|
|
errors = []
|
|
for i, row in enumerate(reader, start=2):
|
|
name = row.get("name", "").strip()
|
|
if not name:
|
|
errors.append(f"Linje {i}: mangler navn")
|
|
continue
|
|
try:
|
|
exists = await db.execute(select(Serie).where(Serie.name == name))
|
|
if exists.scalar_one_or_none():
|
|
errors.append(f"Linje {i}: '{name}' findes allerede")
|
|
continue
|
|
serie = Serie(name=name, description=row.get("description", "").strip() or None)
|
|
db.add(serie)
|
|
created += 1
|
|
except Exception as e:
|
|
errors.append(f"Linje {i}: {str(e)}")
|
|
await db.commit()
|
|
return {"ok": True, "created": created, "errors": errors}
|
|
|
|
|
|
# ─── API: Sets ──────────────────────────────────────────────────────
|
|
|
|
@router.post("/api/admin/series/{serie_name}/sets")
|
|
async def create_set(serie_name: str, data: SetCreate, db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_admin)):
|
|
result = await db.execute(select(Serie).where(Serie.name == serie_name))
|
|
serie = result.scalar_one_or_none()
|
|
if not serie:
|
|
raise HTTPException(status_code=404, detail="Serie ikke fundet")
|
|
exists = await db.execute(select(Set).where(Set.code == data.code.upper()))
|
|
if exists.scalar_one_or_none():
|
|
raise HTTPException(status_code=400, detail="Set-kode findes allerede")
|
|
s = Set(serie_id=serie.id, code=data.code.upper(), name=data.name,
|
|
release_date=data.release_date, description=data.description)
|
|
db.add(s)
|
|
await db.commit()
|
|
return {"ok": True, "id": s.id, "code": s.code, "name": s.name}
|
|
|
|
@router.get("/api/admin/series/{serie_name}/sets")
|
|
async def list_sets(serie_name: str, db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_admin)):
|
|
result = await db.execute(
|
|
select(Serie).where(Serie.name == serie_name).options(selectinload(Serie.sets))
|
|
)
|
|
serie = result.scalar_one_or_none()
|
|
if not serie:
|
|
raise HTTPException(status_code=404, detail="Serie ikke fundet")
|
|
return [{"id": s.id, "code": s.code, "name": s.name, "card_count": len(s.cards)}
|
|
for s in serie.sets]
|
|
|
|
@router.get("/api/admin/sets")
|
|
async def list_all_sets(db: AsyncSession = Depends(get_db), _: User = Depends(require_admin)):
|
|
result = await db.execute(
|
|
select(Set).options(selectinload(Set.serie), selectinload(Set.cards)).order_by(Set.code)
|
|
)
|
|
sets = result.scalars().all()
|
|
return [{"id": s.id, "code": s.code, "name": s.name, "serie": s.serie.name,
|
|
"card_count": len(s.cards), "release_date": s.release_date or ""}
|
|
for s in sets]
|
|
|
|
@router.delete("/api/admin/sets/{set_id}")
|
|
async def delete_set(set_id: int, db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_admin)):
|
|
result = await db.execute(select(Set).where(Set.id == set_id))
|
|
s = result.scalar_one_or_none()
|
|
if not s:
|
|
raise HTTPException(status_code=404, detail="Set ikke fundet")
|
|
await db.delete(s)
|
|
await db.commit()
|
|
return {"ok": True}
|
|
|
|
@router.post("/api/admin/sets/import")
|
|
async def import_sets(file: UploadFile = File(...), db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_admin)):
|
|
content = await file.read()
|
|
text = content.decode("utf-8-sig")
|
|
reader = csv.DictReader(io.StringIO(text))
|
|
created = 0
|
|
errors = []
|
|
for i, row in enumerate(reader, start=2):
|
|
code = row.get("code", "").strip().upper()
|
|
name = row.get("name", "").strip()
|
|
serie_name = row.get("serie", "").strip()
|
|
if not code or not name or not serie_name:
|
|
errors.append(f"Linje {i}: mangler kode, navn eller serie")
|
|
continue
|
|
try:
|
|
serie = await db.execute(select(Serie).where(Serie.name == serie_name))
|
|
serie = serie.scalar_one_or_none()
|
|
if not serie:
|
|
errors.append(f"Linje {i}: serie '{serie_name}' ikke fundet")
|
|
continue
|
|
exists = await db.execute(select(Set).where(Set.code == code))
|
|
if exists.scalar_one_or_none():
|
|
errors.append(f"Linje {i}: kode '{code}' findes allerede")
|
|
continue
|
|
s = Set(serie_id=serie.id, code=code, name=name,
|
|
release_date=row.get("release_date", "").strip() or None,
|
|
description=row.get("description", "").strip() or None)
|
|
db.add(s)
|
|
created += 1
|
|
except Exception as e:
|
|
errors.append(f"Linje {i}: {str(e)}")
|
|
await db.commit()
|
|
return {"ok": True, "created": created, "errors": errors}
|
|
|
|
|
|
# ─── API: Cards ─────────────────────────────────────────────────────
|
|
|
|
@router.post("/api/admin/sets/{set_code}/cards")
|
|
async def create_card(set_code: str, data: CardCreate, db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_admin)):
|
|
result = await db.execute(select(Set).where(Set.code == set_code.upper()))
|
|
s = result.scalar_one_or_none()
|
|
if not s:
|
|
raise HTTPException(status_code=404, detail="Set ikke fundet")
|
|
card = Card(set_id=s.id, number=data.number, name=data.name)
|
|
if data.image:
|
|
card.image_path = data.image
|
|
db.add(card)
|
|
await db.commit()
|
|
return {"ok": True, "id": card.id, "full_code": card.full_code, "name": card.name}
|
|
|
|
@router.get("/api/admin/sets/{set_code}/cards")
|
|
async def list_cards(set_code: str, db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_admin)):
|
|
result = await db.execute(
|
|
select(Set).where(Set.code == set_code.upper()).options(
|
|
selectinload(Set.cards), selectinload(Set.serie)
|
|
)
|
|
)
|
|
s = result.scalar_one_or_none()
|
|
if not s:
|
|
raise HTTPException(status_code=404, detail="Set ikke fundet")
|
|
return [{"id": c.id, "number": c.number, "name": c.name, "full_code": c.full_code,
|
|
"image_url": c.image_url} for c in s.cards]
|
|
|
|
@router.delete("/api/admin/cards/{card_id}")
|
|
async def delete_card(card_id: int, db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_admin)):
|
|
result = await db.execute(select(Card).where(Card.id == card_id))
|
|
card = result.scalar_one_or_none()
|
|
if not card:
|
|
raise HTTPException(status_code=404, detail="Kort ikke fundet")
|
|
# Slet billedfil hvis lokal
|
|
if card.image_path and not card.image_path.startswith(("http://", "https://")):
|
|
fp = Path("app/static") / card.image_path
|
|
if fp.exists():
|
|
fp.unlink()
|
|
await db.delete(card)
|
|
await db.commit()
|
|
return {"ok": True}
|
|
|
|
@router.post("/api/admin/cards/import")
|
|
async def import_cards(file: UploadFile = File(...), db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_admin)):
|
|
content = await file.read()
|
|
text = content.decode("utf-8-sig")
|
|
reader = csv.DictReader(io.StringIO(text))
|
|
created = 0
|
|
errors = []
|
|
for i, row in enumerate(reader, start=2):
|
|
set_code = row.get("set_code", "").strip().upper()
|
|
number = row.get("number", "").strip()
|
|
name = row.get("name", "").strip()
|
|
if not set_code or not number or not name:
|
|
errors.append(f"Linje {i}: mangler set_code, number eller name")
|
|
continue
|
|
try:
|
|
result = await db.execute(select(Set).where(Set.code == set_code))
|
|
s = result.scalar_one_or_none()
|
|
if not s:
|
|
errors.append(f"Linje {i}: set '{set_code}' ikke fundet")
|
|
continue
|
|
image = row.get("image", "").strip() or None
|
|
card = Card(set_id=s.id, number=number, name=name, image_path=image)
|
|
db.add(card)
|
|
created += 1
|
|
except Exception as e:
|
|
errors.append(f"Linje {i}: {str(e)}")
|
|
await db.commit()
|
|
return {"ok": True, "created": created, "errors": errors}
|
|
|
|
@router.post("/api/admin/cards/{card_id}/upload")
|
|
async def upload_card_image(card_id: int, file: UploadFile = File(...),
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(require_admin)):
|
|
result = await db.execute(
|
|
select(Card).where(Card.id == card_id).options(
|
|
selectinload(Card.set).selectinload(Set.serie)
|
|
)
|
|
)
|
|
card = result.scalar_one_or_none()
|
|
if not card:
|
|
raise HTTPException(status_code=404, detail="Kort ikke fundet")
|
|
|
|
# Bestem filtype
|
|
ext = os.path.splitext(file.filename or "image.png")[1].lower()
|
|
if ext not in (".jpg", ".jpeg", ".png", ".gif", ".webp"):
|
|
raise HTTPException(status_code=400, detail="Kun .jpg, .png, .gif, .webp tilladt")
|
|
|
|
serie_name = card.set.serie.name.replace(" ", "_").replace("/", "_")
|
|
set_code = card.set.code
|
|
number = card.number
|
|
folder = UPLOAD_DIR / serie_name / set_code
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
|
|
filename = f"{number}{ext}"
|
|
filepath = folder / filename
|
|
content = await file.read()
|
|
filepath.write_bytes(content)
|
|
|
|
# Gem relativ sti
|
|
rel_path = f"images/{serie_name}/{set_code}/{filename}"
|
|
card.image_path = rel_path
|
|
await db.commit()
|
|
|
|
return {"ok": True, "image_url": f"/static/{rel_path}", "path": rel_path}
|
|
|
|
|
|
# ─── CSV Templates ──────────────────────────────────────────────────
|
|
|
|
@router.get("/api/admin/templates/series.csv", response_class=PlainTextResponse)
|
|
async def series_csv_template(_: User = Depends(require_admin)):
|
|
return "name,description\nSword & Shield,The Sword & Shield series\n"
|
|
|
|
@router.get("/api/admin/templates/sets.csv", response_class=PlainTextResponse)
|
|
async def sets_csv_template(_: User = Depends(require_admin)):
|
|
return "serie,code,name,release_date,description\nSword & Shield,EVS,Evolving Skies,2021-08-27,\n"
|
|
|
|
@router.get("/api/admin/templates/cards.csv", response_class=PlainTextResponse)
|
|
async def cards_csv_template(_: User = Depends(require_admin)):
|
|
return "set_code,number,name,image\nEVS,050,Raichu,\n" |