180 lines
4.6 KiB
Python
180 lines
4.6 KiB
Python
"""
|
|
Admin API routes for managing core data.
|
|
"""
|
|
from typing import List, Optional
|
|
from ninja import Router, Schema
|
|
from ninja.errors import HttpError
|
|
from ninja_jwt.authentication import JWTAuth
|
|
|
|
from apps.users.models import User
|
|
from apps.products.models import Product, Website, Category
|
|
from apps.bounties.models import Bounty, BountyDispute, PaymentEvent
|
|
|
|
router = Router()
|
|
|
|
|
|
def require_admin(user):
|
|
if not user or user.role != 'admin':
|
|
raise HttpError(403, "仅管理员可访问")
|
|
|
|
|
|
class UserOut(Schema):
|
|
id: int
|
|
open_id: str
|
|
name: Optional[str] = None
|
|
email: Optional[str] = None
|
|
role: str
|
|
is_active: bool
|
|
created_at: str
|
|
|
|
|
|
class UserUpdateIn(Schema):
|
|
role: Optional[str] = None
|
|
is_active: Optional[bool] = None
|
|
|
|
|
|
class SimpleOut(Schema):
|
|
id: int
|
|
name: str
|
|
|
|
|
|
class BountyAdminOut(Schema):
|
|
id: int
|
|
title: str
|
|
status: str
|
|
reward: str
|
|
publisher_id: int
|
|
acceptor_id: Optional[int] = None
|
|
is_escrowed: bool
|
|
is_paid: bool
|
|
created_at: str
|
|
|
|
|
|
class PaymentEventOut(Schema):
|
|
id: int
|
|
event_id: str
|
|
event_type: str
|
|
bounty_id: Optional[int] = None
|
|
success: bool
|
|
processed_at: str
|
|
|
|
|
|
class DisputeOut(Schema):
|
|
id: int
|
|
bounty_id: int
|
|
initiator_id: int
|
|
status: str
|
|
created_at: str
|
|
|
|
|
|
@router.get("/users/", response=List[UserOut], auth=JWTAuth())
|
|
def list_users(request):
|
|
require_admin(request.auth)
|
|
users = User.objects.all().order_by('-created_at')
|
|
return [
|
|
UserOut(
|
|
id=u.id,
|
|
open_id=u.open_id,
|
|
name=u.name,
|
|
email=u.email,
|
|
role=u.role,
|
|
is_active=u.is_active,
|
|
created_at=u.created_at.isoformat(),
|
|
)
|
|
for u in users
|
|
]
|
|
|
|
|
|
@router.patch("/users/{user_id}", response=UserOut, auth=JWTAuth())
|
|
def update_user(request, user_id: int, data: UserUpdateIn):
|
|
require_admin(request.auth)
|
|
user = User.objects.get(id=user_id)
|
|
update_data = data.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(user, key, value)
|
|
user.save()
|
|
return UserOut(
|
|
id=user.id,
|
|
open_id=user.open_id,
|
|
name=user.name,
|
|
email=user.email,
|
|
role=user.role,
|
|
is_active=user.is_active,
|
|
created_at=user.created_at.isoformat(),
|
|
)
|
|
|
|
|
|
@router.get("/categories/", response=List[SimpleOut], auth=JWTAuth())
|
|
def list_categories(request):
|
|
require_admin(request.auth)
|
|
return [SimpleOut(id=c.id, name=c.name) for c in Category.objects.all()]
|
|
|
|
|
|
@router.get("/websites/", response=List[SimpleOut], auth=JWTAuth())
|
|
def list_websites(request):
|
|
require_admin(request.auth)
|
|
return [SimpleOut(id=w.id, name=w.name) for w in Website.objects.all()]
|
|
|
|
|
|
@router.get("/products/", response=List[SimpleOut], auth=JWTAuth())
|
|
def list_products(request):
|
|
require_admin(request.auth)
|
|
return [SimpleOut(id=p.id, name=p.name) for p in Product.objects.all()]
|
|
|
|
|
|
@router.get("/bounties/", response=List[BountyAdminOut], auth=JWTAuth())
|
|
def list_bounties(request, status: Optional[str] = None):
|
|
require_admin(request.auth)
|
|
queryset = Bounty.objects.all().order_by('-created_at')
|
|
if status:
|
|
queryset = queryset.filter(status=status)
|
|
return [
|
|
BountyAdminOut(
|
|
id=b.id,
|
|
title=b.title,
|
|
status=b.status,
|
|
reward=str(b.reward),
|
|
publisher_id=b.publisher_id,
|
|
acceptor_id=b.acceptor_id,
|
|
is_escrowed=b.is_escrowed,
|
|
is_paid=b.is_paid,
|
|
created_at=b.created_at.isoformat(),
|
|
)
|
|
for b in queryset
|
|
]
|
|
|
|
|
|
@router.get("/disputes/", response=List[DisputeOut], auth=JWTAuth())
|
|
def list_disputes(request, status: Optional[str] = None):
|
|
require_admin(request.auth)
|
|
disputes = BountyDispute.objects.all().order_by('-created_at')
|
|
if status:
|
|
disputes = disputes.filter(status=status)
|
|
return [
|
|
DisputeOut(
|
|
id=d.id,
|
|
bounty_id=d.bounty_id,
|
|
initiator_id=d.initiator_id,
|
|
status=d.status,
|
|
created_at=d.created_at.isoformat(),
|
|
)
|
|
for d in disputes
|
|
]
|
|
|
|
|
|
@router.get("/payments/", response=List[PaymentEventOut], auth=JWTAuth())
|
|
def list_payment_events(request):
|
|
require_admin(request.auth)
|
|
events = PaymentEvent.objects.all().order_by('-processed_at')
|
|
return [
|
|
PaymentEventOut(
|
|
id=e.id,
|
|
event_id=e.event_id,
|
|
event_type=e.event_type,
|
|
bounty_id=e.bounty_id,
|
|
success=e.success,
|
|
processed_at=e.processed_at.isoformat(),
|
|
)
|
|
for e in events
|
|
]
|