mirror of
https://github.com/boostorg/website-v2.git
synced 2026-01-19 04:42:17 +00:00
130 lines
4.2 KiB
Python
130 lines
4.2 KiB
Python
import csv
|
|
import logging
|
|
import re
|
|
from datetime import datetime
|
|
from io import TextIOWrapper
|
|
|
|
from django import forms
|
|
from django.shortcuts import redirect, render
|
|
from django.urls import path
|
|
from django.http import HttpResponseRedirect
|
|
from django.contrib import admin, messages
|
|
from django.conf import settings
|
|
|
|
from mailing_list.models import EmailData, SubscriptionData, ListPosting
|
|
from mailing_list.tasks import sync_mailinglist_stats
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@admin.register(EmailData)
|
|
class EmailDataAdmin(admin.ModelAdmin):
|
|
list_display = ["author", "version", "count"]
|
|
search_fields = [
|
|
"author__commitauthoremail__email",
|
|
"author__name",
|
|
]
|
|
raw_id_fields = ["author"]
|
|
list_filter = ["version"]
|
|
change_list_template = "admin/mailinglist_change_list.html"
|
|
|
|
def get_urls(self):
|
|
urls = super().get_urls()
|
|
my_urls = [
|
|
path(
|
|
"sync_mailinglist_stats/",
|
|
self.admin_site.admin_view(self.sync_mailinglist_stats),
|
|
name="sync_mailinglist_stats",
|
|
),
|
|
]
|
|
return my_urls + urls
|
|
|
|
def sync_mailinglist_stats(self, request):
|
|
if settings.HYPERKITTY_DATABASE_NAME:
|
|
sync_mailinglist_stats.delay()
|
|
self.message_user(request, "Syncing EmailData.")
|
|
else:
|
|
self.message_user(
|
|
request,
|
|
"HYPERKITTY_DATABASE_NAME setting not configured.",
|
|
level=messages.WARNING,
|
|
)
|
|
return HttpResponseRedirect("../")
|
|
|
|
def has_add_permission(self, request):
|
|
return False
|
|
|
|
|
|
class SubscribesCSVForm(forms.Form):
|
|
csv_file = forms.FileField()
|
|
|
|
|
|
@admin.register(SubscriptionData)
|
|
class SubscriptionDataAdmin(admin.ModelAdmin):
|
|
list_display = ["subscription_dt", "email"]
|
|
search_fields = ["email"]
|
|
change_list_template = "admin/mailinglist_change_list.html"
|
|
|
|
email_regex = re.compile("([a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,})")
|
|
|
|
def get_urls(self):
|
|
return [
|
|
path("import-csv", self.import_csv, name="import_csv")
|
|
] + super().get_urls()
|
|
|
|
def parse_rows(self, reader):
|
|
for row in reader:
|
|
date_str = " ".join(row[0:4])
|
|
try:
|
|
dt = datetime.strptime(date_str, "%b %d %H:%M:%S %Y")
|
|
except ValueError:
|
|
logger.error(f"Error parsing date {date_str} from {row=}")
|
|
dt = None
|
|
# re-merge, the email address isn't always in a consistent position
|
|
email_matches = re.search(self.email_regex, " ".join(row[6:]))
|
|
email = email_matches.group(0) if email_matches else None
|
|
entry_type = row[6]
|
|
# only save confirmed subscriber entries, it's all we need for now
|
|
if entry_type != "new":
|
|
continue
|
|
if not email:
|
|
logger.error(
|
|
f"Invalid email {row=} {email_matches=} {' '.join(row[6:])=}"
|
|
)
|
|
continue
|
|
yield SubscriptionData(
|
|
email=email,
|
|
entry_type=entry_type,
|
|
list=row[5].rstrip(":-1"),
|
|
subscription_dt=dt,
|
|
)
|
|
|
|
def import_csv(self, request):
|
|
if request.method == "POST":
|
|
csv_file = request.FILES["csv_file"]
|
|
rows = TextIOWrapper(csv_file, encoding="ISO-8859-1", newline="")
|
|
reader = csv.reader(rows, delimiter=" ")
|
|
SubscriptionData.objects.bulk_create(
|
|
self.parse_rows(reader), batch_size=500, ignore_conflicts=True
|
|
)
|
|
self.message_user(request, "Subscribe CSV file imported.")
|
|
return redirect("..")
|
|
|
|
payload = {"form": SubscribesCSVForm()}
|
|
return render(request, "admin/mailinglist_subscribe_csv_form.html", payload)
|
|
|
|
|
|
@admin.register(ListPosting)
|
|
class ListPostingAdmin(admin.ModelAdmin):
|
|
list_display = ["id", "date", "sender_id"]
|
|
search_fields = ["sender_id"]
|
|
|
|
def has_add_permission(self, request):
|
|
return False
|
|
|
|
def has_change_permission(self, request, obj=None):
|
|
return False
|
|
|
|
def has_delete_permission(self, request, obj=None):
|
|
return False
|