JakeBreath/NoN-Site
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

annoucment logic implemented, fixing looks

Browse Source
This commit is contained in:
JakeBreath
2026-08-05 15:50:32 -05:00
parent c163271a04
commit f5954247fc
22 changed files with 1020 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nonpacks/announcements/__init__.py
View File
nonpacks/announcements/admin.py
+3
View File
@@ -0,0 +1,3 @@
from django.contrib import admin
# Register your models here.
nonpacks/announcements/apps.py
+5
View File
@@ -0,0 +1,5 @@
from django.apps import AppConfig
class AnnouncementsConfig(AppConfig):
name = 'announcements'
nonpacks/announcements/forms.py
+13
View File
@@ -0,0 +1,13 @@
from django import forms
from .models import Announcement
class AnnouncementForm(forms.ModelForm):
class Meta:
model = Announcement
fields = ['title', 'body', 'published']
widgets = {
'title': forms.TextInput(attrs={'class': 'form-input', 'placeholder': 'Announcement title'}),
'body': forms.Textarea(attrs={'class': 'form-textarea', 'rows': 12, 'placeholder': 'Write in Markdown…'}),
}
nonpacks/announcements/migrations/0001_initial.py
+75
View File
@@ -0,0 +1,75 @@
# Generated by Django 6.0.3 on 2026-08-05 19:53
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
('library', '0013_project_rating_count_project_rating_score_comment_and_more'),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='Announcement',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('title', models.CharField(max_length=200)),
('body', models.TextField(blank=True, default='')),
('published', models.BooleanField(default=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('created_by', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='announcements', to=settings.AUTH_USER_MODEL)),
],
options={
'ordering': ['-created_at', '-pk'],
},
),
migrations.CreateModel(
name='AnnouncementComment',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('body', models.TextField()),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('announcement', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='comments', to='announcements.announcement')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='announcement_comments', to=settings.AUTH_USER_MODEL)),
],
options={
'ordering': ['created_at', 'pk'],
},
),
migrations.CreateModel(
name='Notification',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('kind', models.CharField(choices=[('version', 'New version'), ('asset', 'New media'), ('edit', 'Project edited'), ('comment', 'New comment')], max_length=16)),
('text', models.CharField(max_length=255)),
('read', models.BooleanField(default=False)),
('created_at', models.DateTimeField(auto_now_add=True)),
('actor', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to=settings.AUTH_USER_MODEL)),
('project', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='notifications', to='library.project')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='notifications', to=settings.AUTH_USER_MODEL)),
],
options={
'ordering': ['-created_at', '-pk'],
},
),
migrations.CreateModel(
name='AnnouncementRead',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('read_at', models.DateTimeField(auto_now_add=True)),
('announcement', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='reads', to='announcements.announcement')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='announcement_reads', to=settings.AUTH_USER_MODEL)),
],
options={
'constraints': [models.UniqueConstraint(fields=('announcement', 'user'), name='uniq_announcement_read')],
},
),
]
nonpacks/announcements/migrations/__init__.py
View File
nonpacks/announcements/models.py
+102
View File
@@ -0,0 +1,102 @@
from django.conf import settings
from django.db import models
from library.models import Project
class Announcement(models.Model):
"""A staff-written, site-wide announcement (global). Kept forever."""
title = models.CharField(max_length=200)
body = models.TextField(blank=True, default='') # Markdown
created_by = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
null=True,
related_name='announcements',
)
published = models.BooleanField(default=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
ordering = ['-created_at', '-pk']
def __str__(self):
return self.title
@property
def comment_count(self):
return self.comments.count()
class AnnouncementRead(models.Model):
"""Read receipt: one row per (announcement, user) the first time it's opened."""
announcement = models.ForeignKey(
Announcement, on_delete=models.CASCADE, related_name='reads',
)
user = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name='announcement_reads',
)
read_at = models.DateTimeField(auto_now_add=True)
class Meta:
constraints = [
models.UniqueConstraint(
fields=['announcement', 'user'], name='uniq_announcement_read',
),
]
def __str__(self):
return f'{self.user.username} read {self.announcement.title}'
class AnnouncementComment(models.Model):
"""A comment on an announcement page. Body is Markdown."""
announcement = models.ForeignKey(
Announcement, on_delete=models.CASCADE, related_name='comments',
)
user = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name='announcement_comments',
)
body = models.TextField()
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
ordering = ['created_at', 'pk']
def __str__(self):
return f'{self.user.username}: {self.body[:40]}'
class Notification(models.Model):
"""A personal alert triggered by UGC activity. Retained at ~200 per user."""
KIND_CHOICES = [
('version', 'New version'),
('asset', 'New media'),
('edit', 'Project edited'),
('comment', 'New comment'),
]
user = models.ForeignKey(
settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name='notifications',
)
actor = models.ForeignKey(
settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, related_name='+',
)
project = models.ForeignKey(
Project, on_delete=models.CASCADE, related_name='notifications',
)
kind = models.CharField(max_length=16, choices=KIND_CHOICES)
text = models.CharField(max_length=255)
read = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ['-created_at', '-pk']
def __str__(self):
return f'{self.user.username}: {self.text}'
nonpacks/announcements/notifications.py
+32
View File
@@ -0,0 +1,32 @@
from django.db import transaction
from .models import Notification
PERSONAL_NOTIFICATION_LIMIT = 200
def notify_project_change(project, actor, kind, text):
"""Notify the project's owner + contributors about a change, excluding the actor.
Retention: keeps only the newest ~200 notifications per recipient."""
if actor is None or not actor.is_authenticated:
return
recipient_ids = {project.owner_id}
recipient_ids.update(project.contributors.values_list('user_id', flat=True))
recipient_ids.discard(actor.pk)
if not recipient_ids:
return
with transaction.atomic():
for uid in recipient_ids:
Notification.objects.create(
user_id=uid, actor=actor, project=project, kind=kind, text=text,
)
# Prune each recipient down to the newest limit.
for uid in recipient_ids:
stale = list(
Notification.objects.filter(user_id=uid)
.order_by('-created_at', '-pk')
.values_list('pk', flat=True)[PERSONAL_NOTIFICATION_LIMIT:]
)
if stale:
Notification.objects.filter(pk__in=stale).delete()
nonpacks/announcements/tests.py
+193
View File
@@ -0,0 +1,193 @@
import io
import zipfile
from django.contrib.auth import get_user_model
from django.core.cache import cache
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import TestCase
from django.urls import reverse
from library.models import Project, ProjectContributor
from .models import (
Announcement,
AnnouncementComment,
AnnouncementRead,
Notification,
)
from .notifications import notify_project_change
def _zip_bytes(entries=None):
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w') as zf:
for name, content in (entries or {}).items():
zf.writestr(name, content)
return buf.getvalue()
class AnnouncementsTestCase(TestCase):
def setUp(self):
User = get_user_model()
self.staff = User.objects.create_user(username='Boss', password='pw', is_staff=True)
self.normal = User.objects.create_user(username='NormalUser', password='pw')
self.normal2 = User.objects.create_user(username='NormalUser2', password='pw')
self.project = Project.objects.create(
slug='test-pack', title='Test Pack', category='mod', owner=self.normal,
)
cache.clear()
def _auth(self, user=None):
if user is not None:
self.client.force_login(user)
session = self.client.session
session['authorized'] = True
session.save()
def _upload_version(self, name, content):
self.client.post(
reverse('library:api_upload_temp'),
{'kind': 'version', 'file': SimpleUploadedFile(name, content, content_type='application/zip')},
HTTP_X_REQUESTED_WITH='XMLHttpRequest',
)
# --- Global announcements: staff management ---
def test_staff_can_create_edit_delete(self):
self._auth(self.staff)
resp = self.client.post(
reverse('announcements:create'),
{'title': 'Welcome', 'body': 'Hello **world**', 'published': 'on'},
)
announcement = Announcement.objects.get(title='Welcome')
self.assertRedirects(resp, reverse('announcements:detail', args=[announcement.pk]))
self.assertTrue(announcement.published)
self.assertEqual(announcement.created_by, self.staff)
resp = self.client.post(
reverse('announcements:edit', args=[announcement.pk]),
{'title': 'Welcome v2', 'body': 'Edited', 'published': 'on'},
)
announcement.refresh_from_db()
self.assertEqual(announcement.title, 'Welcome v2')
self.client.post(reverse('announcements:delete', args=[announcement.pk]))
self.assertFalse(Announcement.objects.filter(pk=announcement.pk).exists())
def test_non_staff_cannot_manage(self):
self._auth(self.normal)
resp = self.client.post(
reverse('announcements:create'),
{'title': 'Nope', 'body': 'x', 'published': 'on'},
)
self.assertEqual(resp.status_code, 403)
self.assertFalse(Announcement.objects.exists())
def test_unpublished_hidden_from_list_and_detail(self):
draft = Announcement.objects.create(
title='Draft', body='secret', created_by=self.staff, published=False,
)
published = Announcement.objects.create(
title='Live', body='ok', created_by=self.staff, published=True,
)
self._auth(self.normal)
resp = self.client.get(reverse('announcements:list'))
self.assertContains(resp, 'Live')
self.assertNotContains(resp, 'Draft')
self.assertEqual(self.client.get(reverse('announcements:detail', args=[draft.pk])).status_code, 404)
self.assertEqual(self.client.get(reverse('announcements:detail', args=[published.pk])).status_code, 200)
# --- Read receipts ---
def test_detail_marks_read_once(self):
announcement = Announcement.objects.create(title='A', body='b', created_by=self.staff)
self._auth(self.normal)
self.assertEqual(self.client.get(reverse('announcements:detail', args=[announcement.pk])).status_code, 200)
self.assertEqual(AnnouncementRead.objects.filter(announcement=announcement, user=self.normal).count(), 1)
self.client.get(reverse('announcements:detail', args=[announcement.pk]))
self.assertEqual(AnnouncementRead.objects.filter(announcement=announcement, user=self.normal).count(), 1)
# --- Announcement comments ---
def test_comment_lifecycle(self):
announcement = Announcement.objects.create(title='A', body='b', created_by=self.staff)
self._auth(self.normal)
self.client.post(reverse('announcements:add_comment', args=[announcement.pk]), {'body': 'First!'})
comment = AnnouncementComment.objects.get(announcement=announcement)
self.assertEqual(comment.user, self.normal)
# Owner edits own comment.
self.client.post(reverse('announcements:edit_comment', args=[announcement.pk, comment.pk]), {'body': 'Edited!'})
comment.refresh_from_db()
self.assertEqual(comment.body, 'Edited!')
# Another non-staff user cannot edit or delete it.
self._auth(self.normal2)
self.assertEqual(
self.client.post(reverse('announcements:edit_comment', args=[announcement.pk, comment.pk]), {'body': 'x'}).status_code,
403,
)
self.assertEqual(
self.client.post(reverse('announcements:delete_comment', args=[announcement.pk, comment.pk])).status_code,
403,
)
# Staff can delete any comment.
self._auth(self.staff)
self.client.post(reverse('announcements:delete_comment', args=[announcement.pk, comment.pk]))
self.assertFalse(AnnouncementComment.objects.filter(pk=comment.pk).exists())
# --- Personal notifications ---
def test_version_change_notifies_owner_and_contributors_not_actor(self):
ProjectContributor.objects.create(project=self.project, user=self.normal2, added_by=self.normal)
self._auth(self.normal2)
self._upload_version('pack.zip', _zip_bytes())
self.client.post(reverse('library:version_upload', args=[self.project.slug]), {'version_name': '1.0.0'})
self.assertTrue(Notification.objects.filter(user=self.normal, kind='version', project=self.project).exists())
self.assertFalse(Notification.objects.filter(user=self.normal2, kind='version').exists())
def test_comment_change_notifies_owner(self):
self._auth(self.normal2)
self.client.post(
reverse('library:add_comment', args=[self.project.slug]),
{'body': 'Nice pack'},
)
self.assertTrue(Notification.objects.filter(user=self.normal, kind='comment', project=self.project).exists())
def test_retention_prunes_to_200(self):
for i in range(210):
notify_project_change(self.project, self.normal2, 'version', f'change {i}')
self.assertEqual(Notification.objects.filter(user=self.normal).count(), 200)
def test_mark_all_read(self):
for i in range(3):
notify_project_change(self.project, self.normal2, 'edit', f'edit {i}')
self._auth(self.normal)
self.client.post(reverse('announcements:mark_all_read'))
self.assertFalse(Notification.objects.filter(user=self.normal, read=False).exists())
def test_project_detail_auto_marks_read(self):
notify_project_change(self.project, self.normal2, 'comment', 'someone commented')
self._auth(self.normal)
self.client.get(reverse('library:project_detail', args=[self.project.slug]))
self.assertFalse(Notification.objects.filter(user=self.normal, read=False).exists())
# --- Home ---
def test_home_guests_see_global_only(self):
Announcement.objects.create(title='Global', body='hi', created_by=self.staff)
notify_project_change(self.project, self.normal2, 'edit', 'edited it')
self._auth()
resp = self.client.get(reverse('landing:home'))
self.assertContains(resp, 'Global')
self.assertNotContains(resp, 'For you')
def test_home_logged_sees_global_and_personal(self):
Announcement.objects.create(title='Global', body='hi', created_by=self.staff)
notify_project_change(self.project, self.normal2, 'edit', 'edited the pack')
self._auth(self.normal)
resp = self.client.get(reverse('landing:home'))
self.assertContains(resp, 'Global')
self.assertContains(resp, 'For you')
self.assertContains(resp, 'edited the pack')
nonpacks/announcements/urls.py
+18
View File
@@ -0,0 +1,18 @@
from django.urls import path
from . import views
app_name = 'announcements'
urlpatterns = [
path('announcements/', views.announcement_list, name='list'),
path('announcements/new/', views.announcement_create, name='create'),
path('announcements/personal/', views.personal_notifications, name='personal'),
path('announcements/mark-all-read/', views.mark_all_notifications_read, name='mark_all_read'),
path('announcements/<int:pk>/', views.announcement_detail, name='detail'),
path('announcements/<int:pk>/edit/', views.announcement_edit, name='edit'),
path('announcements/<int:pk>/delete/', views.announcement_delete, name='delete'),
path('announcements/<int:pk>/comments/add/', views.announcement_add_comment, name='add_comment'),
path('announcements/<int:pk>/comments/<int:comment_id>/edit/', views.announcement_edit_comment, name='edit_comment'),
path('announcements/<int:pk>/comments/<int:comment_id>/delete/', views.announcement_delete_comment, name='delete_comment'),
]
nonpacks/announcements/views.py
+173
View File
@@ -0,0 +1,173 @@
from django.contrib import messages
from django.contrib.auth.decorators import login_required
from django.core.paginator import Paginator
from django.db.models import Count
from django.http import HttpResponseForbidden
from django.shortcuts import get_object_or_404, redirect, render
from django.urls import reverse
from common.markdown import render_markdown
from .forms import AnnouncementForm
from .models import Announcement, AnnouncementComment, AnnouncementRead, Notification
def _staff_only(user):
return user.is_authenticated and user.is_staff
def announcement_list(request):
"""Every published global announcement."""
announcements = (
Announcement.objects.filter(published=True)
.annotate(comments_count=Count('comments'))
.order_by('-created_at', '-pk')
)
paginator = Paginator(announcements, 10)
page = paginator.get_page(request.GET.get('page'))
read_ids = set()
if request.user.is_authenticated:
read_ids = set(
AnnouncementRead.objects.filter(user=request.user)
.values_list('announcement_id', flat=True)
)
return render(request, 'announcements/announcement_list.html', {
'page': page,
'read_ids': read_ids,
})
def announcement_detail(request, pk):
"""One announcement + its comment section. Marks it read on view."""
announcement = get_object_or_404(
Announcement.objects.annotate(comments_count=Count('comments')),
pk=pk, published=True,
)
if request.user.is_authenticated:
AnnouncementRead.objects.get_or_create(
announcement=announcement, user=request.user,
)
comments = announcement.comments.select_related('user').all()
return render(request, 'announcements/announcement_detail.html', {
'announcement': announcement,
'comments': comments,
'can_comment': request.user.is_authenticated,
})
@login_required
def announcement_create(request):
if not _staff_only(request.user):
return HttpResponseForbidden('Only staff can manage announcements.')
if request.method == 'POST':
form = AnnouncementForm(request.POST)
if form.is_valid():
form.instance.created_by = request.user
form.save()
messages.success(request, 'Announcement published.')
return redirect('announcements:detail', pk=form.instance.pk)
else:
form = AnnouncementForm()
return render(request, 'announcements/announcement_form.html', {
'form': form,
'editing': False,
})
@login_required
def announcement_edit(request, pk):
if not _staff_only(request.user):
return HttpResponseForbidden('Only staff can manage announcements.')
announcement = get_object_or_404(Announcement, pk=pk)
if request.method == 'POST':
form = AnnouncementForm(request.POST, instance=announcement)
if form.is_valid():
form.save()
messages.success(request, 'Announcement updated.')
return redirect('announcements:detail', pk=announcement.pk)
else:
form = AnnouncementForm(instance=announcement)
return render(request, 'announcements/announcement_form.html', {
'form': form,
'announcement': announcement,
'editing': True,
})
@login_required
def announcement_delete(request, pk):
if not _staff_only(request.user):
return HttpResponseForbidden('Only staff can manage announcements.')
announcement = get_object_or_404(Announcement, pk=pk)
if request.method == 'POST':
announcement.delete()
messages.success(request, 'Announcement deleted.')
return redirect('announcements:list')
@login_required
def announcement_add_comment(request, pk):
announcement = get_object_or_404(Announcement, pk=pk)
if request.method == 'POST':
body = request.POST.get('body', '').strip()
if body:
AnnouncementComment.objects.create(
announcement=announcement, user=request.user, body=body,
)
messages.success(request, 'Comment added.')
else:
messages.error(request, 'Comment cannot be empty.')
return redirect('announcements:detail', pk=announcement.pk)
@login_required
def announcement_edit_comment(request, pk, comment_id):
announcement = get_object_or_404(Announcement, pk=pk)
comment = get_object_or_404(
AnnouncementComment, pk=comment_id, announcement=announcement,
)
if comment.user_id != request.user.pk and not request.user.is_staff:
return HttpResponseForbidden('You can only edit your own comments.')
if request.method == 'POST':
body = request.POST.get('body', '').strip()
if body:
comment.body = body
comment.save(update_fields=['body', 'updated_at'])
messages.success(request, 'Comment updated.')
else:
messages.error(request, 'Comment cannot be empty.')
return redirect('announcements:detail', pk=announcement.pk)
@login_required
def announcement_delete_comment(request, pk, comment_id):
announcement = get_object_or_404(Announcement, pk=pk)
comment = get_object_or_404(
AnnouncementComment, pk=comment_id, announcement=announcement,
)
if comment.user_id != request.user.pk and not request.user.is_staff:
return HttpResponseForbidden('You can only delete your own comments.')
if request.method == 'POST':
comment.delete()
messages.success(request, 'Comment deleted.')
return redirect('announcements:detail', pk=announcement.pk)
@login_required
def personal_notifications(request):
"""The user's own personal announcement page."""
notifications = Notification.objects.filter(user=request.user)
paginator = Paginator(notifications, 20)
page = paginator.get_page(request.GET.get('page'))
return render(request, 'announcements/notifications.html', {
'page': page,
})
@login_required
def mark_all_notifications_read(request):
if request.method == 'POST':
Notification.objects.filter(user=request.user, read=False).update(read=True)
messages.success(request, 'All notifications marked as read.')
return redirect(request.POST.get('next') or reverse('announcements:personal'))