warrenchen/innovedus_cms
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add contact form spam processing and rate limiting

Browse Source
This commit is contained in:
warrenchen 2026-08-05 16:10:24 +09:00
parent 555462c71a
commit 61a67b8d4a
13 changed files with 1281 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
docs/contact_form_processing.md Normal file
View File

@ -0,0 +1,40 @@
# Contact form processing
Contact form submissions are stored first and processed asynchronously by:
```bash
python manage.py run_contact_submission_worker
```
The `scheduler` application role runs this worker and the newsletter scheduler in independent loops. A database lease allows only one spam-detection request at a time across scheduler replicas, while newsletter and SMTP work continue independently.
## Environment variables
Required in production:
```text
CONTACT_SPAM_DETECTION_ENABLED=true
CONTACT_SPAM_DETECTION_URL=https://spamd.innovedus.com/classify
CONTACT_SPAM_DETECTION_API_KEY=replace-me
```
Optional tuning:
```text
CONTACT_SPAM_DETECTION_TIMEOUT_SECONDS=180
CONTACT_SPAM_DETECTION_RETRY_DELAY_SECONDS=300
CONTACT_SPAM_DETECTION_MAX_ATTEMPTS=3
CONTACT_FORM_WORKER_BATCH_SIZE=20
CONTACT_FORM_WORKER_MAX_RUNTIME_SECONDS=600
CONTACT_FORM_WORKER_LEASE_SECONDS=600
CONTACT_FORM_MAIL_RETRY_DELAY_SECONDS=300
CONTACT_FORM_MAIL_MAX_ATTEMPTS=3
CONTACT_SCHEDULER_INTERVAL_SECONDS=60
NEWSLETTER_SCHEDULER_INTERVAL_SECONDS=60
```
When spam detection is disabled or its URL/key is missing, queued submissions stay in the database and no contact email is sent. A service error opens a database-backed cooldown until the configured retry time, so later scheduler runs do not repeatedly call an AI box that is restarting.
Rate limit count/window and the submitter-copy switch are managed in Wagtail under Contact Form Settings. The default is three submissions per ten minutes per IP, and submitter copies are disabled by default.
The application must only be publicly reachable through an Application Load Balancer whose `routing.http.xff_header_processing.mode` is `append` (the AWS default). The client IP is read from the rightmost valid `X-Forwarded-For` entry, with `REMOTE_ADDR` used only when that header is absent.

41
innovedus_cms/base/admin.py
View File

@ -4,9 +4,10 @@ from .models import ContactFormSubmission
@admin.register(ContactFormSubmission)
class ContactFormSubmissionAdmin(admin.ModelAdmin):
list_display = ("created_at", "category", "name", "email", "contact")
list_filter = ("category", "created_at")
search_fields = ("name", "email", "contact", "message", "source_page")
list_display = ("created_at", "processing_status", "spam_category", "category", "name", "email", "ip_address")
list_filter = ("processing_status", "spam_category", "category", "created_at")
search_fields = ("name", "email", "contact", "message", "source_page", "ip_address")
actions = ("requeue_submissions",)
readonly_fields = (
"name",
"email",
@ -16,5 +17,39 @@ class ContactFormSubmissionAdmin(admin.ModelAdmin):
"source_page",
"ip_address",
"user_agent",
"processing_status",
"spam_category",
"spam_source",
"spam_checked_at",
"scan_attempt_count",
"send_attempt_count",
"next_attempt_at",
"processing_started_at",
"notification_sent_at",
"user_copy_sent_at",
"last_error",
"created_at",
"updated_at",
)
@admin.action(description="Requeue selected contact submissions")
def requeue_submissions(self, request, queryset):
eligible = queryset.filter(
processing_status__in=[
ContactFormSubmission.STATUS_BLOCKED,
ContactFormSubmission.STATUS_FAILED,
ContactFormSubmission.STATUS_RATE_LIMITED,
]
)
updated = eligible.update(
processing_status=ContactFormSubmission.STATUS_QUEUED,
spam_category=None,
spam_source="",
spam_checked_at=None,
scan_attempt_count=0,
send_attempt_count=0,
next_attempt_at=None,
processing_started_at=None,
last_error="",
)
self.message_user(request, f"Requeued {updated} contact submission(s).")

437
innovedus_cms/base/contact_submission_processor.py Normal file
View File

