This commit is contained in:
nvrl committed 2026-04-11 10:37:49 +02:00
1 parent 5e3474dd59
commit e4d59a13ba
119 files changed
+15490 -9975

No files matched your search

+6 -7
View File
@@ -1,14 +1,13 @@
from django.contrib import admin
from django.contrib.auth.admin import UserAdmin
from .models import User, Category, Flavor, Rating, Banner
from .models import Banner, Category, Flavor, Rating, User
class CustomUserAdmin(UserAdmin):
fieldsets = UserAdmin.fieldsets + (
(None, {'fields': ('theme',)}),
)
add_fieldsets = UserAdmin.add_fieldsets + (
(None, {'fields': ('theme',)}),
)
fieldsets = UserAdmin.fieldsets + ((None, {"fields": ("theme",)}),)
add_fieldsets = UserAdmin.add_fieldsets + ((None, {"fields": ("theme",)}),)
admin.site.register(User, CustomUserAdmin)
admin.site.register(Category)
+1 -1
View File
@@ -2,4 +2,4 @@ from django.apps import AppConfig
class ApiConfig(AppConfig):
name = 'api'
name = "api"
+5 -2
View File
@@ -1,6 +1,8 @@
import ssl
from django.core.mail.backends.smtp import EmailBackend
from django.conf import settings
from django.core.mail.backends.smtp import EmailBackend
class InsecureEmailBackend(EmailBackend):
"""
@@ -8,9 +10,10 @@ class InsecureEmailBackend(EmailBackend):
Useful for local development or internal SMTP servers with self-signed certificates.
Set EMAIL_SKIP_CERT_VERIFICATION = True in settings to use.
"""
@property
def ssl_context(self):
if getattr(settings, 'EMAIL_SKIP_CERT_VERIFICATION', False):
if getattr(settings, "EMAIL_SKIP_CERT_VERIFICATION", False):
# The most robust way to bypass verification for self-signed certs
context = ssl._create_unverified_context()
return context
+36 -26
View File
@@ -1,38 +1,43 @@
import sqlite3
import os
import shutil
import sqlite3
from datetime import datetime, timedelta
from django.conf import settings
from django.core.management.base import BaseCommand
class Command(BaseCommand):
help = 'Creates a safe, consistent backup of the SQLite database and manages rotation.'
help = "Creates a safe, consistent backup of the SQLite database and manages rotation."
def add_arguments(self, parser):
parser.add_argument('--output', type=str, help='Specific output path for the backup file')
parser.add_argument('--retention', type=int, default=7, help='Number of days to keep old backups')
parser.add_argument('--full', action='store_true', help='Create a full archive including media')
parser.add_argument("--output", type=str, help="Specific output path for the backup file")
parser.add_argument(
"--retention", type=int, default=7, help="Number of days to keep old backups"
)
parser.add_argument(
"--full", action="store_true", help="Create a full archive including media"
)
def handle(self, *args, **options):
db_conf = settings.DATABASES['default']
if db_conf['ENGINE'] != 'django.db.backends.sqlite3':
db_conf = settings.DATABASES["default"]
if db_conf["ENGINE"] != "django.db.backends.sqlite3":
self.stdout.write(self.style.ERROR("This command only supports SQLite."))
return
db_path = str(db_conf['NAME'])
backup_dir = os.path.join(settings.BASE_DIR, 'backups')
db_path = str(db_conf["NAME"])
backup_dir = os.path.join(settings.BASE_DIR, "backups")
os.makedirs(backup_dir, exist_ok=True)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
# 1. Database Backup
if options['output']:
db_backup_path = options['output']
if options["output"]:
db_backup_path = options["output"]
else:
db_backup_path = os.path.join(backup_dir, f'db_backup_{timestamp}.sqlite3')
db_backup_path = os.path.join(backup_dir, f"db_backup_{timestamp}.sqlite3")
self.stdout.write(f"Cloning database {db_path}...")
try:
src = sqlite3.connect(db_path)
dst = sqlite3.connect(db_backup_path)
@@ -46,12 +51,13 @@ class Command(BaseCommand):
return
# 2. Full Archive (Optional)
if options['full']:
if options["full"]:
import tarfile
archive_name = f"full_backup_{timestamp}.tar.gz"
archive_path = os.path.join(backup_dir, archive_name)
self.stdout.write(f"Creating full archive at {archive_path}...")
try:
with tarfile.open(archive_path, "w:gz") as tar:
# Add database
@@ -60,19 +66,19 @@ class Command(BaseCommand):
media_dir = str(settings.MEDIA_ROOT)
if os.path.exists(media_dir):
tar.add(media_dir, arcname="media")
# Remove the raw sqlite backup after bundling it into the archive
if not options['output']:
if not options["output"]:
os.remove(db_backup_path)
self.stdout.write(self.style.SUCCESS(f"Full archive created successfully."))
self.stdout.write(self.style.SUCCESS("Full archive created successfully."))
except Exception as e:
self.stdout.write(self.style.ERROR(f"Archive creation failed: {e}"))
# 3. Cleanup Old Backups
retention_days = options['retention']
retention_days = options["retention"]
cutoff = datetime.now() - timedelta(days=retention_days)
self.stdout.write(f"Cleaning up backups older than {retention_days} days...")
for filename in os.listdir(backup_dir):
file_path = os.path.join(backup_dir, filename)
@@ -81,6 +87,10 @@ class Command(BaseCommand):
if file_time < cutoff:
try:
os.remove(file_path)
self.stdout.write(self.style.SUCCESS(f" -> Deleted old backup: {filename}"))
self.stdout.write(
self.style.SUCCESS(f" -> Deleted old backup: {filename}")
)
except Exception as e:
self.stdout.write(self.style.WARNING(f" -> Failed to delete {filename}: {e}"))
self.stdout.write(
self.style.WARNING(f" -> Failed to delete {filename}: {e}")
)
@@ -1,34 +1,39 @@
from django.core.management.base import BaseCommand
from api.models import Flavor, Rating
from django.db.models import Count
from api.models import Flavor, Rating
class Command(BaseCommand):
help = 'Identifies and merges duplicate flavors (same name and category)'
help = "Identifies and merges duplicate flavors (same name and category)"
def handle(self, *args, **options):
# Find duplicates based on name and category
duplicates = Flavor.objects.values('name', 'category').annotate(
name_count=Count('id')
).filter(name_count__gt=1)
duplicates = (
Flavor.objects.values("name", "category")
.annotate(name_count=Count("id"))
.filter(name_count__gt=1)
)
self.stdout.write(f"Found {duplicates.count()} sets of duplicates.")
for dup in duplicates:
name = dup['name']
category_id = dup['category']
name = dup["name"]
category_id = dup["category"]
# Get all flavors in this set, ordered by having an external_id first, then by date
flavor_set = Flavor.objects.filter(
name=name,
category_id=category_id
).order_by('-external_id', 'created_at')
flavor_set = Flavor.objects.filter(name=name, category_id=category_id).order_by(
"-external_id", "created_at"
)
# Keep the first one, merge others into it
keep = flavor_set[0]
others = flavor_set[1:]
self.stdout.write(f"Merging {len(others)} duplicates into '{keep.name}' (ID: {keep.id}, Ext: {keep.external_id})")
self.stdout.write(
f"Merging {len(others)} duplicates into '{keep.name}' (ID: {keep.id}, Ext: {keep.external_id})"
)
for other in others:
# Re-assign ratings
for rating in Rating.objects.filter(flavor=other):
@@ -39,7 +44,7 @@ class Command(BaseCommand):
# User already has a rating for the 'keep' flavor
# Delete the duplicate rating instead
rating.delete()
# Delete the duplicate flavor
other.delete()
+33 -22
View File
@@ -1,64 +1,75 @@
from django.core.management.base import BaseCommand
from api.models import Banner
import os
import json
import os
from django.core.management.base import BaseCommand
from api.models import Banner
class Command(BaseCommand):
help = 'Seeds banner models and missing settings from JSON files in backend/banners/'
help = "Seeds banner models and missing settings from JSON files in backend/banners/"
def handle(self, *args, **options):
# Determine the banners directory relative to the project root
project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
banners_dir = os.path.join(project_root, 'banners')
banners_dir = os.path.join(project_root, "banners")
if not os.path.exists(banners_dir):
self.stdout.write(self.style.WARNING(f"Banners directory not found at {banners_dir}"))
return
for filename in os.listdir(banners_dir):
if filename.endswith('.json'):
if filename.endswith(".json"):
filepath = os.path.join(banners_dir, filename)
try:
with open(filepath, 'r') as f:
with open(filepath) as f:
data = json.load(f)
slug = data.get('slug')
slug = data.get("slug")
if not slug:
continue
banner, created = Banner.objects.get_or_create(
slug=slug,
defaults={
'name': data.get('name', slug),
'description': data.get('description', ''),
'settings': data.get('settings', {})
}
"name": data.get("name", slug),
"description": data.get("description", ""),
"settings": data.get("settings", {}),
},
)
if not created:
# Sync missing settings from JSON to DB without overwriting existing keys
db_settings = banner.settings or {}
json_settings = data.get('settings', {})
json_schema = data.get('schema', [])
json_settings = data.get("settings", {})
json_schema = data.get("schema", [])
updated = False
# Sync missing settings
for key, value in json_settings.items():
if key not in db_settings:
db_settings[key] = value
updated = True
# Always sync schema as it represents the UI structure
if banner.schema != json_schema:
banner.schema = json_schema
updated = True
if updated:
banner.settings = db_settings
banner.save()
self.stdout.write(self.style.SUCCESS(f"Synced new settings for banner '{slug}' from {filename}"))
self.stdout.write(
self.style.SUCCESS(
f"Synced new settings for banner '{slug}' from {filename}"
)
)
else:
self.stdout.write(self.style.SUCCESS(f"Created new banner '{slug}' from {filename}"))
self.stdout.write(
self.style.SUCCESS(f"Created new banner '{slug}' from {filename}")
)
except Exception as e:
self.stdout.write(self.style.ERROR(f"Failed to sync banner from {filename}: {e}"))
self.stdout.write(
self.style.ERROR(f"Failed to sync banner from {filename}: {e}")
)
@@ -1,41 +1,45 @@
import json
import os
import requests
from django.core.management.base import BaseCommand
from django.core.files import File
from django.conf import settings
from django.core.management.base import BaseCommand
from django.utils.text import slugify
from api.models import Category, Flavor
class Command(BaseCommand):
help = 'Adds legacy flavors from JSON files in the legacy/ folder'
help = "Adds legacy flavors from JSON files in the legacy/ folder"
def download_image(self, url, flavor_name):
if not url:
return None
try:
safe_name = slugify(flavor_name)
ext = url.split('.')[-1].split('?')[0].lower()
if ext not in ['jpg', 'jpeg', 'png', 'webp', 'gif']:
ext = 'png'
ext = url.split(".")[-1].split("?")[0].lower()
if ext not in ["jpg", "jpeg", "png", "webp", "gif"]:
ext = "png"
filename = f"legacy_{safe_name}.{ext}"
filepath = os.path.join(settings.MEDIA_ROOT, 'flavors', filename)
filepath = os.path.join(settings.MEDIA_ROOT, "flavors", filename)
os.makedirs(os.path.dirname(filepath), exist_ok=True)
response = requests.get(url, headers={'User-Agent': 'Mozilla/5.0'}, timeout=10)
response = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=10)
response.raise_for_status()
if os.path.exists(filepath):
os.remove(filepath)
with open(filepath, 'wb') as f:
with open(filepath, "wb") as f:
f.write(response.content)
return f"flavors/{filename}"
except Exception as e:
self.stdout.write(self.style.WARNING(f"Failed to download image for {flavor_name}: {e}"))
self.stdout.write(
self.style.WARNING(f"Failed to download image for {flavor_name}: {e}")
)
return None
def handle(self, *args, **options):
@@ -43,88 +47,103 @@ class Command(BaseCommand):
# Instead, we update existing entries.
cat_map = {
'energy': Category.objects.get_or_create(name='Energy', defaults={'slug': 'energy'})[0],
'hydration': Category.objects.get_or_create(name='Hydration', defaults={'slug': 'hydration'})[0],
'iced-tea': Category.objects.get_or_create(name='Iced Tea', defaults={'slug': 'iced-tea'})[0],
'milkshake': Category.objects.get_or_create(name='Milkshake', defaults={'slug': 'milkshake'})[0],
'packs': Category.objects.get_or_create(name='Packs and other', defaults={'slug': 'packs-and-other'})[0],
"energy": Category.objects.get_or_create(name="Energy", defaults={"slug": "energy"})[0],
"hydration": Category.objects.get_or_create(
name="Hydration", defaults={"slug": "hydration"}
)[0],
"iced-tea": Category.objects.get_or_create(
name="Iced Tea", defaults={"slug": "iced-tea"}
)[0],
"milkshake": Category.objects.get_or_create(
name="Milkshake", defaults={"slug": "milkshake"}
)[0],
"packs": Category.objects.get_or_create(
name="Packs and other", defaults={"slug": "packs-and-other"}
)[0],
}
legacy_dir = os.path.join(settings.BASE_DIR, 'legacy')
legacy_dir = os.path.join(settings.BASE_DIR, "legacy")
if not os.path.exists(legacy_dir):
legacy_dir = os.path.join(settings.BASE_DIR, '..', 'legacy')
legacy_dir = os.path.join(settings.BASE_DIR, "..", "legacy")
if not os.path.exists(legacy_dir):
self.stdout.write(self.style.ERROR(f"Legacy directory not found."))
self.stdout.write(self.style.ERROR("Legacy directory not found."))
return
json_files = [f for f in os.listdir(legacy_dir) if f.endswith('.json')]
json_files = [f for f in os.listdir(legacy_dir) if f.endswith(".json")]
processed_count = 0
for json_file in json_files:
file_path = os.path.join(legacy_dir, json_file)
file_lower = json_file.lower()
# Determine category from filename
if 'energy' in file_lower:
file_cat = cat_map['energy']
elif 'iced_tea' in file_lower or 'iced-tea' in file_lower:
file_cat = cat_map['iced-tea']
elif 'hydration' in file_lower:
file_cat = cat_map['hydration']
elif 'milkshake' in file_lower:
file_cat = cat_map['milkshake']
if "energy" in file_lower:
file_cat = cat_map["energy"]
elif "iced_tea" in file_lower or "iced-tea" in file_lower:
file_cat = cat_map["iced-tea"]
elif "hydration" in file_lower:
file_cat = cat_map["hydration"]
elif "milkshake" in file_lower:
file_cat = cat_map["milkshake"]
else:
file_cat = None
with open(file_path, 'r', encoding='utf-8') as f:
with open(file_path, encoding="utf-8") as f:
data_list = json.load(f)
for data in data_list:
name = data.get('Name')
desc = data.get('Beschreibung')
taste = data.get('Geschmack')
status_text = data.get('Status')
image_url = data.get('Bild_URL')
name = data.get("Name")
desc = data.get("Beschreibung")
taste = data.get("Geschmack")
status_text = data.get("Status")
image_url = data.get("Bild_URL")
full_desc = f"**Geschmack:** {taste}\n\n{desc}\n\n**Status:** {status_text}"
# Determine final category
cat = file_cat
if not cat:
# Fallback to name/desc detection if filename was ambiguous
if "hydration" in name.lower() or "hydration" in desc.lower():
cat = cat_map['hydration']
cat = cat_map["hydration"]
elif "iced tea" in name.lower() or "eistee" in name.lower():
cat = cat_map['iced-tea']
cat = cat_map["iced-tea"]
elif "milkshake" in name.lower():
cat = cat_map['milkshake']
elif any(k in name.lower() for k in ['bundle', 'set', 'box', 'probe', 'sample', 'taster', 'shaker', 'starter']):
cat = cat_map['packs']
cat = cat_map["milkshake"]
elif any(
k in name.lower()
for k in [
"bundle",
"set",
"box",
"probe",
"sample",
"taster",
"shaker",
"starter",
]
):
cat = cat_map["packs"]
else:
cat = cat_map['energy'] # Ultimate fallback
cat = cat_map["energy"] # Ultimate fallback
# Try to find existing flavor by name and category regardless of legacy status
flavor = Flavor.objects.filter(name__iexact=name, category=cat).first()
# If it exists and is an active Shopify product, skip legacy processing
if flavor and not flavor.is_legacy:
self.stdout.write(f"Skipping active flavor: {name}")
continue
if not flavor:
flavor = Flavor(
name=name,
category=cat,
is_legacy=True,
external_id=None
)
flavor = Flavor(name=name, category=cat, is_legacy=True, external_id=None)
self.stdout.write(f"Adding new legacy flavor: {name}")
else:
self.stdout.write(f"Updating legacy flavor: {name}")
flavor.description = full_desc
flavor.is_available = False # Legacy is always unavailable via API
flavor.is_available = False # Legacy is always unavailable via API
if image_url:
# Check if local file exists
@@ -140,9 +159,11 @@ class Command(BaseCommand):
if rel_path:
flavor.image = rel_path
self.stdout.write(f" -> Updated image for: {name}")
flavor.image_url = image_url
flavor.save()
processed_count += 1
self.stdout.write(self.style.SUCCESS(f"Finished! Processed {processed_count} legacy flavors."))
self.stdout.write(
self.style.SUCCESS(f"Finished! Processed {processed_count} legacy flavors.")
)
+103 -67
View File
@@ -1,21 +1,22 @@
import json
import requests
import re
import os
from django.core.management.base import BaseCommand
from django.core.files import File
import re
import requests
from django.conf import settings
from django.core.management.base import BaseCommand
from django.utils.text import slugify
from api.models import Category, Flavor
class Command(BaseCommand):
help = 'Syncs flavors from Holy Energy Shopify API'
help = "Syncs flavors from Holy Energy Shopify API"
def clean_html(self, raw_html):
if not raw_html:
return ""
cleanr = re.compile('<.*?>')
cleantext = re.sub(cleanr, '', raw_html)
cleanr = re.compile("<.*?>")
cleantext = re.sub(cleanr, "", raw_html)
return cleantext.strip()
def download_image(self, url, flavor_name):
@@ -23,50 +24,60 @@ class Command(BaseCommand):
return None
try:
safe_name = slugify(flavor_name)
ext = url.split('.')[-1].split('?')[0].lower()
if ext not in ['jpg', 'jpeg', 'png', 'webp', 'gif']:
ext = 'png'
ext = url.split(".")[-1].split("?")[0].lower()
if ext not in ["jpg", "jpeg", "png", "webp", "gif"]:
ext = "png"
filename = f"{safe_name}.{ext}"
filepath = os.path.join(settings.MEDIA_ROOT, 'flavors', filename)
filepath = os.path.join(settings.MEDIA_ROOT, "flavors", filename)
os.makedirs(os.path.dirname(filepath), exist_ok=True)
response = requests.get(url, headers={'User-Agent': 'Mozilla/5.0'}, timeout=10)
response = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=10)
response.raise_for_status()
if os.path.exists(filepath):
os.remove(filepath)
with open(filepath, 'wb') as f:
with open(filepath, "wb") as f:
f.write(response.content)
return f"flavors/{filename}"
except Exception as e:
self.stdout.write(self.style.WARNING(f"Failed to download image for {flavor_name}: {e}"))
self.stdout.write(
self.style.WARNING(f"Failed to download image for {flavor_name}: {e}")
)
return None
def handle(self, *args, **options):
url = "https://weareholy.com/products.json?limit=250"
self.stdout.write(f"Fetching from {url}...")
try:
response = requests.get(url, headers={'User-Agent': 'Mozilla/5.0'}, timeout=15)
response = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=15)
response.raise_for_status()
data = response.json()
except Exception as e:
self.stdout.write(self.style.ERROR(f"Failed to fetch data: {e}"))
return
products = data.get('products', [])
products = data.get("products", [])
self.stdout.write(f"Found {len(products)} products.")
cat_map = {
'energy': Category.objects.get_or_create(name='Energy', defaults={'slug': 'energy'})[0],
'hydration': Category.objects.get_or_create(name='Hydration', defaults={'slug': 'hydration'})[0],
'iced-tea': Category.objects.get_or_create(name='Iced Tea', defaults={'slug': 'iced-tea'})[0],
'milkshake': Category.objects.get_or_create(name='Milkshake', defaults={'slug': 'milkshake'})[0],
'packs': Category.objects.get_or_create(name='Packs and other', defaults={'slug': 'packs-and-other'})[0],
"energy": Category.objects.get_or_create(name="Energy", defaults={"slug": "energy"})[0],
"hydration": Category.objects.get_or_create(
name="Hydration", defaults={"slug": "hydration"}
)[0],
"iced-tea": Category.objects.get_or_create(
name="Iced Tea", defaults={"slug": "iced-tea"}
)[0],
"milkshake": Category.objects.get_or_create(
name="Milkshake", defaults={"slug": "milkshake"}
)[0],
"packs": Category.objects.get_or_create(
name="Packs and other", defaults={"slug": "packs-and-other"}
)[0],
}
created_count = 0
@@ -74,82 +85,97 @@ class Command(BaseCommand):
synced_external_ids = []
for p in products:
title = p.get('title', '').strip()
tags = p.get('tags', [])
title = p.get("title", "").strip()
tags = p.get("tags", [])
if isinstance(tags, str):
tags = [t.strip() for t in tags.split(',')]
tags = [t.strip() for t in tags.split(",")]
tags_lower = [t.lower() for t in tags]
title_lower = title.lower()
is_pack = False
pack_keywords = ['bundle', 'set', 'box', 'probe', 'sample', 'taster', 'starter', 'collection', 'probier']
pack_keywords = [
"bundle",
"set",
"box",
"probe",
"sample",
"taster",
"starter",
"collection",
"probier",
]
if any(k in title_lower for k in pack_keywords):
is_pack = True
if 'shaker' in title_lower or 'merch' in tags_lower:
if "shaker" in title_lower or "merch" in tags_lower:
is_pack = True
category = None
if is_pack:
category = cat_map['packs']
elif 'holy energy' in tags_lower or 'energy' in tags_lower:
category = cat_map['energy']
elif 'holy hydration' in tags_lower or 'hydration' in tags_lower:
category = cat_map['hydration']
elif 'holy iced tea' in tags_lower or 'iced tea' in tags_lower:
category = cat_map['iced-tea']
elif 'milkshake' in tags_lower:
category = cat_map['milkshake']
category = cat_map["packs"]
elif "holy energy" in tags_lower or "energy" in tags_lower:
category = cat_map["energy"]
elif "holy hydration" in tags_lower or "hydration" in tags_lower:
category = cat_map["hydration"]
elif "holy iced tea" in tags_lower or "iced tea" in tags_lower:
category = cat_map["iced-tea"]
elif "milkshake" in tags_lower:
category = cat_map["milkshake"]
if not category:
if 'shaker' in tags_lower:
category = cat_map['packs']
if "shaker" in tags_lower:
category = cat_map["packs"]
else:
continue
variants = p.get('variants', [])
is_available = any(v.get('available') for v in variants)
images = p.get('images', [])
image_url = images[0].get('src') if images else None
description = self.clean_html(p.get('body_html', ''))
handle = p.get('handle')
variants = p.get("variants", [])
is_available = any(v.get("available") for v in variants)
images = p.get("images", [])
image_url = images[0].get("src") if images else None
description = self.clean_html(p.get("body_html", ""))
handle = p.get("handle")
shop_url = f"https://weareholy.com/products/{handle}" if handle else None
# Find or create - be very specific to avoid duplicates
# 1. Try external_id
flavor = Flavor.objects.filter(external_id=p['id']).first()
flavor = Flavor.objects.filter(external_id=p["id"]).first()
if not flavor:
# 2. Try exact name match in category
# REMOVED: is_legacy=False
flavor = Flavor.objects.filter(name=title, category=category).first()
if not flavor:
# 3. Try case-insensitive name match in category (safety for whitespace/case)
# REMOVED: is_legacy=False
flavor = Flavor.objects.filter(name__iexact=title, category=category).first()
if not flavor:
flavor = Flavor(external_id=p['id'], name=title, category=category, is_legacy=False)
flavor = Flavor(external_id=p["id"], name=title, category=category, is_legacy=False)
created_count += 1
else:
updated_count += 1
# Ensure external_id is updated if it was missing
if not flavor.external_id:
flavor.external_id = p['id']
flavor.external_id = p["id"]
# Reactivate the flavor if it was previously marked as legacy
if flavor.is_legacy:
flavor.is_legacy = False
# Ensure all synced fields are accurately reflected
if flavor.name != title and not Flavor.objects.filter(name=title, category=category).exclude(pk=flavor.pk).exists():
if (
flavor.name != title
and not Flavor.objects.filter(name=title, category=category)
.exclude(pk=flavor.pk)
.exists()
):
flavor.name = title
flavor.description = description
flavor.shop_url = shop_url
flavor.is_available = is_available
if image_url:
# Check if local file exists
file_exists = False
@@ -165,11 +191,21 @@ class Command(BaseCommand):
flavor.image = rel_path
self.stdout.write(f"Downloaded image for: {title}")
else:
self.stdout.write(self.style.WARNING(f"Failed to ensure image for: {title}"))
self.stdout.write(
self.style.WARNING(f"Failed to ensure image for: {title}")
)
flavor.image_url = image_url
flavor.save()
synced_external_ids.append(p['id'])
synced_external_ids.append(p["id"])
discontinued_count = Flavor.objects.filter(external_id__isnull=False).exclude(external_id__in=synced_external_ids).update(is_available=False)
self.stdout.write(self.style.SUCCESS(f"Finished! Created: {created_count}, Updated: {updated_count}, Marked Unavailable: {discontinued_count}"))
discontinued_count = (
Flavor.objects.filter(external_id__isnull=False)
.exclude(external_id__in=synced_external_ids)
.update(is_available=False)
)
self.stdout.write(
self.style.SUCCESS(
f"Finished! Created: {created_count}, Updated: {updated_count}, Marked Unavailable: {discontinued_count}"
)
)
-81
View File
@@ -1,81 +0,0 @@
import time
import sys
from datetime import datetime, timedelta
from django.core.management.base import BaseCommand
from django.core.management import call_command
from django.utils import timezone
from django.db import models
from api.models import Job
from io import StringIO
class Command(BaseCommand):
help = 'Background worker to process scheduled and on-demand jobs.'
def handle(self, *args, **options):
self.stdout.write(self.style.SUCCESS("Starting Holy Archive Worker..."))
# Ensure all core jobs exist in the DB
for job_type, label in Job.JOB_TYPES:
Job.objects.get_or_create(name=job_type)
while True:
# 1. Find jobs that are either explicitly set to 'pending' (manual trigger)
# or those where next_run has passed.
now = timezone.now()
jobs_to_run = Job.objects.filter(
~models.Q(status='running')
).filter(
# Run if pending OR (next_run is set AND has passed)
models.Q(status='pending') |
models.Q(next_run__lte=now, interval_hours__gt=0)
)
for job in jobs_to_run:
self.run_job(job)
time.sleep(30) # Check every 30 seconds
def run_job(self, job):
self.stdout.write(f"Executing Job: {job.get_name_display()}...")
job.status = 'running'
job.last_run = timezone.now()
job.last_output = "" # Reset output for new run
job.save()
# Capture management command output
class LiveStream(StringIO):
def write(self, s):
super().write(s)
# Periodically update the DB with current output
job.last_output = self.getvalue()
job.save(update_fields=['last_output'])
out = LiveStream()
error = ""
try:
if job.name == 'sync_flavors':
call_command('sync_flavors', stdout=out)
elif job.name == 'cleanup_duplicates':
call_command('cleanup_duplicates', stdout=out)
elif job.name == 'backup_db':
call_command('backup_db', '--full', stdout=out)
elif job.name == 'seed_legacy':
call_command('seed_legacy_flavors', stdout=out)
job.status = 'completed'
except Exception as e:
job.status = 'failed'
error = str(e)
self.stdout.write(self.style.ERROR(f"Job failed: {error}"))
# Update scheduling for next run if interval is set
if job.interval_hours > 0:
job.next_run = timezone.now() + timedelta(hours=job.interval_hours)
else:
job.next_run = None
job.last_output = out.getvalue()
job.error_message = error
job.save()
self.stdout.write(self.style.SUCCESS(f"Job finished: {job.status}"))
@@ -0,0 +1,18 @@
# Generated by Django 6.0.4 on 2026-04-11 00:37
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('api', '0029_alter_job_name'),
]
operations = [
migrations.AddField(
model_name='flavor',
name='image_urls',
field=models.JSONField(blank=True, default=list),
),
]
-266
View File
@@ -1,266 +0,0 @@
from django.db import models
from django.contrib.auth.models import AbstractUser
from django.core.validators import MinValueValidator, MaxValueValidator
class User(AbstractUser):
THEME_CHOICES = [
('holy_light', 'Holy Light'),
('holy_dark', 'Holy Dark'),
('latte', 'Catppuccin Latte'),
('frappe', 'Catppuccin Frappé'),
('macchiato', 'Catppuccin Macchiato'),
('mocha', 'Catppuccin Mocha'),
('pink_pastel', 'Pink Pastel'),
('mint_pastel', 'Mint Pastel'),
('lavender_pastel', 'Lavender Pastel'),
('dracula', 'Dracula'),
('nord', 'Nordic Frost'),
('gruvbox', 'Gruvbox Retro'),
('oceanic', 'Oceanic Deep'),
('t0p_sai', 'SAI (T0P)'),
('t0p_trench', 'Trench (T0P)'),
('t0p_blurryface', 'Blurryface (T0P)'),
('t0p_clancy', 'Clancy (T0P)'),
]
LANGUAGE_CHOICES = [
('en', 'English'),
('de', 'Deutsch'),
]
theme = models.CharField(max_length=20, choices=THEME_CHOICES, default='holy_light')
language = models.CharField(max_length=2, choices=LANGUAGE_CHOICES, default='en')
avatar = models.ImageField(upload_to='avatars/', blank=True, null=True)
pending_email = models.EmailField(max_length=254, blank=True, null=True)
email_confirmation_code = models.CharField(max_length=6, blank=True, null=True)
following = models.ManyToManyField('self', symmetrical=False, related_name='followers', blank=True)
selected_banner = models.ForeignKey('Banner', on_delete=models.SET_NULL, null=True, blank=True, related_name='users')
def save(self, *args, **kwargs):
super().save(*args, **kwargs)
if self.avatar:
try:
from PIL import Image
img = Image.open(self.avatar.path)
if img.height > 256 or img.width > 256 or img.format != 'JPEG':
output_size = (256, 256)
img.thumbnail(output_size)
if img.mode in ("RGBA", "P"):
img = img.convert("RGB")
img.save(self.avatar.path, 'JPEG', quality=90)
except Exception:
pass
def __str__(self):
return self.username
class Category(models.Model):
name = models.CharField(max_length=50, unique=True)
slug = models.SlugField(unique=True)
class Meta:
verbose_name_plural = "Categories"
ordering = ['name']
def __str__(self):
return self.name
class Flavor(models.Model):
category = models.ForeignKey(Category, related_name='flavors', on_delete=models.CASCADE)
name = models.CharField(max_length=100) # Removed unique=True here
description = models.TextField(blank=True)
image_url = models.URLField(max_length=1000, blank=True, null=True)
image = models.ImageField(upload_to='flavors/', blank=True, null=True)
shop_url = models.URLField(max_length=500, blank=True, null=True)
is_available = models.BooleanField(default=True)
is_legacy = models.BooleanField(default=False)
external_id = models.BigIntegerField(unique=True, null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
# This allows "Erdbeere" in Iced Tea and "Erdbeere" in Milkshake
constraints = [
models.UniqueConstraint(fields=['name', 'category'], name='unique_flavor_name_per_category')
]
ordering = ['name']
def __str__(self):
return f"{self.name} ({self.category.name})"
@property
def get_average_rating(self):
ratings = self.ratings.all()
if ratings.exists():
return sum(r.score for r in ratings) / ratings.count()
return 0.0
@property
def cached_image_url(self):
if self.image:
return self.image.url
return self.image_url
class Rating(models.Model):
user = models.ForeignKey(User, related_name='ratings', on_delete=models.CASCADE)
flavor = models.ForeignKey(Flavor, related_name='ratings', on_delete=models.CASCADE)
score = models.IntegerField(validators=[MinValueValidator(1), MaxValueValidator(10)])
comment = models.TextField(blank=True, null=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
unique_together = ('user', 'flavor')
ordering = ['-created_at']
def __str__(self):
return f"{self.user.username} - {self.flavor.name}: {self.score}"
class Reply(models.Model):
user = models.ForeignKey(User, related_name='replies', on_delete=models.CASCADE)
rating = models.ForeignKey(Rating, related_name='replies', on_delete=models.CASCADE)
text = models.TextField()
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ['created_at']
def __str__(self):
return f"Reply by {self.user.username} on {self.rating.id}"
class Notification(models.Model):
TYPE_CHOICES = [
('reply', 'Reply'),
('mention', 'Mention'),
('follow', 'Follow'),
('ticket_new', 'New Ticket'),
('ticket_reply', 'Ticket Reply'),
('profile_comment', 'Profile Comment'),
]
recipient = models.ForeignKey(User, related_name='notifications', on_delete=models.CASCADE)
actor = models.ForeignKey(User, related_name='actions', on_delete=models.CASCADE)
notification_type = models.CharField(max_length=20, choices=TYPE_CHOICES)
rating = models.ForeignKey(Rating, on_delete=models.CASCADE, null=True, blank=True)
reply = models.ForeignKey(Reply, on_delete=models.CASCADE, null=True, blank=True)
ticket = models.ForeignKey('Ticket', on_delete=models.CASCADE, null=True, blank=True)
profile_comment = models.ForeignKey('ProfileComment', on_delete=models.CASCADE, null=True, blank=True)
is_read = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ['-created_at']
def __str__(self):
return f"{self.notification_type} for {self.recipient.username}"
class UserIP(models.Model):
user = models.ForeignKey(User, related_name='ips', on_delete=models.CASCADE)
ip_address = models.GenericIPAddressField()
last_login = models.DateTimeField(auto_now=True)
class Meta:
unique_together = ('user', 'ip_address')
ordering = ['-last_login']
class Ticket(models.Model):
STATUS_CHOICES = [
('open', 'Open'),
('in_progress', 'In Progress'),
('resolved', 'Resolved'),
('closed', 'Closed'),
]
user = models.ForeignKey(User, related_name='tickets', on_delete=models.CASCADE)
subject = models.CharField(max_length=200)
description = models.TextField()
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='open')
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
ordering = ['-updated_at']
class TicketMessage(models.Model):
ticket = models.ForeignKey(Ticket, related_name='messages', on_delete=models.CASCADE)
user = models.ForeignKey(User, on_delete=models.CASCADE)
text = models.TextField()
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ['created_at']
class ProfileComment(models.Model):
profile_owner = models.ForeignKey(User, related_name='profile_comments', on_delete=models.CASCADE)
author = models.ForeignKey(User, related_name='authored_profile_comments', on_delete=models.CASCADE)
text = models.TextField()
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ['-created_at']
class Banner(models.Model):
name = models.CharField(max_length=100)
slug = models.SlugField(unique=True)
description = models.TextField(blank=True)
is_active = models.BooleanField(default=False, help_text="Designates this as the global default banner.")
is_enabled = models.BooleanField(default=True, help_text="Allows users to select this banner.")
settings = models.JSONField(default=dict, blank=True)
schema = models.JSONField(default=list, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
def save(self, *args, **kwargs):
if self.is_active:
# Set all others to inactive
Banner.objects.filter(is_active=True).exclude(pk=self.pk).update(is_active=False)
elif not Banner.objects.filter(is_active=True).exclude(pk=self.pk).exists():
# If this was the only active banner, don't allow deactivating it
# unless there's another one to take its place.
self.is_active = True
super().save(*args, **kwargs)
def __str__(self):
return self.name
class SystemConfig(models.Model):
"""Singleton model for application-wide settings editable via UI."""
site_name = models.CharField(max_length=100, default="Holy Flavors Archive")
maintenance_mode = models.BooleanField(default=False)
allow_new_signups = models.BooleanField(default=True)
require_email_verification = models.BooleanField(default=True)
updated_at = models.DateTimeField(auto_now=True)
def save(self, *args, **kwargs):
self.pk = 1 # Force singleton
super().save(*args, **kwargs)
@classmethod
def get_solo(cls):
obj, _ = cls.objects.get_or_create(pk=1)
return obj
class Meta:
verbose_name = "System Configuration"
class Job(models.Model):
"""Tracks background jobs, their status and scheduling."""
JOB_TYPES = [
('sync_flavors', 'Sync Holy Flavors'),
('cleanup_duplicates', 'Cleanup Duplicate Flavors'),
('backup_db', 'Database Backup'),
('seed_legacy', 'Seed Legacy Data'),
]
STATUS_CHOICES = [
('pending', 'Pending'),
('running', 'Running'),
('completed', 'Completed'),
('failed', 'Failed'),
]
name = models.CharField(max_length=100, choices=JOB_TYPES)
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending')
last_run = models.DateTimeField(null=True, blank=True)
next_run = models.DateTimeField(null=True, blank=True)
interval_hours = models.IntegerField(default=0, help_text="Set to 0 to disable periodic run.")
last_output = models.TextField(blank=True)
error_message = models.TextField(blank=True)
def __str__(self):
return f"{self.get_name_display()} ({self.status})"
+22
View File
@@ -0,0 +1,22 @@
from api.models.flavor import Category, Flavor
from api.models.rating import Rating, Reply
from api.models.social import Notification, ProfileComment
from api.models.support import Ticket, TicketMessage
from api.models.system import Banner, Job, SystemConfig
from api.models.user import User, UserIP
__all__ = [
"Banner",
"Category",
"Flavor",
"Job",
"Notification",
"ProfileComment",
"Rating",
"Reply",
"SystemConfig",
"Ticket",
"TicketMessage",
"User",
"UserIP",
]
+51
View File
@@ -0,0 +1,51 @@
from django.db import models
from django.db.models import Avg
class Category(models.Model):
name = models.CharField(max_length=50, unique=True)
slug = models.SlugField(unique=True)
class Meta:
verbose_name_plural = "Categories"
ordering = ["name"]
def __str__(self) -> str:
return self.name
class Flavor(models.Model):
category = models.ForeignKey(Category, related_name="flavors", on_delete=models.CASCADE)
name = models.CharField(max_length=100)
description = models.TextField(blank=True)
image_url = models.URLField(max_length=1000, blank=True, null=True)
image_urls = models.JSONField(blank=True, default=list)
image = models.ImageField(upload_to="flavors/", blank=True, null=True)
shop_url = models.URLField(max_length=500, blank=True, null=True)
is_available = models.BooleanField(default=True)
is_legacy = models.BooleanField(default=False)
external_id = models.BigIntegerField(unique=True, null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
constraints = [
models.UniqueConstraint(
fields=["name", "category"],
name="unique_flavor_name_per_category",
)
]
ordering = ["name"]
def __str__(self) -> str:
return f"{self.name} ({self.category.name})"
@property
def get_average_rating(self) -> float:
result = self.ratings.aggregate(avg=Avg("score"))["avg"]
return result or 0.0
@property
def cached_image_url(self) -> str | None:
if self.image:
return self.image.url
return self.image_url
+30
View File
@@ -0,0 +1,30 @@
from django.core.validators import MaxValueValidator, MinValueValidator
from django.db import models
class Rating(models.Model):
user = models.ForeignKey("api.User", related_name="ratings", on_delete=models.CASCADE)
flavor = models.ForeignKey("api.Flavor", related_name="ratings", on_delete=models.CASCADE)
score = models.IntegerField(validators=[MinValueValidator(1), MaxValueValidator(10)])
comment = models.TextField(blank=True, null=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
unique_together = ("user", "flavor")
ordering = ["-created_at"]
def __str__(self) -> str:
return f"{self.user.username} - {self.flavor.name}: {self.score}"
class Reply(models.Model):
user = models.ForeignKey("api.User", related_name="replies", on_delete=models.CASCADE)
rating = models.ForeignKey(Rating, related_name="replies", on_delete=models.CASCADE)
text = models.TextField()
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ["created_at"]
def __str__(self) -> str:
return f"Reply by {self.user.username} on {self.rating.id}"
+45
View File
@@ -0,0 +1,45 @@
from django.db import models
class Notification(models.Model):
TYPE_CHOICES = [
("reply", "Reply"),
("mention", "Mention"),
("follow", "Follow"),
("ticket_new", "New Ticket"),
("ticket_reply", "Ticket Reply"),
("profile_comment", "Profile Comment"),
]
recipient = models.ForeignKey(
"api.User", related_name="notifications", on_delete=models.CASCADE
)
actor = models.ForeignKey("api.User", related_name="actions", on_delete=models.CASCADE)
notification_type = models.CharField(max_length=20, choices=TYPE_CHOICES)
rating = models.ForeignKey("api.Rating", on_delete=models.CASCADE, null=True, blank=True)
reply = models.ForeignKey("api.Reply", on_delete=models.CASCADE, null=True, blank=True)
ticket = models.ForeignKey("api.Ticket", on_delete=models.CASCADE, null=True, blank=True)
profile_comment = models.ForeignKey(
"api.ProfileComment", on_delete=models.CASCADE, null=True, blank=True
)
is_read = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ["-created_at"]
def __str__(self) -> str:
return f"{self.notification_type} for {self.recipient.username}"
class ProfileComment(models.Model):
profile_owner = models.ForeignKey(
"api.User", related_name="profile_comments", on_delete=models.CASCADE
)
author = models.ForeignKey(
"api.User", related_name="authored_profile_comments", on_delete=models.CASCADE
)
text = models.TextField()
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ["-created_at"]
+29
View File
@@ -0,0 +1,29 @@
from django.db import models
class Ticket(models.Model):
STATUS_CHOICES = [
("open", "Open"),
("in_progress", "In Progress"),
("resolved", "Resolved"),
("closed", "Closed"),
]
user = models.ForeignKey("api.User", related_name="tickets", on_delete=models.CASCADE)
subject = models.CharField(max_length=200)
description = models.TextField()
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default="open")
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
ordering = ["-updated_at"]
class TicketMessage(models.Model):
ticket = models.ForeignKey(Ticket, related_name="messages", on_delete=models.CASCADE)
user = models.ForeignKey("api.User", on_delete=models.CASCADE)
text = models.TextField()
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ["created_at"]
+77
View File
@@ -0,0 +1,77 @@
from django.db import models
class Banner(models.Model):
name = models.CharField(max_length=100)
slug = models.SlugField(unique=True)
description = models.TextField(blank=True)
is_active = models.BooleanField(
default=False, help_text="Designates this as the global default banner."
)
is_enabled = models.BooleanField(default=True, help_text="Allows users to select this banner.")
settings = models.JSONField(default=dict, blank=True)
schema = models.JSONField(default=list, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
def save(self, *args, **kwargs):
if self.is_active:
Banner.objects.filter(is_active=True).exclude(pk=self.pk).update(is_active=False)
elif not Banner.objects.filter(is_active=True).exclude(pk=self.pk).exists():
self.is_active = True
super().save(*args, **kwargs)
def __str__(self) -> str:
return self.name
class SystemConfig(models.Model):
"""Singleton model for application-wide settings editable via UI."""
site_name = models.CharField(max_length=100, default="Holy Flavors Archive")
maintenance_mode = models.BooleanField(default=False)
allow_new_signups = models.BooleanField(default=True)
require_email_verification = models.BooleanField(default=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
verbose_name = "System Configuration"
def save(self, *args, **kwargs):
self.pk = 1
super().save(*args, **kwargs)
@classmethod
def get_solo(cls) -> "SystemConfig":
obj, _ = cls.objects.get_or_create(pk=1)
return obj
class Job(models.Model):
"""Tracks background jobs, their status and scheduling."""
JOB_TYPES = [
("sync_flavors", "Sync Holy Flavors"),
("cleanup_duplicates", "Cleanup Duplicate Flavors"),
("backup_db", "Database Backup"),
("seed_legacy", "Seed Legacy Data"),
]
STATUS_CHOICES = [
("pending", "Pending"),
("running", "Running"),
("completed", "Completed"),
("failed", "Failed"),
]
name = models.CharField(max_length=100, choices=JOB_TYPES)
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default="pending")
last_run = models.DateTimeField(null=True, blank=True)
next_run = models.DateTimeField(null=True, blank=True)
interval_hours = models.IntegerField(default=0, help_text="Set to 0 to disable periodic run.")
last_output = models.TextField(blank=True)
error_message = models.TextField(blank=True)
def __str__(self) -> str:
return f"{self.get_name_display()} ({self.status})"
+72
View File
@@ -0,0 +1,72 @@
from django.contrib.auth.models import AbstractUser
from django.db import models
class User(AbstractUser):
THEME_CHOICES = [
("holy_light", "Holy Light"),
("holy_dark", "Holy Dark"),
("latte", "Catppuccin Latte"),
("frappe", "Catppuccin Frappé"),
("macchiato", "Catppuccin Macchiato"),
("mocha", "Catppuccin Mocha"),
("pink_pastel", "Pink Pastel"),
("mint_pastel", "Mint Pastel"),
("lavender_pastel", "Lavender Pastel"),
("dracula", "Dracula"),
("nord", "Nordic Frost"),
("gruvbox", "Gruvbox Retro"),
("oceanic", "Oceanic Deep"),
("t0p_sai", "SAI (T0P)"),
("t0p_trench", "Trench (T0P)"),
("t0p_blurryface", "Blurryface (T0P)"),
("t0p_clancy", "Clancy (T0P)"),
]
LANGUAGE_CHOICES = [
("en", "English"),
("de", "Deutsch"),
]
theme = models.CharField(max_length=20, choices=THEME_CHOICES, default="holy_light")
language = models.CharField(max_length=2, choices=LANGUAGE_CHOICES, default="en")
avatar = models.ImageField(upload_to="avatars/", blank=True, null=True)
pending_email = models.EmailField(max_length=254, blank=True, null=True)
email_confirmation_code = models.CharField(max_length=6, blank=True, null=True)
following = models.ManyToManyField(
"self", symmetrical=False, related_name="followers", blank=True
)
selected_banner = models.ForeignKey(
"api.Banner",
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="users",
)
def save(self, *args, **kwargs):
super().save(*args, **kwargs)
if self.avatar:
try:
from PIL import Image
img = Image.open(self.avatar.path)
if img.height > 256 or img.width > 256 or img.format != "JPEG":
output_size = (256, 256)
img.thumbnail(output_size)
if img.mode in ("RGBA", "P"):
img = img.convert("RGB")
img.save(self.avatar.path, "JPEG", quality=90)
except Exception:
pass
def __str__(self) -> str:
return self.username
class UserIP(models.Model):
user = models.ForeignKey(User, related_name="ips", on_delete=models.CASCADE)
ip_address = models.GenericIPAddressField()
last_login = models.DateTimeField(auto_now=True)
class Meta:
unique_together = ("user", "ip_address")
ordering = ["-last_login"]
-237
View File
@@ -1,237 +0,0 @@
from rest_framework import serializers
from .models import User, Flavor, Category, Rating, Reply, Notification, Ticket, TicketMessage, UserIP, Banner, SystemConfig, Job
class CategorySerializer(serializers.ModelSerializer):
class Meta:
model = Category
fields = ['id', 'name', 'slug']
class SystemConfigSerializer(serializers.ModelSerializer):
class Meta:
model = SystemConfig
fields = ['site_name', 'maintenance_mode', 'allow_new_signups', 'require_email_verification', 'updated_at']
class JobSerializer(serializers.ModelSerializer):
name_display = serializers.CharField(source='get_name_display', read_only=True)
class Meta:
model = Job
fields = ['id', 'name', 'name_display', 'status', 'last_run', 'next_run', 'interval_hours', 'last_output', 'error_message']
read_only_fields = ['status', 'last_run', 'next_run', 'last_output', 'error_message']
class BannerSerializer(serializers.ModelSerializer):
class Meta:
model = Banner
fields = ['id', 'name', 'slug', 'description', 'is_active', 'is_enabled', 'settings', 'schema', 'created_at', 'updated_at']
class ReplySerializer(serializers.ModelSerializer):
user = serializers.StringRelatedField(read_only=True)
class Meta:
model = Reply
fields = ['id', 'user', 'rating', 'text', 'created_at']
read_only_fields = ['rating']
class RatingSerializer(serializers.ModelSerializer):
user = serializers.StringRelatedField(read_only=True)
user_id = serializers.IntegerField(source='user.id', read_only=True)
user_avatar = serializers.SerializerMethodField()
flavor_name = serializers.CharField(source='flavor.name', read_only=True)
flavor_image = serializers.SerializerMethodField()
category_name = serializers.CharField(source='flavor.category.name', read_only=True)
category_slug = serializers.CharField(source='flavor.category.slug', read_only=True)
is_available = serializers.BooleanField(source='flavor.is_available', read_only=True)
is_legacy = serializers.BooleanField(source='flavor.is_legacy', read_only=True)
replies = ReplySerializer(many=True, read_only=True)
class Meta:
model = Rating
fields = ['id', 'user', 'user_id', 'user_avatar', 'flavor', 'flavor_name', 'flavor_image', 'category_name', 'category_slug', 'is_available', 'is_legacy', 'score', 'comment', 'created_at', 'replies']
def get_user_avatar(self, obj):
if obj.user.avatar:
request = self.context.get('request')
if request:
return request.build_absolute_uri(obj.user.avatar.url)
return obj.user.avatar.url
return None
def get_flavor_image(self, obj):
if obj.flavor.image:
request = self.context.get('request')
if request:
return request.build_absolute_uri(obj.flavor.image.url)
return obj.flavor.image.url
return obj.flavor.image_url
class FlavorSerializer(serializers.ModelSerializer):
category_name = serializers.CharField(source='category.name', read_only=True)
category_slug = serializers.CharField(source='category.slug', read_only=True)
average_rating = serializers.FloatField(read_only=True)
followed_average_rating = serializers.FloatField(read_only=True)
user_rating = serializers.SerializerMethodField()
ratings = RatingSerializer(many=True, read_only=True)
image_url = serializers.SerializerMethodField()
class Meta:
model = Flavor
fields = ['id', 'name', 'category', 'category_name', 'category_slug', 'description', 'average_rating', 'followed_average_rating', 'user_rating', 'ratings', 'image_url', 'image', 'is_available', 'is_legacy', 'shop_url']
extra_kwargs = {
'image': {'write_only': True}
}
def get_image_url(self, obj):
if obj.image:
request = self.context.get('request')
if request:
return request.build_absolute_uri(obj.image.url)
return obj.image.url
return obj.image_url
def get_user_rating(self, obj):
request = self.context.get('request')
if request and request.user.is_authenticated:
try:
rating = request.user.ratings.get(flavor=obj)
return rating.score
except Rating.DoesNotExist:
return None
return None
class UserSerializer(serializers.ModelSerializer):
avatar = serializers.SerializerMethodField()
following_count = serializers.IntegerField(source='following.count', read_only=True)
followers_count = serializers.IntegerField(source='followers.count', read_only=True)
is_following = serializers.SerializerMethodField()
unread_notifications_count = serializers.SerializerMethodField()
is_superuser = serializers.BooleanField(read_only=True)
selected_banner_slug = serializers.CharField(source='selected_banner.slug', read_only=True, allow_null=True)
class Meta:
model = User
fields = ['id', 'username', 'email', 'theme', 'language', 'avatar', 'following_count', 'followers_count', 'is_following', 'unread_notifications_count', 'is_superuser', 'selected_banner', 'selected_banner_slug']
def get_avatar(self, obj):
if obj.avatar:
request = self.context.get('request')
if request:
return request.build_absolute_uri(obj.avatar.url)
return obj.avatar.url
return None
def get_is_following(self, obj):
request = self.context.get('request')
if request and request.user.is_authenticated:
return request.user.following.filter(pk=obj.pk).exists()
return False
def get_unread_notifications_count(self, obj):
return obj.notifications.filter(is_read=False).count()
class UserIPSerializer(serializers.ModelSerializer):
class Meta:
model = UserIP
fields = ['id', 'ip_address', 'last_login']
class TicketMessageSerializer(serializers.ModelSerializer):
username = serializers.CharField(source='user.username', read_only=True)
is_admin = serializers.BooleanField(source='user.is_superuser', read_only=True)
class Meta:
model = TicketMessage
fields = ['id', 'username', 'text', 'created_at', 'is_admin']
class TicketSerializer(serializers.ModelSerializer):
username = serializers.CharField(source='user.username', read_only=True)
user_email = serializers.EmailField(source='user.email', read_only=True)
user_avatar = serializers.SerializerMethodField()
messages = TicketMessageSerializer(many=True, read_only=True)
class Meta:
model = Ticket
fields = ['id', 'user', 'username', 'user_email', 'user_avatar', 'subject', 'description', 'status', 'created_at', 'updated_at', 'messages']
read_only_fields = ['status', 'user']
def get_user_avatar(self, obj):
if obj.user.avatar:
request = self.context.get('request')
if request:
return request.build_absolute_uri(obj.user.avatar.url)
return obj.user.avatar.url
return None
class ProfileCommentSerializer(serializers.ModelSerializer):
author_username = serializers.CharField(source='author.username', read_only=True)
author_avatar = serializers.SerializerMethodField()
class Meta:
from .models import ProfileComment
model = ProfileComment
fields = ['id', 'author_username', 'author_avatar', 'text', 'created_at']
def get_author_avatar(self, obj):
if obj.author.avatar:
request = self.context.get('request')
if request:
return request.build_absolute_uri(obj.author.avatar.url)
return obj.author.avatar.url
return None
class AdminUserListSerializer(serializers.ModelSerializer):
ips = serializers.SerializerMethodField()
class Meta:
model = User
fields = ['id', 'username', 'email', 'is_active', 'is_superuser', 'date_joined', 'last_login', 'ips']
def get_ips(self, obj):
return [ip.ip_address for ip in obj.ips.all()[:3]]
class AdminUserDetailSerializer(serializers.ModelSerializer):
ips = serializers.SerializerMethodField()
ratings = RatingSerializer(many=True, read_only=True)
class Meta:
model = User
fields = ['id', 'username', 'email', 'is_active', 'is_superuser', 'date_joined', 'last_login', 'ips', 'ratings']
def get_ips(self, obj):
return [ip.ip_address for ip in obj.ips.all()]
class NotificationSerializer(serializers.ModelSerializer):
actor_username = serializers.CharField(source='actor.username', read_only=True)
actor_avatar = serializers.SerializerMethodField()
flavor_name = serializers.SerializerMethodField()
flavor_id = serializers.SerializerMethodField()
profile_owner_username = serializers.SerializerMethodField()
class Meta:
model = Notification
fields = ['id', 'actor_username', 'actor_avatar', 'notification_type', 'rating', 'reply', 'ticket', 'profile_comment', 'profile_owner_username', 'is_read', 'created_at', 'flavor_name', 'flavor_id']
def get_actor_avatar(self, obj):
if obj.actor.avatar:
request = self.context.get('request')
if request:
return request.build_absolute_uri(obj.actor.avatar.url)
return obj.actor.avatar.url
return None
def get_flavor_name(self, obj):
if obj.rating:
return obj.rating.flavor.name
if obj.reply:
return obj.reply.rating.flavor.name
return None
def get_flavor_id(self, obj):
if obj.rating:
return obj.rating.flavor.id
if obj.reply:
return obj.reply.rating.flavor.id
return None
def get_profile_owner_username(self, obj):
if obj.profile_comment:
return obj.profile_comment.profile_owner.username
return None
+33
View File
@@ -0,0 +1,33 @@
from api.serializers.flavor import CategorySerializer, FlavorSerializer
from api.serializers.rating import RatingSerializer, ReplySerializer
from api.serializers.social import NotificationSerializer, ProfileCommentSerializer
from api.serializers.support import TicketMessageSerializer, TicketSerializer
from api.serializers.system import (
BannerSerializer,
JobSerializer,
SystemConfigSerializer,
)
from api.serializers.user import (
AdminUserDetailSerializer,
AdminUserListSerializer,
UserIPSerializer,
UserSerializer,
)
__all__ = [
"AdminUserDetailSerializer",
"AdminUserListSerializer",
"BannerSerializer",
"CategorySerializer",
"FlavorSerializer",
"JobSerializer",
"NotificationSerializer",
"ProfileCommentSerializer",
"RatingSerializer",
"ReplySerializer",
"SystemConfigSerializer",
"TicketMessageSerializer",
"TicketSerializer",
"UserIPSerializer",
"UserSerializer",
]
+10
View File
@@ -0,0 +1,10 @@
from typing import Any
def absolute_image_url(serializer: Any, image_field: Any) -> str | None:
if not image_field:
return None
request = serializer.context.get("request")
if request:
return request.build_absolute_uri(image_field.url)
return image_field.url
+56
View File
@@ -0,0 +1,56 @@
from rest_framework import serializers
from api.models import Category, Flavor, Rating
from api.serializers._helpers import absolute_image_url
from api.serializers.rating import RatingSerializer
class CategorySerializer(serializers.ModelSerializer):
class Meta:
model = Category
fields = ["id", "name", "slug"]
class FlavorSerializer(serializers.ModelSerializer):
category_name = serializers.CharField(source="category.name", read_only=True)
category_slug = serializers.CharField(source="category.slug", read_only=True)
average_rating = serializers.FloatField(read_only=True)
followed_average_rating = serializers.FloatField(read_only=True)
user_rating = serializers.SerializerMethodField()
ratings = RatingSerializer(many=True, read_only=True)
image_url = serializers.SerializerMethodField()
class Meta:
model = Flavor
fields = [
"id",
"name",
"category",
"category_name",
"category_slug",
"description",
"average_rating",
"followed_average_rating",
"user_rating",
"ratings",
"image_url",
"image",
"is_available",
"is_legacy",
"shop_url",
]
extra_kwargs = {"image": {"write_only": True}}
def get_image_url(self, obj: Flavor) -> str | None:
if obj.image:
return absolute_image_url(self, obj.image)
return obj.image_url
def get_user_rating(self, obj: Flavor) -> int | None:
request = self.context.get("request")
if request and request.user.is_authenticated:
try:
return request.user.ratings.get(flavor=obj).score
except Rating.DoesNotExist:
return None
return None
+54
View File
@@ -0,0 +1,54 @@
from rest_framework import serializers
from api.models import Rating, Reply
from api.serializers._helpers import absolute_image_url
class ReplySerializer(serializers.ModelSerializer):
user = serializers.StringRelatedField(read_only=True)
class Meta:
model = Reply
fields = ["id", "user", "rating", "text", "created_at"]
read_only_fields = ["rating"]
class RatingSerializer(serializers.ModelSerializer):
user = serializers.StringRelatedField(read_only=True)
user_id = serializers.IntegerField(source="user.id", read_only=True)
user_avatar = serializers.SerializerMethodField()
flavor_name = serializers.CharField(source="flavor.name", read_only=True)
flavor_image = serializers.SerializerMethodField()
category_name = serializers.CharField(source="flavor.category.name", read_only=True)
category_slug = serializers.CharField(source="flavor.category.slug", read_only=True)
is_available = serializers.BooleanField(source="flavor.is_available", read_only=True)
is_legacy = serializers.BooleanField(source="flavor.is_legacy", read_only=True)
replies = ReplySerializer(many=True, read_only=True)
class Meta:
model = Rating
fields = [
"id",
"user",
"user_id",
"user_avatar",
"flavor",
"flavor_name",
"flavor_image",
"category_name",
"category_slug",
"is_available",
"is_legacy",
"score",
"comment",
"created_at",
"replies",
]
def get_user_avatar(self, obj: Rating) -> str | None:
return absolute_image_url(self, obj.user.avatar)
def get_flavor_image(self, obj: Rating) -> str | None:
if obj.flavor.image:
return absolute_image_url(self, obj.flavor.image)
return obj.flavor.image_url
+64
View File
@@ -0,0 +1,64 @@
from rest_framework import serializers
from api.models import Notification, ProfileComment
from api.serializers._helpers import absolute_image_url
class ProfileCommentSerializer(serializers.ModelSerializer):
author_username = serializers.CharField(source="author.username", read_only=True)
author_avatar = serializers.SerializerMethodField()
class Meta:
model = ProfileComment
fields = ["id", "author_username", "author_avatar", "text", "created_at"]
def get_author_avatar(self, obj: ProfileComment) -> str | None:
return absolute_image_url(self, obj.author.avatar)
class NotificationSerializer(serializers.ModelSerializer):
actor_username = serializers.CharField(source="actor.username", read_only=True)
actor_avatar = serializers.SerializerMethodField()
flavor_name = serializers.SerializerMethodField()
flavor_id = serializers.SerializerMethodField()
profile_owner_username = serializers.SerializerMethodField()
class Meta:
model = Notification
fields = [
"id",
"actor_username",
"actor_avatar",
"notification_type",
"rating",
"reply",
"ticket",
"profile_comment",
"profile_owner_username",
"is_read",
"created_at",
"flavor_name",
"flavor_id",
]
def get_actor_avatar(self, obj: Notification) -> str | None:
return absolute_image_url(self, obj.actor.avatar)
def get_flavor_name(self, obj: Notification) -> str | None:
if obj.rating:
return obj.rating.flavor.name
if obj.reply:
return obj.reply.rating.flavor.name
return None
def get_flavor_id(self, obj: Notification) -> int | None:
if obj.rating:
return obj.rating.flavor.id
if obj.reply:
return obj.reply.rating.flavor.id
return None
def get_profile_owner_username(self, obj: Notification) -> str | None:
if obj.profile_comment:
return obj.profile_comment.profile_owner.username
return None
+40
View File
@@ -0,0 +1,40 @@
from rest_framework import serializers
from api.models import Ticket, TicketMessage
from api.serializers._helpers import absolute_image_url
class TicketMessageSerializer(serializers.ModelSerializer):
username = serializers.CharField(source="user.username", read_only=True)
is_admin = serializers.BooleanField(source="user.is_superuser", read_only=True)
class Meta:
model = TicketMessage
fields = ["id", "username", "text", "created_at", "is_admin"]
class TicketSerializer(serializers.ModelSerializer):
username = serializers.CharField(source="user.username", read_only=True)
user_email = serializers.EmailField(source="user.email", read_only=True)
user_avatar = serializers.SerializerMethodField()
messages = TicketMessageSerializer(many=True, read_only=True)
class Meta:
model = Ticket
fields = [
"id",
"user",
"username",
"user_email",
"user_avatar",
"subject",
"description",
"status",
"created_at",
"updated_at",
"messages",
]
read_only_fields = ["status", "user"]
def get_user_avatar(self, obj: Ticket) -> str | None:
return absolute_image_url(self, obj.user.avatar)
+57
View File
@@ -0,0 +1,57 @@
from rest_framework import serializers
from api.models import Banner, Job, SystemConfig
class SystemConfigSerializer(serializers.ModelSerializer):
class Meta:
model = SystemConfig
fields = [
"site_name",
"maintenance_mode",
"allow_new_signups",
"require_email_verification",
"updated_at",
]
class JobSerializer(serializers.ModelSerializer):
name_display = serializers.CharField(source="get_name_display", read_only=True)
class Meta:
model = Job
fields = [
"id",
"name",
"name_display",
"status",
"last_run",
"next_run",
"interval_hours",
"last_output",
"error_message",
]
read_only_fields = [
"status",
"last_run",
"next_run",
"last_output",
"error_message",
]
class BannerSerializer(serializers.ModelSerializer):
class Meta:
model = Banner
fields = [
"id",
"name",
"slug",
"description",
"is_active",
"is_enabled",
"settings",
"schema",
"created_at",
"updated_at",
]
+95
View File
@@ -0,0 +1,95 @@
from rest_framework import serializers
from api.models import User, UserIP
from api.serializers._helpers import absolute_image_url
from api.serializers.rating import RatingSerializer
class UserSerializer(serializers.ModelSerializer):
avatar = serializers.SerializerMethodField()
following_count = serializers.IntegerField(source="following.count", read_only=True)
followers_count = serializers.IntegerField(source="followers.count", read_only=True)
is_following = serializers.SerializerMethodField()
unread_notifications_count = serializers.SerializerMethodField()
is_superuser = serializers.BooleanField(read_only=True)
selected_banner_slug = serializers.CharField(
source="selected_banner.slug", read_only=True, allow_null=True
)
class Meta:
model = User
fields = [
"id",
"username",
"email",
"theme",
"language",
"avatar",
"following_count",
"followers_count",
"is_following",
"unread_notifications_count",
"is_superuser",
"selected_banner",
"selected_banner_slug",
]
def get_avatar(self, obj: User) -> str | None:
return absolute_image_url(self, obj.avatar)
def get_is_following(self, obj: User) -> bool:
request = self.context.get("request")
if request and request.user.is_authenticated:
return request.user.following.filter(pk=obj.pk).exists()
return False
def get_unread_notifications_count(self, obj: User) -> int:
return obj.notifications.filter(is_read=False).count()
class UserIPSerializer(serializers.ModelSerializer):
class Meta:
model = UserIP
fields = ["id", "ip_address", "last_login"]
class AdminUserListSerializer(serializers.ModelSerializer):
ips = serializers.SerializerMethodField()
class Meta:
model = User
fields = [
"id",
"username",
"email",
"is_active",
"is_superuser",
"date_joined",
"last_login",
"ips",
]
def get_ips(self, obj: User) -> list[str]:
return [ip.ip_address for ip in obj.ips.all()[:3]]
class AdminUserDetailSerializer(serializers.ModelSerializer):
ips = serializers.SerializerMethodField()
ratings = RatingSerializer(many=True, read_only=True)
class Meta:
model = User
fields = [
"id",
"username",
"email",
"is_active",
"is_superuser",
"date_joined",
"last_login",
"ips",
"ratings",
]
def get_ips(self, obj: User) -> list[str]:
return [ip.ip_address for ip in obj.ips.all()]
View File
Whitespace-only changes.
+14
View File
@@ -0,0 +1,14 @@
from rest_framework.request import Request
from api.models import User, UserIP
def log_user_ip(user: User, request: Request) -> None:
x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
if x_forwarded_for:
ip = x_forwarded_for.split(",")[0].strip()
else:
ip = request.META.get("REMOTE_ADDR")
if ip and user.is_authenticated:
UserIP.objects.update_or_create(user=user, ip_address=ip)
+29
View File
@@ -0,0 +1,29 @@
import re
from api.models import Notification, Rating, Reply, User
_MENTION_RE = re.compile(r"@(\w+)")
def parse_mentions(
text: str,
actor: User,
rating: Rating | None = None,
reply: Reply | None = None,
) -> None:
if not text:
return
for username in set(_MENTION_RE.findall(text)):
try:
recipient = User.objects.get(username__iexact=username)
except User.DoesNotExist:
continue
if recipient == actor:
continue
Notification.objects.get_or_create(
recipient=recipient,
actor=actor,
notification_type="mention",
rating=rating,
reply=reply,
)
+44
View File
@@ -0,0 +1,44 @@
from django.db.models import Q, QuerySet
from api.models import Flavor
CATEGORY_KEYWORDS: dict[str, str] = {
"iced tea": "iced-tea",
"eistee": "iced-tea",
"energy": "energy",
"hydration": "hydration",
"milkshake": "milkshake",
}
def extract_category_slug(query: str) -> tuple[str | None, str]:
lowered = query.lower().strip()
for keyword, slug in CATEGORY_KEYWORDS.items():
if keyword in lowered:
return slug, lowered.replace(keyword, "").strip()
return None, lowered
def filter_flavors_by_query(queryset: QuerySet[Flavor], query: str) -> QuerySet[Flavor]:
slug, remaining = extract_category_slug(query)
if slug:
queryset = queryset.filter(category__slug=slug)
if remaining:
word_query = Q()
for word in remaining.split():
if len(word) > 2:
word_query |= Q(name__icontains=word) | Q(description__icontains=word)
if word_query:
queryset = queryset.filter(word_query)
return queryset
def score_relevance(name: str, query: str) -> int:
name_lower = name.lower()
if name_lower == query:
return 3
if name_lower.startswith(query):
return 2
if query in name_lower:
return 1
return 0
+96
View File
@@ -0,0 +1,96 @@
"""Celery tasks for Holy Flavors.
Each @shared_task wraps an existing management command so that the same
code path remains usable from the CLI for manual ops. The legacy ``Job``
rows are kept as an execution log: tasks update the matching row with
status/output/timestamps so the admin UI keeps working.
"""
from __future__ import annotations
from datetime import timedelta
from io import StringIO
from typing import Any
from celery import shared_task
from celery.utils.log import get_task_logger
from django.core.mail import send_mail
from django.core.management import call_command
from django.utils import timezone
logger = get_task_logger(__name__)
def _run_command_job(job_name: str, command: str, *args: str) -> str:
"""Run a management command and mirror status/output into the Job row."""
from api.models import Job
job, _ = Job.objects.get_or_create(name=job_name)
job.status = "running"
job.last_run = timezone.now()
job.last_output = ""
job.error_message = ""
job.save(update_fields=["status", "last_run", "last_output", "error_message"])
out = StringIO()
try:
call_command(command, *args, stdout=out)
job.status = "completed"
except Exception as exc: # noqa: BLE001
job.status = "failed"
job.error_message = str(exc)
logger.exception("Job %s failed", job_name)
raise
finally:
job.last_output = out.getvalue()
if job.interval_hours > 0:
job.next_run = timezone.now() + timedelta(hours=job.interval_hours)
else:
job.next_run = None
job.save()
return job.last_output
@shared_task(name="api.sync_flavors")
def sync_flavors_task() -> str:
return _run_command_job("sync_flavors", "sync_flavors")
@shared_task(name="api.cleanup_duplicates")
def cleanup_duplicates_task() -> str:
return _run_command_job("cleanup_duplicates", "cleanup_duplicates")
@shared_task(name="api.backup_db")
def backup_db_task() -> str:
return _run_command_job("backup_db", "backup_db", "--full")
@shared_task(name="api.seed_legacy")
def seed_legacy_task() -> str:
return _run_command_job("seed_legacy", "seed_legacy_flavors")
@shared_task(
name="api.send_email",
autoretry_for=(Exception,),
retry_backoff=True,
retry_kwargs={"max_retries": 3},
)
def send_email_task(
subject: str,
message: str,
recipient_list: list[str],
from_email: str | None = None,
**kwargs: Any,
) -> int:
"""Async wrapper around django.core.mail.send_mail."""
return send_mail(
subject=subject,
message=message,
from_email=from_email,
recipient_list=recipient_list,
fail_silently=False,
**kwargs,
)
-45
View File
@@ -1,45 +0,0 @@
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from .models import Category, Flavor, User, Rating
class FlavorTopTests(APITestCase):
def setUp(self):
self.category_energy = Category.objects.create(name='Energy', slug='energy')
self.category_iced_tea = Category.objects.create(name='Iced Tea', slug='iced-tea')
self.user = User.objects.create_user(username='testuser', password='password')
# Create flavors
self.flavor1 = Flavor.objects.create(name='Energy 1', category=self.category_energy)
self.flavor2 = Flavor.objects.create(name='Iced Tea 1', category=self.category_iced_tea)
# Add ratings to make them appear in top (since we filter by ratings__isnull=False)
Rating.objects.create(user=self.user, flavor=self.flavor1, score=5)
Rating.objects.create(user=self.user, flavor=self.flavor2, score=4)
def test_top_flavors_all(self):
url = reverse('flavor-top')
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 2)
def test_top_flavors_filtered_energy(self):
url = reverse('flavor-top')
response = self.client.get(url, {'category': 'energy'})
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]['name'], 'Energy 1')
def test_top_flavors_filtered_iced_tea(self):
url = reverse('flavor-top')
response = self.client.get(url, {'category': 'iced-tea'})
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]['name'], 'Iced Tea 1')
def test_top_flavors_filtered_non_existent(self):
url = reverse('flavor-top')
response = self.client.get(url, {'category': 'non-existent'})
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 0)
View File
Whitespace-only changes.
+55
View File
@@ -0,0 +1,55 @@
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from api.models import User
class JWTAuthTests(APITestCase):
def setUp(self) -> None:
self.username = "jwtuser"
self.password = "correct-horse-battery-staple"
self.user = User.objects.create_user(username=self.username, password=self.password)
def test_obtain_pair_returns_access_and_refresh(self) -> None:
url = reverse("token_obtain_pair")
response = self.client.post(
url,
{"username": self.username, "password": self.password},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn("access", response.data)
self.assertIn("refresh", response.data)
def test_access_token_authenticates_protected_endpoint(self) -> None:
pair = self.client.post(
reverse("token_obtain_pair"),
{"username": self.username, "password": self.password},
format="json",
).data
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {pair['access']}")
response = self.client.get("/api/notifications/", follow=True)
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_refresh_issues_new_access_token(self) -> None:
pair = self.client.post(
reverse("token_obtain_pair"),
{"username": self.username, "password": self.password},
format="json",
).data
response = self.client.post(
reverse("token_refresh"),
{"refresh": pair["refresh"]},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn("access", response.data)
def test_invalid_refresh_is_rejected(self) -> None:
response = self.client.post(
reverse("token_refresh"),
{"refresh": "not-a-real-token"},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+45
View File
@@ -0,0 +1,45 @@
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from api.models import Category, Flavor, Rating, User
class FlavorTopTests(APITestCase):
def setUp(self) -> None:
self.category_energy = Category.objects.create(name="Energy", slug="energy")
self.category_iced_tea = Category.objects.create(name="Iced Tea", slug="iced-tea")
self.user = User.objects.create_user(username="testuser", password="password")
self.flavor1 = Flavor.objects.create(name="Energy 1", category=self.category_energy)
self.flavor2 = Flavor.objects.create(name="Iced Tea 1", category=self.category_iced_tea)
Rating.objects.create(user=self.user, flavor=self.flavor1, score=5)
Rating.objects.create(user=self.user, flavor=self.flavor2, score=4)
def test_top_flavors_all(self) -> None:
url = reverse("flavor-top")
response = self.client.get(url, follow=True)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 2)
def test_top_flavors_filtered_energy(self) -> None:
url = reverse("flavor-top")
response = self.client.get(url, {"category": "energy"}, follow=True)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]["name"], "Energy 1")
def test_top_flavors_filtered_iced_tea(self) -> None:
url = reverse("flavor-top")
response = self.client.get(url, {"category": "iced-tea"}, follow=True)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]["name"], "Iced Tea 1")
def test_top_flavors_filtered_non_existent(self) -> None:
url = reverse("flavor-top")
response = self.client.get(url, {"category": "non-existent"}, follow=True)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 0)
+19
View File
@@ -0,0 +1,19 @@
from django.test import TestCase
from api.models import Category, Flavor, Rating, User
class FlavorAverageRatingTests(TestCase):
def setUp(self) -> None:
self.category = Category.objects.create(name="Energy", slug="energy")
self.flavor = Flavor.objects.create(name="Test", category=self.category)
self.u1 = User.objects.create_user(username="u1", password="x")
self.u2 = User.objects.create_user(username="u2", password="x")
def test_average_zero_when_no_ratings(self) -> None:
self.assertEqual(self.flavor.get_average_rating, 0.0)
def test_average_uses_aggregate(self) -> None:
Rating.objects.create(user=self.u1, flavor=self.flavor, score=4)
Rating.objects.create(user=self.u2, flavor=self.flavor, score=8)
self.assertEqual(self.flavor.get_average_rating, 6.0)
+12 -11
View File
@@ -1,18 +1,19 @@
from django.urls import path, include
from django.urls import include, path
from rest_framework import routers
from . import views
router = routers.DefaultRouter()
router.register('users', views.UserViewSet)
router.register('flavors', views.FlavorViewSet)
router.register('categories', views.CategoryViewSet)
router.register('ratings', views.RatingViewSet)
router.register('replies', views.ReplyViewSet)
router.register('notifications', views.NotificationViewSet, basename='notification')
router.register('tickets', views.TicketViewSet, basename='ticket')
router.register('admin-custom', views.AdminViewSet, basename='admin-custom')
router.register('banners', views.BannerViewSet, basename='banner')
router.register("users", views.UserViewSet)
router.register("flavors", views.FlavorViewSet)
router.register("categories", views.CategoryViewSet)
router.register("ratings", views.RatingViewSet)
router.register("replies", views.ReplyViewSet)
router.register("notifications", views.NotificationViewSet, basename="notification")
router.register("tickets", views.TicketViewSet, basename="ticket")
router.register("admin-custom", views.AdminViewSet, basename="admin-custom")
router.register("banners", views.BannerViewSet, basename="banner")
urlpatterns = [
path('', include(router.urls)),
path("", include(router.urls)),
]
-979
View File
@@ -1,979 +0,0 @@
from django.core.mail import send_mail
from django.conf import settings
from rest_framework import viewsets, permissions, status, filters, serializers
from rest_framework.response import Response
from rest_framework.decorators import action
from django.db.models import Avg, Q
from django_filters.rest_framework import DjangoFilterBackend
from django_ratelimit.decorators import ratelimit
from django.utils.decorators import method_decorator
from rest_framework.pagination import PageNumberPagination
from .models import User, Flavor, Category, Rating, Reply, Notification, Ticket, TicketMessage, UserIP, ProfileComment, Banner, SystemConfig, Job
from .serializers import (
UserSerializer, FlavorSerializer, CategorySerializer, RatingSerializer,
ReplySerializer, NotificationSerializer, TicketSerializer,
TicketMessageSerializer, AdminUserListSerializer, AdminUserDetailSerializer,
ProfileCommentSerializer, BannerSerializer, SystemConfigSerializer, JobSerializer
)
def log_user_ip(user, request):
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
if x_forwarded_for:
ip = x_forwarded_for.split(',')[0]
else:
ip = request.META.get('REMOTE_ADDR')
if ip and user.is_authenticated:
UserIP.objects.update_or_create(user=user, ip_address=ip)
def parse_mentions(text, actor, rating=None, reply=None):
import re
mentions = re.findall(r'@(\w+)', text)
for username in set(mentions):
try:
recipient = User.objects.get(username__iexact=username)
if recipient != actor:
Notification.objects.get_or_create(
recipient=recipient,
actor=actor,
notification_type='mention',
rating=rating,
reply=reply
)
except User.DoesNotExist:
continue
class FeedPagination(PageNumberPagination):
page_size = 10
class FlavorViewSet(viewsets.ModelViewSet):
queryset = Flavor.objects.annotate(average_rating=Avg('ratings__score')).order_by('-average_rating')
serializer_class = FlavorSerializer
filter_backends = [DjangoFilterBackend, filters.SearchFilter]
filterset_fields = ['category', 'category__slug']
search_fields = ['name', 'description']
permission_classes = [permissions.IsAuthenticatedOrReadOnly]
def get_queryset(self):
qs = Flavor.objects.select_related('category').annotate(average_rating=Avg('ratings__score')).order_by('-average_rating').distinct()
# Handle category keywords in search query for main results too
search_query = self.request.query_params.get('search', '').lower().strip()
if search_query:
cat_keywords = {
'iced tea': 'iced-tea',
'eistee': 'iced-tea',
'energy': 'energy',
'hydration': 'hydration',
'milkshake': 'milkshake'
}
for keyword, slug in cat_keywords.items():
if keyword in search_query:
qs = qs.filter(category__slug=slug)
break # Only one category at a time
return qs
@action(detail=False, methods=['get'], permission_classes=[permissions.AllowAny])
def search(self, request):
query = request.query_params.get('q', '').lower().strip()
if not query:
return Response([])
# Start with all flavors
flavors = Flavor.objects.select_related('category').distinct()
# Intelligent category filtering based on keywords
cat_keywords = {
'iced tea': 'iced-tea',
'eistee': 'iced-tea',
'energy': 'energy',
'hydration': 'hydration',
'milkshake': 'milkshake'
}
active_cat_slug = None
remaining_query = query
for keyword, slug in cat_keywords.items():
if keyword in query:
active_cat_slug = slug
remaining_query = remaining_query.replace(keyword, '').strip()
break
if active_cat_slug:
flavors = flavors.filter(category__slug=active_cat_slug)
# Filter by remaining query words in name or description
if remaining_query:
words = remaining_query.split()
# Construct a combined OR query for all words to be more permissive (fuzzy-ish)
word_query = Q()
for word in words:
if len(word) > 2: # Only filter by words longer than 2 chars to avoid noise
word_query |= Q(name__icontains=word) | Q(description__icontains=word)
if word_query:
flavors = flavors.filter(word_query)
# Limit results and deduplicate in Python to be safe against DB inconsistencies
results_list = []
seen_keys = set()
# Priority sorting in Python
search_lower = remaining_query.lower()
# Order by name, but we'll also try to prioritize exact matches in Python
for f in flavors.order_by('name'):
# Create a unique key for this flavor (name + category)
key = f"{f.name.strip().lower()}|{f.category.slug}"
if key in seen_keys:
continue
seen_keys.add(key)
name_lower = f.name.lower()
relevance = 0
if name_lower == search_lower:
relevance = 3
elif name_lower.startswith(search_lower):
relevance = 2
elif search_lower in name_lower:
relevance = 1
results_list.append({
'id': f.id,
'name': f.name,
'type': 'flavor',
'subtitle': f.category.name,
'image_url': request.build_absolute_uri(f.image.url) if f.image else f.image_url,
'slug': None,
'relevance': relevance
})
# Sort by relevance desc, then name
results_list.sort(key=lambda x: (-x['relevance'], x['name']))
# Remove relevance helper and limit to 15
final_results = []
for r in results_list:
del r['relevance']
final_results.append(r)
if len(final_results) >= 15:
break
return Response(final_results)
@action(detail=False, methods=['get'], permission_classes=[permissions.AllowAny])
def top(self, request):
queryset = self.get_queryset().filter(ratings__isnull=False)
category_slug = request.query_params.get('category')
if category_slug:
queryset = queryset.filter(category__slug=category_slug)
top_flavors = queryset.distinct()[:10]
serializer = self.get_serializer(top_flavors, many=True)
return Response(serializer.data)
@action(detail=False, methods=['get'], permission_classes=[permissions.AllowAny])
def newest(self, request):
# Get the 10 newest flavors
newest_flavors = Flavor.objects.select_related('category').order_by('-created_at')[:10]
serializer = self.get_serializer(newest_flavors, many=True)
return Response(serializer.data)
@action(detail=False, methods=['get'], permission_classes=[permissions.IsAuthenticated])
def followed_top(self, request):
followed_users = request.user.following.all()
top_flavors = Flavor.objects.filter(ratings__user__in=followed_users) \
.annotate(average_rating=Avg('ratings__score')) \
.order_by('-average_rating') \
.distinct()[:10]
serializer = self.get_serializer(top_flavors, many=True)
return Response(serializer.data)
class CategoryViewSet(viewsets.ReadOnlyModelViewSet):
queryset = Category.objects.all()
serializer_class = CategorySerializer
class RatingViewSet(viewsets.ModelViewSet):
queryset = Rating.objects.all()
serializer_class = RatingSerializer
permission_classes = [permissions.IsAuthenticatedOrReadOnly]
@method_decorator(ratelimit(key='user', rate='10/m', method='POST', block=True))
def create(self, request, *args, **kwargs):
return super().create(request, *args, **kwargs)
def perform_create(self, serializer):
# Ensure user can only have one rating per flavor
flavor = serializer.validated_data['flavor']
if Rating.objects.filter(user=self.request.user, flavor=flavor).exists():
raise serializers.ValidationError('You have already rated this flavor.')
rating = serializer.save(user=self.request.user)
# Parse mentions in the rating comment
if rating.comment:
parse_mentions(rating.comment, self.request.user, rating=rating)
def get_queryset(self):
return Rating.objects.select_related('user', 'flavor', 'flavor__category').prefetch_related('replies', 'replies__user').order_by('-created_at')
@action(detail=False, methods=['get'], permission_classes=[permissions.IsAuthenticated])
def feed(self, request):
# Get ratings from people the user follows
followed_users = request.user.following.all()
feed_ratings = Rating.objects.filter(user__in=followed_users).select_related('user', 'flavor', 'flavor__category').prefetch_related('replies', 'replies__user').order_by('-created_at')
paginator = FeedPagination()
page = paginator.paginate_queryset(feed_ratings, request)
if page is not None:
serializer = self.get_serializer(page, many=True)
return paginator.get_paginated_response(serializer.data)
serializer = self.get_serializer(feed_ratings, many=True)
return Response(serializer.data)
def check_object_permissions(self, request, obj):
super().check_object_permissions(request, obj)
if request.method in ['PUT', 'PATCH', 'DELETE'] and obj.user != request.user and not request.user.is_superuser:
self.permission_denied(request, message='You cannot edit/delete this rating.')
@action(detail=False, methods=['get'], permission_classes=[permissions.AllowAny])
def recent(self, request):
recent_ratings = Rating.objects.filter(comment__isnull=False).exclude(comment='').select_related('user', 'flavor').order_by('-created_at')[:10]
serializer = self.get_serializer(recent_ratings, many=True)
return Response(serializer.data)
@action(detail=True, methods=['post'], permission_classes=[permissions.IsAuthenticated])
def reply(self, request, pk=None):
rating = self.get_object()
text = request.data.get('text')
if not text:
return Response({'error': 'Text is required'}, status=status.HTTP_400_BAD_REQUEST)
reply = Reply.objects.create(user=request.user, rating=rating, text=text)
# Notify the rating author if it's not the same person
if rating.user != request.user:
Notification.objects.create(
recipient=rating.user,
actor=request.user,
notification_type='reply',
rating=rating,
reply=reply
)
# Parse mentions in the reply text
parse_mentions(text, request.user, rating=rating, reply=reply)
return Response(ReplySerializer(reply, context={'request': request}).data, status=status.HTTP_201_CREATED)
class ReplyViewSet(viewsets.ModelViewSet):
queryset = Reply.objects.all()
serializer_class = ReplySerializer
permission_classes = [permissions.IsAuthenticated]
def get_queryset(self):
return Reply.objects.select_related('user', 'rating').order_by('created_at')
def perform_update(self, serializer):
reply = self.get_object()
if reply.user != self.request.user and not self.request.user.is_superuser:
raise permissions.PermissionDenied("You cannot edit this reply.")
new_reply = serializer.save()
# Parse mentions in the updated text
parse_mentions(new_reply.text, self.request.user, rating=new_reply.rating, reply=new_reply)
def perform_destroy(self, instance):
if instance.user != self.request.user and not self.request.user.is_superuser:
raise permissions.PermissionDenied("You cannot delete this reply.")
instance.delete()
class NotificationViewSet(viewsets.ReadOnlyModelViewSet):
serializer_class = NotificationSerializer
permission_classes = [permissions.IsAuthenticated]
pagination_class = None
def get_queryset(self):
return self.request.user.notifications.select_related('actor', 'rating__flavor', 'reply__rating__flavor').order_by('-created_at')
@action(detail=False, methods=['post'])
def mark_all_read(self, request):
request.user.notifications.filter(is_read=False).update(is_read=True)
return Response({'status': 'all read'})
@action(detail=True, methods=['post'])
def mark_read(self, request, pk=None):
notification = self.get_object()
notification.is_read = True
notification.save()
return Response({'status': 'marked read'})
class UserViewSet(viewsets.ModelViewSet):
queryset = User.objects.all()
serializer_class = UserSerializer
permission_classes = [permissions.IsAuthenticated]
def get_permissions(self):
if self.action in ['signup', 'verify_signup', 'request_password_reset', 'complete_password_reset']:
return [permissions.AllowAny()]
return super().get_permissions()
@action(detail=False, methods=['get'], permission_classes=[permissions.IsAuthenticated])
def following_list(self, request):
following = request.user.following.all()
serializer = self.get_serializer(following, many=True)
return Response(serializer.data)
@action(detail=False, methods=['post'], permission_classes=[permissions.AllowAny])
@method_decorator(ratelimit(key='ip', rate='5/h', method='POST', block=True))
def signup(self, request):
username = request.data.get('username')
email = request.data.get('email')
password = request.data.get('password')
if not username or not email or not password:
return Response({'error': 'Username, email and password required'}, status=status.HTTP_400_BAD_REQUEST)
if User.objects.filter(username=username).exists():
return Response({'error': 'Username already exists'}, status=status.HTTP_400_BAD_REQUEST)
if User.objects.filter(email=email).exists():
return Response({'error': 'Email already exists'}, status=status.HTTP_400_BAD_REQUEST)
import secrets
import string
code = ''.join(secrets.choice(string.digits) for _ in range(6))
user = User.objects.create_user(
username=username,
email=email,
password=password,
is_active=False,
email_confirmation_code=code
)
verification_link = f"{settings.FRONTEND_URL}/verify-email?username={username}&code={code}"
try:
send_mail(
'Verify your Holy Flavors account',
f'Hi {username},\n\nYour verification code is: {code}\n\nAlternatively, you can complete your registration by clicking the link below:\n{verification_link}\n\nWelcome to the archive!',
settings.DEFAULT_FROM_EMAIL,
[email],
fail_silently=False,
)
except Exception as e:
# If email fails for any reason, delete the user so they can try again
user.delete()
return Response(
{'error': f'Failed to send verification email. Please check your SMTP settings in the .env file. The server error was: {e}'},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
return Response({'status': 'User created, please verify your email', 'username': username, 'email': email}, status=status.HTTP_201_CREATED)
@action(detail=False, methods=['post'], permission_classes=[permissions.AllowAny])
def resend_verification(self, request):
username = request.data.get('username')
if not username:
return Response({'error': 'Username required'}, status=status.HTTP_400_BAD_REQUEST)
try:
user = User.objects.get(username=username, is_active=False)
except User.DoesNotExist:
return Response({'error': 'User not found or already active'}, status=status.HTTP_404_NOT_FOUND)
import secrets
import string
code = ''.join(secrets.choice(string.digits) for _ in range(6))
user.email_confirmation_code = code
user.save()
verification_link = f"{settings.FRONTEND_URL}/verify-email?username={user.username}&code={code}"
try:
send_mail(
'Verify your Holy Flavors account',
f'Hi {user.username},\n\nYour new verification code is: {code}\n\nAlternatively, you can complete your registration by clicking the link below:\n{verification_link}',
settings.DEFAULT_FROM_EMAIL,
[user.email],
fail_silently=False,
)
except Exception as e:
return Response(
{'error': f'Failed to send email: {e}'},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
return Response({'status': 'Verification code resent!'})
@action(detail=False, methods=['post'], permission_classes=[permissions.AllowAny])
def verify_signup(self, request):
username = request.data.get('username')
code = request.data.get('code')
try:
user = User.objects.get(username=username, is_active=False)
except User.DoesNotExist:
return Response({'error': 'User not found or already active'}, status=status.HTTP_404_NOT_FOUND)
if user.email_confirmation_code == code:
user.is_active = True
user.email_confirmation_code = None
user.save()
return Response({'status': 'Account verified successfully!'})
else:
return Response({'error': 'Invalid verification code'}, status=status.HTTP_400_BAD_REQUEST)
@action(detail=False, methods=['get'], permission_classes=[permissions.AllowAny], url_path='profile/(?P<username>[^/.]+)')
def public_profile(self, request, username=None):
try:
user = User.objects.get(username=username)
except User.DoesNotExist:
return Response({'error': 'User not found'}, status=status.HTTP_404_NOT_FOUND)
ratings = user.ratings.select_related('flavor', 'flavor__category').order_by('-score')
comments = user.profile_comments.select_related('author').order_by('-created_at')
followers = user.followers.all()
following = user.following.all()
return Response({
'id': user.id,
'username': user.username,
'theme': user.theme,
'avatar': request.build_absolute_uri(user.avatar.url) if user.avatar else None,
'following_count': following.count(),
'followers_count': followers.count(),
'is_following': request.user.following.filter(pk=user.pk).exists() if request.user.is_authenticated else False,
'ratings': RatingSerializer(ratings, many=True, context={'request': request}).data,
'comments': ProfileCommentSerializer(comments, many=True, context={'request': request}).data,
'followers': UserSerializer(followers, many=True, context={'request': request}).data,
'following': UserSerializer(following, many=True, context={'request': request}).data
})
@action(detail=True, methods=['post'], permission_classes=[permissions.IsAuthenticated])
def add_comment(self, request, pk=None):
profile_owner = self.get_object()
text = request.data.get('text')
if not text:
return Response({'error': 'Text is required'}, status=status.HTTP_400_BAD_REQUEST)
comment = ProfileComment.objects.create(
profile_owner=profile_owner,
author=request.user,
text=text
)
# Notify profile owner
if profile_owner != request.user:
Notification.objects.create(
recipient=profile_owner,
actor=request.user,
notification_type='profile_comment',
profile_comment=comment
)
return Response(ProfileCommentSerializer(comment, context={'request': request}).data, status=status.HTTP_201_CREATED)
@action(detail=True, methods=['delete'], url_path='delete_comment/(?P<comment_id>[^/.]+)', permission_classes=[permissions.IsAuthenticated])
def delete_comment(self, request, pk=None, comment_id=None):
profile_owner = self.get_object()
try:
comment = ProfileComment.objects.get(pk=comment_id, profile_owner=profile_owner)
except ProfileComment.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND)
if comment.author != request.user and not request.user.is_superuser and profile_owner != request.user:
return Response(status=status.HTTP_403_FORBIDDEN)
comment.delete()
return Response(status=status.HTTP_204_NO_CONTENT)
@action(detail=False, methods=['get'])
def me(self, request):
log_user_ip(request.user, request)
serializer = self.get_serializer(request.user)
return Response(serializer.data)
@action(detail=False, methods=['patch'])
def update_preferences(self, request):
user = request.user
theme = request.data.get('theme')
language = request.data.get('language')
banner_id = request.data.get('selected_banner')
updated = False
if theme in dict(User.THEME_CHOICES):
user.theme = theme
updated = True
if language in dict(User.LANGUAGE_CHOICES):
user.language = language
updated = True
if banner_id is not None:
if banner_id == "":
user.selected_banner = None
updated = True
else:
try:
banner = Banner.objects.get(pk=banner_id, is_enabled=True)
user.selected_banner = banner
updated = True
except Banner.DoesNotExist:
return Response({'error': 'Invalid or disabled banner'}, status=status.HTTP_400_BAD_REQUEST)
if updated:
user.save()
return Response({
'status': 'preferences updated',
'theme': user.theme,
'language': user.language,
'selected_banner': user.selected_banner.id if user.selected_banner else None
})
return Response({'error': 'No valid updates provided'}, status=status.HTTP_400_BAD_REQUEST)
@action(detail=False, methods=['post'])
def update_avatar(self, request):
user = request.user
avatar = request.FILES.get('avatar')
if not avatar:
return Response({'error': 'No avatar provided'}, status=status.HTTP_400_BAD_REQUEST)
# Check size (2MB = 2 * 1024 * 1024 bytes)
if avatar.size > 2 * 1024 * 1024:
return Response({'error': 'File size exceeds 2MB limit'}, status=status.HTTP_400_BAD_REQUEST)
user.avatar = avatar
user.save()
serializer = self.get_serializer(user)
return Response(serializer.data)
@action(detail=False, methods=['post'])
def change_password(self, request):
user = request.user
old_password = request.data.get('old_password')
new_password = request.data.get('new_password')
if not user.check_password(old_password):
return Response({'error': 'Wrong old password'}, status=status.HTTP_400_BAD_REQUEST)
user.set_password(new_password)
user.save()
return Response({'status': 'password changed'})
@action(detail=False, methods=['patch'])
def update_profile(self, request):
user = request.user
username = request.data.get('username')
email = request.data.get('email')
message = ""
if username and username != user.username:
if User.objects.filter(username=username).exists():
return Response({'error': 'Username already taken'}, status=status.HTTP_400_BAD_REQUEST)
user.username = username
if email and email != user.email:
import secrets
import string
code = ''.join(secrets.choice(string.digits) for _ in range(6))
user.pending_email = email
user.email_confirmation_code = code
send_mail(
'Confirm your new email',
f'Hi {user.username},\n\nYour confirmation code is: {code}',
settings.DEFAULT_FROM_EMAIL,
[email],
fail_silently=False,
)
message = " Confirmation code sent to new email."
user.save()
serializer = self.get_serializer(user)
data = serializer.data
if message:
data['message'] = message
return Response(data)
@action(detail=True, methods=['post'], permission_classes=[permissions.IsAuthenticated])
def follow(self, request, pk=None):
user_to_follow = self.get_object()
if user_to_follow == request.user:
return Response({'error': 'You cannot follow yourself'}, status=status.HTTP_400_BAD_REQUEST)
if not request.user.following.filter(pk=user_to_follow.pk).exists():
request.user.following.add(user_to_follow)
# Notify the user being followed
Notification.objects.create(
recipient=user_to_follow,
actor=request.user,
notification_type='follow'
)
return Response({'status': 'followed'})
@action(detail=True, methods=['post'], permission_classes=[permissions.IsAuthenticated])
def unfollow(self, request, pk=None):
user_to_unfollow = self.get_object()
request.user.following.remove(user_to_unfollow)
return Response({'status': 'unfollowed'})
@action(detail=False, methods=['post'])
def confirm_email(self, request):
user = request.user
code = request.data.get('code')
if not user.email_confirmation_code or code != user.email_confirmation_code:
return Response({'error': 'Invalid or expired code'}, status=status.HTTP_400_BAD_REQUEST)
user.email = user.pending_email
user.pending_email = None
user.email_confirmation_code = None
user.save()
return Response({'status': 'Email confirmed', 'email': user.email})
@action(detail=False, methods=['post'], permission_classes=[permissions.AllowAny])
@method_decorator(ratelimit(key='ip', rate='3/h', method='POST', block=True))
def request_password_reset(self, request):
email = request.data.get('email')
try:
user = User.objects.get(email=email, is_active=True)
except User.DoesNotExist:
return Response({'status': 'If an account exists with this email, a reset code has been sent.'})
import secrets
import string
code = ''.join(secrets.choice(string.digits) for _ in range(6))
user.email_confirmation_code = code
user.save()
send_mail(
'Password Reset Request',
f'Hi {user.username},\n\nYour password reset code is: {code}',
settings.DEFAULT_FROM_EMAIL,
[email],
fail_silently=False,
)
return Response({'status': 'If an account exists with this email, a reset code has been sent.'})
@action(detail=False, methods=['post'], permission_classes=[permissions.AllowAny])
@method_decorator(ratelimit(key='ip', rate='5/h', method='POST', block=True))
def complete_password_reset(self, request):
email = request.data.get('email')
code = request.data.get('code')
new_password = request.data.get('password')
try:
user = User.objects.get(email=email, email_confirmation_code=code)
except User.DoesNotExist:
return Response({'error': 'Invalid email or code'}, status=status.HTTP_400_BAD_REQUEST)
user.set_password(new_password)
user.email_confirmation_code = None
user.save()
return Response({'status': 'Password reset successful!'})
@action(detail=False, methods=['post'])
@method_decorator(ratelimit(key='user', rate='3/h', method='POST', block=True))
def request_account_deletion(self, request):
import secrets
import string
code = ''.join(secrets.choice(string.digits) for _ in range(6))
user = request.user
user.email_confirmation_code = code
user.save()
send_mail(
'Confirm Account Deletion - Holy Flavors Archive',
f'Hi {user.username},\n\nYou requested to delete your account. This action is permanent.\n\nYour deletion code is: {code}',
settings.DEFAULT_FROM_EMAIL,
[user.email],
fail_silently=False,
)
return Response({'status': 'Code sent to your email'})
@action(detail=False, methods=['post'])
def confirm_account_deletion(self, request):
user = request.user
code = request.data.get('code')
if not user.email_confirmation_code or code != user.email_confirmation_code:
return Response({'error': 'Invalid or expired code'}, status=status.HTTP_400_BAD_REQUEST)
user.delete()
return Response({'status': 'Account deleted'})
@action(detail=False, methods=['get'])
def dashboard(self, request):
user = request.user
rated_ids = user.ratings.values_list('flavor_id', flat=True)
followed_users = user.following.all()
# Missing flavors with community avg and followed circle avg
missing_flavors = Flavor.objects.exclude(id__in=rated_ids) \
.exclude(category__name='Packs and other') \
.select_related('category') \
.annotate(
average_rating=Avg('ratings__score'),
followed_average_rating=Avg('ratings__score', filter=Q(ratings__user__in=followed_users))
).order_by('category__name', 'name')
rated_flavors = user.ratings.select_related('flavor', 'flavor__category') \
.prefetch_related('replies', 'replies__user') \
.order_by('-created_at')
return Response({
'user': UserSerializer(user).data,
'rated_count': rated_flavors.count(),
'missing_count': missing_flavors.count(),
'missing_flavors': FlavorSerializer(missing_flavors, many=True, context={'request': request}).data,
'my_ratings': RatingSerializer(rated_flavors, many=True, context={'request': request}).data
})
class TicketViewSet(viewsets.ModelViewSet):
serializer_class = TicketSerializer
permission_classes = [permissions.IsAuthenticated]
def get_queryset(self):
if self.request.user.is_superuser:
return Ticket.objects.all().prefetch_related('messages', 'messages__user').order_by('-updated_at')
return Ticket.objects.filter(user=self.request.user).prefetch_related('messages', 'messages__user').order_by('-updated_at')
def perform_create(self, serializer):
ticket = serializer.save(user=self.request.user)
# Notify all superusers about new ticket
admins = User.objects.filter(is_superuser=True).exclude(pk=self.request.user.pk)
for admin in admins:
Notification.objects.create(
recipient=admin,
actor=self.request.user,
notification_type='ticket_new',
ticket=ticket
)
@action(detail=True, methods=['post'])
def add_message(self, request, pk=None):
ticket = self.get_object()
text = request.data.get('text')
if not text:
return Response({'error': 'Text is required'}, status=status.HTTP_400_BAD_REQUEST)
msg = TicketMessage.objects.create(ticket=ticket, user=request.user, text=text)
# Notification logic
if request.user.is_superuser:
# Notify the user who opened the ticket
if ticket.user != request.user:
Notification.objects.create(
recipient=ticket.user,
actor=request.user,
notification_type='ticket_reply',
ticket=ticket
)
else:
# Notify all superusers
admins = User.objects.filter(is_superuser=True).exclude(pk=request.user.pk)
for admin in admins:
Notification.objects.create(
recipient=admin,
actor=request.user,
notification_type='ticket_reply',
ticket=ticket
)
return Response(TicketMessageSerializer(msg).data, status=status.HTTP_201_CREATED)
@action(detail=True, methods=['post'], permission_classes=[permissions.IsAdminUser])
def update_status(self, request, pk=None):
ticket = self.get_object()
new_status = request.data.get('status')
if new_status in dict(Ticket.STATUS_CHOICES):
ticket.status = new_status
ticket.save()
return Response({'status': 'updated'})
return Response({'error': 'Invalid status'}, status=status.HTTP_400_BAD_REQUEST)
def destroy(self, request, *args, **kwargs):
if not request.user.is_superuser:
raise permissions.PermissionDenied("Only admins can delete tickets.")
return super().destroy(request, *args, **kwargs)
class BannerViewSet(viewsets.ModelViewSet):
queryset = Banner.objects.all()
serializer_class = BannerSerializer
permission_classes = [permissions.IsAuthenticatedOrReadOnly]
pagination_class = None
def get_queryset(self):
if self.request.user.is_authenticated and self.request.user.is_superuser:
return Banner.objects.all()
return Banner.objects.filter(is_enabled=True)
def get_permissions(self):
if self.action in ['list', 'retrieve', 'active']:
return [permissions.AllowAny()]
return [permissions.IsAdminUser()]
@action(detail=False, methods=['get'])
def active(self, request):
# Determine which banner to show
# Priority:
# 1. Requested user's preference (if username provided in query)
# 2. Logged in user's preference
# 3. Global default
username = request.query_params.get('username')
user = None
if username:
try:
user = User.objects.get(username=username)
except User.DoesNotExist:
pass
if not user and request.user.is_authenticated:
user = request.user
if user and user.selected_banner and user.selected_banner.is_enabled:
serializer = self.get_serializer(user.selected_banner)
return Response(serializer.data)
active_banner = Banner.objects.filter(is_active=True).first()
if active_banner:
serializer = self.get_serializer(active_banner)
return Response(serializer.data)
return Response(None)
@action(detail=True, methods=['post'])
def activate(self, request, pk=None):
banner = self.get_object()
banner.is_active = True
banner.is_enabled = True # Active banner must be enabled
banner.save()
return Response({'status': 'banner set as global default'})
@action(detail=True, methods=['post'])
def toggle_enabled(self, request, pk=None):
banner = self.get_object()
if banner.is_active:
return Response({'error': 'Cannot disable the global default banner.'}, status=status.HTTP_400_BAD_REQUEST)
banner.is_enabled = not banner.is_enabled
banner.save()
return Response({'status': 'enabled' if banner.is_enabled else 'disabled'})
class AdminViewSet(viewsets.ViewSet):
permission_classes = [permissions.IsAdminUser]
@action(detail=False, methods=['get'])
def stats(self, request):
return Response({
'total_users': User.objects.count(),
'total_ratings': Rating.objects.count(),
'total_replies': Reply.objects.count(),
'open_tickets': Ticket.objects.filter(status='open').count(),
'email_config': {
'host': getattr(settings, 'EMAIL_HOST', 'None'),
'port': getattr(settings, 'EMAIL_PORT', 'None'),
'use_tls': getattr(settings, 'EMAIL_USE_TLS', False),
'use_ssl': getattr(settings, 'EMAIL_USE_SSL', False),
'skip_verify': getattr(settings, 'EMAIL_SKIP_CERT_VERIFICATION', False),
},
'server_info': {
'debug': settings.DEBUG,
'allowed_hosts': settings.ALLOWED_HOSTS,
'frontend_url': getattr(settings, 'FRONTEND_URL', 'Not set'),
'media_root': str(settings.MEDIA_ROOT),
'static_root': str(settings.STATIC_ROOT),
}
})
@action(detail=False, methods=['get', 'patch'])
def config(self, request):
config = SystemConfig.get_solo()
if request.method == 'PATCH':
serializer = SystemConfigSerializer(config, data=request.data, partial=True)
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(serializer.data)
serializer = SystemConfigSerializer(config)
return Response(serializer.data)
@action(detail=False, methods=['get'])
def jobs(self, request):
# Ensure all jobs exist
for job_type, label in Job.JOB_TYPES:
Job.objects.get_or_create(name=job_type)
jobs = Job.objects.all().order_by('name')
serializer = JobSerializer(jobs, many=True)
return Response(serializer.data)
@action(detail=True, methods=['post'])
def trigger_job(self, request, pk=None):
try:
job = Job.objects.get(pk=pk)
job.status = 'pending'
job.save()
return Response({'status': 'Job queued'})
except Job.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND)
@action(detail=True, methods=['patch'])
def update_job_schedule(self, request, pk=None):
try:
job = Job.objects.get(pk=pk)
interval = request.data.get('interval_hours')
next_run = request.data.get('next_run')
if interval is not None:
job.interval_hours = int(interval)
if next_run is not None:
from django.utils.dateparse import parse_datetime
from django.utils import timezone
parsed_dt = parse_datetime(next_run)
if parsed_dt:
# If it's naive (has no timezone info), assume it's local time
if timezone.is_naive(parsed_dt):
job.next_run = timezone.make_aware(parsed_dt, timezone.get_current_timezone())
else:
job.next_run = parsed_dt
elif interval is not None and job.interval_hours > 0 and not job.next_run:
from django.utils import timezone
job.next_run = timezone.now()
job.save()
return Response(JobSerializer(job).data)
except Job.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND)
@action(detail=False, methods=['post'])
def send_test_email(self, request):
try:
send_mail(
'Holy Flavors Admin Test Email',
f'This is a test email sent to {request.user.email} from the Holy Flavors Admin Interface.',
settings.DEFAULT_FROM_EMAIL,
[request.user.email],
fail_silently=False,
)
return Response({'status': 'Test email sent!'})
except Exception as e:
return Response({'error': str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
@action(detail=False, methods=['get'])
def users(self, request):
users = User.objects.all().prefetch_related('ips').order_by('-date_joined')
return Response(AdminUserListSerializer(users, many=True).data)
@action(detail=True, methods=['get', 'patch', 'delete'])
def user_detail(self, request, pk=None):
try:
user = User.objects.get(pk=pk)
except User.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND)
if request.method == 'GET':
return Response(AdminUserDetailSerializer(user, context={'request': request}).data)
elif request.method == 'PATCH':
is_active = request.data.get('is_active')
if is_active is not None:
user.is_active = is_active
user.save()
return Response(AdminUserDetailSerializer(user, context={'request': request}).data)
elif request.method == 'DELETE':
user.delete()
return Response(status=status.HTTP_204_NO_CONTENT)
+20
View File
@@ -0,0 +1,20 @@
from api.views._pagination import FeedPagination
from api.views.admin import AdminViewSet
from api.views.banner import BannerViewSet
from api.views.flavor import CategoryViewSet, FlavorViewSet
from api.views.rating import RatingViewSet, ReplyViewSet
from api.views.support import TicketViewSet
from api.views.user import NotificationViewSet, UserViewSet
__all__ = [
"AdminViewSet",
"BannerViewSet",
"CategoryViewSet",
"FeedPagination",
"FlavorViewSet",
"NotificationViewSet",
"RatingViewSet",
"ReplyViewSet",
"TicketViewSet",
"UserViewSet",
]
+5
View File
@@ -0,0 +1,5 @@
from rest_framework.pagination import PageNumberPagination
class FeedPagination(PageNumberPagination):
page_size = 10
+145
View File
@@ -0,0 +1,145 @@
from django.conf import settings
from django.core.mail import send_mail
from rest_framework import permissions, status, viewsets
from rest_framework.decorators import action
from rest_framework.request import Request
from rest_framework.response import Response
from api.models import Job, Rating, Reply, SystemConfig, Ticket, User
from api.serializers import (
AdminUserDetailSerializer,
AdminUserListSerializer,
JobSerializer,
SystemConfigSerializer,
)
class AdminViewSet(viewsets.ViewSet):
permission_classes = [permissions.IsAdminUser]
@action(detail=False, methods=["get"])
def stats(self, request: Request) -> Response:
return Response(
{
"total_users": User.objects.count(),
"total_ratings": Rating.objects.count(),
"total_replies": Reply.objects.count(),
"open_tickets": Ticket.objects.filter(status="open").count(),
"email_config": {
"host": getattr(settings, "EMAIL_HOST", "None"),
"port": getattr(settings, "EMAIL_PORT", "None"),
"use_tls": getattr(settings, "EMAIL_USE_TLS", False),
"use_ssl": getattr(settings, "EMAIL_USE_SSL", False),
"skip_verify": getattr(settings, "EMAIL_SKIP_CERT_VERIFICATION", False),
},
"server_info": {
"debug": settings.DEBUG,
"allowed_hosts": settings.ALLOWED_HOSTS,
"frontend_url": getattr(settings, "FRONTEND_URL", "Not set"),
"media_root": str(settings.MEDIA_ROOT),
"static_root": str(settings.STATIC_ROOT),
},
}
)
@action(detail=False, methods=["get", "patch"])
def config(self, request: Request) -> Response:
config = SystemConfig.get_solo()
if request.method == "PATCH":
serializer = SystemConfigSerializer(config, data=request.data, partial=True)
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(serializer.data)
return Response(SystemConfigSerializer(config).data)
@action(detail=False, methods=["get"])
def jobs(self, request: Request) -> Response:
for job_type, _ in Job.JOB_TYPES:
Job.objects.get_or_create(name=job_type)
jobs = Job.objects.all().order_by("name")
return Response(JobSerializer(jobs, many=True).data)
@action(detail=True, methods=["post"])
def trigger_job(self, request: Request, pk=None) -> Response:
try:
job = Job.objects.get(pk=pk)
job.status = "pending"
job.save()
return Response({"status": "Job queued"})
except Job.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND)
@action(detail=True, methods=["patch"])
def update_job_schedule(self, request: Request, pk=None) -> Response:
try:
job = Job.objects.get(pk=pk)
interval = request.data.get("interval_hours")
next_run = request.data.get("next_run")
if interval is not None:
job.interval_hours = int(interval)
if next_run is not None:
from django.utils import timezone
from django.utils.dateparse import parse_datetime
parsed_dt = parse_datetime(next_run)
if parsed_dt:
if timezone.is_naive(parsed_dt):
job.next_run = timezone.make_aware(
parsed_dt, timezone.get_current_timezone()
)
else:
job.next_run = parsed_dt
elif interval is not None and job.interval_hours > 0 and not job.next_run:
from django.utils import timezone
job.next_run = timezone.now()
job.save()
return Response(JobSerializer(job).data)
except Job.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND)
@action(detail=False, methods=["post"])
def send_test_email(self, request: Request) -> Response:
try:
send_mail(
"Holy Flavors Admin Test Email",
(
f"This is a test email sent to {request.user.email} from the "
f"Holy Flavors Admin Interface."
),
settings.DEFAULT_FROM_EMAIL,
[request.user.email],
fail_silently=False,
)
return Response({"status": "Test email sent!"})
except Exception as e:
return Response({"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
@action(detail=False, methods=["get"])
def users(self, request: Request) -> Response:
users = User.objects.all().prefetch_related("ips").order_by("-date_joined")
return Response(AdminUserListSerializer(users, many=True).data)
@action(detail=True, methods=["get", "patch", "delete"])
def user_detail(self, request: Request, pk=None) -> Response:
try:
user = User.objects.get(pk=pk)
except User.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND)
if request.method == "GET":
return Response(AdminUserDetailSerializer(user, context={"request": request}).data)
elif request.method == "PATCH":
is_active = request.data.get("is_active")
if is_active is not None:
user.is_active = is_active
user.save()
return Response(AdminUserDetailSerializer(user, context={"request": request}).data)
elif request.method == "DELETE":
user.delete()
return Response(status=status.HTTP_204_NO_CONTENT)
+65
View File
@@ -0,0 +1,65 @@
from rest_framework import permissions, status, viewsets
from rest_framework.decorators import action
from rest_framework.request import Request
from rest_framework.response import Response
from api.models import Banner, User
from api.serializers import BannerSerializer
class BannerViewSet(viewsets.ModelViewSet):
queryset = Banner.objects.all()
serializer_class = BannerSerializer
permission_classes = [permissions.IsAuthenticatedOrReadOnly]
pagination_class = None
def get_queryset(self):
if self.request.user.is_authenticated and self.request.user.is_superuser:
return Banner.objects.all()
return Banner.objects.filter(is_enabled=True)
def get_permissions(self):
if self.action in ["list", "retrieve", "active"]:
return [permissions.AllowAny()]
return [permissions.IsAdminUser()]
@action(detail=False, methods=["get"])
def active(self, request: Request) -> Response:
username = request.query_params.get("username")
user = None
if username:
try:
user = User.objects.get(username=username)
except User.DoesNotExist:
pass
if not user and request.user.is_authenticated:
user = request.user
if user and user.selected_banner and user.selected_banner.is_enabled:
return Response(self.get_serializer(user.selected_banner).data)
active_banner = Banner.objects.filter(is_active=True).first()
if active_banner:
return Response(self.get_serializer(active_banner).data)
return Response(None)
@action(detail=True, methods=["post"])
def activate(self, request: Request, pk=None) -> Response:
banner = self.get_object()
banner.is_active = True
banner.is_enabled = True
banner.save()
return Response({"status": "banner set as global default"})
@action(detail=True, methods=["post"])
def toggle_enabled(self, request: Request, pk=None) -> Response:
banner = self.get_object()
if banner.is_active:
return Response(
{"error": "Cannot disable the global default banner."},
status=status.HTTP_400_BAD_REQUEST,
)
banner.is_enabled = not banner.is_enabled
banner.save()
return Response({"status": "enabled" if banner.is_enabled else "disabled"})
+105
View File
@@ -0,0 +1,105 @@
from django.db.models import Avg
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework import filters, permissions, viewsets
from rest_framework.decorators import action
from rest_framework.request import Request
from rest_framework.response import Response
from api.models import Category, Flavor
from api.serializers import CategorySerializer, FlavorSerializer
from api.services.search import (
extract_category_slug,
filter_flavors_by_query,
score_relevance,
)
class CategoryViewSet(viewsets.ReadOnlyModelViewSet):
queryset = Category.objects.all()
serializer_class = CategorySerializer
class FlavorViewSet(viewsets.ModelViewSet):
queryset = Flavor.objects.annotate(average_rating=Avg("ratings__score")).order_by(
"-average_rating"
)
serializer_class = FlavorSerializer
filter_backends = [DjangoFilterBackend, filters.SearchFilter]
filterset_fields = ["category", "category__slug"]
search_fields = ["name", "description"]
permission_classes = [permissions.IsAuthenticatedOrReadOnly]
def get_queryset(self):
qs = (
Flavor.objects.select_related("category")
.annotate(average_rating=Avg("ratings__score"))
.order_by("-average_rating")
.distinct()
)
search_query = self.request.query_params.get("search", "")
slug, _ = extract_category_slug(search_query)
if slug:
qs = qs.filter(category__slug=slug)
return qs
@action(detail=False, methods=["get"], permission_classes=[permissions.AllowAny])
def search(self, request: Request) -> Response:
query = request.query_params.get("q", "").lower().strip()
if not query:
return Response([])
flavors = Flavor.objects.select_related("category").distinct()
flavors = filter_flavors_by_query(flavors, query)
_, remaining = extract_category_slug(query)
seen_keys: set[str] = set()
results: list[dict] = []
for f in flavors.order_by("name"):
key = f"{f.name.strip().lower()}|{f.category.slug}"
if key in seen_keys:
continue
seen_keys.add(key)
results.append(
{
"id": f.id,
"name": f.name,
"type": "flavor",
"subtitle": f.category.name,
"image_url": (
request.build_absolute_uri(f.image.url) if f.image else f.image_url
),
"slug": None,
"relevance": score_relevance(f.name, remaining),
}
)
results.sort(key=lambda x: (-x["relevance"], x["name"]))
for r in results:
del r["relevance"]
return Response(results[:15])
@action(detail=False, methods=["get"], permission_classes=[permissions.AllowAny])
def top(self, request: Request) -> Response:
queryset = self.get_queryset().filter(ratings__isnull=False)
category_slug = request.query_params.get("category")
if category_slug:
queryset = queryset.filter(category__slug=category_slug)
top_flavors = queryset.distinct()[:10]
return Response(self.get_serializer(top_flavors, many=True).data)
@action(detail=False, methods=["get"], permission_classes=[permissions.AllowAny])
def newest(self, request: Request) -> Response:
newest_flavors = Flavor.objects.select_related("category").order_by("-created_at")[:10]
return Response(self.get_serializer(newest_flavors, many=True).data)
@action(detail=False, methods=["get"], permission_classes=[permissions.IsAuthenticated])
def followed_top(self, request: Request) -> Response:
followed_users = request.user.following.all()
top_flavors = (
Flavor.objects.filter(ratings__user__in=followed_users)
.annotate(average_rating=Avg("ratings__score"))
.order_by("-average_rating")
.distinct()[:10]
)
return Response(self.get_serializer(top_flavors, many=True).data)
+120
View File
@@ -0,0 +1,120 @@
from django.utils.decorators import method_decorator
from django_ratelimit.decorators import ratelimit
from rest_framework import permissions, serializers, status, viewsets
from rest_framework.decorators import action
from rest_framework.exceptions import PermissionDenied
from rest_framework.request import Request
from rest_framework.response import Response
from api.models import Notification, Rating, Reply
from api.serializers import RatingSerializer, ReplySerializer
from api.services.mentions import parse_mentions
from api.views._pagination import FeedPagination
class RatingViewSet(viewsets.ModelViewSet):
queryset = Rating.objects.all()
serializer_class = RatingSerializer
permission_classes = [permissions.IsAuthenticatedOrReadOnly]
@method_decorator(ratelimit(key="user", rate="10/m", method="POST", block=True))
def create(self, request, *args, **kwargs):
return super().create(request, *args, **kwargs)
def perform_create(self, serializer):
flavor = serializer.validated_data["flavor"]
if Rating.objects.filter(user=self.request.user, flavor=flavor).exists():
raise serializers.ValidationError("You have already rated this flavor.")
rating = serializer.save(user=self.request.user)
if rating.comment:
parse_mentions(rating.comment, self.request.user, rating=rating)
def get_queryset(self):
return (
Rating.objects.select_related("user", "flavor", "flavor__category")
.prefetch_related("replies", "replies__user")
.order_by("-created_at")
)
@action(detail=False, methods=["get"], permission_classes=[permissions.IsAuthenticated])
def feed(self, request: Request) -> Response:
followed_users = request.user.following.all()
feed_ratings = (
Rating.objects.filter(user__in=followed_users)
.select_related("user", "flavor", "flavor__category")
.prefetch_related("replies", "replies__user")
.order_by("-created_at")
)
paginator = FeedPagination()
page = paginator.paginate_queryset(feed_ratings, request)
if page is not None:
serializer = self.get_serializer(page, many=True)
return paginator.get_paginated_response(serializer.data)
return Response(self.get_serializer(feed_ratings, many=True).data)
def check_object_permissions(self, request, obj):
super().check_object_permissions(request, obj)
if (
request.method in ["PUT", "PATCH", "DELETE"]
and obj.user != request.user
and not request.user.is_superuser
):
self.permission_denied(request, message="You cannot edit/delete this rating.")
@action(detail=False, methods=["get"], permission_classes=[permissions.AllowAny])
def recent(self, request: Request) -> Response:
recent_ratings = (
Rating.objects.filter(comment__isnull=False)
.exclude(comment="")
.select_related("user", "flavor")
.order_by("-created_at")[:10]
)
return Response(self.get_serializer(recent_ratings, many=True).data)
@action(detail=True, methods=["post"], permission_classes=[permissions.IsAuthenticated])
def reply(self, request: Request, pk=None) -> Response:
rating = self.get_object()
text = request.data.get("text")
if not text:
return Response({"error": "Text is required"}, status=status.HTTP_400_BAD_REQUEST)
reply = Reply.objects.create(user=request.user, rating=rating, text=text)
if rating.user != request.user:
Notification.objects.create(
recipient=rating.user,
actor=request.user,
notification_type="reply",
rating=rating,
reply=reply,
)
parse_mentions(text, request.user, rating=rating, reply=reply)
return Response(
ReplySerializer(reply, context={"request": request}).data,
status=status.HTTP_201_CREATED,
)
class ReplyViewSet(viewsets.ModelViewSet):
queryset = Reply.objects.all()
serializer_class = ReplySerializer
permission_classes = [permissions.IsAuthenticated]
def get_queryset(self):
return Reply.objects.select_related("user", "rating").order_by("created_at")
def perform_update(self, serializer):
reply = self.get_object()
if reply.user != self.request.user and not self.request.user.is_superuser:
raise PermissionDenied("You cannot edit this reply.")
new_reply = serializer.save()
parse_mentions(new_reply.text, self.request.user, rating=new_reply.rating, reply=new_reply)
def perform_destroy(self, instance):
if instance.user != self.request.user and not self.request.user.is_superuser:
raise PermissionDenied("You cannot delete this reply.")
instance.delete()
+81
View File
@@ -0,0 +1,81 @@
from rest_framework import permissions, status, viewsets
from rest_framework.decorators import action
from rest_framework.exceptions import PermissionDenied
from rest_framework.request import Request
from rest_framework.response import Response
from api.models import Notification, Ticket, TicketMessage, User
from api.serializers import TicketMessageSerializer, TicketSerializer
class TicketViewSet(viewsets.ModelViewSet):
serializer_class = TicketSerializer
permission_classes = [permissions.IsAuthenticated]
def get_queryset(self):
if self.request.user.is_superuser:
return (
Ticket.objects.all()
.prefetch_related("messages", "messages__user")
.order_by("-updated_at")
)
return (
Ticket.objects.filter(user=self.request.user)
.prefetch_related("messages", "messages__user")
.order_by("-updated_at")
)
def perform_create(self, serializer):
ticket = serializer.save(user=self.request.user)
admins = User.objects.filter(is_superuser=True).exclude(pk=self.request.user.pk)
for admin in admins:
Notification.objects.create(
recipient=admin,
actor=self.request.user,
notification_type="ticket_new",
ticket=ticket,
)
@action(detail=True, methods=["post"])
def add_message(self, request: Request, pk=None) -> Response:
ticket = self.get_object()
text = request.data.get("text")
if not text:
return Response({"error": "Text is required"}, status=status.HTTP_400_BAD_REQUEST)
msg = TicketMessage.objects.create(ticket=ticket, user=request.user, text=text)
if request.user.is_superuser:
if ticket.user != request.user:
Notification.objects.create(
recipient=ticket.user,
actor=request.user,
notification_type="ticket_reply",
ticket=ticket,
)
else:
admins = User.objects.filter(is_superuser=True).exclude(pk=request.user.pk)
for admin in admins:
Notification.objects.create(
recipient=admin,
actor=request.user,
notification_type="ticket_reply",
ticket=ticket,
)
return Response(TicketMessageSerializer(msg).data, status=status.HTTP_201_CREATED)
@action(detail=True, methods=["post"], permission_classes=[permissions.IsAdminUser])
def update_status(self, request: Request, pk=None) -> Response:
ticket = self.get_object()
new_status = request.data.get("status")
if new_status in dict(Ticket.STATUS_CHOICES):
ticket.status = new_status
ticket.save()
return Response({"status": "updated"})
return Response({"error": "Invalid status"}, status=status.HTTP_400_BAD_REQUEST)
def destroy(self, request, *args, **kwargs):
if not request.user.is_superuser:
raise PermissionDenied("Only admins can delete tickets.")
return super().destroy(request, *args, **kwargs)
+546
View File
@@ -0,0 +1,546 @@
import secrets
import string
from django.conf import settings
from django.utils.decorators import method_decorator
from django_ratelimit.decorators import ratelimit
from rest_framework import permissions, status, viewsets
from rest_framework.decorators import action
from rest_framework.request import Request
from rest_framework.response import Response
from api.models import Banner, Flavor, Notification, ProfileComment, User
from api.serializers import (
FlavorSerializer,
NotificationSerializer,
ProfileCommentSerializer,
RatingSerializer,
UserSerializer,
)
from api.services.ip_logging import log_user_ip
from api.tasks import send_email_task
def _generate_code() -> str:
return "".join(secrets.choice(string.digits) for _ in range(6))
class UserViewSet(viewsets.ModelViewSet):
queryset = User.objects.all()
serializer_class = UserSerializer
permission_classes = [permissions.IsAuthenticated]
def get_permissions(self):
if self.action in [
"signup",
"verify_signup",
"request_password_reset",
"complete_password_reset",
]:
return [permissions.AllowAny()]
return super().get_permissions()
@action(detail=False, methods=["get"], permission_classes=[permissions.IsAuthenticated])
def following_list(self, request: Request) -> Response:
following = request.user.following.all()
return Response(self.get_serializer(following, many=True).data)
@action(detail=False, methods=["post"], permission_classes=[permissions.AllowAny])
@method_decorator(ratelimit(key="ip", rate="5/h", method="POST", block=True))
def signup(self, request: Request) -> Response:
username = request.data.get("username")
email = request.data.get("email")
password = request.data.get("password")
if not username or not email or not password:
return Response(
{"error": "Username, email and password required"},
status=status.HTTP_400_BAD_REQUEST,
)
if User.objects.filter(username=username).exists():
return Response(
{"error": "Username already exists"}, status=status.HTTP_400_BAD_REQUEST
)
if User.objects.filter(email=email).exists():
return Response({"error": "Email already exists"}, status=status.HTTP_400_BAD_REQUEST)
code = _generate_code()
user = User.objects.create_user(
username=username,
email=email,
password=password,
is_active=False,
email_confirmation_code=code,
)
verification_link = f"{settings.FRONTEND_URL}/verify-email?username={username}&code={code}"
try:
send_email_task.delay(
subject="Verify your Holy Flavors account",
message=(
f"Hi {username},\n\nYour verification code is: {code}\n\n"
f"Alternatively, you can complete your registration by clicking the link below:\n"
f"{verification_link}\n\nWelcome to the archive!"
),
recipient_list=[email],
from_email=settings.DEFAULT_FROM_EMAIL,
)
except Exception as e:
user.delete()
return Response(
{
"error": (
f"Failed to send verification email. Please check your SMTP settings "
f"in the .env file. The server error was: {e}"
)
},
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
return Response(
{
"status": "User created, please verify your email",
"username": username,
"email": email,
},
status=status.HTTP_201_CREATED,
)
@action(detail=False, methods=["post"], permission_classes=[permissions.AllowAny])
def resend_verification(self, request: Request) -> Response:
username = request.data.get("username")
if not username:
return Response({"error": "Username required"}, status=status.HTTP_400_BAD_REQUEST)
try:
user = User.objects.get(username=username, is_active=False)
except User.DoesNotExist:
return Response(
{"error": "User not found or already active"},
status=status.HTTP_404_NOT_FOUND,
)
code = _generate_code()
user.email_confirmation_code = code
user.save()
verification_link = (
f"{settings.FRONTEND_URL}/verify-email?username={user.username}&code={code}"
)
try:
send_email_task.delay(
subject="Verify your Holy Flavors account",
message=(
f"Hi {user.username},\n\nYour new verification code is: {code}\n\n"
f"Alternatively, you can complete your registration by clicking the link below:\n"
f"{verification_link}"
),
recipient_list=[user.email],
from_email=settings.DEFAULT_FROM_EMAIL,
)
except Exception as e:
return Response(
{"error": f"Failed to send email: {e}"},
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
return Response({"status": "Verification code resent!"})
@action(detail=False, methods=["post"], permission_classes=[permissions.AllowAny])
def verify_signup(self, request: Request) -> Response:
username = request.data.get("username")
code = request.data.get("code")
try:
user = User.objects.get(username=username, is_active=False)
except User.DoesNotExist:
return Response(
{"error": "User not found or already active"},
status=status.HTTP_404_NOT_FOUND,
)
if user.email_confirmation_code == code:
user.is_active = True
user.email_confirmation_code = None
user.save()
return Response({"status": "Account verified successfully!"})
return Response({"error": "Invalid verification code"}, status=status.HTTP_400_BAD_REQUEST)
@action(
detail=False,
methods=["get"],
permission_classes=[permissions.AllowAny],
url_path="profile/(?P<username>[^/.]+)",
)
def public_profile(self, request: Request, username=None) -> Response:
try:
user = User.objects.get(username=username)
except User.DoesNotExist:
return Response({"error": "User not found"}, status=status.HTTP_404_NOT_FOUND)
ratings = user.ratings.select_related("flavor", "flavor__category").order_by("-score")
comments = user.profile_comments.select_related("author").order_by("-created_at")
followers = user.followers.all()
following = user.following.all()
return Response(
{
"id": user.id,
"username": user.username,
"theme": user.theme,
"avatar": (request.build_absolute_uri(user.avatar.url) if user.avatar else None),
"following_count": following.count(),
"followers_count": followers.count(),
"is_following": (
request.user.following.filter(pk=user.pk).exists()
if request.user.is_authenticated
else False
),
"ratings": RatingSerializer(ratings, many=True, context={"request": request}).data,
"comments": ProfileCommentSerializer(
comments, many=True, context={"request": request}
).data,
"followers": UserSerializer(
followers, many=True, context={"request": request}
).data,
"following": UserSerializer(
following, many=True, context={"request": request}
).data,
}
)
@action(detail=True, methods=["post"], permission_classes=[permissions.IsAuthenticated])
def add_comment(self, request: Request, pk=None) -> Response:
profile_owner = self.get_object()
text = request.data.get("text")
if not text:
return Response({"error": "Text is required"}, status=status.HTTP_400_BAD_REQUEST)
comment = ProfileComment.objects.create(
profile_owner=profile_owner,
author=request.user,
text=text,
)
if profile_owner != request.user:
Notification.objects.create(
recipient=profile_owner,
actor=request.user,
notification_type="profile_comment",
profile_comment=comment,
)
return Response(
ProfileCommentSerializer(comment, context={"request": request}).data,
status=status.HTTP_201_CREATED,
)
@action(
detail=True,
methods=["delete"],
url_path="delete_comment/(?P<comment_id>[^/.]+)",
permission_classes=[permissions.IsAuthenticated],
)
def delete_comment(self, request: Request, pk=None, comment_id=None) -> Response:
profile_owner = self.get_object()
try:
comment = ProfileComment.objects.get(pk=comment_id, profile_owner=profile_owner)
except ProfileComment.DoesNotExist:
return Response(status=status.HTTP_404_NOT_FOUND)
if (
comment.author != request.user
and not request.user.is_superuser
and profile_owner != request.user
):
return Response(status=status.HTTP_403_FORBIDDEN)
comment.delete()
return Response(status=status.HTTP_204_NO_CONTENT)
@action(detail=False, methods=["get"])
def me(self, request: Request) -> Response:
log_user_ip(request.user, request)
return Response(self.get_serializer(request.user).data)
@action(detail=False, methods=["patch"])
def update_preferences(self, request: Request) -> Response:
user = request.user
theme = request.data.get("theme")
language = request.data.get("language")
banner_id = request.data.get("selected_banner")
updated = False
if theme in dict(User.THEME_CHOICES):
user.theme = theme
updated = True
if language in dict(User.LANGUAGE_CHOICES):
user.language = language
updated = True
if banner_id is not None:
if banner_id == "":
user.selected_banner = None
updated = True
else:
try:
banner = Banner.objects.get(pk=banner_id, is_enabled=True)
user.selected_banner = banner
updated = True
except Banner.DoesNotExist:
return Response(
{"error": "Invalid or disabled banner"},
status=status.HTTP_400_BAD_REQUEST,
)
if updated:
user.save()
return Response(
{
"status": "preferences updated",
"theme": user.theme,
"language": user.language,
"selected_banner": (user.selected_banner.id if user.selected_banner else None),
}
)
return Response({"error": "No valid updates provided"}, status=status.HTTP_400_BAD_REQUEST)
@action(detail=False, methods=["post"])
def update_avatar(self, request: Request) -> Response:
user = request.user
avatar = request.FILES.get("avatar")
if not avatar:
return Response({"error": "No avatar provided"}, status=status.HTTP_400_BAD_REQUEST)
if avatar.size > 2 * 1024 * 1024:
return Response(
{"error": "File size exceeds 2MB limit"},
status=status.HTTP_400_BAD_REQUEST,
)
user.avatar = avatar
user.save()
return Response(self.get_serializer(user).data)
@action(detail=False, methods=["post"])
def change_password(self, request: Request) -> Response:
user = request.user
old_password = request.data.get("old_password")
new_password = request.data.get("new_password")
if not user.check_password(old_password):
return Response({"error": "Wrong old password"}, status=status.HTTP_400_BAD_REQUEST)
user.set_password(new_password)
user.save()
return Response({"status": "password changed"})
@action(detail=False, methods=["patch"])
def update_profile(self, request: Request) -> Response:
user = request.user
username = request.data.get("username")
email = request.data.get("email")
message = ""
if username and username != user.username:
if User.objects.filter(username=username).exists():
return Response(
{"error": "Username already taken"},
status=status.HTTP_400_BAD_REQUEST,
)
user.username = username
if email and email != user.email:
code = _generate_code()
user.pending_email = email
user.email_confirmation_code = code
send_email_task.delay(
subject="Confirm your new email",
message=f"Hi {user.username},\n\nYour confirmation code is: {code}",
recipient_list=[email],
from_email=settings.DEFAULT_FROM_EMAIL,
)
message = " Confirmation code sent to new email."
user.save()
data = self.get_serializer(user).data
if message:
data["message"] = message
return Response(data)
@action(detail=True, methods=["post"], permission_classes=[permissions.IsAuthenticated])
def follow(self, request: Request, pk=None) -> Response:
user_to_follow = self.get_object()
if user_to_follow == request.user:
return Response(
{"error": "You cannot follow yourself"},
status=status.HTTP_400_BAD_REQUEST,
)
if not request.user.following.filter(pk=user_to_follow.pk).exists():
request.user.following.add(user_to_follow)
Notification.objects.create(
recipient=user_to_follow,
actor=request.user,
notification_type="follow",
)
return Response({"status": "followed"})
@action(detail=True, methods=["post"], permission_classes=[permissions.IsAuthenticated])
def unfollow(self, request: Request, pk=None) -> Response:
user_to_unfollow = self.get_object()
request.user.following.remove(user_to_unfollow)
return Response({"status": "unfollowed"})
@action(detail=False, methods=["post"])
def confirm_email(self, request: Request) -> Response:
user = request.user
code = request.data.get("code")
if not user.email_confirmation_code or code != user.email_confirmation_code:
return Response(
{"error": "Invalid or expired code"}, status=status.HTTP_400_BAD_REQUEST
)
user.email = user.pending_email
user.pending_email = None
user.email_confirmation_code = None
user.save()
return Response({"status": "Email confirmed", "email": user.email})
@action(detail=False, methods=["post"], permission_classes=[permissions.AllowAny])
@method_decorator(ratelimit(key="ip", rate="3/h", method="POST", block=True))
def request_password_reset(self, request: Request) -> Response:
email = request.data.get("email")
try:
user = User.objects.get(email=email, is_active=True)
except User.DoesNotExist:
return Response(
{"status": "If an account exists with this email, a reset code has been sent."}
)
code = _generate_code()
user.email_confirmation_code = code
user.save()
send_email_task.delay(
subject="Password Reset Request",
message=f"Hi {user.username},\n\nYour password reset code is: {code}",
recipient_list=[email],
from_email=settings.DEFAULT_FROM_EMAIL,
)
return Response(
{"status": "If an account exists with this email, a reset code has been sent."}
)
@action(detail=False, methods=["post"], permission_classes=[permissions.AllowAny])
@method_decorator(ratelimit(key="ip", rate="5/h", method="POST", block=True))
def complete_password_reset(self, request: Request) -> Response:
email = request.data.get("email")
code = request.data.get("code")
new_password = request.data.get("password")
try:
user = User.objects.get(email=email, email_confirmation_code=code)
except User.DoesNotExist:
return Response({"error": "Invalid email or code"}, status=status.HTTP_400_BAD_REQUEST)
user.set_password(new_password)
user.email_confirmation_code = None
user.save()
return Response({"status": "Password reset successful!"})
@action(detail=False, methods=["post"])
@method_decorator(ratelimit(key="user", rate="3/h", method="POST", block=True))
def request_account_deletion(self, request: Request) -> Response:
code = _generate_code()
user = request.user
user.email_confirmation_code = code
user.save()
send_email_task.delay(
subject="Confirm Account Deletion - Holy Flavors Archive",
message=(
f"Hi {user.username},\n\nYou requested to delete your account. "
f"This action is permanent.\n\nYour deletion code is: {code}"
),
recipient_list=[user.email],
from_email=settings.DEFAULT_FROM_EMAIL,
)
return Response({"status": "Code sent to your email"})
@action(detail=False, methods=["post"])
def confirm_account_deletion(self, request: Request) -> Response:
user = request.user
code = request.data.get("code")
if not user.email_confirmation_code or code != user.email_confirmation_code:
return Response(
{"error": "Invalid or expired code"}, status=status.HTTP_400_BAD_REQUEST
)
user.delete()
return Response({"status": "Account deleted"})
@action(detail=False, methods=["get"])
def dashboard(self, request: Request) -> Response:
from django.db.models import Avg, Q
user = request.user
rated_ids = user.ratings.values_list("flavor_id", flat=True)
followed_users = user.following.all()
missing_flavors = (
Flavor.objects.exclude(id__in=rated_ids)
.exclude(category__name="Packs and other")
.select_related("category")
.annotate(
average_rating=Avg("ratings__score"),
followed_average_rating=Avg(
"ratings__score", filter=Q(ratings__user__in=followed_users)
),
)
.order_by("category__name", "name")
)
rated_flavors = (
user.ratings.select_related("flavor", "flavor__category")
.prefetch_related("replies", "replies__user")
.order_by("-created_at")
)
return Response(
{
"user": UserSerializer(user).data,
"rated_count": rated_flavors.count(),
"missing_count": missing_flavors.count(),
"missing_flavors": FlavorSerializer(
missing_flavors, many=True, context={"request": request}
).data,
"my_ratings": RatingSerializer(
rated_flavors, many=True, context={"request": request}
).data,
}
)
class NotificationViewSet(viewsets.ReadOnlyModelViewSet):
serializer_class = NotificationSerializer
permission_classes = [permissions.IsAuthenticated]
pagination_class = None
def get_queryset(self):
return self.request.user.notifications.select_related(
"actor", "rating__flavor", "reply__rating__flavor"
).order_by("-created_at")
@action(detail=False, methods=["post"])
def mark_all_read(self, request: Request) -> Response:
request.user.notifications.filter(is_read=False).update(is_read=True)
return Response({"status": "all read"})
@action(detail=True, methods=["post"])
def mark_read(self, request: Request, pk=None) -> Response:
notification = self.get_object()
notification.is_read = True
notification.save()
return Response({"status": "marked read"})