Admin: per-user activity file (logins, logouts, orders) with CSV export
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
42
accounts/activity.py
Normal file
42
accounts/activity.py
Normal file
@@ -0,0 +1,42 @@
|
||||
"""
|
||||
«پروندهی کاربر»: ورود، خروج و خریدهای یک کاربر به ترتیبِ زمان، برای ادمین.
|
||||
|
||||
فقط میخواند؛ چیزی جدا ذخیره نمیکند، پس با جدولهای اصلی (LoginEvent، Order، StatusChangeLog) هرگز
|
||||
ناهماهنگ نمیشود. تصمیمِ ادغامنکردنِ اینها در یک جدول: ستونهایشان فرق دارد.
|
||||
آزموده نشده: کاربری با دهها هزار ردیف (همهی ردیفها در حافظه میآیند).
|
||||
"""
|
||||
from django.utils import timezone
|
||||
|
||||
from payment.models import Order
|
||||
from statuslog.models import StatusChangeLog
|
||||
|
||||
from .models import LoginEvent, User
|
||||
|
||||
_STATUS_FA = dict(Order.STATUS_CHOICES)
|
||||
_EVENT_FA = dict(LoginEvent.EVENT_CHOICES)
|
||||
COLUMNS = ['زمان', 'نوع', 'شرح', 'مبلغ', 'آیپی']
|
||||
|
||||
|
||||
def user_activity(user: User) -> list[dict]:
|
||||
rows = [{'at': user.created_at, 'kind': 'ثبتنام', 'detail': '', 'amount': '', 'ip': ''}]
|
||||
for e in LoginEvent.objects.filter(user_id=user.id):
|
||||
rows.append({'at': e.created_at, 'kind': _EVENT_FA.get(e.event, e.event),
|
||||
'detail': e.user_agent[:80], 'amount': '', 'ip': e.ip or ''})
|
||||
orders = list(Order.objects.filter(user_id=user.id))
|
||||
for o in orders:
|
||||
rows.append({
|
||||
'at': o.created_at, 'kind': 'سفارش',
|
||||
'detail': f'#{o.id} {o.plan_title or "بدون پلن"} ({o.get_provider_display()})',
|
||||
'amount': f'{o.amount} {o.currency}', 'ip': '',
|
||||
})
|
||||
for c in StatusChangeLog.objects.filter(label='payment.Order', object_id__in=[o.id for o in orders]):
|
||||
rows.append({
|
||||
'at': c.created_at, 'kind': 'تغییر وضعیت سفارش',
|
||||
'detail': f'#{c.object_id}: {_STATUS_FA.get(c.old_status, c.old_status or "—")} ← '
|
||||
f'{_STATUS_FA.get(c.new_status, c.new_status)} ({c.event})',
|
||||
'amount': '', 'ip': '',
|
||||
})
|
||||
rows.sort(key=lambda r: r['at'])
|
||||
for r in rows:
|
||||
r['at_local'] = timezone.localtime(r['at']).strftime('%Y-%m-%d %H:%M:%S')
|
||||
return rows
|
||||
@@ -51,9 +51,52 @@ class SendSMSForm(forms.Form):
|
||||
|
||||
@admin.register(User)
|
||||
class UserAdmin(admin.ModelAdmin):
|
||||
list_display = ('id', 'public_id', 'name', 'phone', 'grade_id', 'created_at')
|
||||
list_display = ('id', 'public_id', 'name', 'phone', 'grade_id', 'created_at', 'activity_link')
|
||||
search_fields = ('name', 'phone')
|
||||
|
||||
@admin.display(description='پرونده')
|
||||
def activity_link(self, obj):
|
||||
from django.urls import reverse
|
||||
from django.utils.html import format_html
|
||||
return format_html('<a href="{}">فعالیت</a>', reverse('admin:accounts_user_activity', args=[obj.pk]))
|
||||
|
||||
def get_urls(self):
|
||||
from django.urls import path
|
||||
return [
|
||||
path('<int:user_id>/activity/', self.admin_site.admin_view(self.activity_view),
|
||||
name='accounts_user_activity'),
|
||||
] + super().get_urls()
|
||||
|
||||
def activity_view(self, request, user_id):
|
||||
"""ورود/خروج و خریدهای یک کاربر؛ ?format=csv برای اکسل. فقط با اجازهی دیدنِ کاربر و ورود و سفارش."""
|
||||
import csv
|
||||
from django.http import Http404, HttpResponse
|
||||
from django.shortcuts import render
|
||||
from payment.models import Order
|
||||
from .activity import COLUMNS, user_activity
|
||||
from .public_id import encode
|
||||
|
||||
perms = ('accounts.view_user', 'accounts.view_loginevent', 'payment.view_order')
|
||||
if not all(request.user.has_perm(p) for p in perms):
|
||||
raise PermissionDenied
|
||||
user = User.objects.filter(pk=user_id).first()
|
||||
if user is None:
|
||||
raise Http404
|
||||
rows = user_activity(user)
|
||||
if request.GET.get('format') == 'csv':
|
||||
response = HttpResponse(content_type='text/csv; charset=utf-8')
|
||||
response['Content-Disposition'] = f'attachment; filename="user-{user.pk}-activity.csv"'
|
||||
response.write('\ufeff') # BOM: اکسل بیآن فارسی را خراب میخواند
|
||||
writer = csv.writer(response)
|
||||
writer.writerow(COLUMNS)
|
||||
for r in rows:
|
||||
writer.writerow([r['at_local'], r['kind'], r['detail'], r['amount'], r['ip']])
|
||||
return response
|
||||
return render(request, 'admin/accounts_user_activity.html', {
|
||||
**self.admin_site.each_context(request), 'target': user, 'public_id': encode(user.pk),
|
||||
'rows': rows, 'columns': COLUMNS, 'title': 'پروندهی کاربر',
|
||||
})
|
||||
|
||||
@admin.display(description='شناسهی نمایشی')
|
||||
def public_id(self, obj):
|
||||
from .public_id import encode
|
||||
|
||||
41
accounts/tests_user_activity.py
Normal file
41
accounts/tests_user_activity.py
Normal file
@@ -0,0 +1,41 @@
|
||||
"""پروندهی کاربر در ادمین: ترتیب، CSV، و دسترسی."""
|
||||
from django.test import TestCase
|
||||
from django.utils import timezone
|
||||
|
||||
from accounts.models import LoginEvent, User
|
||||
from payment.models import Order
|
||||
|
||||
|
||||
class UserActivityTests(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(phone='09121117001', name='علی')
|
||||
LoginEvent.objects.create(event='login_ok', user_id=self.user.id, phone=self.user.phone, ip='203.0.113.9')
|
||||
Order.objects.create(user_id=self.user.id, amount=45000, plan_title='بسته ۲۰', idempotency_key='k1', status='paid')
|
||||
from django.contrib.auth.models import User as AuthUser
|
||||
self.admin = AuthUser.objects.create_superuser('boss', 'b@x.ir', 'pw')
|
||||
self.client.force_login(self.admin)
|
||||
|
||||
def test_page_lists_login_and_order(self):
|
||||
r = self.client.get(f'/admin/accounts/user/{self.user.id}/activity/')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
body = r.content.decode()
|
||||
self.assertIn('ورود موفق', body)
|
||||
self.assertIn('بسته ۲۰', body)
|
||||
self.assertIn('45000 IRT', body)
|
||||
|
||||
def test_csv_has_bom_and_rows(self):
|
||||
r = self.client.get(f'/admin/accounts/user/{self.user.id}/activity/?format=csv')
|
||||
text = r.content.decode('utf-8')
|
||||
self.assertTrue(text.startswith(''))
|
||||
self.assertIn('203.0.113.9', text)
|
||||
self.assertEqual(len(text.strip().splitlines()), 4) # سرستون + ثبتنام + ورود + سفارش
|
||||
|
||||
def test_view_only_staff_without_order_permission_is_refused(self):
|
||||
from django.contrib.auth.models import Permission, User as AuthUser
|
||||
s = AuthUser.objects.create_user('sup', 's@x.ir', 'pw', is_staff=True)
|
||||
s.user_permissions.add(Permission.objects.get(codename='view_user', content_type__app_label='accounts'))
|
||||
self.client.force_login(s)
|
||||
self.assertEqual(self.client.get(f'/admin/accounts/user/{self.user.id}/activity/').status_code, 403)
|
||||
|
||||
def test_unknown_user_is_404(self):
|
||||
self.assertEqual(self.client.get('/admin/accounts/user/999999/activity/').status_code, 404)
|
||||
16
templates/admin/accounts_user_activity.html
Normal file
16
templates/admin/accounts_user_activity.html
Normal file
@@ -0,0 +1,16 @@
|
||||
{% extends "admin/base_site.html" %}
|
||||
{% block content %}
|
||||
<div dir="rtl">
|
||||
<h1>پروندهی کاربر {{ public_id }} — {{ target.phone|default:"بدون شماره" }}</h1>
|
||||
<p>{{ target.name|default:"بینام" }} · {{ rows|length }} رویداد · زمانها به وقتِ تهران
|
||||
· <a href="?format=csv">خروجی CSV (اکسل)</a></p>
|
||||
<table>
|
||||
<thead><tr>{% for c in columns %}<th>{{ c }}</th>{% endfor %}</tr></thead>
|
||||
<tbody>
|
||||
{% for r in rows %}
|
||||
<tr><td dir="ltr">{{ r.at_local }}</td><td>{{ r.kind }}</td><td>{{ r.detail }}</td><td>{{ r.amount }}</td><td dir="ltr">{{ r.ip }}</td></tr>
|
||||
{% endfor %}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
{% endblock %}
|
||||
Reference in New Issue
Block a user