@ -0,0 +1,437 @@
import json
import logging
import time
import uuid
from dataclasses import dataclass
from datetime import timedelta
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from django.conf import settings
from django.db import transaction
from django.db.models import Q
from django.utils import timezone
from django.utils.html import escape
from .models import (
ContactFormSettings,
ContactFormSubmission,
ContactSpamDetectionState,
MailSmtpSettings,
SystemNotificationMailSettings,
)
from .newsletter import send_contact_notification_email, send_contact_user_email
logger = logging.getLogger(__name__)
class SpamDetectionError(Exception):
pass
@dataclass(frozen=True)
class SpamDetectionResult:
category: int
source: str
class SpamDetectionClient:
def __init__(self):
self.url = settings.CONTACT_SPAM_DETECTION_URL
self.api_key = settings.CONTACT_SPAM_DETECTION_API_KEY
self.timeout_seconds = max(1, int(settings.CONTACT_SPAM_DETECTION_TIMEOUT_SECONDS))
def validate_config(self):
if not self.url:
raise SpamDetectionError("CONTACT_SPAM_DETECTION_URL is not configured")
if not self.api_key:
raise SpamDetectionError("CONTACT_SPAM_DETECTION_API_KEY is not configured")
def classify(self, message: str) -> SpamDetectionResult:
self.validate_config()
request = Request(
self.url,
data=json.dumps({"message": message}).encode("utf-8"),
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"Accept": "application/json",
},
method="POST",
)
try:
with urlopen(request, timeout=self.timeout_seconds) as response:
response_data = json.loads(response.read().decode("utf-8"))
except HTTPError as exc:
try:
detail = exc.read().decode("utf-8")
except Exception:
detail = ""
raise SpamDetectionError(f"spam detection HTTP {exc.code}: {detail[:500]}") from exc
except (URLError, TimeoutError, OSError) as exc:
raise SpamDetectionError(f"spam detection request failed: {exc}") from exc
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise SpamDetectionError("spam detection returned invalid JSON") from exc
if not isinstance(response_data, dict):
raise SpamDetectionError("spam detection response must be an object")
if response_data.get("error"):
raise SpamDetectionError(f"spam detection service error: {response_data['error']}")
category = response_data.get("category")
if type(category) is not int or category not in {0, 1, 2, 3}:
raise SpamDetectionError(f"spam detection returned invalid category: {category!r}")
return SpamDetectionResult(
category=category,
source=str(response_data.get("source") or "")[:32],
)
def _render_contact_template(template: str, values: dict[str, str]) -> str:
rendered = template or ""
for key, value in values.items():
rendered = rendered.replace(f"{{{{{key}}}}}", value)
return rendered
def _claim_mail_submission(now):
stale_before = now - timedelta(seconds=max(1, int(settings.CONTACT_FORM_WORKER_LEASE_SECONDS)))
eligible = Q(
processing_status=ContactFormSubmission.STATUS_SEND_RETRY,
next_attempt_at__lte=now,
) | Q(
processing_status=ContactFormSubmission.STATUS_SEND_RETRY,
next_attempt_at__isnull=True,
) | Q(
processing_status=ContactFormSubmission.STATUS_SENDING,
processing_started_at__lte=stale_before,
) | Q(
processing_status=ContactFormSubmission.STATUS_SENDING,
processing_started_at__isnull=True,
)
with transaction.atomic():
submission = (
ContactFormSubmission.objects.select_for_update(skip_locked=True)
.filter(eligible)
.order_by("created_at", "id")
.first()
)
if submission is None:
return None
submission.processing_status = ContactFormSubmission.STATUS_SENDING
submission.processing_started_at = now
submission.next_attempt_at = None
submission.save(update_fields=["processing_status", "processing_started_at", "next_attempt_at", "updated_at"])
return submission
def _claim_scan_submission(now):
stale_before = now - timedelta(seconds=max(1, int(settings.CONTACT_FORM_WORKER_LEASE_SECONDS)))
eligible = Q(processing_status=ContactFormSubmission.STATUS_QUEUED) | Q(
processing_status=ContactFormSubmission.STATUS_SCAN_RETRY,
next_attempt_at__lte=now,
) | Q(
processing_status=ContactFormSubmission.STATUS_SCAN_RETRY,
next_attempt_at__isnull=True,
) | Q(
processing_status=ContactFormSubmission.STATUS_SCANNING,
processing_started_at__lte=stale_before,
)
with transaction.atomic():
submission = (
ContactFormSubmission.objects.select_for_update(skip_locked=True)
.filter(eligible)
.order_by("created_at", "id")
.first()
)
if submission is None:
return None
submission.processing_status = ContactFormSubmission.STATUS_SCANNING
submission.processing_started_at = now
submission.next_attempt_at = None
submission.scan_attempt_count += 1
submission.save(
update_fields=[
"processing_status",
"processing_started_at",
"next_attempt_at",
"scan_attempt_count",
"updated_at",
]
)
return submission
def _acquire_detection_lease(now):
with transaction.atomic():
state, _ = ContactSpamDetectionState.objects.get_or_create(pk=1)
state = ContactSpamDetectionState.objects.select_for_update().get(pk=state.pk)
if state.cooldown_until and state.cooldown_until > now:
return "", state.cooldown_until, False
if state.locked_until and state.locked_until > now:
return "", None, True
lease_seconds = max(
int(settings.CONTACT_FORM_WORKER_LEASE_SECONDS),
int(settings.CONTACT_SPAM_DETECTION_TIMEOUT_SECONDS) + 60,
)
token = uuid.uuid4().hex
state.lock_token = token
state.locked_until = now + timedelta(seconds=max(1, lease_seconds))
state.cooldown_until = None
state.save(update_fields=["lock_token", "locked_until", "cooldown_until", "updated_at"])
return token, None, False
def _release_detection_lease(token: str, *, cooldown_until=None, error=""):
ContactSpamDetectionState.objects.filter(pk=1, lock_token=token).update(
lock_token="",
locked_until=None,
cooldown_until=cooldown_until,
last_error=error[:4000],
updated_at=timezone.now(),
)
def _record_scan_failure(submission: ContactFormSubmission, error: str):
now = timezone.now()
retry_at = now + timedelta(seconds=max(1, int(settings.CONTACT_SPAM_DETECTION_RETRY_DELAY_SECONDS)))
if submission.scan_attempt_count >= max(1, int(settings.CONTACT_SPAM_DETECTION_MAX_ATTEMPTS)):
submission.processing_status = ContactFormSubmission.STATUS_FAILED
else:
submission.processing_status = ContactFormSubmission.STATUS_SCAN_RETRY
submission.processing_started_at = None
submission.next_attempt_at = retry_at
submission.last_error = error[:4000]
submission.save(
update_fields=[
"processing_status",
"processing_started_at",
"next_attempt_at",
"last_error",
"updated_at",
]
)
return retry_at
def _scan_submission(submission: ContactFormSubmission, client: SpamDetectionClient):
result = client.classify(submission.message)
now = timezone.now()
submission.spam_category = result.category
submission.spam_source = result.source
submission.spam_checked_at = now
submission.next_attempt_at = None
submission.last_error = ""
if result.category == 0:
submission.processing_status = ContactFormSubmission.STATUS_SENDING
submission.processing_started_at = now
else:
submission.processing_status = ContactFormSubmission.STATUS_BLOCKED
submission.processing_started_at = None
submission.save(
update_fields=[
"spam_category",
"spam_source",
"spam_checked_at",
"processing_status",
"processing_started_at",
"next_attempt_at",
"last_error",
"updated_at",
]
)
return result
def _notification_content(submission: ContactFormSubmission, notification_settings):
subject_prefix = (notification_settings.contact_form_subject_prefix or "").strip()
subject = f"{subject_prefix} {submission.get_category_display()}".strip()
escaped_message_html = escape(submission.message).replace("\n", "<br>")
text_body = (
f"Name: {submission.name}\n"
f"Email: {submission.email}\n"
f"Contact: {submission.contact}\n"
f"Category: {submission.get_category_display()}\n"
f"Source Page: {submission.source_page}\n"
f"IP Address: {submission.ip_address or ''}\n\n"
f"Message:\n{submission.message}\n"
)
html_body = (
f"<p><strong>Name:</strong> {escape(submission.name)}</p>"
f"<p><strong>Email:</strong> {escape(submission.email)}</p>"
f"<p><strong>Contact:</strong> {escape(submission.contact)}</p>"
f"<p><strong>Category:</strong> {escape(submission.get_category_display())}</p>"
f"<p><strong>Source Page:</strong> {escape(submission.source_page or '')}</p>"
f"<p><strong>IP Address:</strong> {escape(str(submission.ip_address or ''))}</p>"
f"<p><strong>Message:</strong></p><p>{escaped_message_html}</p>"
)
return subject, text_body, html_body
def _user_copy_content(submission: ContactFormSubmission, notification_settings):
values_text = {
"name": submission.name,
"email": submission.email,
"contact": submission.contact,
"category": submission.get_category_display(),
"message": submission.message,
"source_page": submission.source_page or "",
}
values_html = {
"name": escape(submission.name),
"email": escape(submission.email),
"contact": escape(submission.contact),
"category": escape(submission.get_category_display()),
"message": escape(submission.message).replace("\n", "<br>"),
"source_page": escape(submission.source_page or ""),
}
return (
_render_contact_template(notification_settings.contact_form_user_subject_template, values_text),
_render_contact_template(notification_settings.contact_form_user_text_template, values_text),
_render_contact_template(notification_settings.contact_form_user_html_template, values_html),
)
def _send_submission(submission: ContactFormSubmission):
now = timezone.now()
submission.send_attempt_count += 1
submission.processing_status = ContactFormSubmission.STATUS_SENDING
submission.processing_started_at = now
submission.save(
update_fields=["send_attempt_count", "processing_status", "processing_started_at", "updated_at"]
)
notification_settings = SystemNotificationMailSettings.load()
smtp_settings = MailSmtpSettings.load()
contact_settings = ContactFormSettings.load()
try:
if submission.notification_sent_at is None:
subject, text_body, html_body = _notification_content(submission, notification_settings)
send_contact_notification_email(
subject=subject,
text_body=text_body,
html_body=html_body,
notification_config=notification_settings,
smtp_config=smtp_settings,
)
submission.notification_sent_at = timezone.now()
submission.save(update_fields=["notification_sent_at", "updated_at"])
if contact_settings.send_user_copy and submission.email and submission.user_copy_sent_at is None:
user_subject, user_text, user_html = _user_copy_content(submission, notification_settings)
send_contact_user_email(
to_email=submission.email,
subject=user_subject,
text_body=user_text,
html_body=user_html,
notification_config=notification_settings,
smtp_config=smtp_settings,
)
submission.user_copy_sent_at = timezone.now()
submission.save(update_fields=["user_copy_sent_at", "updated_at"])
except Exception as exc:
logger.warning("contact submission email failed submission_id=%s: %s", submission.pk, exc)
submission.last_error = str(exc)[:4000]
submission.processing_started_at = None
delay = max(1, int(settings.CONTACT_FORM_MAIL_RETRY_DELAY_SECONDS))
submission.next_attempt_at = timezone.now() + timedelta(seconds=delay)
if submission.send_attempt_count >= max(1, int(settings.CONTACT_FORM_MAIL_MAX_ATTEMPTS)):
submission.processing_status = ContactFormSubmission.STATUS_FAILED
else:
submission.processing_status = ContactFormSubmission.STATUS_SEND_RETRY
submission.save(
update_fields=[
"processing_status",
"processing_started_at",
"next_attempt_at",
"last_error",
"updated_at",
]
)
return False
submission.processing_status = ContactFormSubmission.STATUS_SENT
submission.processing_started_at = None
submission.next_attempt_at = None
submission.last_error = ""
submission.save(
update_fields=[
"processing_status",
"processing_started_at",
"next_attempt_at",
"last_error",
"updated_at",
]
)
return True
def process_contact_submissions(*, limit=None, max_runtime_seconds=None) -> dict:
limit = max(1, int(limit or settings.CONTACT_FORM_WORKER_BATCH_SIZE))
max_runtime_seconds = max(
1,
int(max_runtime_seconds or settings.CONTACT_FORM_WORKER_MAX_RUNTIME_SECONDS),
)
result = {
"processed": 0,
"allowed": 0,
"blocked": 0,
"sent": 0,
"failed": 0,
"cooldown_until": None,
"detector_busy": False,
"disabled": not settings.CONTACT_SPAM_DETECTION_ENABLED,
}
started_at = time.monotonic()
client = SpamDetectionClient()
while result["processed"] < limit and time.monotonic() - started_at < max_runtime_seconds:
now = timezone.now()
submission = _claim_mail_submission(now)
if submission is not None:
result["processed"] += 1
if _send_submission(submission):
result["sent"] += 1
else:
result["failed"] += 1
continue
if not settings.CONTACT_SPAM_DETECTION_ENABLED:
break
lease_token, cooldown_until, detector_busy = _acquire_detection_lease(now)
if not lease_token:
result["cooldown_until"] = cooldown_until
result["detector_busy"] = detector_busy
break
submission = _claim_scan_submission(now)
if submission is None:
_release_detection_lease(lease_token)
break
result["processed"] += 1
try:
scan_result = _scan_submission(submission, client)
except SpamDetectionError as exc:
logger.warning("contact spam detection failed submission_id=%s: %s", submission.pk, exc)
retry_at = _record_scan_failure(submission, str(exc))
_release_detection_lease(lease_token, cooldown_until=retry_at, error=str(exc))
result["failed"] += 1
result["cooldown_until"] = retry_at
break
_release_detection_lease(lease_token)
if scan_result.category != 0:
result["blocked"] += 1
continue
result["allowed"] += 1
if _send_submission(submission):
result["sent"] += 1
else:
result["failed"] += 1
return result

