Files
mandelstudio/mandelstudio/idea_marketplace.py

477 lines
19 KiB
Python

from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
from django.conf import settings
from django.db.models import Q
from django.utils.text import slugify
from oscar.core.loading import get_model
from mandelstudio.content_hygiene import DEMO_MARKERS
IDEA_PRODUCT_CLASS_NAME = "Idea Product"
DIGITAL_IDEAS_CATEGORY_NAME = "Digital Ideas"
SHORT_DESCRIPTION_ATTRIBUTE_CODE = "short_description"
FULL_DESCRIPTION_ATTRIBUTE_CODE = "full_description"
IDEA_PARTNER_NAME = "Mandel Blog Studio"
@dataclass(frozen=True)
class IdeaSeedItem:
title: str
short_description: str
full_description: str
price_eur: Decimal
IDEA_PRODUCTS: tuple[IdeaSeedItem, ...] = (
IdeaSeedItem(
title="B2B Webshop Starter Blueprint",
short_description=(
"Launch a B2B webshop with a quote-first buying flow and enterprise-ready trust structure. "
"Get a clear execution path from positioning to first qualified orders."
),
full_description=(
"Introduction\n"
"A practical B2B ecommerce blueprint for teams that need to sell complex offers with confidence.\n\n"
"Problem it solves\n"
"- Generic webshop setups underperform in B2B because they ignore quote-first journeys and multi-stakeholder buying.\n"
"- Sales and marketing handoff is often unclear, which slows deal velocity.\n\n"
"Step-by-step concept\n"
"1. Define ICP and buying committee signals.\n"
"2. Map quote-first vs direct checkout decision rules.\n"
"3. Build page architecture for trust, proof, and qualification.\n"
"4. Implement lead-to-order routing between website and sales ops.\n"
"5. Run a 90-day optimization loop with conversion checkpoints.\n\n"
"Tech stack\n"
"- Django + Oscar commerce core\n"
"- Wagtail CMS for structured sales content\n"
"- Analytics and event tracking for funnel visibility\n\n"
"Business value\n"
"- Faster sales-qualified lead capture\n"
"- Lower friction for enterprise buyers\n"
"- Higher conversion from product page to qualified pipeline\n\n"
"Who it is for\n"
"Founders, growth teams, and B2B operators launching or rebuilding a serious ecommerce motion."
),
price_eur=Decimal("99.00"),
),
IdeaSeedItem(
title="AI Product Description System",
short_description=(
"Scale product copy with AI while preserving brand tone, SEO intent, and quality control. "
"Turn catalog chaos into a repeatable content engine your team can trust."
),
full_description=(
"Introduction\n"
"A production content system for generating and governing high-quality product descriptions at scale.\n\n"
"Problem it solves\n"
"- Manual copywriting does not scale across growing catalogs.\n"
"- Uncontrolled AI output introduces inconsistency and factual risk.\n\n"
"Step-by-step concept\n"
"1. Define attribute schema and content rules per category.\n"
"2. Build prompt templates linked to taxonomy fields.\n"
"3. Add QA gates for accuracy, tone, and compliance.\n"
"4. Localize with multilingual adaptation rules.\n"
"5. Monitor quality with an editorial review workflow.\n\n"
"Tech stack\n"
"- Django/Wagtail content governance\n"
"- AI model orchestration with prompt templates\n"
"- Validation layer for quality and policy checks\n\n"
"Business value\n"
"- Faster time-to-publish for new products\n"
"- Consistent conversion-focused copy\n"
"- Reduced editorial costs with better control\n\n"
"Who it is for\n"
"Ecommerce teams, marketplaces, and catalog-heavy brands that need reliable AI-assisted copy operations."
),
price_eur=Decimal("49.00"),
),
IdeaSeedItem(
title="High-Converting Landing Page Framework",
short_description=(
"Build landing pages that convert with a proven structure for message clarity, proof, and CTA flow. "
"Stop guessing and launch with a repeatable conversion framework."
),
full_description=(
"Introduction\n"
"A practical landing-page framework focused on conversion, not visual noise.\n\n"
"Problem it solves\n"
"- Teams often launch pages without a clear conversion narrative.\n"
"- Weak proof and CTA sequencing create drop-off before action.\n\n"
"Step-by-step concept\n"
"1. Align offer with one core audience intent.\n"
"2. Build headline and subheadline hierarchy.\n"
"3. Add objection-handling proof blocks and trust signals.\n"
"4. Design CTA progression for low and high intent visitors.\n"
"5. Define test plan for copy, layout, and offer variants.\n\n"
"Tech stack\n"
"- Wagtail page composition\n"
"- Bootstrap 5 component patterns\n"
"- Event tracking for funnel diagnostics\n\n"
"Business value\n"
"- Higher lead quality from the same traffic\n"
"- Faster launch cycles with reusable page logic\n"
"- Better conversion through structured experimentation\n\n"
"Who it is for\n"
"Service businesses, SaaS teams, and agencies that rely on landing pages for growth."
),
price_eur=Decimal("29.00"),
),
IdeaSeedItem(
title="Subscription-Based Service Website Model",
short_description=(
"Design a subscription service website that improves activation, retention, and recurring revenue. "
"Package offers clearly and reduce churn with lifecycle-aware UX."
),
full_description=(
"Introduction\n"
"A complete website model for subscription-first service businesses.\n\n"
"Problem it solves\n"
"- Subscription sites often sell features, not ongoing outcomes.\n"
"- Poor onboarding and renewal communication increases churn risk.\n\n"
"Step-by-step concept\n"
"1. Structure offer tiers by business outcome and support level.\n"
"2. Build onboarding pages for fast activation.\n"
"3. Add lifecycle messaging for renewal and expansion.\n"
"4. Map churn-risk touchpoints and intervention moments.\n"
"5. Track retention metrics and optimize plan positioning.\n\n"
"Tech stack\n"
"- Django + Oscar for billing-ready commerce foundations\n"
"- Wagtail for lifecycle content and onboarding assets\n"
"- Event instrumentation for retention analytics\n\n"
"Business value\n"
"- Improved activation-to-retention conversion\n"
"- More predictable recurring revenue\n"
"- Clearer upgrade path across plan tiers\n\n"
"Who it is for\n"
"Founders and operators running service subscriptions with monthly or annual plans."
),
price_eur=Decimal("69.00"),
),
IdeaSeedItem(
title="Marketplace Platform Architecture (Django)",
short_description=(
"Get a scalable marketplace architecture for Django from MVP to multi-vendor growth. "
"Includes domain boundaries, payments, moderation, and operations blueprint."
),
full_description=(
"Introduction\n"
"A technical blueprint for launching and scaling a marketplace platform on Django.\n\n"
"Problem it solves\n"
"- Marketplace projects fail when core domains and workflows are not separated early.\n"
"- Teams underestimate moderation, payout, and operational complexity.\n\n"
"Step-by-step concept\n"
"1. Define bounded domains for buyers, sellers, listings, and transactions.\n"
"2. Design catalog and search architecture for growth.\n"
"3. Implement payment orchestration and settlement flow.\n"
"4. Add moderation, permissions, and abuse controls.\n"
"5. Plan observability and phased scaling from MVP to expansion.\n\n"
"Tech stack\n"
"- Django service layer and domain modules\n"
"- Oscar commerce primitives where applicable\n"
"- Queue/events for async marketplace operations\n"
"- Monitoring and operational alerting baseline\n\n"
"Business value\n"
"- Lower re-architecture risk at scale\n"
"- Faster delivery of revenue-critical flows\n"
"- Better reliability for multi-sided operations\n\n"
"Who it is for\n"
"Technical founders, CTOs, and product teams building marketplace businesses with Django."
),
price_eur=Decimal("149.00"),
),
)
def _get_attribute_text(product, code: str) -> str:
value = (
product.attribute_values.select_related("attribute")
.filter(attribute__code=code)
.first()
)
if value is None:
return ""
for field_name in (
"value_text",
"value_richtext",
"value_option",
"value_file",
"value_image",
):
field_value = getattr(value, field_name, None)
if field_value:
return str(field_value)
return ""
def _set_attribute_text(product, attribute, text: str) -> None:
ProductAttributeValue = get_model("catalogue", "ProductAttributeValue")
value_field = (
"value_richtext"
if getattr(attribute, "type", "text") == "richtext"
else "value_text"
)
value, _created = ProductAttributeValue.objects.get_or_create(
product=product,
attribute=attribute,
)
if getattr(value, value_field, "") != text:
setattr(value, value_field, text)
value.save(update_fields=[value_field])
def is_idea_product(product) -> bool:
product_class = getattr(product, "product_class", None)
return bool(product_class and product_class.name == IDEA_PRODUCT_CLASS_NAME)
def get_idea_short_description(product) -> str:
return _get_attribute_text(product, SHORT_DESCRIPTION_ATTRIBUTE_CODE) or (
getattr(product, "description", "") or ""
)
def get_idea_full_description(product) -> str:
return _get_attribute_text(product, FULL_DESCRIPTION_ATTRIBUTE_CODE)
def get_unlockable_description(product, user) -> tuple[str, bool]:
unlocked = user_has_unlocked_idea(user, product)
if unlocked:
return get_idea_full_description(product) or get_idea_short_description(
product
), True
return get_idea_short_description(product), False
def user_has_unlocked_idea(user, product) -> bool:
if not getattr(user, "is_authenticated", False):
return False
if not is_idea_product(product):
return True
Line = get_model("order", "Line")
PaymentEventQuantity = get_model("order", "PaymentEventQuantity")
paid_statuses = {
getattr(settings, "OSCAR_PAID_ORDER_STATUS", None),
getattr(settings, "OSCAR_COMPLETE_ORDER_STATUS", None),
"paid",
"complete",
"payment-complete",
"delayed-payment",
}
paid_statuses = {
status.strip().lower()
for status in paid_statuses
if isinstance(status, str) and status.strip()
}
status_match = Line.objects.filter(
order__user=user,
product_id=product.id,
).filter(
Q(order__status__in=paid_statuses)
| Q(order__status__icontains="paid")
| Q(order__status__icontains="complete")
)
if status_match.exists():
return True
# Fallback to payment event evidence so unlocking still works when status names differ per provider.
return PaymentEventQuantity.objects.filter(
line__order__user=user,
line__product_id=product.id,
quantity__gt=0,
).exists()
def _ensure_digital_ideas_category():
Category = get_model("catalogue", "Category")
existing = Category.objects.filter(name=DIGITAL_IDEAS_CATEGORY_NAME).first()
if existing:
return existing
root = (
Category.objects.filter(depth=1).order_by("path").first()
if hasattr(Category, "depth")
else None
)
if root and hasattr(root, "add_child"):
return root.add_child(name=DIGITAL_IDEAS_CATEGORY_NAME)
if hasattr(Category, "add_root"):
return Category.add_root(name=DIGITAL_IDEAS_CATEGORY_NAME)
category = Category(name=DIGITAL_IDEAS_CATEGORY_NAME)
if hasattr(category, "slug"):
category.slug = slugify(DIGITAL_IDEAS_CATEGORY_NAME)
category.save()
return category
def _ensure_product_class():
ProductClass = get_model("catalogue", "ProductClass")
product_class, _created = ProductClass.objects.get_or_create(
name=IDEA_PRODUCT_CLASS_NAME,
defaults={
"requires_shipping": False,
"track_stock": False,
},
)
if product_class.requires_shipping:
product_class.requires_shipping = False
product_class.save(update_fields=["requires_shipping"])
return product_class
def _ensure_product_attributes(product_class):
ProductAttribute = get_model("catalogue", "ProductAttribute")
text_type = getattr(ProductAttribute, "TEXT", "text")
richtext_type = getattr(ProductAttribute, "RICHTEXT", "richtext")
short_attr, _ = ProductAttribute.objects.get_or_create(
product_class=product_class,
code=SHORT_DESCRIPTION_ATTRIBUTE_CODE,
defaults={
"name": "Short description",
"type": text_type,
"required": False,
},
)
full_attr, _ = ProductAttribute.objects.get_or_create(
product_class=product_class,
code=FULL_DESCRIPTION_ATTRIBUTE_CODE,
defaults={
"name": "Full description",
"type": richtext_type,
"required": False,
},
)
return short_attr, full_attr
def _ensure_partner():
Partner = get_model("partner", "Partner")
partner, _ = Partner.objects.get_or_create(name=IDEA_PARTNER_NAME)
return partner
def _upsert_stockrecord(product, partner, price_eur: Decimal):
StockRecord = get_model("partner", "StockRecord")
defaults = {
"partner_sku": f"idea-{product.id}",
"price_currency": "EUR",
"price_excl_tax": price_eur,
"num_in_stock": 99999,
}
stockrecord, _created = StockRecord.objects.get_or_create(
product=product,
partner=partner,
defaults=defaults,
)
dirty_fields: list[str] = []
for field_name, field_value in defaults.items():
if getattr(stockrecord, field_name, None) != field_value:
setattr(stockrecord, field_name, field_value)
dirty_fields.append(field_name)
if dirty_fields:
stockrecord.save(update_fields=dirty_fields)
def seed_idea_marketplace_products(
*, purge_demo_products: bool = True, retire_non_idea_products: bool = True
) -> dict[str, int]:
Product = get_model("catalogue", "Product")
product_class = _ensure_product_class()
category = _ensure_digital_ideas_category()
short_attr, full_attr = _ensure_product_attributes(product_class)
partner = _ensure_partner()
created = 0
updated = 0
for item in IDEA_PRODUCTS:
product = Product.objects.filter(title=item.title).first()
if product is None:
product = Product(
title=item.title,
slug=slugify(item.title),
product_class=product_class,
description=item.short_description,
)
if hasattr(Product, "STANDALONE") and hasattr(product, "structure"):
product.structure = Product.STANDALONE
if hasattr(product, "is_public") and not getattr(
product, "is_public", False
):
product.is_public = True
product.save()
created += 1
else:
dirty_fields: list[str] = []
if product.product_class_id != product_class.id:
product.product_class = product_class
dirty_fields.append("product_class")
if product.description != item.short_description:
product.description = item.short_description
dirty_fields.append("description")
if hasattr(product, "slug") and product.slug != slugify(item.title):
product.slug = slugify(item.title)
dirty_fields.append("slug")
if hasattr(product, "is_public") and not getattr(
product, "is_public", False
):
product.is_public = True
dirty_fields.append("is_public")
if dirty_fields:
product.save(update_fields=dirty_fields)
updated += 1
product.categories.add(category)
_set_attribute_text(product, short_attr, item.short_description)
_set_attribute_text(product, full_attr, item.full_description)
_upsert_stockrecord(product, partner, item.price_eur)
deleted_demo = 0
if purge_demo_products:
keep_titles = {item.title for item in IDEA_PRODUCTS}
demo_filter = Q()
for marker in DEMO_MARKERS:
demo_filter |= Q(title__icontains=marker) | Q(slug__icontains=marker)
demo_queryset = Product.objects.filter(demo_filter).exclude(
title__in=keep_titles
)
# Also purge any non-canonical products lingering in the Idea Product class
# or explicitly grouped under the Digital Ideas category.
non_canonical_ideas_queryset = (
Product.objects.filter(
Q(product_class=product_class)
| Q(categories__name__iexact=DIGITAL_IDEAS_CATEGORY_NAME)
)
.exclude(title__in=keep_titles)
.distinct()
)
delete_ids = set(demo_queryset.values_list("id", flat=True)) | set(
non_canonical_ideas_queryset.values_list("id", flat=True)
)
deleted_demo = len(delete_ids)
if deleted_demo:
Product.objects.filter(id__in=delete_ids).delete()
retired_non_idea = 0
if retire_non_idea_products:
keep_titles = {item.title for item in IDEA_PRODUCTS}
non_idea_public_qs = Product.objects.exclude(title__in=keep_titles).filter(
is_public=True
)
retired_non_idea = non_idea_public_qs.update(is_public=False)
return {
"created": created,
"updated": updated,
"deleted_demo": deleted_demo,
"retired_non_idea": retired_non_idea,
}