chengma/cmhub
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: improve operations admin

Browse Source
This commit is contained in:
QiuSW
2026-07-03 16:36:06 +08:00
parent 547196f39a
commit 79605449ef
15 changed files with 520 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/billing/admin.py
+11 -1
View File
@@ -29,8 +29,10 @@ class PointsLedgerAdmin(ReadOnlyLedgerAdmin):
"ref_call",
)
list_filter = ("change_type", "created_at")
search_fields = ("user__username", "user__email", "reason")
search_fields = ("user__username", "user__email", "=ref_order_id", "reason")
ordering = ("-created_at", "-id")
date_hierarchy = "created_at"
list_select_related = ("user", "ref_call")
@admin.register(PricingRule)
@@ -47,6 +49,7 @@ class PricingRuleAdmin(admin.ModelAdmin):
search_fields = ("alias", "resolution")
ordering = ("operation_type", "alias", "resolution")
readonly_fields = ("created_at", "updated_at")
list_editable = ("points_cost", "is_active")
@admin.display(description="resolution")
def resolution_display(self, obj):
@@ -66,6 +69,8 @@ class ExchangeRateAdmin(admin.ModelAdmin):
search_fields = ("currency", "note")
ordering = ("-effective_from", "-id")
readonly_fields = ("created_at", "updated_at")
list_editable = ("is_active",)
date_hierarchy = "effective_from"
@admin.register(RechargeOrder)
@@ -90,6 +95,8 @@ class RechargeOrderAdmin(ReadOnlyLedgerAdmin):
"payment_txn_no",
)
ordering = ("-created_at", "-id")
date_hierarchy = "created_at"
list_select_related = ("user",)
@admin.register(CallRecord)
@@ -112,8 +119,11 @@ class CallRecordAdmin(ReadOnlyLedgerAdmin):
"api_key__key_prefix",
"alias",
"model_used",
"prompt",
"error_message",
"result_ref",
"result_summary",
)
ordering = ("-created_at", "-id")
date_hierarchy = "created_at"
list_select_related = ("user", "api_key")
apps/billing/services.py
+99
View File
@@ -85,6 +85,19 @@ class InvalidRechargePayMethodError(RechargeOrderCreateError):
super().__init__("Unsupported recharge payment method.")
class WalletAdjustmentError(BillingOperationError):
code = "invalid_wallet_adjustment"
class WalletAdjustmentWouldOverdrawError(WalletAdjustmentError):
code = "adjustment_would_overdraw"
def __init__(self, *, points_delta: int, available_points: int):
self.points_delta = points_delta
self.available_points = available_points
super().__init__("Wallet adjustment would make the balance negative.")
@dataclass(frozen=True)
class CallCharge:
call_record: CallRecord
@@ -126,12 +139,56 @@ class RechargeResult:
applied: bool
@dataclass(frozen=True)
class WalletAdjustment:
wallet: UserWallet
ledger_entry: PointsLedger
points_delta: int
balance_after: int
def _validate_positive_points(points: int) -> int:
if isinstance(points, bool) or not isinstance(points, int) or points <= 0:
raise ValueError("points must be a positive integer")
return points
def _validate_adjustment_delta(points_delta: int) -> int:
if (
isinstance(points_delta, bool)
or not isinstance(points_delta, int)
or points_delta == 0
):
raise WalletAdjustmentError("points_delta must be a non-zero integer.")
return points_delta
def _normalize_adjustment_reason(reason: str) -> str:
normalized_reason = str(reason or "").strip()
if not normalized_reason:
raise WalletAdjustmentError("Adjustment reason is required.")
return normalized_reason
def _format_operator(actor) -> str:
if actor is None:
return ""
username_getter = getattr(actor, "get_username", None)
username = (
username_getter()
if callable(username_getter)
else getattr(actor, "username", "")
)
actor_pk = getattr(actor, "pk", None)
if username and actor_pk:
return f"{username}#{actor_pk}"
if username:
return str(username)
if actor_pk:
return f"id#{actor_pk}"
return str(actor)
def _locked_wallet_for_user(user) -> UserWallet:
wallet, _created = UserWallet.objects.select_for_update().get_or_create(user=user)
return wallet
@@ -234,6 +291,48 @@ def get_balance_snapshot(user) -> BalanceSnapshot:
)
def adjust_wallet_points(
*,
user,
points_delta: int,
reason: str,
actor=None,
) -> WalletAdjustment:
points_delta = _validate_adjustment_delta(points_delta)
normalized_reason = _normalize_adjustment_reason(reason)
operator = _format_operator(actor)
ledger_reason = (
f"{normalized_reason}\n\noperator={operator}" if operator else normalized_reason
)
with transaction.atomic():
wallet = _locked_wallet_for_user(user)
balance_after = wallet.points_balance + points_delta
if balance_after < 0:
raise WalletAdjustmentWouldOverdrawError(
points_delta=points_delta,
available_points=wallet.points_balance,
)
wallet.points_balance = balance_after
wallet.save(update_fields=("points_balance", "updated_at"))
ledger_entry = PointsLedger.objects.create(
user=user,
change_type=PointsLedger.ChangeType.ADJUST,
points_delta=points_delta,
balance_after=wallet.points_balance,
reason=ledger_reason,
)
return WalletAdjustment(
wallet=wallet,
ledger_entry=ledger_entry,
points_delta=points_delta,
balance_after=ledger_entry.balance_after,
)
def apply_recharge_payment(payment: RechargePayment) -> RechargeResult:
order_no = str(payment.order_no or "").strip()
pay_method = _normalize_pay_method(payment.pay_method)
apps/billing/tests.py
+69
View File
@@ -34,6 +34,9 @@ from apps.billing.services import (
RechargeAmountMismatchError,
RechargeOrderCreateError,
RechargePayment,
WalletAdjustmentError,
WalletAdjustmentWouldOverdrawError,
adjust_wallet_points,
apply_recharge_payment,
create_recharge_order,
mark_call_success,
@@ -683,6 +686,72 @@ class BillingServiceTests(TestCase):
self.assertEqual(order.status, RechargeOrder.Status.PAID)
self.assertEqual(order.payment_txn_no, "queried-txn-001")
def test_adjust_wallet_points_credits_wallet_and_writes_adjust_ledger(self):
result = adjust_wallet_points(
user=self.user,
points_delta=25,
reason="运营补偿",
)
self.wallet.refresh_from_db()
self.assertEqual(result.points_delta, 25)
self.assertEqual(result.balance_after, 125)
self.assertEqual(self.wallet.points_balance, 125)
ledger = PointsLedger.objects.get(
user=self.user,
change_type=PointsLedger.ChangeType.ADJUST,
)
self.assertEqual(ledger.points_delta, 25)
self.assertEqual(ledger.balance_after, 125)
self.assertEqual(ledger.reason, "运营补偿")
def test_adjust_wallet_points_debits_wallet_without_overdraft(self):
result = adjust_wallet_points(
user=self.user,
points_delta=-40,
reason="纠正误充值",
)
self.wallet.refresh_from_db()
self.assertEqual(result.balance_after, 60)
self.assertEqual(self.wallet.points_balance, 60)
self.assertEqual(result.ledger_entry.points_delta, -40)
self.assertEqual(result.ledger_entry.reason, "纠正误充值")
def test_adjust_wallet_points_requires_non_zero_delta_and_reason(self):
with self.assertRaises(WalletAdjustmentError):
adjust_wallet_points(user=self.user, points_delta=0, reason="无变化")
with self.assertRaises(WalletAdjustmentError):
adjust_wallet_points(user=self.user, points_delta=10, reason=" ")
self.wallet.refresh_from_db()
self.assertEqual(self.wallet.points_balance, 100)
self.assertFalse(
PointsLedger.objects.filter(
user=self.user,
change_type=PointsLedger.ChangeType.ADJUST,
).exists()
)
def test_adjust_wallet_points_rejects_overdraft_without_writing_ledger(self):
with self.assertRaises(WalletAdjustmentWouldOverdrawError) as context:
adjust_wallet_points(
user=self.user,
points_delta=-101,
reason="扣回异常点数",
)
self.assertEqual(context.exception.available_points, 100)
self.wallet.refresh_from_db()
self.assertEqual(self.wallet.points_balance, 100)
self.assertFalse(
PointsLedger.objects.filter(
user=self.user,
change_type=PointsLedger.ChangeType.ADJUST,
).exists()
)
class ConcurrentDebitTests(TransactionTestCase):
def setUp(self):
apps/users/admin.py
+134 -4
View File
@@ -1,9 +1,42 @@
from django.contrib import admin
from django import forms
from django.contrib import admin, messages
from django.contrib.admin.utils import quote, unquote
from django.contrib.auth.admin import UserAdmin as DjangoUserAdmin
from django.core.exceptions import PermissionDenied
from django.http import Http404, HttpResponseRedirect
from django.template.response import TemplateResponse
from django.urls import path, reverse
from django.utils.html import format_html
from apps.billing.services import WalletAdjustmentError, adjust_wallet_points
from .models import ApiKey, User, UserWallet
class WalletAdjustmentForm(forms.Form):
points_delta = forms.IntegerField(
label="点数变动",
help_text="正数加点，负数扣点；不可为 0。",
)
reason = forms.CharField(
label="调整原因",
widget=forms.Textarea(attrs={"rows": 4}),
help_text="必填，会写入点数流水用于审计。",
)
def clean_points_delta(self):
points_delta = self.cleaned_data["points_delta"]
if points_delta == 0:
raise forms.ValidationError("点数变动不可为 0。")
return points_delta
def clean_reason(self):
reason = self.cleaned_data["reason"].strip()
if not reason:
raise forms.ValidationError("必须填写调整原因。")
return reason
@admin.register(User)
class UserAdmin(DjangoUserAdmin):
fieldsets = DjangoUserAdmin.fieldsets + (
@@ -14,6 +47,7 @@ class UserAdmin(DjangoUserAdmin):
"username",
"email",
"status",
"wallet_points_balance",
"is_staff",
"is_active",
"created_at",
@@ -21,13 +55,35 @@ class UserAdmin(DjangoUserAdmin):
list_filter = DjangoUserAdmin.list_filter + ("status",)
search_fields = ("username", "email", "payment_user_id")
def get_queryset(self, request):
return super().get_queryset(request).select_related("wallet")
@admin.display(description="points")
def wallet_points_balance(self, obj):
wallet = getattr(obj, "wallet", None)
return wallet.points_balance if wallet else 0
@admin.register(UserWallet)
class UserWalletAdmin(admin.ModelAdmin):
list_display = ("user", "points_balance", "updated_at")
search_fields = ("user__username", "user__email")
readonly_fields = ("user", "points_balance", "created_at", "updated_at")
list_display = ("user", "points_balance", "updated_at", "adjust_points_link")
search_fields = ("user__username", "user__email", "user__payment_user_id")
readonly_fields = (
"user",
"points_balance",
"created_at",
"updated_at",
"adjust_points_link",
)
fields = (
"user",
"points_balance",
"created_at",
"updated_at",
"adjust_points_link",
)
ordering = ("user__username",)
list_select_related = ("user",)
def has_add_permission(self, request):
return False
@@ -35,12 +91,86 @@ class UserWalletAdmin(admin.ModelAdmin):
def has_delete_permission(self, request, obj=None):
return False
def get_urls(self):
urls = super().get_urls()
custom_urls = [
path(
"<path:object_id>/adjust-points/",
self.admin_site.admin_view(self.adjust_points_view),
name="users_userwallet_adjust_points",
),
]
return custom_urls + urls
@admin.display(description="手工调点")
def adjust_points_link(self, obj):
if obj is None:
return ""
url = reverse(
"admin:users_userwallet_adjust_points",
args=(quote(obj.pk),),
current_app=self.admin_site.name,
)
return format_html('<a class="button" href="{}">调点</a>', url)
def adjust_points_view(self, request, object_id):
wallet = self.get_object(request, unquote(object_id))
if wallet is None:
raise Http404("UserWallet does not exist.")
if not self.has_change_permission(request, wallet):
raise PermissionDenied
if request.method == "POST":
form = WalletAdjustmentForm(request.POST)
if form.is_valid():
try:
result = adjust_wallet_points(
user=wallet.user,
points_delta=form.cleaned_data["points_delta"],
reason=form.cleaned_data["reason"],
actor=request.user,
)
except WalletAdjustmentError as exc:
form.add_error(None, str(exc))
else:
messages.success(
request,
(
f"已调整 {wallet.user} 点数 {result.points_delta:+d}，"
f"当前余额 {result.balance_after}。"
),
)
change_url = reverse(
"admin:users_userwallet_change",
args=(quote(wallet.pk),),
current_app=self.admin_site.name,
)
return HttpResponseRedirect(change_url)
else:
form = WalletAdjustmentForm()
context = {
**self.admin_site.each_context(request),
"title": "手工调整点数",
"opts": self.model._meta,
"original": wallet,
"form": form,
"media": self.media + form.media,
"has_change_permission": self.has_change_permission(request, wallet),
}
return TemplateResponse(
request,
"admin/users/userwallet/adjust_points.html",
context,
)
@admin.register(ApiKey)
class ApiKeyAdmin(admin.ModelAdmin):
list_display = ("key_prefix", "user", "name", "status", "last_used_at", "created_at")
list_filter = ("status", "created_at")
search_fields = ("key_prefix", "user__username", "user__email", "name")
list_select_related = ("user",)
readonly_fields = (
"user",
"key_prefix",
apps/users/templates/admin/users/userwallet/adjust_points.html
+58
View File
@@ -0,0 +1,58 @@
{% extends "admin/base_site.html" %}
{% load admin_urls i18n %}
{% block breadcrumbs %}
<div class="breadcrumbs">
<a href="{% url 'admin:index' %}">首页</a>
&rsaquo; <a href="{% url 'admin:app_list' app_label=opts.app_label %}">{{ opts.app_config.verbose_name }}</a>
&rsaquo; <a href="{% url opts|admin_urlname:'changelist' %}">{{ opts.verbose_name_plural|capfirst }}</a>
&rsaquo; <a href="{% url opts|admin_urlname:'change' original.pk|admin_urlquote %}">{{ original }}</a>
&rsaquo; 手工调整点数
</div>
{% endblock %}
{% block content %}
<div id="content-main">
<form method="post" novalidate>
{% csrf_token %}
<div>
{% if form.non_field_errors %}
<p class="errornote">{{ form.non_field_errors }}</p>
{% endif %}
<fieldset class="module aligned">
<div class="form-row">
<div>
<label>用户:</label>
<div class="readonly">{{ original.user }}</div>
</div>
</div>
<div class="form-row">
<div>
<label>当前余额:</label>
<div class="readonly">{{ original.points_balance }}</div>
</div>
</div>
{% for field in form %}
<div class="form-row{% if field.errors %} errors{% endif %}">
{{ field.errors }}
<div>
{{ field.label_tag }}
{{ field }}
{% if field.help_text %}
<div class="help">{{ field.help_text }}</div>
{% endif %}
</div>
</div>
{% endfor %}
</fieldset>
<div class="submit-row">
<input type="submit" value="保存调整" class="default">
<a class="closelink" href="{% url opts|admin_urlname:'change' original.pk|admin_urlquote %}">取消</a>
</div>
</div>
</form>
</div>
{% endblock %}
apps/users/tests.py
+96
View File
@@ -3,6 +3,9 @@ from django.contrib.auth import get_user_model
from django.test import TestCase
from django.urls import reverse
from apps.billing.models import PointsLedger
from apps.users.models import ApiKey, UserWallet
class AdminSmokeTests(TestCase):
def test_custom_user_is_registered_in_admin(self):
@@ -21,3 +24,96 @@ class AdminSmokeTests(TestCase):
response = self.client.get(reverse("admin:index"))
self.assertEqual(response.status_code, 200)
class UserAdminOperationTests(TestCase):
def setUp(self):
user_model = get_user_model()
self.admin_user = user_model.objects.create_superuser(
username="ops-admin",
email="ops-admin@example.com",
password="test-password",
)
self.user = user_model.objects.create_user(
username="client-user",
email="client-user@example.com",
password="test-password",
)
self.wallet = UserWallet.objects.create(user=self.user, points_balance=100)
self.client.force_login(self.admin_user)
def test_wallet_adjust_admin_view_calls_billing_service_and_writes_ledger(self):
url = reverse("admin:users_userwallet_adjust_points", args=(self.wallet.pk,))
get_response = self.client.get(url)
self.assertEqual(get_response.status_code, 200)
self.assertContains(get_response, "手工调整点数")
self.assertContains(get_response, "当前余额")
response = self.client.post(
url,
{
"points_delta": "30",
"reason": "运营补偿",
},
)
self.assertEqual(response.status_code, 302)
self.wallet.refresh_from_db()
self.assertEqual(self.wallet.points_balance, 130)
ledger = PointsLedger.objects.get(
user=self.user,
change_type=PointsLedger.ChangeType.ADJUST,
)
self.assertEqual(ledger.points_delta, 30)
self.assertEqual(ledger.balance_after, 130)
self.assertIn("运营补偿", ledger.reason)
self.assertIn("operator=ops-admin#", ledger.reason)
def test_wallet_adjust_admin_requires_reason_and_keeps_balance_unchanged(self):
url = reverse("admin:users_userwallet_adjust_points", args=(self.wallet.pk,))
response = self.client.post(
url,
{
"points_delta": "30",
"reason": "",
},
)
self.assertEqual(response.status_code, 200)
self.wallet.refresh_from_db()
self.assertEqual(self.wallet.points_balance, 100)
self.assertFalse(
PointsLedger.objects.filter(
user=self.user,
change_type=PointsLedger.ChangeType.ADJUST,
).exists()
)
def test_wallet_admin_change_form_does_not_directly_edit_balance(self):
url = reverse("admin:users_userwallet_change", args=(self.wallet.pk,))
response = self.client.post(url, {"points_balance": "999"}, follow=True)
self.assertEqual(response.status_code, 200)
self.wallet.refresh_from_db()
self.assertEqual(self.wallet.points_balance, 100)
self.assertFalse(
PointsLedger.objects.filter(
user=self.user,
change_type=PointsLedger.ChangeType.ADJUST,
).exists()
)
def test_api_key_admin_hides_plaintext_and_full_hash(self):
api_key, raw_key = ApiKey.create_for_user(self.user, name="desktop")
url = reverse("admin:users_apikey_change", args=(api_key.pk,))
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertContains(response, api_key.key_prefix)
self.assertNotContains(response, raw_key)
self.assertNotContains(response, api_key.key_hash)