31
innovedus_cms/base/management/commands/run_contact_submission_worker.py Normal file
View File

@ -0,0 +1,31 @@
from django.conf import settings
from django.core.management.base import BaseCommand
from base.contact_submission_processor import process_contact_submissions
class Command(BaseCommand):
help = "Scan and send queued contact form submissions"
def add_arguments(self, parser):
parser.add_argument("--limit", type=int, default=settings.CONTACT_FORM_WORKER_BATCH_SIZE)
parser.add_argument(
"--max-runtime-seconds",
type=int,
default=settings.CONTACT_FORM_WORKER_MAX_RUNTIME_SECONDS,
)
def handle(self, *args, **options):
result = process_contact_submissions(
limit=options["limit"],
max_runtime_seconds=options["max_runtime_seconds"],
)
cooldown_until = result["cooldown_until"].isoformat() if result["cooldown_until"] else ""
self.stdout.write(
"processed={processed} allowed={allowed} blocked={blocked} sent={sent} "
"failed={failed} disabled={disabled} detector_busy={detector_busy} "
"cooldown_until={cooldown}".format(
**result,
cooldown=cooldown_until,
)
)

127
innovedus_cms/base/migrations/0014_contactformratelimitbucket_contactformsettings_and_more.py Normal file
View File

@ -0,0 +1,127 @@
# Generated by Django 5.2.15 on 2026-08-05 05:22
from django.db import migrations, models
def mark_existing_submissions_as_legacy(apps, schema_editor):
ContactFormSubmission = apps.get_model("base", "ContactFormSubmission")
ContactFormSubmission.objects.update(processing_status="legacy")
class Migration(migrations.Migration):
dependencies = [
('base', '0013_send_engine_oauth_client_credentials'),
]
operations = [
migrations.CreateModel(
name='ContactFormRateLimitBucket',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('ip_address', models.GenericIPAddressField(unique=True, verbose_name='IP Address')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='Updated At')),
],
options={
'verbose_name': 'Contact Form Rate Limit Bucket',
'verbose_name_plural': 'Contact Form Rate Limit Buckets',
},
),
migrations.CreateModel(
name='ContactFormSettings',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('rate_limit_enabled', models.BooleanField(default=True, verbose_name='Enable IP Rate Limit')),
('rate_limit_max_submissions', models.PositiveIntegerField(default=3, verbose_name='Maximum Submissions Per Window')),
('rate_limit_window_minutes', models.PositiveIntegerField(default=10, verbose_name='Rate Limit Window Minutes')),
('send_user_copy', models.BooleanField(default=False, help_text='Send a copy to the submitter only after spam detection allows the message.', verbose_name='Send User Copy')),
],
options={
'verbose_name': 'Contact Form Settings',
},
),
migrations.CreateModel(
name='ContactSpamDetectionState',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('lock_token', models.CharField(blank=True, max_length=64, verbose_name='Lock Token')),
('locked_until', models.DateTimeField(blank=True, null=True, verbose_name='Locked Until')),
('cooldown_until', models.DateTimeField(blank=True, null=True, verbose_name='Cooldown Until')),
('last_error', models.TextField(blank=True, verbose_name='Last Error')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='Updated At')),
],
options={
'verbose_name': 'Contact Spam Detection State',
'verbose_name_plural': 'Contact Spam Detection State',
},
),
migrations.AddField(
model_name='contactformsubmission',
name='last_error',
field=models.TextField(blank=True, verbose_name='Last Error'),
),
migrations.AddField(
model_name='contactformsubmission',
name='next_attempt_at',
field=models.DateTimeField(blank=True, db_index=True, null=True, verbose_name='Next Attempt At'),
),
migrations.AddField(
model_name='contactformsubmission',
name='notification_sent_at',
field=models.DateTimeField(blank=True, null=True, verbose_name='Notification Sent At'),
),
migrations.AddField(
model_name='contactformsubmission',
name='processing_started_at',
field=models.DateTimeField(blank=True, null=True, verbose_name='Processing Started At'),
),
migrations.AddField(
model_name='contactformsubmission',
name='processing_status',
field=models.CharField(choices=[('queued', 'Queued'), ('scanning', 'Scanning'), ('scan_retry', 'Scan Retry'), ('blocked', 'Blocked as Spam'), ('sending', 'Sending'), ('send_retry', 'Send Retry'), ('sent', 'Sent'), ('failed', 'Failed'), ('rate_limited', 'Rate Limited'), ('legacy', 'Legacy Submission')], db_index=True, default='queued', max_length=20, verbose_name='Processing Status'),
),
migrations.AddField(
model_name='contactformsubmission',
name='scan_attempt_count',
field=models.PositiveIntegerField(default=0, verbose_name='Scan Attempt Count'),
),
migrations.AddField(
model_name='contactformsubmission',
name='send_attempt_count',
field=models.PositiveIntegerField(default=0, verbose_name='Send Attempt Count'),
),
migrations.AddField(
model_name='contactformsubmission',
name='spam_category',
field=models.PositiveSmallIntegerField(blank=True, choices=[(0, 'Allowed'), (1, 'Unsolicited Advertising'), (2, 'Malicious'), (3, 'Meaningless')], null=True, verbose_name='Spam Category'),
),
migrations.AddField(
model_name='contactformsubmission',
name='spam_checked_at',
field=models.DateTimeField(blank=True, null=True, verbose_name='Spam Checked At'),
),
migrations.AddField(
model_name='contactformsubmission',
name='spam_source',
field=models.CharField(blank=True, max_length=32, verbose_name='Spam Detection Source'),
),
migrations.AddField(
model_name='contactformsubmission',
name='updated_at',
field=models.DateTimeField(auto_now=True, verbose_name='Updated At'),
),
migrations.AddField(
model_name='contactformsubmission',
name='user_copy_sent_at',
field=models.DateTimeField(blank=True, null=True, verbose_name='User Copy Sent At'),
),
migrations.AddIndex(
model_name='contactformsubmission',
index=models.Index(fields=['ip_address', 'created_at'], name='contact_ip_created_idx'),
),
migrations.AddIndex(
model_name='contactformsubmission',
index=models.Index(fields=['processing_status', 'next_attempt_at', 'created_at'], name='contact_process_due_idx'),
),
migrations.RunPython(mark_existing_submissions_as_legacy, migrations.RunPython.noop),
]

114
innovedus_cms/base/models.py
View File

@ -171,6 +171,39 @@ class MailSmtpSettings(BaseGenericSetting):
super().save(*args, **kwargs)
@register_setting
class ContactFormSettings(BaseGenericSetting):
rate_limit_enabled = models.BooleanField(default=True, verbose_name=_("Enable IP Rate Limit"))
rate_limit_max_submissions = models.PositiveIntegerField(
default=3,
verbose_name=_("Maximum Submissions Per Window"),
)
rate_limit_window_minutes = models.PositiveIntegerField(
default=10,
verbose_name=_("Rate Limit Window Minutes"),
)
send_user_copy = models.BooleanField(
default=False,
help_text=_("Send a copy to the submitter only after spam detection allows the message."),
verbose_name=_("Send User Copy"),
)
panels = [
MultiFieldPanel(
[
FieldPanel("rate_limit_enabled"),
FieldPanel("rate_limit_max_submissions"),
FieldPanel("rate_limit_window_minutes"),
],
heading=_("Contact Form Rate Limit"),
),
FieldPanel("send_user_copy"),
]
class Meta:
verbose_name = _("Contact Form Settings")
@register_setting
class NewsletterSystemSettings(BaseGenericSetting):
member_center_base_url = models.URLField(blank=True, verbose_name=_("Member Center Base URL"))
@ -804,6 +837,36 @@ class ContactFormSubmission(models.Model):
(CATEGORY_OTHER, _("Other")),
]
STATUS_QUEUED = "queued"
STATUS_SCANNING = "scanning"
STATUS_SCAN_RETRY = "scan_retry"
STATUS_BLOCKED = "blocked"
STATUS_SENDING = "sending"
STATUS_SEND_RETRY = "send_retry"
STATUS_SENT = "sent"
STATUS_FAILED = "failed"
STATUS_RATE_LIMITED = "rate_limited"
STATUS_LEGACY = "legacy"
STATUS_CHOICES = [
(STATUS_QUEUED, _("Queued")),
(STATUS_SCANNING, _("Scanning")),
(STATUS_SCAN_RETRY, _("Scan Retry")),
(STATUS_BLOCKED, _("Blocked as Spam")),
(STATUS_SENDING, _("Sending")),
(STATUS_SEND_RETRY, _("Send Retry")),
(STATUS_SENT, _("Sent")),
(STATUS_FAILED, _("Failed")),
(STATUS_RATE_LIMITED, _("Rate Limited")),
(STATUS_LEGACY, _("Legacy Submission")),
]
SPAM_CATEGORY_CHOICES = [
(0, _("Allowed")),
(1, _("Unsolicited Advertising")),
(2, _("Malicious")),
(3, _("Meaningless")),
]
name = models.CharField(max_length=100, verbose_name=_("Name"))
email = models.EmailField(blank=True, verbose_name=_("Email"))
contact = models.CharField(max_length=255, verbose_name=_("Contact"))
@ -812,12 +875,63 @@ class ContactFormSubmission(models.Model):
source_page = models.CharField(max_length=512, blank=True, verbose_name=_("Source Page"))
ip_address = models.GenericIPAddressField(null=True, blank=True, verbose_name=_("IP Address"))
user_agent = models.TextField(blank=True, verbose_name=_("User Agent"))
processing_status = models.CharField(
max_length=20,
choices=STATUS_CHOICES,
default=STATUS_QUEUED,
db_index=True,
verbose_name=_("Processing Status"),
)
spam_category = models.PositiveSmallIntegerField(
choices=SPAM_CATEGORY_CHOICES,
null=True,
blank=True,
verbose_name=_("Spam Category"),
)
spam_source = models.CharField(max_length=32, blank=True, verbose_name=_("Spam Detection Source"))
spam_checked_at = models.DateTimeField(null=True, blank=True, verbose_name=_("Spam Checked At"))
scan_attempt_count = models.PositiveIntegerField(default=0, verbose_name=_("Scan Attempt Count"))
send_attempt_count = models.PositiveIntegerField(default=0, verbose_name=_("Send Attempt Count"))
next_attempt_at = models.DateTimeField(null=True, blank=True, db_index=True, verbose_name=_("Next Attempt At"))
processing_started_at = models.DateTimeField(null=True, blank=True, verbose_name=_("Processing Started At"))
notification_sent_at = models.DateTimeField(null=True, blank=True, verbose_name=_("Notification Sent At"))
user_copy_sent_at = models.DateTimeField(null=True, blank=True, verbose_name=_("User Copy Sent At"))
last_error = models.TextField(blank=True, verbose_name=_("Last Error"))
created_at = models.DateTimeField(auto_now_add=True, verbose_name=_("Created At"))
updated_at = models.DateTimeField(auto_now=True, verbose_name=_("Updated At"))
class Meta:
ordering = ["-created_at"]
indexes = [
models.Index(fields=["ip_address", "created_at"], name="contact_ip_created_idx"),
models.Index(
fields=["processing_status", "next_attempt_at", "created_at"],
name="contact_process_due_idx",
),
]
verbose_name = _("Contact Form Submission")
verbose_name_plural = _("Contact Form Submissions")
def __str__(self):
return f"{self.get_category_display()} - {self.name}"
class ContactFormRateLimitBucket(models.Model):
ip_address = models.GenericIPAddressField(unique=True, verbose_name=_("IP Address"))
updated_at = models.DateTimeField(auto_now=True, verbose_name=_("Updated At"))
class Meta:
verbose_name = _("Contact Form Rate Limit Bucket")
verbose_name_plural = _("Contact Form Rate Limit Buckets")
class ContactSpamDetectionState(models.Model):
lock_token = models.CharField(max_length=64, blank=True, verbose_name=_("Lock Token"))
locked_until = models.DateTimeField(null=True, blank=True, verbose_name=_("Locked Until"))
cooldown_until = models.DateTimeField(null=True, blank=True, verbose_name=_("Cooldown Until"))
last_error = models.TextField(blank=True, verbose_name=_("Last Error"))
updated_at = models.DateTimeField(auto_now=True, verbose_name=_("Updated At"))
class Meta:
verbose_name = _("Contact Spam Detection State")
verbose_name_plural = _("Contact Spam Detection State")

199
innovedus_cms/base/tests.py
View File

@ -1,11 +1,19 @@
from datetime import date
import json
from datetime import date, timedelta
from types import SimpleNamespace
from urllib.parse import parse_qs
from unittest.mock import Mock, patch
from django.test import TestCase
from django.test import TestCase, override_settings
from django.urls import reverse
from django.utils import timezone
from .contact_submission_processor import (
SpamDetectionClient,
SpamDetectionError,
SpamDetectionResult,
process_contact_submissions,
)
from .newsletter import (
APIResult,
MemberCenterClient,
@ -17,7 +25,12 @@ from .newsletter import (
render_newsletter_html,
verify_one_click_token,
)
from .models import ContactFormSubmission, SystemNotificationMailSettings
from .models import (
ContactFormSettings,
ContactFormSubmission,
ContactSpamDetectionState,
SystemNotificationMailSettings,
)
from .security import decrypt_text, encrypt_text
@ -265,6 +278,7 @@ class NewsletterTemplateTests(TestCase):
self.assertEqual(submission.name, "Tester")
self.assertEqual(submission.email, "tester@example.com")
self.assertEqual(submission.category, "other")
self.assertEqual(submission.processing_status, ContactFormSubmission.STATUS_QUEUED)
def test_contact_form_submit_rejects_invalid_email(self):
response = self.client.post(
@ -281,3 +295,182 @@ class NewsletterTemplateTests(TestCase):
)
self.assertEqual(response.status_code, 400)
self.assertEqual(ContactFormSubmission.objects.count(), 0)
class ContactFormSubmissionFlowTests(TestCase):
def _post(self, *, remote_addr="10.0.1.10", forwarded_for="198.51.100.20"):
return self.client.post(
reverse("contact_form_submit"),
data={
"name": "Tester",
"contact": "tester@example.com",
"email": "tester@example.com",
"category": "other",
"message": "hello",
"source_page": "/contact/",
},
REMOTE_ADDR=remote_addr,
HTTP_X_FORWARDED_FOR=forwarded_for,
HTTP_X_REQUESTED_WITH="XMLHttpRequest",
)
def test_records_rightmost_ip_appended_by_alb(self):
response = self._post(forwarded_for="203.0.113.99, 198.51.100.20")
self.assertEqual(response.status_code, 200)
self.assertEqual(ContactFormSubmission.objects.get().ip_address, "198.51.100.20")
def test_falls_back_to_remote_address_without_forwarded_header(self):
response = self._post(remote_addr="10.0.1.10", forwarded_for="")
self.assertEqual(response.status_code, 200)
self.assertEqual(ContactFormSubmission.objects.get().ip_address, "10.0.1.10")
def test_fourth_submission_in_default_window_is_rate_limited(self):
for _ in range(4):
response = self._post()
self.assertEqual(response.status_code, 200)
statuses = list(ContactFormSubmission.objects.order_by("created_at", "id").values_list("processing_status", flat=True))
self.assertEqual(
statuses,
[
ContactFormSubmission.STATUS_QUEUED,
ContactFormSubmission.STATUS_QUEUED,
ContactFormSubmission.STATUS_QUEUED,
ContactFormSubmission.STATUS_RATE_LIMITED,
],
)
@override_settings(
CONTACT_SPAM_DETECTION_ENABLED=True,
CONTACT_SPAM_DETECTION_URL="https://spam.example.com/classify",
CONTACT_SPAM_DETECTION_API_KEY="test-key",
CONTACT_SPAM_DETECTION_RETRY_DELAY_SECONDS=300,
CONTACT_SPAM_DETECTION_MAX_ATTEMPTS=3,
CONTACT_FORM_MAIL_RETRY_DELAY_SECONDS=300,
CONTACT_FORM_MAIL_MAX_ATTEMPTS=3,
)
class ContactSubmissionProcessorTests(TestCase):
def _submission(self, **overrides):
values = {
"name": "Tester",
"contact": "tester@example.com",
"email": "tester@example.com",
"category": ContactFormSubmission.CATEGORY_OTHER,
"message": "hello",
"source_page": "/contact/",
"ip_address": "198.51.100.20",
}
values.update(overrides)
return ContactFormSubmission.objects.create(**values)
@patch("base.contact_submission_processor.send_contact_user_email")
@patch("base.contact_submission_processor.send_contact_notification_email")
@patch("base.contact_submission_processor.SpamDetectionClient.classify")
def test_allowed_submission_is_sent_after_scan(self, classify, send_notification, send_user):
submission = self._submission()
classify.return_value = SpamDetectionResult(category=0, source="kneo300")
result = process_contact_submissions(limit=1)
submission.refresh_from_db()
self.assertEqual(result["sent"], 1)
self.assertEqual(submission.processing_status, ContactFormSubmission.STATUS_SENT)
self.assertEqual(submission.spam_category, 0)
self.assertIsNotNone(submission.notification_sent_at)
send_notification.assert_called_once()
send_user.assert_not_called()
@patch("base.contact_submission_processor.send_contact_notification_email")
@patch("base.contact_submission_processor.SpamDetectionClient.classify")
def test_spam_submission_is_recorded_and_not_sent(self, classify, send_notification):
submission = self._submission()
classify.return_value = SpamDetectionResult(category=2, source="kneo300")
result = process_contact_submissions(limit=1)
submission.refresh_from_db()
self.assertEqual(result["blocked"], 1)
self.assertEqual(submission.processing_status, ContactFormSubmission.STATUS_BLOCKED)
self.assertEqual(submission.spam_category, 2)
send_notification.assert_not_called()
@patch("base.contact_submission_processor.send_contact_notification_email")
@patch("base.contact_submission_processor.SpamDetectionClient.classify")
def test_detection_error_schedules_cooldown_and_does_not_send(self, classify, send_notification):
submission = self._submission()
classify.side_effect = SpamDetectionError("AI box unavailable")
result = process_contact_submissions(limit=1)
submission.refresh_from_db()
self.assertEqual(result["failed"], 1)
self.assertEqual(submission.processing_status, ContactFormSubmission.STATUS_SCAN_RETRY)
self.assertGreater(submission.next_attempt_at, timezone.now())
send_notification.assert_not_called()
@patch("base.contact_submission_processor.SpamDetectionClient.classify")
def test_active_detection_cooldown_prevents_calls_for_other_rows(self, classify):
ContactSpamDetectionState.objects.create(
pk=1,
cooldown_until=timezone.now() + timedelta(minutes=5),
last_error="AI box unavailable",
)
self._submission(email="second@example.com")
result = process_contact_submissions(limit=2)
self.assertEqual(result["processed"], 0)
self.assertIsNotNone(result["cooldown_until"])
classify.assert_not_called()
@patch("base.contact_submission_processor.SpamDetectionClient.classify")
def test_active_detection_lease_prevents_parallel_ai_calls(self, classify):
ContactSpamDetectionState.objects.create(
pk=1,
lock_token="another-worker",
locked_until=timezone.now() + timedelta(minutes=5),
)
self._submission()
result = process_contact_submissions(limit=1)
self.assertEqual(result["processed"], 0)
self.assertTrue(result["detector_busy"])
classify.assert_not_called()
@patch("base.contact_submission_processor.send_contact_user_email")
@patch("base.contact_submission_processor.send_contact_notification_email")
@patch("base.contact_submission_processor.SpamDetectionClient.classify")
def test_mail_retry_does_not_repeat_successful_notification(self, classify, send_notification, send_user):
ContactFormSettings.objects.create(send_user_copy=True)
submission = self._submission()
classify.return_value = SpamDetectionResult(category=0, source="kneo300")
send_user.side_effect = [OSError("SMTP unavailable"), 1]
process_contact_submissions(limit=1)
submission.refresh_from_db()
self.assertEqual(submission.processing_status, ContactFormSubmission.STATUS_SEND_RETRY)
self.assertIsNotNone(submission.notification_sent_at)
submission.next_attempt_at = timezone.now() - timedelta(seconds=1)
submission.save(update_fields=["next_attempt_at", "updated_at"])
process_contact_submissions(limit=1)
submission.refresh_from_db()
self.assertEqual(submission.processing_status, ContactFormSubmission.STATUS_SENT)
send_notification.assert_called_once()
self.assertEqual(send_user.call_count, 2)
@patch("base.contact_submission_processor.urlopen")
def test_category_zero_with_service_error_is_not_accepted(self, urlopen_mock):
response = Mock()
response.read.return_value = json.dumps(
{"category": 0, "source": "kneo300", "error": "timeout"}
).encode("utf-8")
urlopen_mock.return_value.__enter__.return_value = response
with self.assertRaises(SpamDetectionError):
SpamDetectionClient().classify("hello")

159
innovedus_cms/base/views.py
View File

@ -2,6 +2,8 @@ import logging
from urllib.parse import urlencode
import hashlib
import json
from datetime import timedelta
from ipaddress import ip_address as parse_ip_address
from django.contrib import messages
from django.contrib.admin.views.decorators import staff_member_required
@ -9,14 +11,18 @@ from django.contrib.auth.decorators import login_required
from django.core.files.storage import default_storage
from django.core.exceptions import PermissionDenied, ValidationError
from django.core.validators import validate_email
from django.db import transaction
from django.http import HttpResponseNotAllowed, JsonResponse
from django.shortcuts import get_object_or_404, redirect, render
from django.urls import reverse
from django.utils.html import escape
from django.utils import timezone
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_GET, require_http_methods, require_POST
from .forms import ContactForm, NewsletterSubscribeForm, NewsletterUnsubscribeForm
from .models import (
ContactFormRateLimitBucket,
ContactFormSettings,
ContactFormSubmission,
MailSmtpSettings,
NewsletterCampaign,
NewsletterSystemSettings,
@ -32,8 +38,6 @@ from .newsletter import (
extract_token,
render_placeholders,
render_newsletter_html_for_send_job,
send_contact_notification_email,
send_contact_user_email,
send_subscribe_email,
)
from .newsletter_scheduler import dispatch_campaign
@ -84,11 +88,65 @@ def _build_context(*, title: str, message: str, success: bool):
}
def _render_contact_template(template: str, values: dict[str, str]) -> str:
rendered = template or ""
for key, value in values.items():
rendered = rendered.replace(f"{{{{{key}}}}}", value)
return rendered
def _normalize_ip(value: str):
candidate = (value or "").strip()
if not candidate:
return None
try:
return parse_ip_address(candidate)
except ValueError:
pass
if candidate.startswith("[") and "]:" in candidate:
candidate = candidate[1 : candidate.rfind("]")]
elif candidate.count(":") == 1:
candidate = candidate.rsplit(":", 1)[0]
try:
return parse_ip_address(candidate)
except ValueError:
return None
def _get_client_ip(request) -> str:
forwarded_ips = [
parsed
for item in request.META.get("HTTP_X_FORWARDED_FOR", "").split(",")
if (parsed := _normalize_ip(item)) is not None
]
if forwarded_ips:
return str(forwarded_ips[-1])
remote_ip = _normalize_ip(request.META.get("REMOTE_ADDR", ""))
return str(remote_ip) if remote_ip is not None else ""
def _save_contact_submission(*, form: ContactForm, request) -> ContactFormSubmission:
contact_settings = ContactFormSettings.load(request_or_site=request)
client_ip = _get_client_ip(request)
now = timezone.now()
rate_limited = False
with transaction.atomic():
if contact_settings.rate_limit_enabled and client_ip:
bucket, _ = ContactFormRateLimitBucket.objects.get_or_create(ip_address=client_ip)
ContactFormRateLimitBucket.objects.select_for_update().get(pk=bucket.pk)
window_minutes = max(1, int(contact_settings.rate_limit_window_minutes or 10))
max_submissions = max(0, int(contact_settings.rate_limit_max_submissions or 0))
recent_count = ContactFormSubmission.objects.filter(
ip_address=client_ip,
created_at__gte=now - timedelta(minutes=window_minutes),
).count()
rate_limited = recent_count >= max_submissions
submission = form.save(commit=False)
submission.ip_address = client_ip or None
submission.user_agent = request.META.get("HTTP_USER_AGENT", "")
if rate_limited:
submission.processing_status = ContactFormSubmission.STATUS_RATE_LIMITED
submission.save()
return submission
@require_POST
@ -100,88 +158,9 @@ def contact_form_submit(request):
messages.error(request, "表單欄位未填完整，請確認後再送出。")
return redirect(request.META.get("HTTP_REFERER") or "/")
submission = form.save(commit=False)
submission.ip_address = request.META.get("REMOTE_ADDR") or ""
submission.user_agent = request.META.get("HTTP_USER_AGENT", "")
submission.save()
notification_settings = SystemNotificationMailSettings.load(request_or_site=request)
smtp_settings = MailSmtpSettings.load(request_or_site=request)
subject_prefix = (notification_settings.contact_form_subject_prefix or "").strip()
subject = f"{subject_prefix} {submission.get_category_display()}".strip()
escaped_message_html = escape(submission.message).replace("\n", "<br>")
text_body = (
f"Name: {submission.name}\n"
f"Email: {submission.email}\n"
f"Contact: {submission.contact}\n"
f"Category: {submission.get_category_display()}\n"
f"Source Page: {submission.source_page}\n\n"
f"Message:\n{submission.message}\n"
)
html_body = (
"<p><strong>Name:</strong> "
f"{escape(submission.name)}</p>"
"<p><strong>Email:</strong> "
f"{escape(submission.email)}</p>"
"<p><strong>Contact:</strong> "
f"{escape(submission.contact)}</p>"
"<p><strong>Category:</strong> "
f"{escape(submission.get_category_display())}</p>"
"<p><strong>Source Page:</strong> "
f"{escape(submission.source_page or '')}</p>"
"<p><strong>Message:</strong></p>"
f"<p>{escaped_message_html}</p>"
)
try:
send_contact_notification_email(
subject=subject,
text_body=text_body,
html_body=html_body,
notification_config=notification_settings,
smtp_config=smtp_settings,
)
except Exception as exc:
logger.warning("contact form admin notification email failed: %s", exc)
# if submission.email:
# values_text = {
# "name": submission.name,
# "email": submission.email,
# "contact": submission.contact,
# "category": submission.get_category_display(),
# "message": submission.message,
# "source_page": submission.source_page or "",
# }
# values_html = {
# "name": escape(submission.name),
# "email": escape(submission.email),
# "contact": escape(submission.contact),
# "category": escape(submission.get_category_display()),
# "message": escape(submission.message).replace("\n", "<br>"),
# "source_page": escape(submission.source_page or ""),
# }
# user_subject = _render_contact_template(
# notification_settings.contact_form_user_subject_template,
# values_text,
# )
# user_text = _render_contact_template(
# notification_settings.contact_form_user_text_template,
# values_text,
# )
# user_html = _render_contact_template(
# notification_settings.contact_form_user_html_template,
# values_html,
# )
# try:
# send_contact_user_email(
# to_email=submission.email,
# subject=user_subject,
# text_body=user_text,
# html_body=user_html,
# notification_config=notification_settings,
# smtp_config=smtp_settings,
# )
# except Exception as exc:
# logger.warning("contact form user copy email failed: %s", exc)
submission = _save_contact_submission(form=form, request=request)
if submission.processing_status == ContactFormSubmission.STATUS_RATE_LIMITED:
logger.info("contact form submission rate limited submission_id=%s ip=%s", submission.pk, submission.ip_address)
if request.headers.get("x-requested-with") == "XMLHttpRequest":
return JsonResponse({"success": True})

86
innovedus_cms/base/wagtail_hooks.py
View File

@ -6,7 +6,9 @@ from django.utils.translation import gettext as _
from django.shortcuts import redirect
from django.shortcuts import render
from wagtail import hooks
from wagtail.admin.menu import MenuItem
from wagtail.admin.panels import FieldPanel
from wagtail.admin.ui.tables import Column
from wagtail.admin.widgets import Button
from wagtail.permission_policies import ModelPermissionPolicy
from wagtail.snippets.models import register_snippet
@ -18,11 +20,31 @@ from .forms import (
NewsletterHtmlEditorWidget,
NewsletterTemplateAdminForm,
)
from .models import NewsletterCampaign, NewsletterDispatchRecord, NewsletterSystemSettings, NewsletterTemplate
from .models import (
ContactFormSubmission,
NewsletterCampaign,
NewsletterDispatchRecord,
NewsletterSystemSettings,
NewsletterTemplate,
)
SEND_NEWSLETTER_PERMISSION = "base.send_newslettercampaign"
def _contact_status_value(submission):
return submission.get_processing_status_display()
def _contact_spam_category_value(submission):
if submission.spam_category is None:
return "尚未檢查"
return submission.get_spam_category_display()
def _contact_category_value(submission):
return submission.get_category_display()
def _newsletter_campaign_add_url() -> str:
candidates = [
"wagtailsnippets_base_newslettercampaign:add",
@ -188,6 +210,51 @@ class ReadOnlySnippetPermissionPolicy(ModelPermissionPolicy):
return super().user_has_permission(user, action)
class ContactFormSubmissionViewSet(SnippetViewSet):
model = ContactFormSubmission
icon = "form"
menu_label = "聯絡表單留言"
menu_order = 240
add_to_admin_menu = False
inspect_view_enabled = True
copy_view_enabled = False
permission_policy = ReadOnlySnippetPermissionPolicy(ContactFormSubmission)
list_display = [
"created_at",
Column("processing_status", label="處理狀態", accessor=_contact_status_value, sort_key="processing_status"),
Column("spam_category", label="垃圾訊息判定", accessor=_contact_spam_category_value, sort_key="spam_category"),
Column("category", label="留言類型", accessor=_contact_category_value, sort_key="category"),
"name",
"email",
"ip_address",
]
list_filter = ["processing_status", "spam_category", "category", "created_at"]
search_fields = ["name", "email", "contact", "message", "source_page", "ip_address"]
inspect_view_fields = [
"name",
"email",
"contact",
"category",
"message",
"source_page",
"ip_address",
"user_agent",
"processing_status",
"spam_category",
"spam_source",
"spam_checked_at",
"scan_attempt_count",
"send_attempt_count",
"next_attempt_at",
"processing_started_at",
"notification_sent_at",
"user_copy_sent_at",
"last_error",
"created_at",
"updated_at",
]
class NewsletterDispatchRecordViewSet(SnippetViewSet):
model = NewsletterDispatchRecord
icon = "tasks"
@ -253,6 +320,23 @@ class NewsletterAdminGroup(SnippetViewSetGroup):
register_snippet(NewsletterAdminGroup)
register_snippet(ContactFormSubmissionViewSet)
class ContactFormSubmissionReportMenuItem(MenuItem):
def is_shown(self, request):
return request.user.has_perm("base.view_contactformsubmission")
@hooks.register("register_reports_menu_item")
def register_contact_form_submission_report_menu_item():
return ContactFormSubmissionReportMenuItem(
"聯絡表單留言",
reverse("wagtailsnippets_base_contactformsubmission:list"),
name="contact-form-submissions",
icon_name="form",
order=600,
)
@hooks.register("register_snippet_listing_buttons")

37
innovedus_cms/entrypoint.sh Normal file → Executable file
View File

@ -7,11 +7,38 @@ APP_ROLE="${APP_ROLE:-web}"
python manage.py migrate --noinput
if [ "$APP_ROLE" = "scheduler" ]; then
SCHEDULER_INTERVAL_SECONDS="${SCHEDULER_INTERVAL_SECONDS:-60}"
while true; do
python manage.py run_newsletter_scheduler --limit 20
sleep "$SCHEDULER_INTERVAL_SECONDS"
done
NEWSLETTER_SCHEDULER_INTERVAL_SECONDS="${NEWSLETTER_SCHEDULER_INTERVAL_SECONDS:-${SCHEDULER_INTERVAL_SECONDS:-60}}"
CONTACT_SCHEDULER_INTERVAL_SECONDS="${CONTACT_SCHEDULER_INTERVAL_SECONDS:-60}"
run_newsletter_loop() {
while true; do
if ! python manage.py run_newsletter_scheduler --limit 20; then
echo "newsletter scheduler command failed" >&2
fi
sleep "$NEWSLETTER_SCHEDULER_INTERVAL_SECONDS"
done
}
run_contact_loop() {
while true; do
if ! python manage.py run_contact_submission_worker; then
echo "contact submission worker command failed" >&2
fi
sleep "$CONTACT_SCHEDULER_INTERVAL_SECONDS"
done
}
run_newsletter_loop &
newsletter_pid=$!
run_contact_loop &
contact_pid=$!
stop_scheduler_loops() {
kill "$newsletter_pid" "$contact_pid" 2>/dev/null || true
wait "$newsletter_pid" "$contact_pid" 2>/dev/null || true
}
trap stop_scheduler_loops EXIT TERM INT
wait
else
python manage.py collectstatic --noinput
exec "$@"

BIN
innovedus_cms/locale/zh_Hant/LC_MESSAGES/django.mo
View File

Binary file not shown.

93
innovedus_cms/locale/zh_Hant/LC_MESSAGES/django.po
View File

@ -270,7 +270,7 @@ msgid "Sending"
msgstr "寄送中"
msgid "Sent"
msgstr "已送出"
msgstr "已寄出"
msgid "Failed"
msgstr "失敗"
@ -333,10 +333,97 @@ msgid "Created At"
msgstr "建立時間"
msgid "Contact Form Submission"
msgstr "聯絡表單提交"
msgstr "聯絡表單留言"
msgid "Contact Form Submissions"
msgstr "聯絡表單提交"
msgstr "聯絡表單留言"
msgid "Contact form submissions"
msgstr "聯絡表單留言"
msgid "Contact Form Settings"
msgstr "聯絡表單設定"
msgid "Contact Form Rate Limit"
msgstr "聯絡表單頻率限制"
msgid "Enable IP Rate Limit"
msgstr "啟用 IP 留言頻率限制"
msgid "Maximum Submissions Per Window"
msgstr "限制區間內最多留言數"
msgid "Rate Limit Window Minutes"
msgstr "限制區間（分鐘）"
msgid "Send User Copy"
msgstr "寄送留言副本給填表人"
msgid "Send a copy to the submitter only after spam detection allows the message."
msgstr "僅在垃圾訊息檢測通過後，寄送留言副本給填表人。"
msgid "Processing Status"
msgstr "處理狀態"
msgid "Queued"
msgstr "等待垃圾訊息檢查"
msgid "Scanning"
msgstr "檢查垃圾訊息中"
msgid "Scan Retry"
msgstr "等待重新檢查"
msgid "Blocked as Spam"
msgstr "已阻擋垃圾訊息"
msgid "Send Retry"
msgstr "等待重新寄送"
msgid "Rate Limited"
msgstr "超過留言頻率限制"
msgid "Legacy Submission"
msgstr "舊資料（未重新檢查）"
msgid "Spam Category"
msgstr "垃圾訊息分類"
msgid "Allowed"
msgstr "正常留言（允許寄送）"
msgid "Unsolicited Advertising"
msgstr "廣告／推銷訊息"
msgid "Malicious"
msgstr "惡意／釣魚訊息"
msgid "Meaningless"
msgstr "無意義／測試內容"
msgid "Spam Detection Source"
msgstr "垃圾訊息判定來源"
msgid "Spam Checked At"
msgstr "垃圾訊息檢測時間"
msgid "Scan Attempt Count"
msgstr "檢測嘗試次數"
msgid "Send Attempt Count"
msgstr "寄送嘗試次數"
msgid "Next Attempt At"
msgstr "下次嘗試時間"
msgid "Processing Started At"
msgstr "開始處理時間"
msgid "Notification Sent At"
msgstr "內部通知寄送時間"
msgid "User Copy Sent At"
msgstr "填表人副本寄送時間"
msgid "Newsletter campaigns"
msgstr "電子報"

22
innovedus_cms/mysite/settings/base.py
View File

@ -55,6 +55,16 @@ def env_optional(name, default=None):
return normalized
def env_int(name, default):
value = os.environ.get(name)
if value is None:
return default
try:
return int(value.strip())
except (TypeError, ValueError):
return default
def build_media_storage_options():
options = {
"access_key": os.environ.get("AWS_ACCESS_KEY_ID"),
@ -111,6 +121,18 @@ def build_allowed_hosts():
GA4_MEASUREMENT_ID = os.environ.get("GA4_MEASUREMENT_ID", "").strip()
SECRET_KEY = os.environ.get("SECRET_KEY", "").strip()
CONTACT_SPAM_DETECTION_ENABLED = env_bool("CONTACT_SPAM_DETECTION_ENABLED", default=False)
CONTACT_SPAM_DETECTION_URL = os.environ.get("CONTACT_SPAM_DETECTION_URL", "").strip()
CONTACT_SPAM_DETECTION_API_KEY = os.environ.get("CONTACT_SPAM_DETECTION_API_KEY", "").strip()
CONTACT_SPAM_DETECTION_TIMEOUT_SECONDS = env_int("CONTACT_SPAM_DETECTION_TIMEOUT_SECONDS", 180)
CONTACT_SPAM_DETECTION_RETRY_DELAY_SECONDS = env_int("CONTACT_SPAM_DETECTION_RETRY_DELAY_SECONDS", 300)
CONTACT_SPAM_DETECTION_MAX_ATTEMPTS = env_int("CONTACT_SPAM_DETECTION_MAX_ATTEMPTS", 3)
CONTACT_FORM_WORKER_BATCH_SIZE = env_int("CONTACT_FORM_WORKER_BATCH_SIZE", 20)
CONTACT_FORM_WORKER_MAX_RUNTIME_SECONDS = env_int("CONTACT_FORM_WORKER_MAX_RUNTIME_SECONDS", 600)
CONTACT_FORM_WORKER_LEASE_SECONDS = env_int("CONTACT_FORM_WORKER_LEASE_SECONDS", 600)
CONTACT_FORM_MAIL_RETRY_DELAY_SECONDS = env_int("CONTACT_FORM_MAIL_RETRY_DELAY_SECONDS", 300)
CONTACT_FORM_MAIL_MAX_ATTEMPTS = env_int("CONTACT_FORM_MAIL_MAX_ATTEMPTS", 3)
# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/5.2/howto/deployment/checklist/